Add API to grpc event engines to forcibly set underlying fd to be readable/writable/errored

src/core/lib/iomgr/ev_epoll1_linux.cc

@@ -397,6 +397,12 @@ static void fd_notify_on_error(grpc_fd* fd, grpc_closure* closure) {
   fd->error_closure->NotifyOn(closure);
 }
 
+static void fd_set_readable(grpc_fd* fd) { fd->read_closure->SetReady(); }
+
+static void fd_set_writable(grpc_fd* fd) { fd->write_closure->SetReady(); }
+
+static void fd_set_error(grpc_fd* fd) { fd->error_closure->SetReady(); }
+
 static void fd_become_readable(grpc_fd* fd, grpc_pollset* notifier) {
   fd->read_closure->SetReady();
   /* Use release store to match with acquire load in fd_get_read_notifier */
@@ -1217,6 +1223,9 @@ static const grpc_event_engine_vtable vtable = {
     fd_notify_on_read,
     fd_notify_on_write,
     fd_notify_on_error,
+    fd_set_readable,
+    fd_set_writable,
+    fd_set_error,
     fd_is_shutdown,
     fd_get_read_notifier_pollset,
 

src/core/lib/iomgr/ev_epollex_linux.cc

@@ -550,6 +550,12 @@ static void fd_notify_on_error(grpc_fd* fd, grpc_closure* closure) {
   fd->error_closure->NotifyOn(closure);
 }
 
+static void fd_set_readable(grpc_fd* fd) { fd->read_closure->SetReady(); }
+
+static void fd_set_writable(grpc_fd* fd) { fd->write_closure->SetReady(); }
+
+static void fd_set_error(grpc_fd* fd) { fd->error_closure->SetReady(); }
+
 /*******************************************************************************
  * Pollable Definitions
  */
@@ -1636,6 +1642,9 @@ static const grpc_event_engine_vtable vtable = {
     fd_notify_on_read,
     fd_notify_on_write,
     fd_notify_on_error,
+    fd_set_readable,
+    fd_set_writable,
+    fd_set_error,
     fd_is_shutdown,
     fd_get_read_notifier_pollset,
 

src/core/lib/iomgr/ev_epollsig_linux.cc

@@ -958,6 +958,12 @@ static void fd_notify_on_error(grpc_fd* fd, grpc_closure* closure) {
   fd->error_closure->NotifyOn(closure);
 }
 
+static void fd_set_readable(grpc_fd* fd) { fd->read_closure->SetReady(); }
+
+static void fd_set_writable(grpc_fd* fd) { fd->write_closure->SetReady(); }
+
+static void fd_set_error(grpc_fd* fd) { fd->error_closure->SetReady(); }
+
 /*******************************************************************************
  * Pollset Definitions
  */
@@ -1667,6 +1673,9 @@ static const grpc_event_engine_vtable vtable = {
     fd_notify_on_read,
     fd_notify_on_write,
     fd_notify_on_error,
+    fd_set_readable,
+    fd_set_writable,
+    fd_set_error,
     fd_is_shutdown,
     fd_get_read_notifier_pollset,
 

src/core/lib/iomgr/ev_poll_posix.cc

@@ -557,6 +557,23 @@ static void fd_notify_on_error(grpc_fd* fd, grpc_closure* closure) {
   abort();
 }
 
+static void fd_set_readable(grpc_fd* fd) {
+  gpr_mu_lock(&fd->mu);
+  set_ready_locked(fd, &fd->read_closure);
+  gpr_mu_unlock(&fd->mu);
+}
+
+static void fd_set_writable(grpc_fd* fd) {
+  gpr_mu_lock(&fd->mu);
+  set_ready_locked(fd, &fd->write_closure);
+  gpr_mu_unlock(&fd->mu);
+}
+
+static void fd_set_error(grpc_fd* fd) {
+  gpr_log(GPR_ERROR, "Polling engine does not support tracking errors.");
+  abort();
+}
+
 static uint32_t fd_begin_poll(grpc_fd* fd, grpc_pollset* pollset,
                               grpc_pollset_worker* worker, uint32_t read_mask,
                               uint32_t write_mask, grpc_fd_watcher* watcher) {
@@ -1723,6 +1740,9 @@ static const grpc_event_engine_vtable vtable = {
     fd_notify_on_read,
     fd_notify_on_write,
     fd_notify_on_error,
+    fd_set_readable,
+    fd_set_writable,
+    fd_set_error,
     fd_is_shutdown,
     fd_get_read_notifier_pollset,
 

src/core/lib/iomgr/ev_posix.cc

@@ -239,6 +239,12 @@ void grpc_fd_notify_on_error(grpc_fd* fd, grpc_closure* closure) {
   g_event_engine->fd_notify_on_error(fd, closure);
 }
 
+void grpc_fd_set_readable(grpc_fd* fd) { g_event_engine->fd_set_readable(fd); }
+
+void grpc_fd_set_writable(grpc_fd* fd) { g_event_engine->fd_set_writable(fd); }
+
+void grpc_fd_set_error(grpc_fd* fd) { g_event_engine->fd_set_error(fd); }
+
 static size_t pollset_size(void) { return g_event_engine->pollset_size; }
 
 static void pollset_init(grpc_pollset* pollset, gpr_mu** mu) {

src/core/lib/iomgr/ev_posix.h

@@ -51,6 +51,9 @@ typedef struct grpc_event_engine_vtable {
   void (*fd_notify_on_read)(grpc_fd* fd, grpc_closure* closure);
   void (*fd_notify_on_write)(grpc_fd* fd, grpc_closure* closure);
   void (*fd_notify_on_error)(grpc_fd* fd, grpc_closure* closure);
+  void (*fd_set_readable)(grpc_fd* fd);
+  void (*fd_set_writable)(grpc_fd* fd);
+  void (*fd_set_error)(grpc_fd* fd);
   bool (*fd_is_shutdown)(grpc_fd* fd);
   grpc_pollset* (*fd_get_read_notifier_pollset)(grpc_fd* fd);
 
@@ -142,6 +145,21 @@ void grpc_fd_notify_on_write(grpc_fd* fd, grpc_closure* closure);
  * needs to have been set on grpc_fd_create */
 void grpc_fd_notify_on_error(grpc_fd* fd, grpc_closure* closure);
 
+/* Forcibly set the fd to be readable, resulting in the closure registered with
+ * grpc_fd_notify_on_read being invoked.
+ */
+void grpc_fd_set_readable(grpc_fd* fd);
+
+/* Forcibly set the fd to be writable, resulting in the closure registered with
+ * grpc_fd_notify_on_write being invoked.
+ */
+void grpc_fd_set_writable(grpc_fd* fd);
+
+/* Forcibly set the fd to have errored, resulting in the closure registered with
+ * grpc_fd_notify_on_error being invoked.
+ */
+void grpc_fd_set_error(grpc_fd* fd);
+
 /* Return the read notifier pollset from the fd */
 grpc_pollset* grpc_fd_get_read_notifier_pollset(grpc_fd* fd);