analysis.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f9c0bcbf-b498-45d5-a12a-fee2d4220a88",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "%matplotlib widget\n",
    "\n",
    "import collections\n",
    "import json\n",
    "import math\n",
    "import os\n",
    "from os import path\n",
    "\n",
    "import numpy as np\n",
    "from matplotlib import pyplot as plt\n",
    "\n",
    "# TODO: Add burstiness analysis from receiver pcap, flow level\n",
    "\n",
    "FONTSIZE = 15\n",
    "LINESIZE = 3\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    RUN = True\n",
    "else:\n",
    "    RUN = False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4915fdcb-ffa2-4dc1-83d0-df65b094043b",
   "metadata": {},
   "outputs": [],
   "source": [
    "if RUN:\n",
    "    EXP_DIR = \"/data_ssd/ccanel/data/imc2024/sweep/background-senders/15ms-100-0-11-TcpDctcp-10000mbps-2000000B-10icwnd-0offset-none-rwnd0B-20tokens-4g-65ecn-1_0da\"\n",
    "    EXP = path.basename(EXP_DIR)\n",
    "    GRAPH_DIR = path.join(EXP_DIR, \"graphs\")\n",
    "    if not path.isdir(GRAPH_DIR):\n",
    "        os.makedirs(GRAPH_DIR)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "458d03cb-7e6d-4a8a-a28a-c39e3b2fc772",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def show(fig):\n",
    "    plt.tight_layout()\n",
    "    # Change the toolbar position\n",
    "    fig.canvas.toolbar_position = \"left\"\n",
    "    # If true then scrolling while the mouse is over the canvas will not\n",
    "    # move the entire notebook\n",
    "    fig.canvas.capture_scroll = True\n",
    "    fig.show()\n",
    "\n",
    "\n",
    "def save(graph_dir, prefix=None, suffix=None, extent=None):\n",
    "    \"\"\"Save the entire figure.\"\"\"\n",
    "    assert prefix is not None or suffix is not None\n",
    "    both_defined = prefix is not None and suffix is not None\n",
    "    out_flp = path.join(\n",
    "        graph_dir,\n",
    "        (\"\" if prefix is None else prefix)\n",
    "        + (\"_\" if both_defined else \"\")\n",
    "        + (\"\" if suffix is None else suffix),\n",
    "    )\n",
    "    plt.tight_layout()\n",
    "    plt.savefig(out_flp + \".pdf\", bbox_inches=\"tight\" if extent is None else extent)\n",
    "    plt.savefig(\n",
    "        out_flp + \".png\", dpi=300, bbox_inches=\"tight\" if extent is None else extent\n",
    "    )\n",
    "\n",
    "\n",
    "def save_axes(figure, axes, graph_dir, prefix=None, suffix=None):\n",
    "    \"\"\"Save a single axes, instead of the entire figure.\"\"\"\n",
    "    extent = axes.get_window_extent().transformed(figure.dpi_scale_trans.inverted())\n",
    "    save(graph_dir, prefix, suffix, extent.expanded(1.3, 1.2))\n",
    "\n",
    "\n",
    "def get_axes(rows=1, width=8, height=3, cols=1):\n",
    "    with plt.ioff():\n",
    "        fig, axes = plt.subplots(figsize=(width, height * rows), nrows=rows, ncols=cols)\n",
    "\n",
    "    if rows == 1:\n",
    "        axes = [axes]\n",
    "    elif cols == 1:\n",
    "        axes = axes.flatten()\n",
    "    return fig, axes\n",
    "\n",
    "\n",
    "def get_aligned_xs(old_start_sec, old_end_sec, interp_delta):\n",
    "    # Create a new xs ndarray, ranging from >= old_start_sec to <= old_end_sec, aligned\n",
    "    # at intervals of 1 / interp_delta.\n",
    "\n",
    "    # Round the start *up* to the nearest multiple of 1 / interp_delta.\n",
    "    #     math.ceil(start  / (1 / interp_delta)) * (1 / interp_delta)\n",
    "    new_start_sec = math.ceil(old_start_sec * interp_delta) / interp_delta\n",
    "\n",
    "    # Round the end *down* to the nearest multiple of 1 / interp_delta.\n",
    "    #     math.floor(end / (1 / interp_delta)) * (1 / interp_delta)\n",
    "    new_end_sec = math.floor(old_end_sec * interp_delta) / interp_delta\n",
    "\n",
    "    # If the old start and end are so close together that they do not overlap an\n",
    "    # aligned interval, then we cannot do anything.\n",
    "    if new_start_sec > new_end_sec:\n",
    "        return np.empty(0)\n",
    "\n",
    "    return np.array(\n",
    "        [\n",
    "            x / interp_delta\n",
    "            for x in range(\n",
    "                math.ceil(new_start_sec * interp_delta),\n",
    "                math.floor(new_end_sec * interp_delta) + 1,\n",
    "            )\n",
    "        ]\n",
    "    )\n",
    "\n",
    "\n",
    "def filter_samples(samples, start, end):\n",
    "    return [sample for sample in samples if start <= sample[0] <= end]\n",
    "\n",
    "\n",
    "def separate_samples_into_bursts(\n",
    "    samples,\n",
    "    burst_times,\n",
    "    flow_times=None,\n",
    "    filter_on_flow_times=False,\n",
    "    bookend=True,\n",
    "):\n",
    "    num_bursts = len(burst_times)\n",
    "    bursts = []\n",
    "\n",
    "    if filter_on_flow_times:\n",
    "        assert flow_times is not None\n",
    "    else:\n",
    "        assert flow_times is None\n",
    "        flow_times = [(None, None, None, None)] * num_bursts\n",
    "\n",
    "    for burst_idx, (\n",
    "        (burst_start, burst_end),\n",
    "        (flow_start, _, flow_end, _),\n",
    "    ) in enumerate(zip(burst_times, flow_times)):\n",
    "        burst = []\n",
    "        for sample in samples:\n",
    "            if burst_start <= sample[0] <= burst_end and (\n",
    "                not filter_on_flow_times or flow_start <= sample[0] <= flow_end\n",
    "            ):\n",
    "                # This sample is part of the current burst.\n",
    "                burst.append(sample)\n",
    "\n",
    "        # Insert a sample at precisely the start and end time for this burst,\n",
    "        # if possible.\n",
    "        if bookend:\n",
    "            start, end = (\n",
    "                (flow_start, flow_end)\n",
    "                if filter_on_flow_times\n",
    "                else (burst_start, burst_end)\n",
    "            )\n",
    "            if burst_idx > 0:\n",
    "                # Make sure that the burst has a sample at the start time\n",
    "                # Two case: Either we have no samples for this burst, so we take\n",
    "                # the last value from the previous burst, or we do have samples for\n",
    "                # this burst but not at the start time, so we also take the last\n",
    "                # value from the previous burst. In both cases, make sure there is\n",
    "                # a previous burst.\n",
    "                if (not burst and bursts[-1]) or (\n",
    "                    burst and burst[0][0] != start and bursts[-1]\n",
    "                ):\n",
    "                    burst.insert(0, (start, *bursts[-1][-1][1:]))\n",
    "            # Every burst should now have at least one sample: start.\n",
    "            # Note: This will fail if we have no data for the first burst.\n",
    "\n",
    "            if burst:\n",
    "                # Make sure that the burst has a sample at the end time\n",
    "                if burst[-1][0] != end:\n",
    "                    burst.append((end, *burst[-1][1:]))\n",
    "                # Every burst should now have at least two samples: start and end\n",
    "                assert len(burst) >= 2, (burst, start, end)\n",
    "\n",
    "        bursts.append(burst)\n",
    "    # Make sure we have the expected number of bursts\n",
    "    assert len(bursts) == num_bursts\n",
    "\n",
    "    # Adjust the x values of the samples\n",
    "    bursts_ = []\n",
    "    for (start, end), burst in zip(burst_times, bursts):\n",
    "        burst_ = []\n",
    "        for sample in burst:\n",
    "            burst_.append((sample[0] - start, *sample[1:]))\n",
    "        bursts_.append(burst_)\n",
    "    return bursts_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0a292b86-8cfd-42af-b6a5-1b316ec1a8be",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "def parse_times_line(line):\n",
    "    # Format: <start time seconds> <end time seconds>\n",
    "    parts = line.strip().split(\" \")\n",
    "    assert len(parts) == 2\n",
    "    return [float(sec) for sec in parts]\n",
    "\n",
    "\n",
    "def get_burst_times(exp_dir):\n",
    "    with open(\n",
    "        path.join(exp_dir, \"logs\", \"burst_times.log\"), \"r\", encoding=\"utf-8\"\n",
    "    ) as fil:\n",
    "        return [parse_times_line(line) for line in fil if line.strip()[0] != \"#\"]\n",
    "\n",
    "\n",
    "def get_config_json(exp_dir):\n",
    "    with open(path.join(exp_dir, \"config.json\"), \"r\", encoding=\"utf-8\") as fil:\n",
    "        return json.load(fil)\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    BURST_TIMES = get_burst_times(EXP_DIR)\n",
    "    # BURST_TIMES = [(start, (start + 0.03) if (end - start) > 0.03 else end) for start, end in BURST_TIMES]\n",
    "    # BURST_TIMES = BURST_TIMES[:3]\n",
    "    NUM_BURSTS = len(BURST_TIMES)\n",
    "    CONFIG = get_config_json(EXP_DIR)\n",
    "    # assert NUM_BURSTS == CONFIG[\"numBursts\"]\n",
    "\n",
    "    # ideal_sec = CONFIG[\"bytesPerSender\"] * CONFIG[\"numSenders\"] / (\n",
    "    ideal_sec = CONFIG[\"bytesPerBurstSender\"] * CONFIG[\"numBurstSenders\"] / (\n",
    "        CONFIG[\"smallLinkBandwidthMbps\"] * 1e6 / 8\n",
    "    ) + (6 * CONFIG[\"delayPerLinkUs\"] / 1e6)\n",
    "    print(\n",
    "        \"Burst times:\",\n",
    "        f\"Ideal: {ideal_sec * 1e3:.4f} ms\",\n",
    "        *[\n",
    "            (\n",
    "                f\"{burst_idx + 1}: [{start} -> {end}] - \"\n",
    "                f\"{(end - start) * 1e3:.4f} ms - \"\n",
    "                f\"{(end - start) / ideal_sec * 100:.2f} %\"\n",
    "            )\n",
    "            for burst_idx, (start, end) in enumerate(BURST_TIMES)\n",
    "        ],\n",
    "        sep=\"\\n\",\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8a68c936-ec67-4327-8d31-eb40619cb768",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def parse_length_line(line):\n",
    "    # Format: <timestamp seconds> <num packets> <backlog time microseconds>\n",
    "    parts = line.strip().split(\" \")\n",
    "    assert len(parts) == 2\n",
    "    time_sec, packets = parts\n",
    "    time_sec = float(time_sec)\n",
    "    packets = int(packets)\n",
    "    # backlog_us = float(backlog_us)\n",
    "    return time_sec, packets  # , backlog_us\n",
    "\n",
    "\n",
    "def parse_mark_line(line):\n",
    "    # Format <timestamp seconds>\n",
    "    parts = line.strip().split(\" \")\n",
    "    assert len(parts) == 1\n",
    "    return (float(parts[0]), None)\n",
    "\n",
    "\n",
    "def parse_drop_line(line):\n",
    "    # Format: <timestamp seconds> <drop type>\n",
    "    parts = line.strip().split(\" \")\n",
    "    assert len(parts) == 2\n",
    "    time_sec, drop_type = parts\n",
    "    time_sec = float(time_sec)\n",
    "    drop_type = int(drop_type)\n",
    "    return time_sec, drop_type\n",
    "\n",
    "\n",
    "def get_lengths_by_burst(exp_dir, queue_prefix, burst_times):\n",
    "    length_samples = []\n",
    "    with open(\n",
    "        path.join(exp_dir, \"logs\", f\"{queue_prefix}_depth.log\"), \"r\", encoding=\"utf-8\"\n",
    "    ) as fil:\n",
    "        length_samples = [\n",
    "            parse_length_line(line) for line in fil if line.strip()[0] != \"#\"\n",
    "        ]\n",
    "    return separate_samples_into_bursts(length_samples, burst_times)\n",
    "\n",
    "\n",
    "def get_marks_by_burst(exp_dir, queue_prefix, burst_times):\n",
    "    mark_samples = []\n",
    "    with open(\n",
    "        path.join(exp_dir, \"logs\", f\"{queue_prefix}_mark.log\"), \"r\", encoding=\"utf-8\"\n",
    "    ) as fil:\n",
    "        mark_samples = [parse_mark_line(line) for line in fil if line.strip()[0] != \"#\"]\n",
    "    return separate_samples_into_bursts(mark_samples, burst_times, bookend=False)\n",
    "\n",
    "\n",
    "def get_drops_by_burst(exp_dir, queue_prefix, burst_times):\n",
    "    drop_samples = []\n",
    "    with open(\n",
    "        path.join(exp_dir, \"logs\", f\"{queue_prefix}_drop.log\"), \"r\", encoding=\"utf-8\"\n",
    "    ) as fil:\n",
    "        drop_samples = [parse_drop_line(line) for line in fil if line.strip()[0] != \"#\"]\n",
    "    return separate_samples_into_bursts(drop_samples, burst_times, bookend=False)\n",
    "\n",
    "\n",
    "def get_queue_metrics_by_burst(exp_dir, queue_name, burst_times):\n",
    "    queue_prefix = (\n",
    "        \"incast_queue\"\n",
    "        if queue_name == \"Incast Queue\"\n",
    "        else (\"uplink_queue\" if queue_name == \"Uplink Queue\" else None)\n",
    "    )\n",
    "    assert queue_prefix is not None\n",
    "    return {\n",
    "        \"lengths\": get_lengths_by_burst(exp_dir, queue_prefix, burst_times),\n",
    "        \"marks\": get_marks_by_burst(exp_dir, queue_prefix, burst_times),\n",
    "        \"drops\": get_drops_by_burst(exp_dir, queue_prefix, burst_times),\n",
    "    }\n",
    "\n",
    "\n",
    "def graph_queue(\n",
    "    queue_name,\n",
    "    lengths_by_burst,\n",
    "    marks_by_burst,\n",
    "    drops_by_burst,\n",
    "    marking_threshold_packets,\n",
    "    capacity_packets,\n",
    "    graph_dir,\n",
    "    prefix,\n",
    "):\n",
    "    for burst_idx, burst in enumerate(lengths_by_burst):\n",
    "        # if burst_idx != len(lengths_by_burst) - 1:\n",
    "        #     continue\n",
    "        fig, axes = get_axes()\n",
    "        ax = axes[0]\n",
    "\n",
    "        # Plot length\n",
    "        length_xs, length_ys = zip(*burst)\n",
    "        length_xs = np.asarray(length_xs)\n",
    "        length_xs = length_xs * 1e3\n",
    "        ax.plot(\n",
    "            length_xs,\n",
    "            length_ys,\n",
    "            label=\"queue length\",\n",
    "            drawstyle=\"steps-post\",\n",
    "            linewidth=LINESIZE,\n",
    "            alpha=0.8,\n",
    "        )\n",
    "        max_x = length_xs[-1]\n",
    "        max_y = max(length_ys)\n",
    "\n",
    "        # If there are marks, plot them..\n",
    "        if burst_idx < len(marks_by_burst) and marks_by_burst[burst_idx]:\n",
    "            mark_xs, _ = zip(*marks_by_burst[burst_idx])\n",
    "            mark_xs = np.asarray(mark_xs)\n",
    "            mark_xs = mark_xs * 1e3\n",
    "            mark_ys = [marking_threshold_packets] * len(mark_xs)\n",
    "            ax.plot(\n",
    "                mark_xs,\n",
    "                mark_ys,\n",
    "                \"x\",\n",
    "                color=\"orange\",\n",
    "                label=\"ECN marks\",\n",
    "                linewidth=LINESIZE,\n",
    "                alpha=0.8,\n",
    "            )\n",
    "            max_x = max(max_x, mark_xs[-1])\n",
    "\n",
    "        # If there are drops, plot them..\n",
    "        if burst_idx < len(drops_by_burst) and drops_by_burst[burst_idx]:\n",
    "            drop_xs, _ = zip(*drops_by_burst[burst_idx])\n",
    "            drop_xs = np.asarray(drop_xs)\n",
    "            drop_xs = drop_xs * 1e3\n",
    "            drop_ys = [capacity_packets] * len(drop_xs)\n",
    "            ax.plot(\n",
    "                drop_xs,\n",
    "                drop_ys,\n",
    "                \"x\",\n",
    "                color=\"red\",\n",
    "                label=\"drops\",\n",
    "                linewidth=LINESIZE,\n",
    "                alpha=0.8,\n",
    "            )\n",
    "            max_x = max(max_x, drop_xs[-1])\n",
    "\n",
    "        # Draw a line at the marking threshold\n",
    "        ax.plot(\n",
    "            [0, max_x],\n",
    "            [marking_threshold_packets] * 2,\n",
    "            label=\"ECN threshold\",\n",
    "            color=\"orange\",\n",
    "            linestyle=\"dashed\",\n",
    "            linewidth=LINESIZE,\n",
    "            alpha=0.8,\n",
    "        )\n",
    "\n",
    "        # For readability, only draw a line at the capacity if the max y is at least half the capacity.\n",
    "        if max_y > capacity_packets / 2:\n",
    "            # Draw a line at the queue capacity\n",
    "            ax.plot(\n",
    "                [0, max_x],\n",
    "                [capacity_packets] * 2,\n",
    "                label=\"queue capacity\",\n",
    "                color=\"red\",\n",
    "                linestyle=\"dotted\",\n",
    "                linewidth=LINESIZE,\n",
    "                alpha=0.8,\n",
    "            )\n",
    "            max_y = capacity_packets\n",
    "        else:\n",
    "            max_y = capacity_packets / 2\n",
    "\n",
    "        # ax.set_title(f\"{queue_name} Length: Burst {burst_idx + 1} of {num_bursts}\", fontsize=FONTSIZE)\n",
    "        ax.set_xlabel(\"time (ms)\", fontsize=FONTSIZE)\n",
    "        ax.set_ylabel(\"packets\", fontsize=FONTSIZE)\n",
    "        # ax.tick_params(axis='y', labelcolor=blue)\n",
    "        ax.tick_params(axis=\"x\", labelsize=FONTSIZE)\n",
    "        ax.tick_params(axis=\"y\", labelsize=FONTSIZE)\n",
    "        ax.set_xlim(left=-0.01 * max_x, right=1.01 * max_x)\n",
    "        ax.set_ylim(bottom=-0.01 * max_y, top=1.1 * max_y)\n",
    "        ax.legend(fontsize=FONTSIZE, loc=\"upper right\")\n",
    "\n",
    "        with open(\n",
    "            path.join(\n",
    "                graph_dir,\n",
    "                prefix\n",
    "                + \"-\"\n",
    "                + \"_\".join(queue_name.split(\" \")).lower()\n",
    "                + f\"-burst{burst_idx}-length.dat\",\n",
    "            ),\n",
    "            \"w\",\n",
    "            encoding=\"utf-8\",\n",
    "        ) as fil:\n",
    "            fil.write(\"time length\\n\")\n",
    "            fil.write(\"\\n\".join(f\"{x} {y}\" for x, y in burst))\n",
    "\n",
    "        show(fig)\n",
    "        save(\n",
    "            graph_dir,\n",
    "            prefix,\n",
    "            suffix=\"_\".join(queue_name.split(\" \")).lower() + \"_\" + str(burst_idx),\n",
    "        )\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    MARKING_THRESHOLD = CONFIG[\"smallQueueMinThresholdPackets\"]\n",
    "    QUEUE_CAPACITY = CONFIG[\"smallQueueSizePackets\"]\n",
    "    INCAST_Q_METRICS = get_queue_metrics_by_burst(EXP_DIR, \"Incast Queue\", BURST_TIMES)\n",
    "    graph_queue(\n",
    "        \"Incast Queue\",\n",
    "        INCAST_Q_METRICS[\"lengths\"],\n",
    "        INCAST_Q_METRICS[\"marks\"],\n",
    "        INCAST_Q_METRICS[\"drops\"],\n",
    "        MARKING_THRESHOLD,\n",
    "        QUEUE_CAPACITY,\n",
    "        GRAPH_DIR,\n",
    "        EXP,\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f693eb0f-154e-4219-9602-4d14dc7c1928",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def calculate_time_at_or_above_threshold_helper(lengths, thresh, start_sec, end_sec):\n",
    "    # Identify crossover points and above regions points by filtering burst_samples.\n",
    "    above_regions = []\n",
    "    last_length = None\n",
    "    last_cross_up = None\n",
    "    for x, length in lengths:\n",
    "        if length < thresh:\n",
    "            if last_cross_up is not None:\n",
    "                above_regions.append((last_cross_up, x))\n",
    "                last_cross_up = None\n",
    "        elif length >= thresh:\n",
    "            if last_length is None or last_length < thresh:\n",
    "                last_cross_up = x\n",
    "        last_length = length\n",
    "    if last_cross_up is not None:\n",
    "        above_regions.append((last_cross_up, end_sec))\n",
    "\n",
    "    above_sec = sum(\n",
    "        region_end_sec - region_start_sec\n",
    "        for region_start_sec, region_end_sec in above_regions\n",
    "    )\n",
    "    total_sec = end_sec - start_sec\n",
    "    return above_sec, total_sec, above_sec / total_sec * 100\n",
    "\n",
    "\n",
    "def calculate_time_at_or_above_threshold(lengths_by_burst, burst_times, thresh):\n",
    "    return [\n",
    "        calculate_time_at_or_above_threshold_helper(lengths, thresh, start_sec, end_sec)\n",
    "        for burst_idx, (lengths, (start_sec, end_sec)) in enumerate(\n",
    "            zip(lengths_by_burst, burst_times)\n",
    "        )\n",
    "    ]\n",
    "\n",
    "\n",
    "def print_q_above_thresh(lengths_by_burst, burst_times, thresh, label):\n",
    "    num_bursts = len(burst_times)\n",
    "    for burst_idx, (above_sec, _, perc) in enumerate(\n",
    "        calculate_time_at_or_above_threshold(lengths_by_burst, burst_times, thresh)\n",
    "    ):\n",
    "        print(\n",
    "            f\"Burst {burst_idx + 1} of {num_bursts} \"\n",
    "            f\"- Time above {label}: {above_sec * 1e3:.2f} ms ({perc:.2f}%)\"\n",
    "        )\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    print_q_above_thresh(\n",
    "        INCAST_Q_METRICS[\"lengths\"], BURST_TIMES, MARKING_THRESHOLD, \"marking threshold\"\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "91e0ea62-e968-44c0-8775-033b81ddae7d",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "if RUN:\n",
    "    print_q_above_thresh(INCAST_Q_METRICS[\"lengths\"], BURST_TIMES, 1, \"empty\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c65a77e5-9719-4168-beb2-5c8b61be1ea4",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "if RUN:\n",
    "    print_q_above_thresh(\n",
    "        INCAST_Q_METRICS[\"lengths\"], BURST_TIMES, QUEUE_CAPACITY * 0.9, \"90% capacity\"\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ce597b6e-74ef-4963-9935-39603d0e34ad",
   "metadata": {},
   "outputs": [],
   "source": [
    "if RUN:\n",
    "    UPLINK_Q_METRICS = get_queue_metrics_by_burst(EXP_DIR, \"Uplink Queue\", BURST_TIMES)\n",
    "    graph_queue(\n",
    "        \"Uplink Queue\",\n",
    "        UPLINK_Q_METRICS[\"lengths\"],\n",
    "        UPLINK_Q_METRICS[\"marks\"],\n",
    "        UPLINK_Q_METRICS[\"drops\"],\n",
    "        MARKING_THRESHOLD,\n",
    "        QUEUE_CAPACITY,\n",
    "        GRAPH_DIR,\n",
    "        EXP,\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a0be6075-4fbc-4b8a-88a0-d122b3db0d82",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "def parse_flow_times(flow_times_json):\n",
    "    burst_to_sender_to_flow_times = [\n",
    "        {\n",
    "            times[\"id\"]: (times[\"start\"], times[\"firstPacket\"], times[\"end\"], ip)\n",
    "            for ip, times in flows.items()\n",
    "        }\n",
    "        for burst, flows in sorted(flow_times_json.items(), key=lambda p: int(p[0]))\n",
    "    ]\n",
    "    sender_to_flow_times_by_burst = {}\n",
    "    for sender in burst_to_sender_to_flow_times[0].keys():\n",
    "        sender_flow_times_by_burst = []\n",
    "        for burst_idx in range(len(burst_to_sender_to_flow_times)):\n",
    "            sender_flow_times_by_burst.append(\n",
    "                burst_to_sender_to_flow_times[burst_idx][sender]\n",
    "            )\n",
    "        sender_to_flow_times_by_burst[sender] = sender_flow_times_by_burst\n",
    "    return sender_to_flow_times_by_burst\n",
    "\n",
    "\n",
    "def get_sender_to_flow_times_by_burst(exp_dir):\n",
    "    with open(\n",
    "        path.join(exp_dir, \"logs\", \"flow_times.json\"), \"r\", encoding=\"utf-8\"\n",
    "    ) as fil:\n",
    "        return parse_flow_times(json.load(fil))\n",
    "\n",
    "\n",
    "def get_active_flows_by_burst(sender_to_flow_times_by_burst, num_bursts):\n",
    "    active_flows_by_burst = []\n",
    "    for burst_idx in range(num_bursts):\n",
    "        times = [\n",
    "            flow_times_by_burst[burst_idx]\n",
    "            for flow_times_by_burst in sender_to_flow_times_by_burst.values()\n",
    "        ]\n",
    "        starts, _, ends, _ = zip(*times)\n",
    "        serialized = [(start, 1) for start in starts] + [(end, -1) for end in ends]\n",
    "        serialized = sorted(serialized, key=lambda p: p[0])\n",
    "        active = [serialized[0]]\n",
    "        for time, action in serialized[1:]:\n",
    "            active.append((time, active[-1][1] + action))\n",
    "        active_flows_by_burst.append(active)\n",
    "    return active_flows_by_burst\n",
    "\n",
    "\n",
    "def graph_active_flows(active_flows_by_burst, num_bursts, graph_dir, prefix):\n",
    "    for burst_idx in range(num_bursts):\n",
    "        fig, axes = get_axes()\n",
    "        ax = axes[0]\n",
    "\n",
    "        xs, ys = zip(*active_flows_by_burst[burst_idx])\n",
    "        ax.plot(xs, ys, drawstyle=\"steps-post\", alpha=0.8)\n",
    "\n",
    "        # ax.set_title(\n",
    "        #     f\"Active flows over time: Burst {burst_idx + 1} of {num_bursts}\"\n",
    "        # )\n",
    "        ax.set_xlabel(\"time (seconds)\")\n",
    "        ax.set_ylabel(\"active flows\")\n",
    "        ax.set_ylim(bottom=0)\n",
    "\n",
    "        show(fig)\n",
    "        save(graph_dir, prefix, suffix=f\"active_flows_{burst_idx}\")\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    SENDER_TO_FLOW_TIMES_BY_BURST = get_sender_to_flow_times_by_burst(EXP_DIR)\n",
    "    ACTIVE_flowS_BY_BURST = get_active_flows_by_burst(\n",
    "        SENDER_TO_FLOW_TIMES_BY_BURST, NUM_BURSTS\n",
    "    )\n",
    "    graph_active_flows(ACTIVE_flowS_BY_BURST, NUM_BURSTS, GRAPH_DIR, EXP)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c70e9b9c-411e-4461-9869-5d3671dda229",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def graph_cdf_of_flow_duration(\n",
    "    sender_to_flow_times_by_burst, num_bursts, graph_dir, prefix\n",
    "):\n",
    "    for burst_idx in range(num_bursts):\n",
    "        # if burst_idx != num_bursts - 1:\n",
    "        #     continue\n",
    "\n",
    "        fig, axes = get_axes(width=3)\n",
    "        ax = axes[0]\n",
    "\n",
    "        times = [\n",
    "            flow_times_by_burst[burst_idx]\n",
    "            for flow_times_by_burst in sender_to_flow_times_by_burst.values()\n",
    "        ]\n",
    "        durations = [(end - start) * 1e3 for start, _, end, _ in times]\n",
    "        print(f\"min: {min(durations)}, max: {max(durations)}\")\n",
    "\n",
    "        count, bins_count = np.histogram(durations, bins=len(durations))\n",
    "        ax.plot(\n",
    "            bins_count[1:], np.cumsum(count / sum(count)), linewidth=LINESIZE, alpha=0.8\n",
    "        )\n",
    "\n",
    "        # ax.set_title(f\"CDF of flow duration: Burst {burst_idx + 1} of {num_bursts}\", fontsize=FONTSIZE)\n",
    "        ax.set_xlabel(\"FCT (ms)\", fontsize=FONTSIZE)\n",
    "        ax.set_ylabel(\"CDF\", fontsize=FONTSIZE)\n",
    "        ax.tick_params(axis=\"x\", labelsize=FONTSIZE)\n",
    "        ax.tick_params(axis=\"y\", labelsize=FONTSIZE)\n",
    "        ax.set_xticks([5, 10, 15], [5, 10, 15])\n",
    "        ax.set_xlim(left=0)\n",
    "        ax.set_ylim(bottom=0, top=1)\n",
    "\n",
    "        show(fig)\n",
    "        save(graph_dir, prefix, suffix=f\"flow_duration_cdf_{burst_idx}\")\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    graph_cdf_of_flow_duration(\n",
    "        SENDER_TO_FLOW_TIMES_BY_BURST, NUM_BURSTS, GRAPH_DIR, EXP\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4491fc08-1963-4a89-a561-3fdc27561b6b",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def parse_cwnd_line(line):\n",
    "    parts = line.strip().split(\" \")\n",
    "    assert len(parts) == 2\n",
    "    time_sec, cwnd_bytes = parts\n",
    "    time_sec = float(time_sec)\n",
    "    cwnd_bytes = int(cwnd_bytes)\n",
    "    return time_sec, cwnd_bytes\n",
    "\n",
    "\n",
    "def parse_cwnds(flp):\n",
    "    with open(flp, \"r\", encoding=\"utf-8\") as fil:\n",
    "        return [parse_cwnd_line(line) for line in fil if line.strip()[0] != \"#\"]\n",
    "\n",
    "\n",
    "def parse_sender(flp):\n",
    "    return int(path.basename(flp).split(\"_\")[0][6:])\n",
    "\n",
    "\n",
    "def get_sender_to_cwnds_by_burst(\n",
    "    exp_dir, burst_times, sender_to_flow_times_by_burst, suffix=\"cwnd\"\n",
    "):\n",
    "    return {\n",
    "        parse_sender(flp): separate_samples_into_bursts(\n",
    "            # Read all CWND samples for this sender\n",
    "            parse_cwnds(flp),\n",
    "            burst_times,\n",
    "            # Look up the start and end time for this sender\n",
    "            sender_to_flow_times_by_burst[parse_sender(flp)],\n",
    "            filter_on_flow_times=True,\n",
    "            bookend=True,\n",
    "        )\n",
    "        for flp in [\n",
    "            path.join(exp_dir, \"logs\", fln)\n",
    "            for fln in os.listdir(path.join(exp_dir, \"logs\"))\n",
    "            if fln.startswith(\"sender\") and fln.endswith(f\"_{suffix}.log\")\n",
    "        ]\n",
    "    }\n",
    "\n",
    "\n",
    "def graph_sender_cwnd(\n",
    "    sender_to_cwnds_by_burst, num_bursts, graph_dir, prefix, ylabel=\"CWND (bytes)\"\n",
    "):\n",
    "    for burst_idx in range(num_bursts):\n",
    "        fig, axes = get_axes()\n",
    "        ax = axes[0]\n",
    "\n",
    "        # ax.set_title(\n",
    "        #     f\"CWND of active flows: Burst {burst_idx + 1} of {num_bursts}\"\n",
    "        # )\n",
    "        ax.set_xlabel(\"time (seconds)\")\n",
    "        ax.set_ylabel(ylabel)\n",
    "\n",
    "        for sender, bursts in sender_to_cwnds_by_burst.items():\n",
    "            if not bursts[burst_idx]:\n",
    "                continue\n",
    "            xs, ys = zip(*bursts[burst_idx])\n",
    "            ax.plot(xs, ys, label=sender, drawstyle=\"steps-post\", alpha=0.8)\n",
    "\n",
    "        ax.set_ylim(bottom=0)\n",
    "\n",
    "        show(fig)\n",
    "        save(graph_dir, prefix, suffix=f\"cwnd_{burst_idx}\")\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    SENDER_TO_CWNDS_BY_BURST = get_sender_to_cwnds_by_burst(\n",
    "        EXP_DIR, BURST_TIMES, SENDER_TO_FLOW_TIMES_BY_BURST\n",
    "    )\n",
    "    graph_sender_cwnd(SENDER_TO_CWNDS_BY_BURST, NUM_BURSTS, GRAPH_DIR, EXP)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "24e1c1ac-ad3f-4c9d-9d4b-89a53605d8fa",
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_sender_to_inflight_by_burst(\n",
    "    exp_dir, burst_times, sender_to_flow_times_by_burst\n",
    "):\n",
    "    return get_sender_to_cwnds_by_burst(\n",
    "        exp_dir, burst_times, sender_to_flow_times_by_burst, suffix=\"bytes_in_flight\"\n",
    "    )\n",
    "\n",
    "\n",
    "def graph_sender_inflight(sender_to_inflight_by_burst, num_bursts, graph_dir, prefix):\n",
    "    graph_sender_cwnd(\n",
    "        sender_to_inflight_by_burst,\n",
    "        num_bursts,\n",
    "        graph_dir,\n",
    "        prefix,\n",
    "        ylabel=\"in-flight data (bytes)\",\n",
    "    )\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    SENDER_TO_INFLIGHT_BY_BURST = get_sender_to_inflight_by_burst(\n",
    "        EXP_DIR, BURST_TIMES, SENDER_TO_FLOW_TIMES_BY_BURST\n",
    "    )\n",
    "    graph_sender_inflight(SENDER_TO_INFLIGHT_BY_BURST, NUM_BURSTS, GRAPH_DIR, EXP)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5e9d4e81-9f1e-41a6-bf4f-2f2fb660a178",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Inspired by https://stackoverflow.com/questions/10058227/calculating-mean-of-arrays-with-different-lengths\n",
    "def tolerant_metrics(xs, arrays, percentiles):\n",
    "    # Map x value to index. Used to quickly determine where each array starts\n",
    "    # relative to the overall xs.\n",
    "    xs_map = {\n",
    "        # Do not need to do round(x, decimal_places) because xs are already at\n",
    "        # intervals of 1 / interp_delta.\n",
    "        x: idx\n",
    "        for idx, x in enumerate(xs)\n",
    "    }\n",
    "\n",
    "    # Create 2d array to fit the largest array\n",
    "    combined_2d = np.ma.empty((len(xs), len(arrays)))\n",
    "    combined_2d.mask = True\n",
    "\n",
    "    for idx, array in enumerate(arrays):\n",
    "        # Look up this array's start position\n",
    "        # Do not need to do round(x, decimal_places) because xs are already at\n",
    "        # intervals of 1 / interp_delta.\n",
    "        start_idx = xs_map[array[0][0]]\n",
    "\n",
    "        source = array[: (len(combined_2d) - start_idx)]\n",
    "        if len(array) != len(source):\n",
    "            print(f\"Warning: Dropping {len(array) - len(source)} samples!\")\n",
    "\n",
    "        # Verify alignment.\n",
    "        assert xs[start_idx] == source[0][0]\n",
    "        assert xs[start_idx + len(source) - 1] == source[-1][0]\n",
    "\n",
    "        # combined_2d[start_idx : start_idx + len(array), idx] = list(zip(*array))[1]\n",
    "        source = list(zip(*source))[1]\n",
    "        combined_2d[start_idx : start_idx + len(array), idx] = source\n",
    "\n",
    "    return (\n",
    "        combined_2d.mean(axis=-1),\n",
    "        combined_2d.std(axis=-1),\n",
    "        combined_2d.min(axis=-1),\n",
    "        combined_2d.max(axis=-1),\n",
    "        np.nanpercentile(\n",
    "            np.ma.filled(np.ma.masked_where(combined_2d < 0, combined_2d), np.nan),\n",
    "            percentiles,\n",
    "            axis=-1,\n",
    "        ),\n",
    "        combined_2d.sum(axis=-1),\n",
    "    )\n",
    "\n",
    "\n",
    "def step_interp(old_xs, old_ys, new_xs):\n",
    "    # Lengths must be nonzero and agree.\n",
    "    assert len(old_xs) > 0\n",
    "    assert len(old_ys) > 0\n",
    "    assert len(new_xs) > 0\n",
    "    assert len(old_xs) == len(old_ys)\n",
    "    # xs must be strictly non-decreasing.\n",
    "    assert (np.diff(old_xs) >= 0).all(), np.diff(old_xs)\n",
    "    assert (np.diff(new_xs) >= 0).all()\n",
    "    # This is strictly interpolation, not extrapolation.\n",
    "    assert new_xs[0] >= old_xs[0]\n",
    "    assert new_xs[-1] <= old_xs[-1]\n",
    "\n",
    "    new_ys = np.empty(len(new_xs))\n",
    "    # Points to the next value in xs and ys that is past the current x we are\n",
    "    # interpolating.\n",
    "    old_idx = 0\n",
    "    for new_idx, new_x in enumerate(new_xs):\n",
    "        # Move old_idx forward until it is at a position where the next element\n",
    "        # in old_xs is strictly greater than new_x.\n",
    "        #\n",
    "        # old_idx will never grow larger than len(old_xs) - 2\n",
    "        while old_idx < len(old_xs) - 2 and new_x >= old_xs[old_idx + 1]:\n",
    "            old_idx += 1\n",
    "\n",
    "        # If old_idx is immediately before the last element in old_xs, then\n",
    "        # check manually if we need to advance old_idx to the last element in\n",
    "        # old_xs.\n",
    "        if old_idx == len(old_xs) - 2:\n",
    "            if new_x >= old_xs[len(old_xs) - 1]:\n",
    "                old_idx += 1\n",
    "\n",
    "        new_ys[new_idx] = old_ys[old_idx]\n",
    "\n",
    "    assert len(new_xs) == len(new_ys)\n",
    "    return new_ys\n",
    "\n",
    "\n",
    "def interpolate_flows_for_burst(\n",
    "    sender_to_x_by_burst, sender_to_x_by_burst_interp, burst_idx, interp_delta\n",
    "):\n",
    "    # Interpolate each flow at uniform intervals.\n",
    "    for sender, bursts in sender_to_x_by_burst.items():\n",
    "        if bursts[burst_idx]:\n",
    "            assert len(bursts[burst_idx]) > 0\n",
    "            new_xs = get_aligned_xs(\n",
    "                bursts[burst_idx][0][0], bursts[burst_idx][-1][0], interp_delta\n",
    "            )\n",
    "            if len(new_xs) == 0:\n",
    "                new_ys = np.empty(0)\n",
    "            else:\n",
    "                new_ys = step_interp(*zip(*bursts[burst_idx]), new_xs)\n",
    "        else:\n",
    "            new_xs = np.empty(0)\n",
    "            new_ys = np.empty(0)\n",
    "        sender_to_x_by_burst_interp[sender].append(list(zip(new_xs, new_ys)))\n",
    "\n",
    "\n",
    "def get_sender_to_x_by_burst_interp(sender_to_x_by_burst, num_bursts, interp_delta):\n",
    "    sender_to_x_by_burst_interp = collections.defaultdict(list)\n",
    "    for burst_idx in range(num_bursts):\n",
    "        interpolate_flows_for_burst(\n",
    "            sender_to_x_by_burst,\n",
    "            sender_to_x_by_burst_interp,\n",
    "            burst_idx,\n",
    "            interp_delta,\n",
    "        )\n",
    "    for bursts_interp in sender_to_x_by_burst_interp.values():\n",
    "        assert len(bursts_interp) == num_bursts\n",
    "    return sender_to_x_by_burst_interp\n",
    "\n",
    "\n",
    "def get_metrics(\n",
    "    sender_to_x_by_burst_interp,\n",
    "    burst_idx,\n",
    "    interp_delta,\n",
    "    percentiles,\n",
    "):\n",
    "    # Extract the desired burst from each sender.\n",
    "    # Throw away senders that do not have any samples for this burst.\n",
    "    valid_senders = [\n",
    "        bursts[burst_idx]\n",
    "        for bursts in sender_to_x_by_burst_interp.values()\n",
    "        if bursts[burst_idx]\n",
    "    ]\n",
    "    if len(valid_senders) != len(sender_to_x_by_burst_interp):\n",
    "        print(\n",
    "            f\"Warning: Burst {burst_idx} has only \"\n",
    "            f\"{len(valid_senders)}/{len(sender_to_x_by_burst_interp)} \"\n",
    "            \"senders with at least one sample.\"\n",
    "        )\n",
    "    return get_metrics_helper(valid_senders, interp_delta, percentiles)\n",
    "\n",
    "\n",
    "def get_metrics_helper(\n",
    "    senders,\n",
    "    interp_delta,\n",
    "    percentiles,\n",
    "):\n",
    "    # senders is an array of data for burst i, with one element (subarray) for each\n",
    "    # sender:\n",
    "    #     senders = [\n",
    "    #                 [ samples from burst i for sender 0     ],\n",
    "    #                   ...                                    ,\n",
    "    #                 [ samples from burst i for sender n - 1 ]\n",
    "    #               ]\n",
    "\n",
    "    # Determine the overall x-axis range for this burst, across all valid senders.\n",
    "    # print(\"max from senders:\", max(samples[-1][0] for samples in senders))\n",
    "    xs = get_aligned_xs(\n",
    "        min(samples[0][0] for samples in senders),\n",
    "        max(samples[-1][0] for samples in senders),\n",
    "        interp_delta,\n",
    "    )\n",
    "    # print(\"max from aligned xs:\", xs[-1])\n",
    "    # print(\", \".join(str(z) for z in list(zip(*senders[5]))[0]))\n",
    "\n",
    "    # Calculate and verify metrics.\n",
    "    avg_ys, stdev_ys, min_ys, max_ys, percentiles_ys, sum_ys = tolerant_metrics(\n",
    "        xs, senders, percentiles\n",
    "    )\n",
    "    assert len(xs) == len(avg_ys)\n",
    "    assert len(xs) == len(stdev_ys)\n",
    "    assert len(xs) == len(min_ys)\n",
    "    assert len(xs) == len(max_ys)\n",
    "    assert len(xs) == percentiles_ys.shape[1]\n",
    "    assert len(xs) == len(sum_ys)\n",
    "\n",
    "    return xs, avg_ys, stdev_ys, min_ys, max_ys, percentiles_ys, sum_ys\n",
    "\n",
    "\n",
    "def get_metrics_by_burst(\n",
    "    sender_to_x_by_burst_interp, num_bursts, interp_delta, percentiles\n",
    "):\n",
    "    return [\n",
    "        get_metrics(\n",
    "            sender_to_x_by_burst_interp,\n",
    "            burst_idx,\n",
    "            interp_delta,\n",
    "            percentiles,\n",
    "        )\n",
    "        for burst_idx in range(num_bursts)\n",
    "    ]\n",
    "\n",
    "\n",
    "def graph_cwnd_metrics(\n",
    "    cwnd_metrics_by_burst,\n",
    "    num_bursts,\n",
    "    percentiles,\n",
    "    graph_dir,\n",
    "    prefix,\n",
    "    ylabel=\"CWND (bytes)\",\n",
    "    fln=\"cwnd_analysis\",\n",
    "):\n",
    "    for burst_idx in range(num_bursts):\n",
    "        xs, avg_ys, stdev_ys, min_ys, max_ys, percentiles_ys, _ = cwnd_metrics_by_burst[\n",
    "            burst_idx\n",
    "        ]\n",
    "\n",
    "        # Left graph\n",
    "        fig, axes = get_axes()\n",
    "        ax = axes[0]\n",
    "        ax.fill_between(xs, min_ys, max_ys, alpha=0.25, label=\"min/max\")\n",
    "        ax.fill_between(\n",
    "            xs, avg_ys - stdev_ys, avg_ys + stdev_ys, alpha=0.5, label=\"avg +/- stdev\"\n",
    "        )\n",
    "        ax.plot(xs, avg_ys, label=\"avg\", alpha=0.8)\n",
    "        # ax.set_title(\n",
    "        #     f\"CWND of active flows: Burst {burst_idx + 1} of {num_bursts}\"\n",
    "        # )\n",
    "        ax.set_xlabel(\"time (seconds)\")\n",
    "        ax.set_ylabel(ylabel)\n",
    "        ax.set_ylim(bottom=0)\n",
    "        ax.legend()\n",
    "        show(fig)\n",
    "        save(graph_dir, prefix, suffix=f\"{fln}_{burst_idx}_0\")\n",
    "\n",
    "        # Right graph\n",
    "        fig, axes = get_axes()\n",
    "        ax = axes[0]\n",
    "        ax.plot(xs, avg_ys, label=\"avg\", alpha=0.8)\n",
    "        # Bottom of lowest bar is percentiles[0], which is designed to be the min (p0).\n",
    "        for idx in range(1, percentiles_ys.shape[0]):\n",
    "            ax.fill_between(\n",
    "                xs,\n",
    "                percentiles_ys[idx - 1],\n",
    "                percentiles_ys[idx],\n",
    "                alpha=0.5,\n",
    "                label=f\"p{percentiles[idx]}\",\n",
    "            )\n",
    "        # ax.set_title(\n",
    "        #     f\"CWND of active flows: Burst {burst_idx + 1} of {num_bursts}\"\n",
    "        # )\n",
    "        ax.set_xlabel(\"time (seconds)\")\n",
    "        ax.set_ylabel(ylabel)\n",
    "        ax.set_ylim(bottom=0)\n",
    "        ax.legend()\n",
    "        show(fig)\n",
    "        save(graph_dir, prefix, suffix=f\"{fln}_{burst_idx}_1\")\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    INTERP_DELTA = 1e5\n",
    "    PERCENTILES = [0, 25, 50, 75, 95, 100]\n",
    "    SENDER_TO_CWNDS_BY_BURST_INTERP = get_sender_to_x_by_burst_interp(\n",
    "        SENDER_TO_CWNDS_BY_BURST, NUM_BURSTS, INTERP_DELTA\n",
    "    )\n",
    "    CWND_METRICS_BY_BURST = get_metrics_by_burst(\n",
    "        SENDER_TO_CWNDS_BY_BURST_INTERP, NUM_BURSTS, INTERP_DELTA, PERCENTILES\n",
    "    )\n",
    "    graph_cwnd_metrics(\n",
    "        CWND_METRICS_BY_BURST,\n",
    "        NUM_BURSTS,\n",
    "        PERCENTILES,\n",
    "        GRAPH_DIR,\n",
    "        EXP,\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "56728086-a728-49f1-ac33-c9016cef318e",
   "metadata": {},
   "outputs": [],
   "source": [
    "def graph_inflight_metrics(\n",
    "    inflight_metrics_by_burst, num_bursts, percentiles, graph_dir, prefix\n",
    "):\n",
    "    graph_cwnd_metrics(\n",
    "        inflight_metrics_by_burst,\n",
    "        num_bursts,\n",
    "        percentiles,\n",
    "        graph_dir,\n",
    "        prefix,\n",
    "        ylabel=\"in-flight data (bytes)\",\n",
    "        fln=\"inflight_analysis\",\n",
    "    )\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    SENDER_TO_INFLIGHT_BY_BURST_INTERP = get_sender_to_x_by_burst_interp(\n",
    "        SENDER_TO_INFLIGHT_BY_BURST, NUM_BURSTS, INTERP_DELTA\n",
    "    )\n",
    "    INFLIGHT_METRICS_BY_BURST = get_metrics_by_burst(\n",
    "        SENDER_TO_INFLIGHT_BY_BURST_INTERP, NUM_BURSTS, INTERP_DELTA, PERCENTILES\n",
    "    )\n",
    "    graph_inflight_metrics(\n",
    "        INFLIGHT_METRICS_BY_BURST,\n",
    "        NUM_BURSTS,\n",
    "        PERCENTILES,\n",
    "        GRAPH_DIR,\n",
    "        EXP,\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3f19dd8b-4419-42fd-8d99-e372161a95c2",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def calculate_average_queue_length(\n",
    "    lengths_by_burst, interp_delta, bandwidth_bps, bytes_per_packet\n",
    "):\n",
    "    avg_q_length_by_burst = []\n",
    "    for lengths in lengths_by_burst:\n",
    "        old_xs, old_ys = zip(*lengths)\n",
    "        new_xs = get_aligned_xs(old_xs[0], old_xs[-1], interp_delta)\n",
    "        new_ys = step_interp(old_xs, old_ys, new_xs)\n",
    "        avg_q_packets = new_ys.mean()\n",
    "        avg_q_bytes = avg_q_packets * bytes_per_packet\n",
    "        avg_q_us = avg_q_bytes / (bandwidth_bps / 8) * 1e6\n",
    "        avg_q_length_by_burst.append((avg_q_packets, avg_q_bytes, avg_q_us))\n",
    "    return avg_q_length_by_burst\n",
    "\n",
    "\n",
    "def print_avg_q_length(\n",
    "    lengths_by_burst, num_bursts, interp_delta, bandwidth_bps, bytes_per_packet\n",
    "):\n",
    "    for burst_idx, (\n",
    "        avg_q_packets,\n",
    "        avg_q_bytes,\n",
    "        avg_q_us,\n",
    "    ) in enumerate(\n",
    "        calculate_average_queue_length(\n",
    "            lengths_by_burst, interp_delta, bandwidth_bps, bytes_per_packet\n",
    "        )\n",
    "    ):\n",
    "        print(\n",
    "            f\"Burst {burst_idx + 1} of {num_bursts} - \"\n",
    "            f\"Average queue length: {avg_q_packets:.2f} packets, \"\n",
    "            f\"{avg_q_bytes:.2f} bytes, {avg_q_us:.2f} us\"\n",
    "        )\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    BYTES_PER_PACKET = 1500\n",
    "    BANDWIDTH_BITSPS = CONFIG[\"smallLinkBandwidthMbps\"] * 1e6\n",
    "    print_avg_q_length(\n",
    "        INCAST_Q_METRICS[\"lengths\"],\n",
    "        NUM_BURSTS,\n",
    "        INTERP_DELTA,\n",
    "        BANDWIDTH_BITSPS,\n",
    "        BYTES_PER_PACKET,\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d2817c7b-0ee0-4ce1-9282-dd6c1b300342",
   "metadata": {},
   "outputs": [],
   "source": [
    "def graph_estimated_queue_ingress_rate(\n",
    "    lengths_by_burst,\n",
    "    bandwidth_bps,\n",
    "    bytes_per_packet,\n",
    "    interp_delta,\n",
    "    graph_dir,\n",
    "    prefix,\n",
    "):\n",
    "    for burst_idx, lengths in enumerate(lengths_by_burst):\n",
    "        fig, axes = get_axes()\n",
    "        ax = axes[0]\n",
    "        old_xs, old_ys = zip(*lengths)\n",
    "        new_xs = get_aligned_xs(old_xs[0], old_xs[-1], interp_delta)\n",
    "        new_ys = step_interp(old_xs, old_ys, new_xs)\n",
    "        new_ys *= 8 * bytes_per_packet\n",
    "\n",
    "        # ax.set_title(\n",
    "        #     f\"Estimated queue ingress rate: Burst {burst_idx + 1} of {num_bursts}\"\n",
    "        # )\n",
    "        ax.set_xlabel(\"time (seconds)\")\n",
    "        ax.set_ylabel(\"ingress rate (Gbps)\")\n",
    "\n",
    "        dydxs = np.gradient(new_ys, new_xs)\n",
    "        # If queue is not empty, then it is by definition draining at the\n",
    "        # bandwidth. So if y is not 0, then add back the bandwidth to the\n",
    "        # gradient to calculate the true ingress rate instead of the net rate.\n",
    "        dydxs = np.array(\n",
    "            [\n",
    "                dydx if y == 0 else (dydx + bandwidth_bps)\n",
    "                for dydx, y in zip(dydxs, new_ys)\n",
    "            ]\n",
    "        )\n",
    "        dydxs /= 1e9\n",
    "\n",
    "        ax.plot(new_xs, dydxs, alpha=0.8)\n",
    "        ax.set_ylim(bottom=min(0, min(dydxs) * 1.1))\n",
    "\n",
    "        show(fig)\n",
    "        save(graph_dir, prefix, suffix=f\"queue_ingress_rate_{burst_idx}\")\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    graph_estimated_queue_ingress_rate(\n",
    "        INCAST_Q_METRICS[\"lengths\"],\n",
    "        BANDWIDTH_BITSPS,\n",
    "        BYTES_PER_PACKET,\n",
    "        INTERP_DELTA,\n",
    "        GRAPH_DIR,\n",
    "        EXP,\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dfc479fb-5120-48c9-a132-6e3f6f4c3ef5",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "def get_cwnd_metrics_across_bursts(\n",
    "    sender_to_cwnds_by_burst_interp, num_bursts, interp_delta, percentiles\n",
    "):\n",
    "    # We always ignore the first burst, since it is different than the others\n",
    "    # due to slow start.\n",
    "    if num_bursts == 1:\n",
    "        print(\n",
    "            \"Error: No results because we ignore the frst burst, but there is only one burst!\"\n",
    "        )\n",
    "        return\n",
    "\n",
    "    # Flatten all senders and bursts.\n",
    "    to_avg = []\n",
    "    # Throw away the first burst, because it always looks different.\n",
    "    for burst_idx in range(1, num_bursts):\n",
    "        # Find the earliest start time for a flow in this burst.\n",
    "        for bursts in sender_to_cwnds_by_burst_interp.values():\n",
    "            # Throw away bursts with no samples.\n",
    "            if not bursts[burst_idx]:\n",
    "                continue\n",
    "            to_avg.append(bursts[burst_idx])\n",
    "\n",
    "    xs, avg_ys, stdev_ys, min_ys, max_ys, percentiles_ys, _ = get_metrics_helper(\n",
    "        to_avg, interp_delta, percentiles\n",
    "    )\n",
    "\n",
    "    # To calculate the average total CWND across bursts, first we need to sum within each burst\n",
    "    to_avg = []\n",
    "    # Throw away the first burst, because it always looks different.\n",
    "    for burst_idx in range(1, num_bursts):\n",
    "        to_sum = []\n",
    "        for bursts in sender_to_cwnds_by_burst_interp.values():\n",
    "            # Throw away bursts with no samples.\n",
    "            if not bursts[burst_idx]:\n",
    "                continue\n",
    "            to_sum.append(bursts[burst_idx])\n",
    "        # Sum the ys across the flows in this burst\n",
    "        xs, _, _, _, _, _, sum_ys = get_metrics_helper(\n",
    "            to_sum, interp_delta, percentiles\n",
    "        )\n",
    "        to_avg.append(list(zip(xs, sum_ys)))\n",
    "    # Now average the totals across all bursts\n",
    "    xs, total_ys, _, _, _, _, _ = get_metrics_helper(to_avg, interp_delta, percentiles)\n",
    "\n",
    "    return xs, avg_ys, stdev_ys, min_ys, max_ys, percentiles_ys, total_ys\n",
    "\n",
    "\n",
    "def graph_aggregate_cwnd_across_bursts(\n",
    "    cwnd_metrics_across_bursts,\n",
    "    percentiles,\n",
    "    graph_dir,\n",
    "    prefix,\n",
    "    ylabel=\"CWND\",\n",
    "    fln=\"combined_cwnd_analysis\",\n",
    "):\n",
    "    (\n",
    "        xs,\n",
    "        avg_ys,\n",
    "        _,\n",
    "        _,\n",
    "        _,\n",
    "        percentiles_ys,\n",
    "        totals_ys,\n",
    "    ) = cwnd_metrics_across_bursts\n",
    "    xs = xs * 1e3\n",
    "    avg_ys = avg_ys / 1e3\n",
    "\n",
    "    # print(sum_ys[-10:])\n",
    "\n",
    "    # Left graph\n",
    "    fig, axes = get_axes()\n",
    "    ax = axes[0]\n",
    "    totals_ys = totals_ys / BDP_BYTES\n",
    "    ax.plot(xs, totals_ys, linewidth=LINESIZE, alpha=0.8)\n",
    "    ax.set_xlabel(\"time (ms)\", fontsize=FONTSIZE)\n",
    "    ax.set_ylabel(f\"total {ylabel}\\n(x BDP)\", fontsize=FONTSIZE)\n",
    "    ax.tick_params(axis=\"x\", labelsize=FONTSIZE)\n",
    "    ax.tick_params(axis=\"y\", labelsize=FONTSIZE)\n",
    "    ax.set_xlim(left=-0.01 * xs[-1], right=1.01 * xs[-1])\n",
    "    ax.set_ylim(bottom=-0.01 * max(totals_ys), top=1.1 * max(totals_ys))\n",
    "    show(fig)\n",
    "    save(graph_dir, prefix, suffix=f\"{fln}_0\")\n",
    "\n",
    "    # Right graph\n",
    "    fig, axes = get_axes()\n",
    "    ax = axes[0]\n",
    "    ax.plot(xs, avg_ys, linewidth=LINESIZE, label=\"avg\", alpha=0.8)\n",
    "    max_y = max(avg_ys)\n",
    "    for idx in range(1, percentiles_ys.shape[0]):\n",
    "        prev = percentiles_ys[idx - 1]\n",
    "        prev = prev / 1e3\n",
    "        nxt = percentiles_ys[idx]\n",
    "        nxt = nxt / 1e3\n",
    "        ax.fill_between(\n",
    "            xs,\n",
    "            prev,\n",
    "            nxt,\n",
    "            alpha=0.5,\n",
    "            label=f\"p{percentiles[idx]}\",\n",
    "        )\n",
    "        max_y = max(max_y, *nxt)\n",
    "    # ax.set_title(\"CWND of active flows across all bursts\")\n",
    "    ax.set_xlabel(\"time (ms)\", fontsize=FONTSIZE)\n",
    "    ax.set_ylabel(f\"per-flow {ylabel}\\n(KB)\", fontsize=FONTSIZE)\n",
    "    ax.tick_params(axis=\"x\", labelsize=FONTSIZE)\n",
    "    ax.tick_params(axis=\"y\", labelsize=FONTSIZE)\n",
    "    ax.set_xlim(left=-0.01 * xs[-1], right=1.01 * xs[-1])\n",
    "    ax.set_ylim(bottom=-0.01 * max_y, top=1.1 * max_y)\n",
    "    ax.legend(loc=\"upper center\", fontsize=FONTSIZE, ncols=3)\n",
    "    show(fig)\n",
    "    save(graph_dir, prefix, suffix=\"combined_cwnd_analysis_1\")\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    BDP_BYTES = (\n",
    "        CONFIG[\"smallLinkBandwidthMbps\"] * 1e6 / 8 * 6 * CONFIG[\"delayPerLinkUs\"] / 1e6\n",
    "    )\n",
    "    CWND_METRICS_ACROSS_BURSTS = get_cwnd_metrics_across_bursts(\n",
    "        SENDER_TO_CWNDS_BY_BURST_INTERP, NUM_BURSTS, INTERP_DELTA, PERCENTILES\n",
    "    )\n",
    "    graph_aggregate_cwnd_across_bursts(\n",
    "        CWND_METRICS_ACROSS_BURSTS,\n",
    "        PERCENTILES,\n",
    "        GRAPH_DIR,\n",
    "        EXP,\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ca034adc-ad56-4391-ae52-9c4fbe96841f",
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_inflight_metrics_across_bursts(\n",
    "    sender_to_inflight_by_burst_interp, num_bursts, interp_delta, percentiles\n",
    "):\n",
    "    return get_cwnd_metrics_across_bursts(\n",
    "        sender_to_inflight_by_burst_interp, num_bursts, interp_delta, percentiles\n",
    "    )\n",
    "\n",
    "\n",
    "def graph_aggregate_inflight_across_bursts(\n",
    "    inflight_metrics_across_bursts,\n",
    "    percentiles,\n",
    "    graph_dir,\n",
    "    prefix,\n",
    "):\n",
    "    graph_aggregate_cwnd_across_bursts(\n",
    "        inflight_metrics_across_bursts,\n",
    "        percentiles,\n",
    "        graph_dir,\n",
    "        prefix,\n",
    "        ylabel=\"in-flight data\",\n",
    "        fln=\"combined_inflight_analysis\",\n",
    "    )\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    INFLIGHT_METRICS_ACROSS_BURSTS = get_inflight_metrics_across_bursts(\n",
    "        SENDER_TO_INFLIGHT_BY_BURST_INTERP, NUM_BURSTS, INTERP_DELTA, PERCENTILES\n",
    "    )\n",
    "    graph_aggregate_inflight_across_bursts(\n",
    "        INFLIGHT_METRICS_ACROSS_BURSTS,\n",
    "        PERCENTILES,\n",
    "        GRAPH_DIR,\n",
    "        EXP,\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ac3af6be-9dd4-4905-a35a-70ce61eecba1",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def graph_total_cwnd(\n",
    "    cwnd_metrics_by_burst,\n",
    "    num_bursts,\n",
    "    bdp_bytes,\n",
    "    graph_dir,\n",
    "    prefix,\n",
    "    ylabel=\"CWND\",\n",
    "    fln=\"total_cwnd\",\n",
    "):\n",
    "    for burst_idx in range(num_bursts):\n",
    "        xs, _, _, _, _, _, sum_ys = cwnd_metrics_by_burst[burst_idx]\n",
    "        xs = xs * 1e3\n",
    "\n",
    "        fig, axes = get_axes()\n",
    "        ax = axes[0]\n",
    "\n",
    "        sum_ys_KB = sum_ys / 1e3\n",
    "        ax.plot(xs, sum_ys_KB, label=f\"Total {ylabel}\", alpha=0.8)\n",
    "        max_y = max(sum_ys_KB)\n",
    "\n",
    "        # Draw a line at the BDP\n",
    "        bdp_kbytes = bdp_bytes / 1e3\n",
    "        ax.plot(\n",
    "            [xs[0], xs[-1]],\n",
    "            [bdp_kbytes, bdp_kbytes],\n",
    "            label=\"BDP\",\n",
    "            color=\"orange\",\n",
    "            linestyle=\"dashed\",\n",
    "            alpha=0.8,\n",
    "        )\n",
    "\n",
    "        # ax.set_title(f\"Total CWND in bytes: Burst {burst_idx + 1} of {num_bursts}\")\n",
    "        ax.set_xlabel(\"time (seconds)\", fontsize=FONTSIZE)\n",
    "        ax.set_ylabel(f\"total {ylabel}\\n(KB)\", fontsize=FONTSIZE)\n",
    "        ax.set_xlim(left=-0.01 * xs[-1], right=1.01 * xs[-1])\n",
    "        ax.set_ylim(bottom=-0.01 * max_y, top=1.1 * max_y)\n",
    "        ax.tick_params(axis=\"x\", labelsize=FONTSIZE)\n",
    "        ax.tick_params(axis=\"y\", labelsize=FONTSIZE)\n",
    "        ax.legend()\n",
    "        show(fig)\n",
    "        save(graph_dir, prefix, suffix=f\"{fln}_0\")\n",
    "\n",
    "        fig, axes = get_axes()\n",
    "        ax = axes[0]\n",
    "\n",
    "        ys_bdp = [y / bdp_bytes for y in sum_ys]\n",
    "        ax.plot(\n",
    "            xs,\n",
    "            ys_bdp,\n",
    "            # label=\"Total CWND as a multiple of BDP\",\n",
    "            alpha=0.8,\n",
    "        )\n",
    "        max_y = max(ys_bdp)\n",
    "\n",
    "        # ax.set_title(\n",
    "        #     f\"Total CWND in multiples of BDP: Burst {burst_idx + 1} of {num_bursts}\"\n",
    "        # )\n",
    "        ax.set_xlabel(\"time (ms)\", fontsize=FONTSIZE)\n",
    "        ax.set_ylabel(f\"total {ylabel}\\n(x BDP)\", fontsize=FONTSIZE)\n",
    "        ax.set_xlim(left=-0.01 * xs[-1], right=1.01 * xs[-1])\n",
    "        ax.set_ylim(bottom=-0.01 * max_y, top=1.1 * max_y)\n",
    "        ax.tick_params(axis=\"x\", labelsize=FONTSIZE)\n",
    "        ax.tick_params(axis=\"y\", labelsize=FONTSIZE)\n",
    "        show(fig)\n",
    "        save(graph_dir, prefix, suffix=f\"{fln}_1\")\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    graph_total_cwnd(\n",
    "        CWND_METRICS_BY_BURST,\n",
    "        NUM_BURSTS,\n",
    "        BDP_BYTES,\n",
    "        GRAPH_DIR,\n",
    "        EXP,\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c3ac51d9-fdd5-4dc9-a7ae-5541a85565eb",
   "metadata": {},
   "outputs": [],
   "source": [
    "def graph_total_inflight(\n",
    "    inflight_metrics_by_burst,\n",
    "    num_bursts,\n",
    "    bdp_bytes,\n",
    "    graph_dir,\n",
    "    prefix,\n",
    "):\n",
    "    graph_total_cwnd(\n",
    "        inflight_metrics_by_burst,\n",
    "        num_bursts,\n",
    "        bdp_bytes,\n",
    "        graph_dir,\n",
    "        prefix,\n",
    "        ylabel=\"in-flight data\",\n",
    "        fln=\"inflight\",\n",
    "    )\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    graph_total_inflight(\n",
    "        INFLIGHT_METRICS_BY_BURST,\n",
    "        NUM_BURSTS,\n",
    "        BDP_BYTES,\n",
    "        GRAPH_DIR,\n",
    "        EXP,\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "769c89e7-f7ef-4781-a107-41d52f7958f4",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def graph_cwnd_change_cdf(sender_to_cwnds_by_burst, num_bursts, graph_dir, prefix):\n",
    "    for burst_idx in range(num_bursts):\n",
    "        fig, axes = get_axes(width=5)\n",
    "        ax = axes[0]\n",
    "\n",
    "        cwnd_down = []\n",
    "        cwnd_up = []\n",
    "        for sender_cwnds in sender_to_cwnds_by_burst.values():\n",
    "            if len(sender_cwnds[burst_idx]) < 2:\n",
    "                continue\n",
    "            _, sender_cwnds_burst = zip(*sender_cwnds[burst_idx])\n",
    "            # Compute percent difference\n",
    "            cwnd_changes = np.diff(sender_cwnds_burst) / sender_cwnds_burst[:-1] * 100\n",
    "            # Filter based on whether increase or decrease\n",
    "            cwnd_down.extend(abs(x) for x in cwnd_changes if x < 0)\n",
    "            cwnd_up.extend(x for x in cwnd_changes if x > 0)\n",
    "\n",
    "        # Remove inf values\n",
    "        cwnd_down = np.asarray(cwnd_down)\n",
    "        cwnd_down = cwnd_down[np.logical_not(np.isinf(cwnd_down))]\n",
    "        cwnd_up = np.asarray(cwnd_up)\n",
    "        cwnd_up = cwnd_up[np.logical_not(np.isinf(cwnd_up))]\n",
    "\n",
    "        # Plot CWND decreases\n",
    "        count, bins_count = np.histogram(cwnd_down, bins=len(cwnd_down))\n",
    "        ax.plot(\n",
    "            bins_count[1:],\n",
    "            np.cumsum(count / sum(count)),\n",
    "            alpha=0.8,\n",
    "            label=\"CWND decrease\",\n",
    "            color=\"red\",\n",
    "        )\n",
    "\n",
    "        # Plot CWND increases\n",
    "        count, bins_count = np.histogram(cwnd_up, bins=len(cwnd_up))\n",
    "        ax.plot(\n",
    "            bins_count[1:],\n",
    "            np.cumsum(count / sum(count)),\n",
    "            alpha=0.8,\n",
    "            linestyle=\"dashed\",\n",
    "            label=\"CWND increase\",\n",
    "            color=\"green\",\n",
    "        )\n",
    "\n",
    "        # ax.set_title(f\"CDF of CWND change (%): Burst {burst_idx + 1} of {num_bursts}\")\n",
    "        ax.set_xlabel(\"CWND change (%)\")\n",
    "        ax.set_ylabel(\"CDF\")\n",
    "        ax.set_xlim(left=0)\n",
    "        ax.set_ylim(bottom=0, top=1.01)\n",
    "        ax.legend()\n",
    "\n",
    "        show(fig)\n",
    "        save(graph_dir, prefix, suffix=f\"cwnd_change_cdf_{burst_idx}\")\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    graph_cwnd_change_cdf(SENDER_TO_CWNDS_BY_BURST, NUM_BURSTS, GRAPH_DIR, EXP)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5716d23c-7a5c-41a1-bdcd-b8e7f6d09550",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def parse_congest_line(line):\n",
    "    parts = line.strip().split(\" \")\n",
    "    assert len(parts) == 4\n",
    "    time_sec, acked_ecn_bytes, total_acked_bytes, alpha = parts\n",
    "    time_sec = float(time_sec)\n",
    "    acked_ecn_bytes = int(acked_ecn_bytes)\n",
    "    total_acked_bytes = int(total_acked_bytes)\n",
    "    alpha = float(alpha)\n",
    "    return time_sec, acked_ecn_bytes, total_acked_bytes, alpha\n",
    "\n",
    "\n",
    "def parse_congest(flp):\n",
    "    with open(flp, \"r\", encoding=\"utf-8\") as fil:\n",
    "        return [parse_congest_line(line) for line in fil if line.strip()[0] != \"#\"]\n",
    "\n",
    "\n",
    "def get_sender_to_congest_by_burst(exp_dir, burst_times, sender_to_flow_times_by_burst):\n",
    "    return {\n",
    "        parse_sender(flp): separate_samples_into_bursts(\n",
    "            # Read all congestion estimate samples for this sender\n",
    "            parse_congest(flp),\n",
    "            burst_times,\n",
    "            # Look up the start and end time for this sender\n",
    "            sender_to_flow_times_by_burst[parse_sender(flp)],\n",
    "            filter_on_flow_times=True,\n",
    "            bookend=True,\n",
    "        )\n",
    "        # Look up all congestion estimate log files.\n",
    "        for flp in [\n",
    "            path.join(exp_dir, \"logs\", fln)\n",
    "            for fln in os.listdir(path.join(exp_dir, \"logs\"))\n",
    "            if fln.startswith(\"sender\") and fln.endswith(\"_congest.log\")\n",
    "        ]\n",
    "    }\n",
    "\n",
    "\n",
    "def graph_sender_dctcp_alpha(sender_to_congest_by_burst, num_bursts, graph_dir, prefix):\n",
    "    for burst_idx in range(num_bursts):\n",
    "        fig, axes = get_axes()\n",
    "        ax = axes[0]\n",
    "\n",
    "        # ax.set_title(\n",
    "        #     f\"Alpha of active flows: Burst {burst_idx + 1} of {num_bursts}\"\n",
    "        # )\n",
    "        ax.set_xlabel(\"time (seconds)\")\n",
    "        ax.set_ylabel(\"alpha\")\n",
    "\n",
    "        for sender, bursts in sender_to_congest_by_burst.items():\n",
    "            if not bursts[burst_idx]:\n",
    "                continue\n",
    "            xs, _, _, ys = zip(*bursts[burst_idx])\n",
    "            ax.plot(\n",
    "                xs,\n",
    "                ys,\n",
    "                \"o\",\n",
    "                markersize=1.5,\n",
    "                label=sender,\n",
    "                alpha=0.8,\n",
    "            )\n",
    "\n",
    "        ax.set_ylim(bottom=0)\n",
    "\n",
    "        show(fig)\n",
    "        save(graph_dir, prefix, suffix=f\"dctcp_alpha_{burst_idx}\")\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    SENDER_TO_CONGEST_BY_BURST = get_sender_to_congest_by_burst(\n",
    "        EXP_DIR, BURST_TIMES, SENDER_TO_FLOW_TIMES_BY_BURST\n",
    "    )\n",
    "    graph_sender_dctcp_alpha(SENDER_TO_CONGEST_BY_BURST, NUM_BURSTS, GRAPH_DIR, EXP)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a4025c57-8dac-45ee-b250-968fa5d18ac9",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def parse_rtt_line(line):\n",
    "    parts = line.strip().split(\" \")\n",
    "    assert len(parts) == 2\n",
    "    time_sec, rtt_us = parts\n",
    "    time_sec = float(time_sec)\n",
    "    rtt_us = int(rtt_us)\n",
    "    return time_sec, rtt_us\n",
    "\n",
    "\n",
    "def parse_rtts(flp):\n",
    "    with open(flp, \"r\", encoding=\"utf-8\") as fil:\n",
    "        return [parse_rtt_line(line) for line in fil if line.strip()[0] != \"#\"]\n",
    "\n",
    "\n",
    "def get_sender_to_rtts_by_burst(exp_dir, burst_times, sender_to_flow_times_by_burst):\n",
    "    return {\n",
    "        parse_sender(flp): separate_samples_into_bursts(\n",
    "            # Read all congestion estimate samples for this sender\n",
    "            parse_rtts(flp),\n",
    "            burst_times,\n",
    "            # Look up the start and end time for this sender\n",
    "            sender_to_flow_times_by_burst[parse_sender(flp)],\n",
    "            filter_on_flow_times=True,\n",
    "            bookend=True,\n",
    "        )\n",
    "        # Look up all congestion estimate log files.\n",
    "        for flp in [\n",
    "            path.join(exp_dir, \"logs\", fln)\n",
    "            for fln in os.listdir(path.join(exp_dir, \"logs\"))\n",
    "            if fln.startswith(\"sender\") and fln.endswith(\"_rtt.log\")\n",
    "        ]\n",
    "    }\n",
    "\n",
    "\n",
    "def graph_sender_rtt(sender_to_rtts_by_burst, num_bursts, graph_dir, prefix):\n",
    "    for burst_idx in range(num_bursts):\n",
    "        fig, axes = get_axes()\n",
    "        ax = axes[0]\n",
    "\n",
    "        # ax.set_title(\n",
    "        #     \"Sender-measured RTT of active flows: \"\n",
    "        #     f\"Burst {burst_idx + 1} of {num_bursts}\"\n",
    "        # )\n",
    "        ax.set_xlabel(\"time (seconds)\")\n",
    "        ax.set_ylabel(\"Sender-measured RTT (us)\")\n",
    "\n",
    "        for sender, bursts in sender_to_rtts_by_burst.items():\n",
    "            if not bursts[burst_idx]:\n",
    "                continue\n",
    "            xs, ys = zip(*bursts[burst_idx])\n",
    "            ax.plot(xs, ys, \"o\", markersize=1.5, label=sender, alpha=0.8)\n",
    "\n",
    "        ax.set_ylim(bottom=0)\n",
    "\n",
    "        show(fig)\n",
    "        save(graph_dir, prefix, suffix=f\"rtt_{burst_idx}\")\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    SENDER_TO_RTTS_BY_BURST = get_sender_to_rtts_by_burst(\n",
    "        EXP_DIR, BURST_TIMES, SENDER_TO_FLOW_TIMES_BY_BURST\n",
    "    )\n",
    "    graph_sender_rtt(SENDER_TO_RTTS_BY_BURST, NUM_BURSTS, GRAPH_DIR, EXP)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "271a6ad7-5cf1-4303-bb80-416bdcf66773",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def graph_rtt_cdf(sender_to_rtts_by_burst, num_bursts, graph_dir, prefix):\n",
    "    for burst_idx in range(num_bursts):\n",
    "        fig, axes = get_axes(width=5)\n",
    "        ax = axes[0]\n",
    "\n",
    "        rtt_us = [\n",
    "            x[1]\n",
    "            for sender_rtts in sender_to_rtts_by_burst.values()\n",
    "            for x in sender_rtts[burst_idx]\n",
    "        ]\n",
    "\n",
    "        # Plot CDF of ACK size across all senders\n",
    "        count, bins_count = np.histogram(rtt_us, bins=len(rtt_us))\n",
    "        ax.plot(\n",
    "            bins_count[1:],\n",
    "            np.cumsum(count / sum(count)),\n",
    "            alpha=0.8,\n",
    "            label=\"ACKed MSS\",\n",
    "        )\n",
    "\n",
    "        # ax.set_title(f\"CDF of RTT (us): Burst {burst_idx + 1} of {num_bursts}\")\n",
    "        ax.set_xlabel(\"RTT (us)\")\n",
    "        ax.set_ylabel(\"CDF\")\n",
    "        ax.set_xlim(left=0)\n",
    "        ax.set_ylim(bottom=0, top=1.01)\n",
    "\n",
    "        show(fig)\n",
    "        save(graph_dir, prefix, suffix=f\"rtt_cdf_{burst_idx}\")\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    graph_rtt_cdf(SENDER_TO_RTTS_BY_BURST, NUM_BURSTS, GRAPH_DIR, EXP)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "906707b7-ec32-48d4-9449-26a71f9b04e3",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def graph_acks_per_congest_cdf(\n",
    "    sender_to_congest_by_burst, num_bursts, mss, graph_dir, prefix\n",
    "):\n",
    "    for burst_idx in range(num_bursts):\n",
    "        fig, axes = get_axes(width=5)\n",
    "        ax = axes[0]\n",
    "\n",
    "        acks_per_congest_bytes = [\n",
    "            b[2] / mss\n",
    "            for sender_congests in sender_to_congest_by_burst.values()\n",
    "            for b in sender_congests[burst_idx]\n",
    "        ]\n",
    "\n",
    "        # Plot CDF of ACKs per congest across all senders\n",
    "        count, bins_count = np.histogram(\n",
    "            acks_per_congest_bytes, bins=len(acks_per_congest_bytes)\n",
    "        )\n",
    "        ax.plot(\n",
    "            bins_count[1:],\n",
    "            np.cumsum(count / sum(count)),\n",
    "            alpha=0.8,\n",
    "            label=\"ACKs per congestion estimate\",\n",
    "        )\n",
    "\n",
    "        # ax.set_title(\n",
    "        #     f\"CDF of ACKs per congestion estimate: Burst {burst_idx + 1} of {num_bursts}\"\n",
    "        # )\n",
    "        ax.set_xlabel(\"ACKs per congestion estimate\")\n",
    "        ax.set_ylabel(\"CDF\")\n",
    "        ax.set_xlim(left=0)\n",
    "        ax.set_ylim(bottom=0, top=1.01)\n",
    "\n",
    "        show(fig)\n",
    "        save(graph_dir, prefix, suffix=f\"acks_per_congest_cdf_{burst_idx}\")\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    MSS = 1448\n",
    "    graph_acks_per_congest_cdf(\n",
    "        SENDER_TO_CONGEST_BY_BURST, NUM_BURSTS, MSS, GRAPH_DIR, EXP\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8feda1a2-68bb-4a98-9e00-ff862bb33887",
   "metadata": {
    "lines_to_next_cell": 2
   },
   "outputs": [],
   "source": [
    "def parse_bytes_in_ack_line(line):\n",
    "    parts = line.strip().split(\" \")\n",
    "    assert len(parts) == 6\n",
    "    (\n",
    "        time_sec,\n",
    "        sender_ip,\n",
    "        sender_port,\n",
    "        aggregator_ip,\n",
    "        aggregator_port,\n",
    "        bytes_in_ack,\n",
    "    ) = parts\n",
    "    time_sec = float(time_sec)\n",
    "    sender_port = int(sender_port)\n",
    "    aggregator_port = int(aggregator_port)\n",
    "    bytes_in_ack = int(bytes_in_ack)\n",
    "    return (\n",
    "        time_sec,\n",
    "        sender_ip,\n",
    "        sender_port,\n",
    "        aggregator_ip,\n",
    "        aggregator_port,\n",
    "        bytes_in_ack,\n",
    "    )\n",
    "\n",
    "\n",
    "def parse_bytes_in_ack(flp):\n",
    "    with open(flp, \"r\", encoding=\"utf-8\") as fil:\n",
    "        return [parse_bytes_in_ack_line(line) for line in fil if line.strip()[0] != \"#\"]\n",
    "\n",
    "\n",
    "def get_sender_to_bytes_in_ack_by_burst(exp_dir, burst_times):\n",
    "    bytes_in_ack_flp = [\n",
    "        path.join(exp_dir, \"logs\", fln)\n",
    "        for fln in os.listdir(path.join(exp_dir, \"logs\"))\n",
    "        if fln.startswith(\"aggregator\") and fln.endswith(\"_bytes_in_ack.log\")\n",
    "    ]\n",
    "    assert len(bytes_in_ack_flp) == 1\n",
    "    bytes_in_ack_flp = bytes_in_ack_flp[0]\n",
    "\n",
    "    # Parse from disk. All sender stored together.\n",
    "    bytes_in_ack = parse_bytes_in_ack(bytes_in_ack_flp)\n",
    "\n",
    "    # Separate out the senders.\n",
    "    sender_to_bytes_in_ack = collections.defaultdict(list)\n",
    "    for record in bytes_in_ack:\n",
    "        sender_to_bytes_in_ack[record[1]].append((record[0], record[5]))\n",
    "\n",
    "    return {\n",
    "        sender: separate_samples_into_bursts(\n",
    "            # Read all congestion estimate samples for this sender\n",
    "            bytes_in_ack,\n",
    "            burst_times,\n",
    "            None,\n",
    "            filter_on_flow_times=False,\n",
    "            bookend=False,\n",
    "        )\n",
    "        for sender, bytes_in_ack in sender_to_bytes_in_ack.items()\n",
    "    }\n",
    "\n",
    "\n",
    "def graph_ack_size_cdf(\n",
    "    sender_to_bytes_in_ack_by_burst, num_bursts, mss, graph_dir, prefix\n",
    "):\n",
    "    for burst_idx in range(num_bursts):\n",
    "        fig, axes = get_axes(width=5)\n",
    "        ax = axes[0]\n",
    "\n",
    "        bytes_in_ack = [\n",
    "            b[1] / mss\n",
    "            for sender_bytes_in_ack in sender_to_bytes_in_ack_by_burst.values()\n",
    "            for b in sender_bytes_in_ack[burst_idx]\n",
    "        ]\n",
    "\n",
    "        # Plot CDF of ACKs per congest across all senders\n",
    "        count, bins_count = np.histogram(bytes_in_ack, bins=len(bytes_in_ack))\n",
    "        ax.plot(\n",
    "            bins_count[1:],\n",
    "            np.cumsum(count / sum(count)),\n",
    "            alpha=0.8,\n",
    "            label=\"ACK size\",\n",
    "        )\n",
    "\n",
    "        # ax.set_title(f\"CDF of ACK size: Burst {burst_idx + 1} of {num_bursts}\")\n",
    "        ax.set_xlabel(\"ACK size (MSS)\")\n",
    "        ax.set_ylabel(\"CDF\")\n",
    "        ax.set_xlim(left=0)\n",
    "        ax.set_ylim(bottom=0, top=1.01)\n",
    "\n",
    "        show(fig)\n",
    "        save(graph_dir, prefix, suffix=f\"ack_size_cdf_{burst_idx}\")\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    SENDER_TO_BYTES_IN_ACK_BY_BURST = get_sender_to_bytes_in_ack_by_burst(\n",
    "        EXP_DIR, BURST_TIMES\n",
    "    )\n",
    "    graph_ack_size_cdf(SENDER_TO_BYTES_IN_ACK_BY_BURST, NUM_BURSTS, MSS, GRAPH_DIR, EXP)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9eda966c-360d-43a0-94ea-8b686c204b70",
   "metadata": {},
   "outputs": [],
   "source": [
    "# TODO: Graph throughput over time\n",
    "\n",
    "# Graph throughput at the aggregator\n",
    "\n",
    "# TODO: Add JSON log file mapping sender ID to sender IP\n",
    "\n",
    "\n",
    "# Log file will be similar to the\n",
    "\n",
    "# Strategy:\n",
    "#   Create buckets, sum up the number of bytes arriving in each bucket.\n",
    "\n",
    "\n",
    "def parse_data_bytes_line(line):\n",
    "    parts = line.strip().split(\" \")\n",
    "    assert len(parts) == 6\n",
    "    (\n",
    "        time_sec,\n",
    "        sender_ip,\n",
    "        sender_port,\n",
    "        aggregator_ip,\n",
    "        aggregator_port,\n",
    "        frame_bytes,\n",
    "    ) = parts\n",
    "    time_sec = float(time_sec)\n",
    "    sender_port = int(sender_port)\n",
    "    aggregator_port = int(aggregator_port)\n",
    "    frame_bytes = int(frame_bytes)\n",
    "    return (\n",
    "        time_sec,\n",
    "        sender_ip,\n",
    "        sender_port,\n",
    "        aggregator_ip,\n",
    "        aggregator_port,\n",
    "        frame_bytes,\n",
    "    )\n",
    "\n",
    "\n",
    "def parse_data_bytes(flp):\n",
    "    print(flp)\n",
    "    with open(flp, \"r\", encoding=\"utf-8\") as fil:\n",
    "        return [parse_data_bytes_line(line) for line in fil if line.strip()[0] != \"#\"]\n",
    "\n",
    "\n",
    "def get_sender_to_data_bytes_by_burst(exp_dir, burst_times):\n",
    "    data_bytes_flp = [\n",
    "        path.join(exp_dir, \"logs\", fln)\n",
    "        for fln in os.listdir(path.join(exp_dir, \"logs\"))\n",
    "        if fln.startswith(\"aggregator\") and fln.endswith(\"_bytes_received.log\")\n",
    "    ]\n",
    "    assert len(data_bytes_flp) == 1\n",
    "    data_bytes_flp = data_bytes_flp[0]\n",
    "\n",
    "    # Parse from disk. All sender stored together.\n",
    "    data_bytes = parse_data_bytes(data_bytes_flp)\n",
    "\n",
    "    # Separate out the senders.\n",
    "    sender_to_data_bytes = collections.defaultdict(list)\n",
    "    for record in data_bytes:\n",
    "        sender_to_data_bytes[record[1]].append((record[0], record[5]))\n",
    "\n",
    "    return {\n",
    "        sender: separate_samples_into_bursts(\n",
    "            # Read all congestion estimate samples for this sender\n",
    "            data_bytes,\n",
    "            burst_times,\n",
    "            None,\n",
    "            filter_on_flow_times=False,\n",
    "            bookend=False,\n",
    "        )\n",
    "        for sender, data_bytes in sender_to_data_bytes.items()\n",
    "    }\n",
    "\n",
    "\n",
    "def data_bytes_to_throughput(data_bytes, bucket_sec, rate_bps):\n",
    "    rate_bytes_per_sec = rate_bps / 8\n",
    "\n",
    "    times, byts = zip(*data_bytes)\n",
    "    assert (np.diff(times) >= 0).all()\n",
    "    start_sec = times[0]\n",
    "    end_sec = times[-1]\n",
    "\n",
    "    buckets = [\n",
    "        # Bucket: [start time (inclusive), end time (exclusive), bytes]\n",
    "        [start_sec + i * bucket_sec, start_sec + (i + 1) * bucket_sec, 0]\n",
    "        # Start at -1 to add an extra bucket in case the first packet started\n",
    "        # being transmitted before the start of the first bucket.\n",
    "        for i in range(-1, math.ceil((end_sec - start_sec) / bucket_sec))\n",
    "    ]\n",
    "\n",
    "    bucket_idx = 0\n",
    "    for transmit_end_sec, byts in data_bytes:\n",
    "        # Find the bucket in which this transmission ended.\n",
    "        while transmit_end_sec > buckets[bucket_idx][1]:\n",
    "            bucket_idx += 1\n",
    "\n",
    "        transmit_sec = byts / rate_bytes_per_sec\n",
    "        transmit_start_sec = transmit_end_sec - transmit_sec\n",
    "\n",
    "        bucket_start_sec, _, _ = buckets[bucket_idx]\n",
    "        if transmit_start_sec < bucket_start_sec:\n",
    "            # Make sure that this transmission does not span more than two buckets. The start time must be >= the start of the previous bucket.\n",
    "            assert transmit_start_sec >= buckets[bucket_idx - 1][0]\n",
    "\n",
    "            # Split the bytes between previous and current buckets.\n",
    "            prev_bucket_fraction_sec = (\n",
    "                bucket_start_sec - transmit_start_sec\n",
    "            ) / transmit_sec\n",
    "\n",
    "            buckets[bucket_idx - 1][2] += prev_bucket_fraction_sec * byts\n",
    "            buckets[bucket_idx][2] += (1 - prev_bucket_fraction_sec) * byts\n",
    "        else:\n",
    "            # Assign all bytes to the current bucket.\n",
    "            buckets[bucket_idx][2] += byts\n",
    "\n",
    "    # print(buckets[:10], sep=\"\\n\")\n",
    "    return [\n",
    "        # Pick the midpoint of the bucket as the x-coordinate.\n",
    "        ((end_sec + start_sec) / 2, byts * 8 / bucket_sec)\n",
    "        for start_sec, end_sec, byts in buckets\n",
    "    ]\n",
    "\n",
    "\n",
    "def data_bytes_to_throughput_gradient(data_bytes, rate_bps):\n",
    "    bytes_per_sec = rate_bps / 8\n",
    "    sec_per_byte = 1 / bytes_per_sec\n",
    "\n",
    "    times, byts = zip(*data_bytes)\n",
    "    assert (np.diff(times) >= 0).all()\n",
    "\n",
    "    # Calculate the arrival time of each byte in each packet.\n",
    "    one_byte_at_a_time = []\n",
    "    for transmit_end_sec, byts in data_bytes:\n",
    "        transmit_sec = byts / bytes_per_sec\n",
    "        transmit_start_sec = transmit_end_sec - transmit_sec\n",
    "\n",
    "        # Calculate the arrival time of each byte in this packet.\n",
    "        for i in range(1, byts + 1):\n",
    "            one_byte_at_a_time.append((transmit_start_sec + i * sec_per_byte, 1))\n",
    "\n",
    "    # The x-axis is the byte arrival time. The y-axis is the total bytes\n",
    "    # received until that time, which is, in effect, a measure of distance or\n",
    "    # position. We have a measure of position over time, so we can take the\n",
    "    # derivitive to calculate the rate or throughput. If we take the second\n",
    "    # derivitive, then we get the acceleration, which in this case is the rate\n",
    "    # of change in throughput, which can be interpretted as the congestion\n",
    "    # control algorithm's reactivity.\n",
    "    individual_byte_times, individual_bytes = zip(*one_byte_at_a_time)\n",
    "    cummulative_bytes = np.cumsum(individual_bytes)\n",
    "    rates = np.gradient(cummulative_bytes, individual_byte_times)\n",
    "    # Return: (time, position, rate, acceleration)\n",
    "    return zip(\n",
    "        individual_byte_times,\n",
    "        cummulative_bytes,\n",
    "        rates * 8,\n",
    "        np.gradient(rates, individual_byte_times) * 8,\n",
    "    )\n",
    "\n",
    "\n",
    "def graph_throughput(\n",
    "    sender_to_data_bytes_by_burst,\n",
    "    rate_bps,\n",
    "    num_bursts,\n",
    "    graph_dir,\n",
    "    prefix,\n",
    "    bucket_sec,\n",
    "    merge_senders=False,  # Whether to graph the throughput of individual senders separately or all flows summed together.\n",
    "):\n",
    "    for burst_idx in range(num_bursts):\n",
    "        fig, axes = get_axes()\n",
    "        ax = axes[0]\n",
    "\n",
    "        # ax.set_title(\n",
    "        #     \"Throughput at aggregator: \"\n",
    "        #     f\"Burst {burst_idx + 1} of {num_bursts}\"\n",
    "        # )\n",
    "        ax.set_xlabel(\"time (seconds)\")\n",
    "        ax.set_ylabel(\"throughput (bps)\")\n",
    "\n",
    "        if merge_senders:\n",
    "            merged = []\n",
    "            for sender, bursts in sender_to_data_bytes_by_burst.items():\n",
    "                if not bursts[burst_idx]:\n",
    "                    continue\n",
    "\n",
    "                merged.extend(bursts[burst_idx])\n",
    "            merged = sorted(merged, key=lambda x: x[0])\n",
    "\n",
    "            times, rates = zip(*data_bytes_to_throughput(merged, bucket_sec, rate_bps))\n",
    "            ax.plot(times, rates, alpha=0.8)\n",
    "        else:\n",
    "            for sender, bursts in sender_to_data_bytes_by_burst.items():\n",
    "                if not bursts[burst_idx]:\n",
    "                    continue\n",
    "                # times, _, rates, _ = zip(*data_bytes_to_throughput_gradient(bursts[burst_idx], rate_bps))\n",
    "                times, rates = zip(\n",
    "                    *data_bytes_to_throughput(bursts[burst_idx], bucket_sec, rate_bps)\n",
    "                )\n",
    "                ax.plot(times, rates, label=sender, alpha=0.8)\n",
    "\n",
    "        ax.set_ylim(bottom=0)\n",
    "\n",
    "        # # Draw a line at the downlink bandwidth\n",
    "        # ax.plot(\n",
    "        #     [xs[0], xs[-1]],\n",
    "        #     [marking_threshold_packets] * 2,\n",
    "        #     label=\"Marking threshold\",\n",
    "        #     color=\"orange\",\n",
    "        #     linestyle=\"dashed\",\n",
    "        #     alpha=0.8,\n",
    "        # )\n",
    "\n",
    "        show(fig)\n",
    "        save(graph_dir, prefix, suffix=f\"tput_{burst_idx}\")\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    # Set bucket granularity to 10 us. For reference, it takes about 1 us to receive 1 packet.\n",
    "    TPUT_BUCKET_SEC = 1e-5\n",
    "\n",
    "    SENDER_TO_DATA_BYTES_BY_BURST = get_sender_to_data_bytes_by_burst(\n",
    "        EXP_DIR, BURST_TIMES\n",
    "    )\n",
    "\n",
    "    graph_throughput(\n",
    "        SENDER_TO_DATA_BYTES_BY_BURST,\n",
    "        CONFIG[\"smallLinkBandwidthMbps\"] * 1e6,\n",
    "        NUM_BURSTS,\n",
    "        GRAPH_DIR,\n",
    "        EXP,\n",
    "        TPUT_BUCKET_SEC,\n",
    "        merge_senders=False,\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ec98a4ce-9ae0-4129-9be4-e65a083d9c8f",
   "metadata": {},
   "outputs": [],
   "source": [
    "if RUN:\n",
    "    graph_throughput(\n",
    "        SENDER_TO_DATA_BYTES_BY_BURST,\n",
    "        CONFIG[\"smallLinkBandwidthMbps\"] * 1e6,\n",
    "        NUM_BURSTS,\n",
    "        GRAPH_DIR,\n",
    "        EXP,\n",
    "        TPUT_BUCKET_SEC,\n",
    "        merge_senders=True,\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c1ce7e8a-9a29-4533-929a-a9a5196307b7",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "def get_queue_length_across_bursts(\n",
    "    lengths_by_burst, num_bursts, interp_delta, percentiles\n",
    "):\n",
    "    # We always ignore the first burst, since it is different than the others\n",
    "    # due to slow start.\n",
    "    if num_bursts == 1:\n",
    "        print(\n",
    "            \"Error: No results because we ignore the frst burst, but there is only one burst!\"\n",
    "        )\n",
    "        return\n",
    "\n",
    "    # Interp each burst\n",
    "    lengths_by_burst_interp = []\n",
    "    for burst_idx in range(1, num_bursts):\n",
    "        assert len(lengths_by_burst[burst_idx]) > 0\n",
    "        new_xs = get_aligned_xs(\n",
    "            lengths_by_burst[burst_idx][0][0],\n",
    "            lengths_by_burst[burst_idx][-1][0],\n",
    "            interp_delta,\n",
    "        )\n",
    "        assert len(new_xs) > 0\n",
    "        new_ys = step_interp(*zip(*lengths_by_burst[burst_idx]), new_xs)\n",
    "        new_xs -= new_xs[0]\n",
    "        lengths_by_burst_interp.append(list(zip(new_xs, new_ys)))\n",
    "\n",
    "    # Create a new xs array that covers the longest burst\n",
    "    # for points in lengths_by_burst_interp:\n",
    "    #     print(points[0][0], points[-1][0], len(points))\n",
    "    # end_x = max(points[-1][0] for points in lengths_by_burst_interp)\n",
    "    # print(\"end_x\", end_x)\n",
    "    # print(\"interp_delta\", interp_delta)\n",
    "    # print(\"end_x * interp_delta\", end_x * interp_delta)\n",
    "    # print(\"range end\", math.floor(end_x * interp_delta) + 1)\n",
    "    # Add an extra +1 to account for floating point error\n",
    "    # xs = np.array(\n",
    "    #     [x / interp_delta for x in range(math.floor(end_x * interp_delta) + 1 + 1)]\n",
    "    # )\n",
    "    xs = get_aligned_xs(\n",
    "        0, max(points[-1][0] for points in lengths_by_burst_interp), interp_delta\n",
    "    )\n",
    "\n",
    "    # Calculate across bursts.\n",
    "    avg_ys, stdev_ys, min_ys, max_ys, percentiles_ys, sum_ys = tolerant_metrics(\n",
    "        xs, lengths_by_burst_interp, percentiles\n",
    "    )\n",
    "    assert len(xs) == len(avg_ys)\n",
    "    assert len(xs) == len(stdev_ys)\n",
    "    assert len(xs) == len(min_ys)\n",
    "    assert len(xs) == len(max_ys)\n",
    "    assert len(xs) == percentiles_ys.shape[1]\n",
    "    assert len(xs) == len(sum_ys)\n",
    "    return xs, avg_ys, stdev_ys, min_ys, max_ys, percentiles_ys, sum_ys\n",
    "\n",
    "\n",
    "def graph_queue_across_bursts(\n",
    "    queue_name,\n",
    "    lengths_across_bursts,\n",
    "    marking_threshold_packets,\n",
    "    capacity_packets,\n",
    "    graph_dir,\n",
    "    prefix,\n",
    "):\n",
    "    fig, axes = get_axes()\n",
    "    ax = axes[0]\n",
    "\n",
    "    # Plot lengths\n",
    "    xs, avg_ys, _, _, _, _, _ = lengths_across_bursts\n",
    "    xs = xs * 1e3\n",
    "    max_y = max(avg_ys)\n",
    "    ax.plot(\n",
    "        xs,\n",
    "        avg_ys,\n",
    "        label=\"queue length\",\n",
    "        drawstyle=\"steps-post\",\n",
    "        linewidth=LINESIZE,\n",
    "        alpha=0.8,\n",
    "    )\n",
    "    # Draw a line at the marking threshold\n",
    "    ax.plot(\n",
    "        [xs[0], xs[-1]],\n",
    "        [marking_threshold_packets] * 2,\n",
    "        label=\"ECN threshold\",\n",
    "        color=\"orange\",\n",
    "        linestyle=\"dashed\",\n",
    "        linewidth=LINESIZE,\n",
    "        alpha=0.8,\n",
    "    )\n",
    "    # For readability, only draw a line at the capacity if the max y is at least half the capacity.\n",
    "    if max_y > capacity_packets / 2:\n",
    "        # Draw a line at the queue capacity\n",
    "        ax.plot(\n",
    "            [xs[0], xs[-1]],\n",
    "            [capacity_packets] * 2,\n",
    "            label=\"queue capacity\",\n",
    "            color=\"red\",\n",
    "            linestyle=\"dotted\",\n",
    "            linewidth=LINESIZE,\n",
    "            alpha=0.8,\n",
    "        )\n",
    "        max_y = capacity_packets\n",
    "    else:\n",
    "        max_y = capacity_packets / 2\n",
    "\n",
    "    ax.set_xlabel(\"time (ms)\", fontsize=FONTSIZE)\n",
    "    ax.set_ylabel(\"packets\", fontsize=FONTSIZE)\n",
    "    ax.tick_params(axis=\"x\", labelsize=FONTSIZE)\n",
    "    ax.tick_params(axis=\"y\", labelsize=FONTSIZE)\n",
    "    ax.set_xlim(left=-0.01 * xs[-1], right=1.01 * xs[-1])\n",
    "    ax.set_ylim(bottom=-0.01 * max_y, top=1.1 * max_y)\n",
    "    ax.legend(fontsize=FONTSIZE, loc=\"upper right\", ncols=1)\n",
    "\n",
    "    show(fig)\n",
    "    save(\n",
    "        graph_dir,\n",
    "        prefix,\n",
    "        suffix=\"_\".join(queue_name.split(\" \")).lower() + \"_length_across_bursts\",\n",
    "    )\n",
    "\n",
    "\n",
    "if RUN:\n",
    "    INCAST_Q_length_ACROSS_BURSTS = get_queue_length_across_bursts(\n",
    "        INCAST_Q_METRICS[\"lengths\"], NUM_BURSTS, INTERP_DELTA, PERCENTILES\n",
    "    )\n",
    "    graph_queue_across_bursts(\n",
    "        \"Incast Queue\",\n",
    "        INCAST_Q_length_ACROSS_BURSTS,\n",
    "        MARKING_THRESHOLD,\n",
    "        QUEUE_CAPACITY,\n",
    "        GRAPH_DIR,\n",
    "        EXP,\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e4494d04-56f5-447e-805b-2fa58e259a89",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "def get_all_metrics_for_exp(\n",
    "    exp_dir,\n",
    "    interp_delta=1e5,\n",
    "    percentiles=[0, 25, 50, 75, 95, 100],\n",
    "    bytes_per_packet=1500,\n",
    "    filt=None,\n",
    "    desired=None,\n",
    "):\n",
    "    print(f\"Loading: {exp_dir}\")\n",
    "    config = get_config_json(exp_dir)\n",
    "\n",
    "    # If this experiment does not match the filter, then skip it.\n",
    "    if filt is not None and not filt(config):\n",
    "        return None\n",
    "\n",
    "    num_bursts = config[\"numBursts\"]\n",
    "    burst_times = get_burst_times(exp_dir)\n",
    "    sender_to_flow_times_by_burst = get_sender_to_flow_times_by_burst(exp_dir)\n",
    "    sender_to_cwnds_by_burst = get_sender_to_cwnds_by_burst(\n",
    "        exp_dir, burst_times, sender_to_flow_times_by_burst\n",
    "    )\n",
    "    sender_to_cwnds_by_burst_interp = get_sender_to_x_by_burst_interp(\n",
    "        sender_to_cwnds_by_burst, num_bursts, interp_delta\n",
    "    )\n",
    "    sender_to_inflight_by_burst = get_sender_to_inflight_by_burst(\n",
    "        exp_dir, burst_times, sender_to_flow_times_by_burst\n",
    "    )\n",
    "    sender_to_inflight_by_burst_interp = get_sender_to_x_by_burst_interp(\n",
    "        sender_to_inflight_by_burst, num_bursts, interp_delta\n",
    "    )\n",
    "    sender_to_rtts_by_burst = get_sender_to_rtts_by_burst(\n",
    "        exp_dir, burst_times, sender_to_flow_times_by_burst\n",
    "    )\n",
    "    incast_q_metrics = get_queue_metrics_by_burst(exp_dir, \"Incast Queue\", burst_times)\n",
    "    uplink_q_metrics = get_queue_metrics_by_burst(exp_dir, \"Uplink Queue\", burst_times)\n",
    "    return {\n",
    "        \"exp_dir\": exp_dir,\n",
    "        \"config\": config,\n",
    "        \"burst_times\": burst_times,\n",
    "        \"avg_tput_by_burst_bps\": [\n",
    "            config[\"bytesPerBurstSender\"]\n",
    "            * 8\n",
    "            * config[\"numBurstSenders\"]\n",
    "            / (end - start)\n",
    "            for start, end in burst_times\n",
    "        ],\n",
    "        \"sender_to_flow_times_by_burst\": sender_to_flow_times_by_burst,\n",
    "        \"active_flows_by_burst\": None\n",
    "        if (desired is not None and \"active_flows_by_burst\" not in desired)\n",
    "        else get_active_flows_by_burst(sender_to_flow_times_by_burst, num_bursts),\n",
    "        \"ideal_sec\": (\n",
    "            # config[\"bytesPerSender\"]\n",
    "            # * config[\"numSenders\"]\n",
    "            config[\"bytesPerBurstSender\"]\n",
    "            * config[\"numBurstSenders\"]\n",
    "            / (config[\"smallLinkBandwidthMbps\"] * 1e6 / 8)\n",
    "            + (6 * config[\"delayPerLinkUs\"] / 1e6)\n",
    "        ),\n",
    "        # lengths, drops, marks -> [burst 1, burst 2, ..]\n",
    "        \"incast_queue_by_burst\": None\n",
    "        if (desired is not None and \"incast_queue_by_burst\" not in desired)\n",
    "        else incast_q_metrics,\n",
    "        \"uplink_queue_by_burst\": None\n",
    "        if (desired is not None and \"uplink_queue_by_burst\" not in desired)\n",
    "        else uplink_q_metrics,\n",
    "        \"incast_queue_across_bursts\": None\n",
    "        if (desired is not None and \"incast_queue_across_bursts\" not in desired)\n",
    "        else get_queue_length_across_bursts(\n",
    "            incast_q_metrics[\"lengths\"], num_bursts, interp_delta, percentiles\n",
    "        ),\n",
    "        \"uplink_queue_across_bursts\": None\n",
    "        if (desired is not None and \"uplink_queue_by_burst\" not in desired)\n",
    "        else get_queue_length_across_bursts(\n",
    "            uplink_q_metrics[\"lengths\"], num_bursts, interp_delta, percentiles\n",
    "        ),\n",
    "        \"sender_to_cwnds_by_burst\": None\n",
    "        if (desired is not None and \"sender_to_cwnds_by_burst\" not in desired)\n",
    "        else sender_to_cwnds_by_burst,\n",
    "        \"sender_to_cwnds_by_burst_interp\": None\n",
    "        if (desired is not None and \"sender_to_cwnds_by_burst_interp\" not in desired)\n",
    "        else sender_to_cwnds_by_burst_interp,\n",
    "        \"cwnd_metrics_by_burst\": None\n",
    "        if (desired is not None and \"cwnd_metrics_by_burst\" not in desired)\n",
    "        else get_metrics_by_burst(\n",
    "            sender_to_cwnds_by_burst_interp, num_bursts, interp_delta, percentiles\n",
    "        ),\n",
    "        \"cwnd_metrics_across_bursts\": None\n",
    "        if (desired is not None and \"cwnd_metrics_across_bursts\" not in desired)\n",
    "        else get_cwnd_metrics_across_bursts(\n",
    "            sender_to_cwnds_by_burst_interp, num_bursts, interp_delta, percentiles\n",
    "        ),\n",
    "        \"sender_to_inflight_by_burst\": None\n",
    "        if (desired is not None and \"sender_to_inflight_by_burst\" not in desired)\n",
    "        else sender_to_inflight_by_burst,\n",
    "        \"sender_to_inflight_by_burst_interp\": None\n",
    "        if (desired is not None and \"sender_to_inflight_by_burst_interp\" not in desired)\n",
    "        else sender_to_inflight_by_burst_interp,\n",
    "        \"inflight_metrics_by_burst\": None\n",
    "        if (desired is not None and \"inflight_metrics_by_burst\" not in desired)\n",
    "        else get_metrics_by_burst(\n",
    "            sender_to_inflight_by_burst_interp, num_bursts, interp_delta, percentiles\n",
    "        ),\n",
    "        \"inflight_metrics_across_bursts\": None\n",
    "        if (desired is not None and \"inflight_metrics_across_bursts\" not in desired)\n",
    "        else get_inflight_metrics_across_bursts(\n",
    "            sender_to_inflight_by_burst_interp, num_bursts, interp_delta, percentiles\n",
    "        ),\n",
    "        \"sender_to_congest_by_burst\": None\n",
    "        if (desired is not None and \"sender_to_congest_by_burst\" not in desired)\n",
    "        else get_sender_to_congest_by_burst(\n",
    "            exp_dir, burst_times, sender_to_flow_times_by_burst\n",
    "        ),\n",
    "        \"sender_to_bytes_in_ack_by_burst\": None\n",
    "        if (desired is not None and \"sender_to_bytes_in_ack_by_burst\" not in desired)\n",
    "        else get_sender_to_bytes_in_ack_by_burst(exp_dir, burst_times),\n",
    "        \"sender_to_rtts_by_burst\": None\n",
    "        if (desired is not None and \"sender_to_rtts_by_burst\" not in desired)\n",
    "        else sender_to_rtts_by_burst,\n",
    "        # \"sender_to_rtts_by_burst_interp\": get_sender_to_x_by_burst_interp(\n",
    "        #     sender_to_rtts_by_burst, num_bursts, interp_delta\n",
    "        # ),\n",
    "        \"incast_q_above_empty\": calculate_time_at_or_above_threshold(\n",
    "            incast_q_metrics[\"lengths\"],\n",
    "            burst_times,\n",
    "            1,\n",
    "        ),\n",
    "        \"incast_q_above_mark\": calculate_time_at_or_above_threshold(\n",
    "            incast_q_metrics[\"lengths\"],\n",
    "            burst_times,\n",
    "            config[\"smallQueueMinThresholdPackets\"],\n",
    "        ),\n",
    "        \"incast_q_above_90\": calculate_time_at_or_above_threshold(\n",
    "            incast_q_metrics[\"lengths\"],\n",
    "            burst_times,\n",
    "            config[\"smallQueueSizePackets\"] * 0.9,\n",
    "        ),\n",
    "        \"incast_q_avg_length_by_burst\": calculate_average_queue_length(\n",
    "            incast_q_metrics[\"lengths\"],\n",
    "            interp_delta,\n",
    "            config[\"smallLinkBandwidthMbps\"] * 1e6,\n",
    "            bytes_per_packet,\n",
    "        ),\n",
    "        \"uplink_q_avg_length_by_burst\": calculate_average_queue_length(\n",
    "            uplink_q_metrics[\"lengths\"],\n",
    "            interp_delta,\n",
    "            # config[\"largeLinkBandwidthMbps\"] * 1e6,\n",
    "            config[\"largeBurstLinkBandwidthMbps\"] * 1e6,\n",
    "            bytes_per_packet,\n",
    "        ),\n",
    "    }"
   ]
  }
 ],
 "metadata": {
  "jupytext": {
   "formats": "ipynb,py:percent"
  },
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.12"
  },
  "widgets": {
   "application/vnd.jupyter.widget-state+json": {
    "state": {},
    "version_major": 2,
    "version_minor": 0
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}