/*
 * EXERCISE 05 — FIXED VARIANT.  Three changes, all aimed at the same thing:
 * hold the lock for less time, and take it far less often.
 *
 * THE SCENARIO
 *   A metrics aggregator digests events. Each event is scored and the score is
 *   folded into a running total.
 *
 * THE REPAIR
 *   1. The scoring moves OUT of the critical section. Nothing about
 *      score_event() needs the lock: it reads no shared state and writes none.
 *      Only the fold into the total does.
 *   2. Each worker accumulates into a LOCAL total and folds it in once per
 *      BATCH events. That divides the acquisition count by BATCH — from one per
 *      event to one per 64 events — without changing the answer, because
 *      addition is associative.
 *   3. The single hot total becomes SHARDS independent totals, each on its own
 *      cache line, summed once at the end. Contention is divided rather than
 *      removed.
 *
 *   The padding in shard_t is not decoration. Without it the shards share a
 *   cache line, every fold invalidates every other shard, and the measurement
 *   becomes one of false sharing rather than lock contention — a classic way to
 *   "prove" that sharding does not help.
 *
 *   Note what sharding did NOT solve: it works here because each worker owns a
 *   shard and the only shared invariant is a sum, which is associative. Sharding
 *   a structure with an invariant ACROSS shards buys you nothing, because you
 *   would have to hold every shard's lock to preserve it.
 *
 * WHAT THIS PROGRAM MEASURES
 *   A FIXED amount of total work — TOTAL_OPS events, whatever the thread count
 *   — split across 1, 2, 4, 8 and 16 threads. Perfect scaling would keep wall
 *   time FLAT as threads are added, because the work per thread falls by
 *   exactly as much as the thread count rises. Each cell is the median of
 *   REPEATS runs, because a single timing on a loaded machine is noise.
 *
 *   The headline number is the 16-thread time divided by the 1-thread time.
 *   Anything above 1.0 means the lock, not the work, is the bottleneck.
 *
 * Build: clang -O2 -g -Wall -Wextra aggregate.c -o aggregate_fixed
 * Run:   ./aggregate_fixed [total_ops]        (default 1000000)
 *
 * Expected: wall time that FALLS as threads are added, a ratio well below 1.0,
 * and an acquisition count roughly totalOps/64 instead of totalOps.
 * Exit status 0.
 *
 * The total is deterministic and does not depend on the thread count, so the
 * broken and fixed builds must print the SAME checksum. A "speed-up" that
 * changes the answer is not a speed-up.
 */
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <time.h>

#define EXIT_WATCHDOG 75
#define EXIT_INTERNAL 70
#define MAX_THREADS      16
#define REPEATS           5
#define SCORE_ROUNDS     48
#define SHARDS           16
#define BATCH            64

static void *watchdog(void *arg) {
    sleep((unsigned)(long)arg);
    const char m[] = "\nWATCHDOG: EX05 exceeded budget - forcing _exit(75).\n";
    ssize_t n = write(2, m, sizeof(m) - 1); (void)n;
    _exit(EXIT_WATCHDOG);
}

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

/* --------------------------------- the work --------------------------------
 * A deterministic scoring function. Its only job is to take a predictable,
 * non-trivial amount of CPU time so the critical section has some width. It
 * depends only on the event index, so the grand total is the same for any
 * thread count and any scheduling order.
 * -------------------------------------------------------------------------- */
static unsigned long score_event(unsigned long index) {
    unsigned long h = index * 0x9E3779B97F4A7C15UL;
    for (int r = 0; r < SCORE_ROUNDS; r++) {
        h ^= h >> 30; h *= 0xBF58476D1CE4E5B9UL;
        h ^= h >> 27; h *= 0x94D049BB133111EBUL;
        h ^= h >> 31;
    }
    return h;
}

/* ------------------------------- shared state ------------------------------
 * One shard per worker, each padded to its own cache line so that folding into
 * one shard does not invalidate the line holding another.
 * -------------------------------------------------------------------------- */

typedef struct {
    pthread_mutex_t m;
    unsigned long   total;
    unsigned long   acquisitions;
    char            pad[128 - sizeof(pthread_mutex_t) - 2 * sizeof(unsigned long)];
} shard_t;

static shard_t shards[SHARDS];

typedef struct { int id; unsigned long first, count; } work_t;

/* THE CRITICAL SECTION
 *
 * It now contains one addition and one increment, and it is entered once per
 * BATCH events rather than once per event. */
static void fold(int id, unsigned long *local) {
    shard_t *sh = &shards[id % SHARDS];
    pthread_mutex_lock(&sh->m);
    sh->total += *local;
    sh->acquisitions++;
    pthread_mutex_unlock(&sh->m);
    *local = 0;
}

static void *worker(void *arg) {
    work_t *w = (work_t *)arg;
    unsigned long local = 0;
    for (unsigned long i = 0; i < w->count; i++) {
        unsigned long s = score_event(w->first + i);   /* OUTSIDE the lock */
        local += s & 0xFFUL;
        if ((i % BATCH) == BATCH - 1) fold(w->id, &local);
    }
    fold(w->id, &local);                               /* the remainder */
    return NULL;
}

/* --------------------------------- driving --------------------------------- */

static double run_once(int threads, unsigned long total_ops, unsigned long *out_total,
                       unsigned long *out_acq) {
    pthread_t t[MAX_THREADS];
    work_t    w[MAX_THREADS];
    unsigned long per = total_ops / (unsigned long)threads;

    for (int i = 0; i < SHARDS; i++) {
        pthread_mutex_init(&shards[i].m, NULL);
        shards[i].total = 0;
        shards[i].acquisitions = 0;
    }

    double t0 = now_s();
    for (int i = 0; i < threads; i++) {
        w[i].id = i;
        w[i].first = (unsigned long)i * per;
        w[i].count = per;
        if (pthread_create(&t[i], NULL, worker, &w[i]) != 0) { perror("pthread_create"); _exit(EXIT_INTERNAL); }
    }
    for (int i = 0; i < threads; i++) pthread_join(t[i], NULL);
    double elapsed = now_s() - t0;

    unsigned long total = 0, acq = 0;
    for (int i = 0; i < SHARDS; i++) { total += shards[i].total; acq += shards[i].acquisitions; }
    *out_total = total;
    *out_acq = acq;
    return elapsed;
}

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

static double run_median(int threads, unsigned long total_ops, unsigned long *out_total,
                         unsigned long *out_acq) {
    double v[REPEATS];
    for (int r = 0; r < REPEATS; r++) v[r] = run_once(threads, total_ops, out_total, out_acq);
    qsort(v, REPEATS, sizeof(double), cmp_double);
    return v[REPEATS / 2];
}

int main(int argc, char **argv) {
    unsigned long total = argc > 1 ? strtoul(argv[1], NULL, 10) : 1000000UL;
    if (total < 16000UL) total = 16000UL;
    total -= total % 16UL;                      /* divisible by every thread count below */

    pthread_t wd;
    if (pthread_create(&wd, NULL, watchdog, (void *)600L) != 0) { perror("watchdog"); return EXIT_INTERNAL; }
    pthread_detach(wd);
    setvbuf(stdout, NULL, _IOLBF, 0);

    const int counts[] = { 1, 2, 4, 8, 16 };
    const int ncounts = (int)(sizeof(counts) / sizeof(counts[0]));

    printf("EX05 build=fixed  totalOps=%lu (fixed total work, split across N threads)\n", total);
    printf("  each cell is the median of %d runs; perfect scaling would keep it FLAT\n\n", REPEATS);
    printf("  %-10s", "threads");
    for (int i = 0; i < ncounts; i++) printf("%10d", counts[i]);
    printf("    (milliseconds)\n");

    unsigned long checksum = 0, acq = 0, one_acq = 0;
    double one = 0, sixteen = 0, worst = 0;
    int worst_at = 0;
    double cell[8];
    printf("  %-10s", "wall ms");
    for (int i = 0; i < ncounts; i++) {
        double d = run_median(counts[i], total, &checksum, &acq);
        cell[i] = d;
        if (counts[i] == 1)  { one = d; one_acq = acq; }
        if (counts[i] == 16) sixteen = d;
        printf("%10.0f", d * 1000.0);
    }
    printf("\n  %-10s", "vs 1 thread");
    for (int i = 0; i < ncounts; i++) {
        double ratio = one > 0 ? cell[i] / one : 0;
        if (ratio > worst) { worst = ratio; worst_at = counts[i]; }
        printf("%9.2fx", ratio);
    }
    printf("\n\n");

    printf("EX05 build=fixed totalOps=%lu checksum=%lu acquisitions=%lu "
           "oneThreadMs=%.0f sixteenThreadMs=%.0f slowdown=%.2f "
           "worstSlowdown=%.2f worstAtThreads=%d\n",
           total, checksum, one_acq, one * 1000.0, sixteen * 1000.0,
           one > 0 ? sixteen / one : -1.0, worst, worst_at);
    return 0;
}