/* EXERCISE 03 — BROKEN ON PURPOSE. Do not copy this shape into real code.
 *
 * An ingest pipeline. Frames arrive from a capture device roughly every 5 ms.
 * A pool of worker threads picks each frame up and hashes it.
 *
 * Symptom as reported by the field: "the fans spin up and the battery drains
 * even when almost nothing is arriving. Activity Monitor shows us pinned near
 * 100% of several cores while the frame rate is only 200 per second."
 *
 * Build and run:
 *   clang -O2 -g -Wall -Wextra -pthread pipeline.c -o /tmp/pipeline_broken
 *   /tmp/pipeline_broken
 *
 * The output is CORRECT. Every frame is processed exactly once, and the
 * checksum is right. This exercise is not about correctness.
 *
 * Every measurement is printed as key=value so a script can assert on it.
 */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <stdatomic.h>
#include <sys/resource.h>
#include <sys/time.h>

#define FRAMES       200        /* frames the capture device delivers      */
#define FRAME_GAP_US 5000       /* 5 ms between frames = 200 fps           */
#define WATCHDOG_S   60

static double now_ms(void) {
    struct timeval t; gettimeofday(&t, NULL);
    return t.tv_sec * 1000.0 + t.tv_usec / 1000.0;
}

/* ------------------------------------------------------------------ safety */
static void *watchdog(void *unused) {
    (void)unused;
    sleep(WATCHDOG_S);
    const char m[] = "WATCHDOG: no progress within budget - forcing _exit(75).\n";
    ssize_t ignored = write(2, m, sizeof m - 1); (void)ignored;
    _exit(75);
}

/* ------------------------------------------------------------ the pipeline */
/* One slot. The producer publishes a sequence number; consumers claim it. */
static atomic_int  published = 0;      /* last frame number made available */
static atomic_int  claimed   = 0;      /* last frame number taken by a worker */
static atomic_int  processed = 0;
static atomic_ullong checksum = 0;
static atomic_int  shutting_down = 0;

static void hash_frame(int seq) {
    unsigned long long h = 1469598103934665603ULL;
    for (int i = 0; i < 64; i++) { h ^= (unsigned long long)(seq + i); h *= 1099511628211ULL; }
    atomic_fetch_add(&checksum, h);
    atomic_fetch_add(&processed, 1);
}

static void *worker(void *unused) {
    (void)unused;
    for (;;) {
        /* Wait for a frame to become available.
         *
         * This loop never leaves the CPU. The thread stays RUNNABLE, so the
         * scheduler keeps handing it a core to do nothing on. */
        int mine;
        for (;;) {
            if (atomic_load_explicit(&shutting_down, memory_order_acquire)) return NULL;
            int avail = atomic_load_explicit(&published, memory_order_acquire);
            int taken = atomic_load_explicit(&claimed,   memory_order_relaxed);
            if (taken < avail) {
                mine = taken + 1;
                if (atomic_compare_exchange_weak_explicit(
                        &claimed, &taken, mine,
                        memory_order_acq_rel, memory_order_relaxed)) break;
            }
            /* spin again */
        }
        hash_frame(mine);
    }
}

int main(void) {
    pthread_t wd;
    pthread_create(&wd, NULL, watchdog, NULL);
    pthread_detach(wd);

    int ncpu = (int)sysconf(_SC_NPROCESSORS_ONLN);
    int nworkers = ncpu;          /* one worker per core: a reasonable pool */

    pthread_t *t = calloc((size_t)nworkers, sizeof *t);
    for (int i = 0; i < nworkers; i++) pthread_create(&t[i], NULL, worker, NULL);

    struct rusage r0, r1;
    getrusage(RUSAGE_SELF, &r0);
    double t0 = now_ms();

    for (int f = 1; f <= FRAMES; f++) {
        usleep(FRAME_GAP_US);
        atomic_store_explicit(&published, f, memory_order_release);
    }
    /* Wait for the last frames to drain. */
    while (atomic_load(&processed) < FRAMES) usleep(1000);

    atomic_store_explicit(&shutting_down, 1, memory_order_release);
    for (int i = 0; i < nworkers; i++) pthread_join(t[i], NULL);

    double wall = now_ms() - t0;
    getrusage(RUSAGE_SELF, &r1);

    double cpu =
        (double)(r1.ru_utime.tv_sec - r0.ru_utime.tv_sec) +
        (double)(r1.ru_utime.tv_usec - r0.ru_utime.tv_usec) / 1e6 +
        (double)(r1.ru_stime.tv_sec - r0.ru_stime.tv_sec) +
        (double)(r1.ru_stime.tv_usec - r0.ru_stime.tv_usec) / 1e6;
    long vol   = r1.ru_nvcsw  - r0.ru_nvcsw;
    long invol = r1.ru_nivcsw - r0.ru_nivcsw;

    printf("workers=%d\n", nworkers);
    printf("frames=%d\n", FRAMES);
    printf("processed=%d\n", atomic_load(&processed));
    printf("checksum=%llu\n", (unsigned long long)atomic_load(&checksum));
    printf("wallMs=%.1f\n", wall);
    printf("cpuSeconds=%.3f\n", cpu);
    printf("cpuPerWallCore=%.2f\n", cpu / (wall / 1000.0));
    printf("voluntarySwitches=%ld\n", vol);
    printf("involuntarySwitches=%ld\n", invol);
    free(t);
    return atomic_load(&processed) == FRAMES ? 0 : 1;
}