Improve perf infra display for multiple clients (#7770)

Author: eddyashton

tests/infra/basicperf.py

@@ -324,13 +324,25 @@ def run(args):
                 primary_has_stopped = False
                 while True:
                     stop_waiting = True
+                    running = []
+                    done_list = []
                     for i, remote_client in enumerate(clients):
                         done = remote_client.check_done()
-                        # all the clients need to be done
-                        LOG.info(
-                            f"Client {i} has {'completed' if done else 'not completed'} running ({time.time() - start_time:>{format_width}.2f}s / {args.client_timeout_s}s)"
-                        )
+                        if done:
+                            done_list.append(i)
+                        else:
+                            running.append(i)
                         stop_waiting = stop_waiting and done
+                    elapsed = f"{time.time() - start_time:>{format_width}.2f}s / {args.client_timeout_s}s"
+                    if len(clients) <= 3:
+                        for i, remote_client in enumerate(clients):
+                            status = "completed" if i in done_list else "not completed"
+                            LOG.info(f"Client {i} has {status} running ({elapsed})")
+                    else:
+                        parts = [f"{len(running)} running ({elapsed})"]
+                        if done_list:
+                            parts.append(f"{len(done_list)} complete")
+                        LOG.info(f"Clients: {', '.join(parts)}")
                     if stop_waiting:
                         break
                     if time.time() > start_time + args.client_timeout_s:
@@ -444,9 +456,14 @@ def table():
                             f"{remote_client.name}: First send at {first_send}, last receive at {last_recv}"
                         )
                         duration = (last_recv - first_send).total_seconds()
-                        print(
-                            f"{remote_client.name}: {len(overall)} requests in {duration}s => {len(overall)//duration}tx/s"
-                        )
+                        if duration > 0:
+                            print(
+                                f"{remote_client.name}: {len(overall)} requests in {duration}s => {len(overall)//duration}tx/s"
+                            )
+                        else:
+                            print(
+                                f"{remote_client.name}: {len(overall)} requests in {duration}s"
+                            )
                         agg.append(overall)
 
                     table()
@@ -471,6 +488,12 @@ def table():
                 start_send = agg["sendTime"].min()
                 end_recv = agg["receiveTime"].max()
                 duration_s = (end_recv - start_send).total_seconds()
+                if duration_s < 1:
+                    LOG.error(
+                        f"Duration is {duration_s}s (first send: {start_send}, last recv: {end_recv}). "
+                        "Clamping to 1s; results may be inaccurate."
+                    )
+                    duration_s = 1
                 throughput = len(agg) / duration_s
                 statistics["average_throughput_tx/s"] = throughput
                 print(f"Average throughput: {throughput:.2f} tx/s")
@@ -489,6 +512,12 @@ def table():
                     client["receiveTime"].max() for client in each_client
                 )
                 all_active_duration_s = (earliest_end - latest_start).total_seconds()
+                if all_active_duration_s <= 0:
+                    LOG.error(
+                        f"All-clients-active duration is {all_active_duration_s}s. "
+                        "Clamping to 1s; results may be inaccurate."
+                    )
+                    all_active_duration_s = 1
                 statistics["all_clients_active_from"] = latest_start.isoformat()
                 statistics["all_clients_active_to"] = earliest_end.isoformat()
                 statistics["all_clients_active_duration_s"] = all_active_duration_s
@@ -509,81 +538,183 @@ def table():
                 agg_all_active = agg.filter(pl.col("sendTime") > latest_start).filter(
                     pl.col("receiveTime") < earliest_end
                 )
-                all_active_duration_s = (earliest_end - latest_start).total_seconds()
-                all_active_throughput = len(agg_all_active) / all_active_duration_s
-                statistics["all_clients_active_average_throughput_tx/s"] = (
-                    all_active_throughput
-                )
+                n_all_active = max(len(agg_all_active), 1)
+                all_active_throughput = n_all_active / all_active_duration_s
                 writes = len(
                     agg_all_active.filter(pl.col("request").bin.starts_with(b"PUT "))
                 )
-                statistics["all_clients_active_write_fraction"] = writes / len(
-                    agg_all_active
+                write_fraction = writes / n_all_active
+                statistics["all_clients_active_average_throughput_tx/s"] = (
+                    all_active_throughput
                 )
+                statistics["all_clients_active_write_fraction"] = write_fraction
 
                 statistics_path = os.path.join(network.common_dir, "statistics.json")
                 with open(statistics_path, "w") as f:
                     json.dump(statistics, f, indent=2)
                 print(f"Aggregated statistics written to {statistics_path}")
 
-                sent_per_sec = (
-                    agg.with_columns(
-                        (
-                            (pl.col("sendTime").alias("second") - start_send) / 1000000
-                        ).cast(pl.Int64)
+                # Build per-client, per-second breakdown for stacked charts
+                chart_width = 100
+                max_builtin_legend_clients = 3
+
+                client_names = sorted(
+                    agg["client"].unique().to_list(),
+                    key=lambda c: int(c.split("_")[-1]),
+                )
+                n_clients = len(client_names)
+                all_seconds = set()
+
+                # Color gradient helpers: interpolate n RGB colors from start to end.
+                # client_0 gets the darkest shade (matching plotext defaults), higher clients get lighter.
+                def _shades(n, start, end):
+                    return [
+                        tuple(
+                            s + i * (e - s) // max(n - 1, 1) for s, e in zip(start, end)
+                        )
+                        for i in range(n)
+                    ]
+
+                def blue_shades(n):
+                    return _shades(n, (59, 142, 234), (100, 100, 255))
+
+                def green_shades(n):
+                    return _shades(n, (13, 188, 121), (100, 255, 100))
+
+                def red_shades(n):
+                    return _shades(n, (205, 49, 49), (255, 100, 100))
+
+                def format_title(title, width):
+                    """Centered title bar in the style of plotext's simple_bar charts."""
+                    if title is None:
+                        return ""
+                    visible_len = len(plt.uncolorize(title))
+                    w1 = (width - 2 - visible_len) // 2
+                    w2 = width - visible_len - 2 - w1
+                    return (
+                        plt.colorize(
+                            "─" * w1 + " " + title + " " + "─" * w2, "gray+", "bold"
+                        )
+                        + "\n"
                     )
-                    .group_by("second")
+
+                def color_legend(names, *color_groups):
+                    """Print a legend bar with color swatches for first and last client."""
+                    marker = "▇" * 3
+                    parts = []
+                    for color_list, suffix in color_groups:
+                        first = plt.colorize(marker, color_list[0])
+                        last = plt.colorize(marker, color_list[-1])
+                        parts.append(
+                            f"{names[0]}{suffix} {first} .. {last} {names[-1]}{suffix}"
+                        )
+                    print(format_title(" ".join(parts), chart_width), end="")
+
+                # Generate per-client color palettes
+                sent_colors = blue_shades(n_clients)
+                rcvd_colors = green_shades(n_clients)
+                error_colors = red_shades(n_clients)
+
+                # Compute per-client, per-second counts in a single pass
+                agg_with_seconds = agg.with_columns(
+                    ((pl.col("sendTime") - start_send) / 1000000)
+                    .cast(pl.Int64)
+                    .alias("send_second"),
+                    ((pl.col("receiveTime") - start_send) / 1000000)
+                    .cast(pl.Int64)
+                    .alias("recv_second"),
+                    (pl.col("responseStatus") >= 500).alias("is_error"),
+                )
+
+                sent_counts = (
+                    agg_with_seconds.group_by("client", "send_second")
                     .len()
-                    .rename({"len": "sent"})
+                    .rename({"send_second": "second", "len": "count"})
                 )
-                recv_per_sec = (
-                    agg.with_columns(
-                        (
-                            (pl.col("receiveTime").alias("second") - start_send)
-                            / 1000000
-                        ).cast(pl.Int64)
-                    )
-                    .group_by("second")
+                rcvd_counts = (
+                    agg_with_seconds.filter(~pl.col("is_error"))
+                    .group_by("client", "recv_second")
                     .len()
-                    .rename({"len": "rcvd"})
+                    .rename({"recv_second": "second", "len": "count"})
                 )
-                errors_per_sec = (
-                    agg.with_columns(
-                        (
-                            (pl.col("receiveTime").alias("second") - start_send)
-                            / 1000000
-                        ).cast(pl.Int64)
-                    )
-                    .filter(pl.col("responseStatus") >= 500)
-                    .group_by("second")
+                error_counts = (
+                    agg_with_seconds.filter(pl.col("is_error"))
+                    .group_by("client", "recv_second")
                     .len()
-                    .rename({"len": "errors"})
+                    .rename({"recv_second": "second", "len": "count"})
                 )
 
-                per_sec = (
-                    sent_per_sec.join(recv_per_sec, on="second")
-                    .join(errors_per_sec, on="second", how="full")
-                    .sort("second")
-                    .fill_null(0)
-                )
+                client_sent = {}
+                client_rcvd = {}
+                client_errors = {}
+                for cn in client_names:
+                    cs = sent_counts.filter(pl.col("client") == cn)
+                    client_sent[cn] = dict(
+                        zip(cs["second"].to_list(), cs["count"].to_list())
+                    )
+                    all_seconds.update(cs["second"].to_list())
+
+                    cr = rcvd_counts.filter(pl.col("client") == cn)
+                    client_rcvd[cn] = dict(
+                        zip(cr["second"].to_list(), cr["count"].to_list())
+                    )
+                    all_seconds.update(cr["second"].to_list())
 
-                plt.simple_bar(
-                    list(per_sec["second"]),
-                    list(per_sec["sent"]),
-                    width=100,
-                    title="Sent requests per second",
+                    ce = error_counts.filter(pl.col("client") == cn)
+                    client_errors[cn] = dict(
+                        zip(ce["second"].to_list(), ce["count"].to_list())
+                    )
+                    all_seconds.update(ce["second"].to_list())
+
+                seconds = sorted(all_seconds)
+
+                use_builtin_legend = n_clients <= max_builtin_legend_clients
+
+                # Stacked bar: sent per second, blue shades per client
+                sent_stacks = [
+                    [client_sent[cn].get(s, 0) for s in seconds] for cn in client_names
+                ]
+                sent_kwargs = {"colors": sent_colors}
+                if use_builtin_legend:
+                    sent_kwargs["labels"] = client_names
+                plt.simple_stacked_bar(
+                    seconds,
+                    sent_stacks,
+                    width=chart_width,
+                    title="Sent requests per second (by client)",
+                    **sent_kwargs,
                 )
                 plt.show()
+                if not use_builtin_legend:
+                    color_legend(client_names, (sent_colors, ""))
 
+                # Stacked bar: received per second, green shades per client + red for errors
+                rcvd_stacks = [
+                    [client_rcvd[cn].get(s, 0) for s in seconds] for cn in client_names
+                ]
+                error_stacks = [
+                    [client_errors[cn].get(s, 0) for s in seconds]
+                    for cn in client_names
+                ]
+                rcvd_kwargs = {"colors": rcvd_colors + error_colors}
+                if use_builtin_legend:
+                    rcvd_kwargs["labels"] = client_names + [
+                        f"{cn} errors" for cn in client_names
+                    ]
                 plt.simple_stacked_bar(
-                    list(per_sec["second"]),
-                    [list(per_sec["rcvd"]), list(per_sec["errors"])],
-                    width=100,
-                    labels=["rcvd", "errors"],
-                    colors=["green", "red"],
-                    title="Received requests per second",
+                    seconds,
+                    rcvd_stacks + error_stacks,
+                    width=chart_width,
+                    title="Received requests per second (by client)",
+                    **rcvd_kwargs,
                 )
                 plt.show()
+                if not use_builtin_legend:
+                    color_legend(
+                        client_names,
+                        (rcvd_colors, ""),
+                        (error_colors, " errors"),
+                    )
 
                 if number_of_errors and not args.stop_primary_after_s:
                     raise RuntimeError(