/* EXERCISE 04 — REPAIRED.
 *
 * A media app. One latency-sensitive thread refreshes a level meter: it does a
 * small, fixed amount of work, over and over, and must finish each unit before
 * the next frame is due. A background library re-indexes the user's collection
 * on a pool of CPU-bound threads.
 *
 * The repair: the pool declares QOS_CLASS_BACKGROUND, as its first act on each
 * thread. That is not a speed dial. It is a PLACEMENT decision: it moves the
 * pool into a lower scheduling band so it stops taking turns against the work
 * the user can see. The re-index still finishes; it simply yields the tail.
 *
 * Build and run:
 *   clang -O2 -g -Wall -Wextra -pthread heartbeat.c -o /tmp/heartbeat_fixed
 *   /tmp/heartbeat_fixed
 *
 * Read hogBatches as well as the latency figures: this repair buys tail
 * latency by giving the pool less CPU, and pretending otherwise would be
 * dishonest. That trade is the answer to "what does your fix cost?".
 *
 * The fixture measures itself twice — once with the pool idle, to establish
 * this machine's floor, and once with the pool running — so every number it
 * reports is a RATIO against its own baseline rather than a figure from
 * somebody else's laptop.
 *
 * 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/time.h>
#include <pthread/qos.h>

#define UNITS        400        /* work units the meter thread times    */
#define UNIT_LOOPS   200000     /* size of one unit                     */
#define HOG_MULT     4          /* CPU hogs per core                    */
#define SETTLE_US    300000     /* let the pool reach steady state      */
#define WATCHDOG_S   180

static double now_us(void) {
    struct timeval t; gettimeofday(&t, NULL);
    return t.tv_sec * 1e6 + t.tv_usec;
}

/* ------------------------------------------------------------------ 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);
}

static atomic_int stop_hogs = 0;
static atomic_ullong hog_batches = 0;

/* The re-indexing pool. Each thread declares its own quality of service as
 * its first act, so the scheduler knows what this work is worth before it
 * places the thread. QoS is per-thread: setting it on the spawning thread
 * would not have travelled here. */
static void *hog(void *unused) {
    (void)unused;
    pthread_set_qos_class_self_np(QOS_CLASS_BACKGROUND, 0);
    unsigned long long n = 0;
    volatile double x = 0;
    while (!atomic_load_explicit(&stop_hogs, memory_order_relaxed)) {
        for (int i = 0; i < 100000; i++) x += 1.0000001;
        n++;
    }
    atomic_fetch_add(&hog_batches, n);
    return NULL;
}

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);
}

/* Time UNITS identical work units on THIS thread. Returns p50/p99/max in us. */
static void time_units(double *p50, double *p99, double *worst) {
    double *d = calloc(UNITS, sizeof *d);
    volatile double x = 0;
    for (int i = 0; i < UNITS; i++) {
        double a = now_us();
        for (int k = 0; k < UNIT_LOOPS; k++) x += 1.0000001;
        d[i] = now_us() - a;
    }
    qsort(d, UNITS, sizeof *d, cmp_double);
    *p50   = d[UNITS / 2];
    *p99   = d[(UNITS * 99) / 100];
    *worst = d[UNITS - 1];
    free(d);
}

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

    int ncpu  = (int)sysconf(_SC_NPROCESSORS_ONLN);
    int nhogs = ncpu * HOG_MULT;

    /* The meter thread is this one. Asking for the highest ordinary class is
     * necessary but not sufficient: it only means something once the work it
     * competes with has declared something lower. */
    pthread_set_qos_class_self_np(QOS_CLASS_USER_INTERACTIVE, 0);

    double base50, base99, baseMax;
    time_units(&base50, &base99, &baseMax);

    pthread_t *h = calloc((size_t)nhogs, sizeof *h);
    for (int i = 0; i < nhogs; i++) pthread_create(&h[i], NULL, hog, NULL);
    usleep(SETTLE_US);

    double load50, load99, loadMax;
    time_units(&load50, &load99, &loadMax);

    atomic_store(&stop_hogs, 1);
    for (int i = 0; i < nhogs; i++) pthread_join(h[i], NULL);

    printf("cores=%d\n", ncpu);
    printf("hogThreads=%d\n", nhogs);
    printf("hogQos=background\n");
    printf("units=%d\n", UNITS);
    printf("idleP50Us=%.1f\n", base50);
    printf("idleP99Us=%.1f\n", base99);
    printf("loadedP50Us=%.1f\n", load50);
    printf("loadedP99Us=%.1f\n", load99);
    printf("loadedMaxUs=%.1f\n", loadMax);
    printf("p50Inflation=%.2f\n", base50 > 0 ? load50 / base50 : 0.0);
    printf("p99Inflation=%.2f\n", base99 > 0 ? load99 / base99 : 0.0);
    printf("maxInflation=%.2f\n", baseMax > 0 ? loadMax / baseMax : 0.0);
    printf("hogBatches=%llu\n", (unsigned long long)atomic_load(&hog_batches));
    free(h);
    return 0;
}