Non-blocking overlapped receive socket

This tutorial is only for winsock2. Some of the content is from https://www.winsocketdotnetworkprogramming.com/winsock2programming/winsock2advancediomethod5e.html and https://learn.microsoft.com/en-us/windows/win32/api/_winsock/

Why use overlapped socket?

Using winsock2 can be pretty simple if it is used in blocking mode, but we can always see a huge bunch of 0 and nulls being passesed as arguments, does are the capability to did not put our hands on. Overlapped socket can potentially increase throughput when there is a lot of communication needed to handle.

Important Functions

Besides the socket functions you need in blocking mode, you beed to use some microsoft event functions.

WSAWaitForMultipleEvents()  

This function is quite similar to select() in usage. This function can be used to wait for transmition.

WSAResetEvent()

This functions resets the event after that particular event has received something. This is better compared to fd_set() in select() because we don’t have to reset everything everytime we receive something.

WSAGetOverlappedResult()

This function is mainly used to get the amount of data byte’s received.

Flowchart

The flow of this scheme starts with creating multiple requests. After starting, we use WSAWaitforMultipleEvents() to wait for a packet to arrive. It is possible to use a timeout here to check for low traffic. We then reset the event so that it can be raised again. Now we can start looking in to what it is received, we can get the amount of bytes we received from WSAGetOverlappedResult(). We pass the amount of bytes and the pointer to the receive array and any necessary data needed for processing incoming data. Finally, we create a new create a new receive array and create a new receive request in place of the old ones.

SOCKET Information

To keep track of the event and it’s associates , a structure to store all data is recommended.

typedef struct SOCKET_INFORMATION {
 CHAR* Buffer = NULL;     // The actual receive buffer
 WSABUF DataBuf = { 0 };  // Point DataBuf.buf to the buffer above, and set its length accordingly
 SOCKET Socket = 0;
 WSAOVERLAPPED Overlapped = { 0 };  // The overlapped event
 DWORD Flags = 0;
 SOCKADDR_IN SockAddrFrom = { 0 };
} SOCKET_INFORMATION;

A WSAEVENT array are also needed for the WSAWaitForMultipleEvents() to work.

Create the requests

We start with declaring variables to hold data.

WSAEVENT EventArray[NUM_OF_EVENTS];
SOCKET_INFORMATION Sockinfo[NUM_OF_EVENTS];

Remeber to create socket with the overlapped option

 SOCKET Socket = WSASocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP, NULL, 0, WSA_FLAG_OVERLAPPED);

Initialize everything with a loop.

 // Initilize event array and socket information
 for (int i = 0; i < NUM_OF_EVENTS; i++) {
  ZeroMemory(&Sockinfo[i], sizeof(SOCKET_INFORMATION));
  EventArray[i] = WSACreateEvent();  // Event array
  Sockinfo[i].Overlapped.hEvent = EventArray[i];
  Sockinfo[i].Socket = Socket;
  Sockinfo[i].Buffer = new char[UDP_MAX_SIZE + 1];
  Sockinfo[i].DataBuf.buf = Sockinfo[i].Buffer;
  Sockinfo[i].DataBuf.len = UDP_MAX_SIZE;
// create actual receive request
  if (WSARecvFrom(Sockinfo[i].Socket, &Sockinfo[i].DataBuf, 1, NULL,
   &Sockinfo[i].Flags, (SOCKADDR*)&Sockinfo[i].SockAddrFrom,
   &socksize, &Sockinfo[i].Overlapped, NULL) == SOCKET_ERROR)
  {
   if (WSAGetLastError() != WSA_IO_PENDING) {
    std::cout << "recvfrom() failed with error: <" << WSAGetLastError() << ">\n";
    closesocket(Socket);
    WSACleanup();
    return -1;
   }
  }
 }

The main loop waits for the receive. If something is received, use a threadpool to process it then create a receive request to replace the old one.

while (1) {

  DWORD index = WSAWaitForMultipleEvents(NUM_OF_EVENTS, EventArray, FALSE, timeout, FALSE);
  if (index == WSA_WAIT_TIMEOUT) {
   std::cout << "Timeout\n";
   continue;
  }
  // Reset to not trigger again
  WSAResetEvent(EventArray[index - WSA_WAIT_EVENT_0]);
  // check result
  WSAGetOverlappedResult(Sockinfo[index - WSA_WAIT_EVENT_0].Socket, &Sockinfo[index - WSA_WAIT_EVENT_0].Overlapped, &BytesTransferred, FALSE, &Sockinfo[index - WSA_WAIT_EVENT_0].Flags);
  if (BytesTransferred > 0)
  {

   //Process received
   pool.push_task(Do_something, Sockinfo[index - WSA_WAIT_EVENT_0].Buffer, BytesTransferred);

   //Add a new request
   Sockinfo[index - WSA_WAIT_EVENT_0].Buffer = new char[UDP_MAX_SIZE + 1];
   Sockinfo[index - WSA_WAIT_EVENT_0].DataBuf.buf = Sockinfo[index - WSA_WAIT_EVENT_0].Buffer;
   Sockinfo[index - WSA_WAIT_EVENT_0].Overlapped.hEvent = EventArray[index - WSA_WAIT_EVENT_0];
   if (WSARecvFrom(Sockinfo[index - WSA_WAIT_EVENT_0].Socket, &Sockinfo[index - WSA_WAIT_EVENT_0].DataBuf, 1, NULL,
    &Sockinfo[index - WSA_WAIT_EVENT_0].Flags, (SOCKADDR*)&Sockinfo[index - WSA_WAIT_EVENT_0].SockAddrFrom,
    &socksize, &Sockinfo[index - WSA_WAIT_EVENT_0].Overlapped, NULL) == SOCKET_ERROR)
   {
    if (WSAGetLastError() != WSA_IO_PENDING) {
     std::cout<< "recvfrom() failed with error <"<< WSAGetLastError()<<">\n";
     closesocket(Socket);
     WSACleanup();
     return -1;
    }
   }
  }
 }

Working example

#include <iostream>
#include <string>
#include <Winsock2.h> 
#include <Ws2tcpip.h> 
#include <algorithm>
#include <future>
#include "BS_thread_pool.hpp" //Thread pool https://github.com/bshoshany/thread-pool

#pragma comment(lib,"WS2_32")

#define UDP_MAX_SIZE 1500
#define NUM_OF_EVENTS 20

int Port = 15100;
std::string IP = "234.255.255.1";
int optval = 0;

typedef struct SOCKET_INFORMATION {
 CHAR* Buffer = NULL;
 WSABUF DataBuf = { 0 };
 SOCKET Socket = 0;
 WSAOVERLAPPED Overlapped = { 0 };
 DWORD Flags = 0;
 SOCKADDR_IN SockAddrFrom = { 0 };
} SOCKET_INFORMATION;

void Do_something(char* Received, DWORD Length);

int main()
{
 sockaddr_in AllowAddr;
 ip_mreq JoinReq;
 WSAEVENT EventArray[NUM_OF_EVENTS];
 SOCKET_INFORMATION Sockinfo[NUM_OF_EVENTS];
 int socksize = sizeof(SOCKADDR_IN);

 BS::thread_pool pool; //Thread pool https://github.com/bshoshany/thread-pool

 //WSAStartup
 WSADATA wsaData;
 int iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);
 if (iResult != NO_ERROR) {
  std::cout << "WSAStartup function failed with error: <" << iResult << ">\n";
  WSACleanup();
  return -1;
 }
 // Create socket. Note the WSA_FLAG_OVERLAPPED
 SOCKET Socket = WSASocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP, NULL, 0, WSA_FLAG_OVERLAPPED);
 if (Socket == INVALID_SOCKET)
 {
  std::cout << "Can not create socket: <" << WSAGetLastError() << ">\n";
  closesocket(Socket);
  WSACleanup();
  return -1;
 }

 // Allow reuse of port
 optval = 1;
 if ((setsockopt(Socket, SOL_SOCKET, SO_REUSEADDR, (char*)&optval, sizeof(optval))) < 0)
 {
  std::cout << "Socket set SO_REUSEADDR fail\n";
  closesocket(Socket);
  WSACleanup();
  return -1;
 }

 // Allow any address
 memset(&AllowAddr, 0, sizeof(AllowAddr));
 AllowAddr.sin_family = AF_INET;
 AllowAddr.sin_addr.s_addr = htonl(INADDR_ANY);
 AllowAddr.sin_port = htons(Port);

 // Bind socket
 if (bind(Socket, (struct sockaddr*)&AllowAddr, sizeof(AllowAddr)) < 0) {
  std::cout << "Multicast failed to bind socket\n";
  closesocket(Socket);
  WSACleanup();
  return -1;
 }

 // Membership setting
 if (inet_pton(AF_INET, (PCSTR)(IP.c_str()), &JoinReq.imr_multiaddr.s_addr) < 0) {
  std::cout << "IP invalid\n";
  closesocket(Socket);
  WSACleanup();
  return -1;
 }
 // This can be used to restrict to only receive form particular sender
 JoinReq.imr_interface.s_addr = htonl(INADDR_ANY);

 // Join membership
 if ((setsockopt(Socket, IPPROTO_IP, IP_ADD_MEMBERSHIP, (char*)&JoinReq, sizeof(JoinReq))) < 0)
 {
  std::cout << "Multicast join membership fail. Error code: <" << WSAGetLastError() << ">\n";
  closesocket(Socket);
  WSACleanup();
  return -1;
 }
 std::cout << "Multicast group joined\n";
 // Initilize event array and socket information
 for (int i = 0; i < NUM_OF_EVENTS; i++) {
  ZeroMemory(&Sockinfo[i], sizeof(SOCKET_INFORMATION));
  EventArray[i] = WSACreateEvent();
  Sockinfo[i].Overlapped.hEvent = EventArray[i];
  Sockinfo[i].Socket = Socket;
  Sockinfo[i].Buffer = new char[UDP_MAX_SIZE + 1];
  Sockinfo[i].DataBuf.buf = Sockinfo[i].Buffer;
  Sockinfo[i].DataBuf.len = UDP_MAX_SIZE;
  if (WSARecvFrom(Sockinfo[i].Socket, &Sockinfo[i].DataBuf, 1, NULL,
   &Sockinfo[i].Flags, (SOCKADDR*)&Sockinfo[i].SockAddrFrom,
   &socksize, &Sockinfo[i].Overlapped, NULL) == SOCKET_ERROR)
  {
   if (WSAGetLastError() != WSA_IO_PENDING) {
    std::cout << "recvfrom() failed with error: <" << WSAGetLastError() << ">\n";
    closesocket(Socket);
    WSACleanup();
    return -1;
   }
  }
 }

 std::cout << "Ready to receive.\n";
 DWORD index = 0, BytesTransferred = 0, timeout = 5000;
 //Receive loop
 while (1) {

  DWORD index = WSAWaitForMultipleEvents(NUM_OF_EVENTS, EventArray, FALSE, timeout, FALSE);
  if (index == WSA_WAIT_TIMEOUT) {
   std::cout << "Timeout\n";
   continue;
  }
  // Reset to not trigger again
  WSAResetEvent(EventArray[index - WSA_WAIT_EVENT_0]);
  // check result
  WSAGetOverlappedResult(Sockinfo[index - WSA_WAIT_EVENT_0].Socket, &Sockinfo[index - WSA_WAIT_EVENT_0].Overlapped, &BytesTransferred, FALSE, &Sockinfo[index - WSA_WAIT_EVENT_0].Flags);
  if (BytesTransferred > 0)
  {

   //Process received
   pool.push_task(Do_something, Sockinfo[index - WSA_WAIT_EVENT_0].Buffer, BytesTransferred);

   //Add a new request
   Sockinfo[index - WSA_WAIT_EVENT_0].Buffer = new char[UDP_MAX_SIZE + 1];
   Sockinfo[index - WSA_WAIT_EVENT_0].DataBuf.buf = Sockinfo[index - WSA_WAIT_EVENT_0].Buffer;
   Sockinfo[index - WSA_WAIT_EVENT_0].Overlapped.hEvent = EventArray[index - WSA_WAIT_EVENT_0];
   if (WSARecvFrom(Sockinfo[index - WSA_WAIT_EVENT_0].Socket, &Sockinfo[index - WSA_WAIT_EVENT_0].DataBuf, 1, NULL,
    &Sockinfo[index - WSA_WAIT_EVENT_0].Flags, (SOCKADDR*)&Sockinfo[index - WSA_WAIT_EVENT_0].SockAddrFrom,
    &socksize, &Sockinfo[index - WSA_WAIT_EVENT_0].Overlapped, NULL) == SOCKET_ERROR)
   {
    if (WSAGetLastError() != WSA_IO_PENDING) {
     std::cout<< "recvfrom() failed with error <"<< WSAGetLastError()<<">\n";
     closesocket(Socket);
     WSACleanup();
     return -1;
    }
   }
  }
 }
 closesocket(Socket);
 WSACleanup();

 return 1;
}

void Do_something(char* Received, DWORD  Length) {
 // Remember to use locks if needed!
 std::cout << "Received: " << Received << " (" << Length << " Bytes)\n";
 delete(Received);
}