Добрынин В.Ю. Технологии компонентного программирования
Подождите немного. Документ загружается.
Правила обработки вложенного вызова зависят от реентерабельности контекста синхронизации. Если контекст реентерабельный,
то никакой специальной обработки вложенных вызовов производить не надо. В этом случае любой новый вызов (в том числе и
вложенный) имеет право исполняться в реентерабельном контексте в то время, как выполняемый в данный момент вызов
приостановлен на время ожидания
ответа на какой-либо сделаный в его рамках внешний вызов.
Если же контекст синхронизации нереентерабельный, то никакой новый вызов не может выполняться в данном контексте, если в
данном контексте в данное время выполняется какой-либо синхронный вызов. Если при выполнении синхронного вызова была
инициирована цепочка вызовов и последний в этой
цепочке вызов является вызовом в данный контекст синхронизации, то его
блокировка приведет к блокировке всей очереди вызовов (текущий вызов никогда не завершится). Именно в связи с этим в случае
нереентерабельного контекста синхронизации и синхронного исполняемого вызова вложенный вызов должен исполняться вне
очереди.
Рассмотрим определенный в этом же классе
SynchronizationAttribute метод IsNestedCall.
internal bool IsNestedCall(IMessage reqMsg) {
bool bNested = false;
if (!IsReEntrant) {
String lcid = SyncCallOutLCID;
if (lcid != null) {
LogicalCallContext callCtx =
(LogicalCallContext)
reqMsg.Properties[Message.CallContextKey];
if ( callCtx!=null &&
lcid.Equals(callCtx.RemotingData.LogicalCallID)) {
bNested = true;
}
}
if (!bNested && AsyncCallOutLCIDList.Count>0) {
LogicalCallContext callCtx =
(LogicalCallContext)
reqMsg.Properties[Message.CallContextKey];
if (AsyncCallOutLCIDList.Contains(
callCtx.RemotingData.LogicalCallID)) {
bNested = true;
}
}
}
return bNested;
}
Этот метод возвращает true если текущий контекст (точнее, текущий домен синхронизации) нереентерабельный, а новый вызов
(
reqMsg) является вложенным для исполняемого в данный момент синхронного вызова, или для одного из исходящих
асинхронных вызовов. Во всех остальных случаях возвращается
false.
Вначале выясняется синхронность выполняемого в данный момент вызова:
String lcid = SyncCallOutLCID;
if (lcid != null) {
. . . . .
}
Свойство SyncCallOutLCID атрибута синхронизации возвращает идентификатор логического вызова исполняемого в данный
момент вызова, если этот вызов синхронный. В противном случае возвращается
null.
Теперь в случае синхронности исполняемого вызова мы получаем доступ к контексту вызова для вызова
reqMsg :
LogicalCallContext callCtx =
(LogicalCallContext)
reqMsg.Properties[Message.CallContextKey];
Надо заметить, что тут имеется ввиду класс Message из пространства имен System.Runtime.Remoting.Messaging. Реализация
такого класса в этом пространстве имен имеется в Rotor, но отсутствует в .NET. Статическое поле
CallContextKey равно
__CallContext.
Если доступ к контексту вызова получен, то проверяется, что идентификатор логического вызова исполняемого в данный момент
синхронного вызова
lcid совпадает с идентификатором логического вызова для вызова reqMsg. В случае их совпадения флаг
вложенности
bNested получает значение true:
if ( callCtx!=null &&
lcid.Equals(callCtx.RemotingData.LogicalCallID)) {
bNested = true;
}
Здесь опять приходится отметить, что в .NET у класса LogicalCallContext нет свойства RemotingData типа
CallContextRemotingData, нет и самого класса CallContextRemotingData и его свойства LogicalCallID.
Если в данный момент не выполняется синхронный вызов, или новый вызов не является вложенным для исполняемого
синхронного вызова, то начинается проверка вложенности нового вызова в один из исходящих асинхронных вызовов.
Выполнение условия
AsyncCallOutLCIDList.Count>0 означает, что список идентификаторов логических вызовов,
соответствующих исходящим асинхронным вызовам, не пуст. Свойство
AsyncCallOutLCIDList типа ArrayList возвращает
ссылку на этот список.
В этом случае как и ранее выясняется контекст вызова
reqMsg и проверяется, что соответствующий ему идентификатор
логического вызова содержится в списке исходящих асинхронных вызовов. В случае успеха флаг
bNested получает значение
true:
LogicalCallContext callCtx =
(LogicalCallContext)
reqMsg.Properties[Message.CallContextKey];
if (AsyncCallOutLCIDList.Contains(
callCtx.RemotingData.LogicalCallID)) {
bNested = true;
}
Возвращаемся вновь к коду метода HandleWorkRequest, который мы рассматриваем в данный момент только для случая
инкапсулированного синхронного вызова.
Здесь возможны два случая:
z IsNestedCall(work._reqMsg) == false
Это случай соответствует реентерабельному контексту или невложенному вызову.
В случае синхронности инкапсулированного вызова его нужно поставить в очередь (или выполнять без постановки в
очередь, если в данный момент времени в данном домене синхронизации не выполняется ни один поток и очередь работ
пуста).
z IsNestedCall(work._reqMsg) == true
А вот в этом случае при синхронности инкапсулированного в
work вызова и его вложенности в исполняемый синхронный
вызов инкапсулированный вызов следует выполнять сразу же без постановки в очередь, так как его выполнения ожидает
основной выполняемый в данный момент синхронный вызов.
Другая возможность - инкапсулированный вызов является вложенным для исходящего асинхронного вызова. Но можно ли
этот инкапсулированный вызов исполнять вне очереди и
в этом случае? Отложим обсуждение этого вопроса.
Итак, в первом случае (не вложенный вызов), поток входит в критическую секцию
lock(work) {
. . . . .
}
заблокировав доступ к работе work, инкапсулирующей входящий вызов. В рамках этой критической секции поток ожидает входа
в критическую секцию, для доступа к которой надо заблокировать очередь работ:
lock(_workItemQueue) {
. . . . .
}
Войдя в эту вторую критическую секцию, поток проверяет - имеется ли блокировка домена синхронизации. Поле _locked
атрибута синхронизации равно
false в том случае, когда после выполнения очередной работы обнаруживается, что очередь
работ пуста, то есть нет работы, которую следует начинать выполнять в рамках данного домена синхронизации.
Если домен синхронизации не заблокирован, и очередь работ пуста, то домен блокируется, но флаг постановки работы в очередь
bQueued не задается (работа может быть выполнена сразу же):
if ((!_locked) &&
(_workItemQueue.Count == 0)) {
_locked = true;
bQueued = false;
}
В противном случае задается флаг bQueued постановки работы в очередь , в самой работе задается флаг, указывающий на то, что
она стоит в очереди (
work.SetWaiting()) и выполняется реальная запись работы в очередь:
else {
bQueued = true;
work.SetWaiting();
_workItemQueue.Enqueue(work);
}
Теперь поток выходит из второй (внутренней) критической секции, разблокировав очередь работ. Но что ему делать дальше?
Если работа была поставлена в очередь, то в силу синхронности вызова данный поток должен сам перейти в состояние ожидания,
из которого он выйдет только тогда, когда только что поставленная в очередь работа продвинется в ее
начало и будет готова для
выполнения:
if (bQueued == true) {
Monitor.Wait(work);
if (!work.IsDummy()) {
DispatcherCallBack(null, true);
}
else {
lock(_workItemQueue) {
_workItemQueue.Dequeue();
}
}
}
Вызов Monitor.Wait(work) переведет текущий поток в состояние ожидания, причем этот поток освободит ранее
заблокированную им работу
work и будет ожидать сигнала, говорящего о том, что состояние объекта work изменилось, нужно
проснуться и продолжить работу с этим объектом. Это сигнал будет выдан другим потоком, заметившим, что работа
work первая
в очереди и нет препятствий для ее выполнения.
Следующая за
Monitor.Wait(work);
строка кода будет выполняться уже разбуженным потоком, который вновь получает исключительный доступ к объекту work.
Если данная работа не является работой-заглушкой (об этом позже), то вызывается уже рассмотренный метод
DispatcherCallBack, который и извлечет эту работу из очереди, выполнит ее и инициирует выполнение следующей работы. В
случае же работы-заглушки просто блокируется очередь работ и эта работа-заглушка удаляется из очереди.
А вот что происходит с работой, которую не пришлось ставить в очередь:
else {
if (!work.IsDummy()) {
work.SetSignaled();
ExecuteWorkItem(work);
HandleWorkCompletion();
}
}
Если это не заглушка, устанавливается флаг готовности к выполнению, работа выполняется и потом все подготавливается для
выполнения следующей работы.
Здесь надо бы рассмотреть код для метода
HandleWorkCompletion(), который собственно и подготавливает все для выполнения
следующей работы, но в очередной раз отложим это рассмотрение на потом.
Теперь рассмотрим второй случай, когда вызов синхронный вложенный, а контекст нереентерабельный. Такой вызов надо
выполнять вне очереди:
if (!IsNestedCall(work._reqMsg)) {
. . . . .
}
else {
work.SetSignaled();
work.Execute();
}
Тут устанавливается флаг готовности к выполнению, работа выполняется, но никакой подготовки к выполнению следующей
работы не проводится, т.к. уже имеется находящийся в состоянии выполнения вызов, ожидающий выполнения данного
синхронного вызова.
Тут все в порядке, если только инкапсулированный вызов вложен в исполняемый синхронный вызов. Если же этот
инкапсулированный вызов вложен в
некоторый исходящий асинхронный вызов, то могут быть проблемы. Например, в данное
время в данном домене синхронизации уже может выполняться некоторый синхронный вызов и параллельное выполнение еще
одного вызова противоречит логике использования домена синхронизации (?!).
Теперь настало время рассмотреть код ранее упомянутого метода
HandleWorkCompletion(). Как и ранее рассмотрим только ту ветвь, которая связана с обработкой синхронных вызовов.
internal virtual void HandleWorkCompletion() {
WorkItem nextWork = null;
bool bNotify = false;
lock (_workItemQueue) {
if (_workItemQueue.Count >= 1) {
nextWork = (WorkItem) _workItemQueue.Peek();
bNotify = true;
nextWork.SetSignaled();
}
else {
_locked = false;
}
}
if (bNotify) {
if (nextWork.IsAsync()) {
. . . . .
}
else {
lock(nextWork) {
Monitor.Pulse(nextWork);
}
}
}
}
Данный метод вызывается по завершении обработки некоторой работы. В это время домен синхронизации еще блокирован
(
_locked == true).
Прежде всего блокируется очередь работ. Если она не пуста, мы получаем ссылку на работу (
nextWork), стоящую в этой очереди
первой (не извлекая ее из очереди) и устанавливаем флаг готовности данной работы к выполнению. Если же очередь пуста, домен
синхронизации разблокируется (_locked = false). Это означает, что вновь пришедший синхронный вызов будет обработан
сразу же без постановки в очередь.
Если в начале очереди находится синхронная работа (
nextWork), то это означает, что имеется поток, который находится в
состоянии ожидания того момента, когда работа
nextWork будет готова к выполнению. Этот поток в свое время впал в состояние
ожидания находясь в некоторой критической секции, в которую он попал заблокировав объект
nextWork. Находясь в этой
критической секции он был должен освободить эту блокировку, так как в то время работа
nextWork была поставлена в очередь
работ и этот поток не мог продолжить ее обработку. Теперь его пора разбудить. Это делает текущий поток, в котором
выполняется код метода
HandleWorkCompletion:
lock(nextWork) {
Monitor.Pulse(nextWork);
}
Текущий поток входит в критическую секцию (заблокировав nextWork) и уведомляет все заинтересованные потоки о том, что
состояние объекта
nextWork как-то изменилось. Эти находящиеся в состоянии ожидания потоки переходят в состояние
готовности и по выходе текущего потока из данной критической секции пытаются захватить контроль над объектом
nextWork. В
нашем случае такой поток единственный и именно он ответственен за продолжение обработки вызова, инкапсулированного в
работу
nextWork. Конкретнее, это поток, выполняющий код метода HandleWorkRequest начиная со стороки, следующей за
строкой
Monitor.Wait(work);
Как перехватчик обрабатывает асинхронные вызовы
Вновь возвращаемся к классу SynchronizedServerContextSink, который реализует интерфейс IMessageSink. В этом интерфейсе
объявлен метод
AsyncProcessMessage, который и должен реализоать обработку асинхронных вызовов в перехватчике входящих
вызовов. Вот так этот метод реализован в классе
SynchronizedServerContextSink:
public virtual IMessageCtrl AsyncProcessMessage(
IMessage reqMsg, IMessageSink replySink) {
WorkItem work = new WorkItem(
reqMsg,
_nextSink,
replySink);
work.SetAsync();
_property.HandleWorkRequest(work);
return null;
}
В данном случае мы имеем два входных параметра. Первый задает вызов в форме сообщения типа IMessage (как и в случае
синхронного вызова). А вот второй параметр специфичен для асинхронных вызовов. Он задает ссылку на еще один перехватчик,
который ответственен за доставку уведомления о завершении асинхронного вызова. И, наконец, возвращаемое значение имеет
также специфический для асинхронных вызовов тип -
IMessageCtrl. В .NET интерфейс IMessageCtrl объявляет единственный
метод
Cancel, с помощью которого можно прервать выполнение асинхронного вызова. В данном случае метод
AsyncProcessMessage всегда возвращает null, не предоставляя такую возможность.
Функциональность данного метода вполне понятна. Пришедший асинхронный вызов инкапсулируется в работу
work, далее для
нее устанавливается флаг асинхронности, позволяющий различать синхронные и асинхронные работы, и, наконец, данная работа
направляется на обработку в свойство синхронизации
_property.
То, как метод
HandleWorkRequest обрабатывает синхронные запросы, уже было рассмотрено ранее. Теперь изучим ветви его
кода, ответственные за обработку асинхронных запросов.
Как свойство синхронизации обрабатывает инкапсулированный асинхронный
вызов, полученный от перехватчика
Ниже приведена часть кода HandleWorkRequest, которая относится к обработке именно aсинхронных вызовов:
internal virtual void HandleWorkRequest(WorkItem work) {
bool bQueued;
if (!IsNestedCall(work._reqMsg)) {
if (work.IsAsync()) {
bQueued = true;
lock (_workItemQueue) {
work.SetWaiting();
_workItemQueue.Enqueue(work);
if ( (!_locked) &&
(_workItemQueue.Count == 1) ) {
work.SetSignaled();
_locked = true;
_asyncWorkEvent.Set();
}
}
}
else {
. . . . .
}
}
else {
work.SetSignaled();
work.Execute();
}
}
Итак, если вызов не вложенный, происходит следующее. Блокируется очередь работ (текущий поток входит в критическую
секцию)
lock (_workItemQueue) {
. . . . .
}
и текущая работа помечается как ожидающая в очереди
work.SetWaiting();
Потом эта работа становится в очередь работ
_workItemQueue.Enqueue(work);
и, если домен синхронизации не заблокирован и данная работа в очереди единственна, инициируется ее выполнение. Для этого
после установки флага готовности работы к выполнению и блокировки домена событие
_asyncWorkEvent переводится в
состояние signaled. Это приведет к тому, что свободный рабочий поток из пула потоков начнет выполнять метод
DispatcherCallBack, в процессе чего данная работа и будет извлечена из очереди и отправлена на выполнение:
if ( (!_locked) &&
(_workItemQueue.Count == 1) ) {
work.SetSignaled();
_locked = true;
_asyncWorkEvent.Set();
}
Если же очередь была не пуста, то только-что поставленная в эту очередь работа ждет своей очереди и будет извлечена из нее в
свое время.
Если вызов вложенный, то инкапсулирующая его работа выполняется сразу же без постановки в очередь:
internal virtual void HandleWorkRequest(WorkItem work) {
bool bQueued;
if (!IsNestedCall(work._reqMsg)) {
. . . . .
}
else {
work.SetSignaled();
work.Execute();
}
}
Возникающие при этом проблемы уже обсуждались при рассмотрении синхронного случая.
Теперь рассмотрим ту ветвь кода метода
HandleWorkCompletion(), которая связана с обработкой асинхроннных вызовов (в
асинхронном случае этот метод будет вызван из
DispatcherCallBack, который будет выполняться рабочим потоком,
инициированным переводом свойства
_asyncWorkEvent в состояние signaled):
internal virtual void HandleWorkCompletion() {
WorkItem nextWork = null;
bool bNotify = false;
lock (_workItemQueue) {
if (_workItemQueue.Count >= 1) {
nextWork = (WorkItem) _workItemQueue.Peek();
bNotify = true;
nextWork.SetSignaled();
}
else {
_locked = false;
}
}
if (bNotify) {
if (nextWork.IsAsync()) {
_asyncWorkEvent.Set();
}
else {
. . . . .
}
}
}
Критическая секция
lock (_workItemQueue) {
. . . . .
}
уже была рассмотрена ранее.
Пусть теперь в начале очереди находится асинхронная работа (
nextWork). В этом случае событие _asyncWorkEvent
устанавливается в состояние signaled и на этом вся подготовка к обработке новой работы завершается.
Перехват исходящего вызова
Формирование перехватчика исходящих вызовов
Напомним, что с каждым контекстом может быть связано несколько цепочек перехватчиков. Формирование связанного со
свойством синхронизации перехватчика входящих вызовов было рассмотрено в предыдущем разделе. Теперь рассмотрим
формирование перехватчика исходящих вызовов.
Класс
SynchronizationAttribute реализует интерфейс
IContributeClientContextSink.
Благодаря этому факту, при формировании нового контекста синхронизации автоматически вызывается метод
GetClientContextSink, объявленный в данном интерфейсе, который и формирует перехватчик исходящих вызовов для данного
контекста.
Зачем нужен перехватчик исходящих вызовов? Предположим, контекст (домен) синхронизации реентерабельный. Это означает,
что с того момента, когда поток, исполняющий некоторый вызов в данном контексте, инициировал вызов за пределы этого
контекста и вошел в состоянии ожидания ответа, очередная
работа может быть извлечена из очереди и может начаться
выполнение инкапсулированного в ней вызова. Перехватчик исходящих вызовов как раз и замечает момент выдачи внешнего
вызова и инициирует обработку очередной работы.
Ниже приводится код метода
GetClientContextSink из Rotor:
public virtual IMessageSink GetClientContextSink(
IMessageSink nextSink) {
InitIfNecessary();
SynchronizedClientContextSink propertySink =
new SynchronizedClientContextSink(
this,
nextSink);
return (IMessageSink) propertySink;
}
Этот код аналогичен коду метода GetServerContextSink, в связи с чем комментарии опущены.
Как и в случае перехватчика входящих вызовов, при одном на весь домен синхронизации свойстве синхронизации, для каждого
контекста в этом домене формируется свой перехватчик исходящих вызовов, имеющий ссылку на это свойство синхронизации.
Класс
SynchronizedClientContextSink наследует классу InternalSink и реализует интерфейс IMessageSink. Его основная
функциональность определяется двумя методами интерфейса
IMessageSink: SyncProcessMessage и AsyncProcessMessage,
обрабатывающими соответственно синхронные и асинхронные исходящие вызовы.
Перехват исходящих синхронных вызовов
Случай реентерабельного контекста
Начнем со случая реентерабельного контекста (домена). Вот соответствующая ветвь кода метода SyncProcessMessage:
public virtual IMessage SyncProcessMessage(
IMessage reqMsg) {
IMessage replyMsg;
if (_property.IsReEntrant) {
_property.HandleThreadExit();
replyMsg = _nextSink.SyncProcessMessage(reqMsg);
_property.HandleThreadReEntry();
}
else {
. . . . .
}
return replyMsg;
}
Прежде всего нужно уведомить свойство синхронизации (_property)
_property.HandleThreadExit();
о том, что выполняется вызов за пределы текущего контекста. Это позволит свойству синхронизации инициировать выполнение
очередной работы. Рассмотрим код соответствующего метода
HandleThreadExit класса SynchronizationAttribute:
internal virtual void HandleThreadExit() {
HandleWorkCompletion();
}
Код для HandleWorkCompletion уже рассматривался. В результате его выполнения будет проверено состояние очереди работ.
Если она не пуста, то очередная работа будет помечена флагом готовности к выполнению. В противном случае домен
синхронизации будет разблокирован, что просто означает возможность выполнения вновь поступившего синхронного вызова без
записи в очередь. Далее в случае наличия готовой к
выполнению работы ее выполнение инициируется. В случае асинхронной
работы для этого достаточно перевести событие
_asyncWorkEvent в состояние signaled, а в случае синхронной - разбудить
занятый ее выполнением процесс путем вызова
Monitor.Pulse(nextWork), где nextWork - ссылка на готовую к выполнению
синхронную работу.
Теперь вызов
reqMsg в форме сообщения пересылается следующему перехватчику в цепочке перехватчиков исходящих вызовов
для данного контекста. Текущий поток блокируется в ожидании ответа.
replyMsg = _nextSink.SyncProcessMessage(reqMsg);
После получения ответа на внешний вызов, текущий поток не может безоглядно продолжить выполнение основного вызова, так
как в связи с реентерабельностью контекста (домена), возможно, в данном домене уже выполняется какой-либо другой поток.
Таким образом, текущий поток должен ожидать своей очереди. Как ему встать в эту очередь? Можно воспользоваться тем, что
свойство синхронизации уже поддерживает одну очередь - очередь работ. Можно создать фиктивную работу, включив в нее
только информацию о контексте, где этот поток должен выполняться, и о контексте вызова, связанного с этим потоком.
Информацию о самом вызове в работу включать не надо, так как этот поток уже находится в состоянии его
выполнения. После
постановки фиктивной работы в очередь данный поток заснет и будет разбужен только тогда, когда эта фиктивная работа
окажется в очереди работ на первом месте.
Вся вышеописанная логика запускается следующим вызовом:
_property.HandleThreadReEntry();
Ниже приводится код для метода HandleThreadReEntry класса
SynchronizationAttribute:
internal virtual void HandleThreadReEntry() {
WorkItem work = new WorkItem(null, null, null);
work.SetDummy();
HandleWorkRequest(work);
}
Здесь создается фиктивная работа, помечается флагом фиктивности и ставится в очередь в процессе выполнения кода
HandleWorkRequest.
Последний метод уже неоднократно обсуждался, но теперь имеет смысл особо рассмотреть ту его ветвь, которая связана с
обработкой фиктивной работы. По умолчанию эта работа синхронна. Кроме того, в связи с реентерабельностью контекста,
инкапсулированный в ней вызов не вложенный.
internal virtual void HandleWorkRequest(WorkItem work) {
bool bQueued;
if (!IsNestedCall(work._reqMsg)) {
if (work.IsAsync()) {
. . . . .
}
else {
lock(work) {
lock(_workItemQueue) {
if ((!_locked) &&
(_workItemQueue.Count == 0)) {
_locked = true;
bQueued = false;
}
else {
bQueued = true;
work.SetWaiting();
_workItemQueue.Enqueue(work);
}
}
if (bQueued == true) {
Monitor.Wait(work);
if (!work.IsDummy()) {
. . . . .
}
else {
lock(_workItemQueue) {
_workItemQueue.Dequeue();
}
}
}
else {
if (!work.IsDummy()) {
. . . . .
}
}
}
}
}
else {
. . . . .
}
}
Таким образом, если домен синхронизации не блокирован и очередь пуста, то единственной операцией для фиктивной работы
является блокирование домена, так как наш поток может без ожидания продолжать выполнение основного вызова.
Если домен был заблокирован, то фиктивная работа помечается как стоящая в очереди и записывается в очередь работ. Далее
связанный с этой
работой поток засыпает
Monitor.Wait(work);
и спит до тех пор, пока фиктивная работа не будет готова к выполнению, продвинувшись в очереди на первое место.
После пробуждения данного потока он выполняет код
if (!work.IsDummy()) {
. . . . .
}
else {
lock(_workItemQueue) {
_workItemQueue.Dequeue();
}
}
Иными словами, фиктивная работа просто удаляется из очереди, после чего выполнение метода SyncProcessMessage завершается
возвращением результата
replyMsg и разбуженный поток продолжает выполнять основной вызов.
Случай нереентерабельного контекста
Теперь рассмотрим ветвь метода SyncProcessMessage класса
SynchronizedClientContextSink, относящуюся к случаю нереентерабельного контекста:
public virtual IMessage SyncProcessMessage(
IMessage reqMsg) {
IMessage replyMsg;
if (_property.IsReEntrant) {
. . . . .
}
else {
LogicalCallContext cctx =
(LogicalCallContext)
reqMsg.Properties[Message.CallContextKey];
String lcid = cctx.RemotingData.LogicalCallID;
bool bClear = false;
if (lcid == null) {
lcid = Identity.GetNewLogicalCallID();
cctx.RemotingData.LogicalCallID = lcid;
bClear = true;
}
bool bTopLevel=false;
if (_property.SyncCallOutLCID==null) {
_property.SyncCallOutLCID = lcid;
bTopLevel = true;
}
replyMsg = _nextSink.SyncProcessMessage(reqMsg);
if (bTopLevel) {
_property.SyncCallOutLCID = null;
if (bClear) {
LogicalCallContext cctxRet =
(LogicalCallContext)
replyMsg.Properties[Message.CallContextKey];
cctxRet.RemotingData.LogicalCallID = null;
}
}
}
return replyMsg;
}
В нереентерабельном случае для предотвращения блокировок надо заботиться о первоочередном выполнении вновь
поступающих вызовов, инициированных исполняемым в данный момент синхронным вызовом. Для этого используются
уникальные идентификаторы, совпадающие у исходного синхронного вызова и всех инициированных им вызовов. При
совпадении идентификатора нового вызова с идентификатором исполняемого вызова можно сделать заключение о том, что
новый вызов инициирован исполняемым вызовом и должен выполняться без промедления. Итак, при отправлении синхронного
вызова за пределы текущего контекста мы должны запомнить его идентификатор и быть готовыми незамедлительно обработать
любой входящий вызов, если его идентификатор совпадает с идентификатором исходящего в данный момент вызова.
Прежде всего нужно узнать идентификатор исходящего вызова.
Для этого получаем доступ к его контексту вызова
LogicalCallContext cctx =
(LogicalCallContext)
reqMsg.Properties[Message.CallContextKey];
и, затем, к самому идентификатору
String lcid = cctx.RemotingData.LogicalCallID;
Может оказаться, что в данный момент исходящий вызов (reqMsg) еще не имеет идентификатора. В таком случае присвоим ему
новый еще не использованный идентификатор
bool bClear = false;
if (lcid == null) {
lcid = Identity.GetNewLogicalCallID();
cctx.RemotingData.LogicalCallID = lcid;
bClear = true;
}
Здесь статический метод GetNewLogicalCallID класса Identity генерирует новый идентификатор, который и запоминается в
контексте вызова.
Если нам пришлось самим построить новый идентификатор для исходящего синхронного вызова, то его нужно и сохранить в
свойстве синхронизации. В этом случае условное выражение в следующем операторе
if равно true:
bool bTopLevel=false;
if (_property.SyncCallOutLCID==null) {
_property.SyncCallOutLCID = lcid;
bTopLevel = true;
}
Ну а теперь передаем исходящий вызов следующему перехватчику исходящих вызовов и ждем ответа
replyMsg = _nextSink.SyncProcessMessage(reqMsg);
Ну а теперь (после получения ответа) надо за собой почистить (если мы что-то поменяли в контексте вызова и в свойстве
синхронизации):
if (bTopLevel) {
_property.SyncCallOutLCID = null;
if (bClear) {
LogicalCallContext cctxRet =
(LogicalCallContext)
replyMsg.Properties[Message.CallContextKey];
cctxRet.RemotingData.LogicalCallID = null;
}
}
Здесь мы обнуляем свойство SyncCallOutLCID свойства синхронизации (если мы его только что сами установили) и обнуляем
идентификатор для полученного ответа
replyMsg (если мы сами задавали этот идентификатор для исходящего вызова).
Перехват исходящих асинхронных вызовов
Для обработки исходящих асинхронных вызовов перехватчик использует следующий код:
public virtual IMessageCtrl AsyncProcessMessage(
IMessage reqMsg,
IMessageSink replySink) {
IMessageCtrl msgCtrl = null;
if (!_property.IsReEntrant) {
LogicalCallContext cctx =
(LogicalCallContext)
reqMsg.Properties[Message.CallContextKey];
String lcid = Identity.GetNewLogicalCallID();
cctx.RemotingData.LogicalCallID = lcid;
_property.AsyncCallOutLCIDList.Add(lcid);
}
AsyncReplySink mySink =
new AsyncReplySink(replySink, _property);
msgCtrl = _nextSink.AsyncProcessMessage(
reqMsg,
(IMessageSink)mySink);
return msgCtrl;
}
В качестве входных параметров задаются сам вызов reqMsg и перехватчик, на который вызывающая сторона ожидает получения
уведомления о завершении процесса выполнения асинхронного вызова.
В случае нереентерабельного контекста исходящему асинхронному вызову назначается новый идентификатор и этот
идентификатор сохраняется в списке
_asyncLcidList исходящих асинхронных вызовов, поддерживаемому свойством
синхронизации (доступ через свойство
AsyncCallOutLCIDList). Заметим, что в случае асинхронных вызовов нет нужны
сохранять один и тот же идентификатор по всей цепочке вызовов, в связи с чем здесь не проверяется наличие идентификатора, а
сразу же назначается новый:
if (!_property.IsReEntrant) {
LogicalCallContext cctx =
(LogicalCallContext)
reqMsg.Properties[Message.CallContextKey];
String lcid = Identity.GetNewLogicalCallID();
cctx.RemotingData.LogicalCallID = lcid;
_property.AsyncCallOutLCIDList.Add(lcid);
}
Зачем вообще сохраняется список идентификаторов исходящих асинхронных вызовов? И почему он обновляется только для
нереентерабельного случая?
Единственно, для чего этот список необходим (и именно в случае нереентерабельного контекста) - для определения того,
является ли новый входящий вызов вложенным (см. код методов
IsNestedCall и HandleWorkRequest). При изложении этого
вопроса ранее уже отмечалось наличие некоторых проблем, связанных с определение понятия вложенного вызова и его
использования.
Теперь пора отправить исходящий асинхронный вызов следующему перехватчику исходящих асинхронных вызовов. И тут мы
должны указать, куда посылать уведомления о завершении вызова. Непосредственно использовать перехватчик уведомлений
replySink, предоставленный вызывающей стороной, нельзя, т.к. его непосредственное использование может нарушить логику
синхронизации, поддерживаемую в домене синхронизации. В связи с этим создается специальный перехватчик уведомлений
mySink, который придерживается логики синхронизации и обеспечивает безопасную работу с replySink:
AsyncReplySink mySink =
new AsyncReplySink(replySink, _property);
Класс AsyncReplySink будет рассмотрен чуть позже.
И вот, наконец, исходящий асинхронный вызов
reqMsg передается следующему перехватчику исходящих асинхронных вызовов, а
для получения уведомления указывается
mySink:
msgCtrl = _nextSink.AsyncProcessMessage(
reqMsg,
(IMessageSink)mySink);
return msgCtrl;
Теперь рассмотрим класс AsyncReplySink:
internal class AsyncReplySink : IMessageSink {
. . . . .
}
В конструкторе в полях _nextSink и _property сохраняются соответственно ссылка на следующий перехватчик (в нашем случае
это будет
replySink) и ссылка на свойство синхронизации
internal AsyncReplySink(IMessageSink nextSink,
SynchronizationAttribute prop) {