Файл: 1_C#_Введение_в_программирование_Уч_пособ_(Марченко-2005).pdf

последним. Таким образом, обеспечивается обработка ВСЕГО множества данных, размещённых в очереди. Разумеется, это идеальная ситуация. Изменение соответствующих значений может привести к тому, что обработчик данных опустошит очередь и завершит работу до того, как генератор данных разместит в очереди все данные.

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

Подобная ситуация хорошо известна как “Race conditions” – состязание потоков и должна учитываться при реализации многопоточных приложений. Результаты работы потока-обработчика не должны зависеть от быстродействия потока-генератора.

Блокировки и тупики

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

Тупик – взаимная блокировка потоков:

Поток A захватывает ресурс a и не может получить доступа к ресурсу b, который занят потоком B, и может быть им освобождён только по получению доступа к ресурсу a.

В приводимом ниже примере отсутствует ВЗАИМНАЯ блокировка потоков, поскольку описание изощрённых “самодельных” способов преодоления тупика не входит в задачу.

///<summary>

///Взаимодействующие потоки разделяют общие ресурсы - пару очередей.

///Для успешной работы каждый поток должен последовательно получить доступ

///к каждой из очередей. Из одной очереди взять в, другую положить.

///Поток оказывается заблокирован, когда одна из очередей оказывается

///занятой другим потоком.

///</summary>

using System;

using System.Threading;

using System.Collections; namespace CommunicatingThreadsQueue

{

///<summary>

///Модифифицированный вариант очереди - очередь с флажком.

///Захвативший эту очередь поток объявляет очередь «закрытой».

///</summary>

public class myQueue: Queue

{

private bool isFree; public bool IsFree

{

get

{

return isFree;

}

set

{

isFree = value;

}

}

public object myDequeue()

{

if (IsFree) {IsFree = false; return base.Dequeue();} else return null;

}

160

CommonData cd = null; while (true)

{//================================================================

if (cdQueue0.Count > 0) {//======================================================== while (true)

{

cd = (CommonData)cdQueue0.myDequeue(); if (cd != null) break;

Console.WriteLine(“>> Receiver{0} is blocked.”, threadIndex);

}

//Временная задержка “на обработку” полученного блока информации

//влияет на частоту и продолжительность блокировок. Thread.Sleep(cd.iValProp*100);

//И это не ВЗАИМНАЯ блокировка потоков.

//“Обработали” блок - открыли очередь.

//И только потом предпринимается попытка

//обращения к очереди оппонента.

cdQueue0.IsFree = true;

//Записали результат во вторую очередь. while (cdQueue1.myEnqueue(cd) == false)

{

Console.WriteLine(“<< Receiver{0} is blocked.”, threadIndex);

}

//А вот если пытаться освободить захваченную потоком очередь

//в этом месте – то взаимной блокировки потоков не избежать!

//cdQueue0.IsFree = true;

//Сообщили о состоянии очередей.

Console.WriteLine(“Receiver{0}...{1}>{2}”,threadIndex.ToString(), cdQueue0.Count,cdQueue1.Count);

}//========================================================

else

{//======================================================== cdQueue0.IsFree = true; callBack(string.Format(“Receiver{0}”,threadIndex.ToString()));

}//========================================================

}//================================================================

}//====================================

}

class Sender

{

Random rnd; int stopVal;

myQueue cdQueue0; myQueue cdQueue1;

CallBackFromStartClass callBack;

// Конструктор...

public Sender(ref myQueue queueKey0, ref myQueue queueKey1, int key,

CallBackFromStartClass cbKey)

{

rnd = new Random(key); stopVal = key; cdQueue0 = queueKey0; cdQueue1 = queueKey1; callBack = cbKey;

}

public void startSender()

162

{

sendIt();

}

// Тело рабочей функции...

public void sendIt() {//====================================

cdQueue0.IsFree = false; cdQueue1.IsFree = false;

while (true)

{

if (stopVal > 0)

{

// Размещение в очереди нового члена со случайными характеристиками. cdQueue0.Enqueue(new CommonData(rnd.Next(0,stopVal))); cdQueue1.Enqueue(new CommonData(rnd.Next(0,stopVal)));

stopVal--;

}

else

{

cdQueue0.IsFree = true; cdQueue1.IsFree = true; callBack(“Sender”);

}

Console.WriteLine(“Sender. the rest of notifications:{0}, notifications in queue:{1},{2}.”, stopVal, cdQueue0.Count, cdQueue1.Count);

}

}//====================================

}

class StartClass

{

static Thread th0, th1, th2; static myQueue NotificationQueue0; static myQueue NotificationQueue1;

static string[] report = new string[3];

static void Main(string[] args)

{

StartClass.NotificationQueue0 = new myQueue();

StartClass.NotificationQueue1 = new myQueue();

//Конструкторы классов Receiver и Sender несут дополнительную нагрузку.

//Они обеспечивают необходимыми значениями методы,

//выполняемые во вторичных потоках.

Sender sender;

//По окончании работы отправитель вызывает функцию-терминатор.

//Для этого используется специально определяемый и настраиваемый делегат. sender = new Sender(ref NotificationQueue0,

ref NotificationQueue1,

10, new CallBackFromStartClass(StartClass.StopMain));

Receiver receiver0;

// Выбрав всю очередь получатель вызывает функцию-терминатор. receiver0 = new Receiver(ref NotificationQueue0,

ref NotificationQueue1,

new CallBackFromStartClass(StartClass.StopMain),0);

Receiver receiver1;

// Выбрав всю очередь получатель вызывает функцию-терминатор. receiver1 = new Receiver(ref NotificationQueue0,

163

ref NotificationQueue1,

new CallBackFromStartClass(StartClass.StopMain),1);

//Стартовые функции потоков должны соответствовать сигнатуре

//класса делегата ThreadStart. Поэтому они не имеют параметров. ThreadStart t0, t1, t2;

t0 = new ThreadStart(sender.startSender);

t1 = new ThreadStart(receiver0.startReceiver);

t2 = new ThreadStart(receiver1.startReceiver);

//Созданы вторичные потоки.

StartClass.th0 = new Thread(t0);

StartClass.th1 = new Thread(t1);

StartClass.th2 = new Thread(t2);

//Запущены вторичные потоки. StartClass.th0.Start();

//Ещё раз о методе Join():

//Выполнение главного потока приостановлено до завершения

//выполнения вторичного потока закрузки очередей.

//Потоки получателей пока отдыхают. StartClass.th0.Join();

//Отработал поток загрузчика.

//Очередь получателей.

StartClass.th1.Start();

StartClass.th2.Start();

//Метод Join():

//Выполнение главного потока опять остановлено

//приостановлено до завершения выполнения вторичных потоков. StartClass.th1.Join();

StartClass.th2.Join();

//Последнее слово остаётся за главным потоком приложения.

//Но только после того как отработают терминаторы.

Console.WriteLine(“Main(): “+report[0]+”...”+report[1]+”...”+report[2]+”... Bye.”);

}

//Функция-член класса StartClass выполняется во ВТОРИЧНОМ потоке! public static void StopMain(string param)

{

Console.WriteLine(«StopMain: « + param);

//Остановка рабочих потоков. Её выполняет функция-член

//класса StartClass. Этой функции в силу своего определения

//известно ВСЁ о вторичных потоках. Но выполняется она

//в ЧУЖИХ (вторичных) потоках.

if (param.Equals(“Sender”))

{

report[0] = “Sender all did.”; StartClass.th0.Abort();

}

if (param.Equals(“Receiver0”))

{

report[1] = “Receiver0 all did.”; StartClass.th1.Abort();

}

if (param.Equals(“Receiver1”))

{

report[2] = “Receiver1 all did.”; StartClass.th2.Abort();

}

//Этот оператор не выполняется! Поток, в котором выполняется

//метод-член класса StartClass StopMain() остановлен.

164

Console.WriteLine(“StopMain(): bye.”);

}

}

}

Очереди. Основа интерфейса взаимодействия

Queue – класс, представляющий коллекцию объектов (objects), работающую по принципу “первый пришёл, первый ушёл” (first-in, first-out).

Stack - класс, представляющий коллекцию объектов (objects), работающую по принципу “последний пришёл, первый ушёл” (last-in, first-out).

Безопасность данных и критические секции кода

Некоторое значение, связанное с конкретным объектом, подвергается воздействию (изменению, преобразованию) со стороны потока. Это означает, что по отношению к объекту (значению объекта) применяется некоторая фиксированная последовательность операторов, в результате которой происходит КОРРЕКТНОЕ изменение состояния объекта или его значения.

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

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

Управление последовательностью доступа потоков к объекту называют синхронизацией потоков.

Сам же объект называют объектом синхронизации. Типичными средствами синхронизации потоков являются:

критические секции,

мониторы,

мьютексы.

Пример организации многопоточного приложения

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

using System;

using System.Threading;

namespace threads12

{

class TextPresentation

{

Mutex mutex = new Mutex(false); public void showText(string text)

{

int i;

//Объект синхронизации в данном конкретном случае -

//представитель класса TextPresentation. Для его обозначения используется

//первичное выражение this.______________________________________________

//1. Блокировка кода монитором (начало)// Monitor.Enter(this); //2. Критическая секция кода (начало)// lock(this)

//2. Критическая секция кода (начало)// { mutex.WaitOne();//3.Блокировка кода мьютексом (начало)// Console.WriteLine(“\n” + (char)31 + (char)31 + (char)31 + (char)31); for (i = 0; i < 250; i++)

{

Console.Write(text);

}

Console.WriteLine(“\n” + (char)30 + (char)30 + (char)30 + (char)30); mutex.Close();//3.Блокировка кода мьютексом (конец)//

165