[Перевод] Джозеф Албахари. Работа с потоками в C#. Часть 4

Это перевод четвертой части “Part 4. Advanced Threading” из замечательной серии статей Джозефа Албахари (Joseph Albahari)  о работе с потоками в языке C# и, как и при переводе третьей части, сейчас я публикую только, что появилось с выходом нового издания книги  “C# 4.0 in Nutshell: The Definitive Reference” и еще не опубликовано на русском языке на rsdn.ru.

Класс Barrier

Класс Barrier – это сигнальная конструкция, которая появилась в .Net Framework 4.0. Он реализует барьер потока исполнения, который позволяет множеству потоков встречаться в определенном месте во времени. Этот класс очень быстрый и эффективный, поскольку построен на основе Wait, Pulse и спин-блокировок.

Для использования этого класса необходимо:

1. Создать экземпляр, указав количество потоков, которые будут встречаться одновременно(вы можете изменить это значение позднее путем вызова методов AddParticipants/RemoveParticipants).
2. Каждый поток, ожидающий встречи, должен вызвать метод SignalAndWait.

Создание экземпляра класса Barrier со значением 3 приведет к тому, что вызов метода SignalAndWait блокируется до тех пор, пока этот метод не будет вызван трижды. Но, в отличие от CountdownEvent, он автоматически запускает этот процесс заново: вызов метода SignalAndWait снова блокируется до тех пор, пока он не вызовется трижды. Это позволяет идти нескольким потокам «в ногу» во время обработки набора задач.

В следующем примере, каждый из трех потоков выводит числа от 0 до 4, шагая в ногу с другими потоками:

static Barrier _barrier = new Barrier (3);
 
static void Main()
{
  new Thread (Speak).Start();
  new Thread (Speak).Start();
  new Thread (Speak).Start();
}
 
static void Speak()
{
  for (int i = 0; i < 5; i++)
  {
    Console.Write (i + " ");
    _barrier.SignalAndWait();
  }
}

0 0 0 1 1 1 2 2 2 3 3 3 4 4 4

Очень полезной возможностью класса Barrier является возможность указать при его конструировании постдействие (post-phase action). Этот делегат выполняется после того, как метод SignalAndWait вызван n раз, но перед тем, как потоки будут разблокированы. Если в нашем примере мы создадим экземпляр класса барьера следующим образом:

static Barrier _barrier = new Barrier (3, barrier => Console.WriteLine());

то вывод на экран будет таким:

0 0 0 
1 1 1 
2 2 2 
3 3 3 
4 4 4 

Постдействие может быть полезным при объединении данных рабочих потоков. В этом случае не нужно беспокоиться о порядке выполнения, поскольку до окончания выполнения этого действия все рабочие потоки остаются заблокированными.

Блокировки чтения/записи

Довольно часто экземпляры некоторых типов являются потокобезопасными при параллельных операциях чтениях, но не являются таковыми при параллельных обновлениях (при параллельных операциях чтения и обновления). Это также относится к таким ресурсам, как файлы. И хотя защита экземпляров таких типов простой эксклюзивной блокировкой для всех режимов доступа, решает проблему параллельного доступа, это может необоснованно ограничить параллелизм, когда к экземпляру обращаются много читателей при небольшом количестве обновлений. Типичным примером, когда может возникнуть такая ситуация – это сервер бизнес-приложений, в котором часто используемые данные кэшируются в статических полях для последующего быстрого доступа. Класс ReaderWriteLockSlim разработан для обеспечения максимальной доступности ресурса именно для этого случая.

Класс ReaderWriterLockSlim появился в .Net Framework 3.5 и предназначен для замены старого «тяжеловесного» класса ReaderWriterLock. Этот класс обладает аналогичной функциональности, но он в несколько раз медленнее и содержит ошибки проектирования в механизме обработки обновления блокировки.

Оба эти класса предусматривают два типа блокировки – блокировка чтения и блокировка записи:

• Блокировка записи является полностью эксклюзивной.
• Блокировка чтения совместима с другими блокировками чтения.

So, a thread holding a write lock blocks all other threads trying to obtain a read or write lock (and vice versa). But if no thread holds a write lock, any number of threads may concurrently obtain a read lock.

ReaderWriterLockSlim defines the following methods for obtaining and releasing read/write locks:

Т.е. поток, захвативший блокировку записи, блокирует все другие потоки, пытающиеся получить блокировку чтения или записи (и наоборот). Но если ни один поток не держит блокировку записи, любое количество потоков могут одновременно захватить блокировку чтения.

В классе ReaderWriterLockSLim определены следующие методы для захвата и освобождения блокировок чтения/записи:

public void EnterReadLock();
public void ExitReadLock();
public void EnterWriteLock();
public void ExitWriteLock();

В дополнение к этим методом, существуют также “Try” версии всех EnterXXX методов, которые принимают тайм-аут в качестве параметра, аналогично методу Monitor.TryEnter (таймауты могут часто возникать при работе с высокой степенью конкуренции). Класс ReaderWriterLock содержит аналогичные методы с именами AcquireXXX и ReleaseXXX, которые в случае возникновения тайм-аута генерируют ApplicationException, а не возвращают false.

Следующий код демонстрирует использование класса ReaderWriterLockSlim. Три потока постоянно читают список, в то время как два других потока добавляют случайное значение в него каждую секунду. Блокировка чтения защищает читателей списка, а блокировка записи защищает писателей.

class SlimDemo
{
    static ReaderWriterLockSlim _rw = new ReaderWriterLockSlim();
    static List<int> _items = new List<int>();
    static Random _rand = new Random();
 
    static void Main()
    {
        new Thread(Read).Start();
        new Thread(Read).Start();
        new Thread(Read).Start();
 
        new Thread(Write).Start("A");
        new Thread(Write).Start("B");
    }
 
    static void Read()
    {
        while (true)
        {
            _rw.EnterReadLock();
            foreach (int i in _items) Thread.Sleep(10);
            _rw.ExitReadLock();
        }
    }
 
    static void Write(object threadID)
    {
        while (true)
        {
            int newNumber = GetRandNum(100);
            _rw.EnterWriteLock();
            _items.Add(newNumber);
            _rw.ExitWriteLock();
            Console.WriteLine("Thread " + threadID + " added " + newNumber);
            Thread.Sleep(100);
        }
    }
 
    static int GetRandNum(int max) { lock (_rand) return _rand.Next(max); }
}

В промышленном коде вам следует добавить блоки try/finally для гарантии освобождения блокировок при генерации исключения.

Вот результат выполнения:

Thread B added 61
Thread A added 83
Thread B added 55
Thread A added 33
...

Класс ReaderWriterLockSlim позволяет выполнять больше параллельных операций чтения, нежели обычная блокировка. Мы можем продемонстрировать это путем добавления следующей строки в метод Write в начале цикла while:

Console.WriteLine (_rw.CurrentReadCount + " concurrent readers");

Этот код практически всегда выводит “3 concurrent readers” (методы Read проводят основное время выполнения внутри циклов foreach). Помимо свойства ConcurrentReadCount класс ReaderWriterLockSlim предоставляет следующие свойства для мониторинга блокировок:

public bool IsReadLockHeld            { get; }
public bool IsUpgradeableReadLockHeld { get; }
public bool IsWriteLockHeld           { get; }
 
public int  WaitingReadCount          { get; }
public int  WaitingUpgradeCount       { get; }
public int  WaitingWriteCount         { get; }
 
public int  RecursiveReadCount        { get; }
public int  RecursiveUpgradeCount     { get; }
public int  RecursiveWriteCount       { get; }

Обновляемые блокировки и рекурсия

Иногда бывает полезным переключиться с блокировки чтения на блокировку записи одной атомарной операцией. Например, предположим, вы хотите добавить элемент в список, только если этого элемента там еще нет. В идеале, вы хотите минимизировать время удержания эксклюзивной блокировки (блокировки записи), и поступить следующим образом:

1. Получить блокировку чтения
2. Проверить, если элемент уже находится в списке, освободить блокировку и завершить выполнение функции
3. Освободить блокировку чтения
4. Захватить блокировку записи
5. Добавить элемент

Проблема в том, что другой поток может вклиниться и модифицировать список (например, добавить такой же элемент) между шагами 3 и 4. ReaderWriterLockSlim решает эту задачу с помощью третьего типа блокировки, называемой обновляемой блокировкой (upgradeable lock). Обновляемая блокировка аналогична блокировке чтения за исключением того, что позднее с помощью атомарной операции она может быть расширена до блокировки записи. Вот как ею пользоваться:

1. Вызвать метод EnterUpgradeableReadLock
2. Выполнить операции, требующие доступ на чтение (например, проверить находится ли некоторый элемент в списке).
3. Вызвать EnterWriteLock (это преобразует обновляемую блокировку в блокировку записи).
4. Выполнить операции, требующие доступ на запись (например, добавить элемент в список).
5. Вызвать ExitWriteLock (это преобразует блокировку записи обратно к обновляемой блокировке).
6. Выполнить любые другие операции, требующие доступ на чтение
7. Вызвать ExitUpgradeableReadLock.

С точки зрения вызывающего кода это очень похоже на вложенную или рекурсивную блокировку. Но функционально, на третьем шаге, ReaderWriteLockSlim освобождает блокировку чтения и автоматически получает новую блокировку записи.

Существует еще одно важное отличие между обновляемыми блокировками и блокировками чтения. В то время как обновляемая блокировка может совместно существовать с любым количеством блокировок чтения, в один момент времени может быть получена только одна обновляемая блокировка. Это предотвращает взаимоблокировку при преобразовании блокировок во время сериализации нескольких одновременных обновляемых блокировок, аналогично тому, как обновляются блокировки в SQL Server:

SQL Server	ReaderWriterLockSlim
Разделяемая блокировка	Блокировка на чтение
Эксклюзивная блокировка	Блокировка на запись
Обновляемая блокировка	Обновляемая блокировка

Мы можем показать использование обновляемой блокировки, модифицировав метод Write предыдущего примера таким образом, чтобы добавлять значение в список только тогда, когда его там еще нет:

while (true)
{
  int newNumber = GetRandNum (100);
  _rw.EnterUpgradeableReadLock();
  if (!_items.Contains (newNumber))
  {
    _rw.EnterWriteLock();
    _items.Add (newNumber);
    _rw.ExitWriteLock();
    Console.WriteLine ("Thread " + threadID + " added " + newNumber);
  }
  _rw.ExitUpgradeableReadLock();
  Thread.Sleep (100);
}

Класс ReaderWriterLock также поддерживает преобразование блокировок, но процесс этот ненадежен, поскольку он не поддерживает концепцию обновляемых блокировок. Вот почему разработчикам ReaderWriteLockSlim пришлось создавать полностью новый класс.

Рекурсия блокировок

Обычно, вложенные или рекурсивные блокировки с классом ReaderWriteLockSlim запрещены. Вот почему следующий код генерирует исключение:

var rw = new ReaderWriterLockSlim();
rw.EnterReadLock();
rw.EnterReadLock();
rw.ExitReadLock();
rw.ExitReadLock();

Однако этот код выполняется без ошибок, если создать объект ReaderWriteLockSlim следующим образом:

var rw = new ReaderWriterLockSlim (LockRecursionPolicy.SupportsRecursion);

Это гарантирует то, что рекурсивная блокировка может произойти только тогда, когда вы это планировали. Рекурсивная блокировка может добавить нежелательные сложности, поскольку возможно получение более одного типа блокировок:

rw.EnterWriteLock();
rw.EnterReadLock();
Console.WriteLine (rw.IsReadLockHeld);     // True
Console.WriteLine (rw.IsWriteLockHeld);    // True
rw.ExitReadLock();
rw.ExitWriteLock();

Базовое правило состоит в следующем: если вы уже захватили блокировку, последующая рекурсивная блокировка может быть менее, а не более сильной, согласно следующей шкале:

      Блокировка чтения → Обновляемая блокировка → Блокировка записи

Однако запрос на преобразование обновляемой блокировки к блокировке записи, всегда является корректным.