[fix] Socket: Fix stale indexes and surface close/error reliably in _hx_std_socket_poll_events

src/hx/libs/std/Socket.cpp

@@ -25,7 +25,12 @@
 typedef INT (WSAAPI *inet_pton_func)( INT Family, PCSTR pszAddrString, PVOID pAddrBuf);
 typedef PCSTR (WSAAPI *inet_ntop_func)(INT  Family, PVOID pAddr, PSTR pStringBuf, size_t StringBufSize);
 
-#   define FDSIZE(n)   (sizeof(u_int) + (n) * sizeof(SOCKET))
+// fd_array does not begin at sizeof(u_int). On 64-bit Windows SOCKET is
+// eight bytes, so fd_set carries four bytes of padding after fd_count and
+// the array starts at offset 8. Sizing the copy as 4 + n*8 truncated the
+// first handle by half and left the rest as whatever malloc returned, so
+// select() saw a handle that was not a socket and failed with WSAENOTSOCK.
+#   define FDSIZE(n)   (offsetof(fd_set, fd_array) + (n) * sizeof(SOCKET))
 #   define SHUT_WR      SD_SEND
 #   define SHUT_RD      SD_RECEIVE
 #   define SHUT_RDWR   SD_BOTH
@@ -1221,65 +1226,101 @@ Array<Dynamic> _hx_std_socket_poll_prepare( Dynamic pdata, Array<Dynamic> rsocks
    Update the read/write flags arrays that were created with [socket_poll_prepare].
    </doc>
 **/
-void _hx_std_socket_poll_events( Dynamic pdata, double timeout )
+void _hx_std_socket_poll_events(Dynamic pdata, double timeout)
 {
    polldata *p = val_poll(pdata);
 
-   #ifdef NEKO_WINDOWS
-   memcpy(p->outr,p->fdr,FDSIZE(p->fdr->fd_count));
-   memcpy(p->outw,p->fdw,FDSIZE(p->fdw->fd_count));
+   p->ridx[0] = -1;
+   p->widx[0] = -1;
+
+#ifdef NEKO_WINDOWS
+   memcpy(p->outr, p->fdr, FDSIZE(p->fdr->fd_count));
+   memcpy(p->outw, p->fdw, FDSIZE(p->fdw->fd_count));
+
+   fd_set oute;
+   FD_ZERO(&oute);
+   if (p->fdr->fd_count)
+   {
+      for (u_int i = 0; i < p->fdr->fd_count; ++i)
+         FD_SET(p->fdr->fd_array[i], &oute);
+   }
 
    struct timeval t;
-   struct timeval *tt = init_timeval(timeout,&t);
+   struct timeval *tt = init_timeval(timeout, &t);
 
    hx::EnterGCFreeZone();
-   if( select(0/* Ignored */, p->fdr->fd_count ? p->outr : 0, p->fdw->fd_count ?p->outw : 0,NULL,tt) == SOCKET_ERROR )
+   int sel = select(
+       0,
+       p->fdr->fd_count ? p->outr : 0,
+       p->fdw->fd_count ? p->outw : 0,
+       p->fdr->fd_count ? &oute : 0,
+       tt);
+   if (sel == SOCKET_ERROR)
    {
       hx::ExitGCFreeZone();
+
+      int k = 0;
+      for (u_int i = 0; i < p->fdr->fd_count; ++i)
+         p->ridx[k++] = i;
+      p->ridx[k] = -1;
+
       return;
    }
    hx::ExitGCFreeZone();
 
    int k = 0;
-   for(int i=0;i<p->fdr->fd_count;i++)
-      if( FD_ISSET(p->fdr->fd_array[i],p->outr) )
+   for (u_int i = 0; i < p->fdr->fd_count; ++i)
+   {
+      SOCKET fd = p->fdr->fd_array[i];
+      if (FD_ISSET(fd, p->outr) || FD_ISSET(fd, &oute))
          p->ridx[k++] = i;
+   }
    p->ridx[k] = -1;
 
    k = 0;
-   for(int i=0;i<p->fdw->fd_count;i++)
-      if( FD_ISSET(p->fdw->fd_array[i],p->outw) )
+   for (u_int i = 0; i < p->fdw->fd_count; ++i)
+   {
+      if (FD_ISSET(p->fdw->fd_array[i], p->outw))
          p->widx[k++] = i;
+   }
    p->widx[k] = -1;
 
-   #else
-
+#else
    int tot = p->rcount + p->wcount;
+
    hx::EnterGCFreeZone();
-   POSIX_LABEL(poll_events_again);
-   if( poll(p->fds,tot,(int)(timeout * 1000)) < 0 )
+poll_events_again:
+   if (poll(p->fds, tot, (int)(timeout * 1000)) < 0)
    {
-      HANDLE_EINTR(poll_events_again);
+      if (errno == EINTR)
+         goto poll_events_again;
       hx::ExitGCFreeZone();
+
       return;
    }
    hx::ExitGCFreeZone();
 
    int k = 0;
    int i = 0;
-   for(i=0;i<p->rcount;i++)
-      if( p->fds[i].revents & (POLLIN|POLLHUP) )
+
+   for (i = 0; i < p->rcount; ++i)
+   {
+      if (p->fds[i].revents & (POLLIN | POLLHUP | POLLERR | POLLNVAL))
          p->ridx[k++] = i;
+   }
    p->ridx[k] = -1;
+
    k = 0;
-   for(;i<tot;i++)
-      if( p->fds[i].revents & (POLLOUT|POLLHUP) )
+
+   for (; i < tot; ++i)
+   {
+      if (p->fds[i].revents & (POLLOUT | POLLHUP | POLLERR | POLLNVAL))
          p->widx[k++] = i - p->rcount;
+   }
    p->widx[k] = -1;
-   #endif
+#endif
 }
 
-
 /**
    socket_poll : 'socket array -> 'poll -> timeout:float -> 'socket array
    <doc>

test/std/Test.hx

@@ -502,6 +502,144 @@ class Test extends utest.Test
    }
 
 
+   // A listening socket with a connection waiting to be accepted is readable.
+   // _hx_std_socket_poll_events used to be able to return without reporting it,
+   // leaving the caller to conclude nothing had arrived.
+   function testPollReportsReadableSocket()
+   {
+      log("Test poll reports a readable socket");
+
+      var server = new Socket();
+      server.bind(new Host("127.0.0.1"), 0);
+      server.listen(1);
+
+      var client = new Socket();
+      client.connect(new Host("127.0.0.1"), server.host().port);
+
+      var poll = new Poll(4);
+      var readable = -1;
+
+      // Polled in a loop because the connection is not required to be visible
+      // on the first pass; the point is that it becomes visible at all.
+      for(attempt in 0...20)
+      {
+         poll.prepare([server], []);
+         poll.events(0.05);
+         readable = poll.readIndexes[0];
+         if (readable == 0)
+            break;
+         Sys.sleep(0.01);
+      }
+
+      v('readIndexes[0] = ${readable}');
+      Assert.equals(0, readable, "Listening socket with a pending connection was not reported readable");
+
+      client.close();
+      server.close();
+   }
+
+   // The ready lists are the caller's only result, so they have to describe the
+   // poll that just ran and not the one before it. An early return that left
+   // them untouched reported the previous call's sockets as ready again.
+   function testPollClearsIndexesBetweenCalls()
+   {
+      log("Test poll clears stale indexes");
+
+      var server = new Socket();
+      server.bind(new Host("127.0.0.1"), 0);
+      server.listen(1);
+
+      var client = new Socket();
+      client.connect(new Host("127.0.0.1"), server.host().port);
+
+      var poll = new Poll(4);
+
+      for(attempt in 0...20)
+      {
+         poll.prepare([server], []);
+         poll.events(0.05);
+         if (poll.readIndexes[0] == 0)
+            break;
+         Sys.sleep(0.01);
+      }
+      Assert.equals(0, poll.readIndexes[0], "Setup failed: socket never became readable");
+
+      // Nothing is registered now, so nothing can be ready.
+      poll.prepare([], []);
+      poll.events(0.01);
+
+      v('after empty poll: readIndexes[0] = ${poll.readIndexes[0]}, writeIndexes[0] = ${poll.writeIndexes[0]}');
+      Assert.equals(-1, poll.readIndexes[0], "Read index survived a poll with no sockets registered");
+      Assert.equals(-1, poll.writeIndexes[0], "Write index survived a poll with no sockets registered");
+
+      client.close();
+      server.close();
+   }
+
+   // A connected socket with room in its send buffer is writable, and the
+   // write list is filled by the same pass that fills the read list.
+   function testPollReportsWritableSocket()
+   {
+      log("Test poll reports a writable socket");
+
+      var server = new Socket();
+      server.bind(new Host("127.0.0.1"), 0);
+      server.listen(1);
+
+      var client = new Socket();
+      client.connect(new Host("127.0.0.1"), server.host().port);
+      var peer = server.accept();
+
+      var poll = new Poll(4);
+      var writable = -1;
+
+      for(attempt in 0...20)
+      {
+         poll.prepare([], [client]);
+         poll.events(0.05);
+         writable = poll.writeIndexes[0];
+         if (writable == 0)
+            break;
+         Sys.sleep(0.01);
+      }
+
+      v('writeIndexes[0] = ${writable}');
+      Assert.equals(0, writable, "Connected socket was not reported writable");
+
+      peer.close();
+      client.close();
+      server.close();
+   }
+
+   // Readiness has to mean readiness. When select() failed the error path
+   // marked every registered read socket ready, so an idle connection looked
+   // readable and a caller woke up to read nothing -- for as long as it stayed
+   // idle.
+   function testPollDoesNotReportIdleSocketAsReadable()
+   {
+      log("Test poll does not invent readability");
+
+      var server = new Socket();
+      server.bind(new Host("127.0.0.1"), 0);
+      server.listen(1);
+
+      var client = new Socket();
+      client.connect(new Host("127.0.0.1"), server.host().port);
+      var peer = server.accept();
+
+      // Nothing has been sent either way, so neither end has anything to read.
+      var poll = new Poll(4);
+      poll.prepare([client], []);
+      poll.events(0.05);
+
+      v('readIndexes[0] = ${poll.readIndexes[0]}');
+      Assert.equals(-1, poll.readIndexes[0], "Idle socket with no pending data was reported readable");
+
+      peer.close();
+      client.close();
+      server.close();
+   }
+
    function testUdpSocket()
    {
       log("Test UdpSocket");