Fix ASIO one shot lost notifications problem (#2897)

Prior to this commit, the way the ASIO notifications worked on
linux epoll included an edge case/race condition that allowed
for a loss of ASIO events.

As part of testing wallaroo scalability, we encountered an issue
(WallarooLabs/wally#2547) where some of our TCP actors would
stop reading from the socket for no reason even though there
was data waiting on the socket. After much fighting, @SeanTAllen
inferred that the ASIO subsystem was losing notifications and
the TCP actor was never notified about the data waiting on the
socket.

This commit changes things so that going forward:

* ASIO one shot subscriptions do *not* use edge triggered mode, but
  instead use level triggered mode
* Always tries to resubscribe to both read and write depending on
  the state of the `ev->readable` and `ev->writeable` booleans
* Only send exactly one notification to one shot subscribers
  for either reads or writes, until they reset the `ev->readable`
  and/or `ev->writeable` booleans and resubscribe for notifications
  again in case of spurious notifications from epoll due to the
  use of level triggered mode (even though one shot should ensure
  that we only get one notification).

These changes ensure that if there's data on the socket at the
time we (re)subscribe to reads, we will be told immediately by
epoll that there is a read notification.

Similarly, if the socket is writeable at the time we (re)subscribe
to writes, we will be told immediately by epoll that there is a
write notification.

Both of these ensure that the ASIO subsystem will always be told
by epoll when the socket is readable or writeable and avoid any
race conditions due to edge triggering and the state of the socket
at the time the subscription occurs.

Author: dipinhora

src/libponyrt/asio/epoll.c

@@ -144,11 +144,14 @@ void ponyint_asio_backend_final(asio_backend_t* b)
   eventfd_write(b->wakeup, 1);
 }
 
-PONY_API void pony_asio_event_resubscribe_write(asio_event_t* ev)
+// Single function for resubscribing to both reads and writes for an event
+PONY_API void pony_asio_event_resubscribe(asio_event_t* ev)
 {
+  // needs to be a valid event that is one shot enabled
   if((ev == NULL) ||
     (ev->flags == ASIO_DISPOSABLE) ||
-    (ev->flags == ASIO_DESTROYED))
+    (ev->flags == ASIO_DESTROYED) ||
+    !(ev->flags & ASIO_ONESHOT))
   {
     pony_assert(0);
     return;
@@ -160,44 +163,44 @@ PONY_API void pony_asio_event_resubscribe_write(asio_event_t* ev)
   struct epoll_event ep;
   ep.data.ptr = ev;
   ep.events = 0;
+  bool something_to_resub = false;
 
-  if(ev->flags & ASIO_ONESHOT)
-    ep.events |= EPOLLONESHOT;
-
+  // if the event is supposed to be listening for write notifications
+  // and it is currently not writeable
   if((ev->flags & ASIO_WRITE) && !ev->writeable)
+  {
+    something_to_resub = true;
     ep.events |= EPOLLOUT;
-  else
-    return;
-
-  epoll_ctl(b->epfd, EPOLL_CTL_MOD, ev->fd, &ep);
-}
+  }
 
-PONY_API void pony_asio_event_resubscribe_read(asio_event_t* ev)
-{
-  if((ev == NULL) ||
-    (ev->flags == ASIO_DISPOSABLE) ||
-    (ev->flags == ASIO_DESTROYED))
+  // if the event is supposed to be listening for read notifications
+  // and it is currently not readable
+  if((ev->flags & ASIO_READ) && !ev->readable)
   {
-    pony_assert(0);
-    return;
+    something_to_resub = true;
+    ep.events |= EPOLLRDHUP;
+    ep.events |= EPOLLIN;
   }
 
-  asio_backend_t* b = ponyint_asio_get_backend();
-  pony_assert(b != NULL);
-
-  struct epoll_event ep;
-  ep.data.ptr = ev;
-  ep.events = EPOLLRDHUP | EPOLLET;
-
-  if(ev->flags & ASIO_ONESHOT)
-    ep.events |= EPOLLONESHOT;
+  // only resubscribe if there is something to resubscribe to
+  if (something_to_resub)
+    epoll_ctl(b->epfd, EPOLL_CTL_MOD, ev->fd, &ep);
+}
 
-  if((ev->flags & ASIO_READ) && !ev->readable)
-    ep.events |= EPOLLIN;
-  else
-    return;
+// Kept to maintain backwards compatibility so folks don't
+// have to change their code to use `pony_asio_event_resubscribe`
+// immediately
+PONY_API void pony_asio_event_resubscribe_write(asio_event_t* ev)
+{
+  pony_asio_event_resubscribe(ev);
+}
 
-  epoll_ctl(b->epfd, EPOLL_CTL_MOD, ev->fd, &ep);
+// Kept to maintain backwards compatibility so folks don't
+// have to change their code to use `pony_asio_event_resubscribe`
+// immediately
+PONY_API void pony_asio_event_resubscribe_read(asio_event_t* ev)
+{
+  pony_asio_event_resubscribe(ev);
 }
 
 DECLARE_THREAD_FN(ponyint_asio_backend_dispatch)
@@ -235,17 +238,35 @@ DECLARE_THREAD_FN(ponyint_asio_backend_dispatch)
       {
         if(ep->events & (EPOLLIN | EPOLLRDHUP | EPOLLHUP | EPOLLERR))
         {
-          flags |= ASIO_READ;
-          ev->readable = true;
+          // Send read notification to an actor if either
+          // * the event is not a one shot event
+          // * the event is a one shot event and we haven't already sent a notification
+          // if the event is a one shot event and we have already sent a notification
+          // don't send another one until we are asked for it again (i.e. the actor
+          // gets a 0 byte read and sets `readable` to false and resubscribes to reads
+          if(((ev->flags & ASIO_ONESHOT) && !ev->readable) || !(ev->flags & ASIO_ONESHOT))
+          {
+            ev->readable = true;
+            flags |= ASIO_READ;
+          }
         }
       }
 
       if(ev->flags & ASIO_WRITE)
       {
         if(ep->events & EPOLLOUT)
         {
-          flags |= ASIO_WRITE;
-          ev->writeable = true;
+          // Send write notification to an actor if either
+          // * the event is not a one shot event
+          // * the event is a one shot event and we haven't already sent a notification
+          // if the event is a one shot event and we have already sent a notification
+          // don't send another one until we are asked for it again (i.e. the actor
+          // gets partial write and sets `writeable` to false and resubscribes to writes
+          if(((ev->flags & ASIO_ONESHOT) && !ev->writeable) || !(ev->flags & ASIO_ONESHOT))
+          {
+            flags |= ASIO_WRITE;
+            ev->writeable = true;
+          }
         }
       }
 
@@ -272,15 +293,20 @@ DECLARE_THREAD_FN(ponyint_asio_backend_dispatch)
         }
       }
 
+      // if we had a valid event of some type that needs to be sent
+      // to an actor
       if(flags != 0)
       {
-        if (!(flags & ASIO_DESTROYED) && !(flags & ASIO_DISPOSABLE))
-        {
-          if(ev->auto_resub && !(flags & ASIO_WRITE))
-            pony_asio_event_resubscribe_write(ev);
-          if(ev->auto_resub && !(flags & ASIO_READ))
-            pony_asio_event_resubscribe_read(ev);
-        }
+        // if this event hasn't been destroyed or disposed.
+        // to avoid a race condition if destroyed or dispoed events
+        // are resubscribed
+        if((ev->flags != ASIO_DISPOSABLE) && (ev->flags != ASIO_DESTROYED))
+          // if this event is using one shot and should auto resubscribe and
+          // then resubscribe
+          if(ev->flags & ASIO_ONESHOT)
+            pony_asio_event_resubscribe(ev);
+
+        // send the event to the actor
         pony_asio_event_send(ev, flags, count);
       }
     }
@@ -377,9 +403,18 @@ PONY_API void pony_asio_event_subscribe(asio_event_t* ev)
 
   if(ev->flags & ASIO_ONESHOT) {
     ep.events |= EPOLLONESHOT;
-    ev->auto_resub = true;
+    // disable edge triggering if one shot is enabled
+    // this is because of how the runtime gets notifications
+    // from epoll in this ASIO thread and then notifies the
+    // appropriate actor to read/write as necessary.
+    // specifically, it seems there's an edge case/race condition
+    // with edge triggering where if there is already data waiting
+    // on the socket, then epoll might not be triggering immediately
+    // when an edge triggered epoll request is made.
+    ep.events &= ~EPOLLET;
   }
 
+
   epoll_ctl(b->epfd, EPOLL_CTL_ADD, ev->fd, &ep);
 }
 

src/libponyrt/asio/event.h

@@ -22,15 +22,6 @@ typedef struct asio_event_t
   bool noisy;           /* prevents termination? */
   uint64_t nsec;        /* nanoseconds for timers */
 
-#ifdef PLATFORM_IS_LINUX
-  // automagically resubscribe to an event on an FD when another event is
-  // on the same FD is triggered
-  // This is only needed on linux where an event firing on an FD that has
-  // ONESHOT enabled will clear all events for the FD and not only the
-  // event that was fired.
-  bool auto_resub;      /* automagically resubscribe? */
-#endif
-
   bool readable;        /* is fd readable? */
   bool writeable;       /* is fd writeable? */
 #ifdef PLATFORM_IS_WINDOWS