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

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

}

}

class threadsRunners

{

public static TextPresentation tp = new TextPresentation(); public static void Runner1()

{

Console.WriteLine(“thread_1 run!”);

Console.WriteLine(“thread_1 - calling TextPresentation.showText”); tp.showText(“*”);

Console.WriteLine(“thread_1 stop!”);

}

public static void Runner2()

{

Console.WriteLine(“thread_2 run!”);

Console.WriteLine(“thread_2 - calling TextPresentation.showText”); tp.showText(“|”);

Console.WriteLine(“thread_2 stop!”);

}

static void Main(string[] args)

{

ThreadStart runner1 = new ThreadStart(Runner1); ThreadStart runner2 = new ThreadStart(Runner2);

Thread th1 = new Thread(runner1);

Thread th2 = new Thread(runner2);

th1.Start();

th2.Start();

}

}

}

Очередь как объект синхронизации

Достаточно сложное образование с множеством свойств и методов, предназначенное для упорядоченного размещения объектов (все дети класса object). Одновременное воздействие на очередь со стороны кода нескольких потоков представляет серьёзную опасность. И не только для самого объекта очереди в смысле возможного искажения сохраняемой в ней информации, сколько для самого приложения. Класс Queue взаимодействует с окружением через интерфейсы, неупорядоченное воздействие на объект очереди через эти интерфейсы возбуждает исключения. Очередь располагает специальными средствами, позволяющими защитить объект от неупорядоченного воздействия со стороны множества потоков. Назначение некоторых средств и их применение очевидно, как использовать другие средства – пока неясно (я пометил их вопросительным знаком).

Вкласс входят методы и свойства:

множество вариантов конструкторов - Queue(…),

методы, обеспечивающие загрузку и выгрузку данных (объектов-представителей классов-наследников класса object) – void Enqueue(object), object Dequeue(),

методы поиска – bool Contains(object),

методы предъявления – object Peek() (возвращает объект из начала очереди, не удаляя его из очереди),

свойство Count – сохраняет информацию о количестве объектов в очереди,

свойство SyncRoot – предоставляет ссылку на ОБЪЕКТ СИНХРОНИЗАЦИИ, который используется при синхронизации потоков многопоточного приложения,

166

свойство IsSynchronized (?) – предоставляет информацию о том, синхронизирован ли объект для работы в многопоточном приложении. Это всего лишь значение объявленной в классе Queue булевской переменной,

статический метод Synchronized (?) – создающий синхронизированную оболочку вокруг объекта очереди.

Примеры использования очередей в приложении приводятся ниже, а пока –

“опасное” для очереди занятие в условиях многопоточного приложения. И причина всех неприятностей заключается во внутреннем устройстве и особенностях реализации цикла foreach, который при своём выполнении использует множество функций интерфейса очереди:

Queue myCollection = new Queue();

::::::::::::::::::::::::::::::::::

// Перебор элементов очереди – критическая секция кода. foreach ( Object item in myCollection )

{

:::::::::::::::::::::::::::::::::::::::

}

Возможный способ преодоления опасной ситуации – защита кода критической секцией. Суть защиты сводится к следующему. Поток, выполняющий собственный код при “подходе” к критической секции, связанной с конкретным объектом синхронизации, блокируется, если ЭТУ или ДРУГУЮ связанную с данным объектом синхронизации критическую секцию в данное время выполняет другой поток.

Queue myCollection = new Queue();

::::::::::::::::::::::::::::::::::::::

lock( myCollection.SyncRoot )

{// Критическая секция, связанная с объетом

//синхронизации, полученным от очереди

//myCollection обозначена…

foreach ( Object item in myCollection )

{

:::::::::::::::::::::::::::::::::::::::

}

}

Пример синхронизации объекта очереди. Видно, как создавать синхронизированную оболочку вокруг несинхронизированной очереди, как узнавать о том, синхронизирована она или нет, НО ЗАЧЕМ ДЕЛАТЬ ЭТО – не сказано и не показано. Дело в том, что synchronized она или нет, а соответствующий код (критические секции кода) защищать всё равно надо!

using System;

using System.Collections; public class SamplesQueue {

public static void Main() {

//Creates and initializes a new Queue. Queue myQ = new Queue();

myQ.Enqueue( “The” ); myQ.Enqueue( “quick” ); myQ.Enqueue( “brown” ); myQ.Enqueue( “fox” );

//Creates a synchronized wrapper around the Queue. Queue mySyncdQ = Queue.Synchronized( myQ );

//Displays the sychronization status of both Queues. Console.WriteLine(“myQ is {0}.”,

myQ.IsSynchronized ? “synchronized” : “not synchronized” ); Console.WriteLine( “mySyncdQ is {0}.”,

mySyncdQ.IsSynchronized ? “synchronized” : “not synchronized” );

}

167

}

/*

This code produces the following output. myQ is not synchronized.

mySyncdQ is synchronized. */

Синхронизация работы потоков при работе с общими ресурсами

1. Организация критических секций

///<summary>

///Пара потоков «наперегонки» заполняет одну очередь.

///Эти потоки синхронизуются посредством критичесих секций кода,

///связанных с разделяемым ресурсом - общей очередью.

///Третий поток читает из этой очереди.

///Этот поток синхронизуется посредством монитора.

///Методы Enter(...) и Exit(...) обеспечивают вход в критическую секцию кода,

///связанную с конкретным разделяемым объектом и тем самым блокируют

///одновременное выполнение какого-либо связанного с данным ресурсом кода в другом потоке.

///Толчея потоков сопровождается генерацией исключений.

///</summary>

using System;

using System.Threading; using System.Collections;

namespace CommunicatingThreadsQueue {//----------------------------------------------------------------------------------

public delegate void CallBackFromStartClass (string param);

//=================================================================================== // Данные. Предмет и основа взаимодействия двух потоков.

class CommonData

{

private int iVal; public int iValProp

{

get{return iVal;} set{iVal = value;}

}

public CommonData(int key)

{

iVal = key;

}

}

//Классы Sender и Receiver: основа взаимодействующих потоков. class Sender

{

Queue cdQueue; CallBackFromStartClass callBack; int threadIndex;

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

public Sender(ref Queue queueKey, CallBackFromStartClass cbKey, int iKey)

{

cdQueue = queueKey; callBack = cbKey; threadIndex = iKey;

}

public void startSender()

{

DoIt();

}

168

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

public void DoIt() {//=============================================================== Console.WriteLine(“Sender{0}.DoIt()”, threadIndex);

int i;

for (i = 0; i < 100; i++) {//================================================================ try

{

lock(cdQueue.SyncRoot) {//__________________________________________________________________

Console.WriteLine(“Sender{0}.”, threadIndex); cdQueue.Enqueue(new CommonData(i));

Console.WriteLine(“>> Sender{0} >> {1}.”, threadIndex,cdQueue.Count); foreach(CommonData cd in cdQueue)

{

Console.Write(“\rS{0}:{1} “, threadIndex, cd.iValProp);

}

Console.WriteLine(“__ Sender{0} __”, threadIndex); }//__________________________________________________________________

}

catch (ThreadAbortException e)

{

Console.WriteLine(“~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~”); Console.WriteLine(“AbortException from Sender{0}.”, threadIndex); Console.WriteLine(e.ToString()); Console.WriteLine(“~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~”);

}

catch (Exception e)

{

Console.WriteLine(“__________________________________________________”); Console.WriteLine(“Exception from Sender{0}.”, threadIndex); Console.WriteLine(e.ToString()); Console.WriteLine(“__________________________________________________”); callBack(threadIndex.ToString());

}

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

callBack(string.Format(“Sender{0}”,threadIndex.ToString()));

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

}

class Receiver

{

Queue cdQueue; CallBackFromStartClass callBack; int threadIndex;

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

public Receiver(ref Queue queueKey, CallBackFromStartClass cbKey, int iKey)

{

cdQueue = queueKey; callBack = cbKey; threadIndex = iKey;

}

public void startReceiver()

{

DoIt();

}

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

public void DoIt()

{

Console.WriteLine(“Receiver.DoIt()”); int i = 0;

CommonData cd; while (i < 200)

{

try

{

169

Monitor.Enter(cdQueue.SyncRoot);

Console.WriteLine(“Receiver.”); if (cdQueue.Count > 0)

{

cd = (CommonData)cdQueue.Dequeue();

Console.WriteLine(“Receiver.current:{0},in queue:{1}.”, cd.iValProp,cdQueue.Count); foreach(CommonData cdW in cdQueue)

{

Console.Write(“\rR:{0}.”, cdW.iValProp);

}

Console.WriteLine(“__ Receiver __”); i++;

}

Monitor.Exit(cdQueue.SyncRoot);

}

catch (ThreadAbortException e)

{

Console.WriteLine(“~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~”); Console.WriteLine(“AbortException from Receiver.”); Console.WriteLine(e.ToString()); Console.WriteLine(“~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~”);

}

catch (Exception e)

{

Console.WriteLine(“__________________________________________________”); Console.WriteLine(“Exception from Receiver.”); Console.WriteLine(e.ToString()); Console.WriteLine(“__________________________________________________”); callBack(threadIndex.ToString());

}

}

callBack(“Receiver”);

}

}

class StartClass

{

Thread th0, th1, th2; Queue queueX;

string[] report = new string[3]; ThreadStart t0, t1, t2;

Sender sender0; Sender sender1; Receiver receiver;

static void Main(string[] args)

{

StartClass sc = new StartClass(); sc.go();

}

void go()

{

//Простая очередь.

//queueX = new Queue();

//Синхронизированная очередь. Строится на основе простой очереди.

//Свойство синхронизированности дополнительно устанавливается в true

//посредством метода Synchronized.

queueX = Queue.Synchronized(new Queue());

//Но на самом деле никакой разницы между двумя версиями очереди

//(между несинхронизированной очередью и синхронизирванной оболочкой вокруг

//несинхронизированной очереди) мною замечено не было. И в том и в другом

//случае соответствующий код, который обеспечивает перебор элементов очереди

//должен быть закрыт посредством lock блока, с явным указанием ссылки на

//объект синхронизации.

sender0 = new Sender(ref queueX, new CallBackFromStartClass(StopMain), 0); sender1 = new Sender(ref queueX, new CallBackFromStartClass(StopMain), 1); receiver = new Receiver(ref queueX, new CallBackFromStartClass(StopMain), 2);

170

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

//класса делегата ThreadStart. Поэтому они не имеют параметров. t0 = new ThreadStart(sender0.startSender);

t1 = new ThreadStart(sender1.startSender);

t2 = new ThreadStart(receiver.startReceiver);

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

th0 = new Thread(t0); th1 = new Thread(t1); th2 = new Thread(t2);

th0.Start();

th1.Start();

th2.Start();

th0.Join();

th1.Join();

th2.Join();

Console.WriteLine

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

}

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

{

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

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

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

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

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

if (param.Equals(“Sender0”))

{

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

}

if (param.Equals(“Sender1”))

{

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

}

if (param.Equals(“Receiver”))

{

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

}

if (param.Equals(“0”))

{

th1.Abort();

th2.Abort();

th0.Abort();

}

if (param.Equals(“1”))

{

th0.Abort();

th2.Abort();

th1.Abort();

}

if (param.Equals(“2”))

{

th0.Abort();

th1.Abort();

th2.Abort();

}

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

171

// метод-член класса StartClass StopMain() остановлен. Console.WriteLine(“StopMain(): bye.”);

}

}

}//----------------------------------------------------------------------------------

2. Специальные возможности мониторов

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

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

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

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

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

Enter и Exit иетоды используются для обозначения начала или конца критической секции. Если критическая секция представляет собой “непрерывное” множество инструкций, закрытие кода посредством метода Enter гарантирует, что только один единичный поток сможет выполнять код, закрытый объектом синхронизации.

Рекомендуется размещать эти инструкции в try block и помещать Exit instruction в finally блоке.

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

///<summary>

///Синхронизация потоков с использованием класса монитора.

///Синхронизация потоков с использованием класса монитора.

172

///Монитор защищает очередь от параллельного вторжения со стороны

///взаимодействующих потоков из разных фрагментов кода.

///Однако монитор не может защитить потоки от взаимной блокировки.

///Поток просыпается, делает свою работу, будит конкурента, засыпает сам.

///К тому моменту, как поток будит конкурента, конкурент должен спать.

///Активизация незаснувшего потока не имеет никаких последствий.

///Если работающий поток разбудит не успевший заснуть поток - возникает

///тупиковая ситуация. Оба потока оказываются погруженными в сон.

///В этом случае имеет смысл использовать перегруженный вариант метода

///Wait - с указанием временного интервала.

///</summary>

using System;

using System.Threading; using System.Collections;

namespace MonitorCS1

{

class MonitorApplication

{

const int MAX_LOOP_TIME = 100; Queue xQueue;

public MonitorApplication()

{

xQueue = new Queue();

}

public void FirstThread()

{

int counter = 0; while(counter < MAX_LOOP_TIME)

{

Console.WriteLine(“Thread_1___”); counter++;

Console.WriteLine(“Thread_1...{0}”, counter);

try

{

//Push element. xQueue.Enqueue(counter);

foreach(int ctr in xQueue)

{

Console.WriteLine(“:::Thread_1:::{0}”, ctr);

}

}

catch (Exception ex)

{

Console.WriteLine(ex.ToString());

}

//Release the waiting thread. Применяется к конкурирующему потоку. lock(xQueue){Monitor.Pulse(xQueue);}

Console.WriteLine(“>1 Wait<”);

//Wait, if the queue is busy. Применяется к текущему потоку. // Собственное погружение в состояние ожидания. lock(xQueue){Monitor.Wait(xQueue,1000);} Console.WriteLine(“!1 Work!”);

}

Console.WriteLine(“*****1 Finish*****”); lock(xQueue) {Monitor.Pulse(xQueue);}

}

public void SecondThread()

173

{

int counter = 0; while(counter < MAX_LOOP_TIME)

{

//Release the waiting thread. Применяется к конкурирующему потоку. lock(xQueue){Monitor.Pulse(xQueue);}

Console.WriteLine(«>2 Wait<»);

// Собственное погружение в состояние ожидания. lock(xQueue){Monitor.Wait(xQueue,1000);}

Console.WriteLine(“!2 Work!”);

Console.WriteLine(“Thread_2___”);

try

{

foreach(int ctr in xQueue)

{

Console.WriteLine(“:::Thread_2:::{0}”, ctr);

}

//Pop element.

counter = (int)xQueue.Dequeue();

}

catch (Exception ex)

{

counter = MAX_LOOP_TIME; Console.WriteLine(ex.ToString());

}

Console.WriteLine(“Thread_2...{0}”,counter);

}

Console.WriteLine(“*****2 Finish*****”); lock(xQueue) {Monitor.Pulse(xQueue);}

}

static void Main(string[] args)

{

//Create the MonitorApplication object. MonitorApplication test = new MonitorApplication();

Thread tFirst = new Thread(new ThreadStart(test.FirstThread));

//Вторичные потоки созданы!

Thread tSecond = new Thread(new ThreadStart(test.SecondThread)); //Start threads.

tFirst.Start();

tSecond.Start();

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

tSecond.Join();

}

}

}

Рекомендации по недопущению блокировок потоков

Соблюдать определённый порядок при выделении ресурсов.

При освобождении выделенных ресурсов придерживаться обратного (reverse) порядка.

Минимизировать время неопределённого ожидания выделяемого ресурса.

Не захватывать ресурсы без необходимости и при первой возможности освобождать захваченные ресурсы.

Захватывать ресурс только в случае крайней необходимости.

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

Максимально упрощать структуру задачи, решение которой требует захвата ресурсов. Чем проще задача – тем на меньший период времени захватывается ресурс.

174