Если домен синхронизации не заблокирован, и очередь работ пуста, то домен блокируется, но флаг постановки работы в очередь bQueued не задается (работа может быть выполнена сразу же):
if ((!_locked) &&
(_workltemQueue.Count == 0)) {
_locked = true;
bQueued = false;
}
В противном случае задается флаг bQueued постановки работы в очередь, в самой работе задается флаг, указывающий на то, что она стоит в очереди (work.SetWaiting ()) и выполняется реальная запись работы в очередь:
else {
bQueued = true;
work.SetWaiting();
_workltemQueue.Enqueue(work);
}
Теперь поток выходит из второй (внутренней) критической секции, разблокировав очередь работ. Но что ему делать дальше?
Если работа была поставлена в очередь, то в силу синхронности вызова данный поток должен сам перейти в состояние ожидания, из которого он выйдет только тогда, когда только что поставленная в очередь работа продвинется в ее начало и будет готова для выполнения:
if (bQueued == true) {
Monitor.Wait(work);
if (!work.IsDummy()) {
DispatcherCaiiBack(null, true);
}
else {
lock(workltemQueue) {
_workltemQueue.Dequeue();
}
}
}
Вызов Monitor.Wait (work) переведет текущий поток в состояние ожидания, причем этот поток освободит ранее заблокированную им работу work и будет ожидать сигнала, говорящего о том, что состояние объекта work изменилось, нужно проснуться и продолжить работу с этим объектом. Это сигнал будет выдан другим потоком, заметившим, что работа work первая в очереди и нет препятствий для ее выполнения.
Следующая за Monitor.Wait(work); строка кода будет выполняться уже разбуженным потоком, который вновь получает исключительный доступ к объекту work. Если данная работа не является работой-заглушкой (об этом позже), то вызывается уже рассмотренный метод DispatcherCallBack, который и извлечет эту работу из очереди, выполнит ее и инициирует выполнение следующей работы. В случае же работы-заглушки просто блокируется очередь работ и эта работа-заглушка удаляется из очереди.
А вот что происходит с работой, которую не пришлось ставить в очередь:
else {
if (!work.IsDummy()) {
work.SetSignaled();
ExecuteWorkltem(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) _workltemQueue.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. Конкретнее, это поток, выполняющий код метода HandieWorkRequest начиная со строки, следующей за строкой Monitor.Wait(work);
Как перехватчик обрабатывает асинхронные вызовы
Вновь возвращаемся к классу SynchronizedServerContextSink, который реализует интерфейс IMessageSink. В этом интерфейсе объявлен метод AsyncProcessMessage, который и должен реализовать обработку асинхронных вызовов в перехватчике входящих вызовов. Вот так этот метод реализован в классе SynchronizedServerContextSink: