12.2 - מבני נתונים ללא נעילה - פתרון

פתרון לתרגיל 1 - מונה אטומי עם CAS

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

typedef struct {
    _Atomic int value;
} atomic_counter_t;

void counter_init(atomic_counter_t *c, int initial) {
    atomic_store(&c->value, initial);
}

void counter_increment(atomic_counter_t *c) {
    int old = atomic_load(&c->value);
    while (!atomic_compare_exchange_weak(&c->value, &old, old + 1)) {
        // old is automatically updated to the current value, try again
    }
}

void counter_add(atomic_counter_t *c, int amount) {
    int old = atomic_load(&c->value);
    while (!atomic_compare_exchange_weak(&c->value, &old, old + amount)) {
        // retry
    }
}

int counter_get(atomic_counter_t *c) {
    return atomic_load(&c->value);
}

#define NUM_THREADS 8
#define OPS_PER_THREAD 100000

atomic_counter_t counter;

void *increment_thread(void *arg) {
    (void)arg;
    for (int i = 0; i < OPS_PER_THREAD; i++) {
        counter_increment(&counter);
    }
    return NULL;
}

int main(void) {
    counter_init(&counter, 0);

    pthread_t threads[NUM_THREADS];
    for (int i = 0; i < NUM_THREADS; i++) {
        pthread_create(&threads[i], NULL, increment_thread, NULL);
    }
    for (int i = 0; i < NUM_THREADS; i++) {
        pthread_join(threads[i], NULL);
    }

    int result = counter_get(&counter);
    printf("expected: %d\n", NUM_THREADS * OPS_PER_THREAD);
    printf("got:      %d\n", result);
    printf("correct:  %s\n", result == NUM_THREADS * OPS_PER_THREAD ? "yes" : "NO!");

    return 0;
}

קומפילציה:

gcc -O2 -o counter counter.c -lpthread -latomic

---

פתרון לתרגיל 2 - מחסנית Treiber

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

typedef struct node {
    int data;
    struct node *next;
} node_t;

typedef struct {
    _Atomic(node_t *) top;
} treiber_stack_t;

void stack_init(treiber_stack_t *s) {
    atomic_store(&s->top, NULL);
}

void stack_push(treiber_stack_t *s, int value) {
    node_t *new_node = malloc(sizeof(node_t));
    new_node->data = value;

    node_t *old_top = atomic_load(&s->top);
    do {
        new_node->next = old_top;
    } while (!atomic_compare_exchange_weak(&s->top, &old_top, new_node));
}

// returns 1 if successful, 0 if empty
int stack_pop(treiber_stack_t *s, int *value) {
    node_t *old_top = atomic_load(&s->top);
    do {
        if (old_top == NULL) {
            return 0;
        }
    } while (!atomic_compare_exchange_weak(&s->top, &old_top, old_top->next));

    *value = old_top->data;
    // note: in a real program, free here is dangerous because of ABA
    // for the purpose of this exercise it's fine because our pop threads don't interleave with push
    free(old_top);
    return 1;
}

#define PUSH_THREADS 4
#define POP_THREADS 4
#define OPS_PER_THREAD 10000

treiber_stack_t stack;
_Atomic int total_popped = 0;

void *push_thread(void *arg) {
    int id = *(int *)arg;
    for (int i = 0; i < OPS_PER_THREAD; i++) {
        stack_push(&stack, id * OPS_PER_THREAD + i);
    }
    printf("push thread %d: pushed %d items\n", id, OPS_PER_THREAD);
    return NULL;
}

void *pop_thread(void *arg) {
    int id = *(int *)arg;
    int count = 0;
    int value;

    // try to pop until we succeed OPS_PER_THREAD times
    while (count < OPS_PER_THREAD) {
        if (stack_pop(&stack, &value)) {
            count++;
        }
    }

    atomic_fetch_add(&total_popped, count);
    printf("pop thread %d: popped %d items\n", id, count);
    return NULL;
}

int main(void) {
    stack_init(&stack);

    pthread_t push_threads[PUSH_THREADS], pop_threads[POP_THREADS];
    int push_ids[PUSH_THREADS], pop_ids[POP_THREADS];

    // run the push threads first so there's something to read
    for (int i = 0; i < PUSH_THREADS; i++) {
        push_ids[i] = i;
        pthread_create(&push_threads[i], NULL, push_thread, &push_ids[i]);
    }
    for (int i = 0; i < PUSH_THREADS; i++) {
        pthread_join(push_threads[i], NULL);
    }

    printf("total pushed: %d\n", PUSH_THREADS * OPS_PER_THREAD);

    // now the pop threads
    for (int i = 0; i < POP_THREADS; i++) {
        pop_ids[i] = i;
        pthread_create(&pop_threads[i], NULL, pop_thread, &pop_ids[i]);
    }
    for (int i = 0; i < POP_THREADS; i++) {
        pthread_join(pop_threads[i], NULL);
    }

    printf("total popped: %d\n", atomic_load(&total_popped));

    // count what remains
    int remaining = 0;
    int value;
    while (stack_pop(&stack, &value)) {
        remaining++;
    }
    printf("remaining in stack: %d\n", remaining);
    printf("total (popped + remaining): %d\n",
           atomic_load(&total_popped) + remaining);

    return 0;
}

---

פתרון לתרגיל 3 - queue SPSC

#include <stdio.h>
#include <stdlib.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <pthread.h>
#include <time.h>

#define QUEUE_SIZE 4096  // a power of 2
#define NUM_ITEMS 10000000

typedef struct {
    int buffer[QUEUE_SIZE];
    _Atomic int head;
    _Atomic int tail;
} spsc_queue_t;

void spsc_init(spsc_queue_t *q) {
    atomic_store(&q->head, 0);
    atomic_store(&q->tail, 0);
}

bool spsc_push(spsc_queue_t *q, int item) {
    int head = atomic_load_explicit(&q->head, memory_order_relaxed);
    int tail = atomic_load_explicit(&q->tail, memory_order_acquire);

    if ((head - tail) >= QUEUE_SIZE) {
        return false;
    }

    q->buffer[head & (QUEUE_SIZE - 1)] = item;
    atomic_store_explicit(&q->head, head + 1, memory_order_release);
    return true;
}

bool spsc_pop(spsc_queue_t *q, int *item) {
    int tail = atomic_load_explicit(&q->tail, memory_order_relaxed);
    int head = atomic_load_explicit(&q->head, memory_order_acquire);

    if (tail >= head) {
        return false;
    }

    *item = q->buffer[tail & (QUEUE_SIZE - 1)];
    atomic_store_explicit(&q->tail, tail + 1, memory_order_release);
    return true;
}

spsc_queue_t queue;

void *producer(void *arg) {
    (void)arg;
    for (int i = 0; i < NUM_ITEMS; i++) {
        while (!spsc_push(&queue, i)) {
            // spin - the queue is full
        }
    }
    return NULL;
}

void *consumer(void *arg) {
    (void)arg;
    int item;
    long sum = 0;
    int count = 0;

    while (count < NUM_ITEMS) {
        if (spsc_pop(&queue, &item)) {
            sum += item;
            count++;
        }
    }

    printf("consumed %d items, sum = %ld\n", count, sum);
    return NULL;
}

int main(void) {
    spsc_init(&queue);

    struct timespec start, end;
    clock_gettime(CLOCK_MONOTONIC, &start);

    pthread_t prod, cons;
    pthread_create(&prod, NULL, producer, NULL);
    pthread_create(&cons, NULL, consumer, NULL);

    pthread_join(prod, NULL);
    pthread_join(cons, NULL);

    clock_gettime(CLOCK_MONOTONIC, &end);

    double elapsed = (end.tv_sec - start.tv_sec) +
                     (end.tv_nsec - start.tv_nsec) / 1e9;
    double ops_per_sec = NUM_ITEMS / elapsed;

    printf("elapsed: %.4f seconds\n", elapsed);
    printf("throughput: %.0f ops/sec\n", ops_per_sec);

    return 0;
}

---

פתרון לתרגיל 4 - השוואת ביצועים

#include <stdio.h>
#include <stdlib.h>
#include <stdatomic.h>
#include <pthread.h>
#include <time.h>

#define NUM_THREADS 4
#define OPS_PER_THREAD 100000

// --- mutex version ---

typedef struct mutex_node {
    int data;
    struct mutex_node *next;
} mutex_node_t;

typedef struct {
    mutex_node_t *top;
    pthread_mutex_t lock;
} mutex_stack_t;

void mutex_stack_init(mutex_stack_t *s) {
    s->top = NULL;
    pthread_mutex_init(&s->lock, NULL);
}

void mutex_stack_push(mutex_stack_t *s, int val) {
    mutex_node_t *n = malloc(sizeof(mutex_node_t));
    n->data = val;
    pthread_mutex_lock(&s->lock);
    n->next = s->top;
    s->top = n;
    pthread_mutex_unlock(&s->lock);
}

int mutex_stack_pop(mutex_stack_t *s, int *val) {
    pthread_mutex_lock(&s->lock);
    if (s->top == NULL) {
        pthread_mutex_unlock(&s->lock);
        return 0;
    }
    mutex_node_t *n = s->top;
    s->top = n->next;
    pthread_mutex_unlock(&s->lock);
    *val = n->data;
    free(n);
    return 1;
}

// --- lock-free version (Treiber) ---

typedef struct lf_node {
    int data;
    struct lf_node *next;
} lf_node_t;

typedef struct {
    _Atomic(lf_node_t *) top;
} lf_stack_t;

void lf_stack_init(lf_stack_t *s) {
    atomic_store(&s->top, NULL);
}

void lf_stack_push(lf_stack_t *s, int val) {
    lf_node_t *n = malloc(sizeof(lf_node_t));
    n->data = val;
    lf_node_t *old = atomic_load(&s->top);
    do {
        n->next = old;
    } while (!atomic_compare_exchange_weak(&s->top, &old, n));
}

int lf_stack_pop(lf_stack_t *s, int *val) {
    lf_node_t *old = atomic_load(&s->top);
    do {
        if (old == NULL) return 0;
    } while (!atomic_compare_exchange_weak(&s->top, &old, old->next));
    *val = old->data;
    free(old);
    return 1;
}

// --- benchmark ---

mutex_stack_t ms;
lf_stack_t ls;

void *mutex_worker(void *arg) {
    (void)arg;
    for (int i = 0; i < OPS_PER_THREAD; i++) {
        mutex_stack_push(&ms, i);
    }
    int val;
    for (int i = 0; i < OPS_PER_THREAD; i++) {
        while (!mutex_stack_pop(&ms, &val)) {}
    }
    return NULL;
}

void *lf_worker(void *arg) {
    (void)arg;
    for (int i = 0; i < OPS_PER_THREAD; i++) {
        lf_stack_push(&ls, i);
    }
    int val;
    for (int i = 0; i < OPS_PER_THREAD; i++) {
        while (!lf_stack_pop(&ls, &val)) {}
    }
    return NULL;
}

double run_benchmark(void *(*func)(void *), const char *name) {
    struct timespec start, end;
    pthread_t threads[NUM_THREADS];

    clock_gettime(CLOCK_MONOTONIC, &start);

    for (int i = 0; i < NUM_THREADS; i++) {
        pthread_create(&threads[i], NULL, func, NULL);
    }
    for (int i = 0; i < NUM_THREADS; i++) {
        pthread_join(threads[i], NULL);
    }

    clock_gettime(CLOCK_MONOTONIC, &end);

    double elapsed = (end.tv_sec - start.tv_sec) +
                     (end.tv_nsec - start.tv_nsec) / 1e9;
    printf("%s: %.4f seconds\n", name, elapsed);
    return elapsed;
}

int main(void) {
    mutex_stack_init(&ms);
    lf_stack_init(&ls);

    double mutex_time = run_benchmark(mutex_worker, "mutex stack");
    double lf_time = run_benchmark(lf_worker, "lock-free stack");

    printf("lock-free is %.2fx %s than mutex\n",
           mutex_time > lf_time ? mutex_time / lf_time : lf_time / mutex_time,
           mutex_time > lf_time ? "faster" : "slower");

    /*
     * note: the results depend heavily on the chip and the number of cores.
     * with high contention (many threads), lock-free usually wins.
     * with low contention (few threads), mutex is sometimes faster
     * because futex on the fast path is a single atomic operation in user-space.
     * the conclusion: always measure before deciding.
     */

    return 0;
}

---

פתרון לתרגיל 5 - זיהוי בעיית ABA

#include <stdio.h>
#include <stdlib.h>
#include <stdatomic.h>
#include <pthread.h>
#include <unistd.h>

typedef struct node {
    char name;            // name for identification
    struct node *next;
} node_t;

typedef struct {
    _Atomic(node_t *) top;
} stack_t;

node_t node_A, node_B, node_C;
stack_t stack;

void print_stack(const char *label) {
    printf("[%s] stack: ", label);
    node_t *cur = atomic_load(&stack.top);
    while (cur) {
        printf("%c -> ", cur->name);
        cur = cur->next;
    }
    printf("NULL\n");
}

// thread 1: tries to pop, but stops in the middle
void *thread1(void *arg) {
    (void)arg;

    // step 1: read top and next
    node_t *old_top = atomic_load(&stack.top);
    node_t *new_top = old_top->next;

    printf("thread 1: read top=%c, next=%c\n", old_top->name, new_top->name);
    printf("thread 1: going to sleep before CAS...\n");

    // simulate preemption
    usleep(100000);

    printf("thread 1: woke up, attempting CAS(&top, %c, %c)\n",
           old_top->name, new_top->name);

    // the CAS will succeed because top still points to C (after thread2 put it back)
    // but new_top (=B) is no longer part of the stack!
    if (atomic_compare_exchange_strong(&stack.top, &old_top, new_top)) {
        printf("thread 1: CAS succeeded! top is now '%c'\n", new_top->name);
        printf("thread 1: BUT '%c' was already removed! THIS IS THE ABA BUG!\n",
               new_top->name);
    } else {
        printf("thread 1: CAS failed (this would happen with tagged pointers)\n");
    }

    print_stack("after thread 1");
    return NULL;
}

// thread 2: pops C, pops B, pushes C - causes ABA
void *thread2(void *arg) {
    (void)arg;

    usleep(10000);  // give thread1 time to read top first

    printf("thread 2: popping C\n");
    node_t *old = atomic_load(&stack.top);
    atomic_compare_exchange_strong(&stack.top, &old, old->next);
    print_stack("after pop C");

    printf("thread 2: popping B\n");
    old = atomic_load(&stack.top);
    atomic_compare_exchange_strong(&stack.top, &old, old->next);
    print_stack("after pop B");

    printf("thread 2: pushing C back\n");
    node_C.next = atomic_load(&stack.top);
    node_t *expected = node_C.next;
    atomic_compare_exchange_strong(&stack.top, &expected, &node_C);
    print_stack("after push C");

    return NULL;
}

int main(void) {
    // build the stack: top -> C -> B -> A -> NULL
    node_A.name = 'A';
    node_A.next = NULL;
    node_B.name = 'B';
    node_B.next = &node_A;
    node_C.name = 'C';
    node_C.next = &node_B;
    atomic_store(&stack.top, &node_C);

    print_stack("initial");

    pthread_t t1, t2;
    pthread_create(&t1, NULL, thread1, NULL);
    pthread_create(&t2, NULL, thread2, NULL);

    pthread_join(t1, NULL);
    pthread_join(t2, NULL);

    /*
     * the solution to the ABA problem: tagged pointers.
     * every time top changes, a counter (tag) is incremented.
     * the CAS checks both the pointer and the counter.
     * even if the pointer returns to the same value (C), the counter is different -> CAS fails.
     *
     * implementation:
     * typedef struct {
     *     node_t *ptr;
     *     uint64_t tag;
     * } tagged_ptr_t;
     *
     * in a CAS on 128 bits (cmpxchg16b on x86_64), you need to check
     * both ptr and tag. if tag changed - CAS fails, and the problem is avoided.
     */

    return 0;
}