client.c

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <unistd.h>
#include <time.h>
#include <errno.h>

#include <sys/types.h>
#include <sys/socket.h>
#include <sys/uio.h>
#include <netdb.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#include <assert.h>

typedef struct program_options
{
    const char* ip;
    const char* port;
} program_options;

typedef struct stats
{
    long* latency_ns;
    size_t n_latency;
    size_t index;
    size_t count;
} stats;

program_options* parse_options(int argc, char* argv[])
{
    if (argc < 2)
    {
        puts("Usage client host port");
        return NULL;
    }

    program_options* options = malloc(sizeof *options);
    if (options == NULL)
        return NULL;

    const char* ip = argv[1];
    const char* port = argv[2];

    options->ip = ip;
    options->port = port;

    return options;
}

stats* stats_new(size_t count)
{
    stats* stats = malloc(sizeof* stats);
    if (stats == NULL)
        return NULL;

    stats->latency_ns = malloc(count * sizeof(long));
    if (stats->latency_ns == NULL)
        return NULL;

    memset(stats->latency_ns, 0, count);

    stats->n_latency = count;
    stats->index = 0;
    stats->count = 0;

    return stats;
}

void stats_push(stats* stats, long latency_ns)
{
    stats->latency_ns[stats->index] = latency_ns;
    stats->index = (stats->index + 1) % stats->n_latency;
    stats->count = stats->count + 1;
}

static int latency_less(const void* l1, const void *l2)
{
    int v1 = *(long *)l1;
    int v2 = *(long *)l2;

    return v1 - v2;
}

int pct_nth_idx(double percentile, size_t size)
{
    return (int)(percentile * size);
}

long get_pct(long* arr, double percentile, size_t size)
{
    return arr[pct_nth_idx(percentile, size)];
}

enum print_flags
{
    PRINT_NONE = 1 << 0,
    PRINT_PERCENTILES = 1 << 1,
    PRINT_NORMALIZED = 1 << 2
};

typedef struct print_options
{
    int flags;

    double percentiles[100];
    size_t n_percentiles;
} print_options;

int fill_options_percentiles(print_options* options, size_t count, const double* percentiles)
{
    if (count >= 100)
        return -1;

    memcpy(&options->percentiles, percentiles, count * sizeof(*percentiles));
    options->n_percentiles = count;

    return 0;
}

void report_stats(stats* stats, print_options options)
{
    qsort(stats->latency_ns, stats->count, sizeof(long), latency_less);
    unsigned long long total_latency = 0;

    for (size_t i = 0; i < stats->count; ++i) {
        total_latency += stats->latency_ns[i];
    }

    long mean_latency_ns = total_latency / stats->count;


    const long min = stats->latency_ns[0];

    printf("Min: %ldns\n", min);
    printf("Mean: %ldns\n", mean_latency_ns);
    printf("Max: %ldns\n", stats->latency_ns[stats->count - 1]);

    if (options.flags & PRINT_PERCENTILES)
    {
        for (size_t i = 0; i < options.n_percentiles; ++i)
        {
            printf("p(%lf) = %ldns\n", options.percentiles[i], get_pct(stats->latency_ns, options.percentiles[i], stats->count));
        }

        if (options.flags & PRINT_NORMALIZED)
        {
            for (size_t i = 0; i < options.n_percentiles; ++i)
            {
                const long pct = get_pct(stats->latency_ns, options.percentiles[i], stats->count);
                const long normalized_pct = pct - min;
                printf("p(%lf)_normalized = %ldns\n", options.percentiles[i], normalized_pct);
            }
        }
    }
}

void dump_stats(const stats* stats, const char* file_name)
{
    FILE* fp = fopen(file_name, "w");
    if (fp == NULL)
    {
        perror("fopen");
        return;
    }

    for (size_t i = 0; i < stats->n_latency; ++i)
    {
        fprintf(fp, "%ld\n",  stats->latency_ns[i]);
    }

    fclose(fp);
}

int do_connect(const program_options* options)
{
    printf("Connecting to %s:%s ...\n", options->ip, options->port);

    struct addrinfo hints;
    struct addrinfo* result, *rp;

    int sfd, ret;
    sfd = -1;

    memset(&hints, 0, sizeof(hints));
    hints.ai_family = AF_INET;
    hints.ai_socktype = SOCK_STREAM;
    hints.ai_flags = 0;
    hints.ai_protocol = 0;

    ret = getaddrinfo(options->ip, options->port, &hints, &result);
    if (ret < 0)
        return ret;

    for (rp = result; rp != NULL; rp = rp->ai_next)
    {
        sfd = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
        if (sfd < 0)
            continue;

        if (connect(sfd, rp->ai_addr, rp->ai_addrlen) != -1)
            break;

        close(sfd);
    }

    puts("... connected");

    return sfd;
}

static void handle_message(const char* buffer, size_t size, stats* stats)
{
    uint64_t timestamp = *(uint64_t *)buffer;
    struct timespec tp;
    int ret;

    ret = clock_gettime(CLOCK_REALTIME, &tp);
    if (ret == 0)
    {
        uint64_t now_epoch = (uint64_t)tp.tv_sec * 1000000000UL + (uint64_t)tp.tv_nsec;
        uint64_t latency_ns = now_epoch - timestamp;
        stats_push(stats, latency_ns);
    }
}

static void receive_loop(int sfd)
{
    enum {
        MaxMessages = 10000000
    };

    struct msghdr hdr;
    struct iovec iov;
    char buffer[128];

    size_t messages = 0;
    size_t bytes = 0;

    stats* stats = stats_new(MaxMessages);
    if (stats == NULL)
    {
        perror("stats");
        return;
    }

    iov.iov_base = buffer;
    iov.iov_len = sizeof(buffer);
    hdr.msg_name = NULL;
    hdr.msg_namelen = 0;
    hdr.msg_iov = &iov;
    hdr.msg_iovlen = 1;
    hdr.msg_control = NULL;
    hdr.msg_controllen = 0;
    hdr.msg_flags = 0;
    for (;;) {
        ssize_t ret;
        memset(buffer, 0, sizeof(buffer));
        ret = recvmsg(sfd, &hdr, MSG_WAITALL);
        if (ret == -1) {
            perror("recvmsg");
            break;
        } else if (!ret)
            break;

        ++messages;
        bytes += ret;
        handle_message(buffer, ret, stats);
    }

    printf("Received %llu messages (%llu bytes)\n", (unsigned long long) messages, (unsigned long long) bytes);

    printf("Dumping stats...\n");
    dump_stats(stats, "stats.log");
    printf("... Done\n");

    print_options options;
    options.flags = PRINT_PERCENTILES | PRINT_NORMALIZED;
    double percentiles[] = { 0.10, 0.25, 0.50, 0.75, 0.90, 0.99 };
    const size_t n_percentiles = sizeof(percentiles) / sizeof(*percentiles);
    fill_options_percentiles(&options, n_percentiles, percentiles);  

    report_stats(stats, options);
}

int main(int argc, char *argv[])
{
    program_options* options = parse_options(argc, argv);
    if (options == NULL)
        return 0;

    int sockfd = do_connect(options);
    if (sockfd < 0)
    {
        perror("connect");
        return 0;
    }

    receive_loop(sockfd);
    return 0;
}