يمكن حل العديد من مشاكل التزامن (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"); }
ترجمة -وبتصرّف- للفصل Condition variables من كتاب Think OS A Brief Introduction to Operating Systems
أفضل التعليقات
لا توجد أية تعليقات بعد
انضم إلى النقاش
يمكنك أن تنشر الآن وتسجل لاحقًا. إذا كان لديك حساب، فسجل الدخول الآن لتنشر باسم حسابك.