Windows Sockets, IOCP и Delphi

Пролог

Недавно я столкнулся с необходимостью эффективной работы с сокетами в Windows приложении. Задача типичная для нагруженного сервера. Нетипичным тут будет казаться только язык реализации - Delphi.
Я хочу описать способ массовой асинхронной работы с большим количеством сокетов с использованием I/O Completion Ports. Microsoft в своей литературе рекомендует использовать именно эту технологию. Я думаю, многие с ней знакомы, но на всякий случай укажу ссылку на MSDN. Суть технологии в том, что система организует высокоэффективную очередь событий, а программа обрабатывает её из тред-пула, размер которого подобран по количеству вычислительных ядер. Данный подход имеет преимущества при большом количестве одновременно производимых асинхронных операций ввода вывода для разных конечных точек. Готовый исходник можно (лучше) сразу гляуть здесь. Не всё идеально, но для эксперементов сойдёт.

Roadmap

Я, в некотором смысле, буду придерживаться идеологии Node.Js во всём, что касается организации объектов и операций ввода вывода. 
В случае с серверной частью понадобиться реализовать следующее:

• Прослушивание сокета. Принятием или отклонением новых соединений.
• Отслеживание сигнала закрытия клиентских сокетов.

Для клиента первый пункт этого списка не актуален, но необходимо реализовать асинхронное подключение к серверу. В обоих классах будет возможность одновременного и чтения и записи на одну конечную точку.
Все созданные экземпляры клиентских и серверных сокетов будут использовать одну общую очередь сообщений и один тред-пул. Это нужно для возможности использовать оба типа сокетов в одном приложении оптимальным образом.

Реализация

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

• Создание списков сокетов разных типов.
• Инициализация подсистемы сокетов.
• Создание очереди сообщений.
• Создание пула для обработки очереди.
• Создание событий для сокетов.
• Создание потоков отслеживающих сокетные события( например подключение нового клиента).

И так, секция инициализации содержит следующую процедуру, которая реализует список пункт за пунктом.

procedure Init;
var
  WSAData: TWsaData;
  i: Integer;
begin
  gClients := TProtoStore.Create;
  gListeners := TProtoStore.Create;
  gServerClients := TProtoStore.Create;
  if WSAStartup(MAKEWORD(2, 2), WSAData) <> 0 then
    raise IOCPClientException.Create(sErrorInit_WSAtartup);
  gIOCP := CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0, CPUCount * 2);
  if gIOCP = INVALID_HANDLE_VALUE then
    raise IOCPClientException.Create(sErrorInit_CreateIoCompletionPort);
  for i := 1 to CPUCount * 2 do
  begin
    SetLength(gWorkers, Length(gWorkers) + 1);
    gWorkers[Length(gWorkers) - 1] := TWorkerThread.Create();
  end;
  gListenerAcceptEvent := WSACreateEvent;
  if gListenerAcceptEvent = WSA_INVALID_EVENT then
    raise IOCPClientException.Create(sErrorInit_WSACreateEvent);
  gServerClientsCloseEvent := WSACreateEvent;
  if gServerClientsCloseEvent = WSA_INVALID_EVENT then
    raise IOCPClientException.Create(sErrorInit_WSACreateEvent);
  gClisentsConnectAndCloseEvents := WSACreateEvent;
  if gClisentsConnectAndCloseEvents = WSA_INVALID_EVENT then
    raise IOCPClientException.Create(sErrorInit_WSACreateEvent);
  gClientSocketEventThread := TSocketEventThread.Create
    (gClisentsConnectAndCloseEvents, gClients, ET_EVENT_SIGNALED);
  gClientSocketEventThread.Start;
  gServerClientsSocketEventThread := TSocketEventThread.Create
    (gServerClientsCloseEvent, gServerClients, ET_EVENT_SIGNALED);
  gServerClientsSocketEventThread.Start;
  gServerSocketEventThread := TSocketEventThread.Create(gListenerAcceptEvent,
    gListeners, ET_EVENT_SIGNALED);
  gServerSocketEventThread.Start;
end;

Функция CreateIoCompletionPort в данном случае выполняет создание специальной очереди сообщений.
Видно, что для отслеживания событий на сокетах с разным назначением используется один и тот же класс потока TSocketEventThread. Потоки этого класса выполняют процедуру, которая ожидает сокетные события, и сразу же ставят в очередь сообщения (для каждого сокета относящегося к типу, который облуживает этот поток) о том что произошло какое-то событие. 

procedure TSocketEventThread.WaitForClientsEvents;
var
  WaitResult: DWORD;
const
  TimeOut: DWORD = 100;
begin
  WaitResult := WSAWaitForMultipleEvents(1, @fEvent, FALSE, TimeOut, FALSE);
  if WaitResult = WSA_WAIT_FAILED then
    raise IOCPClientException.Create
      (sErrorWaitForClientsEvents_WSAWaitForMultipleEvents);
  if WaitResult = WSA_WAIT_EVENT_0 then
  begin
    if not WSAResetEvent(fEvent) then
      raise IOCPClientException.Create
        (sErrorWaitForClientsEvents_WSAResetEvent);
    fStore.Post(fKey);
  end;
end;

Тут метод fStore.Post(fKey); как раз и выполняет отправку сообщений в очередь.

procedure TProtoStore.Post(CompletionKey: DWORD);
var
  i: Integer;
begin
  fLock.Enter;
  try
    for i := 0 to Length(ProtoArray) - 1 do
    begin
      ProtoArray[i]._AddRef;
      if not PostQueuedCompletionStatus(gIOCP, 0, CompletionKey,
        POverlapped(ProtoArray[i])) then
      begin
        ProtoArray[i]._Release;
        raise IOCPClientException.Create(sErrorPost_PostQueuedCompletionStatus);
      end;
    end;
  finally
    fLock.Leave;
  end;
end;

Особое внимание заслуживает тут использование объектов с интерфейсами.
Метод _AddRef используется для того, чтобы обозначить тот факт, что объект "находится в очереди" и его не следует уничтожать. (Позже после обработки будет вызван _Release). Процедура PostQueuedCompletionStatus непосредственно выполняет постановку сообщения в очередь.
Пул обработает каждое сообщение в асинхронном режиме.
Для этого он выполняет следующую процедуру.

procedure TWorkerThread.ProcessIOCP;
var
  NumberOfBytes: DWORD;
  CompletionKey: NativeUInt;
  Overlapped: POverlapped;
  Proto: TIOCPSocketProto;
begin
  if not((not GetQueuedCompletionStatus(gIOCP, NumberOfBytes, CompletionKey,
    Overlapped, INFINITE)) and (Overlapped = nil)) then
  begin
    if CompletionKey = ET_EVENT_SIGNALED then
    begin
      Proto := TIOCPSocketProto(Overlapped);
      with Proto do
      begin
        IOCPProcessEventsProc();
        _Release;
      end
    end
    else if CompletionKey <> 0 then
    begin
      Proto := TIOCPSocketProto(CompletionKey);
      if Proto.IOCPProcessIOProc(NumberOfBytes, Overlapped) then
        Proto._Release;
    end;
  end
end;

Процедура GetQueuedCompletionStatus служит для получения сообщения из очереди. Далее определяется является ли это сообщение сообщением о завершенном вводе/выводе или это сообщение о произошедшем событии. Тут продемонстрированы два способа передать через очередь какую-то информацию, в данном случае это ссылка на конкретный экземпляр класса сокетов. 
Обработка ведётся унифицировано для всех типов сокетов, это достигнуто с помощью наследования от общего предка который содержит общие обработчики, допускается их переопределение. 
Рассмотрим механизм обработки сокетных событий.

procedure TIOCPSocketProto.IOCPProcessEventsProc();
var
  WSAEvents: TWsaNetworkEvents;
  AcceptedSocket: TSocket;
  RemoteAddress: string;
begin
  if fStateLock <> CLI_SOCKET_LOCK_CLOSED then
  begin
    fClosingLock.BeginRead;
    try
      if (fStateLock <> CLI_SOCKET_LOCK_CLOSED) then
        if WSAEnumNetworkEvents(fSocket, 0, WSAEvents) <> SOCKET_ERROR then
        begin
          if ((WSAEvents.lNetworkEvents and FD_CONNECT) <> 0) then
          begin
            if 0 = WSAEvents.iErrorCode[FD_CONNECT_BIT] then
              InterlockedExchange(fStateLock, CLI_SOCKET_LOCK_CONNECTED);
            CallOnConnect;
          end;
          if ((WSAEvents.lNetworkEvents and FD_CLOSE) <> 0) and
            (0 = WSAEvents.iErrorCode[FD_CLOSE_BIT]) then
            CallOnClose;
          if ((WSAEvents.lNetworkEvents and FD_ACCEPT) <> 0) and
            (0 = WSAEvents.iErrorCode[FD_ACCEPT_BIT]) then
          begin
            AcceptedSocket := DoAccept(RemoteAddress);
            if AcceptedSocket <> INVALID_SOCKET then
            begin
              fClientClass.Create(AcceptedSocket, fOnConnect, fOnClose,
                RemoteAddress).Prepare;
            end;
          end;
        end
    finally
      fClosingLock.EndRead;
    end;
  end;
end;

Здесь интересно применён класс TMultiReadExclusiveWriteSynchronizer. Он используется для предотвращения попытки закрыть сокет и уничтожить объект из другой нити пула (fClosingLock.BeginRead). Все операции с сокетом проходят как операции чтения для этого объекта синхронизации, кроме операции создания и операции закрытия сокета - они являются операциями записи и потому могут выполняться только при монопольном владении ресурсом.
Во всём же остальном работа с сокетами в данной процедуре совершенно обыкновенная.
Единственное что в этой процедуре стоит рассмотреть дополнительно - это подключение нового клиента к серверу, метод DoAccept. 

function TIOCPSocketProto.DoAccept(var RemoteAddress: string): TSocket;
var
  addr: TSockAddr;
  addrlen: Integer;
  dwCallbackData: NativeUInt;
  RemoteAddrLen: DWORD;
begin
  dwCallbackData := NativeUInt(self);
  addrlen := SizeOf(addr);
  Result := WSAAccept(fSocket, @addr, @addrlen, ServerAcceptCallBack,
    dwCallbackData);
  if Result <> INVALID_SOCKET then
  begin
    SetLength(RemoteAddress, 255);
    RemoteAddrLen := Length(RemoteAddress);
    if WSAAddressToString(addr, addrlen, nil, PChar(<hh user=RemoteAddress>[1]),
      RemoteAddrLen) = SOCKET_ERROR then
      raise IOCPClientException.Create(sErrorAccept_WSAAddressToString);
    SetLength(RemoteAddress, RemoteAddrLen - 1)
  end
end;

Здесь ключевым моментом является использование WSAAccept. Эта функция позволяет отклонять подключение клиентов таким образом, что клиент на самом деле получает событие FD_CONNECT.
Это предочтительный путь для организации так называемых чёрных списков.
Идём далее. Расмотрим организацию ввода вывода. Сделаем это на примере операции чтения.

procedure TIOCPSocketProto.Read(Length: DWORD;
  OnRead, OnReadProcess: TOnReadEvent);
var
  Bytes, Flags: DWORD;
  WsaBuf: TWsaBuf;
begin
  fClosingLock.BeginRead;
  try
    if fStateLock = CLI_SOCKET_LOCK_CONNECTED then
    begin
      if InterlockedCompareExchange(fReadLock, IO_PROCESS, IO_IDLE) = IO_IDLE
      then
      begin
        fOnRead := OnRead;
        fOnReadProcess := OnReadProcess;
        fReaded := 0;
        fReadBufLength := Length;
        fReadBuffer := nil;
        GetMem(fReadBuffer, Length);
        if fReadBuffer <> nil then
        begin
          Bytes := 0;
          FillChar(fOverlappedRead, SizeOf(fOverlappedRead), 0);
          WsaBuf.buf := fReadBuffer;
          WsaBuf.len := fReadBufLength;
          Flags := 0;
          Bytes := 0;
          _AddRef;
          if (WSARecv(fSocket, @WsaBuf, 1, Bytes, Flags, @fOverlappedRead, nil)
            = SOCKET_ERROR) and (WSAGetLastError <> WSA_IO_PENDING) then
          begin
            FreeMem(fReadBuffer, Length);
            InterlockedExchange(fReadLock, IO_IDLE);
            _Release;
            raise IOCPClientException.Create(sErrorRead_WSARecv);
          end;
        end
        else
          raise IOCPClientException.Create(sErrorRead_GetMem);
      end
      else
        raise IOCPClientException.Create(sErrorRead_InProcess);
    end
    else
      raise IOCPClientException.Create(sErrorRead_NotConnected);
  finally
    fClosingLock.EndRead;
  end;
end;

Здесь пришлось использовать интерлокед блокировку, т.к. она очень быстрая и удовлетворяет потребность в отсечении попытки повторного вызова опрации ввода/вывода. Память выделяется под буфер единажды в каждой операции. Далее вызывается чтение из сокета в асинхронном режиме. Объект также "помечается" с помощью AddRef, для невозможности его удаления во время нахождения в очереди. По завершении вычитывания пакета сообщения об этом автоматически выставляется в очередь.
Рассмотрим, что происходит при выборке из очереди сообщения о завершенном вводе/выводе.

function TIOCPSocketProto.IOCPProcessIOProc(NumberOfBytes: DWORD;
  Overlapped: POverlapped): Boolean;
var
  Bytes, Flags: DWORD;
  WsaBuf: TWsaBuf;
begin
  Result := FALSE;
  fClosingLock.BeginRead;
  try
    if Overlapped = @fOverlappedRead then
    begin
      if NumberOfBytes <> 0 then
      begin
        if fReadLock = IO_PROCESS then
        begin
          inc(fReaded, NumberOfBytes);
          if fReaded < fReadBufLength then
          begin
            CallOnReadProcess;
            WsaBuf.buf := fReadBuffer;
            inc(WsaBuf.buf, fReaded);
            WsaBuf.len := fReadBufLength;
            dec(WsaBuf.len, fReaded);
            Flags := 0;
            Bytes := 0;
            if (WSARecv(fSocket, @WsaBuf, 1, Bytes, Flags, @fOverlappedRead,
              nil) = SOCKET_ERROR) and (WSAGetLastError <> WSA_IO_PENDING) then
            begin
              CallOnRead;
              Result := True;
            end
          end
          else
          begin
            CallOnReadProcess;
            CallOnRead;
            Result := True;
          end;
        end
      end
      else
      begin
        CallOnRead;
        Result := True;
      end;
    end
    else if Overlapped = @fOverlappedWrite then
    begin
      if NumberOfBytes <> 0 then
      begin
        if fWriteLock = IO_PROCESS then
        begin
          inc(fWrited, NumberOfBytes);
          if fWrited < fWriteBufLength then
          begin
            CallOnWriteProcess;
            WsaBuf.buf := fWriteBuffer;
            inc(WsaBuf.buf, fWrited);
            WsaBuf.len := fWriteBufLength;
            dec(WsaBuf.len, fWrited);
            Flags := 0;
            Bytes := 0;
            if (WSASend(fSocket, @WsaBuf, 1, Bytes, Flags, @fOverlappedWrite,
              nil) = SOCKET_ERROR) and (WSAGetLastError <> WSA_IO_PENDING) then
            begin
              CallOnWrite;
              Result := True;
            end
          end
          else
          begin
            CallOnWriteProcess;
            CallOnWrite;
            Result := True;
          end;
        end
      end
      else
      begin
        CallOnWrite;
        Result := True;
      end;
    end
  finally
    fClosingLock.EndRead;
  end;
end;

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

constructor TIOCPClientSocket.Create(RemoteAddress: string;
  OnConnect, OnClose: TOnSimpleSocketEvenet);
var
  lRemoteAddress: TSockAddr;
  lRemoteAddressLength: Integer;
begin
  inherited Create();
  fStore := gClients;
  fOnConnect := OnConnect;
  fOnClose := OnClose;
  fStateLock := 0;
  fRemoteAddressStr := RemoteAddress;
  fSocket := WSASocket(AF_INET, SOCK_STREAM, 0, nil, 0, WSA_FLAG_OVERLAPPED);
  if fSocket = INVALID_SOCKET then
    raise IOCPClientException.Create(sErrorTIOCPClientSocket_WSASocket);
  if (WSAEventSelect(fSocket, gClisentsConnectAndCloseEvents,
    FD_CONNECT or FD_CLOSE) = SOCKET_ERROR) then
    raise IOCPClientException.Create(sErrorTIOCPClientSocket_WSAEventSelect);
  if CreateIoCompletionPort(fSocket, gIOCP, NativeUInt(self), 0) = 0 then
    raise IOCPClientException.Create
      (sErrorTIOCPClientSocket_CreateIoCompletionPort);
  fStateLock := CLI_SOCKET_LOCK_CREATED;
  fStore.Add(self);
  lRemoteAddressLength := SizeOf(lRemoteAddress);
  lRemoteAddress.sa_family := AF_INET;
  if WSAStringToAddress(PChar(@fRemoteAddressStr[1]), AF_INET, nil,
    lRemoteAddress, lRemoteAddressLength) = SOCKET_ERROR then
    raise IOCPClientException.Create
      (sErrorTIOCPClientSocket_WSAStringToAddress);
  if (WSAConnect(fSocket, lRemoteAddress, lRemoteAddressLength, nil, nil, nil,
    nil) = SOCKET_ERROR) and (WSAGetLastError <> WSAEWOULDBLOCK) then
    raise IOCPClientException.Create(sErrorTIOCPClientSocket_WSAConnect);
end; 

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

Эпилог

В данной статье кратко рассмотрены, на мой взгляд, удачные приёмы создания классов для событийного программирования. 
Удалось создать класс для выскопроизводительной работы с сокетами для Delphi. Тема эта освещена в целом крайне слабо и я планирую продолжить эту публикацию ещё 2 - 3 постами по темам контекстов сокетов при использовании интерфейсов и создание защищённых соединений при использовании IOCP (криптопровайдеры и Winsock Secure Socket Extensions). Полный код примера здесь.