/* dcf_sim: un modelo muy simplificado de CSMA/CA, ranura por ranura, en tráfico saturado.
 *
 *   dcf_sim shared <estaciones> <ranuras>    todos se oyen entre sí
 *   dcf_sim hidden <ranuras>                 dos estaciones que no se oyen, y un access point
 *   dcf_sim hidden_rts <ranuras>             las mismas, con RTS/CTS
 *
 * Una estación espera un backoff al azar y transmite. Sin ACK, duplica la ventana. No es 802.11
 * completo: no hay SIFS ni DIFS ni velocidades, y el ACK y el CTS están dentro de los tiempos de
 * abajo. Sirve para comparar casos, no para predecir megabits. Unidad de tiempo: la ranura.
 */
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#define CW_MIN 16
#define CW_MAX 1024
#define FRAME_SLOTS 150 /* datos más ACK */
#define RTS_SLOTS 4     /* RTS más CTS */
#define MAX_STATIONS 128

static uint64_t rng_state = 0x2545F4914F6CDD1Dull;

static uint32_t next_random(void) {
    rng_state ^= rng_state << 13;
    rng_state ^= rng_state >> 7;
    rng_state ^= rng_state << 17;
    return (uint32_t)(rng_state >> 32);
}

static int new_backoff(int window) {
    return (int)(next_random() % (uint32_t)window);
}

static int next_window(int window) {
    return window * 2 > CW_MAX ? CW_MAX : window * 2;
}

/* Todos se oyen: mientras alguien transmite, los demás congelan su backoff. Solo chocan
 * quienes llegan a cero en la misma ranura. */
static int run_shared(int stations, long slots) {
    int backoff[MAX_STATIONS], window[MAX_STATIONS];
    for (int i = 0; i < stations; i++) {
        window[i] = CW_MIN;
        backoff[i] = new_backoff(window[i]);
    }
    long now = 0, delivered = 0, attempts = 0, failures = 0;
    while (now < slots) {
        int talking = 0;
        for (int i = 0; i < stations; i++) {
            talking += backoff[i] == 0;
        }
        if (talking == 0) {
            for (int i = 0; i < stations; i++) {
                backoff[i]--;
            }
            now++;
            continue;
        }
        attempts += talking;
        if (talking == 1) {
            delivered++;
        } else {
            failures += talking;
        }
        for (int i = 0; i < stations; i++) {
            if (backoff[i] == 0) {
                window[i] = talking == 1 ? CW_MIN : next_window(window[i]);
                backoff[i] = new_backoff(window[i]);
            }
        }
        now += FRAME_SLOTS;
    }
    printf("%3d estaciones que se oyen: tiempo útil %5.1f %%; intentos que chocan %5.1f %%\n", stations,
           100.0 * (double)delivered * FRAME_SLOTS / (double)now,
           attempts ? 100.0 * (double)failures / (double)attempts : 0.0);
    return 0;
}

enum state { BACKOFF, HANDSHAKE, DATA, DEFER };

struct station {
    enum state state;
    int window;
    int left;
    int damaged;
};

static void start_backoff(struct station *station) {
    station->state = BACKOFF;
    station->left = new_backoff(station->window);
}

/* Dos estaciones que el access point oye, pero que no se oyen entre ellas: ninguna congela su
 * backoff por la otra. Si dos transmisiones se solapan en el access point, las dos se dañan.
 * Con RTS/CTS se arriesgan solo unas pocas ranuras, y el CTS hace callar a la otra (DEFER). */
static int run_hidden(int use_rts, long slots) {
    struct station stations[2];
    for (int i = 0; i < 2; i++) {
        stations[i].window = CW_MIN;
        stations[i].damaged = 0;
        start_backoff(&stations[i]);
    }
    long delivered = 0, attempts = 0, failures = 0;
    for (long now = 0; now < slots; now++) {
        for (int i = 0; i < 2; i++) {
            struct station *station = &stations[i];
            if (station->state == BACKOFF && station->left == 0) {
                attempts++;
                station->damaged = 0;
                station->state = use_rts ? HANDSHAKE : DATA;
                station->left = use_rts ? RTS_SLOTS : FRAME_SLOTS;
            }
        }
        int on_air[2];
        for (int i = 0; i < 2; i++) {
            on_air[i] = stations[i].state == HANDSHAKE || stations[i].state == DATA;
        }
        if (on_air[0] && on_air[1]) {
            stations[0].damaged = stations[1].damaged = 1;
        }
        for (int i = 0; i < 2; i++) {
            struct station *station = &stations[i];
            struct station *other = &stations[1 - i];
            station->left--;
            if (station->state == BACKOFF || station->left > 0) {
                if (station->state == BACKOFF && station->left < 0) {
                    station->left = 0;
                }
                continue;
            }
            if (station->state == DEFER) {
                start_backoff(station);
            } else if (station->damaged) {
                failures++;
                station->window = next_window(station->window);
                start_backoff(station);
            } else if (station->state == HANDSHAKE) {
                /* CTS: el access point reserva el medio y la otra estación lo oye. */
                station->state = DATA;
                station->left = FRAME_SLOTS;
                other->state = DEFER;
                other->left = FRAME_SLOTS;
            } else {
                delivered++;
                station->window = CW_MIN;
                start_backoff(station);
            }
        }
    }
    printf("2 estaciones ocultas, %s: tiempo útil %5.1f %%; intentos que chocan %5.1f %%\n",
           use_rts ? "con RTS/CTS" : "sin RTS/CTS",
           100.0 * (double)delivered * FRAME_SLOTS / (double)slots,
           attempts ? 100.0 * (double)failures / (double)attempts : 0.0);
    return 0;
}

static long parse_long(const char *text, long low, long high) {
    char *end;
    long value = strtol(text, &end, 10);
    if (*text == '\0' || *end != '\0' || value < low || value > high) {
        fprintf(stderr, "valor fuera de rango: %s\n", text);
        exit(2);
    }
    return value;
}

int main(int argc, char **argv) {
    if (argc == 4 && strcmp(argv[1], "shared") == 0) {
        return run_shared((int)parse_long(argv[2], 1, MAX_STATIONS), parse_long(argv[3], 1, 1000000000));
    }
    if (argc == 3 && strcmp(argv[1], "hidden") == 0) {
        return run_hidden(0, parse_long(argv[2], 1, 1000000000));
    }
    if (argc == 3 && strcmp(argv[1], "hidden_rts") == 0) {
        return run_hidden(1, parse_long(argv[2], 1, 1000000000));
    }
    fprintf(stderr, "uso: %s shared <estaciones> <ranuras>\n     %s hidden|hidden_rts <ranuras>\n",
            argv[0], argv[0]);
    return 2;
}

/* df_ping: un ping con el bit DF prendido y un tamaño exacto. El ping de la imagen de los nodos
 * (busybox) no tiene esa opción.
 *
 *   df_ping <ip> <bytes_de_datos> [intentos]
 *
 * El paquete mide bytes_de_datos + 28 (20 de IP y 8 de ICMP). Sale con 0 si llegó alguna
 * respuesta, con 1 si no. Necesita un socket crudo, así que corre como root en los nodos.
 */
#include <arpa/inet.h>
#include <errno.h>
#include <netinet/in.h>
#include <poll.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>

#define MAX_DATA 9000
#define ICMP_ECHO_REQUEST 8
#define ICMP_ECHO_REPLY 0
#define ICMP_HEADER 8
#define WAIT_MS 800

static uint16_t checksum(const unsigned char *data, size_t size) {
    uint32_t sum = 0;
    for (size_t i = 0; i + 1 < size; i += 2) {
        sum += (uint32_t)(data[i] << 8 | data[i + 1]);
    }
    if (size % 2) {
        sum += (uint32_t)(data[size - 1] << 8);
    }
    while (sum >> 16) {
        sum = (sum & 0xffff) + (sum >> 16);
    }
    return (uint16_t)~sum;
}

int main(int argc, char **argv) {
    if (argc != 3 && argc != 4) {
        fprintf(stderr, "uso: %s <ip> <bytes_de_datos> [intentos]\n", argv[0]);
        return 2;
    }
    struct sockaddr_in remote = {.sin_family = AF_INET};
    if (inet_pton(AF_INET, argv[1], &remote.sin_addr) != 1) {
        fprintf(stderr, "dirección inválida: %s\n", argv[1]);
        return 2;
    }
    long data = strtol(argv[2], NULL, 10);
    long tries = argc == 4 ? strtol(argv[3], NULL, 10) : 3;
    if (data < 0 || data > MAX_DATA || tries < 1 || tries > 100) {
        fprintf(stderr, "valores fuera de rango\n");
        return 2;
    }
    int fd = socket(AF_INET, SOCK_RAW, IPPROTO_ICMP);
    if (fd < 0) {
        perror("socket");
        return 1;
    }
    int discover = IP_PMTUDISC_DO;
    if (setsockopt(fd, IPPROTO_IP, IP_MTU_DISCOVER, &discover, sizeof discover) < 0) {
        perror("setsockopt");
        return 1;
    }
    uint16_t id = (uint16_t)getpid();
    static unsigned char packet[ICMP_HEADER + MAX_DATA];
    size_t size = ICMP_HEADER + (size_t)data;
    long replies = 0;
    for (long attempt = 0; attempt < tries && replies == 0; attempt++) {
        memset(packet, 0x5a, size);
        packet[0] = ICMP_ECHO_REQUEST;
        packet[1] = 0;
        packet[2] = packet[3] = 0;
        packet[4] = (unsigned char)(id >> 8);
        packet[5] = (unsigned char)id;
        packet[6] = (unsigned char)(attempt >> 8);
        packet[7] = (unsigned char)attempt;
        uint16_t sum = checksum(packet, size);
        packet[2] = (unsigned char)(sum >> 8);
        packet[3] = (unsigned char)sum;
        if (sendto(fd, packet, size, 0, (struct sockaddr *)&remote, sizeof remote) < 0) {
            /* EMSGSIZE: este mismo host ya sabe que el paquete no cabe. */
            if (errno == EMSGSIZE) {
                printf("%zu bytes: el host no lo envía (EMSGSIZE)\n", size + 20);
                close(fd);
                return 1;
            }
            perror("sendto");
            close(fd);
            return 1;
        }
        struct pollfd waiting = {.fd = fd, .events = POLLIN};
        while (replies == 0) {
            int ready = poll(&waiting, 1, WAIT_MS);
            if (ready < 0) {
                perror("poll");
                close(fd);
                return 1;
            }
            if (ready == 0) {
                break;
            }
            unsigned char reply[20 + ICMP_HEADER + MAX_DATA];
            ssize_t got = recv(fd, reply, sizeof reply, 0);
            if (got < 0) {
                if (errno == EMSGSIZE) {
                    printf("%zu bytes: llegó un aviso de que no cabe (EMSGSIZE)\n", size + 20);
                    close(fd);
                    return 1;
                }
                perror("recv");
                close(fd);
                return 1;
            }
            size_t header = got > 0 ? (size_t)(reply[0] & 0x0f) * 4 : 0;
            if ((size_t)got >= header + ICMP_HEADER && reply[header] == ICMP_ECHO_REPLY &&
                reply[header + 4] == (unsigned char)(id >> 8) && reply[header + 5] == (unsigned char)id) {
                replies++;
            }
        }
    }
    if (close(fd) < 0) {
        perror("close");
    }
    printf("%zu bytes: %s\n", size + 20, replies ? "respuesta" : "sin respuesta");
    return replies ? 0 : 1;
}

/* ecmp_sim: cuentas sobre el reparto de flujos con un hash, sin red de por medio.
 *
 *   ecmp_sim balance  <caminos> <flujos> <ensayos>   qué tan parejo sale el reparto
 *   ecmp_sim elephant <caminos> <flujos> <ensayos>   flujos grandes que chocan en un camino
 *   ecmp_sim rehash   <caminos> <flujos>             cuántos flujos cambian si muere un camino
 *
 * El hash es FNV-1a sobre la 5-tupla. No es el del kernel, pero sirve igual: lo que importa
 * es que sea determinista por flujo y que reparta como si fuera al azar.
 */
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#define MAX_PATHS 64

struct five_tuple {
    uint32_t source_ip;
    uint32_t dest_ip;
    uint16_t source_port;
    uint16_t dest_port;
    uint8_t protocol;
};

/* Se recorren los campos uno a uno: una struct puede tener relleno sin inicializar. */
static uint32_t fnv_bytes(uint32_t hash, uint64_t value, int count) {
    for (int i = 0; i < count; i++) {
        hash ^= (uint8_t)(value >> (8 * i));
        hash *= 16777619u;
    }
    return hash;
}

static uint32_t hash_flow(const struct five_tuple *flow) {
    uint32_t hash = 2166136261u;
    hash = fnv_bytes(hash, flow->source_ip, 4);
    hash = fnv_bytes(hash, flow->dest_ip, 4);
    hash = fnv_bytes(hash, flow->source_port, 2);
    hash = fnv_bytes(hash, flow->dest_port, 2);
    hash = fnv_bytes(hash, flow->protocol, 1);
    /* FNV mezcla mal los bits altos con entradas cortas: una vuelta más los reparte. */
    hash ^= hash >> 15;
    hash *= 2246822519u;
    hash ^= hash >> 13;
    return hash;
}

/* Generador xorshift con semilla fija: los resultados se pueden repetir. */
static uint64_t rng_state = 0x9E3779B97F4A7C15ull;

static uint32_t next_random(void) {
    rng_state ^= rng_state << 13;
    rng_state ^= rng_state >> 7;
    rng_state ^= rng_state << 17;
    return (uint32_t)(rng_state >> 32);
}

static struct five_tuple random_flow(void) {
    struct five_tuple flow = {.source_ip = next_random(),
                              .dest_ip = next_random(),
                              .source_port = (uint16_t)next_random(),
                              .dest_port = 443,
                              .protocol = 6};
    return flow;
}

/* Módulo: el hash dividido entre la cantidad de caminos. */
static int pick_modulo(uint32_t hash, int paths) {
    return (int)(hash % (uint32_t)paths);
}

/* Hash-threshold (RFC 2992): el espacio del hash se parte en regiones iguales. */
static int pick_threshold(uint32_t hash, int paths) {
    return (int)(((uint64_t)hash * (uint64_t)paths) >> 32);
}

/* Rendezvous: cada camino saca un puntaje con el hash del flujo y gana el mayor. Si muere un
 * camino, solo se mueven los flujos que estaban en él. */
static int pick_rendezvous(uint32_t hash, const int *alive, int paths) {
    int best = -1;
    uint32_t best_score = 0;
    for (int path = 0; path < paths; path++) {
        if (!alive[path]) {
            continue;
        }
        uint32_t score = fnv_bytes(hash, (uint64_t)path, 2);
        score ^= score >> 15;
        score *= 2246822519u;
        score ^= score >> 13;
        if (best < 0 || score > best_score) {
            best = path;
            best_score = score;
        }
    }
    return best;
}

static int run_balance(int paths, int flows, int trials) {
    double ratio_sum = 0;
    double worst_ratio = 0;
    for (int trial = 0; trial < trials; trial++) {
        int load[MAX_PATHS] = {0};
        for (int i = 0; i < flows; i++) {
            struct five_tuple flow = random_flow();
            load[pick_threshold(hash_flow(&flow), paths)]++;
        }
        int largest = 0;
        for (int path = 0; path < paths; path++) {
            if (load[path] > largest) {
                largest = load[path];
            }
        }
        double ratio = (double)largest / ((double)flows / paths);
        ratio_sum += ratio;
        if (ratio > worst_ratio) {
            worst_ratio = ratio;
        }
    }
    printf("%d caminos, %d flujos: el camino más cargado lleva %.2f veces lo justo en promedio "
           "(peor ensayo: %.2f)\n", paths, flows, ratio_sum / trials, worst_ratio);
    return 0;
}

static int run_elephant(int paths, int flows, int trials) {
    /* Cada flujo quiere un enlace completo. Los que comparten un camino se lo reparten, así
     * que el total es la cantidad de caminos ocupados. */
    long clean = 0;
    long used_sum = 0;
    for (int trial = 0; trial < trials; trial++) {
        int load[MAX_PATHS] = {0};
        for (int i = 0; i < flows; i++) {
            struct five_tuple flow = random_flow();
            load[pick_threshold(hash_flow(&flow), paths)]++;
        }
        int used = 0;
        int collision = 0;
        for (int path = 0; path < paths; path++) {
            if (load[path] > 0) {
                used++;
            }
            if (load[path] > 1) {
                collision = 1;
            }
        }
        used_sum += used;
        clean += !collision;
    }
    printf("%d caminos, %d flujos grandes: sin choques en %.1f %% de los ensayos; "
           "caminos ocupados en promedio %.2f; throughput total %.0f %% del ideal\n",
           paths, flows, 100.0 * (double)clean / trials, (double)used_sum / trials,
           100.0 * ((double)used_sum / trials) / (flows < paths ? flows : paths));
    return 0;
}

static int run_rehash(int paths, int flows) {
    int alive[MAX_PATHS];
    for (int path = 0; path < paths; path++) {
        alive[path] = 1;
    }
    /* Muere un camino de en medio: en módulo y en hash-threshold, la tabla
     * se vuelve a numerar con un camino menos. */
    long moved_modulo = 0, moved_threshold = 0, moved_rendezvous = 0, on_dead = 0;
    int dead = paths / 2;
    alive[dead] = 0;
    int remap[MAX_PATHS];
    for (int path = 0, next = 0; path < paths; path++) {
        if (path != dead) {
            remap[next++] = path;
        }
    }
    int all[MAX_PATHS];
    for (int path = 0; path < paths; path++) {
        all[path] = 1;
    }
    for (int i = 0; i < flows; i++) {
        struct five_tuple flow = random_flow();
        uint32_t hash = hash_flow(&flow);
        int before_modulo = pick_modulo(hash, paths);
        int before_threshold = pick_threshold(hash, paths);
        int before_rendezvous = pick_rendezvous(hash, all, paths);
        on_dead += before_threshold == dead;
        moved_modulo += remap[pick_modulo(hash, paths - 1)] != before_modulo;
        moved_threshold += remap[pick_threshold(hash, paths - 1)] != before_threshold;
        moved_rendezvous += pick_rendezvous(hash, alive, paths) != before_rendezvous;
    }
    printf("%d caminos, muere uno; estaban en él: %.1f %% de los flujos\n", paths,
           100.0 * (double)on_dead / flows);
    printf("  módulo:         se mueven %.1f %%\n", 100.0 * (double)moved_modulo / flows);
    printf("  hash-threshold: se mueven %.1f %%\n", 100.0 * (double)moved_threshold / flows);
    printf("  rendezvous:     se mueven %.1f %%\n", 100.0 * (double)moved_rendezvous / flows);
    return 0;
}

static int parse_count(const char *text, int low, int high) {
    char *end;
    long value = strtol(text, &end, 10);
    if (*text == '\0' || *end != '\0' || value < low || value > high) {
        fprintf(stderr, "valor fuera de rango: %s\n", text);
        exit(2);
    }
    return (int)value;
}

int main(int argc, char **argv) {
    if (argc == 5 && strcmp(argv[1], "balance") == 0) {
        return run_balance(parse_count(argv[2], 1, MAX_PATHS), parse_count(argv[3], 1, 10000000),
                           parse_count(argv[4], 1, 100000));
    }
    if (argc == 5 && strcmp(argv[1], "elephant") == 0) {
        return run_elephant(parse_count(argv[2], 1, MAX_PATHS), parse_count(argv[3], 1, 10000),
                            parse_count(argv[4], 1, 10000000));
    }
    if (argc == 4 && strcmp(argv[1], "rehash") == 0) {
        return run_rehash(parse_count(argv[2], 2, MAX_PATHS), parse_count(argv[3], 1, 100000000));
    }
    fprintf(stderr, "uso: %s balance|elephant <caminos> <flujos> <ensayos>\n"
                    "     %s rehash <caminos> <flujos>\n", argv[0], argv[0]);
    return 2;
}

/* flow_probe: varios flujos UDP lentos, para ver qué le pasa a cada uno cuando falla un camino.
 *
 *   flow_probe recv <puerto> <segundos>
 *   flow_probe send <ip> <puerto> <flujos> <intervalo_ms> <segundos>
 *
 * Cada flujo es un socket con su propio puerto de origen, así que ECMP puede mandar a cada
 * uno por un camino distinto. Solo se mide un sentido: el receptor cuenta lo que llega y, por
 * flujo, el hueco más largo entre dos datagramas. Dirección numérica: los binarios son estáticos.
 *
 * Formato del datagrama (8 bytes, orden de red): flujo (32 bits), secuencia (32 bits).
 */
#include <arpa/inet.h>
#include <errno.h>
#include <netinet/in.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 MAX_FLOWS 256
#define WIRE_SIZE 8

struct flow_stats {
    int seen;
    uint16_t source_port;
    uint32_t received;
    uint32_t highest_seq;
    double last_ms;
    double longest_gap_ms;
};

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

static int parse_int(const char *text, int low, int high, const char *name) {
    char *end;
    long value = strtol(text, &end, 10);
    if (*text == '\0' || *end != '\0' || value < low || value > high) {
        fprintf(stderr, "%s fuera de rango: %s\n", name, text);
        exit(2);
    }
    return (int)value;
}

static void put_u32(unsigned char *where, uint32_t value) {
    uint32_t net = htonl(value);
    memcpy(where, &net, sizeof net);
}

static uint32_t get_u32(const unsigned char *where) {
    uint32_t net;
    memcpy(&net, where, sizeof net);
    return ntohl(net);
}

static int run_receiver(int port, int seconds) {
    static struct flow_stats flows[MAX_FLOWS];
    int fd = socket(AF_INET, SOCK_DGRAM, 0);
    if (fd < 0) {
        perror("socket");
        return 1;
    }
    struct sockaddr_in local = {.sin_family = AF_INET,
                                .sin_port = htons((uint16_t)port),
                                .sin_addr.s_addr = htonl(INADDR_ANY)};
    if (bind(fd, (struct sockaddr *)&local, sizeof local) < 0) {
        perror("bind");
        close(fd);
        return 1;
    }
    double deadline = now_ms() + seconds * 1000.0;
    struct pollfd waiting = {.fd = fd, .events = POLLIN};
    for (double now = now_ms(); now < deadline; now = now_ms()) {
        int ready = poll(&waiting, 1, (int)(deadline - now) + 1);
        if (ready < 0) {
            if (errno == EINTR) {
                continue;
            }
            perror("poll");
            close(fd);
            return 1;
        }
        if (ready == 0) {
            break;
        }
        unsigned char wire[WIRE_SIZE];
        struct sockaddr_in peer;
        socklen_t peer_len = sizeof peer;
        ssize_t got = recvfrom(fd, wire, sizeof wire, 0, (struct sockaddr *)&peer, &peer_len);
        if (got < 0) {
            perror("recvfrom");
            close(fd);
            return 1;
        }
        if (got != WIRE_SIZE) {
            continue;
        }
        uint32_t id = get_u32(wire);
        uint32_t seq = get_u32(wire + 4);
        if (id >= MAX_FLOWS) {
            continue;
        }
        struct flow_stats *flow = &flows[id];
        double arrival = now_ms();
        if (flow->seen && arrival - flow->last_ms > flow->longest_gap_ms) {
            flow->longest_gap_ms = arrival - flow->last_ms;
        }
        flow->seen = 1;
        flow->source_port = ntohs(peer.sin_port);
        flow->received++;
        flow->last_ms = arrival;
        if (seq > flow->highest_seq) {
            flow->highest_seq = seq;
        }
    }
    if (close(fd) < 0) {
        perror("close");
    }
    printf("flujo puerto_origen recibidos perdidos hueco_max_ms\n");
    for (int id = 0; id < MAX_FLOWS; id++) {
        const struct flow_stats *flow = &flows[id];
        if (flow->seen) {
            printf("%5d %13u %9u %8u %12.0f\n", id, flow->source_port, flow->received,
                   flow->highest_seq + 1 - flow->received, flow->longest_gap_ms);
        }
    }
    return 0;
}

static int run_sender(const char *address, int port, int count, int interval_ms, int seconds) {
    struct sockaddr_in remote = {.sin_family = AF_INET, .sin_port = htons((uint16_t)port)};
    if (inet_pton(AF_INET, address, &remote.sin_addr) != 1) {
        fprintf(stderr, "dirección inválida: %s\n", address);
        return 2;
    }
    int sockets[MAX_FLOWS];
    for (int id = 0; id < count; id++) {
        sockets[id] = socket(AF_INET, SOCK_DGRAM, 0);
        if (sockets[id] < 0) {
            perror("socket");
            return 1;
        }
        if (connect(sockets[id], (struct sockaddr *)&remote, sizeof remote) < 0) {
            perror("connect");
            return 1;
        }
    }
    int rounds = seconds * 1000 / interval_ms;
    unsigned long refused = 0;
    double start = now_ms();
    for (int round = 0; round < rounds; round++) {
        for (int id = 0; id < count; id++) {
            unsigned char wire[WIRE_SIZE];
            put_u32(wire, (uint32_t)id);
            put_u32(wire + 4, (uint32_t)round);
            if (send(sockets[id], wire, sizeof wire, 0) < 0) {
                /* Un ICMP de vuelta o una ruta ausente no detienen la medición. */
                if (errno == ECONNREFUSED || errno == ENETUNREACH || errno == EHOSTUNREACH) {
                    refused++;
                } else {
                    perror("send");
                    return 1;
                }
            }
        }
        double next = start + (round + 1) * (double)interval_ms;
        double wait = next - now_ms();
        if (wait > 0) {
            struct timespec pause = {.tv_sec = (time_t)(wait / 1000),
                                     .tv_nsec = (long)((wait - (time_t)(wait / 1000) * 1000.0) * 1e6)};
            if (nanosleep(&pause, NULL) < 0 && errno != EINTR) {
                perror("nanosleep");
                return 1;
            }
        }
    }
    for (int id = 0; id < count; id++) {
        if (close(sockets[id]) < 0) {
            perror("close");
        }
    }
    printf("enviados %d datagramas en %d flujos; %lu envíos con error de red\n", rounds * count, count,
           refused);
    return 0;
}

int main(int argc, char **argv) {
    if (argc == 4 && strcmp(argv[1], "recv") == 0) {
        return run_receiver(parse_int(argv[2], 1, 65535, "puerto"),
                            parse_int(argv[3], 1, 3600, "segundos"));
    }
    if (argc == 7 && strcmp(argv[1], "send") == 0) {
        return run_sender(argv[2], parse_int(argv[3], 1, 65535, "puerto"),
                          parse_int(argv[4], 1, MAX_FLOWS, "flujos"),
                          parse_int(argv[5], 1, 10000, "intervalo"),
                          parse_int(argv[6], 1, 3600, "segundos"));
    }
    fprintf(stderr, "uso: %s recv <puerto> <segundos>\n"
                    "     %s send <ip> <puerto> <flujos> <intervalo_ms> <segundos>\n",
            argv[0], argv[0]);
    return 2;
}

"""Ayudas de la semana 14: 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__))
# La solución no se publica. Sin ella, se usa lo que ustedes escriben en lab/.
ANSWERS = os.path.join(HERE, "solution")
if not os.path.isdir(ANSWERS):
    ANSWERS = os.path.join(HERE, "lab")

# scenarios.py lo prende para mostrar cada comando que corre.
ECHO = False


def answer_path(name):
    """Devuelve la ruta de un archivo del laboratorio, o salta si sigue con el TODO."""
    path = os.path.join(ANSWERS, name)
    with open(path, encoding="utf-8") as file:
        if "TODO" in file.read():
            raise unittest.SkipTest(f"falta completar lab/{name}")
    return path


def run(args, timeout=240, stdin=None):
    return subprocess.run(args, capture_output=True, text=True, timeout=timeout, stdin=stdin)


class Lab:
    """Una topología de Containerlab que se levanta y se destruye desde las pruebas."""

    def __init__(self, folder, name, network):
        self.folder = os.path.join(HERE, folder)
        self.topology = f"{folder}.clab.yml"
        self.prefix = f"clab-{name}-"
        self.network = network

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

    def destroy(self):
        subprocess.run(["containerlab", "destroy", "-t", self.topology, "--cleanup"],
                       cwd=self.folder, capture_output=True, text=True, timeout=400)
        subprocess.run(["docker", "network", "rm", self.network], capture_output=True, text=True)

    def sh(self, node, *command, timeout=90):
        """Corre un comando en un nodo y devuelve el resultado completo."""
        if ECHO:
            print(f"  {node}# {' '.join(command)}", flush=True)
        return run(["docker", "exec", self.prefix + node, *command], timeout=timeout)

    def out(self, node, *command, timeout=90):
        """Corre un comando en un nodo y devuelve stdout y stderr juntos."""
        done = self.sh(node, *command, timeout=timeout)
        return done.stdout + done.stderr

    def daemon(self, node, command):
        """Deja un comando corriendo en segundo plano dentro de un nodo."""
        if ECHO:
            print(f"  {node}# {command} &", flush=True)
        run(["docker", "exec", "-d", self.prefix + node, "sh", "-c", command])


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()

# prioridad | match | acción. Gana la prioridad más alta. Lo que no coincide sigue a BGP.
100 | ip saddr 10.214.1.20 tcp dport 5201 | drop
 90 | tcp dport 5201 | via s1
 80 | tcp dport 5202 | via s2

/* ring_reduce: un all-reduce de verdad sobre TCP, en anillo.
 *
 *   ring_reduce <rango> <n> <puerto> <ip_siguiente> <bytes> <rondas>
 *
 * Cada nodo tiene un vector de enteros de 64 bits y al final de cada ronda todos deben tener la
 * suma de todos los vectores. En cada uno de los n-1 pasos, cada nodo le pasa al siguiente lo
 * último que recibió del anterior y lo suma a su resultado. Nadie termina una ronda hasta que
 * le llegó el aporte de todos. Eso es lo que queremos medir: un enlace lento atrasa al grupo.
 *
 * Enviar y recibir ocurren a la vez, con poll. Si todos enviaran primero y recibieran después,
 * con vectores grandes los buffers se llenarían y el anillo entero se quedaría trabado.
 */
#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.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 CONNECT_TRIES 200

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

static long parse_long(const char *text, long low, long high, const char *name) {
    char *end;
    long value = strtol(text, &end, 10);
    if (*text == '\0' || *end != '\0' || value < low || value > high) {
        fprintf(stderr, "%s fuera de rango: %s\n", name, text);
        exit(2);
    }
    return value;
}

/* Cada entero viaja en dos palabras de 32 bits en orden de red. */
static void encode(unsigned char *wire, const uint64_t *values, size_t count) {
    for (size_t i = 0; i < count; i++) {
        uint32_t high = htonl((uint32_t)(values[i] >> 32));
        uint32_t low = htonl((uint32_t)values[i]);
        memcpy(wire + 8 * i, &high, 4);
        memcpy(wire + 8 * i + 4, &low, 4);
    }
}

static void decode(uint64_t *values, const unsigned char *wire, size_t count) {
    for (size_t i = 0; i < count; i++) {
        uint32_t high, low;
        memcpy(&high, wire + 8 * i, 4);
        memcpy(&low, wire + 8 * i + 4, 4);
        values[i] = ((uint64_t)ntohl(high) << 32) | ntohl(low);
    }
}

static int set_nonblocking(int fd) {
    int flags = fcntl(fd, F_GETFL);
    if (flags < 0 || fcntl(fd, F_SETFL, flags | O_NONBLOCK) < 0) {
        perror("fcntl");
        return -1;
    }
    return 0;
}

/* Abre el puerto propio, se conecta al siguiente y acepta al anterior. */
static int open_ring(int port, const char *next_ip, int *to_next, int *from_prev) {
    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");
        return -1;
    }
    struct sockaddr_in local = {.sin_family = AF_INET,
                                .sin_port = htons((uint16_t)port),
                                .sin_addr.s_addr = htonl(INADDR_ANY)};
    if (bind(listener, (struct sockaddr *)&local, sizeof local) < 0) {
        perror("bind");
        return -1;
    }
    if (listen(listener, 1) < 0) {
        perror("listen");
        return -1;
    }
    struct sockaddr_in remote = {.sin_family = AF_INET, .sin_port = htons((uint16_t)port)};
    if (inet_pton(AF_INET, next_ip, &remote.sin_addr) != 1) {
        fprintf(stderr, "dirección inválida: %s\n", next_ip);
        return -1;
    }
    /* Los nodos no arrancan a la vez: se reintenta mientras el siguiente no esté escuchando. */
    *to_next = -1;
    for (int attempt = 0; attempt < CONNECT_TRIES; attempt++) {
        int fd = socket(AF_INET, SOCK_STREAM, 0);
        if (fd < 0) {
            perror("socket");
            return -1;
        }
        if (connect(fd, (struct sockaddr *)&remote, sizeof remote) == 0) {
            *to_next = fd;
            break;
        }
        if (errno != ECONNREFUSED && errno != ETIMEDOUT && errno != EHOSTUNREACH && errno != ENETUNREACH) {
            perror("connect");
            close(fd);
            return -1;
        }
        close(fd);
        struct timespec pause = {.tv_sec = 0, .tv_nsec = 100000000};
        if (nanosleep(&pause, NULL) < 0 && errno != EINTR) {
            perror("nanosleep");
            return -1;
        }
    }
    if (*to_next < 0) {
        fprintf(stderr, "no se pudo conectar con %s\n", next_ip);
        return -1;
    }
    *from_prev = accept(listener, NULL, NULL);
    if (*from_prev < 0) {
        perror("accept");
        return -1;
    }
    if (close(listener) < 0) {
        perror("close");
    }
    if (setsockopt(*to_next, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof yes) < 0) {
        perror("setsockopt");
        return -1;
    }
    if (set_nonblocking(*to_next) < 0 || set_nonblocking(*from_prev) < 0) {
        return -1;
    }
    return 0;
}

/* Envía out[] y recibe in[], los dos de size bytes, sin que uno bloquee al otro. */
static int exchange(int to_next, int from_prev, const unsigned char *out, unsigned char *in, size_t size) {
    size_t sent = 0, received = 0;
    while (sent < size || received < size) {
        struct pollfd fds[2] = {{.fd = sent < size ? to_next : -1, .events = POLLOUT},
                                {.fd = received < size ? from_prev : -1, .events = POLLIN}};
        if (poll(fds, 2, -1) < 0) {
            if (errno == EINTR) {
                continue;
            }
            perror("poll");
            return -1;
        }
        if (fds[0].revents & (POLLOUT | POLLERR | POLLHUP)) {
            ssize_t n = send(to_next, out + sent, size - sent, MSG_NOSIGNAL);
            if (n < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
                perror("send");
                return -1;
            }
            if (n > 0) {
                sent += (size_t)n;
            }
        }
        if (fds[1].revents & (POLLIN | POLLERR | POLLHUP)) {
            ssize_t n = recv(from_prev, in + received, size - received, 0);
            if (n < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
                perror("recv");
                return -1;
            }
            if (n == 0) {
                fprintf(stderr, "el nodo anterior cerró la conexión\n");
                return -1;
            }
            if (n > 0) {
                received += (size_t)n;
            }
        }
    }
    return 0;
}

int main(int argc, char **argv) {
    if (argc != 7) {
        fprintf(stderr, "uso: %s <rango> <n> <puerto> <ip_siguiente> <bytes> <rondas>\n", argv[0]);
        return 2;
    }
    int nodes = (int)parse_long(argv[2], 2, 1024, "n");
    int rank = (int)parse_long(argv[1], 0, nodes - 1, "rango");
    int port = (int)parse_long(argv[3], 1, 65535, "puerto");
    size_t count = (size_t)parse_long(argv[5], 8, 1L << 30, "bytes") / 8;
    int rounds = (int)parse_long(argv[6], 1, 100000, "rondas");
    size_t size = count * 8;

    uint64_t *transit = malloc(size), *result = malloc(size);
    unsigned char *out = malloc(size), *in = malloc(size);
    if (!transit || !result || !out || !in) {
        perror("malloc");
        return 1;
    }
    int to_next, from_prev;
    if (open_ring(port, argv[4], &to_next, &from_prev) < 0) {
        return 1;
    }
    for (int round = 0; round < rounds; round++) {
        double start = now_ms();
        for (size_t i = 0; i < count; i++) {
            transit[i] = (uint64_t)(rank + 1 + round) + i;
            result[i] = transit[i];
        }
        for (int step = 0; step < nodes - 1; step++) {
            encode(out, transit, count);
            if (exchange(to_next, from_prev, out, in, size) < 0) {
                return 1;
            }
            decode(transit, in, count);
            for (size_t i = 0; i < count; i++) {
                result[i] += transit[i];
            }
        }
        /* Se comprueba la suma: si el anillo perdiera o cruzara datos, aquí se notaría. */
        for (size_t i = 0; i < count; i++) {
            uint64_t expected = (uint64_t)nodes * ((uint64_t)round + i) +
                                (uint64_t)nodes * ((uint64_t)nodes + 1) / 2;
            if (result[i] != expected) {
                fprintf(stderr, "suma incorrecta en la ronda %d\n", round);
                return 1;
            }
        }
        printf("ronda %d %.1f ms\n", round, now_ms() - start);
        if (fflush(stdout) == EOF) {
            perror("fflush");
            return 1;
        }
    }
    if (close(to_next) < 0 || close(from_prev) < 0) {
        perror("close");
    }
    free(transit);
    free(result);
    free(out);
    free(in);
    return 0;
}