12.1 - דפוסי תכנות מקבילי - הרצאה

הקדמה

בפרק 5.8 למדנו על תהליכונים (threads) ועל mutex כדי להגן על משאבים משותפים. בפרק 6.10 ראינו איך הקרנל עצמו מסנכרן גישה למשאבים. עכשיו הגיע הזמן ללמוד את הדפוסים - patterns - שהופכים תכנות מקבילי ממשהו מפחיד למשהו מובנה ובטוח.

הרעיון פשוט: במקום להמציא כל פעם מחדש את הגלגל, יש דפוסים מוכחים שפותרים בעיות נפוצות. בואו נכיר אותם.

---

דפוס יצרן-צרכן - producer-consumer

הרעיון

אחד הדפוסים הכי נפוצים בתכנות מקבילי. הרעיון הוא כזה:
- הthread אחד (או יותר) מייצר נתונים - זה היצרן (producer).
- הthread אחד (או יותר) צורך את הנתונים - זה הצרכן (consumer).
- ביניהם יש buffer משותף - תור (queue) בגודל מוגבל.

דוגמאות מהעולם האמיתי:
- שרת וב: הthread מקבל חיבורים חדשים (יצרן) וthreads אחרים מטפלים בבקשות (צרכנים).
- עיבוד לוגים: הthread קורא שורות מקובץ (יצרן) וthreads אחרים מנתחים אותן (צרכנים).
- צינור עיבוד: כל שלב הוא גם צרכן (של השלב הקודם) וגם יצרן (של השלב הבא).

הבעיה

נניח שיש לנו queue משותף. צריך לפתור שתי בעיות:
1. סנכרון גישה - שני threads לא יכולים לגעת בqueue באותו רגע (כבר מוכר לנו - mutex).
2. תיאום - מה קורה כשהqueue מלא? היצרן צריך לחכות. מה קורה כשהqueue ריק? הצרכן צריך לחכות.

אפשר לפתור את בעיה 2 עם busy-waiting - לולאה שבודקת שוב ושוב אם הqueue התפנה. אבל זה בזבוז מטורף של CPU. צריך מנגנון שנותן לthread לישון עד שמשהו משתנה.

---

משתני תנאי - condition variables

למה mutex לבד לא מספיק

כשthread צריך לחכות עד שתנאי מסוים מתקיים (למשל "הqueue כבר לא ריק"), אנחנו צריכים שהוא ילך לישון ויתעורר כשהתנאי מתקיים. mutex לבד לא עושה את זה - הוא רק מגן על גישה, לא מאותת על שינוי מצב.

הרעיון

משתנה תנאי (condition variable) נותן לנו שתי פעולות:
- המתנה (pthread_cond_wait) - שחרר את הmutex, לך לישון, וכשמישהו יעיר אותך - תחזור להחזיק את הmutex.
- איתות (pthread_cond_signal / pthread_cond_broadcast) - העיר thread אחד (signal) או את כולם (broadcast) שישנים על התנאי הזה.

הAPI

#include <pthread.h>

pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

// Wait - must hold the mutex before calling!
pthread_cond_wait(&cond, &mutex);

// Wake up one thread
pthread_cond_signal(&cond);

// Wake up all the threads
pthread_cond_broadcast(&cond);

נקודה קריטית - לולאת while

חובה לבדוק את התנאי בתוך לולאת while ולא ב-if. למה? כי יכולות לקרות "התעוררויות מדומות" (spurious wakeups) - מערכת ההפעלה יכולה להעיר thread גם בלי שקראו ל-signal. בנוסף, אם יש כמה צרכנים ורק signal אחד, יכול להיות שthread אחר כבר תפס את האלמנט לפני שהתעוררנו.

// Correct:
pthread_mutex_lock(&mutex);
while (queue_is_empty()) {         // while - not if!
    pthread_cond_wait(&cond, &mutex);
}
// Now the queue is definitely not empty
item = dequeue();
pthread_mutex_unlock(&mutex);

// Incorrect:
pthread_mutex_lock(&mutex);
if (queue_is_empty()) {            // Bug! A spurious wakeup will cause a problem
    pthread_cond_wait(&cond, &mutex);
}
item = dequeue();                  // Possible crash - the queue might be empty
pthread_mutex_unlock(&mutex);

---

מימוש תור מוגבל בטוח לתהליכונים - thread-safe bounded queue

בואו נממש queue שמתאים לדפוס יצרן-צרכן. הqueue מבוסס על מערך מעגלי (ring buffer):

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

#define QUEUE_CAPACITY 16

typedef struct {
    int items[QUEUE_CAPACITY];
    int head;           // where we read from
    int tail;           // where we write to
    int count;          // how many items there are

    pthread_mutex_t mutex;
    pthread_cond_t not_full;   // signals when the queue is no longer full
    pthread_cond_t not_empty;  // signals when the queue is no longer empty
} bounded_queue_t;

void queue_init(bounded_queue_t *q) {
    q->head = 0;
    q->tail = 0;
    q->count = 0;
    pthread_mutex_init(&q->mutex, NULL);
    pthread_cond_init(&q->not_full, NULL);
    pthread_cond_init(&q->not_empty, NULL);
}

void queue_destroy(bounded_queue_t *q) {
    pthread_mutex_destroy(&q->mutex);
    pthread_cond_destroy(&q->not_full);
    pthread_cond_destroy(&q->not_empty);
}

// blocks if the queue is full
void queue_push(bounded_queue_t *q, int item) {
    pthread_mutex_lock(&q->mutex);

    while (q->count == QUEUE_CAPACITY) {
        // the queue is full - wait until someone removes an item
        pthread_cond_wait(&q->not_full, &q->mutex);
    }

    q->items[q->tail] = item;
    q->tail = (q->tail + 1) % QUEUE_CAPACITY;
    q->count++;

    // signal consumers that there's something to take
    pthread_cond_signal(&q->not_empty);
    pthread_mutex_unlock(&q->mutex);
}

// blocks if the queue is empty
int queue_pop(bounded_queue_t *q) {
    pthread_mutex_lock(&q->mutex);

    while (q->count == 0) {
        // the queue is empty - wait until someone inserts an item
        pthread_cond_wait(&q->not_empty, &q->mutex);
    }

    int item = q->items[q->head];
    q->head = (q->head + 1) % QUEUE_CAPACITY;
    q->count--;

    // signal producers that there's room
    pthread_cond_signal(&q->not_full);
    pthread_mutex_unlock(&q->mutex);

    return item;
}

דוגמה שלמה - יצרנים וצרכנים

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

// (bounded_queue_t definition from above)

bounded_queue_t queue;

void *producer(void *arg) {
    int id = *(int *)arg;
    for (int i = 0; i < 10; i++) {
        int item = id * 100 + i;
        queue_push(&queue, item);
        printf("producer %d: pushed %d\n", id, item);
    }
    return NULL;
}

void *consumer(void *arg) {
    int id = *(int *)arg;
    for (int i = 0; i < 10; i++) {
        int item = queue_pop(&queue);
        printf("consumer %d: popped %d\n", id, item);
    }
    return NULL;
}

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

    // 2 producers, 2 consumers - each processes 10 items
    pthread_t producers[2], consumers[2];
    int ids[] = {0, 1};

    for (int i = 0; i < 2; i++) {
        pthread_create(&producers[i], NULL, producer, &ids[i]);
        pthread_create(&consumers[i], NULL, consumer, &ids[i]);
    }

    for (int i = 0; i < 2; i++) {
        pthread_join(producers[i], NULL);
        pthread_join(consumers[i], NULL);
    }

    queue_destroy(&queue);
    printf("done.\n");
    return 0;
}

קומפילציה:

gcc -o prodcon prodcon.c -lpthread

---

דפוס בריכת תהליכונים - thread pool

הרעיון

יצירת thread היא פעולה יקרה (syscall, הקצאת מחסנית, ניהול בקרנל). אם יש לנו שרת שמטפל באלפי בקשות בשנייה, לא נרצה ליצור ולהרוס thread לכל בקשה.

הפתרון: ניצור N תהליכונים מראש (worker threads), ונחזיק queue של משימות. כל worker לוקח משימה מהqueue, מבצע אותה, וחוזר לקחת את הבאה.

ככה עובדים:
- שרתי וב (Apache, Nginx worker pool)
- מנועי מסדי נתונים (PostgreSQL, MySQL)
- מערכות הפעלה (thread pool של הקרנל)

מימוש

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

// task - a pointer to a function with an argument
typedef struct task {
    void (*function)(void *arg);
    void *arg;
    struct task *next;
} task_t;

// thread pool
typedef struct {
    pthread_t *threads;
    int num_threads;

    task_t *task_head;         // head of the task queue
    task_t *task_tail;         // tail of the task queue
    int task_count;

    pthread_mutex_t mutex;
    pthread_cond_t task_available;   // signals when there's a new task
    bool shutdown;                   // shutdown flag
} threadpool_t;

// the function that every worker runs
static void *worker_func(void *arg) {
    threadpool_t *pool = (threadpool_t *)arg;

    while (true) {
        pthread_mutex_lock(&pool->mutex);

        // wait for a task or for shutdown
        while (pool->task_count == 0 && !pool->shutdown) {
            pthread_cond_wait(&pool->task_available, &pool->mutex);
        }

        // if shutdown was requested and there are no more tasks - exit
        if (pool->shutdown && pool->task_count == 0) {
            pthread_mutex_unlock(&pool->mutex);
            return NULL;
        }

        // take a task from the queue
        task_t *task = pool->task_head;
        pool->task_head = task->next;
        if (pool->task_head == NULL) {
            pool->task_tail = NULL;
        }
        pool->task_count--;

        pthread_mutex_unlock(&pool->mutex);

        // run the task (without holding the mutex!)
        task->function(task->arg);
        free(task);
    }

    return NULL;
}

threadpool_t *threadpool_create(int num_threads) {
    threadpool_t *pool = calloc(1, sizeof(threadpool_t));
    pool->num_threads = num_threads;
    pool->threads = malloc(sizeof(pthread_t) * num_threads);
    pool->shutdown = false;

    pthread_mutex_init(&pool->mutex, NULL);
    pthread_cond_init(&pool->task_available, NULL);

    for (int i = 0; i < num_threads; i++) {
        pthread_create(&pool->threads[i], NULL, worker_func, pool);
    }

    return pool;
}

void threadpool_submit(threadpool_t *pool, void (*func)(void *), void *arg) {
    task_t *task = malloc(sizeof(task_t));
    task->function = func;
    task->arg = arg;
    task->next = NULL;

    pthread_mutex_lock(&pool->mutex);

    if (pool->task_tail) {
        pool->task_tail->next = task;
    } else {
        pool->task_head = task;
    }
    pool->task_tail = task;
    pool->task_count++;

    pthread_cond_signal(&pool->task_available);
    pthread_mutex_unlock(&pool->mutex);
}

void threadpool_destroy(threadpool_t *pool) {
    pthread_mutex_lock(&pool->mutex);
    pool->shutdown = true;
    pthread_cond_broadcast(&pool->task_available);  // wake everyone up
    pthread_mutex_unlock(&pool->mutex);

    for (int i = 0; i < pool->num_threads; i++) {
        pthread_join(pool->threads[i], NULL);
    }

    free(pool->threads);
    pthread_mutex_destroy(&pool->mutex);
    pthread_cond_destroy(&pool->task_available);
    free(pool);
}

שימוש בthread pool

void print_task(void *arg) {
    int *num = (int *)arg;
    printf("task: processing number %d (thread %lu)\n", *num, pthread_self());
    free(num);
}

int main(void) {
    threadpool_t *pool = threadpool_create(4);

    // submit 20 tasks
    for (int i = 0; i < 20; i++) {
        int *num = malloc(sizeof(int));
        *num = i;
        threadpool_submit(pool, print_task, num);
    }

    sleep(1);  // give the tasks time to run
    threadpool_destroy(pool);
    return 0;
}

שימו לב: ב-threadpool_destroy אנחנו שולחים broadcast (לא signal) כי צריך להעיר את כל הworkers כדי שיראו את דגל הכיבוי.

---

דפוס נעילת קורא-כותב - reader-writer lock

הרעיון

לפעמים יש לנו מבנה נתונים שהרבה threads קוראים ממנו, אבל מעטים כותבים אליו. אם נשתמש ב-mutex רגיל, רק thread אחד יוכל לקרוא בכל רגע - בזבוז! הרי קריאה בלבד לא מסוכנת.

הפתרון - reader-writer lock:
- הרבה קוראים יכולים להחזיק את הנעילה במקביל.
- כותב אחד מחזיק את הנעילה בבלעדיות (אף אחד אחר, לא קורא ולא כותב).

הAPI

#include <pthread.h>

pthread_rwlock_t rwlock = PTHREAD_RWLOCK_INITIALIZER;

// read lock - several threads can hold it in parallel
pthread_rwlock_rdlock(&rwlock);
// ... read from the structure ...
pthread_rwlock_unlock(&rwlock);

// write lock - exclusive
pthread_rwlock_wrlock(&rwlock);
// ... write to the structure ...
pthread_rwlock_unlock(&rwlock);

דוגמה - מטמון (cache) משותף

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

#define CACHE_SIZE 100

typedef struct {
    char key[64];
    int value;
} cache_entry_t;

typedef struct {
    cache_entry_t entries[CACHE_SIZE];
    int count;
    pthread_rwlock_t lock;
} cache_t;

void cache_init(cache_t *c) {
    c->count = 0;
    pthread_rwlock_init(&c->lock, NULL);
}

// read - many threads can read in parallel
int cache_get(cache_t *c, const char *key, int *value) {
    pthread_rwlock_rdlock(&c->lock);
    for (int i = 0; i < c->count; i++) {
        if (strcmp(c->entries[i].key, key) == 0) {
            *value = c->entries[i].value;
            pthread_rwlock_unlock(&c->lock);
            return 1;  // found
        }
    }
    pthread_rwlock_unlock(&c->lock);
    return 0;  // not found
}

// write - exclusive
void cache_set(cache_t *c, const char *key, int value) {
    pthread_rwlock_wrlock(&c->lock);
    // check if the key already exists
    for (int i = 0; i < c->count; i++) {
        if (strcmp(c->entries[i].key, key) == 0) {
            c->entries[i].value = value;
            pthread_rwlock_unlock(&c->lock);
            return;
        }
    }
    // new key
    if (c->count < CACHE_SIZE) {
        strncpy(c->entries[c->count].key, key, 63);
        c->entries[c->count].key[63] = '\0';
        c->entries[c->count].value = value;
        c->count++;
    }
    pthread_rwlock_unlock(&c->lock);
}

העדפת קוראים מול העדפת כותבים - fairness

יש בעיה עם reader-writer lock: מה קורה כשיש כל הזמן קוראים חדשים? הכותב עלול לא לקבל לעולם את הנעילה! זה נקרא "הרעבה" (starvation).

שתי אסטרטגיות:
- העדפת קוראים (reader-preference) - קוראים חדשים תמיד נכנסים אם יש כבר קוראים. כותבים עלולים להירעב.
- העדפת כותבים (writer-preference) - כשכותב מחכה, קוראים חדשים מחכים גם. מונע הרעבת כותבים.

בלינוקס אפשר להגדיר את ההעדפה עם attributes:

pthread_rwlockattr_t attr;
pthread_rwlockattr_init(&attr);
pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP);
pthread_rwlock_init(&rwlock, &attr);

---

דפוס המחסום - barrier

הרעיון

לפעמים יש לנו חישוב מקבילי שמתחלק לשלבים. כל הthreads חייבים לסיים את שלב 1 לפני שמישהו מתחיל את שלב 2. הדפוס הזה נקרא barrier - מחסום שכל הthreads צריכים להגיע אליו לפני שמישהו ממשיך.

דוגמאות:
- חישוב מטריצות: כל thread מחשב חלק מהשורות, ואז כולם ממשיכים לשלב הבא.
- סימולציות פיזיקה: כל thread מעדכן חלק מהחלקיקים, ואז כולם ממשיכים לצעד הבא.

הAPI

#include <pthread.h>

pthread_barrier_t barrier;

// initialization - count = the number of threads that need to reach the barrier
pthread_barrier_init(&barrier, NULL, count);

// wait - blocks until all count threads arrive
pthread_barrier_wait(&barrier);

// release
pthread_barrier_destroy(&barrier);

דוגמה - חישוב מקבילי בשלבים

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

#define NUM_THREADS 4
#define NUM_PHASES 3
#define ARRAY_SIZE 1000

int data[ARRAY_SIZE];
pthread_barrier_t barrier;

void *worker(void *arg) {
    int id = *(int *)arg;
    int start = id * (ARRAY_SIZE / NUM_THREADS);
    int end = start + (ARRAY_SIZE / NUM_THREADS);

    for (int phase = 0; phase < NUM_PHASES; phase++) {
        // each thread processes its own part
        for (int i = start; i < end; i++) {
            data[i] += phase + 1;
        }
        printf("thread %d: finished phase %d\n", id, phase);

        // everyone must finish the phase before continuing
        pthread_barrier_wait(&barrier);
    }

    return NULL;
}

int main(void) {
    pthread_t threads[NUM_THREADS];
    int ids[NUM_THREADS];

    pthread_barrier_init(&barrier, NULL, NUM_THREADS);

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

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

    pthread_barrier_destroy(&barrier);
    printf("all phases complete.\n");
    return 0;
}

---

דפוס המוניטור - monitor

הרעיון

כל הדפוסים שראינו משתמשים ב-mutex ו-condition variables. הדפוס של מוניטור (monitor) אומר: נעטוף את כל הסנכרון בתוך מבנה נתונים, כך שהמשתמש לא צריך לדעת מזה.

במילים אחרות: מוניטור = מבנה נתונים + mutex + condition variables, כאשר כל הפעולות על המבנה כבר כוללות את הנעילה בפנים.

למעשה, הqueue המוגבל שמימשנו למעלה הוא דוגמה מצוינת למוניטור! המשתמש קורא ל-queue_push ו-queue_pop ולא צריך לדעת מהmutex וה-condition variables.

עקרונות

1. כל המצב הפנימי פרטי - המשתמש לא ניגש ישירות לשדות.
2. כל פעולה ציבורית נועלת ומשחררת - המשתמש לא צריך לנעול בעצמו.
3. condition variables לתיאום - כשפעולה לא יכולה להתבצע, הthread ישן ומתעורר כשאפשר.

דוגמה נוספת - מונה בטוח (safe counter) שמחכה עד שהמונה מגיע לערך מסוים:

typedef struct {
    int value;
    pthread_mutex_t mutex;
    pthread_cond_t value_changed;
} monitor_counter_t;

void counter_init(monitor_counter_t *mc, int initial) {
    mc->value = initial;
    pthread_mutex_init(&mc->mutex, NULL);
    pthread_cond_init(&mc->value_changed, NULL);
}

void counter_increment(monitor_counter_t *mc) {
    pthread_mutex_lock(&mc->mutex);
    mc->value++;
    pthread_cond_broadcast(&mc->value_changed);  // signal everyone who's waiting
    pthread_mutex_unlock(&mc->mutex);
}

// blocks until the counter reaches target
void counter_wait_until(monitor_counter_t *mc, int target) {
    pthread_mutex_lock(&mc->mutex);
    while (mc->value < target) {
        pthread_cond_wait(&mc->value_changed, &mc->mutex);
    }
    pthread_mutex_unlock(&mc->mutex);
}

---

סיכום

דפוס	בעיה שהוא פותר	מנגנונים
יצרן-צרכן (producer-consumer)	העברת נתונים בין threads	הqueue + mutex + condition variables
בריכת תהליכונים (thread pool)	עלות יצירת threads	הthreads קבועים + queue משימות
קורא-כותב (reader-writer)	קריאות מקביליות, כתיבה בלעדית	rwlock
מחסום (barrier)	סנכרון שלבים	barrier
מוניטור (monitor)	עטיפת סנכרון במבנה נתונים	mutex + condition variables

כל הדפוסים האלה בנויים על אותם בלוקים בסיסיים שכבר מכירים: mutex, condition variables, ופעולות אטומיות. מה שמשתנה זה האופן שבו אנחנו מרכיבים אותם כדי לפתור בעיות שונות.