풍부한 철학자 또는 경쟁력 있는 .NET 프로그래밍

Philosophers Dining Problem을 예로 들어 .Net에서 동시 및 병렬 프로그래밍이 어떻게 작동하는지 살펴보겠습니다. 계획은 스레드/프로세스의 동기화에서 행위자 모델(다음 부분에서)까지입니다. 이 기사는 처음 아는 사람이나 지식을 새로 고치는 데 유용할 수 있습니다.

왜 그렇게합니까? 트랜지스터는 최소 크기에 도달하고 무어의 법칙은 빛의 속도 제한에 따라 달라지므로 수가 증가하면 더 많은 트랜지스터를 만들 수 있습니다. 동시에 데이터의 양이 증가하고 사용자는 시스템에서 즉각적인 응답을 기대합니다. 이러한 상황에서 하나의 실행 스레드가 있는 "정상적인" 프로그래밍은 더 이상 효과적이지 않습니다. 어떻게든 동시 또는 동시 실행 문제를 해결해야 합니다. 더욱이 이 문제는 스레드 수준, 프로세스 수준, 네트워크(분산 시스템)의 기계 수준 등 다양한 수준에서 존재합니다. .NET에는 이러한 문제를 빠르고 효율적으로 해결하기 위한 고품질의 오랜 테스트를 거친 기술이 있습니다.

태스크

Edsger Dijkstra는 이미 1965년에 그의 학생들에게 이 문제를 제기했습니다. 확립된 공식은 다음과 같습니다. 특정(보통 54명) 수의 철학자와 같은 수의 포크가 있습니다. 그들은 그들 사이에 포크를 끼고 원탁에 앉습니다. 철학자는 끝없는 음식을 접시에서 먹고 생각하거나 기다릴 수 있습니다. 철학자를 먹으려면 포크 두 개를 가져와야 합니다(마지막 포크는 첫 번째 포크와 공유). 포크를 집는 것과 내려놓는 것은 별개의 행동입니다. 모든 철학자는 침묵합니다. 과제는 XNUMX년이 지나도 모두가 생각하고 채워지는 그런 알고리즘을 찾는 것입니다.

먼저 공유 공간을 활용하여 이 문제를 해결해 보도록 하겠습니다. 포크는 일반 테이블에 놓여 있고 철학자들은 포크가 있을 때 가져갔다가 다시 놓습니다. 여기에 동기화에 문제가 있습니다. 언제 확실하게 베팅해야 합니까? 포크가 없다면? 기타 그러나 먼저 철학자부터 시작하겠습니다.

스레드를 시작하기 위해 다음을 통해 스레드 풀을 사용합니다. 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은 이러한 작업의 수에 따라 스레드를 만들거나 제거합니다. 모든 AppDomain을 위한 하나의 풀. 이 풀은 거의 항상 사용해야 합니다. 스레드, 대기열 등을 생성, 삭제하는 데 신경 쓸 필요가 없습니다. 풀 없이도 가능하지만 직접 사용해야 합니다. Thread, 이것은 스레드의 우선 순위를 변경해야 하는 경우, 긴 작업이 있는 경우, Foreground 스레드 등의 경우에 유용합니다.

즉, System.Threading.Tasks.Task 클래스는 동일합니다 Thread, 그러나 모든 종류의 편의: 다른 작업 블록 후에 작업을 실행하고, 함수에서 반환하고, 편리하게 중단하는 등의 기능이 있습니다. 등 async / await 구성을 지원하는 데 필요합니다(태스크 기반 비동기 패턴, 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배 적게 먹는 것으로 나타났습니다. 따라서 코드의 작은 오류는 성능 저하로 이어집니다. 모든 철학자가 왼쪽 포크를 취하고 오른쪽 포크가 없으며 왼쪽을 놓고 기다렸다가 다시 왼쪽을 취하는 등의 드문 상황이 가능하다는 점도 여기서 주목할 가치가 있습니다. 이 상황은 또한 교착 상태에 가까운 기아 상태입니다. 나는 그것을 반복하지 못했습니다. 아래는 두 명의 나쁜 철학자가 포크를 모두 가져갔고 두 명의 좋은 철학자가 굶주린 상황에 대한 사진입니다.

여기에서 스레드가 때때로 깨어나 리소스를 가져오려고 시도하는 것을 볼 수 있습니다. XNUMX개의 코어 중 XNUMX개는 아무 작업도 수행하지 않습니다(위의 녹색 그래프).

철학자의 죽음

글쎄요, 철학자들의 영광스러운 저녁 식사를 방해 할 수있는 또 다른 문제는 그들 중 한 명이 손에 포크를 들고 갑자기 죽는 경우입니다 (그리고 그들은 그를 그렇게 묻을 것입니다). 그러면 이웃들은 저녁을 먹지 않고 남게 될 것입니다. 이 경우에 대한 예제 코드를 직접 생각해 낼 수 있습니다. 예를 들어 버려집니다. 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 (거기서 사용됨). 이제 그는 기다리는 사람들의 수를 세는 방법, 그들을 조금 재우는 방법 등을 알고 있습니다. 기타 일반적으로 가능한 모든 것을 최적화합니다. 그러나 이것은 여전히 ​​프로세서 리소스를 잡아먹고 흐름을 유지하는 동일한 활성 주기라는 점을 기억해야 합니다. 철학자 중 한 사람이 다른 사람보다 우선순위가 높지만 골든 포크(우선순위 역전 문제)가 없는 경우 기아로 이어질 수 있습니다. . 따라서 제XNUMX자 호출, 중첩 잠금 및 기타 예상치 못한 상황 없이 공유 메모리의 매우 짧은 변경에만 사용합니다.

그리기 SpinLock. 시냇물은 황금 포크를 위해 끊임없이 "싸움"하고 있습니다. 실패가 있습니다 - 그림에서 선택한 영역입니다. 코어는 완전히 활용되지 않습니다. 이 2개의 스레드에 의해 약 3/XNUMX만 사용됩니다.

여기서 또 다른 해결책은 Interlocked.CompareExchange 위의 코드(굶주린 철학자에서)에 표시된 것과 동일한 활성 대기를 사용하지만 이미 말했듯이 이론적으로 차단으로 이어질 수 있습니다.

약 Interlocked 뿐만 아니라 CompareExchange, 뿐만 아니라 원자 읽기 및 쓰기를 위한 다른 방법도 있습니다. 그리고 변경의 반복을 통해 다른 스레드가 변경을 할 시간이 있는 경우(read 1, read 2, write 2, write 1 불량) 단일 값에 대한 복잡한 변경에 사용할 수 있습니다(Interlocked Anything 패턴). .

커널 모드 솔루션

루프에서 리소스 낭비를 방지하기 위해 스레드를 차단하는 방법을 살펴보겠습니다. 즉, 우리의 예를 계속해서 웨이터가 어떻게 철학자를 잠들게 하고 필요할 때만 깨우는지 봅시다. 먼저 운영 체제의 커널 모드를 통해 이를 수행하는 방법을 살펴보겠습니다. 모든 구조는 종종 사용자 공간의 구조보다 느립니다. 예를 들어 몇 배 더 느립니다. AutoResetEvent 아마도 53배 더 느릴 것입니다. SpinLock [리히터]. 그러나 도움을 받으면 관리 여부에 관계없이 시스템 전체에서 프로세스를 동기화할 수 있습니다.

여기서 기본 구성은 반세기 전에 Dijkstra가 제안한 세마포어입니다. 세마포어는 간단히 말해서 시스템에서 관리하는 양의 정수이며 증가 및 감소라는 두 가지 작업이 있습니다. XNUMX으로 감소하지 못하면 호출 스레드가 차단됩니다. 다른 활성 스레드/프로세스에 의해 숫자가 증가하면 스레드를 건너뛰고 세마포어는 전달된 숫자만큼 다시 감소합니다. 세마포어가 있는 병목 현상에 있는 기차를 상상할 수 있습니다. .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(뮤텍스)인 동일한 세마포어이지만 루프에서 대기, 재귀, 조건 변수 패턴(아래에서 자세히 설명) 등을 지원합니다. 솔루션을 살펴보겠습니다.

// Спрячем объект для Монитора от всех, чтобы без дедлоков.
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 직접 [Richter] [Eric Lippert]. 그 중 하나는 lock 항상 밖으로 Monitor, 예외가 있는 경우에도 다른 스레드가 공유 메모리 상태를 변경할 수 있습니다. 이러한 경우 교착 상태에 빠지거나 프로그램을 안전하게 종료하는 것이 좋습니다. 또 다른 놀라운 점은 Monitor가 동기화 블록(SyncBlock), 모든 개체에 존재합니다. 따라서 부적절한 개체를 선택하면 쉽게 교착 상태에 빠질 수 있습니다(예: 인턴된 문자열에 잠그는 경우). 이를 위해 항상 숨겨진 개체를 사용합니다.

조건 변수 패턴을 사용하면 복잡한 조건에 대한 예상을 보다 간결하게 구현할 수 있습니다. .NET에서는 불완전하다고 생각합니다. 이론적으로 하나의 로크가 아니라 여러 변수(Posix 스레드에서와 같이)에 여러 큐가 있어야 합니다. 그런 다음 모든 철학자를 위해 만들 수 있습니다. 하지만 이 형태에서도 코드를 줄일 수 있습니다.

많은 철학자 또는 async / await

이제 효과적으로 스레드를 차단할 수 있습니다. 그러나 만약 우리에게 많은 철학자들이 있다면 어떨까요? 100? 10000? 예를 들어 웹 서버에 대한 100000개의 요청을 받았습니다. 각 요청에 대해 스레드를 생성하는 것은 오버헤드가 될 것입니다. 너무 많은 스레드가 병렬로 실행되지 않습니다. 논리 코어가 있는 만큼만 실행됩니다(저는 4개 있습니다). 그리고 다른 모든 사람들은 자원을 빼앗을 것입니다. 이 문제에 대한 한 가지 해결책은 async/await 패턴입니다. 그 아이디어는 무언가가 계속될 때까지 기다려야 하는 경우 함수가 스레드를 보유하지 않는다는 것입니다. 그리고 어떤 작업을 수행하면 실행을 다시 시작합니다(반드시 동일한 스레드에서 수행할 필요는 없음!). 우리의 경우 포크를 기다릴 것입니다.

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. 이를 통해 메서드가 완료될 때까지 기다리거나 취소하는 등 작업으로 수행할 수 있는 모든 작업을 수행할 수 있습니다. 메서드 내에서 상태 시스템이 실행을 제어합니다. 결론은 지연이 없으면 실행이 동기식이며 지연이 있으면 스레드가 해제된다는 것입니다. 이를 더 잘 이해하려면 이 상태 머신을 살펴보는 것이 좋습니다. 다음에서 체인을 만들 수 있습니다. async / await 행동 양식.

테스트 해보자. 100개의 논리 코어가 있는 기계에서 4명의 철학자 작업, 8초. Monitor를 사용한 이전 솔루션은 처음 4개의 스레드만 실행했고 나머지는 전혀 실행하지 않았습니다. 이 4개의 스레드는 각각 약 2ms 동안 유휴 상태였습니다. 그리고 async/await 솔루션은 평균 대기 시간이 각각 100초인 6.8개를 모두 실행했습니다. 물론 실제 시스템에서는 6초 동안 유휴 상태가 되는 것은 허용되지 않으며 이렇게 많은 요청을 처리하지 않는 것이 좋습니다. Monitor를 사용한 솔루션은 전혀 확장할 수 없는 것으로 판명되었습니다.

결론

이러한 작은 예에서 알 수 있듯이 .NET은 많은 동기화 구조를 지원합니다. 그러나 사용 방법이 항상 명확한 것은 아닙니다. 이 기사가 도움이 되었기를 바랍니다. 지금은 이것이 끝이지만 여전히 스레드 안전 컬렉션, TPL 데이터 흐름, 반응형 프로그래밍, 소프트웨어 트랜잭션 모델 등과 같은 흥미로운 것들이 많이 남아 있습니다.

소스

• 흐름 시각화: 동시성 시각화기
• MSDN : 스레딩, 비동기 프로그래밍 패턴 기타 여러 다른 사람
• [Richter] - C#을 통한 CLR, Jeffrey Richter
• [에릭 리퍼트] - 잠금 정보
• 소스 코드
• 사진 - "칼 사이의 춤", G. Semiradsky

출처 : habr.com