/* EXERCISE 03 — REPAIRED.
 *
 * 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.
 *
 * The repair: an idle worker BLOCKS instead of spinning. A blocked thread is
 * off every run queue, so it costs a stack and a scheduler slot and no CPU at
 * all; the kernel makes it runnable again when the producer signals. The
 * condition variable is paired with the mutex that guards the state it tests,
 * and the wait sits in a loop because a wakeup is a hint, not a promise.
 *
 * Build and run:
 *   clang -O2 -g -Wall -Wextra -pthread pipeline.c -o /tmp/pipeline_fixed
 *   /tmp/pipeline_fixed
 *
 * The frames processed and the checksum are unchanged. Only the CPU cost of
 * waiting is different. Every measurement is printed as key=value.
 */
#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.
 *
 * `published`, `claimed` and `shutting_down` are now guarded by `m` rather
 * than being free-standing atomics, because the condition variable and the
 * predicate it tests must be protected by the same mutex. */
static pthread_mutex_t m    = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t  work = PTHREAD_COND_INITIALIZER;
static int  published = 0;             /* last frame number made available */
static int  claimed   = 0;             /* last frame number taken by a worker */
static atomic_int  processed = 0;
static atomic_ullong checksum = 0;
static 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 (;;) {
        int mine;
        pthread_mutex_lock(&m);
        /* Wait for a frame to become available.
         *
         * pthread_cond_wait releases the mutex and parks this thread in the
         * kernel. It is in a WHILE loop, not an if: a wakeup does not prove
         * the predicate, and several workers can be woken for one frame. */
        while (claimed >= published && !shutting_down)
            pthread_cond_wait(&work, &m);
        if (shutting_down && claimed >= published) { pthread_mutex_unlock(&m); return NULL; }
        mine = ++claimed;
        pthread_mutex_unlock(&m);

        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);
        pthread_mutex_lock(&m);
        published = f;
        pthread_cond_signal(&work);     /* one frame, one waiter to wake */
        pthread_mutex_unlock(&m);
    }
    /* Wait for the last frames to drain. */
    while (atomic_load(&processed) < FRAMES) usleep(1000);

    pthread_mutex_lock(&m);
    shutting_down = 1;
    pthread_cond_broadcast(&work);      /* shutdown concerns every waiter */
    pthread_mutex_unlock(&m);
    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;
}