Marimo Notebooks — latency network exploration.

Networking

If you’ve spent much time in Jupyter, you know the rhythm: a mix of coding, testing snippets, plotting data, and writing notes. Marimo feels familiar, but it nudges that workflow into a smoother, more responsive space. You can write Python code in cells, experiment with ideas, and immediately see results, all without juggling multiple scripts or constantly switching contexts.

What stands out is how it handles the runtime. Variables, imports, and objects persist across cells, so you can build on previous experiments without having to restart the kernel for every small change. That’s a small detail, but it changes how you explore. You can prototype a function, test it with real data, tweak parameters, and visualize the output in-line, all without losing context. Plotting works out of the box — matplotlib, seaborn, even interactive charts behave as you’d expect. It’s like your Python REPL grew up and learned to keep your workspace organized.

Marimo also doesn’t get in the way of experimentation. Each notebook runs in its own sandbox, so you can play with network calls, system commands, or new libraries without worrying about breaking other projects. And sharing your experiments is straightforward — whether you want a colleague to reproduce results or just keep a record of your own iterations.

For anyone who codes in Python and enjoys exploring, debugging, or visualizing data, Marimo isn’t a flashy product — it’s a workspace designed to match how you think. It lets you focus on what matters: testing ideas, understanding behavior, and iterating quickly, without the friction of context switching or cumbersome setup.

Below you will see some examples which we have been running from within a Marimo Notebook.

Example 1 — TCP ping

In below example snippet, we’re measuring network latency using a TCP connection instead of the traditional ICMP ping. The tcp_ping function tries to open a connection to a host and port — 443 by default — and records how long it takes for the handshake to complete. If the connection fails, it simply returns None.

To see trends over time, ping_many runs this check repeatedly, printing each result as it goes and keeping a record of the successful measurements. Finally, we visualize the data with matplotlib, plotting each ping in sequence so you can quickly see how latency fluctuates.

Using TCP for pinging has a few advantages over ICMP. Many networks block ICMP packets for security reasons, so a traditional ping might fail even if the host is reachable. A TCP-based approach works through firewalls that allow standard traffic, like HTTPS, and gives a realistic view of the latency you’d experience when actually connecting to a service. It’s not just a workaround — it can be a more practical way to monitor network performance from a developer’s perspective.

Below you can see the full code example as it was part of this specific cell in the Marimo notebook.

import socket
import time
import matplotlib.pyplot as plt

def tcp_ping(host: str, port: int = 443, timeout: float = 1.0) -> float | None:
    start = time.time()
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return (time.time() - start) * 1000
    except OSError:
        return None

def ping_many(host: str, count: int = 100, interval: float = 0.2):
    results = []
    timestamps = []

    for i in range(1, count + 1):
        delay = tcp_ping(host)

        if delay is None:
            print(f"{i:03d}: timeout")
        else:
            print(f"{i:03d}: {delay:.2f} ms")
            results.append(delay)
            timestamps.append(i)

        time.sleep(interval)

    return timestamps, results

# Run the pings
x, y = ping_many("8.8.8.8", count=100)

# Plot the results
if y:
    plt.figure()
    plt.plot(x, y, marker="o")
    plt.xlabel("Ping sequence")
    plt.ylabel("Latency (ms)")
    plt.title("TCP Ping Latency over Time")
    plt.grid(True)
    plt.show()
else:
    print("No successful pings to plot")

Example 2 — Connection process breakdown

This snippet is a small experiment in measuring network latency by breaking down the connection process into two phases: DNS resolution and TCP handshake. Instead of relying on ICMP ping, it opens a real TCP connection to a host , in this case, Cloudflare’s DNS at 1.1.1.1 on port 53 , and measures the time it takes to complete each step.

The measure_connection function handles the two phases separately. First, it records how long it takes to resolve the hostname to an IP address using socket.getaddrinfo. Then, it measures the time to perform a TCP handshake by opening a socket and connecting to the resolved address. Each phase’s duration is stored in milliseconds, and the total time is calculated as the sum of DNS and TCP times.

The main loop runs this measurement repeatedly, printing results for each attempt and saving successful measurements. A short delay between attempts keeps things readable and avoids flooding the network.

Finally, the script visualizes the data with a stacked bar chart: DNS times form the bottom of each bar, TCP handshake times are stacked on top. This makes it easy to see which part of the connection contributes most to latency. It also prints a summary with average DNS and TCP times, giving a quick overview of network performance for the target host.

In short, this code turns a simple connection into a step-by-step latency breakdown, letting you see how DNS resolution and TCP handshake individually affect overall responsiveness.

Below you can see the full code example as it was part of this specific cell in the Marimo notebook.

import socket as netsock
import time as timemod
import statistics as stats
import matplotlib.pyplot as mplot

HOST = "1.1.1.1"  # Cloudflare DNS
PORT = 53         # TCP port for DNS
COUNT = 100
TIMEOUT = 5.0

def measure_connection(host: str, port: int):
    timings = {}

    # DNS resolution
    t0 = timemod.perf_counter()
    try:
        addr_info = netsock.getaddrinfo(host, port, netsock.AF_INET, netsock.SOCK_STREAM)
        address = addr_info[0][4]
        timings["dns_ms"] = (timemod.perf_counter() - t0) * 1000
    except OSError:
        return None

    # TCP handshake
    t0 = timemod.perf_counter()
    try:
        sock = netsock.socket(netsock.AF_INET, netsock.SOCK_STREAM)
        sock.settimeout(TIMEOUT)
        sock.connect(address)
        timings["tcp_ms"] = (timemod.perf_counter() - t0) * 1000
    except OSError:
        return None

    sock.close()
    timings["total_ms"] = timings["dns_ms"] + timings["tcp_ms"]
    return timings

results = []

print("seq |   dns ms |   tcp ms | total ms")
print("----+----------+----------+----------")

for i in range(1, COUNT + 1):
    measurement = measure_connection(HOST, PORT)

    if measurement is None:
        print(f"{i:3d} | timeout")
    else:
        results.append(measurement)
        print(
            f"{i:3d} | "
            f"{measurement['dns_ms']:8.2f} | "
            f"{measurement['tcp_ms']:8.2f} | "
            f"{measurement['total_ms']:8.2f}"
        )

    timemod.sleep(0.2)

# Stacked bar chart
if results:
    dns = [r["dns_ms"] for r in results]
    tcp = [r["tcp_ms"] for r in results]
    indices = list(range(1, len(results) + 1))

    mplot.figure(figsize=(12, 6))
    mplot.bar(indices, dns, label="DNS")
    mplot.bar(indices, tcp, bottom=dns, label="TCP handshake")
    mplot.xlabel("Sample")
    mplot.ylabel("Latency (ms)")
    mplot.title("Connection step latency breakdown (stacked)")
    mplot.legend()
    mplot.grid(axis="y")
    mplot.show()

    print("\nSummary (ms)")
    print(f"DNS   avg: {stats.mean(dns):.2f}")
    print(f"TCP   avg: {stats.mean(tcp):.2f}")
else:
    print("\nNo successful measurements collected")

Example 3 — Measure DNS, TCP, TTFB

This snippet takes the idea of measuring network latency a step further by breaking down the connection to an external HTTP server into three phases: DNS resolution, TCP handshake, and Time to First Byte (TTFB). Instead of just checking if a host is reachable, it measures how long each stage of a real HTTP request takes.

The measure_http_connection_safe function handles this step by step. First, it resolves the hostname to an IP address and records how long that takes. Next, it establishes a TCP connection to the server’s HTTP port and times the handshake. Finally, it sends a minimal HEAD request and waits for the first byte of the response, capturing TTFB — the moment the server actually starts sending data. Each measurement is stored in milliseconds, and the total is computed as the sum of all three phases.

The main loop repeats this process multiple times, printing each attempt and collecting successful measurements. A short delay between attempts prevents overwhelming the server and makes the results easier to read.

To visualize the data, the script uses a stacked bar chart: DNS times form the base, TCP handshake times are stacked above, and TTFB sits on top. This clearly shows which part of the connection contributes most to latency, and the script also prints average timings for each phase, giving a concise summary of network performance for the server.

In essence, this approach turns a simple HTTP request into a stepwise latency analysis, letting developers see how DNS, TCP, and server response times individually affect perceived performance.

Below you can see the full code example as it was part of this specific cell in the Marimo notebook.

import socket as netsock_safe
import time as timemod_safe
import matplotlib.pyplot as mplot_safe
import statistics as stats_safe

# --- Configuration ---
HOST_SAFE = "mail.ru"  # External HTTP server
PORT_SAFE = 80             # HTTP port
COUNT_SAFE = 100
TIMEOUT_SAFE = 5.0

def measure_http_connection_safe(host: str, port: int):
    timings_safe = {}

    # --- DNS resolution ---
    t0_safe = timemod_safe.perf_counter()
    try:
        addr_info_safe = netsock_safe.getaddrinfo(host, port, netsock_safe.AF_INET, netsock_safe.SOCK_STREAM)
        address_safe = addr_info_safe[0][4]
        timings_safe["dns_ms"] = (timemod_safe.perf_counter() - t0_safe) * 1000
    except OSError:
        return None

    # --- TCP handshake ---
    t0_safe = timemod_safe.perf_counter()
    try:
        sock_safe = netsock_safe.socket(netsock_safe.AF_INET, netsock_safe.SOCK_STREAM)
        sock_safe.settimeout(TIMEOUT_SAFE)
        sock_safe.connect(address_safe)
        timings_safe["tcp_ms"] = (timemod_safe.perf_counter() - t0_safe) * 1000
    except OSError:
        return None

    # --- Time to first byte (TTFB) ---
    t0_safe = timemod_safe.perf_counter()
    try:
        request_safe = f"HEAD / HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n"
        sock_safe.sendall(request_safe.encode())
        sock_safe.recv(1)  # Wait for first byte
        timings_safe["ttfb_ms"] = (timemod_safe.perf_counter() - t0_safe) * 1000
    except OSError:
        sock_safe.close()
        return None

    sock_safe.close()
    timings_safe["total_ms"] = timings_safe["dns_ms"] + timings_safe["tcp_ms"] + timings_safe["ttfb_ms"]
    return timings_safe

# --- Run measurements ---
results_safe = []

print("seq |   dns ms |   tcp ms | ttfb ms | total ms")
print("----+----------+----------+---------+----------")

for i_safe in range(1, COUNT_SAFE + 1):
    measurement_safe = measure_http_connection_safe(HOST_SAFE, PORT_SAFE)

    if measurement_safe is None:
        print(f"{i_safe:3d} | timeout")
    else:
        results_safe.append(measurement_safe)
        print(
            f"{i_safe:3d} | "
            f"{measurement_safe['dns_ms']:8.2f} | "
            f"{measurement_safe['tcp_ms']:8.2f} | "
            f"{measurement_safe['ttfb_ms']:7.2f} | "
            f"{measurement_safe['total_ms']:8.2f}"
        )

    timemod_safe.sleep(0.5)

# --- Plot stacked bar chart ---
if results_safe:
    dns_safe = [r["dns_ms"] for r in results_safe]
    tcp_safe = [r["tcp_ms"] for r in results_safe]
    ttfb_safe = [r["ttfb_ms"] for r in results_safe]
    indices_safe = list(range(1, len(results_safe) + 1))

    mplot_safe.figure(figsize=(12, 6))
    mplot_safe.bar(indices_safe, dns_safe, label="DNS")
    mplot_safe.bar(indices_safe, tcp_safe, bottom=dns_safe, label="TCP handshake")
    bottom_cumulative_safe = [d + t for d, t in zip(dns_safe, tcp_safe)]
    mplot_safe.bar(indices_safe, ttfb_safe, bottom=bottom_cumulative_safe, label="TTFB")
    mplot_safe.xlabel("Sample")
    mplot_safe.ylabel("Latency (ms)")
    mplot_safe.title("Connection latency breakdown: DNS → TCP → TTFB (mail.ru)")
    mplot_safe.legend()
    mplot_safe.grid(axis="y")
    mplot_safe.show()

    # --- Summary ---
    print("\nSummary (ms)")
    print(f"DNS          avg: {stats_safe.mean(dns_safe):.2f}")
    print(f"TCP handshake avg: {stats_safe.mean(tcp_safe):.2f}")
    print(f"TTFB         avg: {stats_safe.mean(ttfb_safe):.2f}")
else:
    print("\nNo successful measurements collected")

Closing words

You can find the full Marimo notebook containing the examples shown above in the following file on Github: https://github.com/louwersj/notebooks_marimo/blob/main/pingtest.py

About the author(s) Johan Louwers is currently Chief Enterprise Architect for a large global tech company as well as the lead architect for NATO and a number of militaries. Johan has a strong and long background in the field of Enterprise Architecture and complex system engineering. Having worked with enterprises in a diverse set of industries as (enterprise) architect, CTO and technical and strategic business advisor Johan brings both deep technical knowledge to the table as well as strong business oriented expertise. In addition to this Johan is a tech addict who tends to enjoy supporting open source initiatives and actively coding as a hobby. Views expressed in this post are personnel and do not necessarily reflect the views of my current employer.