2.11. Синхронизация потоков

В предыдущем параграфе рассматривались простейшие ситуации, когда несколько потоков работали с независимыми данными. Однако в реальной программе потоки обычно логически связаны друг с другом по данным или ресурсам. Для того чтобы скоординировать их совместную работу, требуются специальные механизмы синхронизации.

Критические секции являются простейшим сервисом синхронизации кода. Они позволяют предотвратить одновременное исполнение защищенных участков кода из различных потоков. Рассмотрим следующий пример. Пусть два потока пытаются выводить данные на консоль порциями по 10 символов:

using System;

using System.Threading;

class MyApp {

static void PrintText(string text) {

for(int i = 0; i < 10; i++) {

Console.Write(text);

Thread.Sleep(100);

}

}

static void FirstPrinter() {

while(true) PrintText("x");

}

static void SecondPrinter() {

while(true) PrintText("o");

}

static void Main() {

Thread th1 = new Thread(new ThreadStart(FirstPrinter));

Thread th2 = new Thread(new ThreadStart(SecondPrinter));

th1.Start();

th2.Start();

}

}

Работа данной программы отличается от ожидаемой: на консоль выводятся символы "x" и "o" в случайном порядке, так как в цикл вывод символов на консоль может «вклиниться» другой поток. В данном примере консоль выступает в качестве такого ресурса, доступ к которому требуется заблокировать, чтобы с этим ресурсом мог работать только один поток. Вывод последовательности из десяти символов является критической секцией программы.

Язык C# содержит специальный оператор lock, задающий критическую секцию. Формат данного оператора следующий:

lock(<выражение>) { <блок критической секции> }

<Выражение> является идентификатором критической секции. В качестве выражения выступает переменная ссылочного типа. Для lock-секций, размещенных в экземплярных методах класса, выражение обычно равно this, для критических секций в статических методах в качестве выражения используется typeof(<имя класса>).

Изменим предыдущий пример следующим образом:

using System;

using System.Threading;

class MyApp {

static void PrintText(string text) {

// Задаем критическую секцию

lock(typeof(MyApp)) {

for(int i = 0; i < 10; i++) {

Console.Write(text);

Thread.Sleep(100);

}

}

}

. . .

}

После подобного изменения данные на консоль выводятся правильно – порциями по 10 символов.

Рассмотрим еще один пример, в котором необходимо использование критической секции. Пусть имеется класс с целочисленным массивом и методами, работающими с данным массивом:

using System;

using System.Threading;

class MyApp {

// В buffer хранятся данные, с которыми работают потоки

static int[] buffer = new int[100];

static Thread writer;

static void Main() {

// Инициализируем the buffer

for(int i=0; i<100; i++)

buffer[i] = i + 1;

// Запустим поток для перезаписи данных

writer = new Thread(new ThreadStart(WriterFunc));

writer.Start();

// запустим 10 потоков для чтения данных

Thread[] readers = new Thread[10];

for(int i=0; i<10; i++) {

readers[i] =

new Thread(new ThreadStart(ReaderFunc));

readers[i].Start();

}

}

static void ReaderFunc() {

// Работаем, пока работает поток записи

while(writer.IsAlive) {

int sum = 0;

// Считаем сумму элементов из buffer

for(int k=0; k<100; k++) sum += buffer[k];

// Если сумма неправильная, сигнализируем

if(sum != 5050) {

Console.WriteLine("Error in sum!");

return;

}

}

}

static void WriterFunc() {

Random rnd = new Random();

// Цикл на 10 секунд

DateTime start = DateTime.Now;

while((DateTime.Now - start).Seconds < 10) {

int j = rnd.Next(0, 100);

int k = rnd.Next(0, 100);

int tmp = buffer[j];

buffer[j] = buffer[k];

buffer[k] = tmp;

}

}

}

При работе данного приложения периодически возникают сообщения о неправильно посчитанной сумме. Причина этого заключается в том, что метод WriterFunc() может изменить данные в массиве buffer во время подсчета суммы. Решение проблемы: объявим критическую секцию, содержащую код, работающий с массивом buffer.

static void ReaderFunc() {

while(writer.IsAlive) {

int sum = 0;

lock(buffer) {

for(int k=0; k<100; k++) sum += buffer[k];

}

// Далее по тексту

. . .

}

}

static void WriterFunc() {

Random rnd = new Random();

DateTime start = DateTime.Now;

while((DateTime.Now - start).Seconds < 10) {

int j = rnd.Next(0, 100);

int k = rnd.Next(0, 100);

lock(buffer) {

int tmp = buffer[j];

buffer[j] = buffer[k];

buffer[k] = tmp;

}

}

}

Обратите внимание на использование одинаковых идентификаторов при указании критической секции (в разных частях программы).

Команда lock языка C# – это всего лишь скрытый способ работы со специальным классом System.Threading.Monitor. А именно, объявление вида

lock(buffer){ . . . }

эквивалентно следующему:

Monitor.Enter(buffer);

try {

. . .

}

finally {

Monitor.Exit(buffer);

}

Статический метод Monitor.Enter() определяет вход в критическую секцию, статический метод Monitor.Exit() – выход из секции. Параметрами данных методов является объект – идентификатор критической секции.

Коротко опишем базовые принципы внутренней организации критической секции. Любой объект имеет скрытое поле syncnum, которое хранит указатель на элемент таблицы блокировок. Если некоторый поток пытается войти в критическую секцию, выполняется проверка значения syncnum. Если данное значение равно null, то код критической секции «свободен» и его можно выполнять. В противном случае поток ставиться в системную очередь, из которой извлекается для выполнения тогда, когда критическая секция освободиться.

Вернемся к предыдущему примеру. Требование наличия критической секции в методе WriterFunc() очевидно: иначе подсчет суммы может вклиниться между инструкциями buffer[j] = buffer[k] и buffer[k] = tmp и получить неверное значение. Когда мы считаем сумму в методе ReaderFunc(), то очевидно, что мы не должны менять значение массива. Однако и в первом и во втором случае требуется блокировать потоки на одном ресурсе. Соответственно, речь идет об одной критической секции, но как бы «размазанной» по двум методам. Не важно, что мы используем buffer в качестве идентификатора критической секции. Это может быть любой инициализированный объект. Таким образом, следующий код также обеспечивает правильную работу:

class MyApp {

. . .

static object someObj = new Random(); // Какой-то объект

. . .

static void ReaderFunc() {

while(. . .) {

. . .

lock(someObj) {

. . .

}

. . .

}

}

static void WriterFunc() {

. . .

while(. . .) {

. . .

lock(someObj) {

. . .

}

}

}

}

Если требуется простая синхронизация потоковых действий с целочисленной переменной, то для этих целей можно использовать класс System.Threading.Interlocked. Данный класс располагает следующими четырьмя статическими методами:

• Increment() – Увеличивает на единицу переменную типа int или long;

• Decrement() – Уменьшает на единицу переменную типа int или long;

• Exchange() – Обменивает значения двух переменных типа int, long или любых двух объектов;

• CompareExchange() – Сравнивает значения первых двух параметров, в случае совпадения заменяет этим значением значение третьего параметра. Тип параметров: int, float, object.

Платформа .NET предоставляет простой способ синхронизации доступа к методам на основе атрибутов. В пространстве имен System.Runtime.CompilerServices описан атрибут MethodImplAttribute, который может применяться к конструкторам и методам и указывает для компилятора особенности реализации метода. Аргументом атрибута являются элементы перечисления MethodImplOptions. В контексте рассматриваемой темы представляет интерес элемент MethodImplOptions.Synchronized. Для того чтобы запретить одновременное выполнение некоторого метода в разных потоках, достаточно объявить метод следующим образом:

[MethodImpl(MethodImplOptions.Synchronized)]

void TransformData(byte[] buffer) { . . . }

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

В заключение рассмотрим класс System.Threading.ThreadPool. Данный класс предназначен для поддержки пула потоков. Пул потоков автоматически запускает указанные методы в различных потоках. Одновременно пул поддерживает 25 запущенных потоков, другие потоки ожидают своей очереди в пуле.

Для регистрации методов в пуле потока служит статический метод QueueUserWorkItem(). Его параметр – это делегат типа WaitCallback:

public delegate void WaitCallback(object state);

При помощи объекта state в метод потока передаются параметры.

Рассмотрим пример приложения, использующего ThreadPool. В приложении в пул помещается 5 одинаковых методов, выводящих значение счетчика на экран:

using System;

using System.Threading;

class MyApp {

static int count = 0; // счетчик

static void Main() {

WaitCallback callback =

new WaitCallback(ProcessRequest);

ThreadPool.QueueUserWorkItem(callback);

ThreadPool.QueueUserWorkItem(callback);

ThreadPool.QueueUserWorkItem(callback);

ThreadPool.QueueUserWorkItem(callback);

ThreadPool.QueueUserWorkItem(callback);

// Приостанавливаемся, чтобы выполнились методы

Thread.Sleep(5000);

}

static void ProcessRequest(object state) {

int n = Interlocked.Increment(ref count);

Console.WriteLine(n);

}

}

Перегруженная версия метода ThreadPool.QueueUserWorkItem() имеет два параметра: первый – это делегат, второй – объект, при помощи которого делегату можно передать информацию:

int[] vals = new int[5]{1, 2, 3, 4, 5};

ThreadPool.QueueUserWorkItem(callback, vals);

// Объявление и реализация ProcessRequest()

static void ProcessRequest(object state) {

int[] vals = (int[])state;

. . .

}

Поток из пула никогда не должен уничтожаться «вручную». Автоматический менеджер пула потоков берет на себя работу по созданию потока в пуле, он же будет уничтожать потоки. Для того чтобы определить вид потока, можно использовать свойство IsThreadPoolThread класса Thread. В следующем примере поток уничтожает себя только в том случае, если он не запущен из пула:

if (!Thread.CurrentThread.IsThreadPoolThread)

Thread.CurrentThread.Abort();