#!/usr/bin/env python3
"""Mediciones de la semana 11 sobre la topología de dumbbell/.

    python3 bench.py link --rate 50 --delay 40 --loss 0 --buffer 250
    python3 bench.py flow --cc reno --seconds 20 --csv reno.csv
    python3 bench.py bloat --cc cubic --seconds 15
    python3 bench.py pair --cc reno --cc2 cubic --seconds 30
    python3 bench.py pair --cc cubic --parallel2 4 --seconds 30
    python3 bench.py queue

Todo corre dentro de los nodos con docker exec. Las cifras dependen de cada computadora.
"""
import argparse
import json
import re
import statistics
import time

from netlab import RECEIVER, sh, spawn

BASE_PORT = 5201
DATA_PORT = 40001  # puerto local del stream de datos, para no confundirlo con el de control
CWND = re.compile(r"\bcwnd:(\d+)")
SSTHRESH = re.compile(r"\bssthresh:(\d+)")
RTT = re.compile(r"\brtt:([\d.]+)/")


def set_link(rate=50, delay=40, loss=0.0, buffer_kb=250, extra_s1=0, extra_s2=0,
             queue="fifo", ecn=False):
    """Configura el enlace. rate en Mbit/s, delay y extra_* en ms, loss en %, buffer en KiB."""
    # 1: el emisor pide ECN al conectarse. 2: el valor por defecto; lo acepta si se lo piden.
    for node in ("s1", "s2"):
        sh(node, f"sysctl -qw net.ipv4.tcp_ecn={1 if ecn else 2}")
    if queue == "red-ecn":
        # RED marca paquetes con ECN entre min y max, antes de que la cola se llene.
        sh("r1", "tc qdisc del dev eth3 root 2>/dev/null; "
                 f"tc qdisc add dev eth3 root handle 1: tbf rate {rate}mbit burst 15kb "
                 f"limit {buffer_kb}kb && "
                 f"tc qdisc add dev eth3 parent 1:1 handle 10: red limit {buffer_kb}kb "
                 f"min 30kb max 90kb avpkt 1500 burst 30 bandwidth {rate}mbit "
                 "probability 0.1 ecn")
    else:
        sh("r1", "tc qdisc del dev eth3 root 2>/dev/null; "
                 f"tc qdisc add dev eth3 root tbf rate {rate}mbit burst 15kb "
                 f"limit {buffer_kb}kb")
    netem = f"netem delay {delay}ms limit 20000"
    if loss > 0:
        netem += f" loss {loss}%"
    sh("r2", f"tc qdisc replace dev eth2 root {netem}")
    # El retardo extra va en el sentido de los ACK: hacia cada emisor, nunca en su propia
    # salida, donde TCP lo notaría antes de que el paquete salga de la máquina.
    for device, extra in (("eth1", extra_s1), ("eth2", extra_s2)):
        if extra > 0:
            sh("r1", f"tc qdisc replace dev {device} root netem delay {extra}ms limit 20000")
        else:
            sh("r1", f"tc qdisc del dev {device} root 2>/dev/null; true")


def start_servers():
    sh("rx", "pkill iperf3; true", check=False)
    time.sleep(0.2)
    for port in (BASE_PORT, BASE_PORT + 1):
        sh("rx", f"iperf3 -s -D -p {port}")
    time.sleep(0.5)


def stop_servers():
    sh("rx", "pkill iperf3; true", check=False)


def iperf_command(cc, seconds, port=BASE_PORT, cport=DATA_PORT, parallel=1):
    return (f"iperf3 -c {RECEIVER} -p {port} --cport {cport} -t {seconds} -C {cc} "
            f"-P {parallel} -i 1 -J")


def sampler_command(cport, seconds, interval=0.1):
    """Un ciclo dentro del nodo: es la única forma de muestrear cada 100 ms."""
    return (
        'read start rest </proc/uptime; limit=${start%.*}; limit=$((limit+' f"{seconds + 2}" '));'
        ' while :; do read now rest </proc/uptime; [ "${now%.*}" -ge "$limit" ] && break;'
        ' echo "@ $now"; ss -tinH "sport = :' f"{cport}" '";'
        f' sleep {interval}; done')


def parse_ss(text):
    """Convierte la salida del muestreo en filas (t, cwnd, ssthresh o None, rtt en ms)."""
    rows, now, first = [], None, None
    for line in text.splitlines():
        if line.startswith("@ "):
            now = float(line[2:])
            continue
        cwnd, ssthresh, rtt = CWND.search(line), SSTHRESH.search(line), RTT.search(line)
        if cwnd and rtt and now is not None:
            if first is None:
                first = now
            rows.append((round(now - first, 2), int(cwnd.group(1)),
                         int(ssthresh.group(1)) if ssthresh else None, float(rtt.group(1))))
    return rows


def write_csv(path, rows):
    with open(path, "w", encoding="utf-8") as file:
        file.write("t_s,cwnd_segments,ssthresh_segments,rtt_ms\n")
        for time_s, cwnd, ssthresh, rtt in rows:
            file.write(f"{time_s},{cwnd},{'' if ssthresh is None else ssthresh},{rtt}\n")


def summary(report):
    """Saca del JSON de iperf3 lo que nos importa."""
    end = report["end"]
    return {
        "cc": end.get("sender_tcp_congestion", "?"),
        "mbps": end["sum_received"]["bits_per_second"] / 1e6,
        "retrans": end["sum_sent"]["retransmits"],
    }


def run_flow(cc, seconds, node="s1", cport=DATA_PORT, sample=True):
    """Un flujo de iperf3 con su muestreo de ss. Devuelve (resumen, filas)."""
    sampler = spawn(node, sampler_command(cport, seconds)) if sample else None
    time.sleep(0.05)
    output = sh(node, iperf_command(cc, seconds, cport=cport), timeout=seconds + 40)
    rows = []
    if sampler:
        text, _ = sampler.communicate(timeout=30)
        rows = parse_ss(text)
    return summary(json.loads(output)), rows


def run_bloat(cc, seconds):
    """Mide el RTT de un ping antes y durante una transferencia que cruza la misma cola."""
    idle = sh("s2", f"ping -c 10 -i 0.2 {RECEIVER}", timeout=30)
    base = [float(x) for x in re.findall(r"time=([\d.]+)", idle)]
    pinger = spawn("s2", f"sleep 4; ping -c {(seconds - 5) * 5} -i 0.2 {RECEIVER}")
    result, rows = run_flow(cc, seconds)
    text, _ = pinger.communicate(timeout=30)
    during = [float(x) for x in re.findall(r"time=([\d.]+)", text)]
    return {
        "mbps": result["mbps"],
        "retrans": result["retrans"],
        "ping_idle_ms": statistics.median(base),
        "ping_load_ms": statistics.median(during) if during else float("nan"),
        "ping_max_ms": max(during) if during else float("nan"),
        "ping_sent": (seconds - 5) * 5,
        "ping_received": len(during),
        "rows": rows,
    }


def queue_stats():
    """Los contadores de la cola del cuello de botella."""
    return sh("r1", "tc -s qdisc show dev eth3")


def run_pair(cc1, cc2, seconds, parallel2=1):
    """Dos emisores a la vez, s1 y s2. Devuelve los dos resúmenes."""
    first = spawn("s1", iperf_command(cc1, seconds, BASE_PORT, DATA_PORT))
    second = spawn("s2", iperf_command(cc2, seconds, BASE_PORT + 1, DATA_PORT, parallel2))
    out1, _ = first.communicate(timeout=seconds + 40)
    out2, _ = second.communicate(timeout=seconds + 40)
    return summary(json.loads(out1)), summary(json.loads(out2))


def jain(values):
    """Índice de equidad de Jain: (suma x)^2 / (n * suma x^2)."""
    return sum(values) ** 2 / (len(values) * sum(x * x for x in values))


def main():
    parser = argparse.ArgumentParser(description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("what", choices=["link", "flow", "bloat", "pair", "queue"])
    parser.add_argument("--rate", type=int, default=50, help="Mbit/s del cuello de botella")
    parser.add_argument("--delay", type=int, default=40, help="ms de retardo en un sentido")
    parser.add_argument("--loss", type=float, default=0.0, help="%% de pérdida aleatoria")
    parser.add_argument("--buffer", type=int, default=250, help="KiB de cola en el cuello")
    parser.add_argument("--extra-s1", type=int, default=0, help="ms extra de RTT para s1")
    parser.add_argument("--extra-s2", type=int, default=0, help="ms extra de RTT para s2")
    parser.add_argument("--queue", choices=["fifo", "red-ecn"], default="fifo")
    parser.add_argument("--ecn", action="store_true", help="los emisores piden ECN")
    parser.add_argument("--cc", default="cubic", help="algoritmo: reno o cubic")
    parser.add_argument("--cc2", default=None, help="algoritmo del segundo flujo en pair")
    parser.add_argument("--parallel2", type=int, default=1,
                        help="conexiones paralelas del segundo emisor en pair")
    parser.add_argument("--seconds", type=int, default=20)
    parser.add_argument("--csv", default=None, help="archivo para guardar cwnd en el tiempo")
    args = parser.parse_args()

    if args.what == "link":
        set_link(args.rate, args.delay, args.loss, args.buffer, args.extra_s1, args.extra_s2,
                 args.queue, args.ecn)
        print(sh("r1", "tc qdisc show dev eth3"), end="")
        print(sh("r2", "tc qdisc show dev eth2"), end="")
        return
    if args.what == "queue":
        print(queue_stats(), end="")
        return

    start_servers()
    try:
        if args.what == "flow":
            result, rows = run_flow(args.cc, args.seconds)
            print(f"{result['cc']}: {result['mbps']:.1f} Mbit/s, "
                  f"{result['retrans']} retransmisiones")
            if args.csv:
                write_csv(args.csv, rows)
                print(f"{len(rows)} muestras en {args.csv}")
        elif args.what == "bloat":
            found = run_bloat(args.cc, args.seconds)
            print(f"{args.cc}: {found['mbps']:.1f} Mbit/s, {found['retrans']} retransmisiones")
            print(f"ping sin carga: {found['ping_idle_ms']:.1f} ms; con carga: mediana "
                  f"{found['ping_load_ms']:.1f} ms, máximo {found['ping_max_ms']:.1f} ms; "
                  f"recibidos {found['ping_received']} de {found['ping_sent']}")
        else:
            one, two = run_pair(args.cc, args.cc2 or args.cc, args.seconds, args.parallel2)
            print(f"s1 {one['cc']}: {one['mbps']:.1f} Mbit/s, {one['retrans']} retransmisiones")
            print(f"s2 {two['cc']}: {two['mbps']:.1f} Mbit/s, {two['retrans']} retransmisiones")
            print(f"Jain: {jain([one['mbps'], two['mbps']]):.3f}")
    finally:
        stop_servers()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""La curva de CUBIC según el RFC 9438: W(t) = C (t - K)^3 + Wmax, frente a Reno.

Solo evalúa la fórmula con C = 0,4 y beta = 0,7, sin la región amigable con Reno ni la
convergencia rápida. Compara cuánto tarda cada uno en volver a Wmax después de una pérdida.

    python3 cubic_curve.py --wmax 167 --rtt 40
    python3 cubic_curve.py --recovery
"""
import argparse

C = 0.4
BETA = 0.7


def k_seconds(wmax):
    """Tiempo que tarda W(t) en volver a Wmax, con la ventana recién reducida a BETA * Wmax."""
    return (wmax * (1 - BETA) / C) ** (1 / 3)


def window(t, wmax):
    return C * (t - k_seconds(wmax)) ** 3 + wmax


def reno_recovery_seconds(wmax, rtt_ms):
    """Reno baja a Wmax / 2 y sube un paquete por RTT."""
    return (wmax / 2) * rtt_ms / 1000


def packets_for(rate_mbps, rtt_ms, packet=1500):
    return rate_mbps * 1e6 * (rtt_ms / 1000) / (8 * packet)


def recovery_table():
    print(f"{'enlace':>10} {'RTT':>7} {'BDP (paq.)':>11} {'Reno':>12} {'CUBIC':>10}")
    for rate, rtt in ((50, 40), (50, 200), (1000, 40), (1000, 200), (10000, 100)):
        wmax = packets_for(rate, rtt)
        print(f"{rate:>5} Mbps {rtt:>4} ms {wmax:>11.0f} "
              f"{reno_recovery_seconds(wmax, rtt):>10.1f} s {k_seconds(wmax):>8.1f} s")


def main():
    parser = argparse.ArgumentParser(description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--wmax", type=float, default=167)
    parser.add_argument("--rtt", type=float, default=40)
    parser.add_argument("--recovery", action="store_true")
    args = parser.parse_args()
    if args.recovery:
        recovery_table()
        return
    k = k_seconds(args.wmax)
    print(f"Wmax = {args.wmax:.0f} paquetes, K = {k:.2f} s")
    print(f"{'t (s)':>6} {'W cubic':>8} {'W reno':>8}")
    reno_slope = 1000 / args.rtt  # paquetes por segundo
    for tenth in range(0, int(k * 2.2 * 2) + 1):
        t = tenth / 2
        reno = args.wmax / 2 + reno_slope * t
        print(f"{t:>6.1f} {window(t, args.wmax):>8.0f} {reno:>8.0f}")


if __name__ == "__main__":
    main()

"""Ayudas de la semana 11: levantar la topología y correr comandos en sus nodos."""
import os
import subprocess
import time
import unittest

HERE = os.path.dirname(os.path.abspath(__file__))
FOLDER = os.path.join(HERE, "dumbbell")
TOPOLOGY = "dumbbell.clab.yml"
PREFIX = "clab-w11-dumbbell-"
NETWORK = "redes-w11"
RECEIVER = "10.211.4.10"


def deploy():
    done = subprocess.run(["containerlab", "deploy", "-t", TOPOLOGY, "--reconfigure"],
                          cwd=FOLDER, capture_output=True, text=True, timeout=300)
    if done.returncode != 0:
        raise RuntimeError(done.stdout + done.stderr)


def destroy():
    subprocess.run(["containerlab", "destroy", "-t", TOPOLOGY, "--cleanup"],
                   cwd=FOLDER, capture_output=True, text=True, timeout=300)
    subprocess.run(["docker", "network", "rm", NETWORK], capture_output=True, text=True)


def sh(node, command, timeout=120, check=True):
    """Corre un comando de shell en un nodo y devuelve su stdout."""
    done = subprocess.run(["docker", "exec", PREFIX + node, "sh", "-c", command],
                          capture_output=True, text=True, timeout=timeout)
    if check and done.returncode != 0:
        raise RuntimeError(f"{node}: {command}\n{done.stdout}{done.stderr}")
    return done.stdout


def spawn(node, command):
    """Arranca un comando en un nodo sin esperarlo; su stdout queda en un pipe."""
    return subprocess.Popen(["docker", "exec", PREFIX + node, "sh", "-c", command],
                            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)


def wait_until(predicate, seconds=30, step=0.5):
    """Espera a que una condición se cumpla; devuelve si se cumplió."""
    deadline = time.monotonic() + seconds
    while time.monotonic() < deadline:
        if predicate():
            return True
        time.sleep(step)
    return predicate()


def load_metrics():
    """Devuelve lab/metrics.py (o la solución, si existe); salta si sigue con el TODO."""
    import importlib.util
    folder = os.path.join(HERE, "solution")
    if not os.path.isdir(folder):
        folder = os.path.join(HERE, "lab")
    path = os.path.join(folder, "metrics.py")
    with open(path, encoding="utf-8") as file:
        if "TODO" in file.read():
            raise unittest.SkipTest("falta completar lab/metrics.py")
    spec = importlib.util.spec_from_file_location("metrics", path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module

#!/usr/bin/env python3
"""El modelo de Chiu y Jain: dos flujos, un enlace y una señal común de congestión.

En cada paso ambos reciben la misma señal: si x1 + x2 supera la capacidad, bajan; si no,
suben. Se comparan las cuatro combinaciones de aumento y reducción.

    python3 phase.py aimd --x1 10 --x2 70 --steps 12
    python3 phase.py all --png fases.png
"""
import argparse

RULES = {
    # nombre: (aumento, reducción)
    "aimd": (lambda x: x + 5, lambda x: x * 0.5),
    "aiad": (lambda x: x + 5, lambda x: max(x - 20, 0)),
    "mimd": (lambda x: x * 1.25, lambda x: x * 0.5),
    "miad": (lambda x: x * 1.25, lambda x: max(x - 20, 0)),
}


def walk(rule, x1, x2, capacity=100, steps=200):
    """Devuelve la lista de puntos (x1, x2) por los que pasa el sistema."""
    increase, decrease = RULES[rule]
    points = [(x1, x2)]
    for _ in range(steps):
        change = decrease if x1 + x2 > capacity else increase
        x1, x2 = change(x1), change(x2)
        points.append((x1, x2))
    return points


def jain(x1, x2):
    if x1 == 0 and x2 == 0:
        return 0.0
    return (x1 + x2) ** 2 / (2 * (x1 * x1 + x2 * x2))


def plot(path, x1, x2, capacity):
    import matplotlib
    matplotlib.use("Agg")
    import matplotlib.pyplot as plt
    figure, axes = plt.subplots(1, 4, figsize=(14, 3.6), sharex=True, sharey=True)
    for ax, rule in zip(axes, RULES):
        points = walk(rule, x1, x2, capacity, steps=60)
        ax.plot([0, capacity], [capacity, 0], color="gray", label="eficiencia")
        ax.plot([0, capacity], [0, capacity], color="gray", linestyle="--", label="equidad")
        ax.plot(*zip(*points), marker=".", linewidth=0.7)
        ax.set_title(rule.upper())
        ax.set_xlim(0, capacity * 1.3)
        ax.set_ylim(0, capacity * 1.3)
        ax.set_xlabel("x1")
    axes[0].set_ylabel("x2")
    axes[0].legend()
    figure.tight_layout()
    figure.savefig(path, dpi=150)


def main():
    parser = argparse.ArgumentParser(description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("rule", choices=[*RULES, "all"])
    parser.add_argument("--x1", type=float, default=10)
    parser.add_argument("--x2", type=float, default=70)
    parser.add_argument("--capacity", type=float, default=100)
    parser.add_argument("--steps", type=int, default=12)
    parser.add_argument("--png", default=None)
    args = parser.parse_args()
    if args.png:
        plot(args.png, args.x1, args.x2, args.capacity)
        print(f"gráfico en {args.png}")
        return
    if args.rule == "all":
        print(f"{'regla':<6} {'x1':>7} {'x2':>7} {'suma':>7} {'|x1-x2|':>8} {'Jain':>6}")
        for rule in RULES:
            # El último punto: el sistema oscila alrededor de la capacidad, pero x1/x2 se ve bien.
            last = walk(rule, args.x1, args.x2, args.capacity)[-1]
            print(f"{rule:<6} {last[0]:>7.1f} {last[1]:>7.1f} {last[0] + last[1]:>7.1f} "
                  f"{abs(last[0] - last[1]):>8.1f} {jain(*last):>6.3f}")
        return
    print(f"{'paso':>4} {'x1':>7} {'x2':>7} {'suma':>7} {'|x1-x2|':>8} {'Jain':>6}")
    for number, (x1, x2) in enumerate(walk(args.rule, args.x1, args.x2, args.capacity,
                                           args.steps)):
        print(f"{number:>4} {x1:>7.1f} {x2:>7.1f} {x1 + x2:>7.1f} {abs(x1 - x2):>8.1f} "
              f"{jain(x1, x2):>6.3f}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Grafica cwnd en el tiempo a partir de los CSV que deja bench.py flow --csv.

    python3 plot_cwnd.py cwnd.png reno.csv cubic.csv
"""
import csv
import os
import sys

import matplotlib

matplotlib.use("Agg")
import matplotlib.pyplot as plt  # noqa: E402


def read_rows(path):
    with open(path, encoding="utf-8") as file:
        return list(csv.DictReader(file))


def main():
    if len(sys.argv) < 3:
        print(__doc__)
        return 1
    target, *sources = sys.argv[1:]
    figure, (top, bottom) = plt.subplots(2, 1, sharex=True, figsize=(9, 5.5))
    for source in sources:
        rows = read_rows(source)
        if not rows:
            print(f"{source}: no tiene muestras", file=sys.stderr)
            return 1
        times = [float(row["t_s"]) for row in rows]
        label = os.path.splitext(os.path.basename(source))[0]
        top.plot(times, [int(row["cwnd_segments"]) for row in rows], label=label)
        bottom.plot(times, [float(row["rtt_ms"]) for row in rows], label=label)
    top.set_ylabel("cwnd (segmentos)")
    bottom.set_ylabel("RTT suavizado (ms)")
    bottom.set_xlabel("tiempo (s)")
    top.legend()
    figure.tight_layout()
    figure.savefig(target, dpi=150)
    print(f"gráfico en {target}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""cwnd de Reno, una fila por RTT, con las reglas del libro y del RFC 5681.

Es un modelo para hacer la cuenta en papel: cuenta en paquetes y por RTT. No tiene ACK, ni
temporizadores, ni datos; no es un transporte.

    python3 reno_trace.py --ssthresh 8 --rounds 12 --dup-at-cwnd 12
    python3 reno_trace.py --ssthresh 16 --rounds 26 --dup-at-round 9 --timeout-at-round 17
"""
import argparse


def trace(rounds, ssthresh, initial=1, dup_rounds=(), timeout_rounds=(), dup_at_cwnd=None):
    """Devuelve filas (ronda, cwnd, ssthresh, fase, evento al final de esa ronda)."""
    rows = []
    cwnd = initial
    fired = False
    for number in range(1, rounds + 1):
        phase = "slow start" if cwnd < ssthresh else "congestion avoidance"
        dup = number in dup_rounds
        if dup_at_cwnd is not None and cwnd == dup_at_cwnd and not fired:
            dup, fired = True, True
        timeout = number in timeout_rounds
        event = "3 ACK dup." if dup else ("timeout" if timeout else "")
        rows.append((number, cwnd, ssthresh, phase, event))
        if dup or timeout:
            ssthresh = max(cwnd // 2, 2)
            cwnd = ssthresh if dup else 1
        elif cwnd < ssthresh:
            cwnd = min(cwnd * 2, ssthresh)
        else:
            cwnd += 1
    return rows


def main():
    parser = argparse.ArgumentParser(description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--rounds", type=int, default=20)
    parser.add_argument("--ssthresh", type=int, default=8)
    parser.add_argument("--initial", type=int, default=1)
    parser.add_argument("--dup-at-round", type=int, action="append", default=[])
    parser.add_argument("--timeout-at-round", type=int, action="append", default=[])
    parser.add_argument("--dup-at-cwnd", type=int, default=None)
    args = parser.parse_args()
    print(f"{'RTT':>3} {'cwnd':>5} {'ssthresh':>8}  {'fase':<21} evento")
    for number, cwnd, ssthresh, phase, event in trace(
            args.rounds, args.ssthresh, args.initial, args.dup_at_round,
            args.timeout_at_round, args.dup_at_cwnd):
        print(f"{number:>3} {cwnd:>5} {ssthresh:>8}  {phase:<21} {event}".rstrip())


if __name__ == "__main__":
    main()

// Pide al kernel el estado de control de congestión de un socket TCP mientras envía.
//
//   ./tcp_info recv <puerto>
//   ./tcp_info send <ip> <puerto> <reno|cubic> <segundos> [ms entre muestras]
//
// El control de congestión lo hace el kernel: este programa solo lo elige con
// TCP_CONGESTION y lo observa con TCP_INFO, que es de donde saca `ss` sus números.
#include <arpa/inet.h>
#include <errno.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <poll.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <time.h>
#include <unistd.h>

#define BUFFER_SIZE 65536
#define NO_SSTHRESH 0x7fffffffu  // el kernel usa este valor mientras no hubo ninguna pérdida

static double now_ms(void) {
    struct timespec ts;
    if (clock_gettime(CLOCK_MONOTONIC, &ts) < 0) {
        perror("clock_gettime");
        exit(EXIT_FAILURE);
    }
    return (double)ts.tv_sec * 1000.0 + (double)ts.tv_nsec / 1e6;
}

static const char *state_name(uint8_t state) {
    // Los nombres del kernel: enum tcp_ca_state en include/uapi/linux/tcp.h.
    switch (state) {
    case 0: return "open";
    case 1: return "disorder";
    case 2: return "cwr";
    case 3: return "recovery";
    case 4: return "loss";
    default: return "?";
    }
}

static int print_sample(int fd, double elapsed) {
    struct tcp_info info;
    socklen_t size = sizeof info;
    if (getsockopt(fd, IPPROTO_TCP, TCP_INFO, &info, &size) < 0) {
        perror("getsockopt TCP_INFO");
        return -1;
    }
    char ssthresh[16] = "-";
    if (info.tcpi_snd_ssthresh < NO_SSTHRESH) {
        snprintf(ssthresh, sizeof ssthresh, "%u", info.tcpi_snd_ssthresh);
    }
    printf("%8.1f %6u %8s %8.1f %7u %s\n", elapsed, info.tcpi_snd_cwnd, ssthresh,
           info.tcpi_rtt / 1000.0, info.tcpi_total_retrans, state_name(info.tcpi_ca_state));
    return 0;
}

static int run_sender(const char *ip, int port, const char *algorithm, int seconds,
                      int interval_ms) {
    struct sockaddr_in address = {.sin_family = AF_INET, .sin_port = htons((uint16_t)port)};
    if (inet_pton(AF_INET, ip, &address.sin_addr) != 1) {
        fprintf(stderr, "dirección no válida: %s\n", ip);
        return -1;
    }
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd < 0) {
        perror("socket");
        return -1;
    }
    // Se elige antes de conectar, para que también valga durante el slow start.
    if (setsockopt(fd, IPPROTO_TCP, TCP_CONGESTION, algorithm, (socklen_t)strlen(algorithm)) < 0) {
        perror("setsockopt TCP_CONGESTION");
        close(fd);
        return -1;
    }
    if (connect(fd, (struct sockaddr *)&address, sizeof address) < 0) {
        perror("connect");
        close(fd);
        return -1;
    }
    char in_use[16] = "";
    socklen_t in_use_size = sizeof in_use;
    if (getsockopt(fd, IPPROTO_TCP, TCP_CONGESTION, in_use, &in_use_size) < 0) {
        perror("getsockopt TCP_CONGESTION");
        close(fd);
        return -1;
    }
    printf("# algoritmo: %s\n", in_use);
    printf("# %6s %6s %8s %8s %7s %s\n", "t_ms", "cwnd", "ssthresh", "rtt_ms", "retrans", "estado");

    static char buffer[BUFFER_SIZE];
    double start = now_ms();
    double next_sample = start;
    double end = start + seconds * 1000.0;
    int status = 0;
    for (;;) {
        double now = now_ms();
        if (now >= end) {
            break;
        }
        if (now >= next_sample) {
            if (print_sample(fd, now - start) < 0) {
                status = -1;
                break;
            }
            next_sample += interval_ms;
        }
        // Esperamos con poll, sin bloquear en send: la muestra debe salir a su hora
        // aunque la ventana esté llena.
        struct pollfd ready = {.fd = fd, .events = POLLOUT};
        int wait = (int)(next_sample - now) + 1;
        int events = poll(&ready, 1, wait);
        if (events < 0) {
            if (errno == EINTR) {
                continue;
            }
            perror("poll");
            status = -1;
            break;
        }
        if (events == 0) {
            continue;
        }
        if (send(fd, buffer, sizeof buffer, MSG_DONTWAIT) < 0
            && errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
            perror("send");
            status = -1;
            break;
        }
    }
    if (close(fd) < 0) {
        perror("close");
        return -1;
    }
    return status;
}

static int run_receiver(int port) {
    int listener = socket(AF_INET, SOCK_STREAM, 0);
    if (listener < 0) {
        perror("socket");
        return -1;
    }
    int yes = 1;
    if (setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof yes) < 0) {
        perror("setsockopt SO_REUSEADDR");
        close(listener);
        return -1;
    }
    struct sockaddr_in address = {.sin_family = AF_INET, .sin_port = htons((uint16_t)port),
                                  .sin_addr.s_addr = htonl(INADDR_ANY)};
    if (bind(listener, (struct sockaddr *)&address, sizeof address) < 0) {
        perror("bind");
        close(listener);
        return -1;
    }
    if (listen(listener, 1) < 0) {
        perror("listen");
        close(listener);
        return -1;
    }
    int fd = accept(listener, NULL, NULL);
    if (fd < 0) {
        perror("accept");
        close(listener);
        return -1;
    }
    static char buffer[BUFFER_SIZE];
    unsigned long long total = 0;
    int status = 0;
    for (;;) {
        ssize_t got = recv(fd, buffer, sizeof buffer, 0);
        if (got < 0) {
            if (errno == EINTR) {
                continue;
            }
            perror("recv");
            status = -1;
            break;
        }
        if (got == 0) {
            break;
        }
        total += (unsigned long long)got;
    }
    printf("recibidos %llu bytes\n", total);
    if (close(fd) < 0 || close(listener) < 0) {
        perror("close");
        return -1;
    }
    return status;
}

int main(int argc, char *argv[]) {
    if (argc == 3 && strcmp(argv[1], "recv") == 0) {
        return run_receiver(atoi(argv[2])) == 0 ? EXIT_SUCCESS : EXIT_FAILURE;
    }
    if ((argc == 6 || argc == 7) && strcmp(argv[1], "send") == 0) {
        int interval_ms = argc == 7 ? atoi(argv[6]) : 100;
        if (interval_ms <= 0) {
            fprintf(stderr, "el intervalo debe ser positivo\n");
            return EXIT_FAILURE;
        }
        return run_sender(argv[2], atoi(argv[3]), argv[4], atoi(argv[5]), interval_ms) == 0
                   ? EXIT_SUCCESS : EXIT_FAILURE;
    }
    fprintf(stderr, "uso: %s recv <puerto>\n"
                    "     %s send <ip> <puerto> <reno|cubic> <segundos> [ms entre muestras]\n",
            argv[0], argv[0]);
    return EXIT_FAILURE;
}

# Dos emisores, un cuello de botella y un receptor.
#
#   s1 --\
#         r1 ==== cuello de botella ==== r2 --- rx
#   s2 --/
#
# La cola se forma en la salida de r1 hacia r2 (eth3). El retardo y la pérdida se aplican en
# la salida de r2 hacia rx (eth2). Las reglas de tc las pone bench.py link.
name: w11-dumbbell
mgmt:
  network: redes-w11
  ipv4-subnet: 172.90.11.0/26
  ipv6-subnet: 3fff:172:90:11::/64
topology:
  defaults:
    kind: linux
    image: redes-router
    binds: [..:/lab]
  nodes:
    s1:
      exec:
        - ip route del default
        - sysctl -w net.ipv6.conf.all.disable_ipv6=1
        - sysctl -w net.ipv4.tcp_no_metrics_save=1
        - ip addr add 10.211.1.10/24 dev eth1
        - ip route add default via 10.211.1.1
    s2:
      exec:
        - ip route del default
        - sysctl -w net.ipv6.conf.all.disable_ipv6=1
        - sysctl -w net.ipv4.tcp_no_metrics_save=1
        - ip addr add 10.211.2.10/24 dev eth1
        - ip route add default via 10.211.2.1
    r1:
      exec:
        - ip route del default
        - sysctl -w net.ipv6.conf.all.disable_ipv6=1
        - sysctl -w net.ipv4.ip_forward=1
        - ip addr add 10.211.1.1/24 dev eth1
        - ip addr add 10.211.2.1/24 dev eth2
        - ip addr add 10.211.3.1/24 dev eth3
        - ip route add 10.211.4.0/24 via 10.211.3.2
    r2:
      exec:
        - ip route del default
        - sysctl -w net.ipv6.conf.all.disable_ipv6=1
        - sysctl -w net.ipv4.ip_forward=1
        - ip addr add 10.211.3.2/24 dev eth1
        - ip addr add 10.211.4.1/24 dev eth2
        - ip route add 10.211.1.0/24 via 10.211.3.1
        - ip route add 10.211.2.0/24 via 10.211.3.1
    rx:
      exec:
        - ip route del default
        - sysctl -w net.ipv6.conf.all.disable_ipv6=1
        - ip addr add 10.211.4.10/24 dev eth1
        - ip route add default via 10.211.4.1
  links:
    - endpoints: ["s1:eth1", "r1:eth1"]
      mtu: 1500
    - endpoints: ["s2:eth1", "r1:eth2"]
      mtu: 1500
    - endpoints: ["r1:eth3", "r2:eth1"]
      mtu: 1500
    - endpoints: ["r2:eth2", "rx:eth1"]
      mtu: 1500

"""Las afirmaciones de los apuntes, medidas de verdad sobre dumbbell/dumbbell.clab.yml.

Todo mide tendencias con márgenes amplios: las cifras cambian según la computadora y
según lo que esté corriendo en ella.
"""
import csv
import os
import re
import statistics
import tempfile
import time
import unittest

import bench
import netlab
from netlab import RECEIVER, sh


def setUpModule():
    unittest.addModuleCleanup(netlab.destroy)
    netlab.deploy()
    if not netlab.wait_until(lambda: "bytes from" in sh("s1", f"ping -c 1 -W 2 {RECEIVER}", check=False)
                             and "bytes from" in sh("s2", f"ping -c 1 -W 2 {RECEIVER}", check=False),
                             seconds=60):
        raise RuntimeError("la topología no quedó lista")


def ratios(rows):
    """ssthresh nuevo sobre el cwnd más alto desde el evento anterior."""
    found, last, peak = [], None, 0
    for _, cwnd, ssthresh, _ in rows:
        peak = max(peak, cwnd)
        if ssthresh is not None and ssthresh != last:
            found.append(ssthresh / peak)
            last, peak = ssthresh, cwnd
    return found


class Measured(unittest.TestCase):
    def setUp(self):
        bench.start_servers()
        self.addCleanup(bench.stop_servers)

    def test_kernel_offers_reno_and_cubic(self):
        offered = sh("s1", "cat /proc/sys/net/ipv4/tcp_available_congestion_control").split()
        self.assertIn("reno", offered)
        self.assertIn("cubic", offered)

    def test_single_flow_fills_the_link_and_shows_its_decrease(self):
        bench.set_link()
        for cc, low, high in (("reno", 0.30, 0.62), ("cubic", 0.60, 0.85)):
            result, rows = bench.run_flow(cc, 20)
            self.assertEqual(result["cc"], cc)
            self.assertGreater(result["mbps"], 30, cc)
            self.assertLess(result["mbps"], 52, cc)
            self.assertGreater(len(rows), 50, cc)
            self.assertGreater(min(row[3] for row in rows), 38, "el RTT no puede bajar de 40 ms")
            found = ratios(rows)
            self.assertGreater(len(found), 1, f"{cc}: casi no se vieron reducciones")
            self.assertGreater(statistics.median(found), low, f"{cc}: {found}")
            self.assertLess(statistics.median(found), high, f"{cc}: {found}")

    def test_plot_script_makes_a_png(self):
        bench.set_link()
        _, rows = bench.run_flow("cubic", 3)
        with tempfile.TemporaryDirectory() as folder:
            source = os.path.join(folder, "cubic.csv")
            target = os.path.join(folder, "cwnd.png")
            bench.write_csv(source, rows)
            with open(source, encoding="utf-8") as file:
                self.assertEqual(next(csv.reader(file))[1], "cwnd_segments")
            done = os.system(f"python3 {netlab.HERE}/plot_cwnd.py {target} {source} >/dev/null")
            self.assertEqual(done, 0)
            self.assertGreater(os.path.getsize(target), 1000)

    def test_tcp_info_shows_exponential_slow_start(self):
        bench.set_link()
        receiver = netlab.spawn("rx", "/lab/tcp_info recv 6000")
        time.sleep(0.5)
        output = sh("s1", f"/lab/tcp_info send {RECEIVER} 6000 reno 2 10", timeout=30)
        receiver.communicate(timeout=20)
        self.assertIn("algoritmo: reno", output)
        rows = [line.split() for line in output.splitlines() if not line.startswith("#")]
        self.assertEqual(rows[0][1], "10", "Linux empieza con 10 segmentos")
        self.assertEqual(rows[0][2], "-", "todavía no hay ssthresh")
        # Antes del primer RTT no ha vuelto ningún ACK de datos: la ventana sigue en 10.
        early = [int(row[1]) for row in rows if float(row[0]) < 30]
        self.assertTrue(early and all(value == 10 for value in early), early)
        # Una muestra puede caer a mitad de un RTT, así que no se exigen valores exactos. A los
        # 300 ms, unos 7 RTT, sumar uno por RTT daría 17; doblar da cientos.
        later = [int(row[1]) for row in rows if float(row[0]) >= 300]
        self.assertGreater(later[0], 80, "slow start debe crecer de forma exponencial")
        self.assertTrue(any(row[2] != "-" for row in rows), "nunca hubo pérdida")
        self.assertTrue(any(row[5] == "recovery" for row in rows), "nunca entró a recovery")

    def test_random_loss_hurts_and_matches_the_formula(self):
        bench.set_link(loss=1)
        for cc in ("reno", "cubic"):
            result, _ = bench.run_flow(cc, 15, sample=False)
            self.assertLess(result["mbps"], 15, f"{cc}: con 1 % de pérdida ya no llena el enlace")
            if cc == "reno":
                expected = 1.22 * 1448 * 8 / (0.040 * 0.1) / 1e6
                self.assertGreater(result["mbps"], expected / 3)
                self.assertLess(result["mbps"], expected * 3)

    def test_large_buffer_means_large_delay(self):
        bench.set_link(buffer_kb=2000)
        large = bench.run_bloat("reno", 20)
        bench.set_link(buffer_kb=50)
        small = bench.run_bloat("reno", 20)
        for found in (large, small):
            self.assertLess(found["ping_idle_ms"], 60)
            self.assertGreater(found["ping_received"], 0.6 * found["ping_sent"])
        self.assertGreater(large["ping_load_ms"], large["ping_idle_ms"] + 80)
        self.assertLess(small["ping_load_ms"], small["ping_idle_ms"] + 40)
        self.assertGreater(large["ping_load_ms"], small["ping_load_ms"] * 2)

    def test_same_algorithm_shares_the_link(self):
        bench.set_link()
        one, two = bench.run_pair("reno", "reno", 30)
        self.assertGreater(bench.jain([one["mbps"], two["mbps"]]), 0.7, (one, two))
        self.assertGreater(one["mbps"] + two["mbps"], 30)

    def test_short_rtt_beats_long_rtt_with_reno(self):
        bench.set_link(extra_s2=80)
        short, long = bench.run_pair("reno", "reno", 30)
        self.assertGreater(short["mbps"], 1.5 * long["mbps"], (short, long))

    def test_parallel_connections_take_a_larger_share(self):
        bench.set_link()
        one, many = bench.run_pair("cubic", "cubic", 30, parallel2=4)
        self.assertGreater(many["mbps"], 1.3 * one["mbps"], (one, many))

    def counters(self):
        stats = bench.queue_stats()
        marked = re.search(r"marked (\d+) early (\d+) pdrop (\d+)", stats)
        self.assertIsNotNone(marked, stats)
        return [int(value) for value in marked.groups()]

    def test_ecn_marks_where_red_would_drop(self):
        bench.set_link(queue="red-ecn", buffer_kb=1000, ecn=True)
        self.addCleanup(bench.set_link)
        bench.run_flow("cubic", 12, sample=False)
        marked, early, pdrop = self.counters()
        self.assertGreater(marked, 0, "con ECN, RED marca")
        self.assertEqual(early, 0, "con ECN, RED no bota por anticipado")

        bench.set_link(queue="red-ecn", buffer_kb=1000, ecn=False)
        bench.run_flow("cubic", 12, sample=False)
        marked, early, pdrop = self.counters()
        self.assertEqual(marked, 0, "sin ECN, no hay a quién marcar")
        self.assertGreater(early + pdrop, 0, "sin ECN, la misma cola bota")


class Compiled(unittest.TestCase):
    def test_compiles_without_warnings(self):
        with tempfile.TemporaryDirectory() as folder:
            target = os.path.join(folder, "tcp_info")
            source = os.path.join(netlab.HERE, "tcp_info.c")
            done = os.system(f"gcc -Wall -Wextra -Werror -std=c17 -D_DEFAULT_SOURCE -o {target} {source}")
            self.assertEqual(done, 0)


if __name__ == "__main__":
    unittest.main()