Refactor Endpoint API

- Allow reads to complete immediately
- Unify read/write signatures
- Simplify memory management to allow future optimization work

4 files changed, 199 insertions, 224 deletions

diff --git a/test/core/bad_client/bad_client.c b/test/core/bad_client/bad_client.c
index 24bf5d3625..1d98879662 100644
--- a/test/core/bad_client/bad_client.c
+++ b/test/core/bad_client/bad_client.c
@@ -59,7 +59,7 @@ static void thd_func(void *arg) {
   gpr_event_set(&a->done_thd, (void *)1);
 }
 
-static void done_write(void *arg, grpc_endpoint_cb_status status) {
+static void done_write(void *arg, int success) {
   thd_args *a = arg;
   gpr_event_set(&a->done_write, (void *)1);
 }
@@ -85,6 +85,8 @@ void grpc_run_bad_client_test(grpc_bad_client_server_side_validator validator,
   grpc_mdctx *mdctx = grpc_mdctx_create();
   gpr_slice slice =
       gpr_slice_from_copied_buffer(client_payload, client_payload_length);
+  gpr_slice_buffer outgoing;
+  grpc_iomgr_closure done_write_closure;
 
   hex = gpr_dump(client_payload, client_payload_length,
                  GPR_DUMP_HEX | GPR_DUMP_ASCII);
@@ -122,14 +124,18 @@ void grpc_run_bad_client_test(grpc_bad_client_server_side_validator validator,
   /* Start validator */
   gpr_thd_new(&id, thd_func, &a, NULL);
 
+  gpr_slice_buffer_init(&outgoing);
+  gpr_slice_buffer_add(&outgoing, slice);
+  grpc_iomgr_closure_init(&done_write_closure, done_write, &a);
+
   /* Write data */
-  switch (grpc_endpoint_write(sfd.client, &slice, 1, done_write, &a)) {
-    case GRPC_ENDPOINT_WRITE_DONE:
+  switch (grpc_endpoint_write(sfd.client, &outgoing, &done_write_closure)) {
+    case GRPC_ENDPOINT_DONE:
       done_write(&a, 1);
       break;
-    case GRPC_ENDPOINT_WRITE_PENDING:
+    case GRPC_ENDPOINT_PENDING:
       break;
-    case GRPC_ENDPOINT_WRITE_ERROR:
+    case GRPC_ENDPOINT_ERROR:
       done_write(&a, 0);
       break;
   }
@@ -155,6 +161,7 @@ void grpc_run_bad_client_test(grpc_bad_client_server_side_validator validator,
                  .type == GRPC_OP_COMPLETE);
   grpc_server_destroy(a.server);
   grpc_completion_queue_destroy(a.cq);
+  gpr_slice_buffer_destroy(&outgoing);
 
   grpc_shutdown();
 }
diff --git a/test/core/iomgr/endpoint_tests.c b/test/core/iomgr/endpoint_tests.c
index 8186c96da1..b4f170cfbe 100644
--- a/test/core/iomgr/endpoint_tests.c
+++ b/test/core/iomgr/endpoint_tests.c
@@ -59,8 +59,7 @@
 
 static grpc_pollset *g_pollset;
 
-size_t count_and_unref_slices(gpr_slice *slices, size_t nslices,
-                              int *current_data) {
+size_t count_slices(gpr_slice *slices, size_t nslices, int *current_data) {
   size_t num_bytes = 0;
   size_t i;
   size_t j;
@@ -72,7 +71,6 @@ size_t count_and_unref_slices(gpr_slice *slices, size_t nslices,
       *current_data = (*current_data + 1) % 256;
     }
     num_bytes += GPR_SLICE_LENGTH(slices[i]);
-    gpr_slice_unref(slices[i]);
   }
   return num_bytes;
 }
@@ -121,86 +119,76 @@ struct read_and_write_test_state {
   int current_write_data;
   int read_done;
   int write_done;
+  gpr_slice_buffer incoming;
+  gpr_slice_buffer outgoing;
+  grpc_iomgr_closure done_read;
+  grpc_iomgr_closure done_write;
 };
 
-static void read_and_write_test_read_handler(void *data, gpr_slice *slices,
-                                             size_t nslices,
-                                             grpc_endpoint_cb_status error) {
+static void read_and_write_test_read_handler(void *data, int success) {
   struct read_and_write_test_state *state = data;
-  GPR_ASSERT(error != GRPC_ENDPOINT_CB_ERROR);
-  if (error == GRPC_ENDPOINT_CB_SHUTDOWN) {
-    gpr_log(GPR_INFO, "Read handler shutdown");
-    gpr_mu_lock(GRPC_POLLSET_MU(g_pollset));
-    state->read_done = 1;
-    grpc_pollset_kick(g_pollset, NULL);
-    gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset));
-    return;
-  }
 
-  state->bytes_read +=
-      count_and_unref_slices(slices, nslices, &state->current_read_data);
-  if (state->bytes_read == state->target_bytes) {
+  state->bytes_read += count_slices(
+      state->incoming.slices, state->incoming.count, &state->current_read_data);
+  if (state->bytes_read == state->target_bytes || !success) {
     gpr_log(GPR_INFO, "Read handler done");
     gpr_mu_lock(GRPC_POLLSET_MU(g_pollset));
-    state->read_done = 1;
+    state->read_done = 1 + success;
     grpc_pollset_kick(g_pollset, NULL);
     gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset));
-  } else {
-    grpc_endpoint_notify_on_read(state->read_ep,
-                                 read_and_write_test_read_handler, data);
+  } else if (success) {
+    switch (grpc_endpoint_read(state->read_ep, &state->incoming,
+                               &state->done_read)) {
+      case GRPC_ENDPOINT_ERROR:
+        read_and_write_test_read_handler(data, 0);
+        break;
+      case GRPC_ENDPOINT_DONE:
+        read_and_write_test_read_handler(data, 1);
+        break;
+      case GRPC_ENDPOINT_PENDING:
+        break;
+    }
   }
 }
 
-static void read_and_write_test_write_handler(void *data,
-                                              grpc_endpoint_cb_status error) {
+static void read_and_write_test_write_handler(void *data, int success) {
   struct read_and_write_test_state *state = data;
   gpr_slice *slices = NULL;
   size_t nslices;
-  grpc_endpoint_write_status write_status;
-
-  GPR_ASSERT(error != GRPC_ENDPOINT_CB_ERROR);
-
-  gpr_log(GPR_DEBUG, "%s: error=%d", "read_and_write_test_write_handler",
-          error);
-
-  if (error == GRPC_ENDPOINT_CB_SHUTDOWN) {
-    gpr_log(GPR_INFO, "Write handler shutdown");
-    gpr_mu_lock(GRPC_POLLSET_MU(g_pollset));
-    state->write_done = 1;
-    grpc_pollset_kick(g_pollset, NULL);
-    gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset));
-    return;
-  }
-
-  for (;;) {
-    /* Need to do inline writes until they don't succeed synchronously or we
-       finish writing */
-    state->bytes_written += state->current_write_size;
-    if (state->target_bytes - state->bytes_written <
-        state->current_write_size) {
-      state->current_write_size = state->target_bytes - state->bytes_written;
-    }
-    if (state->current_write_size == 0) {
-      break;
-    }
-
-    slices = allocate_blocks(state->current_write_size, 8192, &nslices,
-                             &state->current_write_data);
-    write_status =
-        grpc_endpoint_write(state->write_ep, slices, nslices,
-                            read_and_write_test_write_handler, state);
-    gpr_log(GPR_DEBUG, "write_status=%d", write_status);
-    GPR_ASSERT(write_status != GRPC_ENDPOINT_WRITE_ERROR);
-    free(slices);
-    if (write_status == GRPC_ENDPOINT_WRITE_PENDING) {
-      return;
+  grpc_endpoint_op_status write_status;
+
+  if (success) {
+    for (;;) {
+      /* Need to do inline writes until they don't succeed synchronously or we
+         finish writing */
+      state->bytes_written += state->current_write_size;
+      if (state->target_bytes - state->bytes_written <
+          state->current_write_size) {
+        state->current_write_size = state->target_bytes - state->bytes_written;
+      }
+      if (state->current_write_size == 0) {
+        break;
+      }
+
+      slices = allocate_blocks(state->current_write_size, 8192, &nslices,
+                               &state->current_write_data);
+      gpr_slice_buffer_reset_and_unref(&state->outgoing);
+      gpr_slice_buffer_addn(&state->outgoing, slices, nslices);
+      write_status = grpc_endpoint_write(state->write_ep, &state->outgoing,
+                                         &state->done_write);
+      gpr_log(GPR_DEBUG, "write_status=%d", write_status);
+      GPR_ASSERT(write_status != GRPC_ENDPOINT_ERROR);
+      free(slices);
+      if (write_status == GRPC_ENDPOINT_PENDING) {
+        return;
+      }
     }
+    GPR_ASSERT(state->bytes_written == state->target_bytes);
   }
-  GPR_ASSERT(state->bytes_written == state->target_bytes);
 
   gpr_log(GPR_INFO, "Write handler done");
   gpr_mu_lock(GRPC_POLLSET_MU(g_pollset));
-  state->write_done = 1;
+  state->write_done = 1 + success;
   grpc_pollset_kick(g_pollset, NULL);
   gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset));
 }
@@ -234,16 +222,31 @@ static void read_and_write_test(grpc_endpoint_test_config config,
   state.write_done = 0;
   state.current_read_data = 0;
   state.current_write_data = 0;
+  grpc_iomgr_closure_init(&state.done_read, read_and_write_test_read_handler,
+                          &state);
+  grpc_iomgr_closure_init(&state.done_write, read_and_write_test_write_handler,
+                          &state);
+  gpr_slice_buffer_init(&state.outgoing);
+  gpr_slice_buffer_init(&state.incoming);
 
   /* Get started by pretending an initial write completed */
   /* NOTE: Sets up initial conditions so we can have the same write handler
      for the first iteration as for later iterations. It does the right thing
      even when bytes_written is unsigned. */
   state.bytes_written -= state.current_write_size;
-  read_and_write_test_write_handler(&state, GRPC_ENDPOINT_CB_OK);
+  read_and_write_test_write_handler(&state, 1);
 
-  grpc_endpoint_notify_on_read(state.read_ep, read_and_write_test_read_handler,
-                               &state);
+  switch (
+      grpc_endpoint_read(state.read_ep, &state.incoming, &state.done_read)) {
+    case GRPC_ENDPOINT_PENDING:
+      break;
+    case GRPC_ENDPOINT_ERROR:
+      read_and_write_test_read_handler(&state, 0);
+      break;
+    case GRPC_ENDPOINT_DONE:
+      read_and_write_test_read_handler(&state, 1);
+      break;
+  }
 
   if (shutdown) {
     gpr_log(GPR_DEBUG, "shutdown read");
@@ -262,6 +265,8 @@ static void read_and_write_test(grpc_endpoint_test_config config,
 
   grpc_endpoint_destroy(state.read_ep);
   grpc_endpoint_destroy(state.write_ep);
+  gpr_slice_buffer_destroy(&state.outgoing);
+  gpr_slice_buffer_destroy(&state.incoming);
   end_test(config);
 }
 
@@ -272,36 +277,40 @@ struct timeout_test_state {
 typedef struct {
   int done;
   grpc_endpoint *ep;
+  gpr_slice_buffer incoming;
+  grpc_iomgr_closure done_read;
 } shutdown_during_write_test_state;
 
-static void shutdown_during_write_test_read_handler(
-    void *user_data, gpr_slice *slices, size_t nslices,
-    grpc_endpoint_cb_status error) {
-  size_t i;
+static void shutdown_during_write_test_read_handler(void *user_data,
+                                                    int success) {
   shutdown_during_write_test_state *st = user_data;
 
-  for (i = 0; i < nslices; i++) {
-    gpr_slice_unref(slices[i]);
-  }
-
-  if (error != GRPC_ENDPOINT_CB_OK) {
+  if (!success) {
     grpc_endpoint_destroy(st->ep);
     gpr_mu_lock(GRPC_POLLSET_MU(g_pollset));
-    st->done = error;
+    st->done = 1;
     grpc_pollset_kick(g_pollset, NULL);
     gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset));
   } else {
-    grpc_endpoint_notify_on_read(
-        st->ep, shutdown_during_write_test_read_handler, user_data);
+    switch (grpc_endpoint_read(st->ep, &st->incoming, &st->done_read)) {
+      case GRPC_ENDPOINT_PENDING:
+        break;
+      case GRPC_ENDPOINT_ERROR:
+        shutdown_during_write_test_read_handler(user_data, 0);
+        break;
+      case GRPC_ENDPOINT_DONE:
+        shutdown_during_write_test_read_handler(user_data, 1);
+        break;
+    }
   }
 }
 
-static void shutdown_during_write_test_write_handler(
-    void *user_data, grpc_endpoint_cb_status error) {
+static void shutdown_during_write_test_write_handler(void *user_data,
+                                                     int success) {
   shutdown_during_write_test_state *st = user_data;
-  gpr_log(GPR_INFO, "shutdown_during_write_test_write_handler: error = %d",
-          error);
-  if (error == 0) {
+  gpr_log(GPR_INFO, "shutdown_during_write_test_write_handler: success = %d",
+          success);
+  if (success) {
     /* This happens about 0.5% of the time when run under TSAN, and is entirely
        legitimate, but means we aren't testing the path we think we are. */
     /* TODO(klempner): Change this test to retry the write in that case */
@@ -324,6 +333,8 @@ static void shutdown_during_write_test(grpc_endpoint_test_config config,
   shutdown_during_write_test_state read_st;
   shutdown_during_write_test_state write_st;
   gpr_slice *slices;
+  gpr_slice_buffer outgoing;
+  grpc_iomgr_closure done_write;
   grpc_endpoint_test_fixture f =
       begin_test(config, "shutdown_during_write_test", slice_size);
 
@@ -334,19 +345,26 @@ static void shutdown_during_write_test(grpc_endpoint_test_config config,
   read_st.done = 0;
   write_st.done = 0;
 
-  grpc_endpoint_notify_on_read(
-      read_st.ep, shutdown_during_write_test_read_handler, &read_st);
+  grpc_iomgr_closure_init(&done_write, shutdown_during_write_test_write_handler,
+                          &write_st);
+  grpc_iomgr_closure_init(&read_st.done_read,
+                          shutdown_during_write_test_read_handler, &read_st);
+  gpr_slice_buffer_init(&read_st.incoming);
+  gpr_slice_buffer_init(&outgoing);
+
+  GPR_ASSERT(grpc_endpoint_read(read_st.ep, &read_st.incoming,
+                                &read_st.done_read) == GRPC_ENDPOINT_PENDING);
   for (size = 1;; size *= 2) {
     slices = allocate_blocks(size, 1, &nblocks, &current_data);
-    switch (grpc_endpoint_write(write_st.ep, slices, nblocks,
-                                shutdown_during_write_test_write_handler,
-                                &write_st)) {
-      case GRPC_ENDPOINT_WRITE_DONE:
+    gpr_slice_buffer_reset_and_unref(&outgoing);
+    gpr_slice_buffer_addn(&outgoing, slices, nblocks);
+    switch (grpc_endpoint_write(write_st.ep, &outgoing, &done_write)) {
+      case GRPC_ENDPOINT_DONE:
         break;
-      case GRPC_ENDPOINT_WRITE_ERROR:
+      case GRPC_ENDPOINT_ERROR:
         gpr_log(GPR_ERROR, "error writing");
         abort();
-      case GRPC_ENDPOINT_WRITE_PENDING:
+      case GRPC_ENDPOINT_PENDING:
         grpc_endpoint_shutdown(write_st.ep);
         deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10);
         gpr_mu_lock(GRPC_POLLSET_MU(g_pollset));
@@ -365,6 +383,8 @@ static void shutdown_during_write_test(grpc_endpoint_test_config config,
         }
         gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset));
         gpr_free(slices);
+        gpr_slice_buffer_destroy(&read_st.incoming);
+        gpr_slice_buffer_destroy(&outgoing);
         end_test(config);
         return;
     }
diff --git a/test/core/iomgr/tcp_posix_test.c b/test/core/iomgr/tcp_posix_test.c
index 17a85ceaec..a2e3adcf29 100644
--- a/test/core/iomgr/tcp_posix_test.c
+++ b/test/core/iomgr/tcp_posix_test.c
@@ -118,10 +118,12 @@ struct read_socket_state {
   grpc_endpoint *ep;
   ssize_t read_bytes;
   ssize_t target_read_bytes;
+  gpr_slice_buffer incoming;
+  grpc_iomgr_closure read_cb;
 };
 
-static ssize_t count_and_unref_slices(gpr_slice *slices, size_t nslices,
-                                      int *current_data) {
+static ssize_t count_slices(gpr_slice *slices, size_t nslices,
+                            int *current_data) {
   ssize_t num_bytes = 0;
   unsigned i, j;
   unsigned char *buf;
@@ -132,31 +134,41 @@ static ssize_t count_and_unref_slices(gpr_slice *slices, size_t nslices,
       *current_data = (*current_data + 1) % 256;
     }
     num_bytes += GPR_SLICE_LENGTH(slices[i]);
-    gpr_slice_unref(slices[i]);
   }
   return num_bytes;
 }
 
-static void read_cb(void *user_data, gpr_slice *slices, size_t nslices,
-                    grpc_endpoint_cb_status error) {
+static void read_cb(void *user_data, int success) {
   struct read_socket_state *state = (struct read_socket_state *)user_data;
   ssize_t read_bytes;
   int current_data;
 
-  GPR_ASSERT(error == GRPC_ENDPOINT_CB_OK);
+  GPR_ASSERT(success);
 
   gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));
   current_data = state->read_bytes % 256;
-  read_bytes = count_and_unref_slices(slices, nslices, &current_data);
+  read_bytes = count_slices(state->incoming.slices, state->incoming.count,
+                            &current_data);
   state->read_bytes += read_bytes;
   gpr_log(GPR_INFO, "Read %d bytes of %d", read_bytes,
           state->target_read_bytes);
   if (state->read_bytes >= state->target_read_bytes) {
-    /* empty */
+    gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));
   } else {
-    grpc_endpoint_notify_on_read(state->ep, read_cb, state);
+    switch (grpc_endpoint_read(state->ep, &state->incoming, &state->read_cb)) {
+      case GRPC_ENDPOINT_DONE:
+        gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));
+        read_cb(user_data, 1);
+        break;
+      case GRPC_ENDPOINT_ERROR:
+        gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));
+        read_cb(user_data, 0);
+        break;
+      case GRPC_ENDPOINT_PENDING:
+        gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));
+        break;
+    }
   }
-  gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));
 }
 
 /* Write to a socket, then read from it using the grpc_tcp API. */
@@ -181,8 +193,19 @@ static void read_test(ssize_t num_bytes, ssize_t slice_size) {
   state.ep = ep;
   state.read_bytes = 0;
   state.target_read_bytes = written_bytes;
+  gpr_slice_buffer_init(&state.incoming);
+  grpc_iomgr_closure_init(&state.read_cb, read_cb, &state);
 
-  grpc_endpoint_notify_on_read(ep, read_cb, &state);
+  switch (grpc_endpoint_read(ep, &state.incoming, &state.read_cb)) {
+    case GRPC_ENDPOINT_DONE:
+      read_cb(&state, 1);
+      break;
+    case GRPC_ENDPOINT_ERROR:
+      read_cb(&state, 0);
+      break;
+    case GRPC_ENDPOINT_PENDING:
+      break;
+  }
 
   gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));
   while (state.read_bytes < state.target_read_bytes) {
@@ -192,6 +215,7 @@ static void read_test(ssize_t num_bytes, ssize_t slice_size) {
   GPR_ASSERT(state.read_bytes == state.target_read_bytes);
   gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));
 
+  gpr_slice_buffer_destroy(&state.incoming);
   grpc_endpoint_destroy(ep);
 }
 
@@ -218,8 +242,19 @@ static void large_read_test(ssize_t slice_size) {
   state.ep = ep;
   state.read_bytes = 0;
   state.target_read_bytes = written_bytes;
+  gpr_slice_buffer_init(&state.incoming);
+  grpc_iomgr_closure_init(&state.read_cb, read_cb, &state);
 
-  grpc_endpoint_notify_on_read(ep, read_cb, &state);
+  switch (grpc_endpoint_read(ep, &state.incoming, &state.read_cb)) {
+    case GRPC_ENDPOINT_DONE:
+      read_cb(&state, 1);
+      break;
+    case GRPC_ENDPOINT_ERROR:
+      read_cb(&state, 0);
+      break;
+    case GRPC_ENDPOINT_PENDING:
+      break;
+  }
 
   gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));
   while (state.read_bytes < state.target_read_bytes) {
@@ -229,6 +264,7 @@ static void large_read_test(ssize_t slice_size) {
   GPR_ASSERT(state.read_bytes == state.target_read_bytes);
   gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));
 
+  gpr_slice_buffer_destroy(&state.incoming);
   grpc_endpoint_destroy(ep);
 }
 
@@ -260,8 +296,7 @@ static gpr_slice *allocate_blocks(ssize_t num_bytes, ssize_t slice_size,
   return slices;
 }
 
-static void write_done(void *user_data /* write_socket_state */,
-                       grpc_endpoint_cb_status error) {
+static void write_done(void *user_data /* write_socket_state */, int success) {
   struct write_socket_state *state = (struct write_socket_state *)user_data;
   gpr_log(GPR_INFO, "Write done callback called");
   gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));
@@ -336,6 +371,8 @@ static void write_test(ssize_t num_bytes, ssize_t slice_size) {
   size_t num_blocks;
   gpr_slice *slices;
   int current_data = 0;
+  gpr_slice_buffer outgoing;
+  grpc_iomgr_closure write_done_closure;
   gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(20);
 
   gpr_log(GPR_INFO, "Start write test with %d bytes, slice size %d", num_bytes,
@@ -352,73 +389,21 @@ static void write_test(ssize_t num_bytes, ssize_t slice_size) {
 
   slices = allocate_blocks(num_bytes, slice_size, &num_blocks, &current_data);
 
-  if (grpc_endpoint_write(ep, slices, num_blocks, write_done, &state) ==
-      GRPC_ENDPOINT_WRITE_DONE) {
-    /* Write completed immediately */
-    read_bytes = drain_socket(sv[0]);
-    GPR_ASSERT(read_bytes == num_bytes);
-  } else {
-    drain_socket_blocking(sv[0], num_bytes, num_bytes);
-    gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));
-    for (;;) {
-      grpc_pollset_worker worker;
-      if (state.write_done) {
-        break;
-      }
-      grpc_pollset_work(&g_pollset, &worker, deadline);
-    }
-    gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));
-  }
-
-  grpc_endpoint_destroy(ep);
-  gpr_free(slices);
-}
-
-static void read_done_for_write_error(void *ud, gpr_slice *slices,
-                                      size_t nslices,
-                                      grpc_endpoint_cb_status error) {
-  GPR_ASSERT(error != GRPC_ENDPOINT_CB_OK);
-  GPR_ASSERT(nslices == 0);
-}
-
-/* Write to a socket using the grpc_tcp API, then drain it directly.
-   Note that if the write does not complete immediately we need to drain the
-   socket in parallel with the read. */
-static void write_error_test(ssize_t num_bytes, ssize_t slice_size) {
-  int sv[2];
-  grpc_endpoint *ep;
-  struct write_socket_state state;
-  size_t num_blocks;
-  gpr_slice *slices;
-  int current_data = 0;
-  grpc_pollset_worker worker;
-  gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(20);
-
-  gpr_log(GPR_INFO, "Start write error test with %d bytes, slice size %d",
-          num_bytes, slice_size);
-
-  create_sockets(sv);
+  gpr_slice_buffer_init(&outgoing);
+  gpr_slice_buffer_addn(&outgoing, slices, num_blocks);
+  grpc_iomgr_closure_init(&write_done_closure, write_done, &state);
 
-  ep = grpc_tcp_create(grpc_fd_create(sv[1], "write_error_test"),
-                       GRPC_TCP_DEFAULT_READ_SLICE_SIZE, "test");
-  grpc_endpoint_add_to_pollset(ep, &g_pollset);
-
-  close(sv[0]);
-
-  state.ep = ep;
-  state.write_done = 0;
-
-  slices = allocate_blocks(num_bytes, slice_size, &num_blocks, &current_data);
-
-  switch (grpc_endpoint_write(ep, slices, num_blocks, write_done, &state)) {
-    case GRPC_ENDPOINT_WRITE_DONE:
-    case GRPC_ENDPOINT_WRITE_ERROR:
+  switch (grpc_endpoint_write(ep, &outgoing, &write_done_closure)) {
+    case GRPC_ENDPOINT_DONE:
       /* Write completed immediately */
+      read_bytes = drain_socket(sv[0]);
+      GPR_ASSERT(read_bytes == num_bytes);
       break;
-    case GRPC_ENDPOINT_WRITE_PENDING:
-      grpc_endpoint_notify_on_read(ep, read_done_for_write_error, NULL);
+    case GRPC_ENDPOINT_PENDING:
+      drain_socket_blocking(sv[0], num_bytes, num_bytes);
       gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));
       for (;;) {
+        grpc_pollset_worker worker;
         if (state.write_done) {
           break;
         }
@@ -426,10 +411,14 @@ static void write_error_test(ssize_t num_bytes, ssize_t slice_size) {
       }
       gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));
       break;
+    case GRPC_ENDPOINT_ERROR:
+      gpr_log(GPR_ERROR, "endpoint got error");
+      abort();
   }
 
+  gpr_slice_buffer_destroy(&outgoing);
   grpc_endpoint_destroy(ep);
-  free(slices);
+  gpr_free(slices);
 }
 
 void run_tests(void) {
@@ -449,10 +438,6 @@ void run_tests(void) {
   write_test(100000, 137);
 
   for (i = 1; i < 1000; i = GPR_MAX(i + 1, i * 5 / 4)) {
-    write_error_test(40320, i);
-  }
-
-  for (i = 1; i < 1000; i = GPR_MAX(i + 1, i * 5 / 4)) {
     write_test(40320, i);
   }
 }
diff --git a/test/core/security/secure_endpoint_test.c b/test/core/security/secure_endpoint_test.c
index a8368fc842..c76ddcd194 100644
--- a/test/core/security/secure_endpoint_test.c
+++ b/test/core/security/secure_endpoint_test.c
@@ -135,62 +135,26 @@ static grpc_endpoint_test_config configs[] = {
      secure_endpoint_create_fixture_tcp_socketpair_leftover, clean_up},
 };
 
-static void verify_leftover(void *user_data, gpr_slice *slices, size_t nslices,
-                            grpc_endpoint_cb_status error) {
-  gpr_slice s =
-      gpr_slice_from_copied_string("hello world 12345678900987654321");
-
-  GPR_ASSERT(error == GRPC_ENDPOINT_CB_OK);
-  GPR_ASSERT(nslices == 1);
-
-  GPR_ASSERT(0 == gpr_slice_cmp(s, slices[0]));
-  gpr_slice_unref(slices[0]);
-  gpr_slice_unref(s);
-  *(int *)user_data = 1;
-}
-
 static void test_leftover(grpc_endpoint_test_config config, size_t slice_size) {
   grpc_endpoint_test_fixture f = config.create_fixture(slice_size);
-  int verified = 0;
+  gpr_slice_buffer incoming;
+  gpr_slice s =
+      gpr_slice_from_copied_string("hello world 12345678900987654321");
   gpr_log(GPR_INFO, "Start test left over");
 
-  grpc_endpoint_notify_on_read(f.client_ep, verify_leftover, &verified);
-  GPR_ASSERT(verified == 1);
+  gpr_slice_buffer_init(&incoming);
+  GPR_ASSERT(grpc_endpoint_read(f.client_ep, &incoming, NULL) ==
+             GRPC_ENDPOINT_DONE);
+  GPR_ASSERT(incoming.count == 1);
+  GPR_ASSERT(0 == gpr_slice_cmp(s, incoming.slices[0]));
 
   grpc_endpoint_shutdown(f.client_ep);
   grpc_endpoint_shutdown(f.server_ep);
   grpc_endpoint_destroy(f.client_ep);
   grpc_endpoint_destroy(f.server_ep);
-  clean_up();
-}
-
-static void destroy_early(void *user_data, gpr_slice *slices, size_t nslices,
-                          grpc_endpoint_cb_status error) {
-  grpc_endpoint_test_fixture *f = user_data;
-  gpr_slice s =
-      gpr_slice_from_copied_string("hello world 12345678900987654321");
-
-  GPR_ASSERT(error == GRPC_ENDPOINT_CB_OK);
-  GPR_ASSERT(nslices == 1);
-
-  grpc_endpoint_shutdown(f->client_ep);
-  grpc_endpoint_destroy(f->client_ep);
-
-  GPR_ASSERT(0 == gpr_slice_cmp(s, slices[0]));
-  gpr_slice_unref(slices[0]);
   gpr_slice_unref(s);
-}
+  gpr_slice_buffer_destroy(&incoming);
 
-/* test which destroys the ep before finishing reading */
-static void test_destroy_ep_early(grpc_endpoint_test_config config,
-                                  size_t slice_size) {
-  grpc_endpoint_test_fixture f = config.create_fixture(slice_size);
-  gpr_log(GPR_INFO, "Start test destroy early");
-
-  grpc_endpoint_notify_on_read(f.client_ep, destroy_early, &f);
-
-  grpc_endpoint_shutdown(f.server_ep);
-  grpc_endpoint_destroy(f.server_ep);
   clean_up();
 }
 
@@ -203,7 +167,6 @@ int main(int argc, char **argv) {
   grpc_pollset_init(&g_pollset);
   grpc_endpoint_tests(configs[0], &g_pollset);
   test_leftover(configs[1], 1);
-  test_destroy_ep_early(configs[1], 1);
   grpc_pollset_shutdown(&g_pollset, destroy_pollset, &g_pollset);
   grpc_iomgr_shutdown();