Добрынин В.Ю. Технологии компонентного программирования
Подождите немного. Документ загружается.
. . . . .
}
. . . . .
}
. . . . .
public class News : ContextBoundObject {
private SynchronizationAttribute _syncProperty;
public News(){
_syncProperty =
(SynchronizationAttribute)
Thread.CurrentContext.GetProperty(
"Synchronization");
. . . . .
}
. . . . .
internal SynchronizationAttribute syncProperty{
get { return _syncProperty;}
}
. . . . .
}
Просматривая консоль сервера убеждаемся в том, что компонент News размещается в новом домене синхронизации:
Server is listening
News context = 3 News constructor thread = 9 IsPoolThread = True
Tax syncProperty == News syncProperty False
Tax context = 2 Tax constructor thread = 9 IsPoolThread = True
Account context = 1 Account constructor thread = 9 IsPoolThread = True
Account syncProperty == Tax syncProperty True
. . . . .
И последнее замечание касается файла LogFile. Просматривая его, можно заметить, что теперь перехватываются все вызовы,
идущие к компонентам
Tax и News, идущие не только от компонента Account, но и от компонента Tax к компоненту News. Это
объясняется тем, что все компоненты живут в различных контекстах и все вызовы пересекают границу контекста и,
следовательно, перехватываются.
Еще раз про атрибут синхронизации
Эта глава продолжает изучение кода атрибута синхронизации из Rotor, рассмотрение которого было начато в предыдущей главе.
Там мы рассмотрели основные механизмы, связанные с определением контекста и домена синхронизации, в которых будет
размещен новый объект - экземпляр класса, которому приписан атрибут
SynchronizationAttribute. Это конструкторы (четыре
варианта) и методы
IsContextOK и GetPropertiesForNewContext класса SynchronizationAttribute. Теперь мы сосредоточимся
на самом алгоритме синхронизации и попутно рассмотрим несколько важных понятий, связанных с программированием в CLR.
Инициализация свойства синхронизации в домене
синхронизации
Начнем с метода InitIfNecessary класса SynchronizationAttribute:
internal virtual void InitIfNecessary() {
lock(this) {
if (_asyncWorkEvent == null) {
_asyncWorkEvent = new AutoResetEvent(false);
_workItemQueue = new Queue();
_asyncLcidList = new ArrayList();
WaitOrTimerCallback callBackDelegate =
new WaitOrTimerCallback(this.DispatcherCallBack);
ThreadPool.RegisterWaitForSingleObject(
_asyncWorkEvent,
callBackDelegate,
null,
_timeOut,
false);
}
}
}
Данный метод вызывается при формировании каждого нового контекста синхронизации, однако делает он что-либо только в том
случае, когда этот контекст начинает собой новый домен синхронизации. В этом случае инициализируется свойство
синхронизации данного контекста, которое одновременно будет и свойством синхронизации всего домена синхронизации. При
включении в этот домен синхронизации нового контекста
синхронизации новое свойство синхронизации не создается и его
инициализация не требуется.
В коде данного метода мы сталкиваемся с рядом ранее не рассмотренных понятий, таких как критические секции, делегаты, пул
р
абочих потоков, события. Прежде чем двигаться дальше,
р
ассмотрим упомянутые понятия.
Критические секции
Если атрибут синхронизации позволяет управлять синхронизацией декларативно, то критическая секция обеспечивает решение
данной проблемы в рамках парадигмы процедурного программирования. И судя по всему, более эффективно, так как при этом
нет необходимости создавать контексты, перехватчики, преобразовывать вызовы в сообщения и обратно из собщение
формировать вызовы.
Рассмотрим несколько примеров.
Ранее мы уже рассматривали
консольное серверное приложение MyServer, поддерживающее некоторый банковский счет.
Клиентские приложения могли параллельно делать вклады на этот счет. Синхронизация обеспечивалась за счет использования
атрибута синхронизации SynchronizationAttribute, который приписывался классу Account, и наследования этого класса от
класса
ContextBoundObject.
Теперь мы обеспечим синхронизацию за счет использования критических секций.
Простейший способ связан с приписыванием методу
Add атрибута
[MethodImpl(MethodImplOptions.Synchronized)]. Данный атрибут запретит вход в тело метода Add какого-либо потока, если
этот метод уже выполняется в другом потоке. В данном случае мы полностью полагаемся на компилятор, который должен
обеспечить требуемую функциональность.
. . . . .
namespace MyServer {
. . . . .
public class Account: MarshalByRefObject,
IAccumulator, IAudit {
. . . . .
[MethodImpl(MethodImplOptions.Synchronized)]
public void Add(int sum) {
_sum += sum;
}
. . . . .
}
}
Заметим, что теперь достаточно наследования класса Account от класса MarshalByRefObject, так как привязка экземпляра этого
класса к контексту более не нужна.
Использование атрибута
[MethodImpl(MethodImplOptions.Synchronized)] конечно удобно, однако и накладывает на
программиста определенные ограничения:
z Критическая секция охватывает все тело метода
Можно представить ситуацию, когда критичная операция, требующая защиты от параллельного выполнения в нескольких
потоках, выполняется в данном методе только в некоторых случаях (в зависимости от значений входных аргументов). В
этом случае включение всего метода в критическую секцию неоправдано и будет снижать общую эффективность системы.
z Нет возможности запретить параллельный доступ к совместно используемым объектам
Предположим, в данном методе выполняется работа с некоторой очередью (экземпляр класса
Queue). Конечно, благодаря
наличию атрибута
[MethodImpl(MethodImplOptions.Synchronized)] в рамках данного метода два потока не смогут
параллельно работать с этой очередью и целостность данных будет обеспечена. Однако, ничто не запрещает какому-то
другому потоку обратиться к этой же самой очереди в процессе выполнения какого-либо другого метода. Вот тут и
возможны нарушения целостности, т.к. между различными потоками,
выполняющими параллельно различные методы, нет
никакой коммуникации.
Указанные выше проблемы решаются при использовании класса
Monitor.
. . . . .
namespace MyServer {
. . . . .
public class Account: MarshalByRefObject,
IAccumulator, IAudit {
. . . . .
public void Add(int sum) {
. . . . .
Monitor.Enter(this);
try {
_sum += sum;
}
finally {
Monitor.Exit(this);
}
. . . . .
}
. . . . .
}
}
Вызов статического метода Monitor.Enter() помечает начало критической секции, а вызов метода Monitor.Exit() - ее конец.
Аргумент в методе
Enter представляет собой ссылку на некоторый объект. В данном случае это ссылка на экземпляр класса
Account, на котором и вызван метод Enter, однако ничто не мешает указать ссылку на какой-либо другой объект.
Оъект, на который указывает ссылка при вызове
Enter, начинает играть роль "эстафетной палочки". Поток, которому удалось
вызвать
Monitor.Enter(obj), входит в данную критическую секцию, и никакой другой поток не получит ответа от вызова Monitor.Enter
(obj)
, пока первый поток не вызовет Monitor.Exit(obj). Все потоки, сделавшие вызов
Monitor.Enter(obj), находятся в одной очереди потоков готовых к выполнению, и эта очередь связана с объектом obj.
Использование блока
try и включение вызова Monitor.Exit(obj) в блок finally способствует повышению надежности
программирования. Если даже после входа в критическую секцию будет сгенерировано какое-то исключение, вызов
Monitor.Exit(obj) будет выполнен в любом случае, и очередной готовый к выполнению поток, заблокированный при вызове
Monitor.Enter(obj), начнет выполняться.
Хотя, как указывалось ранее, в качестве "эстафетной палочки" можно использовать любой объект, разумно использовать именно
тот объект, ради безопасного доступа к которому и была сформирована данная критическая секция. В этом случае ( если такой же
подход будет использован при формировании всех критических секций) два различных потока не будут параллельно
выполнять
критичные для целостности данных операции над одним и тем же объектом.
Компилятор для C# допускает использование конструкции
lock(obj){} для задания критичеcкой секции. При этом неявно
используется тот же класс
Monitor:
. . . . .
namespace MyServer {
. . . . .
public class Account: MarshalByRefObject,
IAccumulator, IAudit {
. . . . .
public void Add(int sum) {
lock(this) {
_sum += sum;
}
}
. . . . .
}
}
Имеются еще два метода класса Monitor, которые используются в коде атрибута синхронизации. Это Monitor.Wait() и
Monitor.Pulse().
Рассмотрим следующую модификацию предыдущего примера:
. . . . .
namespace MyServer {
. . . . .
public class Account: MarshalByRefObject,
IAccumulator, IAudit {
. . . . .
public void Add(int sum) {
lock(this) {
Console.WriteLine(Thread.CurrentThread.GetHashCode());
int s = _sum;
Thread.Sleep(1);
_sum = s + sum;
if (_sum == 5) {Monitor.Wait(this);}
if (_sum == 505) {Monitor.Pulse(this);}
}
}
. . . . .
}
}
Напомним, что данный фрагмент кода выполняется на сервере MyServer.exe, к которому параллельно могут обращаться
несколько клиентов. Каждый клиент (приложение
MyApp) посылает на сервер 100 раз по 5 условных единиц.
Выводя на консоль хеш потока, мы можем отследить чередование рабочих потоков в очереди готовых к выполнению потоков.
Сохранение текущей величины счета в локальной переменной
s и вызов Thread.Sleep(1) используются для более явного
выявления эффектов, связанных с многопоточностью.
Как правило (если в предыдущем фрагменте кода закоментировать строки с вызовами
Monitor.Wait() и Monitor.Pulse), один и
тот же поток может несколько раз подряд войти в данную критическую секцию и положить на счет очередные 5 условных
единиц, прежде чем выделенный ему квант времени закончится и начнет исполняться другой рабочий поток. После нескольких
циклов вновь начинает работать первый поток и так далее. Используя методы
Wait и Pulse класса Monitor мы можем управлять
очередностью входа различных потоков в данную критическую секцию.
Как только первый поток входит в нашу критическую секцию, он выводит на консоль свой идентификатор, запоминает текущее
значение счета в локальной переменной и засыпает на 1 миллисекунду. Пробудившись, этот поток обновляет счет (его величина
становится равной 5). В связи с выполнением
условия _sum == 5 выполняется вызов Monitor.Wait(this). В этот момент первый
поток освобождает объект
this и становится в очередь ожидания. Эта еще одна, связанная с объектом очередь (наряду с
очередью потоков, готовых к выполнению). Разница между ними состоит в следующем. Очередной поток из очереди готовых к
выполнению потоков начинает выполняться, если текущий исполняемый поток завершил выполнение критической секции
(вызвал
Monitor.Exit(this), то есть освободил объект this). Потоки из очереди ожидания становятся в очередь потоков
готовых к выполнению, если текущий исполняемый поток вызвал
Monitor.Pulse(this), сигнализируя тем самым, что состояние
объекта
this изменилось и ожидающие потоки могут работать с данным объектом.
Таким образом, первый поток стоит в очереди ожидания, а тем временем второй (и, возможно, другие потоки) пополняет счет.
Как только счет достигнет суммы в 505 условных единиц, первый поток попадает в очередь готовых к выполнению потоков и
начинает работать.
Делегаты, регистрация callback делегата в пуле рабочих потоков
В коде метода InitIfNecessary класса SynchronizationAttribute используются упомянутые в заголовке данного раздела
сущности. Познакомимся с их применением в процессе разбора следующего примера:
using System;
using System.Threading;
public class Test {
private AutoResetEvent _myEvent;
private int _count = 0;
public Test() {
Console.WriteLine(">>> Test constructor thread = " +
Thread.CurrentThread.GetHashCode() +
" IsPoolThread = " +
Thread.CurrentThread.IsThreadPoolThread);
_myEvent = new AutoResetEvent(false);
WaitOrTimerCallback myCallBackDelegate =
new WaitOrTimerCallback(this.MyCallBack);
ThreadPool.RegisterWaitForSingleObject(
_myEvent,
myCallBackDelegate,
null,
100,
false);
}
public int count {
get { return _count; }
set { _count = value;}
}
private delegate String MyDelegate ();
private void MyCallBack(Object state, bool timedOut) {
Console.WriteLine(">>> MyCallback thread = " +
Thread.CurrentThread.GetHashCode() +
" IsPoolThread = " +
Thread.CurrentThread.IsThreadPoolThread);
MyDelegate hello = new MyDelegate(MyHello);
count++;
Console.WriteLine(hello() + " Count = " + count +
" timedOut = " + timedOut);
}
private String MyHello() {
return "Test_" + count +" : " ;
}
public void NewEvent() {
_myEvent.Set();
}
}
public class MyApp {
public static void Main () {
Console.WriteLine(">>> MyApp thread = " +
Thread.CurrentThread.GetHashCode() +
" IsPoolThread = " +
Thread.CurrentThread.IsThreadPoolThread);
Test test = new Test();
test.NewEvent();
Thread.Sleep(500);
test.NewEvent();
Thread.Sleep(1000);
}
}
Опишем прежде всего семантику нашего приложения.
Метод
Main выполняется в основном потоке приложения. Все приложение в целом завершается по завершении этого метода
(после возвращения из вызова функции
Thread.Sleep(1000)). Параллельно с выполнением основного потока несколько раз
успевает выполниться метод
MyCallBack класса Test. Заметим, что этот метод выполняется так называемыми рабочими
потоками, извлекаемыми системой из пула рабочих потоков. Рабочий поток не может пережить основной поток и по завершении
последнего завершается и текущий рабочий поток.
Данное консольное приложение после запуска прежде всего выводит на консоль информацию о потоке, выполняющем код
функции Main. Это хеш потока и информация о том,
я
вляется ли данный поток рабочим потоком. Очевидно, что в данном случае
значение последнего параметра должно быть равно
false так как текущий поток является основным потоком приложения.
Далее вызывается конструктор класса
Test. Здесь также выводится информация о потоке - том потоке, который выполняет код
конструктора. Этот тот же самый основной поток приложения.
Далее в конструкторе создается экземпляр
_myEvent события типа
AutoResetEvent. События в .NET еще не обсуждались в данном курсе, но в данном случае используется событие специального
типа, реализованное в системе. В связи с этим пока будет достаточно рассмотреть только это событие и только в контексте
данного примера.
Для работы с пулом потоков мы регистрируем с помощью статического метода
ThreadPool.RegisterWaitForSingleObject
делегат
myCallBackDelegate специального известного системе типа
WaitOrTimerCallback. Делегат является некоторым аналогом указателя на функцию (в данном случае на MyCallBack), которая
должна иметь сигнатуру, заданную при объявлении делегата. В случае делегата типа
WaitOrTimerCallback возвращаемое значение должно отсутствовать (void), первый аргумент должен иметь тип System.Object и
может использоваться для передачи вызываемой функции произвольных данных, второй параметр должен иметь тип
bool и
система через него передает значение
true, если данный вызов произошел в связи с истечением времени ожидания (см.
объяснение ниже).
Функция, ссылка на которую передана в делегат, будет вызываться и исполняться некоторым рабочим потоком из пула как
только произойдет одно из двух событий:
z Событие _myEvent (зарегестрированное вместе с делегатом) будет установлено в состояние signaled
Для установки события типа
AutoResetEvent в данное состояние достаточно вызвать его метод Set().
z Время ожидания превысило пороговое значение (четвертый параметр в ThreadPool.RegisterWaitForSingleObject)
Отсчет времени идет от момента регистрации делегата или от момента последнего его вызова.
Заметим, что при создании
_myEvent вызывался конструктор
AutoResetEvent(false). Задание параметра false привело к созданию события, не находящегося в состоянии signaled. При
задании в этом конструкторе параметра
true инициированное событие сразу же находится в состоянии signaled, и делегат
вызывается сразу же после его регистрации.
Событие
AutoResetEvent обладает еще одним важным свойством, отраженным в его названии - оно переходит в исходное
состояние автоматически после очередного вызова делегата.
Третий параметр в
ThreadPool.RegisterWaitForSingleObject задает ссылку на объект, содержащий данные передаваемые
связанной с делегатом функции при вызове последнего (в данном случае ничего не передается).
Последний параметр определяет, что делегат зарегистрирован навсегда (точнее в данном случае до момента уничтожения
эземпляра класса
Test, в конструкторе которого выполняется регистрация). Если бы последний параметр равнялся true,
регистрация делегата была бы действительна только на один вызов.
Метод
MyCallBack, ссылка на который передана конструктору при создании делегата myCallBackDelegate, прежде всего выводит
на консоль информацию о потоке, в котором он исполняется. Как и раньше это хеш потока и данные о его типе. В данном случае
мы полагаем, что значение последнего параметра при каждом вызове этого метода будет равно
true, так как выполняться этот
метод должен рабочим потоком.
Далее
MyCallBack выводит на консоль некоторое сообщение. Это сообщение состоит из следующих частей:
1. Префикс
Ради демонстрации того, как можно объявлять делегат нового типа, префикс формируется излишне сложно - посредством
использования делегата нового типа -
MyDelegate. Из его объявления
private delegate String MyDelegate ();
видно, что с делегатом данного типа можно связать любую функцию, возвращающую String и не имеющую аргументов. В
теле метода
MyCallBack создается новый делегат hello
MyDelegate hello = new MyDelegate(MyHello);
которому передается ссылка на метод MyHello этого же класса Test. Именно этот метод и формирует префикс вида
Test_XXX : , где вместо XXX будет подставляться порядковый номер текущего вызова метода MyCallBack.
2. Порядковый номер вызова данного метода
При каждом вызове метода
MyCallBack счетчик count увеличивается на единицу.
3. Информация о причине вызова данного метода
Через параметр timedOut метод MyCallBack получает от системы информацию о причине его вызова. Если получено
значение
false, то этот метод был вызван благодаря тому, что кто-то установил событие _myEvent в состояние signaled.
Значение
true будет получено в том случае, если метод был вызван по причине завершения срока ожидания.
Последний метод
NewEvent класса Test как раз и может использоваться клиентами для перевода события _myEvent в состояние
signaled.
Теперь вновь обратимся к коду метода
Main.
После создания экземпляра test класса Test вызывается его метод
NewEvent в результате чего из пула рабочих потоков извлекается новый поток, который и выполняет метод MyCallBack.
Напомним, что после этого событие
_myEvent автоматически переходит в начальное состояние.
Далее основной поток засыпает на 500 mc. В связи с тем, что интервал ожидания, заданный четвертым параметром в
ThreadPool.RegisterWaitForSingleObject равен 100 mc, метод MyCallBack будет вызван несколько раз по причине завершения
периода ожидания.
Далее во второй раз вызывается метод
NewEvent, и MyCallBack вызывается по причине перехода события _myEvent в состояние
signaled.
И, наконец, основной поток засыпает еще на 1000 mc, в течении которых
MyCallBack вызывается с интервалом 100 mc по
причине завершения времени ожидания.
Через 1000 mc основной поток просыпается и выполнение всего приложения (включая все рабочие потоки) завершается.
Ниже приводится вывод на консоль, полученный после запуска данного приложения:
>>> MyApp thread = 16 IsPoolThread = False
>>> Test constructor thread = 16 IsPoolThread = False
>>> MyCallback thread = 18 IsPoolThread = True
Test_1 : Count = 1 timedOut = False
>>> MyCallback thread = 18 IsPoolThread = True
Test_2 : Count = 2 timedOut = True
>>> MyCallback thread = 18 IsPoolThread = True
Test_3 : Count = 3 timedOut = True
>>> MyCallback thread = 18 IsPoolThread = True
Test_4 : Count = 4 timedOut = True
>>> MyCallback thread = 18 IsPoolThread = True
Test_5 : Count = 5 timedOut = True
>>> MyCallback thread = 18 IsPoolThread = True
Test_6 : Count = 6 timedOut = True
>>> MyCallback thread = 18 IsPoolThread = True
Test_7 : Count = 7 timedOut = False
>>> MyCallback thread = 18 IsPoolThread = True
Test_8 : Count = 8 timedOut = True
>>> MyCallback thread = 18 IsPoolThread = True
Test_9 : Count = 9 timedOut = True
>>> MyCallback thread = 18 IsPoolThread = True
Test_10 : Count = 10 timedOut = True
>>> MyCallback thread = 18 IsPoolThread = True
Test_11 : Count = 11 timedOut = True
>>> MyCallback thread = 18 IsPoolThread = True
Test_12 : Count = 12 timedOut = True
>>> MyCallback thread = 18 IsPoolThread = True
Test_13 : Count = 13 timedOut = True
>>> MyCallback thread = 18 IsPoolThread = True
Test_14 : Count = 14 timedOut = True
>>> MyCallback thread = 18 IsPoolThread = True
Test_15 : Count = 15 timedOut = True
>>> MyCallback thread = 18 IsPoolThread = True
Test_16 : Count = 16 timedOut = True
Возвращаемся к коду инициализации атрибута
Теперь можно более подробно обсудить код метода InitIfNecessary. Все тело этого метода включено в критическую секцию
lock(this) {}. Здесь this является ссылкой на экземпляр текущего класса
(
SynchronizationAttribute), который и является собственно свойством синхронизации (как контекста, так и домена
синхронизации). Таким образом, при входе текущего потока в данную критическую секцию никакой другой поток не может
войти в эту секцию (и в любую другую типа
lock(obj) {}, где obj является ссылкой на данное свойство синхронизации).
Далее проверяется условие
_asyncWorkEvent == null. Это условие выполняется только тогда, когда текущее свойство
синхронизации еще не инициализовано, т.е. в данный момент формируется новый домен синхронизации и текущее свойство
будет его свойством синхронизации. Именно в этом случае выполняется инициализация свойства. В противном случае код
инициализации пропускается, т.к. текущее свойство уже инициализированно ранее.
Инициализация состоит
из следующих шагов:
z Создается экземпляр _asyncWorkEvent события AutoResetEvent
Данное событие будет использовано для уведомления системы о том, что очередной рабочий поток из пула потоков может
выполнить очередной вызов, сохраненный в очереди вызовов (см. следующий пункт). Начальное состояние данного
события не равно signaled, и для уведомления системы это событие надо перевести в состояние signaled (после чего оно
автоматически вернется в исходное
состояние).
z Создается экземпляр _workItemQueue очереди Queue
Поддержание этой очереди - основная задача свойства синхронизации. Как правило, внешние вызовы, приходящие к
объектам некоторого домена синхронизации, преобразуются в специальную форму - так называемую работу, и сохраняются
в данной очереди (в некоторых случаях вызов не сохраняется в очереди и выполняется сразу же). Очередная работа
извлекается из этой очереди и выполняется при
готовности системы выполнять новую работу.
z Создается список _asyncLcidList
Данный список будет использоваться для хранения идентификаторов логических вызовов для асинхронных вызовов,
исходящих из данного домена синхронизации. Подробнее это будет обсуждаться далее.
z Создается делегат callBackDelegate типа WaitOrTimerCallBack
Этот делегат хранит ссылку на функцию DispatcherCallBack, которая и будет обрабатывать вызовы, извлекаемые из
очереди вызовов.
z Регистрация делегата callBackDelegate и события _asyncWorkEvent в пуле рабочих потоков
Для регистрации используется статический метод
RegisterWaitForSingleObject класса ThreadPool. Третий параметр в вызове данного метода равен null, что говорит о
том, что функции
DispatcherCallBack не передаются никакие данные. Величина интервала ожидания _timeOut, по
истечении которого автоматически вызывается делегат (если ранее состояние
_asyncWorkEvent не было переведено в
состояние signaled), задается при инициализации атрибута синхронизации и доступна только для чтения:
private static readonly UInt32 _timeOut = (UInt32)0x7fffffff;
Последний параметр в вызове метода RegisterWaitForSingleObject равен false, что означает, что данная регистрация
сохраняется до момента уничтожения самого свойства синхронизации.
Обработка вызова, извлекаемого
из очереди вызовов
Прежде всего рассмотрим класс
internal class WorkItem {. . . . .}
экземпляры которого используются для хранения информации о вызовах в очереди вызовов.
Представление вызова в виде работы -
экземпляра класса
WorkItem
Каждый вызов представляется экземпляром класса WorkItem, в котором необходимая информация задается следующими полями:
z internal IMessage _reqMsg
Это поле хранит ссылку на объект, представляющий собственно вызов в форме сообщения. Именно в этой форме вызов
передается между контекстами клиента и сервера. Соответствующий класс должен реализовать интерфейс IMessage.
z internal IMessageSink _nextSink
Вызов попадает в контекст, пройдя некоторую цепочку перехватчиков. Перехватчик, ассоциированный со свойством
синхронизации, отправляет вызов в очередь вызовов (если его нельзя выполнить сразу же). Вся семантика свойства
синхронизации связана с поддержкой этой очереди. Отстояв свое время в этой очереди, вызов должен продолжить свой
путь через цепочку перехватчиков. Данное поле
_nextSink хранит ссылку на следующий передатчик, которому должен
быть передан вызов. Интерфейс
IMessageSink должен быть реализован каждым перехватчиком.
z internal IMessageSink _replySink
Вызовы разделяются на два типа: синхронные и асинхронные. В случае синхронного вызова вызывающая сторона
блокируется до получения ответа, в асинхронном случае такая блокировка не выполняется. Однако, в асинхронном случае
может оказаться необходимым как-то обеспечить уведомление вызывающей стороны о завершении вызова и о его
результатах. В данном случае в вызов
включается ссылка _replySink на специальный перехватчик, которому система
должна передать уведомление. В случае синхронного вызова значение данного поля равно
null.
z internal IMessage _replyMsg
В случае синхронного вызова результат отправляется вызывающей стороне опять же в форме сообщения. Данное поле
будет хранить ссылку на это сообщение по завершении вызова.
z internal Context _ctx
Домен синхронизации может состоять из нескольких контекстов, и только один из них содержит свойство синхронизации
данного домена. С каждым контекстом связана своя цепочка перехватчиков. В каждой такой цепочке имеется перехватчик,
ассоциированный со свойством синхронизации. Вызов, перемещающийся по некоторой цепочке перехватчиков в некоторый
контекст, прерывает свое путешествие в перехватчике, ассоциированном со
свойством синхронизации, и отправляется в
очередь вызовов в виде работы - экземпляра класса
WorkItem. Эта очередь поддерживается в свойстве синхронизации
данного домена, которое может размещаться совсем не в том контексте, куда первоначально направлялся вызов. Поле
_ctx
используется для хранения информации о первоначальном контексте, в котором должен выполняться данный вызов после
его освобождения из очереди.
z internal LogicalCallContext _callCtx
Условия выполнения вызова определяются не только контекстом, в котором он должен выполняться, но и контекстом
логического вызова, сопровождающим данный вызов. Подробнее это понятие будет объяснено далее. Поле
_callCtx
хранит ссылку на контекст логического вызова для данного вызова.
Немного про асинхронные вызовы
При разборе кода атрибута синхронизации нам придется часто упоминать такое понятие как асинхронный вызов. Уместно
сделать отступление и разобрать код, демонстрирующий работу с асинхронными вызовами.
Рассмотрим следующую ситуацию. Сервер предоставляет услуги по проведению сложных математических вычислений,
требующих значительных временных затрат.
При использовании синхронных вызовов клиент должен был бы последовательно вызывать необходимые
методы. При этом при
каждом синхронном вызове клиент блокируется до получения ответа, что не дает ему возможность вызывать методы сервера
параллельно и выполнять какую-либо другую полезную работу во время ожидания результатов от сервера.
При использовании асинхронных вызовов клиент может вызывать методы сервера параллельно и в процессе ожидания выполнять
дополнительную работу.
В данном случае услуги, предоставляемые сервером, сводятся к выполнению двух арифметических операций (сложение и
умножение на 2). Работа, которую клиент выполняет в ожидании ответов от сервера, состоит в выводе на консоль отметок о
завершении очередного
100 mc временного интервала.
using System;
using System.Threading;
using System.Runtime.Remoting;
public class Server {
public static bool Sum(int x, int y, out int z) {
Console.WriteLine(
"Server (Sum method) thread ="+
Thread.CurrentThread.GetHashCode()+
"; PoolThread = "+
Thread.CurrentThread.IsThreadPoolThread);
Thread.Sleep(1000);
z = 0;
try {
z = checked((int)(x + y));
}
catch (Exception){
return false;
}
return true;
}
public static bool MultBy2(int x, out int y) {
Console.WriteLine(
"Server (MultBy2 method) thread ="+
Thread.CurrentThread.GetHashCode()+
"; PoolThread = "+
Thread.CurrentThread.IsThreadPoolThread);
Thread.Sleep(1000);
y = 0;
try {
y = checked((int)(x*2));
}
catch (Exception){
return false;
}
return true;
}
}
public class Client {
private static int workCount = 0;
private delegate bool HardFunction2Args(
int x, int y, out int result);
private delegate bool HardFunction1Arg(
int x, out int result);
private static void SumCallback(IAsyncResult ar) {
int z;
HardFunction2Args sum =
(HardFunction2Args)ar.AsyncState;
bool result = sum.EndInvoke(out z, ar);
if (result) Console.WriteLine(
"SumCallback: Sum = " + z);
else Console.WriteLine(
"SumCallback: Bad arguments for Server.Sum");
workCount++;
}
private static void MultCallback(IAsyncResult ar) {
int z;
HardFunction1Arg mult =
(HardFunction1Arg)ar.AsyncState;
bool result = mult.EndInvoke(out z, ar);
if (result) Console.WriteLine(
"MultCallback: MultBy2 = " + z);
else Console.WriteLine(
"MultCallback: Bad argument for MultBy2");
workCount++;
}
public static void Main() {
int sumResult, multResult, count = 0;
Console.WriteLine("Client thread ="+
Thread.CurrentThread.GetHashCode()+
"; PoolThread = "+
Thread.CurrentThread.IsThreadPoolThread);
HardFunction2Args sum =
new HardFunction2Args(Server.Sum);
HardFunction1Arg mult =
new HardFunction1Arg(Server.MultBy2);
AsyncCallback sumCallback =
new AsyncCallback(SumCallback);
AsyncCallback multCallback =
new AsyncCallback(MultCallback);
IAsyncResult arSum = sum.BeginInvoke(3, 4,
out sumResult, sumCallback, sum);
IAsyncResult arMult = mult.BeginInvoke(5,
out multResult, multCallback, mult);
while (workCount < 2) {
Console.WriteLine("Client thread: count = "+
count++);
Thread.Sleep(100);
}
Console.WriteLine("Bye!");
}
}
Комментарии к коду.
Сервер и клиент представлены соответственно классами
Server и Client.
Сервер
Сервер реализует два статических метода:
z Метод
public static bool Sum(int x, int y, out int z) { . . . } обеспечивает сложение двух чисел типа int. Результат
записывается в переменную
z типа int. В случае возникновения переполнения возвращается false, при его отсутствии -
true.
Временная сложность проводимых вычислений имитируется с помощью вызова
Thread.Sleep(1000).
В самом начале, еще до проведения вычислений, на консоль выводится хеш потока, выполняющего вызов
(
Thread.CurrentThread.GetHashCode()), и информация о принадлежности данного потока классу рабочих потоков из пула
потоков (
Thread.CurrentThread.IsThreadPoolThread)).
z Метод
public static bool MultBy2(int x, out int y) { . . . } обеспечивает умножение числа на 2 и реализован
аналогично предыдущему методу.
Здесь важно отметить, что разработчик сервера не заботится о том, как именно будут вызываться методы сервера клиентами -
синхронно или асинхронно. Все зависит от клиента. Он может вызывать методы сервера как синхронно, так и асинхронно.
Клиент. Типы используемых делегатов
В данном примере клиент вызывает методы сервера асинхронно, что достигается за счет использования делегатов.
В коде клиента используются делегаты трех типов:
z HardFunction2Args
Этот тип определяется в классе
Client:
private delegate bool HardFunction2Args( int x, int y, out int result);
Данный делегат может делегировать вызов (как синхронный так и асинхронный) любому методу (как статическому так и
нестатическому) любого класса с заданной сигнатурой (два входных параметра типа
int, один выходной типа int,
возвращаемое значение типа bool). В нашем случае вызов будет делегироваться методу Server::Sum.
z HardFunction1Arg
Этот тип также определяется в классе
Client:
private delegate bool HardFunction1Arg( int x, out int result);
Данный делегат может делегировать вызов (как синхронный так и асинхронный) любому методу (как статическому так и
нестатическому) любого класса с заданной сигнатурой (один входной параметр типа
int, один выходной типа int,
возвращаемое значение типа
bool). В данном случае вызов будет делегироваться методу Server::MultBy2.
z AsyncCallback
Этот тип определен в
System. Он может использоваться для делегирования вызова методу со следующей сигнатурой:
{ один входной параметр типа IAsyncResult (тип определен в System);
{ возвращаемое значение отсутствует (void).
В нашем случае делегаты данного типа будут использоваться для делегирования вызовов методам клиента
Client::SumCallback и
Client::MultCallback.
Метод
private static void SumCallback(IAsyncResult ar) { . . . }
клиента вызывается инфраструктурой асинхронных вызовов для уведомления клиента о том, что сделанный им ранее
асинхронный вызов метода
Sum сервера завершен.
Аналогично, метод
private static void MultCallback(IAsyncResult ar) { . . . }
клиента вызывается инфраструктурой асинхронных вызовов для уведомления клиента о том, что сделанный им ранее
асинхронный вызов метода
MultBy2 сервера завершен также.
Клиент. Инициирование асинхронных вызовов
Прежде чем обсуждать завершение асинхронных вызовов уместно рассмотреть их инициирование. Для этого обратимся к коду
метода
Client::Main.
Прежде всего клиент выводит на консоль хеш основного потока
(
Thread.CurrentThread.GetHashCode()) и информацию о принадлежности данного потока классу рабочих потоков из пула
потоков
(Thread.CurrentThread.IsThreadPoolThread)).
Далее создаются два делегата для инициирования асинхронных вызовов методов сервера. Делегат
sum
HardFunction2Args sum =
new HardFunction2Args(Server.Sum);
используется для асинхронного вызова метода Server::Sum, а делегат mult
HardFunction1Arg mult =
new HardFunction1Arg(Server.MultBy2);
используется для асинхронного вызова метода Server::MultBy2.
Далее формируются делегаты
sumCallback и multCallback
AsyncCallback sumCallback =
new AsyncCallback(SumCallback);
AsyncCallback multCallback =
new AsyncCallback(MultCallback);
которые будут использваться инфраструктурой асинхронных вызовов для уведомления клиента о завершении соответственно
Server::Sum и
Server::MultBy2 вызовов.
Инициирование асинхронных вызовов производится клиентом следующим образом:
IAsyncResult arSum = sum.BeginInvoke(3, 4,
out sumResult, sumCallback, sum);
IAsyncResult arMult = mult.BeginInvoke(5,
out multResult, multCallback, mult);
Немного о делегатах в связи с асинхронными вызовами