[fuchsia,dart:io] Emulate edge-triggering of close events

The asynchronous waiting mechanism used by the eventhandler thread
on Fuchsia is inherently level-triggered. However, since the upper
layers of the dart:io socket implementation expect edge-triggering,
the Fuchsia eventhandler has to emulate edge-triggering.

Previously, this edge-triggering was emulated for kInEvent and
kOutEvent, but not kCloseEvent. It was not thought that simulating
edge-triggering for kCloseEvent was necessary since the socket would
soon be going down, and that a few redundant messages would be
harmless. This turned out not to be the case. In practice, the
eventhandler thread may flood the Dart thread with kCloseEvent
messages.

This CL adds the edge-triggering emulation for kCloseEvent. The
Dart thread will no longer be flooded with kCloseEvent messages.

Unfortunately, the eventhandler thread may still spin on Fuchsia's
equivalent of POLLHUP signals until the Dart thread destroys the
socket, as these cannot be unsubscribed from.

Fixes b/150902962

Change-Id: I2784220197771508689504156a761842ac164c8e
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/143888
Reviewed-by: Zichang Guo <zichangguo@google.com>
Commit-Queue: Zach Anderson <zra@google.com>

runtime/bin/eventhandler_fuchsia.cc

@@ -161,7 +161,12 @@ uint32_t IOHandle::MaskToEpollEvents(intptr_t mask) {
   MutexLocker ml(&mutex_);
   // Do not ask for POLLERR and POLLHUP explicitly as they are
   // triggered anyway.
-  uint32_t events = POLLRDHUP;
+  uint32_t events = 0;
+  // Do not subscribe to read closed events when kCloseEvent has already been
+  // sent to the Dart thread.
+  if (close_events_enabled_) {
+    events |= POLLRDHUP;
+  }
   if (read_events_enabled_ && ((mask & (1 << kInEvent)) != 0)) {
     events |= POLLIN;
   }
@@ -253,19 +258,39 @@ uint32_t IOHandle::WaitEnd(zx_signals_t observed) {
   return events;
 }
 
+// This function controls the simulation of edge-triggering. It is responsible
+// for removing events from the event mask when they should be supressed, and
+// for supressing future events. Events are unsupressed by their respective
+// operations by the Dart thread on the socket---that is, where the
+// *_events_enabled_ flags are set to true.
 intptr_t IOHandle::ToggleEvents(intptr_t event_mask) {
   MutexLocker ml(&mutex_);
+  // If write events are disabled, then remove the kOutEvent bit from the
+  // event mask.
   if (!write_events_enabled_) {
-    LOG_INFO("IOHandle::ToggleEvents: fd = %ld de-asserting write\n", fd_);
+    LOG_INFO(
+        "IOHandle::ToggleEvents: fd = %ld "
+        "de-asserting kOutEvent\n",
+        fd_);
     event_mask = event_mask & ~(1 << kOutEvent);
   }
+  // If the kOutEvent bit is set, then supress future write events until the
+  // Dart thread writes.
   if ((event_mask & (1 << kOutEvent)) != 0) {
-    LOG_INFO("IOHandle::ToggleEvents: fd = %ld asserting write and disabling\n",
-             fd_);
+    LOG_INFO(
+        "IOHandle::ToggleEvents: fd = %ld "
+        "asserting kOutEvent and disabling\n",
+        fd_);
     write_events_enabled_ = false;
   }
+
+  // If read events are disabled, then remove the kInEvent bit from the event
+  // mask.
   if (!read_events_enabled_) {
-    LOG_INFO("IOHandle::ToggleEvents: fd=%ld de-asserting read\n", fd_);
+    LOG_INFO(
+        "IOHandle::ToggleEvents: fd = %ld "
+        "de-asserting kInEvent\n",
+        fd_);
     event_mask = event_mask & ~(1 << kInEvent);
   }
   // We may get In events without available bytes, so we must make sure there
@@ -283,11 +308,43 @@ intptr_t IOHandle::ToggleEvents(intptr_t event_mask) {
   //
   // As a detail, negative available bytes (errors) are handled specially; see
   // IOHandle::AvailableBytes for more information.
-  if ((event_mask & (1 << kInEvent)) != 0 &&
-      FDUtils::AvailableBytes(fd_) != 0) {
-    LOG_INFO("IOHandle::ToggleEvents: fd = %ld asserting read and disabling\n",
-             fd_);
-    read_events_enabled_ = false;
+  if ((event_mask & (1 << kInEvent)) != 0) {
+    if (FDUtils::AvailableBytes(fd_) != 0) {
+      LOG_INFO(
+          "IOHandle::ToggleEvents: fd = %ld "
+          "asserting kInEvent and disabling with bytes available\n",
+          fd_);
+      read_events_enabled_ = false;
+    }
+    // Also supress future read events if we get a kCloseEvent. This is to
+    // account for POLLIN being set by Fuchsia when the socket is read-closed.
+    if ((event_mask & (1 << kCloseEvent)) != 0) {
+      LOG_INFO(
+          "IOHandle::ToggleEvents: fd = %ld "
+          "asserting kInEvent and disabling due to a close event\n",
+          fd_);
+      read_events_enabled_ = false;
+    }
+  }
+
+  // If the close events are disabled, then remove the kCloseEvent bit from the
+  // event mask.
+  if (!close_events_enabled_) {
+    LOG_INFO(
+        "IOHandle::ToggleEvents: fd = %ld "
+        "de-asserting kCloseEvent\n",
+        fd_);
+    event_mask = event_mask & ~(1 << kCloseEvent);
+  }
+  // If the kCloseEvent bit is set, then supress future close events, they will
+  // be ignored by the Dart thread. See _NativeSocket.multiplex in
+  // socket_patch.dart.
+  if ((event_mask & (1 << kCloseEvent)) != 0) {
+    LOG_INFO(
+        "IOHandle::ToggleEvents: fd = %ld "
+        "asserting kCloseEvent and disabling\n",
+        fd_);
+    close_events_enabled_ = false;
   }
   return event_mask;
 }

runtime/bin/eventhandler_fuchsia.h

@@ -34,6 +34,7 @@ class IOHandle : public ReferenceCounted<IOHandle> {
         mutex_(),
         write_events_enabled_(true),
         read_events_enabled_(true),
+        close_events_enabled_(true),
         fd_(fd),
         handle_(ZX_HANDLE_INVALID),
         wait_key_(0),
@@ -73,6 +74,8 @@ class IOHandle : public ReferenceCounted<IOHandle> {
   Mutex mutex_;
   bool write_events_enabled_;
   bool read_events_enabled_;
+  bool close_events_enabled_;
+
   // Bytes remaining to be read from the socket. Read events should only be
   // re-enabled when this drops to zero.
   intptr_t available_bytes_;