Io socket handle for internal socket

source/common/event/file_event_impl.h

@@ -8,6 +8,10 @@
 #include "common/event/event_impl_base.h"
 
 namespace Envoy {
+
+namespace Network {
+class BufferedIoSocketHandleImpl;
+}
 namespace Event {
 
 /**
@@ -41,5 +45,119 @@ class FileEventImpl : public FileEvent, ImplBase {
   // polling and activating new fd events.
   const bool activate_fd_events_next_event_loop_;
 };
+
+// Forward declare for friend class.
+class UserSpaceFileEventFactory;
+
+// The interface of populating event watcher and obtaining the active events. The events includes
+// Read, Write and Closed.
+class EventListener {
+public:
+  virtual ~EventListener() = default;
+
+  // Provide the activated events.
+  virtual uint32_t triggeredEvents() PURE;
+  virtual uint32_t getAndClearEphemeralEvents() PURE;
+
+  // Callbacks of the event operation.
+  virtual void onEventEnabled(uint32_t enabled_events) PURE;
+  virtual void onEventActivated(uint32_t activated_events) PURE;
+};
+
+// Return the enabled events except EV_CLOSED. This implementation is generally good since only
+// epoll supports EV_CLOSED but the entire envoy code base supports another poller. The event owner
+// must assume EV_CLOSED is never activated. Also event owner must tolerate that OS could notify
+// events which are not actually triggered.
+class DefaultEventListener : public EventListener {
+public:
+  ~DefaultEventListener() override = default;
+
+  // Return both read and write.
+  uint32_t triggeredEvents() override { return pending_events_ & (~Event::FileReadyType::Closed); }
+
+  void onEventEnabled(uint32_t enabled_events) override { pending_events_ = enabled_events; }
+
+  void onEventActivated(uint32_t activated_events) override {
+    ephermal_events_ |= activated_events;
+  }
+
+  uint32_t getAndClearEphemeralEvents() override {
+    auto res = ephermal_events_;
+    ephermal_events_ = 0;
+    return res;
+  }
+
+private:
+  // The persisted interested events and ready events.
+  uint32_t pending_events_{};
+  // The events set by activate() and will be cleared after the io callback.
+  uint32_t ephermal_events_{};
+};
+
+// A FileEvent implementation which is
+class UserSpaceFileEventImpl : public FileEvent, Logger::Loggable<Logger::Id::io> {
+public:
+  ~UserSpaceFileEventImpl() override {
+    if (schedulable_.enabled()) {
+      schedulable_.cancel();
+    }
+    ASSERT(event_counter_ == 1);
+    --event_counter_;
+  }
+
+  // Event::FileEvent
+  void activate(uint32_t events) override {
+    event_listener_.onEventActivated(events);
+    if (!schedulable_.enabled()) {
+      schedulable_.scheduleCallbackNextIteration();
+    }
+  }
+
+  void setEnabled(uint32_t events) override {
+    event_listener_.onEventEnabled(events);
+    bool was_enabled = schedulable_.enabled();
+    if (!was_enabled) {
+      schedulable_.scheduleCallbackNextIteration();
+    }
+    ENVOY_LOG(trace, "User space file event {} setEnabled {} on {}. Will {} reschedule.",
+              static_cast<void*>(this), was_enabled ? "not " : "");
+  }
+
+  EventListener& getEventListener() { return event_listener_; }
+  void onEvents() { cb_(); }
+  friend class UserSpaceFileEventFactory;
+  friend class Network::BufferedIoSocketHandleImpl;
+
+private:
+  UserSpaceFileEventImpl(Event::FileReadyCb cb, uint32_t events,
+                         SchedulableCallback& schedulable_cb, int& event_counter)
+      : schedulable_(schedulable_cb), cb_([this, cb]() {
+          auto all_events = getEventListener().triggeredEvents();
+          auto ephemeral_events = getEventListener().getAndClearEphemeralEvents();
+          ENVOY_LOG(trace,
+                    "User space event {} invokes callbacks on allevents = {}, ephermal events = {}",
+                    static_cast<void*>(this), all_events, ephemeral_events);
+          cb(all_events | ephemeral_events);
+        }),
+        event_counter_(event_counter) {
+    event_listener_.onEventEnabled(events);
+  }
+  DefaultEventListener event_listener_;
+  SchedulableCallback& schedulable_;
+  std::function<void()> cb_;
+  int& event_counter_;
+};
+
+class UserSpaceFileEventFactory {
+public:
+  static std::unique_ptr<UserSpaceFileEventImpl>
+  createUserSpaceFileEventImpl(Event::Dispatcher&, Event::FileReadyCb cb, Event::FileTriggerType,
+                               uint32_t events, SchedulableCallback& scheduable_cb,
+                               int& event_counter) {
+    return std::unique_ptr<UserSpaceFileEventImpl>(
+        new UserSpaceFileEventImpl(cb, events, scheduable_cb, event_counter));
+  }
+};
+
 } // namespace Event
 } // namespace Envoy

source/common/network/BUILD

@@ -166,6 +166,7 @@ envoy_cc_library(
     hdrs = ["io_socket_error_impl.h"],
     deps = [
         "//include/envoy/api:io_error_interface",
+        "//include/envoy/api:os_sys_calls_interface",
         "//source/common/common:assert_lib",
         "//source/common/common:utility_lib",
     ],
@@ -469,3 +470,23 @@ envoy_cc_library(
         "//source/common/common:macros",
     ],
 )
+
+envoy_cc_library(
+    name = "peer_buffer_lib",
+    hdrs = ["peer_buffer.h"],
+    deps = [
+        "//source/common/buffer:buffer_lib",
+        "//source/common/buffer:watermark_buffer_lib",
+        "//source/common/common:empty_string",
+    ],
+)
+
+envoy_cc_library(
+    name = "buffered_io_socket_handle_lib",
+    srcs = ["buffered_io_socket_handle_impl.cc"],
+    hdrs = ["buffered_io_socket_handle_impl.h"],
+    deps = [
+        "default_socket_interface_lib",
+        ":peer_buffer_lib",
+    ],
+)

source/common/network/buffered_io_socket_handle_impl.cc

@@ -0,0 +1,228 @@
+#include "common/network/buffered_io_socket_handle_impl.h"
+
+#include "envoy/buffer/buffer.h"
+#include "envoy/common/platform.h"
+
+#include "common/api/os_sys_calls_impl.h"
+#include "common/common/assert.h"
+#include "common/common/utility.h"
+#include "common/event/file_event_impl.h"
+#include "common/network/address_impl.h"
+
+#include "absl/container/fixed_array.h"
+#include "absl/types/optional.h"
+
+namespace Envoy {
+namespace Network {
+
+Api::IoCallUint64Result BufferedIoSocketHandleImpl::close() {
+  ASSERT(!closed_);
+  if (!write_shutdown_) {
+    ASSERT(writable_peer_);
+    // Notify the peer we won't write more data. shutdown(WRITE).
+    writable_peer_->setWriteEnd();
+    // Notify the peer that we no longer accept data. shutdown(RD).
+    writable_peer_->onPeerDestroy();
+    writable_peer_->maybeSetNewData();
+    writable_peer_ = nullptr;
+    write_shutdown_ = true;
+  }
+  closed_ = true;
+  return Api::ioCallUint64ResultNoError();
+}
+
+bool BufferedIoSocketHandleImpl::isOpen() const { return !closed_; }
+
+Api::IoCallUint64Result BufferedIoSocketHandleImpl::readv(uint64_t max_length,
+                                                          Buffer::RawSlice* slices,
+                                                          uint64_t num_slice) {
+  if (owned_buffer_.length() == 0) {
+    if (read_end_stream_) {
+      return {0, Api::IoErrorPtr(nullptr, [](Api::IoError*) {})};
+    } else {
+      return {0, Api::IoErrorPtr(IoSocketError::getIoSocketEagainInstance(),
+                                 IoSocketError::deleteIoError)};
+    }
+  } else {
+    absl::FixedArray<iovec> iov(num_slice);
+    uint64_t num_slices_to_read = 0;
+    uint64_t num_bytes_to_read = 0;
+    for (; num_slices_to_read < num_slice && num_bytes_to_read < max_length; num_slices_to_read++) {
+      auto min_len = std::min(std::min(owned_buffer_.length(), max_length) - num_bytes_to_read,
+                              uint64_t(slices[num_slices_to_read].len_));
+      owned_buffer_.copyOut(num_bytes_to_read, min_len, slices[num_slices_to_read].mem_);
+      num_bytes_to_read += min_len;
+    }
+    ASSERT(num_bytes_to_read <= max_length);
+    owned_buffer_.drain(num_bytes_to_read);
+    ENVOY_LOG(trace, "socket {} readv {} bytes", static_cast<void*>(this), num_bytes_to_read);
+    return {num_bytes_to_read, Api::IoErrorPtr(nullptr, [](Api::IoError*) {})};
+  }
+}
+
+Api::IoCallUint64Result BufferedIoSocketHandleImpl::read(Buffer::Instance& buffer,
+                                                         uint64_t max_length) {
+  if (owned_buffer_.length() == 0) {
+    if (read_end_stream_) {
+      return {0, Api::IoErrorPtr(nullptr, [](Api::IoError*) {})};
+    } else {
+      return {0, Api::IoErrorPtr(IoSocketError::getIoSocketEagainInstance(),
+                                 IoSocketError::deleteIoError)};
+    }
+  } else {
+    // TODO(lambdai): Move at slice boundary to move to reduce the copy.
+    uint64_t min_len = std::min(max_length, owned_buffer_.length());
+    buffer.move(owned_buffer_, min_len);
+    return {min_len, Api::IoErrorPtr(nullptr, [](Api::IoError*) {})};
+  }
+}
+
+Api::IoCallUint64Result BufferedIoSocketHandleImpl::writev(const Buffer::RawSlice* slices,
+                                                           uint64_t num_slice) {
+  if (!writable_peer_) {
+    return sysCallResultToIoCallResult(Api::SysCallSizeResult{-1, SOCKET_ERROR_INTR});
+  }
+  if (writable_peer_->isWriteEndSet() || !writable_peer_->isWritable()) {
+    return {0, Api::IoErrorPtr(IoSocketError::getIoSocketEagainInstance(),
+                               IoSocketError::deleteIoError)};
+  }
+  // Write along with iteration. Buffer guarantee the fragment is always append-able.
+  uint64_t num_bytes_to_write = 0;
+  for (uint64_t i = 0; i < num_slice; i++) {
+    if (slices[i].mem_ != nullptr && slices[i].len_ != 0) {
+      writable_peer_->getWriteBuffer()->add(slices[i].mem_, slices[i].len_);
+      num_bytes_to_write += slices[i].len_;
+    }
+  }
+  writable_peer_->maybeSetNewData();
+  ENVOY_LOG(trace, "socket {} writev {} bytes", static_cast<void*>(this), num_bytes_to_write);
+  return {num_bytes_to_write, Api::IoErrorPtr(nullptr, [](Api::IoError*) {})};
+}
+
+Api::IoCallUint64Result BufferedIoSocketHandleImpl::write(Buffer::Instance& buffer) {
+  if (!writable_peer_) {
+    return sysCallResultToIoCallResult(Api::SysCallSizeResult{-1, SOCKET_ERROR_INTR});
+  }
+  if (writable_peer_->isWriteEndSet() || !writable_peer_->isWritable()) {
+    return {0, Api::IoErrorPtr(IoSocketError::getIoSocketEagainInstance(),
+                               IoSocketError::deleteIoError)};
+  }
+  uint64_t num_bytes_to_write = buffer.length();
+  writable_peer_->getWriteBuffer()->move(buffer);
+  writable_peer_->maybeSetNewData();
+  ENVOY_LOG(trace, "socket {} writev {} bytes", static_cast<void*>(this), num_bytes_to_write);
+  return {num_bytes_to_write, Api::IoErrorPtr(nullptr, [](Api::IoError*) {})};
+}
+
+Api::IoCallUint64Result BufferedIoSocketHandleImpl::sendmsg(const Buffer::RawSlice*, uint64_t, int,
+                                                            const Address::Ip*,
+                                                            const Address::Instance&) {
+  return IoSocketError::ioResultSocketInvalidAddress();
+}
+
+Api::IoCallUint64Result BufferedIoSocketHandleImpl::recvmsg(Buffer::RawSlice*, const uint64_t,
+                                                            uint32_t, RecvMsgOutput&) {
+  return IoSocketError::ioResultSocketInvalidAddress();
+}
+
+Api::IoCallUint64Result BufferedIoSocketHandleImpl::recvmmsg(RawSliceArrays&, uint32_t,
+                                                             RecvMsgOutput&) {
+  return IoSocketError::ioResultSocketInvalidAddress();
+}
+
+Api::IoCallUint64Result BufferedIoSocketHandleImpl::recv(void* buffer, size_t length, int flags) {
+  // No data and the writer closed.
+  if (owned_buffer_.length() == 0) {
+    if (read_end_stream_) {
+      return sysCallResultToIoCallResult(Api::SysCallSizeResult{-1, SOCKET_ERROR_INTR});
+    } else {
+      return {0, Api::IoErrorPtr(IoSocketError::getIoSocketEagainInstance(),
+                                 IoSocketError::deleteIoError)};
+    }
+  } else {
+    auto min_len = std::min(owned_buffer_.length(), length);
+    owned_buffer_.copyOut(0, min_len, buffer);
+    if (!(flags & MSG_PEEK)) {
+      owned_buffer_.drain(min_len);
+    }
+    return {min_len, Api::IoErrorPtr(nullptr, [](Api::IoError*) {})};
+  }
+}
+
+bool BufferedIoSocketHandleImpl::supportsMmsg() const { return false; }
+
+bool BufferedIoSocketHandleImpl::supportsUdpGro() const { return false; }
+
+Api::SysCallIntResult makeInvalidSyscall() {
+  return Api::SysCallIntResult{-1, SOCKET_ERROR_NOT_SUP /*SOCKET_ERROR_NOT_SUP*/};
+}
+
+Api::SysCallIntResult BufferedIoSocketHandleImpl::bind(Address::InstanceConstSharedPtr) {
+  return makeInvalidSyscall();
+}
+
+Api::SysCallIntResult BufferedIoSocketHandleImpl::listen(int) { return makeInvalidSyscall(); }
+
+IoHandlePtr BufferedIoSocketHandleImpl::accept(struct sockaddr*, socklen_t*) {
+  NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
+}
+
+Api::SysCallIntResult BufferedIoSocketHandleImpl::connect(Address::InstanceConstSharedPtr) {
+  // Buffered Io handle should always be considered as connected. Use write to determine if peer is
+  // closed.
+  return {0, 0};
+}
+
+Api::SysCallIntResult BufferedIoSocketHandleImpl::setOption(int, int, const void*, socklen_t) {
+  return makeInvalidSyscall();
+}
+
+Api::SysCallIntResult BufferedIoSocketHandleImpl::getOption(int, int, void*, socklen_t*) {
+  return makeInvalidSyscall();
+}
+
+Api::SysCallIntResult BufferedIoSocketHandleImpl::setBlocking(bool) { return makeInvalidSyscall(); }
+
+absl::optional<int> BufferedIoSocketHandleImpl::domain() { return absl::nullopt; }
+
+Address::InstanceConstSharedPtr BufferedIoSocketHandleImpl::localAddress() {
+  throw EnvoyException(fmt::format("getsockname failed for BufferedIoSocketHandleImpl"));
+}
+
+Address::InstanceConstSharedPtr BufferedIoSocketHandleImpl::peerAddress() {
+
+  throw EnvoyException(fmt::format("getsockname failed for BufferedIoSocketHandleImpl"));
+}
+
+Event::FileEventPtr BufferedIoSocketHandleImpl::createFileEvent(Event::Dispatcher& dispatcher,
+                                                                Event::FileReadyCb cb,
+                                                                Event::FileTriggerType trigger_type,
+                                                                uint32_t events) {
+  ASSERT(event_counter_ == 0);
+  ++event_counter_;
+  io_callback_ = dispatcher.createSchedulableCallback([this]() { user_file_event_->onEvents(); });
+  auto res = Event::UserSpaceFileEventFactory::createUserSpaceFileEventImpl(
+      dispatcher, cb, trigger_type, events, *io_callback_, event_counter_);
+  user_file_event_ = res.get();
+  // Blindly activate the events.
+  io_callback_->scheduleCallbackNextIteration();
+  return res;
+}
+
+Api::SysCallIntResult BufferedIoSocketHandleImpl::shutdown(int how) {
+  // Support shutdown write.
+  if ((how == ENVOY_SHUT_WR) || (how == ENVOY_SHUT_RDWR)) {
+    ASSERT(!closed_);
+    if (!write_shutdown_) {
+      ASSERT(writable_peer_);
+      // Notify the peer we won't write more data. shutdown(WRITE).
+      writable_peer_->setWriteEnd();
+      writable_peer_->maybeSetNewData();
+      write_shutdown_ = true;
+    }
+  }
+  return {0, 0};
+}
+
+} // namespace Network
+} // namespace Envoy