الفصل العاشر: المتغيرات الشرطية وحلها مشاكل التزامن بين العمليات في لغة C

يمكن حل العديد من مشاكل التزامن (synchronization) البسيطة باستخدام كائنات المزامنة (mutexes)، ولكن يوجد مشكلة أكبر هي مشكلة منتج-مستهلك (Producer-Consumer problem) التي تُحل باستخدام أداة جديدة هي المتغير الشرطي (condition variable).

طابور العمل (The work queue)

تُنظَّم الخيوط في بعض البرامج ذات الخيوط المتعددة لتجري عدة مهام، وتتواصل هذه الخيوط مع بعضها البعض غالبًا عن طريق طابور (queue)، حيث تدعى الخيوط التي تضع بيانات في الطابور بالخيوط المنتجة (producers)، وتدعى الخيوط التي تأخذ بيانات من الطابور بالخيوط المستهلكة (consumers). يمكن أن يوجد خيطٌ يشغّل واجهة المستخدم الرسومية (graphical user interface وتختصر إلى GUI) للاستجابة لأحداث المستخدم في التطبيقات التي لديها واجهة مستخدم رسومية على سبيل المثال، ويمكن أن يوجد خيطٌ آخر يعالج طلبات المستخدم، حيث يمكن أن يضع خيطُ واجهة المستخدم الرسومية الطلبات في طابورٍ ثم يأخذ الخيط المقابل هذه الطلبات ويعالجها. تحتاج تطبيق طابور لدعم هذا التنظيم، بحيث يحافظ تطبيق الطابور على الخيوط (thread safe)، وهذا يعني أنه يستطيع كلا الخيطين (أو أكثر من خيطين في بعض الأحيان) الوصول إلى الطابور في نفس الوقت، وتحتاج أيضًا أن تعالج الحالات الخاصة مثل أن يكون الطابور فارغًا (empty) وأن يكون حجم الطابور منتهٍ عندما يمتلئ (full). سأبدأ، يقول الكاتب، بطابورٍ بسيط لا يحافظ على الخيوط ثم ترى كيف يكون ذلك خاطئًا وكيف يُصحَّح ذلك الخطأ. شيفرة هذا المثال موجودة في المجلد queue حيث يتضمن الملف queue.c التطبيق الأساسي للمخزَن الدائري circular buffer. تجد تعريف البنية Queue فيما يلي:

typedef struct
{
    int *array;
    int length;
    int next_in;
    int next_out;
} Queue;

array هو المصفوفة التي تتضمن عناصر الطابور وهي أعداد صحيحة (ints) في هذا المثال، ولكنها يمكن أن تكون بنى (structures) تتضمن أحداث المستخدم وعناصر العمل وغير ذلك. length هو طول المصفوفة، و next_in هو دليل (index) المصفوفة الذي يحدد مكان إضافة العنصر التالي في الطابور، أما next_out هو دليل العنصر التالي الذي يجب حذفه من الطابور. تخصص الدالة make_queue حيزًا للبنية Queue وتهيئ حقولها كما يلي:

Queue *make_queue(int length)
{
    Queue *queue = (Queue *)malloc(sizeof(Queue));
    queue->length = length + 1;
    queue->array = (int *)malloc(length * sizeof(int));
    queue->next_in = 0;
    queue->next_out = 0;
    return queue;
}

تحتاج القيمة الابتدائية للمتغير next_out بعض الشرح، فبما أن الطابور فارغ مبدئيًا فلا وجود لعنصر تالٍ لحذفه، لذلك يكون المتغير next_out غير صالح (invalid)، وضبط next_out == next_in هي حالة خاصة تحدد أن الطابور فارغ، فيمكن كتابة ما يلي:

int queue_empty(Queue *queue)
{
    return (queue->next_in == queue->next_out);
}

يمكنك الآن إضافة عناصر إلى الطابور باستخدام الدالة queue_push:

void queue_push(Queue *queue, int item)
{
    if (queue_full(queue))
    {
        perror_exit("queue is full");
    }

    queue->array[queue->next_in] = item;
    queue->next_in = queue_incr(queue, queue->next_in);
}

إذا كان الطابور ممتلئًا فإن الدالة queue_push تطبع رسالة خطأ وتغادر، أما إذا كان الطابور غير ممتلئ فتدخِل الدالة queue_push عنصرًا جديدًا ثم تزيد قيمة المتغير next_in باستخدام الدالة queue_incr كما يلي:

int queue_incr(Queue *queue, int i)
{
    return (i + 1) % queue->length;
}

تعود قيمة الدليل i إلى الصفر عندما يصل الدليل إلى نهاية المصفوفة، وهذا هو المكان الذي نواجه فيه الجزء الصعب، فإذا واصلنا إضافة عناصر إلى الطابور فسيعود المتغير next_in ويلحق بالمتغير next_out، وإذا كان next_in == next_out ستستنتج بصورة غير صحيحة أن الطابور فارغ، لذلك يجب تعريف حالة خاصة أخرى لتحديد أن الطابور ممتلئ لتجنب ذلك:

int queue_full(Queue *queue)
{
    return (queue_incr(queue, queue->next_in) == queue->next_out);
}

حيث إذا واصلت زيادة المتغير next_in ليصل إلى قيمة المتغير next_out فهذا يعني أنك لا تستطيع إضافة عنصر آخر إلى الطابور بدون جعل الطابور يبدو فارغًا، لذلك يجب التوقف عن إضافة عناصر أخرى قبل نهاية الطابور بعنصر واحد (يجب أن تعرف أن نهاية الطابور يمكن أن تكون في أي مكان وليس بالضرورة عند نهاية المصفوفة). يمكن الآن كتابة الدالة queue_pop التي تحذف وتعيد العنصر التالي من الطابور كما يلي:

int queue_pop(Queue *queue)
{
    if (queue_empty(queue))
    {
        perror_exit("queue is empty");
    }

    int item = queue->array[queue->next_out];
    queue->next_out = queue_incr(queue, queue->next_out);
    return item;
}

وإذا جربت سحب (pop) عنصر من طابور فارغ فستطبع الدالة queue_pop رسالة خطأ وتغادر.

المستهلكون والمنتجون (Producers and consumers)

تنشئ الآن بعض الخيوط لتصل إلى هذا الطابور، حيث شيفرة المنتج (producer) هي كما يلي:

void *producer_entry(void *arg)
{
    Shared *shared = (Shared *)arg;

    for (int i = 0; i < QUEUE_LENGTH - 1; i++)
    {
        printf("adding item %d\n", i);
        queue_push(shared->queue, i);
    }
    pthread_exit(NULL);
}

أما شيفرة المستهلك (consumer) هي:

void *consumer_entry(void *arg)
{
    int item;
    Shared *shared = (Shared *)arg;

    for (int i = 0; i < QUEUE_LENGTH - 1; i++)
    {
        item = queue_pop(shared->queue);
        printf("consuming item %d\n", item);
    }
    pthread_exit(NULL);
}

وشيفرة الخيط الأب الذي يبدأ الخيوط وينتظرها لتنتهي هي:

pthread_t child[NUM_CHILDREN];

Shared *shared = make_shared();

child[0] = make_thread(producer_entry, shared);
child[1] = make_thread(consumer_entry, shared);

for (int i = 0; i < NUM_CHILDREN; i++)
{
    join_thread(child[i]);
}

والبنية المشتركة التي تتضمن الطابور هي:

typedef struct
{
    Queue *queue;
} Shared;

Shared *make_shared()
{
    Shared *shared = check_malloc(sizeof(Shared));
    shared->queue = make_queue(QUEUE_LENGTH);
    return shared;
}

تمثل الشيفرة السابقة التي حصلت عليها حتى الآن بدايةً جيدة ولكن لديها بعض المشاكل هي:

لا يحافظ الوصول إلى الطابور على الخيوط، حيث يمكن أن تصل خيوط متعددة إلى المتغيرات array و next_in و next_out في نفس الوقت، وهذا يترك الطابور تالفًا وفي حالة غير مستقرة.
إذا جُدوِل الخيط المستهلك أولًا فسيجد الطابور فارغًا، وبالتالي يطبع رسالة خطأ وينتهي، لذلك من الأفضل أن يتوقف المستهلك حتى يصبح الطابور غير فارغ. وبالمثل يجب إيقاف المنتج إذا كان الطابور ممتلئًا.

ستُحل المشكلة الأولى في الفقرة القادمة باستخدام Mutex، وستحل المشكلة الثانية في الفقرة التي بعدها باستخدام المتغيرات الشرطية.

الإقصاء المتبادل (Mutual exclusion)

يحافظ الطابور على الخيوط باستخدام كائن المزامنة (mutex)، حيث تضيف أولًا المؤشر Mutex إلى بنية الطابور:

typedef struct
{
    int *array;
    int length;
    int next_in;
    int next_out;
    Mutex *mutex; //-- هذا السطر جديد
} Queue;

ثم تهيئه في الدالة make_queue:

Queue *make_queue(int length)
{
    Queue *queue = (Queue *)malloc(sizeof(Queue));
    queue->length = length;
    queue->array = (int *)malloc(length * sizeof(int));
    queue->next_in = 0;
    queue->next_out = 0;
    queue->mutex = make_mutex(); //-- جديد
    return queue;
}

ثم تضيف شيفرة التزامن إلى الدالة queue_push:

void queue_push(Queue *queue, int item)
{
    mutex_lock(queue->mutex); //-- جديد
    if (queue_full(queue))
    {
        mutex_unlock(queue->mutex); //-- جديد
        perror_exit("queue is full");
    }

    queue->array[queue->next_in] = item;
    queue->next_in = queue_incr(queue, queue->next_in);
    mutex_unlock(queue->mutex); //-- جديد
}

يجب قفل Mutex قبل التحقق إذا كان الطابور ممتلئًا أم لا، فإذا كان ممتلئًا يجب فك قفل Mutex قبل المغادرة، وإلا سيتركه الخيط مقفلًا فلا يستطيع أي خيطٍ آخر أن يستأنف عمله. شيفرة التزامن للدالة queue_pop هي:

int queue_pop(Queue *queue)
{
    mutex_lock(queue->mutex);
    if (queue_empty(queue))
    {
        mutex_unlock(queue->mutex);
        perror_exit("queue is empty");
    }

    int item = queue->array[queue->next_out];
    queue->next_out = queue_incr(queue, queue->next_out);
    mutex_unlock(queue->mutex);
    return item;
}

لاحظ أن دوال Queue الأخرى والتي هي queue_full و queue_empty و queue_incr لا تحاول قفل كائن المزامنة، فيجب على كل خيط يستدعي هذه الدوال أن يقفل كائن المزامنة أولًا. أصبح الطابور محافظًا على الخيوط باستخدام الشيفرة الجديدة التي أضيفت، ولا يجب أن ترى أخطاء تزامن إذا شغلت هذه الشيفرة، ولكنك سترى أن الخيط المستهلك يغادر أحيانًا لأن الطابور فارغ، أو قد ترى الخيط المنتج يغادر بسبب أن الطابور ممتلئ أو كلا الأمرين معًا، وبالتالي الخطوة القادمة هي إضافة المتغيرات الشرطية (condition variables).

المتغيرات الشرطية (Condition variables)

المتغير الشرطي هو عبارة عن بينة بيانات مرتبطة بشرط (condition)، ويسمح المتغير الشرطي بإيقاف الخيوط حتى يتحقق الشرط أو تصبح قيمته true، فقد تتحقق الدالة thread_pop على سبيل المثال فيما إذا كان الطابور فارغًا أم لا، فإذا كان فارغًا تنتظر شرطًا هو (الطابور غير فارغ). وقد تتحقق الدالة thread_push أيضًا فيما إذا كان الطابور ممتلئًا، فإذا كان ممتلئًا تتوقف حتى يصبح غير ممتلئ. تعالج الشيفرة التالية الشرط الأول، حيث تضيف أولًا متغيرًا شرطيًا إلى البنية Queue:

typedef struct
{
    int *array;
    int length;
    int next_in;
    int next_out;
    Mutex *mutex;
    Cond *nonempty; //-- جديد
} Queue;

ثم تهيئه في الدالة make_queue:

Queue *make_queue(int length)
{
    Queue *queue = (Queue *)malloc(sizeof(Queue));
    queue->length = length;
    queue->array = (int *)malloc(length * sizeof(int));
    queue->next_in = 0;
    queue->next_out = 0;
    queue->mutex = make_mutex();
    queue->nonempty = make_cond(); //-- جديد
    return queue;
}

إذا وجدت الطابور فارغًا في الدالة queue_pop لا تغادر بل استخدم المتغير الشرطي لتوقف التنفيذ:

int queue_pop(Queue *queue)
{
    mutex_lock(queue->mutex);
    while (queue_empty(queue))
    {
        cond_wait(queue->nonempty, queue->mutex); //-- جديد
    }
    int item = queue->array[queue->next_out];
    queue->next_out = queue_incr(queue, queue->next_out);
    mutex_unlock(queue->mutex);
    cond_signal(queue->nonfull); //-- جديد
    return item;
}

الدالة cond_wait معقدة، فوسيطها الأول هو المتغير الشرطي والشرط الذي يجب انتظاره في هذه الحالة هو (الطابور غير فارغ)، أما وسيطها الثاني هو كائن المزامنة الذي يحمي الطابور. يفك الخيط قفل كائن المزامنة ثم يتوقف عندما يستدعي الخيط الذي قفل كائن المزامنة الدالةَ cond_wait، وهذا شيء مهم جدًا. إذا لم تقفل الدالة cond_wait كائن المزامنة قبل التوقف فلن يستطيع أي خيطٍ آخر أن يصل إلى الطابور ولن تضاف أي عناصر أخرى إلى الطابور، وبالتالي قد يبقى الطابور فارغًا دائمًا، فيمكن أن يشغّل المنتج بينما يكون المستهلك متوقفًا عند nonempty. تبين الشيفرة التالية ما يحدث عنما يشغّل المنتج الدالة queue_push:

void queue_push(Queue *queue, int item)
{
    mutex_lock(queue->mutex);
    if (queue_full(queue))
    {
        mutex_unlock(queue->mutex);
        perror_exit("queue is full");
    }
    queue->array[queue->next_in] = item;
    queue->next_in = queue_incr(queue, queue->next_in);
    mutex_unlock(queue->mutex);
    cond_signal(queue->nonempty); //-- جديد
}

تقفل الدالة queue_push المتغير Mutex وتتحقق فيما إذا كان الطابور ممتلئًا أم لا، وعلى فرض أن الطابور ليس ممتلئًا حيث تضيف الدالة queue_push عنصرًا جديدًا إلى الطابور ثم تفك قفل المتغير Mutex، ولكن تقوم هذه الدالة بشيء آخر قبل أن تعيد شيئًا، حيث تنبّه (signals) المتغير الشرطي nonempty، ويحدد تنبيه (Signalling) المتغير الشرطي أن الشرط صحيح (true)، وليس لإشارة التنبيه أي تأثير إذا لم يوجد خيوطٌ تنتظر المتغير الشرطي. إذا وجد خيوط تنتظر المتغير الشرطي فيعود أحد هذه الخيوط إلى العمل ويستأنف تنفيذ الدالة cond_wait، ولكن يجب على الخيط الذي استأنف عمله أن ينتظر المتغير Mutex ويقفله مرة أخرى قبل أن ينهي تنفيذ الدالة cond_wait. عُد الآن إلى الدالة queue_pop وشاهد ما يحدث عندما ينهي الخيط تنفيذ الدالة cond_wait، حيث يعود الخيط مرة أخرى إلى بداية حلقة while ويتحقق من الشرط مرة أخرى. افترض أنه تحقق الشرط أي أن الطابور غير فارغ، فعندما يغادر الخيط المستهلك حلقة while، هذا يؤدي إلى شيئين: (1) تحقق الشرط أي يوجد عنصر واحد على الأقل في الطابور، و (2) قُفِل المتغير Mutex أي أن الوصول إلى الطابور آمن.

تفك الدالة queue_pop قفل كائن المزامنة وتنتهي بعد حذف عنصر من الطابور، سأبين، يقول الكاتب، كيفية عمل شيفرة Cond ولكن يجب أولًا الإجابة عن سؤالين مهمين هما:

لماذا توجد الدالة cond_wait ضمن حلقة while بدلًا من وجودها ضمن عبارة if، أي لماذا يجب التحقق من الشرط مرة أخرى بعد انتهاء تنفيذ الدالة cond_wait؟

السبب الرئيسي لإعادة التحقق من الشرط هو إمكانية اعتراض إشارة تنبيه، حيث افترض أن الخيط A ينتظر nonempty، ويضيف الخيط B عنصرًا إلى الطابور ثم ينبه nonempty، فيستيقظ الخيط A ويحاول قفل كائن المزامنة، ولكن قبل أن يقوم الخيط A بذلك، يأتي الخيط الشرير C ويقفل كائن المزامنة ويسحب عنصرًا من الطابور ثم يفك قفل كائن المزامنة، وبالتالي أصبح الطابور فارغًا الآن مرة أخرى، ولكن لا يتوقف الخيط A مرة أخرى، ويستطيع الخيط A قفل كائن المزامنة وينهي تنفيذ الدالة cond_wait، فإذا لم يتحقق الخيط A من الشرط مرة أخرى فقد يحاول سحب عنصر من طابورٍ فارغ، وقد يسبب ذلك خطأ.

أما السؤال الثاني الذي يظهر عندما يتعلم الناس المتغيرات الشرطية هو: كيف يعرف المتغير الشرطي الشرطَ الذي يتعلق به؟

هذا السؤال مفهوم لأنه لا يوجد اتصال صريح بين بنية Cond والشرط المتعلق بها، حيث يكون الاتصال مضمّنًا في طريقة استخدامه، وهذه إحدى الطرق للتفكير في ذلك: فالشرط المتعلق ب Cond هو الشي الذي تكون قيمته خاطئة (false) عندما تستدعي الدالة cond_wait، وتكون قيمته صحيحة (true) عندما تستدعي الدالة cond_signal.

ليس من الضروري تمامًا استدعاء الدالة cond_signal فقط عندما يكون الشرط صحيحًا، نظرًا لأن الخيوط يجب أن تتحقق من الشرط عند انتهاء تنفيذها للدالة cond_wait. إذا كان لديك سبب للاعتقاد بأن الشرط قد يكون صحيحًا فيمكنك استدعاء الدالة cond_signal كاقتراحٍ للتحقق من ذلك.

تطبيق المتغير الشرطي (Condition variable implementation)

البنية Cond التي استخدمت في الفقرة السابقة هي مغلّف لنوعٍ يدعى pthread_cond_t المعرّف في واجهة برمجة التطبيقات للخيوط POSIX. البنية Cond شبيهة جدًا بالبنية Mutex والتي هي مغلّفة للنوع pthread_mutex_t، حيث تعريف النوع Cond كما يلي:

typedef pthread_cond_t Cond;

تخصص الدالة make_cond حيزًا وتهيئ المتغير الشرطي وتعيد مؤشرًا:

Cond *make_cond()
{
    Cond *cond = check_malloc(sizeof(Cond));
    int n = pthread_cond_init(cond, NULL);
    if (n != 0)
        perror_exit("make_cond failed");

    return cond;
}

أما الدالتان المغلّفتان للدالتين cond_wait و cond_signal:

void cond_wait(Cond *cond, Mutex *mutex)
{
    int n = pthread_cond_wait(cond, mutex);
    if (n != 0)
        perror_exit("cond_wait failed");
}

void cond_signal(Cond *cond)
{
    int n = pthread_cond_signal(cond);
    if (n != 0)
        perror_exit("cond_signal failed");
}