🥇Fat Philosophers หรือ Competitive .NET Programming

มาดูกันว่าการเขียนโปรแกรมพร้อมกันและขนานทำงานอย่างไรใน .Net โดยใช้ Philosophers Dining Problem เป็นตัวอย่าง แผนนี้ตั้งแต่การซิงโครไนซ์เธรด / กระบวนการไปจนถึงโมเดลนักแสดง (ในส่วนต่อไปนี้) บทความนี้อาจเป็นประโยชน์สำหรับคนรู้จักครั้งแรกหรือเพื่อฟื้นฟูความรู้ของคุณ

ทำไมถึงทำอย่างนั้น? ทรานซิสเตอร์ถึงขนาดต่ำสุด กฎของมัวร์ขึ้นอยู่กับการจำกัดความเร็วของแสง ดังนั้นจึงสังเกตได้ว่าจำนวนเพิ่มขึ้น สามารถสร้างทรานซิสเตอร์ได้มากขึ้น ในขณะเดียวกัน ปริมาณข้อมูลก็เพิ่มขึ้น และผู้ใช้ก็คาดหวังการตอบสนองทันทีจากระบบ ในสถานการณ์เช่นนี้ การเขียนโปรแกรม "ปกติ" เมื่อเรามีหนึ่งเธรดการดำเนินการ จะไม่มีประสิทธิภาพอีกต่อไป คุณต้องแก้ปัญหาของการดำเนินการพร้อมกันหรือพร้อมกัน ยิ่งกว่านั้น ปัญหานี้มีอยู่ในระดับที่แตกต่างกัน: ที่ระดับของเธรด ที่ระดับของกระบวนการ ที่ระดับของเครื่องในเครือข่าย (ระบบกระจาย) .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แต่ด้วยความสะดวกสบายทุกประเภท: ความสามารถในการเรียกใช้งานหลังจากบล็อกของงานอื่นๆ ส่งคืนจากฟังก์ชันต่างๆ ขัดขวางการทำงานอย่างสะดวก และอื่นๆ อีกมากมาย ฯลฯ จำเป็นสำหรับการสนับสนุน async / wait การก่อสร้าง (รูปแบบ Asynchronous ตามงาน, น้ำตาลวากยสัมพันธ์สำหรับการรอการดำเนินการ IO) เราจะพูดถึงเรื่องนี้ในภายหลัง

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

สิ่งสำคัญเกี่ยวกับรหัสนี้คือนักปรัชญา 5 ใน XNUMX คนลืมที่จะวางส้อมซ้ายลง และปรากฎว่าพวกเขากินอาหารมากขึ้นในขณะที่คนอื่นเริ่มอดอาหารแม้ว่าหัวข้อจะมีลำดับความสำคัญเท่ากันก็ตาม ที่นี่พวกเขาไม่หิวโหยอย่างสมบูรณ์เพราะ นักปรัชญาที่ไม่ดีบางครั้งเอาส้อมของพวกเขากลับ ปรากฎว่าคนดีกินน้อยกว่าคนเลวประมาณ XNUMX เท่า ดังนั้นข้อผิดพลาดเล็กน้อยในโค้ดจึงส่งผลให้ประสิทธิภาพลดลง นอกจากนี้ยังเป็นที่น่าสังเกตว่ามีสถานการณ์ที่หายากที่เป็นไปได้เมื่อนักปรัชญาทุกคนใช้ทางแยกซ้าย ไม่มีทางขวา พวกเขาไปทางซ้าย รอ ไปทางซ้ายอีกครั้ง ฯลฯ สถานการณ์นี้ยังเป็นความอดอยากเหมือนการหยุดชะงัก ฉันล้มเหลวที่จะทำซ้ำ ด้านล่างนี้เป็นภาพของสถานการณ์ที่นักปรัชญาที่ไม่ดีสองคนใช้ส้อมทั้งคู่และคนดีสองคนกำลังหิวโหย

ที่นี่คุณจะเห็นว่าเธรดตื่นขึ้นในบางครั้งและพยายามรับทรัพยากร สองในสี่คอร์ไม่ได้ทำอะไรเลย (กราฟสีเขียวด้านบน)

ความตายของนักปรัชญา

ปัญหาอีกประการหนึ่งที่สามารถขัดจังหวะอาหารค่ำอันรุ่งโรจน์ของนักปรัชญาได้คือถ้าคนใดคนหนึ่งเสียชีวิตด้วยส้อมในมือของเขาอย่างกระทันหัน (และพวกเขาจะฝังเขาไว้อย่างนั้น) จากนั้นเพื่อนบ้านจะถูกทิ้งไว้โดยไม่รับประทานอาหารกลางวัน คุณสามารถสร้างตัวอย่างรหัสสำหรับกรณีนี้ได้ด้วยตัวเอง ตัวอย่างเช่น มันถูกโยนออกไป NullReferenceException หลังจากที่นักปรัชญาใช้ส้อม และอย่างไรก็ตามข้อยกเว้นจะไม่ถูกจัดการและรหัสการโทรจะไม่จับได้ (สำหรับสิ่งนี้ AppDomain.CurrentDomain.UnhandledException และอื่น ๆ.). ดังนั้น ตัวจัดการข้อผิดพลาดจึงจำเป็นในเธรดเองและด้วยการยกเลิกอย่างสง่างาม

บริกร

เอาล่ะ เราจะแก้ปัญหาการหยุดชะงัก ความอดอยาก และความตายนี้อย่างไร เราจะอนุญาตให้นักปรัชญาเพียงคนเดียวเข้าถึงทางแยก เพิ่มการยกเว้นเธรดร่วมกันสำหรับสถานที่นี้ ทำอย่างไร? สมมติว่ามีพนักงานเสิร์ฟอยู่ข้างๆ นักปรัชญาที่อนุญาตให้นักปรัชญาคนใดคนหนึ่งรับส้อม เราจะสร้างบริกรนี้ได้อย่างไร และนักปรัชญาจะถามเขาอย่างไร คำถามนั้นน่าสนใจ

วิธีที่ง่ายที่สุดคือเมื่อนักปรัชญาจะถามบริกรตลอดเวลาเพื่อเข้าถึงส้อม เหล่านั้น. ตอนนี้นักปรัชญาจะไม่รอส้อมใกล้ ๆ แต่รอหรือถามบริกร ในตอนแรกเราใช้เฉพาะ User Space สำหรับสิ่งนี้ เราไม่ใช้การขัดจังหวะเพื่อเรียกขั้นตอนใด ๆ จากเคอร์เนล (เกี่ยวกับพวกเขาด้านล่าง)

โซลูชันในพื้นที่ผู้ใช้

ที่นี่เราจะทำแบบเดียวกับที่เราเคยทำด้วยส้อมเดียวและนักปรัชญาสองคน เราจะหมุนเป็นวงกลมและรอ แต่ตอนนี้จะเป็นนักปรัชญาทั้งหมดและมีเพียงทางแยกเดียวเท่านั้นนั่นคือ อาจกล่าวได้ว่านักปรัชญาเท่านั้นที่รับ "ส้อมทอง" นี้จากบริกรเท่านั้นที่จะกิน สำหรับสิ่งนี้เราใช้ 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 ไม่ดี) สามารถใช้สำหรับการเปลี่ยนแปลงที่ซับซ้อนเป็นค่าเดียว (รูปแบบ Interlocked Anything) .

โซลูชันโหมดเคอร์เนล

เพื่อหลีกเลี่ยงการสูญเสียทรัพยากรในลูป มาดูกันว่าเราจะบล็อกเธรดได้อย่างไร กล่าวอีกนัยหนึ่ง ทำตามตัวอย่างของเรา มาดูกันว่าบริกรทำให้นักปรัชญาหลับได้อย่างไร และปลุกเขาเมื่อจำเป็นเท่านั้น ก่อนอื่นมาดูวิธีการทำสิ่งนี้ผ่านโหมดเคอร์เนลของระบบปฏิบัติการ โครงสร้างทั้งหมดมักจะช้ากว่าโครงสร้างในพื้นที่ของผู้ใช้ ตัวอย่างเช่นช้าลงหลายเท่า 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) จากนั้นนักปรัชญาจะรอนักพัฒนาเพราะ การค้นหาข้อผิดพลาดในรหัสดังกล่าวค่อนข้างยาก ปัญหาอีกประการหนึ่งของวิธีแก้ปัญหานี้คือไม่ได้รับประกันว่านักปรัชญาบางคนจะไม่หิวโหย

โซลูชั่นไฮบริด

เราได้พิจารณาสองวิธีในการกำหนดเวลา เมื่อเราอยู่ในโหมดผู้ใช้และวนซ้ำ และเมื่อเราบล็อกเธรดผ่านเคอร์เนล วิธีแรกนั้นดีสำหรับการล็อคแบบสั้นวิธีที่สองสำหรับการล็อคแบบยาว บ่อยครั้งที่จำเป็นต้องรอสักครู่เพื่อให้ตัวแปรเปลี่ยนแปลงในลูป จากนั้นจึงบล็อกเธรดเมื่อการรอนั้นยาวนาน วิธีการนี้ถูกนำมาใช้ในสิ่งที่เรียกว่า โครงสร้างไฮบริด ต่อไปนี้เป็นโครงสร้างเดียวกันกับโหมดเคอร์เนล แต่ตอนนี้มีลูปโหมดผู้ใช้: SemaphorSlim, ManualResetEventSlim เป็นต้น การออกแบบที่ได้รับความนิยมมากที่สุดคือ Monitor, เพราะ ใน C# มีที่รู้จักกันดี lock ไวยากรณ์ Monitor นี่คือสัญญาณเดียวกันกับค่าสูงสุด 1 (mutex) แต่รองรับการรอในลูป การเรียกซ้ำ รูปแบบตัวแปรเงื่อนไข (เพิ่มเติมด้านล่าง) เป็นต้น มาดูวิธีแก้ปัญหากัน

// Спрячем объект для Монитора от всех, чтобы без дедлоков.
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แม้ว่าจะมีข้อยกเว้น ในกรณีนี้ เธรดอื่นสามารถเปลี่ยนสถานะของหน่วยความจำที่ใช้ร่วมกันได้ ในกรณีเช่นนี้ มักจะเป็นการดีกว่าหากไปที่การหยุดชะงักหรือยุติโปรแกรมอย่างปลอดภัย สิ่งที่น่าประหลาดใจอีกอย่างคือ Monitor ใช้บล็อกการซิงโครไนซ์ (SyncBlock) ซึ่งมีอยู่ในวัตถุทั้งหมด ดังนั้น หากมีการเลือกออบเจ็กต์ที่ไม่เหมาะสม คุณจะเกิดการหยุดชะงักได้ง่าย (เช่น หากคุณล็อกสตริงภายใน) เราใช้วัตถุที่ซ่อนอยู่เสมอสำหรับสิ่งนี้

รูปแบบตัวแปรเงื่อนไขช่วยให้คุณดำเนินการตามความคาดหวังของเงื่อนไขที่ซับซ้อนบางอย่างได้รัดกุมยิ่งขึ้น ในความคิดของฉัน .NET มันไม่สมบูรณ์เพราะ ตามทฤษฎีแล้ว ควรมีหลายคิวในหลายตัวแปร (เช่นใน Posix Threads) ไม่ใช่ใน lok เดียว จากนั้นใคร ๆ ก็สามารถสร้างมันขึ้นมาสำหรับนักปรัชญาทุกคน แต่แม้ในรูปแบบนี้จะช่วยให้คุณลดรหัสได้

นักปรัชญาหลายคนหรือ `async` / `await`

เอาล่ะ ตอนนี้เราสามารถบล็อกเธรดได้อย่างมีประสิทธิภาพแล้ว แต่ถ้าเรามีนักปรัชญาจำนวนมากล่ะ? 100? 10000? ตัวอย่างเช่น เราได้รับคำขอ 100000 รายการไปยังเว็บเซิร์ฟเวอร์ การสร้างเธรดสำหรับแต่ละคำขอจะมีค่าใช้จ่ายสูง เนื่องจาก เธรดจำนวนมากจะไม่ทำงานแบบขนาน จะทำงานได้มากเท่าที่มีแกนตรรกะ (ฉันมี 4) และทุกคนก็จะเอาทรัพยากรไป วิธีหนึ่งในการแก้ปัญหานี้คือรูปแบบ async / wait แนวคิดของมันคือฟังก์ชันไม่เก็บเธรดหากจำเป็นต้องรอให้ดำเนินการต่อ และเมื่อมันทำบางอย่าง มันจะดำเนินการต่อ (แต่ไม่จำเป็นต้องอยู่ในเธรดเดียวกัน!) ในกรณีของเราเราจะรอส้อม

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 มิลลิวินาที และโซลูชัน async / wait รันทั้งหมด 100 ครั้งโดยมีการรอเฉลี่ย 6.8 วินาทีในแต่ละครั้ง แน่นอนว่าในระบบจริง การไม่ได้ใช้งานเป็นเวลา 6 วินาทีเป็นสิ่งที่ยอมรับไม่ได้ และเป็นการดีกว่าที่จะไม่ดำเนินการตามคำขอจำนวนมากเช่นนี้ วิธีแก้ปัญหาด้วย Monitor ไม่สามารถปรับขนาดได้เลย

ข้อสรุป

ดังที่คุณเห็นจากตัวอย่างเล็กๆ เหล่านี้ .NET รองรับโครงสร้างการซิงโครไนซ์จำนวนมาก อย่างไรก็ตาม วิธีการใช้ไม่ชัดเจนเสมอไป ฉันหวังว่าบทความนี้จะเป็นประโยชน์ สำหรับตอนนี้ ก็จบลงแต่ยังมีสิ่งที่น่าสนใจเหลืออยู่อีกมาก เช่น thread-safe collections, TPL Dataflow, Reactive programming, Software Transaction model เป็นต้น

แหล่งที่มา

การแสดงภาพการไหล: Visualizer การทำงานพร้อมกัน
MSDN: Threading, รูปแบบการเขียนโปรแกรมแบบอะซิงโครนัส และอื่น ๆ อีกมากมาย ดร
[ริกเตอร์] - CLR ผ่าน C# เจฟฟรีย์ ริชเตอร์
[เอริค ลิปเพิร์ต] - เกี่ยวกับการล็อค
รหัสที่มา
รูปภาพ - "การเต้นรำท่ามกลางดาบ", G. Semiradsky

ที่มา: will.com

Well-Fed Philosophers หรือการเขียนโปรแกรม .NET ที่แข่งขันได้

งาน

ปัญหาการซิงค์

นักปรัชญาที่ถูกบล็อก

ความหิวโหยของนักปรัชญา

ความตายของนักปรัชญา

บริกร

โซลูชันในพื้นที่ผู้ใช้

โซลูชันโหมดเคอร์เนล

โซลูชั่นไฮบริด

นักปรัชญาหลายคนหรือ `async` / `await`

ข้อสรุป

แหล่งที่มา

เพิ่มความคิดเห็น ยกเลิกการตอบ

Well-Fed Philosophers หรือการเขียนโปรแกรม .NET ที่แข่งขันได้

งาน

ปัญหาการซิงค์

นักปรัชญาที่ถูกบล็อก

ความหิวโหยของนักปรัชญา

ความตายของนักปรัชญา

บริกร

โซลูชันในพื้นที่ผู้ใช้

โซลูชันโหมดเคอร์เนล

โซลูชั่นไฮบริด

นักปรัชญาหลายคนหรือ async / await

ข้อสรุป

แหล่งที่มา

เพิ่มความคิดเห็น ยกเลิกการตอบ

นักปรัชญาหลายคนหรือ `async` / `await`