2.2.8. Пример использования сокетов с событиями
К достоинствам асинхронного режима, основанного на сообщениях, относится то, что нить, обслуживающая сокет, может выходить из состояния ожидания не только при получении данных сокетом, но и по иным сигналам. Протокол обмена который мы до сих пор использовали, не позволяет продемонстрировать это достоинство в полном объеме. Поэтому, прежде чем создавать пример, мы несколько изменим протокол. Формат пакетов оставим прежним, изменим условия, при которых эти пакеты могут быть посланы. Теперь клиент имеет право посылать пакеты, не дожидаясь от сервера ответа на предыдущий пакет, а сервер имеет право посылать пакеты клиенту не только в ответ на его запросы, но и по собственной инициативе. В нашей реализации он будет посылать клиентам строку с текущим временем.
Сервер на асинхронных событиях (пример EventSelectServer на компакт-диске) имеет много общего с рассмотренным ранее многонитевым сервером (пример MultithreadedServer, см. разд. 2.1.12). В нем также есть нить для обработки подключений клиентов и по одной нити на каждого клиента, а главная нить только создает слушающий сокет и запускает обслуживающую его нить.
Еще одним важным отличием нашего сервера от всех предыдущих примеров серверов станет то, что пользователь сможет его остановить в любой момент. Подобную функциональность было бы несложно добавить и к таким серверам, как SelectServer, NonBlockingServer и AsyncSelectServer, которые работают в одной нити. Но остановить нити в многонитевом сервере можно было только одним способом: уничтожив сокеты из главной нити — в этом случае все работающие с этими сокетами нити завершились бы с ошибками. Очевидно, что это порочный подход, не позволяющий корректно завершить работу с клиентами. Режим с использованием событий позволяет предусмотреть реакцию нити на внешний сигнал об отключении. Отключаться сервер будет по нажатию кнопки Остановить.
В листинге 2.63 приведен код нити, взаимодействующей с клиентом (код методов LogMessage и DoLogMessage опущен, т. к. он идентичен приведенному в листингах 2.20 и 2.7 соответственно).
Листинг 2.63. Нить, взаимодействующая с клиентами
unit ClientThread;
{
Нить, обслуживающая одного клиента.
Выполняет цикл, выход из которого возможен по внешнему сигналу или при возникновении ошибки на сокете. Умеет отправлять клиенту сообщения по внешнему сигналу.
}
interface
uses
Windows, Classes, WinSock, Winsock2_Events, ShutdownConst, SysUtils, SyncObjs;
type
TClientThread = class(TThread)
private
// Сообщение, которое нужно добавить в лог,
// хранится в отдельном поле, т. к. метод, вызывающийся через
// Synchronize, не может иметь параметров.
FMessage: string;
// Префикс для всех сообщений лога, связанных с данным клиентом
FHeader: string;
// Сокет для взаимодействия с клиентом
FSocket: TSocket;
// События нити
// FEvents[0] используется для остановки нити
// FEvents[1] используется для отправки сообщения
// FEvents[2] связывается с событиями FD_READ, FD_WRITE и FD_CLOSE
FEvents; array[0..2] of TWSAEvent;
// Критическая секция для доступа к буферу исходящих
FSendBufSection: TCriticalSection;
// Буфер исходящих
FSendBuf: string;
// Вспомогательный метод для вызова через Synchronize
procedure DoLogMessage;
// Функция, проверяющая, завершила ли нить работу
function GetFinished: Boolean;
protected
procedure Execute; override;
// Вывод сообщения в лог главной формы
procedure LogMessage(сonst Msg: string);
// Отправка клиенту данных из буфера исходящих
function DoSendBuf: Boolean;
public
constructor Create(ClientSocket: TSocket; const ClientAddr: TSockAddr);
destructor Destroy; override;
// Добавление строки в буфер исходящих
procedure SendString(const S: string);
// Остановка нити извне
procedure StopThread;
property Finished: Boolean read GetFinished;
end;
ESocketError = class(Exception);
implementation
uses
MainServerUnit;
{ TClientThread }
// Сокет для взаимодействия с клиентом создается в главной нити,
// а сюда передается через параметр конструктора. Для формирования
// заголовка сюда же передается адрес подключившегося клиента
constructor TClientThread.Create(ClientSocket: TSocket; const ClientAddr: TSockAddr);
begin
FSocket:= ClientSocket;
// заголовок содержит адрес и номер порта клиента.
// Этот заголовок будет добавляться ко всем сообщениям в лог
// от данного клиента.
FHeader:=
'Сообщение от клиента ' + inet_ntoa(ClientAddr.sin_addr) +
': ' + IntToStr(ntohs(ClientAddr.sin_port)) + ': ';
// Создаем события и привязываем первое из них к сокету
FEvents[0]:= WSACreateEvent;
if FEvents[0] = WSA_INVALID_EVENT then
raise ESocketError.Create(
FHeader + 'Ошибка при создании события: ' + GetErrorString);
FEvents[1]:= WSACreateEvent;
if FEvents[1] = WSA_INVALID_EVENT then
raise ESocketError.Create(
FHeader + 'Ошибка при создании события: ' + GetErrorString);
FEvents[2]:= WSACreateEvent;
if FEvents[2] = WSA_INVALID_EVENT then raise
ESocketError.Create(
FHeader + 'Ошибка при создании события: ' + GetErrorString);
if WSAEventSelect(FSocket, FEvents[2], FD_READ or FD_WRITE or FD_CLOSE) =
SOCKET_ERROR then
raise ESocketError.Create(
FHeader + 'Ошибка при привязывании сокета к событию: ' + GetErrorString);
FSendBufSection:= TCriticalSection.Create;
// Объект этой нити не должен удаляться сам
FreeOnTerminate:= False;
inherited Create(False);
end;
destructor TClientThread.Destroy;
begin
FSendBufSection.Free;
WSACloseEvent(FEvents[0]);
WSACloseEvent(FEvents[1]);
WSACloseEvent(FEvents[2]);
inherited;
end;
// Функция добавляет строку в буфер для отправки
procedure TClientThread.SendString(const S: string);
begin
FSendBufSection.Enter;
try
FSendBuf:= FSendBuf + S + #0;
finally
FSendBufSection.Leave;
end;
LogMessage('Сообщение "' + S + '" поставлено в очередь для отправки');
// Взводим событие, которое говорит, что нужно отправлять данные
WSASetEvent(FEvents[1]);
end;
// Отправка всех данных, накопленных в буфере
// Функция возвращает False, если произошла ошибка,
// и True, если все в порядке
function TClientThread.DoSendBuf: Boolean;
var
SendRes: Integer;
begin
FSendBufSection.Enter;
try
// Если отправлять нечего, выходим
if FSendBuf = '' then
begin
Result:= True;
Exit;
end;
// Пытаемся отправить все, что есть в буфере
SendRes:= send(FSocket, FSendBuf[1], Length(FSendBuf), 0);
if SendRes > 0 then
begin
// Удаляем из буфера ту часть, которая отправилась клиенту
Delete(FSendBuf, 1, SendRes);
Result:= True;
end
else
begin
Result:= WSAGetLastError = WSAEWOULDBLOCK;
if not Result then
LogMessage('Ошибка при отправке данных: ' + GetErrorString);
end;
finally
FSendBufSection.Leave;
end;
end;
procedure TClientThread.Execute;
const
// размер буфера для приема сообщении
RecvBufSize = 4096;
var
// Буфер для приема сообщений
RecvBuf: array[0..RecvBufSize — 1] of Byte;
RecvRes: Integer;
NetEvents: TWSANetworkEvents;
// Полученная строка
Str: string;
// Длина полученной строки
StrLen: Integer;
// Если ReadLength = True, идет чтение длины строки,
// если False — самой строки
ReadLength: Boolean;
// Смещение от начала приемника
Offset: Integer;
// Число байтов, оставшихся при получении длины строки или самой строки
BytesLeft: Integer;
Р: Integer;
I: Integer;
LoopExit: Boolean;
WaitRes: Cardinal;
begin
LogMessage('Соединение установлено');
ReadLength:= True;
Offset:= 0;
BytesLeft:= SizeOf(Integer);
repeat
WaitRes:= WSAWaitForMultipleEvents(3, @FEvents, False, WSA_INFINITE, False);
case WaitRes of
WSA_WAIT_EVENT_0: begin
// Закрываем соединение с клиентом и останавливаем нить
LogMessage('Получен сигнал об остановке нити');
shutdown(FSocket, SD_BOTH);
Break;
end;
WSA_WAIT_EVENT_0 + 1:
begin
// Сбрасываем событие и отправляем данные
WSAResetEvent(FEvents[1]);
if not DoSendBuf then Break;
end;
WSA_WAIT_EVENT_0 + 2: begin
// Произошло событие, связанное с сокетом.
// Проверяем, какое именно, и заодно сбрасываем его
if WSAEnumNetworkEvents(FSocket, FEvents[2], NetEvents) = SOCKET_ERROR then
begin
LogMessage('Ошибка при получении списка событий: ' + GetErrorString);
Break;
end;
if NetEvents.lNetworkEvents and FD_READ <> 0 then
begin
if NetEvents.iErrorCode[FD_READ_BIT] <> 0 then
begin
LogMessage('Ошибка в событии FD_READ: ' +
GetErrorString(NetEvents.iErrorCode[FD_READ_BIT]));
Break;
end;
// В буфере сокета есть данные.
// Копируем данные из буфера сокета в свой буфер RecvBuf
RecvRes:= recv(FSocket, RecvBuf, SizeOf(RecvBuf), 0);
if RecvRes > 0 then
begin
P:= 0;
// Эта переменная нужна потому, что здесь появляется
// вложенный цикл, при возникновении ошибки в котором нужно
// выйти и из внешнего цикла тоже. Так как в Delphi нет
// конструкции типа Break(2) в Аде, приходится прибегать
// к таким способам: если нужен выход из внешнего цикла,
// во внутреннем цикле выполняется LoopExit:= True,
// а после выполнения внутреннего цикла проверяется
// значение этой переменной и при необходимости выполняется
// выход и из главного цикла.
LoopExit:= False;
// В этом цикле мы извлекаем данные из буфера
// и раскидываем их по приёмникам — Str и StrLen.
while Р < RecvRes do
begin
// Определяем, сколько байтов нам хотелось бы скопировать
L:= BytesLeft;
// Если в буфере нет такого количества,
// довольствуемся тем, что есть
if Р + L > RecvRes then L:= RecvRes — P;
// Копируем в соответствующий приемник
if ReadLength then
Move(RecvBuf[P], (PChar(@StrLen) + Offset)^, L)
else Move(RecvBuf[P], Str(Offset + 1), L);
Dec(BytesLeft, L);
// Если прочитали все, что хотели,
// переходим к следующему
if BytesLeft = 0 then
begin
ReadLength:= not ReadLength;
Offset:= 0;
// Если закончено чтение строки, нужно вывести ее
if ReadLength then
begin
LogMessage('Получена строка: ' + Str);
BytesLeft:= SizeOf(Integer);
// Формируем ответ и записываем его в буфер
Str:=
AnsiUpperCase(StringReplace(Str, #0, '#0',
[rfReplaceAll])) + '(AsyncEvent server)';
SendString(Str);
Str:= '';
end
else
begin
if StrLen <= 0 then
begin
LogMessage('Неверная длина строки от клиента: ' +
IntToStr(StrLen));
LoopExit:= True;
Break;
end;
BytesLeft:= StrLen;
SetLength(Str, StrLen);
end;
end
else Inc(Offset, L);
Inc(P, L);
end;
// Проверяем, был ли аварийный выход из внутреннего цикла,
// и если был, выходим и из внешнего, завершая работу
// с клиентом
if LoopExit then Break;
end
else if RecvRes = 0 then
begin
LogMessage('Клиент закрыл соединение ');
Break;
end
else
begin
if WSAGetLastError <> WSAEWOULDBLOCK then
begin
LogMessage('Ошибка при получении данных от клиента: ' +
GetErrorString);
end;
end;
end;
// Сокет готов к передаче данных
if NetEvents.lNetworkEvents and FD_WRITE <> 0 then
begin
if NetEvents.iErrorCode[FD_WRITE_BIT] <> 0 then
begin
LogMessage('Ошибка в событии FD_WRITE: ' +
GetErrorString(NetEvents.iErrorCode[FD_WRITE_BIT)));
Break;
end;
// Отправляем то, что лежит в буфере
if not DoSendBuf then Break;
end;
if NetEvents.lNetworkEvents and FD_CLOSE <> 0 then
begin
// Клиент закрыл соединение
if NetEvents.iErrorCode[FD_CLOSE_BIT] <> 0 then
begin
LogMessage('Ошибка в событии FD_CLOSE: ' +
GetErrorString(NetEvents.iErrorCode[FD_CLOSE_BIT]));
Break;
end;
LogMessage('Клиент закрыл соединение');
shutdown(FSocket, SD_BOTH);
Break;
end;
end;
WSA_WAIT_FAILED: begin
LogMessage('Ошибка при ожидании сообщения: ' + GetErrorString);
Break;
end;
else begin
LogMessage(
'Внутренняя ошибка сервера — неверный результат ожидания ' +
IntToStr(WaitRes));
Break;
end;
end;
until False;
closesocket(FSocket);
LogMessage('Нить остановлена');
end;
// Функция возвращает True, если нить завершилась
function TClientThread.GetFinished: Boolean;
begin
// Ждем окончания работы нити с нулевым тайм-аутом.
// Если нить завершена, вернется WAIT_OBJECT_0.
// Если еще работает, вернется WAIT_TIMEOUT.
Result:= WaitForSingleObject(Handle, 0) = WAIT_OBJECT_0;
end;
// Метод для остановки нити извне.
// Взводим соответствующее событие, а остальное сделаем
// при обработке события
procedure TClientThread.StopThread;
WSASetEvent(FEvents[0]);
end;
Модуль WinSock2_Events, появившийся в списке uses, содержит объявления констант, типов и функций из WinSock 2, которые понадобятся в программе. Модуль ShutdownConst содержит объявления констант для функции shutdown, которые отсутствуют в модуле WinSock Delphi 5 и более ранних версиях — этот модуль нам понадобился, чтобы программу можно было откомпилировать в Delphi 5.
Нить использует три события, дескрипторы которых хранятся в массиве FEvents. Событие FEvents[0] служит для уведомления нити о том, что необходимо завершиться, FEvents[1] — для уведомления о том, что нужно оправить данные, FEvents[2] связывается с событиями на сокете. Такой порядок выбран не случайно. Если взведено несколько событий, функция WSAWaitForMultipleEvents вернет результат, соответствующий событию с самым младшим из взведенных событий индексом. Соответственно, чем ближе к началу массива, тем более высокий приоритет у события. Событие, связанное с сокетом, имеет наинизший приоритет для того, чтобы повысить устойчивость сервера к DoS-атакам. Если приоритет этого события был бы выше, чем события остановки нити, то в случае закидывания сервера огромным количеством сообщений от клиента, событие FD_READ было бы всегда взведено, и сервер все время тратил бы на обработку этого события, игнорируя сигнал об остановке нити. Соответственно, сигнал об остановке должен иметь самый высокий приоритет, чтобы остановке нити ничего не могло помешать. Тем, как отправляются сообщения, сервер управляет сам. поэтому не приходится ожидать проблем, связанных с тратой излишних ресурсов на обработку сигнала отправки. Соответственно, этому событию присваивается приоритет, промежуточный между событием остановки нити и событием сокета.
Так как клиент по новому протоколу перед отправкой сообщения не обязан ждать, пока сервер ответит на предыдущее, возможны ситуации, когда ответ на следующее сообщение сервер должен готовить уже тогда, когда предыдущее еще не отправлено. Кроме того, сервер может отправить сообщение по собственной инициативе, и этот момент тоже может наступить тогда, когда предыдущее сообщение еще не отправлено. Таким образом, мы вынуждены формировать очередь сообщений в том или ином виде. Так как протокол TCP, с одной стороны, может объединять несколько пакетов в один, а с другой, не обязан отправлять отдельную строку за один раз, проще всего не делать очередь из отдельных строк, а заранее объединять их в одном буфере и затем пытаться отправить все содержимое буфера. Таким буфером в нашем случае является поле FSendBuf, метод SendString добавляет строку в этот буфер, a DoSendBuf отправляет данные из этого буфера. Если все данные отправить за один раз не удалось, отправленные данные удаляются из буфера, а оставшиеся будут отправлены при следующем вызове SendBuf. Все операции с буфером FSendBuf выполняются внутри критической секции, т. к. функция SendString может вызываться из других нитей. К каждой строке добавляется символ #0, который, согласно протоколу, является для клиента разделителем строк в потоке.
Сигналом к отправке данных является событие FEvents[1]. Метод SendString, помещая данные в буфер, взводит это событие. Если все содержимое буфера за один раз отправить не удастся, то через некоторое время возникнет событие FD_WRITE, означающее готовность сокета к приему новых данных. Это событие привязано у нас к FEvents[2], поэтому при наступлении FEvents[2] тоже возможна отправка данных.
Для приема данных здесь также используется буфер. Прямой необходимости в этом нет — можно было, как и раньше, помещать данные непосредственно в переменную, хранящую длину строки, а затем и в саму строку. Сделано это в учебных целях, чтобы показать, как можно работать с подобным буфером. Буфер имеет фиксированный размер. Сначала мы читаем из сокета в этот буфер столько, сколько сможем, а потом начинаем разбирать полученное точно так же, как и раньше, копируя данные то в целочисленную, то в строковую переменную. Когда строковая переменная полностью заполняется, строка считается принятой, для пользователя выводится ответ на нее, а в буфер для отправки добавляется ответная строка. Достоинством такого способа является то, что, с одной стороны, за время обработки одного события сервер может прочитать несколько запросов от клиента (если буфер достаточно велик), но, с другой стороны, это не приводит к зацикливанию, если сообщения поступают непрерывно. Другими словами, разработчик здесь сам определяет, какой максимальный объем данных можно получить от сокета за один раз. Иногда это бывает полезно.
Теперь рассмотрим нить, обслуживающую слушающий сокет. Код этой нити приведен в листинге 2.64.
Листинг 2.64. Код нити, обслуживающей слушающий сокет
unit ListenThread;
{
Нить, следящая за подключением клиента к слушающему сокету.
При обнаружении подключения она создает новую нить для работы с подключившимся клиентом, а сама продолжает обслуживать "слушающий" сокет.
}
interface
uses
SysUtils, Classes, WinSock, WinSock2_Events;
type
TListenThread = class(TThread)
private
// Сообщение, которое нужно добавить в лог.
// Хранится в отдельном поле, т. к. метод, вызывающийся
// через Synchronize, не может иметь параметров.
FMessage: string;
// Сокет, находящийся в режиме прослушивания
FServerSocket: TSocket;
// События нити
// FEvents[0] используется для остановки нити
// FEvents[1] связывается с событием FD_ACCEPT
FEvents: array[0..1] of TWSAEvent;
// Список нитей клиентов
FClientThreads: TList;
// Если True, сервер посылает клиенту сообщения
// по собственной инициативе
FServerMsg: Boolean;
// Вспомогательный метод для вызова через Synchronize
procedure DoLogMessage;
protected
procedure Execute; override;
// Вывод сообщения в лог главной формы
procedure LogMessage(const Msg: string);
public
constructor Create(ServerSocket: TSocket; ServerMsg: Boolean);
destructor Destroy; override;
// Вызывается извне для остановки сервера
procedure StopServer;
end;
implementation
uses
MainServerUnit, ClientThread;
{ TListenThread }
// "Слушающий" сокет создается в главной нити,
// а сюда передается через параметр конструктора
constructor TListenThread.Create(ServerSocket: TSocket; ServerMsg: Boolean);
begin
FServerSocket:= ServerSocket;
FServerMsg:= ServerMsg;
// Создаем события
FEvents[0]:= WSACreateEvent;
if FEvents[0] = WSA_INVALID_EVENT then
raise ESocketError.Create(
'Ошибка при создании события для сервера:' + GetErrorString);
FEvents[1]:= WSACreateEvent;
if FEvents[1] = WSA_INVALID_EVENT then
raise ESocketError.Create(
'Ошибка при создании события для сервера: ' + GetErrorString);
if WSAEventSelect(FServerSocket, FEvents[1], FD_ACCEPT) = SOCKET_ERROR then
raise ESocketError.Create(
'Ошибка при привязывании серверного сокета к событию: ' + GetErrorString);
FClientThreads:= TList.Create;
inherited Create(False);
end;
destructor TListenThread.Destroy;
begin
// Убираем за собой
FClientThreads.Free;
WSACloseEvent(FEvents[0]);
WSACloseEvent(FEvents[1]);
inherited;
end;
procedure TListenThread.Execute;
var
// Сокет, созданный для общения с подключившимся клиентом
ClientSocket: TSocket;
// Адрес подключившегося клиента
ClientAddr: TSockAddr;
ClientAddrLen: Integer;
NetEvents: TWSANetworkEvents;
I: Integer;
WaitRes: Cardinal;
begin
LogMessage('Сервер начал работу');
// Начинаем бесконечный цикл
repeat
// Ожидание события с 15-секундным тайм-аутом
WaitRes:=
WSAWaitForMultipleEvents(2, @FEvents, False, 15000, False);
case WaitRes of
WSA_WAIT_EVENT_0:
// Событие FEvents[0] взведено — это означает, что
// сервер должен остановиться.
begin
LogMessage('Сервер получил сигнал завершения работы');
// Просто выходим из цикла, остальное сделает код после цикла
Break;
end;
WSA_WAIT_EVENT_0 + 1:
// Событие FEvents[1] взведено.
// Это должно означать наступление события FD_ACCEPT.
begin
// Проверяем, почему событие взведено,
// и заодно сбрасываем его
if WSAEnumNetworkEvents(FServerSocket, FEvents[1], NetEvents) = SOCKET_ERROR then
begin
LogMessage('Ошибка при получении списка событий: ' +
GetErrorString);
Break;
end;
// Защита от "тупой" ошибки — проверка того,
// что наступило нужное событие
if NetEvents.lNetworkEvents and FD_ACCEPT = 0 then
begin
LogMessage(
'Внутренняя ошибка сервера — неизвестное событие');
Break;
end;
// Проверка, не было ли ошибок
if NetEvents.iErrorCode[FD_ACCEPT_BIT] <> 0 then
begin
LogMessage('Ошибка при подключении клиента: ' +
GetErrorString(NetEvents.iErrorCode[FD_ACCEPT_BIT]));
Break;
end;
ClientAddrLen:= SizeOf(ClientAddr);
// Проверяем наличие подключения
ClientSocket:=
accept(FServerSocket, @ClientAddr, @ClientAddrLen);
if ClientSocket = INVALID_SOCKET then
begin
// Ошибка в функции accept возникает только тогда, когда
// происходит нечто экстраординарное. Продолжать работу
// в этом случае бессмысленно. Единственное возможное
// в нашем случае исключение — ошибка WSAEWOULDBLOCK,
// которая может возникнуть, если срабатывание события
// было ложным, и подключение от клиента отсутствует
if WSAGetLastError <> WSAEWOULDBLOCK then
begin
LogMessage('Ошибка при подключении клиента: ' +
GetErrorString);
Break;
end;
end;
// Создаем новую нить для обслуживания подключившегося клиента
// и передаем ей сокет, созданный для взаимодействия с ним.
// Указатель на нить сохраняем в списке
FClientThreads.Add(
TClientThread.Create(ClientSocket, ClientAddr));
end;
WSA_WAIT_TIMEOUT:
// Ожидание завершено по тайм-ауту
begin
// Проверяем, есть ли клиентские нити, завершившие работу.
// Если есть такие нити, удаляем их из списка
// и освобождаем объекты
for I:= FClientThreads.Count -1 downto 0 do
if TClientThread(FClientThreads[I]).Finished then
begin
TClientThread(FClientThreads[I]).Free;
FClientThreads.Delete(I);
end;
// Если разрешены сообщения от сервера, отправляем
// всем клиентам сообщение с текущим временем
if FServerMsg then
for I:= 0 to FClientThreads.Count — 1 do
TClientThread(FClientThreads[I]).SendString(
'Время на сервере ' + TimeToStr(Now));
end;
WSA_WAIT_FAILED:
// При ожидании возникла ошибка. Это может означать
// только какой-то серьезный сбой в библиотеке сокетов.
begin
LogMessage('Ошибка при ожидании события сервера: ' +
GetErrorString);
Break;
end;
else
// Неожиданный результат при ожидании
begin
LogMessage(
'Внутренняя ошибка сервера — неожиданный результат ожидания '
+ IntToStr(WaitRes));
Break;
end;
end;
until False;
// Останавливаем и уничтожаем все нити клиентов
for I:= 0 to FClientThreads.Count — 1 do
begin
TClientThread(FClientThreads[I]).StopThread;
TClientThread(FClientThreads[I]).WaitFor;
TClientThread(FClientThreads[I]).Free;
end;
closesocket(FServerSocket);
LogMessage('Сервер завершил работу');
Synchronize(ServerForm.OnStopServer);
end;
// Завершение работы сервера. Просто взводим соответствующее
// событие, а остальное делает код в методе Execute.
procedure TListenThread.StopServer;
begin
WSASetEvent(FEvents[0));
end;
end.
Нить TListenThread реализует сразу несколько функций. Во-первых, она обслуживает подключение клиентов и создает нити для их обслуживания. Во-вторых, уничтожает объекты завершившихся нитей. В-третьих, она с определённой периодичностью ставит в очередь на отправку всем клиентам сообщение с текущим временем сервера. И в-четвертых, управляет уничтожением клиентских нитей при завершении работы сервера.
Здесь следует пояснить, почему выбран такой способ управления временем жизни объектов клиентских нитей. Очевидно, что нужно иметь список всех нитей, чтобы обеспечить возможность останавливать их и ставить в очередь сообщения для отправки клиентам (этот список реализован переменной FClientThreads). Если бы объект TClientThread автоматически удалялся при завершении работы нити, в его деструкторе пришлось бы предусмотреть и удаление ссылки на объект из списка, а это значит, что к списку пришлось бы обращаться из разных нитей. Соответственно, потребовалось бы синхронизировать обращение к списку, и здесь мы бы столкнулись с одной неприятной проблемой. Когда нить TListenThread получает команду завершиться, она должна завершить все клиентские нити. Для этого она должна использовать их список для отправки сигнала и ожидания их завершения. И получилась бы взаимная блокировка, потому что нить TListenThread ждала бы завершения клиентских нитей, а они не могли бы завершиться, потому что им требовался бы список, захваченный нитью TListenThread. Избежать этого можно с помощью асинхронных сообщений, но в нашем случае реализация этого механизма затруднительна (хотя и возможна). Для простоты был выбран другой вариант: клиентские нити сами свои объекты не удаляют, а к списку имеет доступ только нить TListenThread, которая время от времени проходит по по списку и удаляет объекты всех завершившихся нитей. В этом случае клиентские нити не используют синхронизацию при завершении, и нить TListenThread может дожидаться их.
Нить TListenThread использует два события: FEvents[0] для получения сигнала о необходимости закрытия и FEvents[1] для получения уведомлений о возникновении события FD_ACCEPT на слушающем сокете (т. е. о подключении клиента). Порядок следования событий к массиве определяется теми же соображениями, что и в случае клиентской нити: сигнал остановки нити должен иметь более высокий приоритет. чтобы в случае DoS-атаки нить могла быть остановлена.
И поиск завершившихся нитей, и отправка сообщений с текущим временем клиентам осуществляется в том случае, если при ожидании события произошёл тайм-аут (который в нашем случае равен 15 c). Подключение клиента — событие достаточно редкое, поэтому такое решение выгладит вполне оправданным. Для тех экзотических случаев, когда клиенты часто подключаются и отключаются, можно предусмотреть еще одно событие у нити TListenThread, при наступлении которого она будет проверять список клиентов. Клиентская нить при своем завершении будет взводить это событие. Что же касается отправки сообщений клиентам, то в обработчик тайм-аута этот код помещён в демонстрационных целях. В реальной программе инициировать отправку сообщений клиентам будет, скорее всего, другой код, например, код главной нити по команде пользователя.
Несмотря на изменение протокола, новый сервер был бы вполне совместим со старым клиентом SimpleClient (см. разд. 2.1.11), если бы не отправлял сообщения по своей инициативе. Действительно, прочие изменения в протоколе разрешают клиенту отправлять новые сообщения до получения ответа сервера, но не обязывают его делать это. В класс TClientThread добавлено логическое поле FServerMsg. Если оно равно False, то сервер не посылает клиентам сообщений по собственной инициативе, т. е. работает в режиме совместимости со старым клиентом. Поле FServerMsg инициализируется в соответствии с параметром, переданным в конструктор, т. е. в соответствии с состоянием галочки Сообщения от сервера, расположенной на главной форме. Если перед запуском сервера она снята, сервер не будет сам посылать сообщения, и старый клиент сможет обмениваться данными с ним.
Запуск сервера практически не отличается от запуска сервера MultithreadedServer (см. листинг 2.19), только теперь объект, созданный конструктором, запоминается классом главной формы, чтобы потом можно было сервер остановить. Остановка осуществляется методом StopServer (листинг 2.65).
Листинг 2.65. Метод StopServer
// Остановка сервера
procedure TServerForm.StopServer;
begin
// Запрещаем кнопку, чтобы пользователь не мог нажать ее
// еще раз, пока сервер не остановится.
BtnStopServer.Enabled:= False;
// Ожидаем завершения слушавшей нити. Так как вывод сообщений
// эта нить осуществляет через Synchronize, выполняемый главной
// нитью в петле сообщений, вызов метода WaitFor мог бы привести
// к взаимной блокировке: главная нить ждала бы, когда завершится
// нить TListenThread, а та, в свою очередь — когда главная нить
// выполнит Synchronize. Чтобы этого не происходило, организуется
// ожидание с локальной петлей сообщений.
if Assigned(FListenThread) then
begin
FListenThread.StopServer;
while Assigned(FListenThread) do
begin
Application.ProcessMessages;
Sleep(10);
end;
end;
end;
Данный метод вызывается в обработчике нажатия кнопки Остановить и при завершении приложения. Сервер можно многократно останавливать и запуска вновь, не завершая приложение.
Чтобы увидеть все возможности сервера, потребуется новый клиент. На компакт-диске он называется EventSelectClient, но "EventSelect" в данном случае означает только то, что клиент является парным к серверу EventSelectServer. Сам клиент функцию WSAEventSelect не использует, поскольку она неудобна, когда нужно работать только с одним сокетом. Поэтому клиент работает в асинхронном режиме, основанном на сообщениях, т. е. посредством функции WSAAsyncSelect.
Клиент может получать от сервера сообщения двух видов: те. которые сервер посылает в ответ на запросы клиента, и те, которые он посылает по собственной инициативе. Но различить эти сообщения клиент не может: например, если клиент отправляет запрос серверу, а потом получает от него сообщение, он не может определить, то ли это сервер ответил на его запрос, то ли именно в этот момент сервер сам отправил клиенту свое сообщение. Соответственно, сообщения обоих типов читает один и тот же код.
Примечание
В принципе, протокол мог бы быть определен таким образом, что ответы на запросы клиента и сообщения, посылаемые сервером по собственной инициативе, имели бы разный формат, по которому их можно было бы различить и читать по-разному. Но даже при этом форматы нельзя различить, пока сообщение не будет прочитано хотя бы частично, так что начало чтения будет выполняться единообразно в любом случае.
Подключение клиента к серверу выполняется точно так же, как в листинге 2.16, за исключением того, что после выполнения функции connect сокет переводится в асинхронный режим, и его события FD_READ и FD_CLOSE связываются с сообщением WM_SOCKETMESSAGE. Обработчик этого сообщения приведен в листинге 2.66.
Листинг 2.66. Получение данных клиентом
procedure TESClientForm.WMSocketMessage(var Msg: TWMSocketMessage);
const
// Размер буфера для получения данных
RecvBufSize = 4096;
var
// Буфер для получения данных
RecvBuf: array[0..RecvBufSize — 1] of Byte;
RecvRes: Integer;
P: Integer;
begin
// Защита от "тупой" ошибки
if Msg.Socket <> FSocket then
begin
MessageDlg('Внутренняя ошибка программы — неверный сокет',
mtError, [mbOK], 0);
Exit;
end;
if Msg.SockError <> 0 then
begin
MessageDlg('Ошибка при взаимодействии с сервером'#13#10 +
GetErrorString(Msg.SockError), mtError, [mbOK], 0);
OnDisconnect;
Exit;
end;
case Msg.SockEvent of
FD_READ:
// Получено сообщение от сервера
begin
// Читаем столько, сколько можем
RecvRes:= recv(FSocket, RecvBuf, RecvBufSize, 0);
if RecvRes > 0 then
begin
// Увеличиваем строку на размер прочитанных данных
P:= Length(FRecvStr);
SetLength(FRecvStr, P + RecvRes);
// Копируем в строку полученные данные
Move(RecvBuf, FRecvStr[Р + 1], RecvRes);
// В строке может оказаться несколько строк от сервера,
// причем последняя может прийти не целиком.
// Ищем в строке символы #0, которые, согласно протоколу,
// являются разделителями строк.
P:= Pos(#0, FRecvStr));
while P > 0 do
begin
AddMessageToRecvMemo('Сообщение от сервера: ' +
Copy(FRecvStr, 1, P — 1));
// Удаляем из строкового буфера выведенную строку
Delete(FRecvStr, 1, P);
P:= Pos(#0, FRecvStr);
end;
end
else if RecvRes = 0 then
begin
MessageDlg('Сервер закрыл соединение'#13#10 +
GetErrorString, mtError, [mbOK], 0);
OnDisconnect;
end
else
begin
if WSAGetLastError <> WSAEWOULDBLOCK then
begin
MessageDlg('Ошибка при получении данных от клиента'#13#10 +
GetErrorString, mtError, [mbOK], 0);
OnDisconnect;
end;
end;
end;
FD_CLOSE: begin
MessageDlg('Сервер закрыл соединение', mtError, [mbOK], 0);
shutdown(FSocket, SD_BOTH);
OnDisconnect;
end;
else begin
MessageDlg('Внутренняя ошибка программы — неизвестное событие ' +
IntToStr(Msg.SockEvent), mtError, [mbOK], 0);
OnDisconnect;
end;
end;
end;
Здесь мы используем новый способ чтения данных. Он во многом похож на тот, который применен в сервере. Функция recv вызывается один раз за один вызов обработчика значений и передаст данные в буфер фиксированного размера RecvBuf. Затем в буфере ищутся границы отдельных строк (символы #0), строки, полученные целиком, выводятся. Если строка получена частично (а такое может случиться не только из-за того, что она передана по частям, но и из-за того, что в буфере просто не хватило место для приема ее целиком), её начало следует сохранить в отдельном буфере, чтобы добавить к тому, что будет прочитано при следующем событии FD_READ. Этот буфер реализуется полем FRecvStr типа string. После чтения к содержимому этой строки добавляется содержимое буфера RecvBuf, а затем из строки выделяются все подстроки, заканчивающиеся на #0. То, что остается в строке FRecvStr после этого, — это начало строки, прочитанной частично. Оно будет учтено при обработке следующего события FD_READ.
Примечание
Описанный алгоритм разбора буфера прост, но неэффективен с точки зрения нагрузки на процессор и использования динамической памяти, особенно в тех случаях, когда в буфере RecvBuf оказывается сразу несколько строк. Это связано с тем, что при добавлении содержимого RecvBuf к FRecvStr и последующем поочередном удалении строк из FRecvStr происходит многократное перераспределение памяти, выделенной для строки. Алгоритм можно оптимизировать: все строки, которые поместились в RecvBuf целиком, выделять непосредственно из этого буфера, не помещая в FRecvStr , а помещать туда только то, что действительно нужно сохранить между обработкой разных событий FD_READ . Реализацию такого алгоритма рекомендуем выполнить в качестве самостоятельного упражнения.
При отправке данных вероятность того, что функция send не сможет быть выполнена сразу, достаточно мала. Кроме того, как мы уже говорили, блокировка клиента при отправке данных часто бывает вполне приемлема из-за редкости и непродолжительности. Таким образом, блокирующий режим из-за своей простоты наиболее удобен при отправке данных серверу клиентом. Но мы не можем перевести сокет, работающий в асинхронном режиме, в блокирующий режим на время отправки, зато можем этот режим имитировать. Занимается этим метод SendString (листинг 2.67).
Листинг 2.67. Метод SendString , имитирующий блокирующим режим отправки
// Отправка строки серверу. Функция имитирует блокирующий
// режим работы сокета: если за один раз не удается отправить
// данные, попытка отправить их продолжается до тех пор,
// пока все данные не будут отправлены или пока не возникнет ошибка.
procedure TESClientForm.SendString(const S: string);
var
SendRes: Integer;
// Буфер, куда помещается отправляемое сообщение
SendBuf: array of Byte;
// Сколько байтов уже отправлено
BytesSent: Integer;
begin
if Length(S) > 0 then
begin
// Отправляемое сообщение состоит из длины строки и самой строки.
// Выделяем для буфера память, достаточную для хранения
// и того и другого.
SetLength(SendBuf, SizeOf(Integer) + Length(S));
// Копируем в буфер длину строки
PInteger(@SendBuf[0])^:= Length(S);
// А затем — саму строку
Move(S[1], SendBuf[SizeOf(Integer)], Length(S));
BytesSent:= 0;
// повторяем попытку отправить до тех пор, пока все содержимое
// буфера не будет отправлено серверу.
while BytesSent < Length(SendBuf) do
begin
SendRes:=
send(FSocket, SendBuf[BytesSent], Length(SendBuf) — BytesSent, 0);
if SendRes > 0 then Inc(BytesSent, SendRes)
else if WSAGetLastError = WSAEWOULDBLOCK then Sleep(10)
else
begin
MessageDlg('Ошибка при отправке данных серверу'#13#10 +
GetErrorString, mtError, [mbOK], 0);
OnDisconnect;
Exit;
end;
end;
end;
end;
Имитация блокирующего режима осуществляется очень просто: если сообщение не удалось отправить сразу, после небольшой паузы производится попытка отправить то, что ещё не отправлено, и так до тех пор, пока не будет отправлено все или пока не возникнет ошибка. В программе SimpleClient мы отправляли длину строки и саму строку разными вызовами send. Теперь, из-за того, что функция send может отправить только часть переданных ей данных, это становится неудобным из-за громоздкости многочисленных проверок. Поэтому мы создаем один буфер, куда заносим и длину строки, и саму строку, и затем передаем его как единое целое.
Примечание
Далее мы познакомимся с функцией WSASend , которая позволяет отправлять данные, находящиеся не в одном, а в нескольких разных местах. Если бы мы использовали ее, можно было бы не объединять самостоятельно длину строки и саму строку в специальном буфере, а просто передать два указателя на длину и на строку.
Чтобы продемонстрировать возможности сервера по приему нескольких слившихся запросов, клиент должен отправлять ему несколько строк сразу, поэтому на главной форме клиента мы заменяем однострочное поле ввода на многострочное (т. е. TEdit на TMemo). При нажатии кнопки Отправить клиент отправляет серверу все непустые строки из этого поля ввода.
Других существенных отличий от SimpleClient программа EventSelectClient не имеет. Получившийся пример работает не только с сервером EventSelectServer, но и с любым сервером, написанным нами ранее. Действительно, ни один из этих серверов не требует, чтобы на момент получения запроса от клиента в буфере сокета ничего не было, кроме этого запроса. Поэтому то, что EventSelectClient может отправлять несколько сообщений сразу, не помешает им работать: просто, в отличие от EventSelectServer, они будут обрабатывать эти запросы строго по одному, а не получать из сокета сразу несколько штук.