basic_pipeline_thread_kvs.c

/* Explanation:
 * Basic thread pipeline KV in C.
 * Stage 1: Producer threads forward operations
 * Stage 2: Worker threads perform actual kv_set / kv_get
 *
 *
 * ----------------------
 *  Output:
 *  Stage 1 : Input / Producer
 *  Stage 2 : KV execution
 *  [T0] Stage 1 started (producer)
 *  [T1] Stage 1 started (producer)
 *  [T2] Stage 2 started (worker)
 *  [T0][Stage1] Forward SET key 0 -> val0
 *  [T3] Stage 2 started (worker)
 *  [T1][Stage1] Forward GET key 0
 *  [T1][Stage1] Forward SET key 2 -> val2
 *  [T1][Stage1] Forward GET key 2
 *  [T1][Stage1] Forward GET key 4
 *  [T1][Stage1] Forward SET key 6 -> val6
 *  [T1][Stage1] Forward GET key 6
 *  [T1] Stage 1 exiting
 *  [T0][Stage1] Forward SET key 4 -> val4
 *  [T0] Stage 1 exiting
 *  [T3][SET] Insert key 0 -> val0
 *  [T3][SET] Insert key 2 -> val2
 *  [T3][GET] Key 2 -> val2
 *  [T3][GET] Key 4 not found
 *  [T2][GET] Key 0 -> val0
 *  [T3][SET] Insert key 6 -> val6
 *  [T3] Stage 2 exiting
 *  [T2][GET] Key 6 -> val6
 *  [T2] Stage 2 exiting
 *
 *  === Pipeline completed ===
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>

#define NUM_BUCKETS 1024
#define VALUE_SIZE 32

#define STAGE1_THREADS 2
#define STAGE2_THREADS 2
#define TOTAL_THREADS (STAGE1_THREADS + STAGE2_THREADS)

#define QUEUE_SIZE 16
#define TOTAL_OPS 8

// ================= KV STORE =================

typedef struct Entry {
    size_t key;
    char value[VALUE_SIZE];
    struct Entry* next;
} Entry;

typedef struct {
    Entry* buckets[NUM_BUCKETS];
    pthread_mutex_t locks[NUM_BUCKETS];
} KVStore;

size_t hash(size_t key) { return key % NUM_BUCKETS; }

KVStore* kvstore_create() {
    KVStore* store = malloc(sizeof(KVStore));
    for (int i = 0; i < NUM_BUCKETS; i++) {
        store->buckets[i] = NULL;
        pthread_mutex_init(&store->locks[i], NULL);
    }
    return store;
}

void kv_set(KVStore* store, size_t key, const char* value, int tid) {
    size_t idx = hash(key);
    pthread_mutex_lock(&store->locks[idx]);
    Entry* e = store->buckets[idx];
    while (e) {
        if (e->key == key) {
            strncpy(e->value, value, VALUE_SIZE);
            printf("[T%d][SET] Update key %zu -> %s\n", tid, key, value);
            pthread_mutex_unlock(&store->locks[idx]);
            return;
        }
        e = e->next;
    }
    e = malloc(sizeof(Entry));
    e->key = key;
    strncpy(e->value, value, VALUE_SIZE);
    e->next = store->buckets[idx];
    store->buckets[idx] = e;
    printf("[T%d][SET] Insert key %zu -> %s\n", tid, key, value);
    pthread_mutex_unlock(&store->locks[idx]);
}

void kv_get(KVStore* store, size_t key, int tid) {
    size_t idx = hash(key);
    pthread_mutex_lock(&store->locks[idx]);
    Entry* e = store->buckets[idx];
    while (e) {
        if (e->key == key) {
            printf("[T%d][GET] Key %zu -> %s\n", tid, key, e->value);
            pthread_mutex_unlock(&store->locks[idx]);
            return;
        }
        e = e->next;
    }
    printf("[T%d][GET] Key %zu not found\n", tid, key);
    pthread_mutex_unlock(&store->locks[idx]);
}

// ================= PIPELINE QUEUE =================

typedef enum { OP_SET, OP_GET, OP_EXIT } OpType;

typedef struct {
    OpType type;
    size_t key;
    char value[VALUE_SIZE];
} Operation;

typedef struct {
    Operation q[QUEUE_SIZE];
    int head, tail;
    pthread_mutex_t lock;
    pthread_cond_t not_empty;
    pthread_cond_t not_full;
} Queue;

Queue* queue_create() {
    Queue* q = malloc(sizeof(Queue));
    q->head = q->tail = 0;
    pthread_mutex_init(&q->lock, NULL);
    pthread_cond_init(&q->not_empty, NULL);
    pthread_cond_init(&q->not_full, NULL);
    return q;
}

void queue_push(Queue* q, Operation op) {
    pthread_mutex_lock(&q->lock);
    while ((q->tail + 1) % QUEUE_SIZE == q->head)
        pthread_cond_wait(&q->not_full, &q->lock);
    q->q[q->tail] = op;
    q->tail = (q->tail + 1) % QUEUE_SIZE;
    pthread_cond_signal(&q->not_empty);
    pthread_mutex_unlock(&q->lock);
}

Operation queue_pop(Queue* q) {
    pthread_mutex_lock(&q->lock);
    while (q->head == q->tail)
        pthread_cond_wait(&q->not_empty, &q->lock);
    Operation op = q->q[q->head];
    q->head = (q->head + 1) % QUEUE_SIZE;
    pthread_cond_signal(&q->not_full);
    pthread_mutex_unlock(&q->lock);
    return op;
}

// ================= STAGES =================

typedef struct {
    Queue* in;
    Queue* out;
    KVStore* store;
    int stage;
    int tid;
} StageArg;

// Stage 1: forward operations
void* stage1_worker(void* arg) {
    StageArg* s = arg;
    printf("[T%d] Stage 1 started (producer)\n", s->tid);

    while (1) {
        Operation op = queue_pop(s->in);
        if (op.type == OP_EXIT) {
            // forward exit to Stage 2
            for (int i = 0; i < STAGE2_THREADS; i++)
                queue_push(s->out, op);
            break;
        }

        // Print the operation
        if (op.type == OP_SET)
            printf("[T%d][Stage1] Forward SET key %zu -> %s\n", s->tid, op.key, op.value);
        else
            printf("[T%d][Stage1] Forward GET key %zu\n", s->tid, op.key);

        queue_push(s->out, op);
    }

    printf("[T%d] Stage 1 exiting\n", s->tid);
    return NULL;
}

// Stage 2: actual KV execution
void* stage2_worker(void* arg) {
    StageArg* s = arg;
    printf("[T%d] Stage 2 started (worker)\n", s->tid);

    while (1) {
        Operation op = queue_pop(s->in);
        if (op.type == OP_EXIT) break;

        if (op.type == OP_SET)
            kv_set(s->store, op.key, op.value, s->tid);
        else
            kv_get(s->store, op.key, s->tid);
    }

    printf("[T%d] Stage 2 exiting\n", s->tid);
    return NULL;
}

// ================= MAIN =================

int main() {
    setvbuf(stdout, NULL, _IONBF, 0); // unbuffered output

    KVStore* store = kvstore_create();
    Queue* q1 = queue_create();
    Queue* q2 = queue_create();

    pthread_t threads[TOTAL_THREADS];
    StageArg args[TOTAL_THREADS];

    printf("Stage 1 : Input / Producer\n");
    printf("Stage 2 : KV execution\n\n");

    // Stage 1 threads
    for (int i = 0; i < STAGE1_THREADS; i++) {
        args[i] = (StageArg){q1, q2, store, 1, i};
        pthread_create(&threads[i], NULL, stage1_worker, &args[i]);
    }

    // Stage 2 threads
    for (int i = 0; i < STAGE2_THREADS; i++) {
        int tid = STAGE1_THREADS + i;
        args[tid] = (StageArg){q2, NULL, store, 2, tid};
        pthread_create(&threads[tid], NULL, stage2_worker, &args[tid]);
    }

    // ---------------- Feed operations ----------------
    for (int i = 0; i < TOTAL_OPS; i++) {
        Operation op;
        if (i % 2 == 0) {
            op.type = OP_SET;
            op.key = i;
            snprintf(op.value, VALUE_SIZE, "val%d", i);
        } else {
            op.type = OP_GET;
            op.key = i - 1;
        }
        queue_push(q1, op);
    }

    // Send exit signal for Stage1 threads
    for (int i = 0; i < STAGE1_THREADS; i++)
        queue_push(q1, (Operation){OP_EXIT});

    // Wait for all threads
    for (int i = 0; i < TOTAL_THREADS; i++)
        pthread_join(threads[i], NULL);

    printf("\n=== Pipeline completed ===\n");
    return 0;
}