Using WSAPoll for Windows socket selector.

IOCP is not really compatible with kqueue or epoll, it's
much richer than I expected. So, on Windows, we do the same
as select.

Author: ktakashi

dist.bat

@@ -16,30 +16,30 @@ goto:eof
 
 rem insn
 :insn
-echo "Generating instructions files"
+echo Generating instructions files
 cd src
 %SASH% geninsn %1
 cd ..
 goto:eof
 
 rem precomp
 :precomp
-echo "Generating compiled library files"
+echo Generating compiled library files
 cd src
 %SASH% genlib %1
 cd ..
 call :insn dummy %1
 cd tools\scripts
-echo "Generating builtin keywords"
+echo Generating builtin keywords
 %SASH% builtin-keywords.scm
-echo "Generating builtin symbols"
+echo Generating builtin symbols
 %SASH% builtin-symbols.scm
 cd  ..\..
 goto:eof
 
 rem stub
 :stub
-echo "Generating library from stub"
+echo Generating library from stub
 cd src
 %SASH% genstub %1
 cd ..
@@ -48,18 +48,23 @@ goto:eof
 rem srfi
 :srfi
 echo Generating R7RS style SRFI libraries
-%SASH% -L./sitelib ./tools/scripts/r7rs-srfi-gen.scm -p ./ext -p ./sitelib/srfi %1
+%SASH% -L./sitelib ./tools/scripts/r7rs-srfi-gen.scm^
+      -p ./ext -p ./sitelib/srfi %1
 goto:eof
 
 rem tzdata
 :tz
-echo "Generating TZ database"
-%SASH% ./tools/scripts/compile-tzdatabase.scm -o ext/time/sagittarius/tzdata.scm -w ext/time/sagittarius/win-mappings.scm -l ext/time/sagittarius/leap-table.scm -r %1
+echo Generating TZ database
+%SASH% ./tools/scripts/compile-tzdatabase.scm^
+    -o ext/time/sagittarius/tzdata.scm^
+    -w ext/time/sagittarius/win-mappings.scm^
+    -l ext/time/sagittarius/leap-table.scm^
+    -r %1
 goto:eof
 
 rem unicode
 :unicode
-echo "Generating Unicode codepoints"
+echo Generating Unicode codepoints
 %SASH% ./tools/scripts/compile-unicode.scm %1
 
 if "%1" == "-c" goto:unicode_end
@@ -95,7 +100,7 @@ goto:eof
 
 rem html
 :html
-echo "Generating HTML entities"
+echo Generating HTML entities
 %SASH% ./tools/scripts/html-entities.scm -o sitelib/text/xml/entities-list.scm %1
 goto:eof
 
@@ -127,20 +132,20 @@ for %%x in (%*) do call :%%x
 goto end
 
 :usage
-echo "usage: %0 precomp|stub|srfi|tz|clean"
-echo "    gen:        generates all files"
-echo "    precomp:    generates precompiled files"
-echo "    stub:       generates stub files"
-echo "    srfi:       generates R7RS style SRFI libraries"
-echo "    tz:         generates TZ database"
-echo "    unicode:    generates Unicode codepoints"
-echo "    html:       generates HTML entries"
-echo "    clean:      cleasn generated files"
+echo usage: %0 precomp|stub|srfi|tz|clean
+echo     gen:        generates all files
+echo     precomp:    generates precompiled files
+echo     stub:       generates stub files
+echo     srfi:       generates R7RS style SRFI libraries
+echo     tz:         generates TZ database
+echo     unicode:    generates Unicode codepoints
+echo     html:       generates HTML entries
+echo     clean:      cleasn generated files
 
 goto :end
 
 :err
-echo "Sagittarius is not installed. Default %SASH%"
+echo Sagittarius is not installed. Default %SASH%
 
 :end
 

ext/socket/CMakeLists.txt

@@ -12,11 +12,11 @@ ENDIF()
 # Simple assumption
 # Linux         - epoll
 # *BSD or macOS - kqueue
-# Windows       - iocp
+# Windows       - WSAPoll
 IF (${CMAKE_SYSTEM_NAME} STREQUAL "Linux")
   SET(SELECTOR_SOURCE selector-epoll.c)
 ELSEIF(WIN32)
-  SET(SELECTOR_SOURCE selector-iocp.c)
+  SET(SELECTOR_SOURCE selector-win.c)
 ELSE()
   # Assume *BSD or macOS,
   CHECK_INCLUDE_FILES("stdint.h;sys/types.h;sys/event.h"  HAVE_SYS_EVENT_H)

ext/socket/sagittarius-socket.c

@@ -1640,7 +1640,7 @@ SG_EXTENSION_ENTRY void CDECL Sg_Init_sagittarius__socket()
   SgLibrary *lib;
 #if defined(_WIN32)
   WSADATA wsaData;
-  WSAStartup(2, &wsaData);
+  WSAStartup(MAKEWORD(2, 2), &wsaData);
   Sg_AddCleanupHandler(finish_winsock, NULL);
 #endif
   SG_INIT_EXTENSION(sagittarius__socket);

ext/socket/selector-iocp.c renamed to ext/socket/selector-win.c

@@ -1,4 +1,4 @@
-/* selector-iocp.c                                  -*- mode:c; coding:utf-8; -*-
+/* selector-iocp.c                                 -*- mode:c; coding:utf-8; -*-
  *
  *   Copyright (c) 2023  Takashi Kato <ktakashi@ymail.com>
  *
@@ -34,30 +34,33 @@
 
 #include "socket-selector.incl"
 
-typedef struct iocp_context_rec
+typedef struct win_context_rec
 {
-  HANDLE iocp;
+  HANDLE event;
   HANDLE thread;		/* waiting thread */
-} iocp_context_t;
+} win_context_t;
 
 
 static void system_error(int code)
 {
-  Sg_SystemError(GetLastError(),
+  Sg_SystemError(code,
 		 UC("Setting up IOCP failed: %A"),
 		 Sg_GetLastErrorMessageWithErrorCode(code));
 }
 
 SgObject Sg_MakeSocketSelector()
 {
   SgSocketSelector *selector = SG_NEW(SgSocketSelector);
-  iocp_context_t *ctx = SG_NEW(iocp_context_t);
+  win_context_t *ctx = SG_NEW(win_context_t);
 
   SG_SET_CLASS(selector, SG_CLASS_SOCKET_SELECTOR);
-  ctx->iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
-  if (ctx->iocp == NULL) goto err;
+
+  ctx->event = CreateEvent(NULL, FALSE, FALSE, NULL);
+  if (ctx->event == NULL) goto err;
+  
   ctx->thread = NULL;
   selector->sockets = SG_NIL;
+  selector->context = ctx;
 
   Sg_RegisterFinalizer(selector, selector_finalizer, NULL);
   return SG_OBJ(selector);
@@ -69,32 +72,34 @@ SgObject Sg_MakeSocketSelector()
 
 void Sg_CloseSocketSelector(SgSocketSelector *selector)
 {
-  iocp_context_t *ctx = (iocp_context_t *)selector->context;
-  CloseHandle(ctx->iocp);
+  win_context_t *ctx = (win_context_t *)selector->context;
+  CloseHandle(ctx->event);
   Sg_UnregisterFinalizer(selector);
 }
 
 SgObject Sg_SocketSelectorAdd(SgSocketSelector *selector, SgSocket *socket)
 {
-  iocp_context_t *ctx = (iocp_context_t *)selector->context;
-  HANDLE r;
-  r = CreateIoCompletionPort(ctx->iocp, (HANDLE)socket->socket,
-			     (ULONG_PTR)socket, 0);
-  if (r == NULL) {
-    system_error(Sg_GetLastError());
-  }
+  win_context_t *ctx = (win_context_t *)selector->context;
   selector->sockets = Sg_Cons(socket, selector->sockets);
   selector_sockets(selector);
   return SG_OBJ(selector);
 }
 
+static SgObject select_socket(SOCKET fd, SgObject sockets)
+{
+  SgObject cp;
+  SG_FOR_EACH(cp, sockets) {
+    SgSocket *sock = SG_SOCKET(SG_CAR(cp));
+    if (sock->socket == fd) return sock;
+  }
+  return SG_FALSE;
+}
+
 SgObject Sg_SocketSelectorWait(SgSocketSelector *selector, SgObject timeout)
 {
-  iocp_context_t *ctx = (iocp_context_t *)selector->context;
-  int n = selector_sockets(selector), i, millis = INFINITE;
-  ULONG removed;
-  LPOVERLAPPED_ENTRY entries;
-  BOOL r;
+  win_context_t *ctx = (win_context_t *)selector->context;
+  int n = selector_sockets(selector), millis = INFINITE, r;
+  HANDLE hEvents[2];
   struct timespec spec, *sp;
   SgObject ret = SG_NIL;
 
@@ -103,35 +108,71 @@ SgObject Sg_SocketSelectorWait(SgSocketSelector *selector, SgObject timeout)
   }
   ctx->thread = GetCurrentThread();
 
+  hEvents[0] = CreateEvent(NULL, FALSE, FALSE, NULL);
+  hEvents[1] = ctx->event;
+
+#define SET_EVENT(sockets, event, flags)		\
+  do {							\
+    SgObject cp;					\
+    SG_FOR_EACH(cp, sockets) {				\
+      SG_SET_SOCKET_EVENT(SG_CAR(cp), event, flags);	\
+    }							\
+  } while(0)
+
+  SET_EVENT(selector->sockets, hEvents[0], FD_READ | FD_OOB);
+  
   sp = selector_timespec(timeout, &spec);
   if (sp) {
     millis = sp->tv_sec * 1000;
     millis += sp->tv_nsec / 1000000;
   }
 
-  entries = SG_NEW_ATOMIC2(OVERLAPPED_ENTRY *, n * sizeof(OVERLAPPED_ENTRY));
-  r = GetQueuedCompletionStatusEx(ctx->iocp, entries, n, &removed, millis, TRUE);
-  if (r) {
-    for (i = 0; i < removed; i++) {
-      ret = Sg_Cons((SgObject) entries[i].lpCompletionKey, ret);
+  r = WaitForMultipleObjects(2, hEvents, FALSE, millis);
+  if (r == WAIT_OBJECT_0) {
+    /* Using WSAPoll to detect which sockets are ready to read */
+    WSAPOLLFD *fds = SG_NEW_ATOMIC2(WSAPOLLFD *, n * sizeof(WSAPOLLFD));
+    int i = 0;
+    SgObject cp;
+    SG_FOR_EACH(cp, selector->sockets) {
+      SgSocket *sock = SG_SOCKET(SG_CAR(cp));
+      fds[i].fd = sock->socket;
+      fds[i].events = POLLRDNORM;
+      i++;
     }
-  } else {
-    system_error(Sg_GetLastError());
+    r = WSAPoll(fds, n, 0); /* Some sockets must be ready at this stage */
+    if (r == SOCKET_ERROR) system_error(WSAGetLastError());
+    for (i = 0; i < n; i++) {
+      if (fds[i].revents & POLLRDNORM) {
+	/* collect sockets, should we use hashtable? */
+	SgObject o = select_socket(fds[i].fd, selector->sockets);
+	if (!SG_FALSEP(o)) ret = Sg_Cons(o, ret);
+      }
+    }
+  }
+
+  SET_EVENT(selector->sockets, hEvents[0], 0);
+#undef SET_EVENT
+
+  if (!SG_NULLP(ret)) {
+    /* remove the returned sockets from the targets */
+    SgObject h = SG_NIL, t = SG_NIL, cp;
+    SG_FOR_EACH(cp, selector->sockets) {
+      if (SG_FALSEP(Sg_Memq(SG_CAR(cp), ret))) {
+	SG_APPEND1(h, t, SG_CAR(cp));
+      }
+    }
+    selector->sockets = h;
   }
+  CloseHandle(hEvents[0]);
   ctx->thread = NULL;
   return ret;
 }
 
-static void CALLBACK dummy(ULONG_PTR param)
-{
-  /* Do nothing */
-}
-
 SgObject Sg_SocketSelectorInterrupt(SgSocketSelector *selector)
 {
-  iocp_context_t *ctx = (iocp_context_t *)selector->context;
+  win_context_t *ctx = (win_context_t *)selector->context;
   if (ctx->thread) {
-    QueueUserAPC(dummy, ctx->thread, 0);
+    SetEvent(ctx->event);
   }
   return selector;
 }

ext/socket/socket-selector.h

@@ -35,9 +35,9 @@
   Socket selector.
   The underlying implementations are platform specific. At this moment,
   we support:
-  - kqueue (*BSD and macOS)
-  - IOCP   (Windows)
-  - epoll  (Linux)
+  - kqueue  (*BSD and macOS)
+  - WSAPoll (Windows)
+  - epoll   (Linux)
  */
 typedef struct SgSocketSelectorRec
 {