فلاسفة ذوو تغذية جيدة أو برمجة .NET تنافسية

دعونا نرى كيف تعمل البرمجة المتزامنة والمتوازية في .Net ، باستخدام مشكلة تناول طعام الفلاسفة كمثال. الخطة هي هذا ، من تزامن الخيوط / العمليات ، إلى نموذج الممثل (في الأجزاء التالية). قد تكون المقالة مفيدة للمعارف الأولى أو لتحديث معلوماتك.

لماذا تفعل ذلك على الإطلاق؟ تصل الترانزستورات إلى الحد الأدنى من حجمها ، ويستند قانون مور إلى الحد من سرعة الضوء ، وبالتالي لوحظ زيادة في العدد ، ويمكن صنع المزيد من الترانزستورات. في الوقت نفسه ، تتزايد كمية البيانات ، ويتوقع المستخدمون استجابة فورية من الأنظمة. في مثل هذه الحالة ، فإن البرمجة "العادية" ، عندما يكون لدينا سلسلة تنفيذ واحدة ، لم تعد فعالة. تحتاج إلى حل مشكلة التنفيذ المتزامن أو المتزامن بطريقة ما. علاوة على ذلك ، توجد هذه المشكلة على مستويات مختلفة: على مستوى الخيوط ، على مستوى العمليات ، على مستوى الآلات في الشبكة (الأنظمة الموزعة). تمتلك .NET تقنيات عالية الجودة تم اختبارها على مدار الوقت لحل مثل هذه المشكلات بسرعة وكفاءة.

مهمة

طرح Edsger Dijkstra هذه المشكلة على طلابه في وقت مبكر من عام 1965. والصيغة المعمول بها هي كما يلي. هناك عدد معين (خمسة في العادة) من الفلاسفة ونفس عدد المفترقات. يجلسون على طاولة مستديرة ، وبينهم شوك. يمكن للفلاسفة أن يأكلوا من أطباقهم طعامًا لا نهاية له ، يفكرون أو ينتظرون. لكي تأكل فيلسوفًا ، عليك أن تأخذ شوكتين (الأخيرة تشترك في الشوكة مع الأولى). التقاط الشوكة ووضعها هما إجراءان منفصلان. كل الفلاسفة صامتون. تتمثل المهمة في العثور على خوارزمية تفكر فيها جميعًا وتكون ممتلئة حتى بعد 54 عامًا.

أولاً ، دعنا نحاول حل هذه المشكلة من خلال استخدام مساحة مشتركة. تقع الشوكات على الطاولة المشتركة ويأخذها الفلاسفة عندما يكونون ثم يعيدونها. هنا توجد مشاكل في المزامنة ، متى بالضبط يجب أن تأخذ رهانات مؤكدة؟ ماذا لو لم يكن هناك شوكة؟ إلخ ولكن أولاً ، لنبدأ الفلاسفة.

لبدء المواضيع ، نستخدم تجمع مؤشرات الترابط من خلال Task.Run طريقة:

var cancelTokenSource = new CancellationTokenSource();
Action<int> create = (i) => RunPhilosopher(i, cancelTokenSource.Token);
for (int i = 0; i < philosophersAmount; i++) 
{
    int icopy = i;
    // Поместить задачу в очередь пула потоков. Метод RunDeadlock не запускаеться 
    // сразу, а ждет своего потока. Асинхронный запуск.
    philosophers[i] = Task.Run(() => create(icopy), cancelTokenSource.Token);
}

تم تصميم تجمع مؤشرات الترابط لتحسين إنشاء سلاسل الرسائل وحذفها. يحتوي هذا التجمع على قائمة انتظار تحتوي على مهام ويقوم CLR بإنشاء أو إزالة مؤشرات الترابط بناءً على عدد هذه المهام. تجمع واحد لجميع AppDomains. يجب استخدام هذا التجمع دائمًا تقريبًا ، لأن. لا داعي للقلق بشأن إنشاء سلاسل الرسائل وحذفها وقوائم الانتظار الخاصة بها ، وما إلى ذلك ، فمن الممكن بدون مجموعة ، ولكن بعد ذلك يتعين عليك استخدامها مباشرة Thread، هذا مفيد للحالات التي تحتاج فيها إلى تغيير أولوية مؤشر ترابط ، عندما يكون لدينا عملية طويلة ، بالنسبة إلى مؤشر ترابط المقدمة ، إلخ.

وبعبارة أخرى، System.Threading.Tasks.Task الطبقة هي نفسها Thread، ولكن مع جميع أنواع وسائل الراحة: القدرة على تشغيل مهمة بعد مجموعة من المهام الأخرى ، وإعادتها من الوظائف ، ومقاطعتها بسهولة ، والمزيد. إلخ. هناك حاجة إليها لدعم الإنشاءات غير المتزامنة / المنتظرة (النمط غير المتزامن المستند إلى المهام ، والسكر النحوي في انتظار عمليات الإدخال / الإخراج). سنتحدث عن هذا لاحقًا.

CancelationTokenSource هنا هو مطلوب بحيث يمكن أن ينتهي الخيط من تلقاء نفسه عند إشارة مؤشر الترابط المتصل.

قضايا متزامنة

الفلاسفة المحظورون

حسنًا ، نحن نعرف كيفية إنشاء المواضيع ، دعنا نحاول تناول الغداء:

// Кто какие вилки взял. К примеру: 1 1 3 3 - 1й и 3й взяли первые две пары.
private int[] forks = Enumerable.Repeat(0, philosophersAmount).ToArray();

// То же, что RunPhilosopher()
private void RunDeadlock(int i, CancellationToken token) 
{
    // Ждать вилку, взять её. Эквивалентно: 
    // while(true) 
    //     if forks[fork] == 0 
    //          forks[fork] = i+1
    //          break
    //     Thread.Sleep() или Yield() или SpinWait()
    void TakeFork(int fork) =>
        SpinWait.SpinUntil(() => 
            Interlocked.CompareExchange(ref forks[fork], i+1, 0) == 0);

    // Для простоты, но можно с Interlocked.Exchange:
    void PutFork(int fork) => forks[fork] = 0;

    while (true)
    {
        TakeFork(Left(i));
        TakeFork(Right(i));
        eatenFood[i] = (eatenFood[i] + 1) % (int.MaxValue - 1);
        PutFork(Left(i));
        PutFork(Right(i));
        Think(i);

        // Завершить работу по-хорошему.
        token.ThrowIfCancellationRequested();
    }
}

هنا نحاول أولاً أن نأخذ الشوكة اليسرى ، ثم الشوكة اليمنى ، وإذا نجحت ، فإننا نأكلها ونعيدها. أخذ شوكة واحدة ذري ، أي. لا يمكن أن يأخذ خيطان واحدًا في نفس الوقت (غير صحيح: يقرأ الأول أن الشوكة مجانية ، والثاني - أيضًا ، الأول يأخذ ، الثاني). لهذا Interlocked.CompareExchange، والتي يجب تنفيذها بتعليمات المعالج (TSL, XCHG) ، الذي يقفل قطعة من الذاكرة للقراءة والكتابة التسلسلية الذرية. و SpinWait يعادل البناء while(true) فقط مع القليل من "السحر" - يأخذ الخيط المعالج (Thread.SpinWait) ، ولكن في بعض الأحيان ينقل التحكم إلى مؤشر ترابط آخر (Thread.Yeild) أو ينام (Thread.Sleep).

لكن هذا الحل لا يعمل ، لأن يتم حظر التدفقات قريبًا (بالنسبة لي في غضون ثانية): يأخذ جميع الفلاسفة مفترقهم الأيسر ، لكن ليس الشوكة اليمنى. عندئذٍ يكون لمصفوفة الشوكات القيم: 1 2 3 4 5.

في الشكل ، حجب المواضيع (طريق مسدود). أخضر - تنفيذ ، أحمر - التزامن ، رمادي - الخيط نائم. تشير المعينات إلى وقت بدء المهام.

جوع الفلاسفة

على الرغم من أنه ليس من الضروري التفكير في الكثير من الطعام بشكل خاص ، إلا أن الجوع يجعل أي شخص يتخلى عن الفلسفة. دعونا نحاول محاكاة حالة تجويع الخيوط في مشكلتنا. الجوع هو عندما يتم تشغيل الخيط ، ولكن بدون عمل كبير ، بمعنى آخر ، هذا هو نفس المأزق ، الآن فقط الخيط لا ينام ، ولكنه يبحث بنشاط عن شيء يأكله ، ولكن لا يوجد طعام. من أجل تجنب الحجب المتكرر ، سنعيد الشوكة إذا لم نتمكن من أخذ واحدة أخرى.

// То же что и в RunDeadlock, но теперь кладем вилку назад и добавляем плохих философов.
private void RunStarvation(int i, CancellationToken token)
{
    while (true)
    {
        bool hasTwoForks = false;
        var waitTime = TimeSpan.FromMilliseconds(50);
        // Плохой философов может уже иметь вилку:
        bool hasLeft = forks[Left(i)] == i + 1;
        if (hasLeft || TakeFork(Left(i), i + 1, waitTime))
        {
            if (TakeFork(Right(i), i + 1, TimeSpan.Zero))
                hasTwoForks = true;
            else
                PutFork(Left(i)); // Иногда плохой философ отдает вилку назад.
        } 
        if (!hasTwoForks)
        {
            if (token.IsCancellationRequested) break;
            continue;
        }
        eatenFood[i] = (eatenFood[i] + 1) % (int.MaxValue - 1);
        bool goodPhilosopher = i % 2 == 0;
        // А плохой философ забывает положить свою вилку обратно:
        if (goodPhilosopher)
            PutFork(Left(i));
        // А если и правую не положит, то хорошие будут вообще без еды.
        PutFork(Right(i));

        Think(i);

        if (token.IsCancellationRequested)
            break;
    }
}

// Теперь можно ждать определенное время.
bool TakeFork(int fork, int philosopher, TimeSpan? waitTime = null)
{
    return SpinWait.SpinUntil(
        () => Interlocked.CompareExchange(ref forks[fork], philosopher, 0) == 0,
              waitTime ?? TimeSpan.FromMilliseconds(-1)
    );
}

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

هنا يمكنك أن ترى أن المواضيع تستيقظ أحيانًا وتحاول الحصول على المورد. اثنتان من النوى الأربعة لا تفعل شيئًا (الرسم البياني الأخضر أعلاه).

موت الفيلسوف

حسنًا ، هناك مشكلة أخرى يمكن أن تقطع عشاءًا رائعًا للفلاسفة وهي إذا مات أحدهم فجأة وشوكًا في يديه (وسوف يدفنوه هكذا). ثم يترك الجيران بدون عشاء. يمكنك الخروج بمثال رمز لهذه الحالة بنفسك ، على سبيل المثال ، تم إلقاؤه NullReferenceException بعد أن يأخذ الفيلسوف الشوك. وبالمناسبة ، لن يتم التعامل مع الاستثناء ولن يكتفي رمز الاستدعاء (لهذا الغرض AppDomain.CurrentDomain.UnhandledException وإلخ.). لذلك ، هناك حاجة إلى معالجات الأخطاء في مؤشرات الترابط نفسها ومع الإنهاء الرائع.

النادل

حسنًا ، كيف نحل مشكلة الجمود والجوع والموت؟ سنسمح لفيلسوف واحد فقط بالوصول إلى مفترق الطرق ، مع إضافة استبعاد متبادل للخيوط لهذا المكان. كيف افعلها؟ لنفترض أن نادلًا يقف بجانب الفلاسفة ، الذي يعطي الإذن لأي فيلسوف بأخذ الشوك. كيف نصنع هذا النادل وكيف سيسأله الفلاسفة الأسئلة شيقة.

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

حلول في مساحة المستخدم

هنا سنفعل الشيء نفسه الذي اعتدنا فعله مع شوكة واحدة واثنين من الفلاسفة ، وسوف ندور في دورة وننتظر. ولكن الآن سيكون كل الفلاسفة ، وكما كان الحال ، سيكون هناك شوكة واحدة فقط ، أي. يمكن القول أن الفيلسوف الذي أخذ هذه "الشوكة الذهبية" من النادل فقط هو الذي سيأكل. لهذا نستخدم SpinLock.

private static SpinLock spinLock = new SpinLock();  // Наш "официант"
private void RunSpinLock(int i, CancellationToken token)
{
    while (true)
    {
        // Взаимная блокировка через busy waiting. Вызываем до try, чтобы
        // выбрасить исключение в случае ошибки в самом SpinLock.
        bool hasLock = false;
        spinLock.Enter(ref hasLock);
        try
        {
            // Здесь может быть только один поток (mutual exclusion).
            forks[Left(i)] = i + 1;  // Берем вилку сразу, без ожидания.
            forks[Right(i)] = i + 1;
            eatenFood[i] = (eatenFood[i] + 1) % (int.MaxValue - 1);
            forks[Left(i)] = 0;
            forks[Right(i)] = 0;
        }
        finally
        {
            if(hasLock) spinLock.Exit();  // Избегаем проблемы со смертью философа.
        }

        Think(i);

        if (token.IsCancellationRequested)
            break;
    }
}

SpinLock هذا مانع ، بالمعنى التقريبي ، نفس الشيء while(true) { if (!lock) break; }، ولكن مع "سحر" أكثر مما هو عليه في SpinWait (الذي يستخدم هناك). الآن يعرف كيف يحصي أولئك الذين ينتظرون ، ويضعهم في النوم قليلاً ، وأكثر من ذلك. بشكل عام ، يفعل كل ما هو ممكن لتحسين. لكن يجب أن نتذكر أن هذه لا تزال نفس الدورة النشطة التي تلتهم موارد المعالج وتحافظ على التدفق ، مما قد يؤدي إلى المجاعة إذا أصبح أحد الفلاسفة أولوية أكثر من غيره ، ولكن ليس لديه شوكة ذهبية (مشكلة انعكاس الأولوية) . لذلك ، نستخدمها فقط لإجراء تغييرات قصيرة جدًا في الذاكرة المشتركة ، دون أي مكالمات من جهات خارجية ، وأقفال متداخلة ، ومفاجآت أخرى.

الرسم ل SpinLock. تيارات تتصارع باستمرار من أجل الشوكة الذهبية. هناك حالات فشل - في الشكل ، المنطقة المحددة. لم يتم استخدام النوى بشكل كامل: فقط حوالي 2/3 بواسطة هذه الخيوط الأربعة.

حل آخر هنا هو الاستخدام فقط Interlocked.CompareExchange مع نفس الانتظار النشط كما هو موضح في الكود أعلاه (في الفلاسفة الجائعين) ، لكن هذا ، كما قيل سابقًا ، يمكن أن يؤدي نظريًا إلى الحجب.

حول Interlocked وتجدر الإشارة إلى أنه لا يوجد سوى CompareExchange، ولكن أيضًا طرق أخرى للقراءة الذرية والكتابة. ومن خلال تكرار التغيير ، في حالة وجود خيط آخر لديه الوقت لإجراء تغييراته (اقرأ 1 ، اقرأ 2 ، اكتب 2 ، اكتب 1 سيئًا) ، يمكن استخدامه لإجراء تغييرات معقدة على قيمة واحدة (نمط متشابك أي شيء) .

حلول وضع Kernel

لتجنب إهدار الموارد في حلقة ، دعنا نرى كيف يمكننا حظر موضوع. بمعنى آخر ، استمرارًا لمثالنا ، لنرى كيف يضع النادل الفيلسوف في حالة نومه ويوقظه عند الضرورة فقط. أولاً ، دعنا نلقي نظرة على كيفية القيام بذلك من خلال وضع kernel لنظام التشغيل. غالبًا ما تكون جميع الهياكل هناك أبطأ من تلك الموجودة في مساحة المستخدم. عدة مرات أبطأ ، على سبيل المثال AutoResetEvent ربما 53 مرة أبطأ SpinLock [ريختر]. ولكن بمساعدتهم ، يمكنك مزامنة العمليات في جميع أنحاء النظام ، سواء كانت مُدارة أم لا.

البناء الأساسي هنا هو الإشارة التي اقترحها Dijkstra منذ أكثر من نصف قرن. السيمافور ، ببساطة ، هو عدد صحيح موجب يديره النظام ، وعمليتان عليه ، الزيادة والنقصان. إذا فشل في الانخفاض ، صفر ، فسيتم حظر مؤشر ترابط الاستدعاء. عندما يتم زيادة الرقم بواسطة بعض الخيوط / العمليات النشطة الأخرى ، يتم تخطي الخيوط ويتم إنقاص الإشارة مرة أخرى بواسطة الرقم الذي تم تمريره. يمكن للمرء أن يتخيل قطارات في عنق الزجاجة بإشارة. تقدم .NET العديد من التركيبات ذات الوظائف المماثلة: AutoResetEvent, ManualResetEvent, Mutex ونفسي Semaphore. سوف نستخدم AutoResetEvent، هذا هو أبسط هذه التركيبات: قيمتان فقط 0 و 1 (خطأ ، صحيح). طريقتها WaitOne() يحظر سلسلة الاستدعاء إذا كانت القيمة 0 ، وإذا كانت 1 ، يخفضها إلى 0 ويتخطها. طريقة Set() يرفع إلى 1 ويسمح لنادل واحد بالمرور ، والذي ينخفض ​​مرة أخرى إلى 0. يتصرف مثل باب دوار لمترو الأنفاق.

دعونا نعقد الحل ونستخدم القفل لكل فيلسوف ، وليس للأبد مرة واحدة. أولئك. الآن يمكن أن يكون هناك العديد من الفلاسفة في آنٍ واحد ، وليس واحدًا. لكننا مرة أخرى نمنع الوصول إلى الطاولة من أجل تجنب السباقات (ظروف السباق) ، واتخاذ رهانات مؤكدة بشكل صحيح.

// Для блокирования отдельного философа.
// Инициализируется: new AutoResetEvent(true) для каждого.
private AutoResetEvent[] philosopherEvents;

// Для доступа к вилкам / доступ к столу.
private AutoResetEvent tableEvent = new AutoResetEvent(true);

// Рождение философа.
public void Run(int i, CancellationToken token)
{
    while (true)
    {
        TakeForks(i); // Ждет вилки.
        // Обед. Может быть и дольше.
        eatenFood[i] = (eatenFood[i] + 1) % (int.MaxValue - 1);
        PutForks(i); // Отдать вилки и разблокировать соседей.
        Think(i);
        if (token.IsCancellationRequested) break;
    }
}

// Ожидать вилки в блокировке.
void TakeForks(int i)
{
    bool hasForks = false;
    while (!hasForks) // Попробовать еще раз (блокировка не здесь).
    {
        // Исключающий доступ к столу, без гонок за вилками.
        tableEvent.WaitOne();
        if (forks[Left(i)] == 0 && forks[Right(i)] == 0)
            forks[Left(i)] = forks[Right(i)] = i + 1;
        hasForks = forks[Left(i)] == i + 1 && forks[Right(i)] == i + 1;
        if (hasForks)
            // Теперь философ поест, выйдет из цикла. Если Set 
            // вызван дважды, то значение true.
            philosopherEvents[i].Set();
        // Разблокировать одного ожидающего. После него значение tableEvent в false.
        tableEvent.Set(); 
        // Если имеет true, не блокируется, а если false, то будет ждать Set от соседа.
        philosopherEvents[i].WaitOne();
    }
}

// Отдать вилки и разблокировать соседей.
void PutForks(int i)
{
    tableEvent.WaitOne(); // Без гонок за вилками.
    forks[Left(i)] = 0;
    // Пробудить левого, а потом и правого соседа, либо AutoResetEvent в true.
    philosopherEvents[LeftPhilosopher(i)].Set();
    forks[Right(i)] = 0;
    philosopherEvents[RightPhilosopher(i)].Set();
    tableEvent.Set();
}

لفهم ما يحدث هنا ، ضع في اعتبارك الحالة التي فشل فيها الفيلسوف في أخذ مفترق الطرق ، فستكون أفعاله على النحو التالي. إنه ينتظر الوصول إلى الطاولة. بعد أن استلمها ، حاول أن يأخذ الشوك. لم ينجح في مبتغاه. يتيح الوصول إلى الجدول (الاستبعاد المتبادل). ويمر "الباب الدوار" (AutoResetEvent) (وهي مفتوحة في البداية). يدخل في الدورة مرة أخرى ، لأن ليس لديه شوك. يحاول أن يأخذهم ويتوقف عند "الباب الدوار". بعض الجيران الأكثر حظًا على اليمين أو اليسار ، بعد أن انتهوا من الأكل ، يفتح فيلسوفنا ، "يفتح بابه الدوار". فيلسوفنا يمر به (ويغلق خلفه) للمرة الثانية. حاول للمرة الثالثة أن يأخذ الشوك. حظ سعيد. ويجتاز الباب الدوار الخاص به لتناول العشاء.

عند وجود أخطاء عشوائية في مثل هذا الرمز (توجد دائمًا) ، على سبيل المثال ، يتم تحديد الجار بشكل غير صحيح أو يتم إنشاء نفس الكائن AutoResetEvent للجميع (Enumerable.Repeat) ، فإن الفلاسفة سينتظرون المطورين ، لأن العثور على أخطاء في مثل هذا الرمز مهمة صعبة للغاية. مشكلة أخرى في هذا الحل هي أنه لا يضمن أن بعض الفيلسوف لن يجوع.

حلول هجينة

لقد نظرنا في طريقتين للتوقيت ، عندما نبقى في وضع المستخدم والحلقة ، وعندما نمنع الخيط عبر النواة. الطريقة الأولى جيدة للأقفال القصيرة والثانية للأقفال الطويلة. غالبًا ما يكون من الضروري أولاً الانتظار لفترة وجيزة حتى يتغير المتغير في حلقة ، ثم حظر الخيط عندما يكون الانتظار طويلاً. يتم تنفيذ هذا النهج في ما يسمى ب. الهياكل الهجينة. فيما يلي نفس التركيبات المستخدمة في وضع kernel ، ولكن الآن مع حلقة وضع المستخدم: SemaphorSlim, ManualResetEventSlim إلخ. التصميم الأكثر شيوعًا هنا هو Monitor، لأن في C # هناك ملف مشهور lock بناء الجملة. Monitor هذا هو نفس الإشارة مع قيمة قصوى 1 (كائن المزامنة) ، ولكن مع دعم الانتظار في حلقة ، والتكرار ، ونمط متغير الشرط (المزيد عن ذلك أدناه) ، وما إلى ذلك. دعونا ننظر إلى حل معها.

// Спрячем объект для Монитора от всех, чтобы без дедлоков.
private readonly object _lock = new object();
// Время ожидания потока.
private DateTime?[] _waitTimes = new DateTime?[philosophersAmount];

public void Run(int i, CancellationToken token)
{
    while (true)
    {
        TakeForks(i);
        eatenFood[i] = (eatenFood[i] + 1) % (int.MaxValue - 1);
        PutForks(i);
        Think(i);
        if (token.IsCancellationRequested) break;
    }
}

// Наше сложное условие для Condition Variable паттерна.
bool CanIEat(int i)
{
    // Если есть вилки:
    if (forks[Left(i)] != 0 && forks[Right(i)] != 0)
        return false;
    var now = DateTime.Now;
    // Может, если соседи не более голодные, чем текущий.
    foreach(var p in new int[] {LeftPhilosopher(i), RightPhilosopher(i)})
        if (_waitTimes[p] != null && now - _waitTimes[p] > now - _waitTimes[i])
            return false;
    return true;
}

void TakeForks(int i)
{
    // Зайти в Монитор. То же самое: lock(_lock) {..}.
    // Вызываем вне try, чтобы возможное исключение выбрасывалось выше.
    bool lockTaken = false;
    Monitor.Enter(_lock, ref lockTaken);
    try
    {
        _waitTimes[i] = DateTime.Now;
        // Condition Variable паттерн. Освобождаем лок, если не выполненно 
        // сложное условие. И ждем пока кто-нибудь сделает Pulse / PulseAll.
        while (!CanIEat(i))
            Monitor.Wait(_lock); 
        forks[Left(i)] = i + 1;
        forks[Right(i)] = i + 1;
        _waitTimes[i] = null;
    }
    finally
    {
        if (lockTaken) Monitor.Exit(_lock);
    }
}

void PutForks(int i)
{
    // То же самое: lock (_lock) {..}.
    bool lockTaken = false;
    Monitor.Enter(_lock, ref lockTaken);
    try
    {
        forks[Left(i)] = 0;
        forks[Right(i)] = 0;
        // Освободить все потоки в очереди ПОСЛЕ вызова Monitor.Exit.
        Monitor.PulseAll(_lock); 
    }
    finally
    {
        if (lockTaken) Monitor.Exit(_lock);
    }
}

نحن هنا مرة أخرى نحظر الجدول بأكمله للوصول إلى الشوكات ، لكننا الآن نقوم بإلغاء حظر جميع الخيوط دفعة واحدة ، وليس الجيران عندما ينتهي شخص ما من تناول الطعام. أولئك. أولاً ، يأكل شخص ما ويسد الجيران ، وعندما ينتهي هذا الشخص ، لكنه يريد أن يأكل مرة أخرى على الفور ، يذهب إلى منع جيرانه ويوقظهم ، لأن. وقت الانتظار أقل.

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

У lock بناء الجملة له مفاجآت سيئة. يوصى باستخدامه Monitor مباشرة [ريختر] [إريك ليبرت]. واحد منهم هو ذلك lock دائما خارج Monitor، حتى إذا كان هناك استثناء ، ففي هذه الحالة يمكن أن يؤدي مؤشر ترابط آخر إلى تغيير حالة الذاكرة المشتركة. في مثل هذه الحالات ، غالبًا ما يكون من الأفضل الوصول إلى طريق مسدود أو إنهاء البرنامج بأمان بطريقة أو بأخرى. مفاجأة أخرى هي أن الشاشة تستخدم كتل التزامن (SyncBlock) ، والموجودة في جميع الأشياء. لذلك ، إذا تم تحديد كائن غير مناسب ، فيمكنك بسهولة الحصول على طريق مسدود (على سبيل المثال ، إذا قمت بإغلاق سلسلة داخلية). نحن نستخدم الشيء المخفي دائمًا لهذا الغرض.

يسمح لك نمط متغير الشرط بتنفيذ توقعات بعض الحالات المعقدة بدقة أكبر. في .NET ، هو غير مكتمل ، في رأيي ، لأن من الناحية النظرية ، يجب أن يكون هناك العديد من قوائم الانتظار على العديد من المتغيرات (كما في Posix Threads) ، وليس على lok واحد. ثم يمكن للمرء أن يصنعها لجميع الفلاسفة. ولكن حتى في هذا الشكل ، فإنه يسمح لك بتقليل الكود.

العديد من الفلاسفة أو async / await

حسنًا ، يمكننا الآن منع الخيوط بفعالية. لكن ماذا لو كان لدينا الكثير من الفلاسفة؟ 100؟ 10000؟ على سبيل المثال ، تلقينا 100000 طلب إلى خادم الويب. سيكون إنشاء مؤشر ترابط لكل طلب عبئًا لأن الكثير من المواضيع لن تعمل بالتوازي. لن يعمل إلا بقدر ما توجد نوى منطقية (لدي 4). وكل شخص آخر سيأخذ الموارد فقط. أحد الحلول لهذه المشكلة هو نمط عدم التزامن / الانتظار. فكرتها هي أن الوظيفة لا تحمل الخيط إذا كانت بحاجة إلى الانتظار حتى يستمر شيء ما. وعندما يفعل شيئًا ما ، فإنه يستأنف تنفيذه (ولكن ليس بالضرورة على نفس الموضوع!). في حالتنا ، سننتظر الشوكة.

SemaphoreSlim لهذا WaitAsync() طريقة. هنا تطبيق باستخدام هذا النمط.

// Запуск такой же, как раньше. Где-нибудь в программе:
Task.Run(() => Run(i, cancelTokenSource.Token));

// Запуск философа.
// Ключевое слово async -- компилятор транслирует этот метот в асинхронный.
public async Task Run(int i, CancellationToken token)
{
    while (true)
    {
        // await -- будем ожидать какого-то события.
        await TakeForks(i);
        // После await, продолжение возможно в другом потоке.
        eatenFood[i] = (eatenFood[i] + 1) % (int.MaxValue - 1);
        // Может быть несколько событий для ожидания.
        await PutForks(i);

        Think(i);

        if (token.IsCancellationRequested) break;
    }
}

async Task TakeForks(int i)
{
    bool hasForks = false;
    while (!hasForks)
    {
        // Взаимоисключающий доступ к столу:
        await _tableSemaphore.WaitAsync();
        if (forks[Left(i)] == 0 && forks[Right(i)] == 0)
        {
            forks[Left(i)] = i+1;
            forks[Right(i)] = i+1;
            hasForks = true;
        }
        _tableSemaphore.Release();
        // Будем ожидать, чтобы сосед положил вилки:
        if (!hasForks)
            await _philosopherSemaphores[i].WaitAsync();
    }
}

// Ждем доступа к столу и кладем вилки.
async Task PutForks(int i)
{
    await _tableSemaphore.WaitAsync();
    forks[Left(i)] = 0;
    // "Пробудить" соседей, если они "спали".
    _philosopherSemaphores[LeftPhilosopher(i)].Release();
    forks[Right(i)] = 0;
    _philosopherSemaphores[RightPhilosopher(i)].Release();
    _tableSemaphore.Release();
}

الطريقة مع async / await تُترجم إلى آلة حالة صعبة تعيد على الفور ما هو داخلي Task. من خلاله ، يمكنك انتظار اكتمال الطريقة وإلغائها وكل شيء آخر يمكنك القيام به مع Task. داخل الطريقة ، تتحكم آلة الحالة في التنفيذ. خلاصة القول هي أنه إذا لم يكن هناك تأخير ، فسيكون التنفيذ متزامنًا ، وإذا كان هناك ، فسيتم تحرير الخيط. لفهم هذا بشكل أفضل ، من الأفضل النظر إلى آلة الدولة هذه. يمكنك إنشاء سلاسل من هؤلاء async / await طُرق.

دعونا نختبر. عمل 100 فلاسفة على آلة ذات 4 نوى منطقية ، 8 ثوان. كان الحل السابق مع Monitor يعمل فقط على أول 4 سلاسل ولم يتم تشغيل الباقي على الإطلاق. كان كل من هذه الخيوط الأربعة خاملًا لحوالي 4 مللي ثانية. وشغل الحل غير المتزامن / انتظار كل 2 ، بمتوسط ​​انتظار يبلغ 100 ثانية لكل منهما. بالطبع ، في الأنظمة الحقيقية ، يعد الخمول لمدة 6.8 ثوانٍ غير مقبول ومن الأفضل عدم معالجة العديد من الطلبات مثل هذا. تبين أن الحل مع Monitor غير قابل للتطوير على الإطلاق.

اختتام

كما ترى من هذه الأمثلة الصغيرة ، يدعم .NET العديد من بنيات المزامنة. ومع ذلك ، ليس من الواضح دائمًا كيفية استخدامها. آمل أن تكون هذه المقالة مفيدة. في الوقت الحالي ، هذه هي النهاية ، ولكن لا يزال هناك الكثير من الأشياء المثيرة للاهتمام المتبقية ، على سبيل المثال ، مجموعات الخيوط الآمنة ، وتدفق بيانات TPL ، والبرمجة التفاعلية ، ونموذج معاملة البرامج ، وما إلى ذلك.

مصادر

• تصور التدفق: مصور التزامن
• MSDN: خيوط, أنماط البرمجة غير المتزامنة وغيرها الكثير وآخرون.
• [ريختر] - CLR عبر C # ، جيفري ريختر
• [إريك ليبرت] - حول القفل
• شفرة المصدر
• اللوحة - "Dance، بين swords"، G. Semiradsky

المصدر: www.habr.com