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

הקדמה

בפרק הקודם (12.1) למדנו דפוסי תכנות מקבילי שמבוססים על נעילות - mutex, rwlock, barrier. הנעילות עובדות מצוין ברוב המקרים, אבל יש להן בעיות:

• מתחרות (contention) - כשהרבה threads מנסים לנעול את אותו mutex, רובם חוסמים וממתינים. במקום לעבוד, הם ישנים.
• היפוך עדיפויות (priority inversion) - thread בעדיפות נמוכה מחזיק נעילה שthread בעדיפות גבוהה צריך. התוצאה: הthread החשוב מחכה לthread הפחות חשוב.
• קיפאון (deadlock) - שני threads מחכים אחד לשני. אף אחד לא מתקדם. לעולם.

אז האם אפשר לעשות תכנות מקבילי בלי נעילות בכלל? התשובה: כן, אבל זה הרבה יותר קשה. בואו נראה איך.

---

השוואה-והחלפה - compare-and-swap (CAS)

אבן הבניין הבסיסית

כמו שלמדנו בפרק 10.4, המעבד תומך בפעולות אטומיות - פעולות שמתבצעות "בבת אחת" מבחינת כל השאר. הפעולה האטומית הכי חשובה לתכנות ללא נעילה היא CAS - השוואה והחלפה (Compare-And-Swap).

הרעיון:

CAS(address, expected, desired):
    if *address == expected:
        *address = desired
        return true
    else:
        expected = *address    // update expected to the current value
        return false

כל הפעולה הזו מתבצעת באופן אטומי - אף thread אחר לא יכול לגשת ל-address באמצע. ב-x86 זו הוראת cmpxchg אחת.

בC11

#include <stdatomic.h>

_Atomic int counter = 0;

// CAS - returns true if we succeeded, false if someone changed it before us
int expected = 5;
int desired = 6;
bool success = atomic_compare_exchange_strong(&counter, &expected, desired);
// if counter was 5: now counter=6, success=true
// if counter wasn't 5: expected is updated to the current value, success=false

הדפוס הבסיסי - לולאת CAS

הדפוס הנפוץ ביותר הוא לולאת retry:

void atomic_add(_Atomic int *val, int amount) {
    int old = atomic_load(val);
    while (!atomic_compare_exchange_weak(val, &old, old + amount)) {
        // CAS failed - old was updated to the current value, try again
    }
}

למה weak ולא strong? ההבדל: weak יכול להחזיר false גם כשהערך לא השתנה (spurious failure), אבל הוא יותר מהיר בלולאה. strong מובטח שנכשל רק כשהערך באמת השתנה.

---

מחסנית ללא נעילה - Treiber stack

הרעיון

מחסנית מקושרת (linked list stack) שמאפשרת push ו-pop מקביליים בלי שום mutex. הרעיון של Treiber (1986):

• push: צור צומת חדש, הצבע אותו לראש הנוכחי, ואז החלף את הראש לצומת החדש עם CAS.
• pop: קרא את הראש, והחלף אותו לnext שלו עם CAS.

אם ה-CAS נכשל (מישהו שינה את הראש בינתיים) - פשוט ננסה שוב.

מימוש

#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;
        // try to swap top for new_node
        // if top is still == old_top, it will succeed
        // if someone changed top, old_top will be updated and we try again
    } while (!atomic_compare_exchange_weak(&s->top, &old_top, new_node));
}

// returns -1 if the stack is empty
int stack_pop(treiber_stack_t *s) {
    node_t *old_top = atomic_load(&s->top);
    do {
        if (old_top == NULL) {
            return -1;  // empty
        }
        // try to swap top for old_top->next
    } while (!atomic_compare_exchange_weak(&s->top, &old_top, old_top->next));

    int value = old_top->data;
    free(old_top);
    return value;
}

בדיקה

#define NUM_THREADS 4
#define OPS_PER_THREAD 10000

treiber_stack_t stack;

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

void *pop_thread(void *arg) {
    int count = 0;
    for (int i = 0; i < OPS_PER_THREAD; i++) {
        if (stack_pop(&stack) != -1) {
            count++;
        }
    }
    printf("popped %d items\n", count);
    return NULL;
}

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

    pthread_t threads[NUM_THREADS];
    int ids[NUM_THREADS];

    // half push, half pop
    for (int i = 0; i < NUM_THREADS / 2; i++) {
        ids[i] = i;
        pthread_create(&threads[i], NULL, push_thread, &ids[i]);
    }
    for (int i = NUM_THREADS / 2; i < NUM_THREADS; i++) {
        ids[i] = i;
        pthread_create(&threads[i], NULL, pop_thread, &ids[i]);
    }

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

    return 0;
}

קומפילציה:

gcc -o treiber treiber.c -lpthread -latomic

---

בעיית ABA

הבעיה

המחסנית של Treiber סובלת מבעיה מפורסמת בשם ABA. הנה התרחיש:

1. הthread 1 רוצה לעשות pop. הוא קורא top = A (שמצביע ל-B). הוא מתכנן לעשות CAS(&top, A, B).
2. הthread 1 נכנס לשינה (preempted) לפני שה-CAS מתבצע.
3. הthread 2 עושה pop של A. עכשיו top = B.
4. הthread 2 עושה pop של B. עכשיו top = C.
5. הthread 2 עושה push של A בחזרה. עכשיו top = A, ו-A מצביע ל-C.
6. הthread 1 מתעורר. הוא עושה CAS(&top, A, B) - וזה מצליח כי top אכן שווה A! אבל B כבר שוחרר, וזה גישה לזכרון לא חוקי.

הבעיה: ה-CAS בודק רק אם הערך שווה, לא אם הוא אותו אחד. הערך חזר להיות A, אבל כל מבנה הנתונים השתנה.

פתרון - מצביעים מתויגים (tagged pointers)

הרעיון: נוסיף מונה לכל מצביע. כל פעם שמשנים את הראש, מעלים את המונה. גם אם המצביע חוזר לאותו ערך, המונה יהיה שונה וה-CAS ייכשל.

#include <stdatomic.h>
#include <stdint.h>

// we'll use 128 bits: 64 bits pointer + 64 bits counter
typedef struct {
    node_t *ptr;
    uint64_t tag;
} tagged_ptr_t;

// we'll need a CAS on 128 bits (on x86_64 this is cmpxchg16b)
typedef _Atomic __int128 atomic_tagged_ptr_t;

// conversion
static inline __int128 pack(node_t *ptr, uint64_t tag) {
    __int128 result = ((__int128)tag << 64) | (uint64_t)ptr;
    return result;
}

static inline node_t *unpack_ptr(__int128 val) {
    return (node_t *)(uint64_t)val;
}

static inline uint64_t unpack_tag(__int128 val) {
    return (uint64_t)(val >> 64);
}

בפועל, הרבה מימושים משתמשים בביטים העליונים של מצביע 64 ביט (שלא בשימוש במרחב כתובות של 48 ביט) כמונה, כדי להימנע מ-128 ביט CAS.

הערה: בעיית ABA היא בעיקר בעיה כשמשחררים ומקצים מחדש זכרון. אם משתמשים בפתרונות כמו hazard pointers או epoch-based reclamation (מנגנונים שמבטיחים שלא משחררים צומת בזמן שמישהו עוד מחזיק מצביע אליו), אפשר להימנע מהבעיה גם בלי תגיות.

---

תור SPSC ללא נעילה - single-producer single-consumer lock-free queue

למה SPSC מיוחד

כשיש רק יצרן אחד וצרכן אחד, הדברים הרבה יותר פשוטים:
- רק היצרן כותב ל-head (אינדקס הכתיבה).
- רק הצרכן כותב ל-tail (אינדקס הקריאה).
- לא צריך CAS בכלל! מספיקים loads/stores אטומיים עם סדר זכרון מתאים.

זה הסוג הכי מהיר של queue מקבילי, ומשתמשים בו ב:
- מערכות אודיו (thread אחד מקליט, אחד מנגן)
- הbuffers רשתיים (כרטיס רשת כותב, תוכנה קוראת)
- תקשורת בין threads בתוכנה (inter-thread communication)

מימוש - ring buffer

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

#define QUEUE_SIZE 1024  // must be a power of 2

typedef struct {
    int buffer[QUEUE_SIZE];
    // head - next to write (only the producer writes)
    // tail - next to read (only the consumer writes)
    _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);

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

    q->buffer[head & (QUEUE_SIZE - 1)] = item;  // head % QUEUE_SIZE

    // publish the new item - the consumer will see it after the release
    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);

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

    *item = q->buffer[tail & (QUEUE_SIZE - 1)];  // tail % QUEUE_SIZE

    // release the slot - the producer will see it after the release
    atomic_store_explicit(&q->tail, tail + 1, memory_order_release);
    return true;
}

שימו לב: אין כאן mutex, אין CAS, אין חסימה. רק load ו-store אטומיים עם סדר acquire/release. זה הסוד של הביצועים הגבוהים.

למה acquire/release

חזרה קצרה לחומר מפרק 10.4 - סדר זכרון (memory ordering):
- memory_order_release בstore: כל הכתיבות שלפניו מובטחות להיות גלויות לפני שהstore הזה נראה.
- memory_order_acquire בload: כל הקריאות שאחריו מובטחות לראות את מה שהיה לפני הrelease התואם.

בqueue שלנו:
- היצרן כותב לbuffer, ואז עושה release store ל-head. כך מובטח שהצרכן יראה את הנתון בbuffer לפני שהוא רואה את הhead החדש.
- הצרכן קורא מהbuffer, ואז עושה release store ל-tail. כך מובטח שהיצרן יראה שהמקום התפנה.

דוגמה שלמה

spsc_queue_t queue;

void *producer(void *arg) {
    for (int i = 0; i < 1000000; i++) {
        while (!spsc_push(&queue, i)) {
            // the queue is full, wait
            // in a real environment: sched_yield() or pause
        }
    }
    return NULL;
}

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

    while (count < 1000000) {
        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);

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

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

    return 0;
}

---

תור MPSC - ריבוי יצרנים, צרכן יחיד - multi-producer single-consumer

למה MPSC

דפוס נפוץ מאוד: כמה threads עובדים שולחים הודעות לthread מרכזי. למשל:
- מערכת לוגים: כל הthreads כותבים, thread אחד כותב לקובץ.
- תור אירועים (event queue): מקורות שונים שולחים אירועים, event loop אחד מעבד אותם.

מימוש עם CAS

ההבדל מ-SPSC: עכשיו כמה יצרנים מתחרים על head, אז צריך CAS:

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

typedef struct mpsc_node {
    int data;
    _Atomic(struct mpsc_node *) next;
} mpsc_node_t;

typedef struct {
    _Atomic(mpsc_node_t *) head;  // producers add here
    mpsc_node_t *tail;            // the consumer reads from here (doesn't need to be atomic)
    mpsc_node_t stub;             // permanent dummy node
} mpsc_queue_t;

void mpsc_init(mpsc_queue_t *q) {
    atomic_store(&q->stub.next, NULL);
    atomic_store(&q->head, &q->stub);
    q->tail = &q->stub;
}

// producer - several producers can call this in parallel
void mpsc_push(mpsc_queue_t *q, mpsc_node_t *node) {
    atomic_store_explicit(&node->next, NULL, memory_order_relaxed);

    // swap head for the new node, get the old head
    mpsc_node_t *prev = atomic_exchange_explicit(&q->head, node,
                                                  memory_order_acq_rel);

    // link the previous node to the new child
    atomic_store_explicit(&prev->next, node, memory_order_release);
}

// consumer - only one consumer!
mpsc_node_t *mpsc_pop(mpsc_queue_t *q) {
    mpsc_node_t *tail = q->tail;
    mpsc_node_t *next = atomic_load_explicit(&tail->next, memory_order_acquire);

    if (tail == &q->stub) {
        if (next == NULL) {
            return NULL;  // empty
        }
        q->tail = next;
        tail = next;
        next = atomic_load_explicit(&tail->next, memory_order_acquire);
    }

    if (next != NULL) {
        q->tail = next;
        return tail;
    }

    return NULL;  // empty, or a producer is mid-push
}

---

רמות התקדמות - progress guarantees

הגדרות

יש שלוש רמות של ערבויות התקדמות במבני נתונים מקביליים:

ללא נעילה - lock-free:
- לפחות thread אחד מתקדם בכל רגע.
- אם thread אחד נתקע (נכנס לשינה, מת), השאר ממשיכים.
- אין deadlock. אבל thread בודד עלול להירעב (starvation).
- דוגמה: מחסנית Treiber - אם CAS נכשל, זה אומר שthread אחר הצליח.

ללא חסימה - wait-free:
- כל thread מתקדם בזמן חסום (bounded).
- הערבות הכי חזקה: אף thread לא יכול להירעב.
- הרבה יותר קשה לממש. מבני נתונים wait-free מורכבים מאוד.
- דוגמה: הqueue SPSC - ה-push וה-pop תמיד מסתיימים בזמן חסום.

ללא חסימה חלקי - obstruction-free:
- הthread מתקדם אם הוא רץ לבד (בלי מתחרים).
- הערבות הכי חלשה: שני threads יכולים "לדרוך" אחד על השני שוב ושוב (livelock).
- דוגמה: אלגוריתם שעושה retry אבל בלי backoff - שני threads עלולים לנסות ולהיכשל שוב ושוב.

סיכום

רמה	ערבות	הthread יחיד	ביצועים
ללא חסימה חלקי (obstruction-free)	התקדמות רק בלי מתחרים	כן	נמוכים
ללא נעילה (lock-free)	לפחות אחד מתקדם	כן	טובים
ללא חסימה (wait-free)	כולם מתקדמים תמיד	כן	תלוי במימוש

---

מתי להשתמש בlock-free

מתי כן

• נתיבים חמים מאוד - קוד שרץ מיליוני פעמים בשנייה ו-mutex contention הוא צוואר בקבוק.
• מערכות זמן אמת - כשאי אפשר להרשות לthread להיחסם לזמן לא ידוע.
• מתחרות גבוהה - כשהרבה threads מתחרים על אותו משאב ו-mutex גורם לירידת ביצועים.
• הbuffers בין שכבות - queue SPSC בין thread שמקליט אודיו לthread שמעבד אותו.

מתי לא

• ברוב המקרים - mutex מודרני (futex בלינוקס) הוא מהיר מאוד כשאין מתחרות. ה-fast path (נעילה בלי מתחרים) הוא פעולה אטומית אחת בuser-space.
• כשהנכונות חשובה יותר מביצועים - מבני נתונים lock-free קשים מאוד לכתיבה נכונה ולדיבוג.
• כשאין מדידה - אל תעברו ל-lock-free בלי למדוד קודם שה-mutex הוא באמת הבעיה.

כלל אצבע: התחילו עם mutex. מדדו. אם יש בעיית contention ספציפית, שקלו lock-free רק לאותו מקום.