/* bench.c -- what synchronisation actually costs, measured.
 *
 * Every number on concurrency.html comes from running this file. It is one
 * translation unit, C11 plus POSIX threads, no dependencies:
 *
 *     cc -O2 -std=c11 -pthread -o bench data/sync/bench.c && ./bench
 *
 * Each case is run REPEATS times and the median is reported, because a single
 * run on a shared machine measures the neighbours as much as the code. The
 * baseline case is a non-atomic increment on a private variable: it is what a
 * counter costs when nothing is coordinating, and every other number here is
 * only meaningful against it.
 */
#define _GNU_SOURCE
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <sched.h>

#define REPEATS      5
#define CACHELINE    64
#define OPS_FAST  20000000L   /* uncontended paths */
#define OPS_CONT   2000000L   /* per thread, contended paths */
#define OPS_QUEUE  1000000L   /* items through a queue */

static int THREADS = 4;

static double now_ns(void)
{
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return ts.tv_sec * 1e9 + ts.tv_nsec;
}

static int cmp_double(const void *a, const void *b)
{
    double x = *(const double *)a, y = *(const double *)b;
    return (x > y) - (x < y);
}

static double median(double *v, int n)
{
    qsort(v, n, sizeof *v, cmp_double);
    return n % 2 ? v[n / 2] : (v[n / 2 - 1] + v[n / 2]) / 2;
}

/* Pinning each thread to its own core removes migration noise. It is a hint:
   a container may forbid it, and the benchmark is still valid without it. */
static void pin(int cpu)
{
    cpu_set_t set;
    CPU_ZERO(&set);
    CPU_SET(cpu, &set);
    pthread_setaffinity_np(pthread_self(), sizeof set, &set);
}

/* ------------------------------------------------------------ primitives */

typedef struct {
    atomic_flag flag;
} spinlock_t;

static void spin_lock(spinlock_t *s)
{
    while (atomic_flag_test_and_set_explicit(&s->flag, memory_order_acquire)) {
#if defined(__x86_64__) || defined(__i386__)
        __builtin_ia32_pause();      /* PAUSE: yields the pipeline, not the CPU */
#endif
    }
}

static void spin_unlock(spinlock_t *s)
{
    atomic_flag_clear_explicit(&s->flag, memory_order_release);
}

/* ------------------------------------------------- 1. uncontended costs */

static long plain_counter;
static atomic_long atomic_counter;
static pthread_mutex_t plain_mutex = PTHREAD_MUTEX_INITIALIZER;
static spinlock_t plain_spin;

typedef enum { OP_PLAIN, OP_RELAXED, OP_SEQCST, OP_CAS, OP_MUTEX, OP_SPIN } op_t;

static void run_ops(op_t op, long n)
{
    switch (op) {
    case OP_PLAIN:
        for (long i = 0; i < n; i++) {
            /* volatile so the loop is not folded away; still non-atomic */
            *(volatile long *)&plain_counter += 1;
        }
        break;
    case OP_RELAXED:
        for (long i = 0; i < n; i++)
            atomic_fetch_add_explicit(&atomic_counter, 1, memory_order_relaxed);
        break;
    case OP_SEQCST:
        for (long i = 0; i < n; i++)
            atomic_fetch_add(&atomic_counter, 1);
        break;
    case OP_CAS:
        for (long i = 0; i < n; i++) {
            long cur = atomic_load_explicit(&atomic_counter, memory_order_relaxed);
            while (!atomic_compare_exchange_weak_explicit(
                       &atomic_counter, &cur, cur + 1,
                       memory_order_release, memory_order_relaxed)) { }
        }
        break;
    case OP_MUTEX:
        for (long i = 0; i < n; i++) {
            pthread_mutex_lock(&plain_mutex);
            plain_counter++;
            pthread_mutex_unlock(&plain_mutex);
        }
        break;
    case OP_SPIN:
        for (long i = 0; i < n; i++) {
            spin_lock(&plain_spin);
            plain_counter++;
            spin_unlock(&plain_spin);
        }
        break;
    }
}

static double bench_uncontended(op_t op)
{
    double runs[REPEATS];
    for (int r = 0; r < REPEATS; r++) {
        double t0 = now_ns();
        run_ops(op, OPS_FAST);
        runs[r] = (now_ns() - t0) / OPS_FAST;
    }
    return median(runs, REPEATS);
}

/* --------------------------------------------------- 2. contended costs */

typedef struct {
    op_t op;
    int cpu;
    long ops;
} worker_arg_t;

static void *contend_worker(void *p)
{
    worker_arg_t *a = p;
    pin(a->cpu);
    run_ops(a->op, a->ops);
    return NULL;
}

static double bench_contended(op_t op)
{
    double runs[REPEATS];
    for (int r = 0; r < REPEATS; r++) {
        pthread_t th[64];
        worker_arg_t args[64];
        double t0 = now_ns();
        for (int i = 0; i < THREADS; i++) {
            args[i] = (worker_arg_t){ .op = op, .cpu = i, .ops = OPS_CONT };
            pthread_create(&th[i], NULL, contend_worker, &args[i]);
        }
        for (int i = 0; i < THREADS; i++)
            pthread_join(th[i], NULL);
        /* ns per operation, counting every thread's operations */
        runs[r] = (now_ns() - t0) / (OPS_CONT * (double)THREADS);
    }
    return median(runs, REPEATS);
}

/* ------------------------------------------------------ 3. false sharing */

typedef struct { atomic_long v; } packed_counter_t;
typedef struct { _Alignas(CACHELINE) atomic_long v; } padded_counter_t;

/* The packed array is aligned as a whole so the first eight counters land in
   one 64-byte line deterministically. Without this the array's own alignment
   is 8, the counters may straddle a boundary, and the comparison measures a
   different amount of sharing on every build -- which would make the headline
   ratio an accident rather than a measurement. */
static _Alignas(CACHELINE) packed_counter_t packed[64];
static padded_counter_t padded[64];

typedef struct { int idx; int padded; long ops; } share_arg_t;

static void *share_worker(void *p)
{
    share_arg_t *a = p;
    pin(a->idx);
    atomic_long *slot = a->padded ? &padded[a->idx].v : &packed[a->idx].v;
    for (long i = 0; i < a->ops; i++)
        atomic_fetch_add_explicit(slot, 1, memory_order_relaxed);
    return NULL;
}

static double bench_sharing(int use_padding)
{
    double runs[REPEATS];
    for (int r = 0; r < REPEATS; r++) {
        pthread_t th[64];
        share_arg_t args[64];
        memset(packed, 0, sizeof packed);
        memset(padded, 0, sizeof padded);
        double t0 = now_ns();
        for (int i = 0; i < THREADS; i++) {
            args[i] = (share_arg_t){ .idx = i, .padded = use_padding, .ops = OPS_CONT };
            pthread_create(&th[i], NULL, share_worker, &args[i]);
        }
        for (int i = 0; i < THREADS; i++)
            pthread_join(th[i], NULL);
        runs[r] = (now_ns() - t0) / (OPS_CONT * (double)THREADS);
    }
    return median(runs, REPEATS);
}

/* --------------------------------------------- 4. SPSC queue, two designs */

#define QCAP 1024
#define QMASK (QCAP - 1)

/* Lock-free: the producer owns head, the consumer owns tail, and the two
   indices sit on separate cache lines so the hand-off is the only sharing.
   store-release on the index publishes the slot written before it; the
   matching load-acquire is what makes that write visible to the reader. */
typedef struct {
    _Alignas(CACHELINE) atomic_size_t head;
    _Alignas(CACHELINE) atomic_size_t tail;
    _Alignas(CACHELINE) void *slot[QCAP];
} spsc_t;

static void spsc_init(spsc_t *q)
{
    atomic_init(&q->head, 0);
    atomic_init(&q->tail, 0);
}

static void spsc_push(spsc_t *q, void *v)
{
    size_t h = atomic_load_explicit(&q->head, memory_order_relaxed);
    while (h - atomic_load_explicit(&q->tail, memory_order_acquire) == QCAP)
        ;                                   /* full: spin (SPSC, so brief) */
    q->slot[h & QMASK] = v;
    atomic_store_explicit(&q->head, h + 1, memory_order_release);
}

static void *spsc_pop(spsc_t *q)
{
    size_t t = atomic_load_explicit(&q->tail, memory_order_relaxed);
    while (t == atomic_load_explicit(&q->head, memory_order_acquire))
        ;                                   /* empty */
    void *v = q->slot[t & QMASK];
    atomic_store_explicit(&q->tail, t + 1, memory_order_release);
    return v;
}

/* Mutex + two condition variables: the textbook bounded queue. */
typedef struct {
    void *slot[QCAP];
    size_t head, tail, count;
    pthread_mutex_t lock;
    pthread_cond_t not_empty, not_full;
} mq_t;

static void mq_reset(mq_t *q)
{
    q->head = q->tail = q->count = 0;
}

static void mq_init(mq_t *q)
{
    mq_reset(q);
    pthread_mutex_init(&q->lock, NULL);
    pthread_cond_init(&q->not_empty, NULL);
    pthread_cond_init(&q->not_full, NULL);
}

static void mq_push(mq_t *q, void *v)
{
    pthread_mutex_lock(&q->lock);
    while (q->count == QCAP)
        pthread_cond_wait(&q->not_full, &q->lock);
    q->slot[q->tail] = v;
    q->tail = (q->tail + 1) & QMASK;
    q->count++;
    pthread_cond_signal(&q->not_empty);
    pthread_mutex_unlock(&q->lock);
}

static void *mq_pop(mq_t *q)
{
    pthread_mutex_lock(&q->lock);
    while (q->count == 0)
        pthread_cond_wait(&q->not_empty, &q->lock);
    void *v = q->slot[q->head];
    q->head = (q->head + 1) & QMASK;
    q->count--;
    pthread_cond_signal(&q->not_full);
    pthread_mutex_unlock(&q->lock);
    return v;
}

static spsc_t lockfree_q;
static mq_t mutex_q;

static void *spsc_producer(void *unused)
{
    (void)unused; pin(0);
    for (long i = 1; i <= OPS_QUEUE; i++)
        spsc_push(&lockfree_q, (void *)(intptr_t)i);
    return NULL;
}

static void *spsc_consumer(void *unused)
{
    (void)unused; pin(1);
    long sum = 0;
    for (long i = 0; i < OPS_QUEUE; i++)
        sum += (intptr_t)spsc_pop(&lockfree_q);
    return (void *)(intptr_t)sum;
}

static void *mq_producer(void *unused)
{
    (void)unused; pin(0);
    for (long i = 1; i <= OPS_QUEUE; i++)
        mq_push(&mutex_q, (void *)(intptr_t)i);
    return NULL;
}

static void *mq_consumer(void *unused)
{
    (void)unused; pin(1);
    long sum = 0;
    for (long i = 0; i < OPS_QUEUE; i++)
        sum += (intptr_t)mq_pop(&mutex_q);
    return (void *)(intptr_t)sum;
}

static double bench_queue(int lockfree, long *checksum)
{
    double runs[REPEATS];
    void *sum = NULL;
    for (int r = 0; r < REPEATS; r++) {
        pthread_t p, c;
        /* Re-initialising a live mutex or condvar is undefined, and doing it
           inside the timed region would charge the queue for it. Both queues
           are drained to empty by construction -- the consumer takes exactly
           what the producer puts in -- so the same objects are reusable. */
        if (lockfree) spsc_init(&lockfree_q);
        else if (r == 0) mq_init(&mutex_q);
        else mq_reset(&mutex_q);
        double t0 = now_ns();
        pthread_create(&p, NULL, lockfree ? spsc_producer : mq_producer, NULL);
        pthread_create(&c, NULL, lockfree ? spsc_consumer : mq_consumer, NULL);
        pthread_join(p, NULL);
        pthread_join(c, &sum);
        runs[r] = (now_ns() - t0) / OPS_QUEUE;
    }
    *checksum = (long)(intptr_t)sum;
    return median(runs, REPEATS);
}


/* --------------------------------------------- 5. wake-up: sleep vs spin
 * A ping-pong round trip. The condvar pair parks the waiter in the kernel;
 * the atomic pair keeps it burning a core. This is the cost that decides
 * between them, and it is why "spinlocks are for short sections" is a claim
 * about *how long the holder holds*, not about how short your own work is. */

static pthread_mutex_t pp_lock = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t  pp_cond = PTHREAD_COND_INITIALIZER;
static int pp_turn;
static atomic_int pp_spin_turn;
#define PP_ROUNDS 200000L

static void *pp_condvar_peer(void *unused)
{
    (void)unused; pin(1);
    for (long i = 0; i < PP_ROUNDS; i++) {
        pthread_mutex_lock(&pp_lock);
        while (pp_turn != 1)
            pthread_cond_wait(&pp_cond, &pp_lock);
        pp_turn = 0;
        pthread_cond_signal(&pp_cond);
        pthread_mutex_unlock(&pp_lock);
    }
    return NULL;
}

static void *pp_spin_peer(void *unused)
{
    (void)unused; pin(1);
    for (long i = 0; i < PP_ROUNDS; i++) {
        while (atomic_load_explicit(&pp_spin_turn, memory_order_acquire) != 1)
#if defined(__x86_64__) || defined(__i386__)
            __builtin_ia32_pause();
#else
            ;
#endif
        atomic_store_explicit(&pp_spin_turn, 0, memory_order_release);
    }
    return NULL;
}

static double bench_pingpong(int spin)
{
    double runs[REPEATS];
    for (int r = 0; r < REPEATS; r++) {
        pthread_t peer;
        pp_turn = 0;
        atomic_init(&pp_spin_turn, 0);
        pthread_create(&peer, NULL, spin ? pp_spin_peer : pp_condvar_peer, NULL);
        pin(0);
        double t0 = now_ns();
        for (long i = 0; i < PP_ROUNDS; i++) {
            if (spin) {
                atomic_store_explicit(&pp_spin_turn, 1, memory_order_release);
                while (atomic_load_explicit(&pp_spin_turn, memory_order_acquire) != 0)
#if defined(__x86_64__) || defined(__i386__)
                    __builtin_ia32_pause();
#else
                    ;
#endif
            } else {
                pthread_mutex_lock(&pp_lock);
                pp_turn = 1;
                pthread_cond_signal(&pp_cond);
                while (pp_turn != 0)
                    pthread_cond_wait(&pp_cond, &pp_lock);
                pthread_mutex_unlock(&pp_lock);
            }
        }
        runs[r] = (now_ns() - t0) / PP_ROUNDS;
        pthread_join(peer, NULL);
    }
    return median(runs, REPEATS);
}

/* ------------------------------------ 6. rwlock vs mutex, read-dominated
 * The claim under test: a reader-writer lock buys throughput when readers
 * dominate. It is true only when the read section is long enough to pay for
 * the lock's own atomic traffic, which is what varies here. */

static pthread_rwlock_t rw = PTHREAD_RWLOCK_INITIALIZER;
static pthread_mutex_t rd_mutex = PTHREAD_MUTEX_INITIALIZER;
static volatile long shared_payload[16];

typedef struct { int use_rwlock; int work; long ops; int cpu; } rd_arg_t;

static long read_work(int units)
{
    long acc = 0;
    for (int i = 0; i < units; i++)
        acc += shared_payload[i & 15];
    return acc;
}

static void *reader_worker(void *p)
{
    rd_arg_t *a = p;
    pin(a->cpu);
    long acc = 0;
    for (long i = 0; i < a->ops; i++) {
        if (a->use_rwlock) {
            pthread_rwlock_rdlock(&rw);
            acc += read_work(a->work);
            pthread_rwlock_unlock(&rw);
        } else {
            pthread_mutex_lock(&rd_mutex);
            acc += read_work(a->work);
            pthread_mutex_unlock(&rd_mutex);
        }
    }
    return (void *)(intptr_t)acc;
}

static double bench_readers(int use_rwlock, int work)
{
    double runs[REPEATS];
    long ops = OPS_CONT / (work > 8 ? 8 : 1);
    for (int r = 0; r < REPEATS; r++) {
        pthread_t th[64];
        rd_arg_t args[64];
        double t0 = now_ns();
        for (int i = 0; i < THREADS; i++) {
            args[i] = (rd_arg_t){ .use_rwlock = use_rwlock, .work = work,
                                  .ops = ops, .cpu = i };
            pthread_create(&th[i], NULL, reader_worker, &args[i]);
        }
        for (int i = 0; i < THREADS; i++)
            pthread_join(th[i], NULL);
        runs[r] = (now_ns() - t0) / (ops * (double)THREADS);
    }
    return median(runs, REPEATS);
}

/* ------------------------------------------- 7. how contention scales */

static double bench_scaling(op_t op, int threads)
{
    int saved = THREADS;
    THREADS = threads;
    double v = bench_contended(op);
    THREADS = saved;
    return v;
}

/* ------------------------------------------------------------------ main */

int main(int argc, char **argv)
{
    if (argc > 1) THREADS = atoi(argv[1]);
    if (THREADS < 1 || THREADS > 64) THREADS = 4;

    printf("# threads=%d repeats=%d cacheline=%d\n", THREADS, REPEATS, CACHELINE);
    printf("case,ns_per_op\n");

    printf("uncontended_plain,%.3f\n",   bench_uncontended(OP_PLAIN));
    printf("uncontended_relaxed,%.3f\n", bench_uncontended(OP_RELAXED));
    printf("uncontended_seqcst,%.3f\n",  bench_uncontended(OP_SEQCST));
    printf("uncontended_cas,%.3f\n",     bench_uncontended(OP_CAS));
    printf("uncontended_mutex,%.3f\n",   bench_uncontended(OP_MUTEX));
    printf("uncontended_spin,%.3f\n",    bench_uncontended(OP_SPIN));

    printf("contended_relaxed,%.3f\n",   bench_contended(OP_RELAXED));
    printf("contended_seqcst,%.3f\n",    bench_contended(OP_SEQCST));
    printf("contended_cas,%.3f\n",       bench_contended(OP_CAS));
    printf("contended_mutex,%.3f\n",     bench_contended(OP_MUTEX));
    printf("contended_spin,%.3f\n",      bench_contended(OP_SPIN));

    printf("false_sharing_packed,%.3f\n", bench_sharing(0));
    printf("false_sharing_padded,%.3f\n", bench_sharing(1));

    long ck_lf = 0, ck_mq = 0;
    double lf = bench_queue(1, &ck_lf);
    double mu = bench_queue(0, &ck_mq);
    long expect = OPS_QUEUE * (OPS_QUEUE + 1) / 2;
    printf("queue_lockfree,%.3f\n", lf);
    printf("queue_mutex_condvar,%.3f\n", mu);

    printf("pingpong_condvar,%.3f\n", bench_pingpong(0));
    printf("pingpong_spin,%.3f\n",    bench_pingpong(1));

    printf("read_short_mutex,%.3f\n",  bench_readers(0, 1));
    printf("read_short_rwlock,%.3f\n", bench_readers(1, 1));
    printf("read_long_mutex,%.3f\n",   bench_readers(0, 256));
    printf("read_long_rwlock,%.3f\n",  bench_readers(1, 256));

    for (int t = 1; t <= THREADS; t++) {
        printf("scale_%d_mutex,%.3f\n",   t, bench_scaling(OP_MUTEX, t));
        printf("scale_%d_atomic,%.3f\n",  t, bench_scaling(OP_RELAXED, t));
        printf("scale_%d_spin,%.3f\n",    t, bench_scaling(OP_SPIN, t));
    }
    printf("# checksums lockfree=%ld mutex=%ld expected=%ld %s\n",
           ck_lf, ck_mq, expect,
           (ck_lf == expect && ck_mq == expect) ? "OK" : "MISMATCH");
    return (ck_lf == expect && ck_mq == expect) ? 0 : 1;
}