/* 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;
}