[Level Events] manage level events registration mask

include/envoy/api/io_error.h

@@ -71,6 +71,12 @@ template <typename ReturnValue> struct IoCallResult {
    */
   bool ok() const { return err_ == nullptr; }
 
+  /**
+   * This return code is frequent enough that we have a separate function to check.
+   * @return true if the system call failed because the socket would block.
+   */
+  bool wouldBlock() const { return !ok() && err_->getErrorCode() == IoError::IoErrorCode::Again; }
+
   // TODO(danzh): rename it to be more meaningful, i.e. return_value_.
   ReturnValue rc_;
   IoErrorPtr err_;

include/envoy/event/file_event.h

@@ -18,15 +18,27 @@ struct FileReadyType {
   static const uint32_t Closed = 0x4;
 };
 
-enum class FileTriggerType { Level, Edge };
+enum class FileTriggerType { Level, Edge, EmulatedEdge };
 
-static constexpr FileTriggerType PlatformDefaultTriggerType
-#ifdef WIN32
-    // Libevent only supports Level trigger on Windows.
-    {FileTriggerType::Level};
+// For POSIX developers to get the Windows behavior of file events
+// you need to add the following definitions:
+// 1. `FORCE_LEVEL_EVENTS`
+// 2. `OPTIMIZE_LEVEL_EVENTS`.
+// You can do this with bazel if you add the following build/test options
+// `--copt="-DFORCE_LEVEL_EVENTS" --copt="-DOPTIMIZE_LEVEL_EVENTS"`
+constexpr FileTriggerType determinePlatformPreferredEventType() {
+#if defined(WIN32) || defined(FORCE_LEVEL_EVENTS)
+#ifdef OPTIMIZE_LEVEL_EVENTS
+  return FileTriggerType::EmulatedEdge;
 #else
-    {FileTriggerType::Edge};
+  return FileTriggerType::Level;
 #endif
+#else
+  return FileTriggerType::Edge;
+#endif
+}
+
+static constexpr FileTriggerType PlatformDefaultTriggerType = determinePlatformPreferredEventType();
 
 /**
  * Callback invoked when a FileEvent is ready for reading or writing.
@@ -53,6 +65,16 @@ class FileEvent {
    * registered events and fire callbacks when they are active.
    */
   virtual void setEnabled(uint32_t events) PURE;
+
+  /**
+   * Add a single event from the event registration mark.
+   */
+  virtual void registerEventIfEmulatedEdge(uint32_t event) PURE;
+
+  /**
+   * Remove a single event from the event registration mark.
+   */
+  virtual void unregisterEventIfEmulatedEdge(uint32_t event) PURE;
 };
 
 using FileEventPtr = std::unique_ptr<FileEvent>;

source/common/event/file_event_impl.cc

@@ -13,7 +13,7 @@ namespace Event {
 
 FileEventImpl::FileEventImpl(DispatcherImpl& dispatcher, os_fd_t fd, FileReadyCb cb,
                              FileTriggerType trigger, uint32_t events)
-    : cb_(cb), fd_(fd), trigger_(trigger),
+    : cb_(cb), fd_(fd), trigger_(trigger), enabled_events_(events),
       activate_fd_events_next_event_loop_(
           // Only read the runtime feature if the runtime loader singleton has already been created.
           // Attempts to access runtime features too early in the initialization sequence triggers
@@ -28,9 +28,13 @@ FileEventImpl::FileEventImpl(DispatcherImpl& dispatcher, os_fd_t fd, FileReadyCb
   // an OOM condition and just crash.
   RELEASE_ASSERT(SOCKET_VALID(fd), "");
 #ifdef WIN32
-  RELEASE_ASSERT(trigger_ == FileTriggerType::Level,
-                 "libevent does not support edge triggers on Windows");
+  ASSERT(trigger_ != FileTriggerType::Edge, "libevent does not support edge triggers on Windows");
 #endif
+  if constexpr (PlatformDefaultTriggerType != FileTriggerType::EmulatedEdge) {
+    ASSERT(trigger_ != FileTriggerType::EmulatedEdge,
+           "Cannot use EmulatedEdge events if they are not the default platform type");
+  }
+
   assignEvents(events, &dispatcher.base());
   event_add(&raw_event_, nullptr);
   if (activate_fd_events_next_event_loop_) {
@@ -81,9 +85,10 @@ void FileEventImpl::activate(uint32_t events) {
 
 void FileEventImpl::assignEvents(uint32_t events, event_base* base) {
   ASSERT(base != nullptr);
+  enabled_events_ = events;
   event_assign(
       &raw_event_, base, fd_,
-      EV_PERSIST | (trigger_ == FileTriggerType::Level ? 0 : EV_ET) |
+      EV_PERSIST | (trigger_ == FileTriggerType::Edge ? EV_ET : 0) |
           (events & FileReadyType::Read ? EV_READ : 0) |
           (events & FileReadyType::Write ? EV_WRITE : 0) |
           (events & FileReadyType::Closed ? EV_CLOSED : 0),
@@ -108,6 +113,16 @@ void FileEventImpl::assignEvents(uint32_t events, event_base* base) {
       this);
 }
 
+void FileEventImpl::updateEvents(uint32_t events) {
+  if (events == enabled_events_) {
+    return;
+  }
+  auto* base = event_get_base(&raw_event_);
+  event_del(&raw_event_);
+  assignEvents(events, base);
+  event_add(&raw_event_, nullptr);
+}
+
 void FileEventImpl::setEnabled(uint32_t events) {
   if (activate_fd_events_next_event_loop_ && injected_activation_events_ != 0) {
     // Clear pending events on updates to the fd event mask to avoid delivering events that are no
@@ -117,19 +132,64 @@ void FileEventImpl::setEnabled(uint32_t events) {
     injected_activation_events_ = 0;
     activation_cb_->cancel();
   }
+  updateEvents(events);
+}
 
-  auto* base = event_get_base(&raw_event_);
-  event_del(&raw_event_);
-  assignEvents(events, base);
-  event_add(&raw_event_, nullptr);
+void FileEventImpl::unregisterEventIfEmulatedEdge(uint32_t event) {
+  // This constexpr if allows the compiler to optimize away the function on POSIX
+  if constexpr (PlatformDefaultTriggerType == FileTriggerType::EmulatedEdge) {
+    ASSERT((event & (FileReadyType::Read | FileReadyType::Write)) == event);
+    if (trigger_ == FileTriggerType::EmulatedEdge) {
+      auto new_event_mask = enabled_events_ & ~event;
+      updateEvents(new_event_mask);
+    }
+  }
+}
+
+void FileEventImpl::registerEventIfEmulatedEdge(uint32_t event) {
+  // This constexpr if allows the compiler to optimize away the function on POSIX
+  if constexpr (PlatformDefaultTriggerType == FileTriggerType::EmulatedEdge) {
+    ASSERT((event & (FileReadyType::Read | FileReadyType::Write)) == event);
+    if (trigger_ == FileTriggerType::EmulatedEdge) {
+      auto new_event_mask = enabled_events_ | event;
+      if (event & FileReadyType::Read && (enabled_events_ & FileReadyType::Closed)) {
+        // We never ask for both early close and read at the same time.
+        new_event_mask = new_event_mask & ~FileReadyType::Read;
+      }
+      if (event == 0) {
+        return;
+      }
+      updateEvents(new_event_mask);
+    }
+  }
 }
 
 void FileEventImpl::mergeInjectedEventsAndRunCb(uint32_t events) {
   if (activate_fd_events_next_event_loop_ && injected_activation_events_ != 0) {
+    // TODO(antoniovicente) remove this adjustment to activation events once ConnectionImpl can
+    // handle Read and Close events delivered together.
+    if constexpr (PlatformDefaultTriggerType == FileTriggerType::EmulatedEdge) {
+      if (events & FileReadyType::Closed &&
+          activate_fd_events_next_event_loop_ & FileReadyType::Read) {
+        // We never ask for both early close and read at the same time. If close is requested
+        // keep that instead.
+        injected_activation_events_ = injected_activation_events_ & ~FileReadyType::Read;
+      }
+    }
+
     events |= injected_activation_events_;
     injected_activation_events_ = 0;
     activation_cb_->cancel();
   }
+
+  // TODO(davinci26): This can be optimized further in (w)epoll backends using the `EPOLLONESHOT`
+  // flag. With this flag `EPOLLIN`/`EPOLLOUT` are automatically disabled when the event is
+  // activated.
+  if (trigger_ == FileTriggerType::EmulatedEdge) {
+    unregisterEventIfEmulatedEdge(events &
+                                  (Event::FileReadyType::Write | Event::FileReadyType::Read));
+  }
+
   cb_(events);
 }
 

source/common/event/file_event_impl.h

@@ -22,14 +22,19 @@ class FileEventImpl : public FileEvent, ImplBase {
   // Event::FileEvent
   void activate(uint32_t events) override;
   void setEnabled(uint32_t events) override;
+  void unregisterEventIfEmulatedEdge(uint32_t event) override;
+  void registerEventIfEmulatedEdge(uint32_t event) override;
 
 private:
   void assignEvents(uint32_t events, event_base* base);
   void mergeInjectedEventsAndRunCb(uint32_t events);
+  void updateEvents(uint32_t events);
 
   FileReadyCb cb_;
   os_fd_t fd_;
   FileTriggerType trigger_;
+  // Enabled events for this fd.
+  uint32_t enabled_events_;
 
   // Injected FileReadyType events that were scheduled by recent calls to activate() and are pending
   // delivery.

source/common/event/libevent_scheduler.h

@@ -110,10 +110,9 @@ class LibeventScheduler : public Scheduler, public CallbackScheduler {
 
   static constexpr int flagsBasedOnEventType() {
     if constexpr (Event::PlatformDefaultTriggerType == FileTriggerType::Level) {
-      // On Windows, EVLOOP_NONBLOCK will cause the libevent event_base_loop to run forever.
-      // This is because libevent only supports level triggering on Windows, and so the write
-      // event callbacks will trigger every time through the loop. Adding EVLOOP_ONCE ensures the
-      // loop will run at most once
+      // With level events, EVLOOP_NONBLOCK will cause the libevent event_base_loop to run
+      // forever. This is because the write event callbacks will trigger every time through the
+      // loop. Adding EVLOOP_ONCE ensures the loop will run at most once
       return EVLOOP_NONBLOCK | EVLOOP_ONCE;
     }
     return EVLOOP_NONBLOCK;

source/common/network/io_socket_handle_impl.cc

@@ -84,8 +84,14 @@ Api::IoCallUint64Result IoSocketHandleImpl::readv(uint64_t max_length, Buffer::R
     num_bytes_to_read += slice_length;
   }
   ASSERT(num_bytes_to_read <= max_length);
-  return sysCallResultToIoCallResult(Api::OsSysCallsSingleton::get().readv(
+  auto result = sysCallResultToIoCallResult(Api::OsSysCallsSingleton::get().readv(
       fd_, iov.begin(), static_cast<int>(num_slices_to_read)));
+
+  // Some tests try to read without initializing the file_event.
+  if (result.wouldBlock() && file_event_) {
+    file_event_->registerEventIfEmulatedEdge(Event::FileReadyType::Read);
+  }
+  return result;
 }
 
 Api::IoCallUint64Result IoSocketHandleImpl::read(Buffer::Instance& buffer, uint64_t max_length) {
@@ -103,6 +109,11 @@ Api::IoCallUint64Result IoSocketHandleImpl::read(Buffer::Instance& buffer, uint6
     bytes_to_commit -= slices[i].len_;
   }
   buffer.commit(slices, num_slices);
+
+  // Some tests try to read without initializing the file_event.
+  if (result.wouldBlock() && file_event_) {
+    file_event_->registerEventIfEmulatedEdge(Event::FileReadyType::Read);
+  }
   return result;
 }
 
@@ -120,8 +131,14 @@ Api::IoCallUint64Result IoSocketHandleImpl::writev(const Buffer::RawSlice* slice
   if (num_slices_to_write == 0) {
     return Api::ioCallUint64ResultNoError();
   }
-  return sysCallResultToIoCallResult(
+  auto result = sysCallResultToIoCallResult(
       Api::OsSysCallsSingleton::get().writev(fd_, iov.begin(), num_slices_to_write));
+
+  // Some tests try to write without initializing the file_event.
+  if (result.wouldBlock() && file_event_) {
+    file_event_->registerEventIfEmulatedEdge(Event::FileReadyType::Write);
+  }
+  return result;
 }
 
 Api::IoCallUint64Result IoSocketHandleImpl::write(Buffer::Instance& buffer) {
@@ -131,6 +148,11 @@ Api::IoCallUint64Result IoSocketHandleImpl::write(Buffer::Instance& buffer) {
   if (result.ok() && result.rc_ > 0) {
     buffer.drain(static_cast<uint64_t>(result.rc_));
   }
+
+  // Some tests try to read without initializing the file_event.
+  if (result.wouldBlock() && file_event_) {
+    file_event_->registerEventIfEmulatedEdge(Event::FileReadyType::Write);
+  }
   return result;
 }
 
@@ -168,7 +190,11 @@ Api::IoCallUint64Result IoSocketHandleImpl::sendmsg(const Buffer::RawSlice* slic
     message.msg_control = nullptr;
     message.msg_controllen = 0;
     const Api::SysCallSizeResult result = os_syscalls.sendmsg(fd_, &message, flags);
-    return sysCallResultToIoCallResult(result);
+    auto io_result = sysCallResultToIoCallResult(result);
+    if (io_result.wouldBlock() && file_event_) {
+      file_event_->registerEventIfEmulatedEdge(Event::FileReadyType::Write);
+    }
+    return io_result;
   } else {
     const size_t space_v6 = CMSG_SPACE(sizeof(in6_pktinfo));
     const size_t space_v4 = CMSG_SPACE(sizeof(in_pktinfo));
@@ -210,7 +236,11 @@ Api::IoCallUint64Result IoSocketHandleImpl::sendmsg(const Buffer::RawSlice* slic
       *(reinterpret_cast<absl::uint128*>(pktinfo->ipi6_addr.s6_addr)) = self_ip->ipv6()->address();
     }
     const Api::SysCallSizeResult result = os_syscalls.sendmsg(fd_, &message, flags);
-    return sysCallResultToIoCallResult(result);
+    auto io_result = sysCallResultToIoCallResult(result);
+    if (io_result.wouldBlock() && file_event_) {
+      file_event_->registerEventIfEmulatedEdge(Event::FileReadyType::Write);
+    }
+    return io_result;
   }
 }
 
@@ -298,7 +328,11 @@ Api::IoCallUint64Result IoSocketHandleImpl::recvmsg(Buffer::RawSlice* slices,
   hdr.msg_controllen = cmsg_space_;
   const Api::SysCallSizeResult result = Api::OsSysCallsSingleton::get().recvmsg(fd_, &hdr, 0);
   if (result.rc_ < 0) {
-    return sysCallResultToIoCallResult(result);
+    auto io_result = sysCallResultToIoCallResult(result);
+    if (io_result.wouldBlock() && file_event_) {
+      file_event_->registerEventIfEmulatedEdge(Event::FileReadyType::Read);
+    }
+    return io_result;
   }
 
   RELEASE_ASSERT((hdr.msg_flags & MSG_CTRUNC) == 0,
@@ -381,7 +415,11 @@ Api::IoCallUint64Result IoSocketHandleImpl::recvmmsg(RawSliceArrays& slices, uin
       fd_, mmsg_hdr.data(), num_packets_per_mmsg_call, MSG_TRUNC | MSG_WAITFORONE, nullptr);
 
   if (result.rc_ <= 0) {
-    return sysCallResultToIoCallResult(result);
+    auto io_result = sysCallResultToIoCallResult(result);
+    if (io_result.wouldBlock() && file_event_) {
+      file_event_->registerEventIfEmulatedEdge(Event::FileReadyType::Read);
+    }
+    return io_result;
   }
 
   int num_packets_read = result.rc_;
@@ -435,7 +473,11 @@ Api::IoCallUint64Result IoSocketHandleImpl::recvmmsg(RawSliceArrays& slices, uin
 Api::IoCallUint64Result IoSocketHandleImpl::recv(void* buffer, size_t length, int flags) {
   const Api::SysCallSizeResult result =
       Api::OsSysCallsSingleton::get().recv(fd_, buffer, length, flags);
-  return sysCallResultToIoCallResult(result);
+  auto io_result = sysCallResultToIoCallResult(result);
+  if (io_result.wouldBlock() && file_event_) {
+    file_event_->registerEventIfEmulatedEdge(Event::FileReadyType::Read);
+  }
+  return io_result;
 }
 
 bool IoSocketHandleImpl::supportsMmsg() const {

source/extensions/transport_sockets/tls/ssl_socket.cc

@@ -293,6 +293,17 @@ void SslSocket::shutdownSsl() {
   if (info_->state() != Ssl::SocketState::ShutdownSent &&
       callbacks_->connection().state() != Network::Connection::State::Closed) {
     int rc = SSL_shutdown(rawSsl());
+    if constexpr (Event::PlatformDefaultTriggerType == Event::FileTriggerType::EmulatedEdge) {
+      // Windows operate under `EmulatedEdge`. These are level events that are artificially
+      // made to behave like edge events. And if the rc is 0 then in that case we want read
+      // activation resumption. This code is protected with an `constexpr` if, to minimize the tax
+      // on POSIX systems that operate in Edge events.
+      if (rc == 0) {
+        // See https://www.openssl.org/docs/manmaster/man3/SSL_shutdown.html
+        // if return value is 0,  Call SSL_read() to do a bidirectional shutdown.
+        callbacks_->setReadBufferReady();
+      }
+    }
     ENVOY_CONN_LOG(debug, "SSL shutdown: rc={}", callbacks_->connection(), rc);
     drainErrorQueue();
     info_->setState(Ssl::SocketState::ShutdownSent);

source/extensions/transport_sockets/tls/ssl_socket.h

@@ -70,8 +70,6 @@ class SslSocket : public Network::TransportSocket,
   void onFailure() override;
   Network::TransportSocketCallbacks* transportSocketCallbacks() override { return callbacks_; }
 
-  SSL* rawSslForTest() const { return rawSsl(); }
-
 protected:
   SSL* rawSsl() const { return info_->ssl_.get(); }
 

source/server/hot_restarting_parent.cc

@@ -29,7 +29,7 @@ void HotRestartingParent::initialize(Event::Dispatcher& dispatcher, Server::Inst
         ASSERT(events == Event::FileReadyType::Read);
         onSocketEvent();
       },
-      Event::PlatformDefaultTriggerType, Event::FileReadyType::Read);
+      Event::FileTriggerType::Edge, Event::FileReadyType::Read);
   internal_ = std::make_unique<Internal>(&server);
 }
 

test/common/event/BUILD

@@ -11,7 +11,6 @@ envoy_package()
 envoy_cc_test(
     name = "dispatcher_impl_test",
     srcs = ["dispatcher_impl_test.cc"],
-    tags = ["fails_on_windows"],
     deps = [
         "//source/common/api:api_lib",
         "//source/common/event:deferred_task",
@@ -30,7 +29,6 @@ envoy_cc_test(
 envoy_cc_test(
     name = "file_event_impl_test",
     srcs = ["file_event_impl_test.cc"],
-    tags = ["fails_on_windows"],
     deps = [
         "//include/envoy/event:file_event_interface",
         "//source/common/event:dispatcher_includes",