extension: User space event

CODEOWNERS

@@ -162,3 +162,5 @@ extensions/filters/http/oauth2 @rgs1 @derekargueta @snowp
 /*/extensions/filters/http/kill_request @qqustc @htuch
 # Rate limit expression descriptor
 /*/extensions/rate_limit_descriptors/expr @kyessenov @lizan
+# user space socket pair and event
+/*/extensions/io_socket/user_space_io_socket @lambdai @antoniovicente

source/extensions/extensions_build_config.bzl

@@ -220,7 +220,7 @@ EXTENSIONS = {
     #
 
     "envoy.watchdog.profile_action":                    "//source/extensions/watchdog/profile_action:config",
-
+    
     #
     # WebAssembly runtimes
     #
@@ -235,6 +235,12 @@ EXTENSIONS = {
     #
 
     "envoy.rate_limit_descriptors.expr":                "//source/extensions/rate_limit_descriptors/expr:config",
+
+    #
+    # IO socket
+    #
+
+    "envoy.io_socket.user_space_socket":                "//source/extensions/io_socket/user_space_io_socket:user_space_io_socket_handle_lib",
 }
 
 # These can be changed to ["//visibility:public"], for  downstream builds which

source/extensions/io_socket/BUILD

@@ -0,0 +1,8 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_extension_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_extension_package()

source/extensions/io_socket/user_space_io_socket/BUILD

@@ -0,0 +1,37 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_extension",
+    "envoy_extension_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_extension_package()
+
+envoy_cc_extension(
+    name = "peer_buffer_lib",
+    hdrs = ["peer_buffer.h"],
+    security_posture = "unknown",
+    status = "alpha",
+    deps = [
+        "//source/common/buffer:buffer_lib",
+        "//source/common/buffer:watermark_buffer_lib",
+        "//source/common/common:empty_string",
+    ],
+)
+
+envoy_cc_extension(
+    name = "users_space_file_event_lib",
+    srcs = [
+        "user_space_file_event_impl.cc",
+    ],
+    hdrs = [
+        "user_space_file_event_impl.h",
+    ],
+    security_posture = "unknown",
+    status = "alpha",
+    deps = [
+        ":peer_buffer_lib",
+        "//source/common/event:dispatcher_includes",
+    ],
+)

source/extensions/io_socket/user_space_io_socket/peer_buffer.h

@@ -0,0 +1,67 @@
+#pragma once
+
+#include "envoy/buffer/buffer.h"
+#include "envoy/common/pure.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace IoSocket {
+namespace UserSpaceIoSocket {
+
+/**
+ * The interface for the peer as a writer and supplied read status query.
+ */
+class UserspaceIoHandle {
+public:
+  virtual ~UserspaceIoHandle() = default;
+
+  /**
+   * Set the flag to indicate no further write from peer.
+   */
+  virtual void setWriteEnd() PURE;
+
+  /**
+   * @return true if the peer promise no more write.
+   */
+  virtual bool isPeerShutDownWrite() const PURE;
+
+  /**
+   * Raised when peer is destroyed. No further write to peer is allowed.
+   */
+  virtual void onPeerDestroy() PURE;
+
+  /**
+   * Notify that consumable data arrived. The consumable data can be either data to read, or the end
+   * of stream event.
+   */
+  virtual void setNewDataAvailable() PURE;
+
+  /**
+   * @return the buffer to be written.
+   */
+  virtual Buffer::Instance* getWriteBuffer() PURE;
+
+  /**
+   * @return true if more data is acceptable at the destination buffer.
+   */
+  virtual bool isWritable() const PURE;
+
+  /**
+   * @return true if peer is valid and writable.
+   */
+  virtual bool isPeerWritable() const PURE;
+
+  /**
+   * Raised by the peer when the peer switch from high water mark to low.
+   */
+  virtual void onPeerBufferLowWatermark() PURE;
+
+  /**
+   * @return true if the pending receive buffer is not empty or read_end is set.
+   */
+  virtual bool isReadable() const PURE;
+};
+} // namespace UserSpaceIoSocket
+} // namespace IoSocket
+} // namespace Extensions
+} // namespace Envoy

source/extensions/io_socket/user_space_io_socket/user_space_file_event_impl.cc

@@ -0,0 +1,89 @@
+#include "extensions/io_socket/user_space_io_socket/user_space_file_event_impl.h"
+
+#include <cstdint>
+
+#include "common/common/assert.h"
+
+#include "extensions/io_socket/user_space_io_socket/peer_buffer.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace IoSocket {
+namespace UserSpaceIoSocket {
+
+UserSpaceFileEventImpl::UserSpaceFileEventImpl(Event::Dispatcher& dispatcher, Event::FileReadyCb cb,
+                                               uint32_t events, UserspaceIoHandle& io_source)
+    : schedulable_(dispatcher.createSchedulableCallback([this, cb]() {
+        auto ephemeral_events = event_listener_.getAndClearEphemeralEvents();
+        ENVOY_LOG(trace, "User space event {} invokes callbacks on events = {}",
+                  static_cast<void*>(this), ephemeral_events);
+        cb(ephemeral_events);
+      })),
+      io_source_(io_source) {
+  setEnabled(events);
+}
+
+void EventListenerImpl::clearEphemeralEvents() {
+  // Clear ephemeral events to align with FileEventImpl::setEnable().
+  ephemeral_events_ = 0;
+}
+
+void EventListenerImpl::onEventActivated(uint32_t activated_events) {
+  ephemeral_events_ |= activated_events;
+}
+
+void EventListenerImpl::setEnabledEvents(uint32_t enabled_events) {
+  enabled_events_ = enabled_events;
+}
+
+void UserSpaceFileEventImpl::activate(uint32_t events) {
+  // Only supported event types are set.
+  ASSERT((events & (Event::FileReadyType::Read | Event::FileReadyType::Write |
+                    Event::FileReadyType::Closed)) == events);
+  event_listener_.onEventActivated(events);
+  schedulable_->scheduleCallbackNextIteration();
+}
+
+void UserSpaceFileEventImpl::setEnabled(uint32_t events) {
+  // Only supported event types are set.
+  ASSERT((events & (Event::FileReadyType::Read | Event::FileReadyType::Write |
+                    Event::FileReadyType::Closed)) == events);
+  event_listener_.clearEphemeralEvents();
+  event_listener_.setEnabledEvents(events);
+  bool was_enabled = schedulable_->enabled();
+  // Recalculate activated events.
+  uint32_t events_to_notify = 0;
+  if ((events & Event::FileReadyType::Read) && io_source_.isReadable()) {
+    events_to_notify |= Event::FileReadyType::Read;
+  }
+  if ((events & Event::FileReadyType::Write) && io_source_.isPeerWritable()) {
+    events_to_notify |= Event::FileReadyType::Write;
+  }
+  if ((events & Event::FileReadyType::Closed) && io_source_.isPeerShutDownWrite()) {
+    events_to_notify |= Event::FileReadyType::Closed;
+  }
+  if (events_to_notify != 0) {
+    activate(events_to_notify);
+  } else {
+    schedulable_->cancel();
+  }
+  ENVOY_LOG(
+      trace,
+      "User space file event {} set enabled events {} and events {} is active. Will {} reschedule.",
+      static_cast<void*>(this), events, was_enabled ? "not " : "");
+}
+
+void UserSpaceFileEventImpl::poll(uint32_t events) {
+  ASSERT((events & (Event::FileReadyType::Read | Event::FileReadyType::Write |
+                    Event::FileReadyType::Closed)) == events);
+  // filtered out disabled events.
+  uint32_t filter_enabled = events & event_listener_.getEnabledEvents();
+  if (filter_enabled == 0) {
+    return;
+  }
+  activate(filter_enabled);
+}
+} // namespace UserSpaceIoSocket
+} // namespace IoSocket
+} // namespace Extensions
+} // namespace Envoy

source/extensions/io_socket/user_space_io_socket/user_space_file_event_impl.h

@@ -0,0 +1,74 @@
+#pragma once
+
+#include <cstdint>
+
+#include "envoy/event/file_event.h"
+
+#include "common/event/dispatcher_impl.h"
+#include "common/event/event_impl_base.h"
+
+#include "extensions/io_socket/user_space_io_socket/peer_buffer.h"
+
+namespace Envoy {
+
+namespace Extensions {
+namespace IoSocket {
+namespace UserSpaceIoSocket {
+
+// This class maintains the ephemeral events and enabled events.
+// getAndClearEphemeralEvents
+class EventListenerImpl {
+public:
+  ~EventListenerImpl() = default;
+
+  // Reset the enabled events. The caller must refresh the triggered events.
+  void setEnabledEvents(uint32_t enabled_events);
+  // Return the enabled events.
+  uint32_t getEnabledEvents() { return enabled_events_; }
+
+  void clearEphemeralEvents();
+  void onEventActivated(uint32_t activated_events);
+
+  uint32_t getAndClearEphemeralEvents() { return std::exchange(ephemeral_events_, 0); }
+
+private:
+  // The events set by activate() and will be cleared after the io callback.
+  uint32_t ephemeral_events_{};
+  // The events set by setEnabled(). The new value replaces the old value.
+  uint32_t enabled_events_{};
+};
+
+// A FileEvent implementation which is used to drive UserSpaceIoSocketHandle.
+// Declare the class final to safely call virtual function setEnabled in constructor.
+class UserSpaceFileEventImpl final : public Event::FileEvent, Logger::Loggable<Logger::Id::io> {
+public:
+  UserSpaceFileEventImpl(Event::Dispatcher& dispatcher, Event::FileReadyCb cb, uint32_t events,
+                         UserspaceIoHandle& io_source);
+
+  // Event::FileEvent
+  void activate(uint32_t events) override;
+  void setEnabled(uint32_t events) override;
+
+  // `UserspaceFileEvent` acts always as edge triggered regardless the underlying OS is level or
+  // edge triggered. The event owner on windows platform should not emulate edge events.
+  void unregisterEventIfEmulatedEdge(uint32_t) override {}
+  void registerEventIfEmulatedEdge(uint32_t) override {}
+
+  // Notify events. Unlike activate() method, this method activates the given events only if the
+  // events are enabled.
+  void poll(uint32_t events);
+
+private:
+  // Used to populate the event operations of enable and activate.
+  EventListenerImpl event_listener_;
+
+  // The handle to registered async callback from dispatcher.
+  Event::SchedulableCallbackPtr schedulable_;
+
+  // Supplies readable and writable status.
+  UserspaceIoHandle& io_source_;
+};
+} // namespace UserSpaceIoSocket
+} // namespace IoSocket
+} // namespace Extensions
+} // namespace Envoy

test/extensions/io_socket/user_space_io_socket/BUILD

@@ -0,0 +1,29 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_package",
+)
+load(
+    "//test/extensions:extensions_build_system.bzl",
+    "envoy_extension_cc_test",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_package()
+
+envoy_extension_cc_test(
+    name = "user_space_file_event_impl_test",
+    srcs = ["user_space_file_event_impl_test.cc"],
+    extension_name = "envoy.io_socket.user_space_socket",
+    deps = [
+        "//include/envoy/event:file_event_interface",
+        "//source/common/event:dispatcher_includes",
+        "//source/common/event:dispatcher_lib",
+        "//source/extensions/io_socket/user_space_io_socket:peer_buffer_lib",
+        "//source/extensions/io_socket/user_space_io_socket:users_space_file_event_lib",
+        "//test/mocks:common_lib",
+        "//test/test_common:environment_lib",
+        "//test/test_common:test_runtime_lib",
+        "//test/test_common:utility_lib",
+    ],
+)