Initial import.

2 files changed, 925 insertions, 0 deletions

diff --git a/test/core/eventmanager/em_pipe_test.c b/test/core/eventmanager/em_pipe_test.c
new file mode 100644
index 0000000000..5411142c89
--- /dev/null
+++ b/test/core/eventmanager/em_pipe_test.c
@@ -0,0 +1,200 @@
+/*
+ *
+ * Copyright 2014, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+/* Test grpc_em_fd with pipe. The test creates a pipe with non-blocking mode,
+   sends a stream of bytes through the pipe, and verifies that all bytes are
+   received. */
+#include "src/core/eventmanager/em.h"
+
+#include <errno.h>
+#include <fcntl.h>
+#include <pthread.h>
+#include <string.h>
+#include <stdio.h>
+#include <unistd.h>
+
+#include <grpc/support/log.h>
+#include "test/core/util/test_config.h"
+
+/* Operation for fcntl() to set pipe buffer size. */
+#ifndef F_SETPIPE_SZ
+#define F_SETPIPE_SZ (1024 + 7)
+#endif
+
+#define TOTAL_WRITE 3 /* total number of times that the write buffer is full. \
+                         */
+#define BUF_SIZE 1024
+char read_buf[BUF_SIZE];
+char write_buf[BUF_SIZE];
+
+typedef struct {
+  int fd[2];
+  grpc_em em;
+  grpc_em_fd read_em_fd;
+  grpc_em_fd write_em_fd;
+  int num_write; /* number of times that the write buffer is full*/
+  ssize_t bytes_written_total; /* total number of bytes written to the pipe */
+  ssize_t bytes_read_total;    /* total number of bytes read from the pipe */
+  pthread_mutex_t mu;          /* protect cv and done */
+  pthread_cond_t cv;           /* signaled when read finished */
+  int done;                    /* set to 1 when read finished */
+} async_pipe;
+
+void write_shutdown_cb(void *arg, /*async_pipe*/
+                       enum grpc_em_cb_status status) {
+  async_pipe *ap = arg;
+  close(ap->fd[1]);
+  grpc_em_fd_destroy(&ap->write_em_fd);
+}
+
+void write_cb(void *arg, /*async_pipe*/ enum grpc_em_cb_status status) {
+  async_pipe *ap = arg;
+  ssize_t bytes_written = 0;
+
+  if (status == GRPC_CALLBACK_CANCELLED) {
+    write_shutdown_cb(arg, GRPC_CALLBACK_SUCCESS);
+    return;
+  }
+
+  do {
+    bytes_written = write(ap->fd[1], write_buf, BUF_SIZE);
+    if (bytes_written > 0) ap->bytes_written_total += bytes_written;
+  } while (bytes_written > 0);
+
+  if (errno == EAGAIN) {
+    if (ap->num_write < TOTAL_WRITE) {
+      ap->num_write++;
+      grpc_em_fd_notify_on_write(&ap->write_em_fd, write_cb, ap,
+                                 gpr_inf_future);
+    } else {
+      /* Note that this could just shut down directly; doing a trip through the
+         shutdown path serves only a demonstration of the API. */
+      grpc_em_fd_shutdown(&ap->write_em_fd);
+      grpc_em_fd_notify_on_write(&ap->write_em_fd, write_cb, ap,
+                                 gpr_inf_future);
+    }
+  } else {
+    GPR_ASSERT(0 && strcat("unknown errno: ", strerror(errno)));
+  }
+}
+
+void read_shutdown_cb(void *arg, /*async_pipe*/ enum grpc_em_cb_status status) {
+  async_pipe *ap = arg;
+  close(ap->fd[0]);
+  grpc_em_fd_destroy(&ap->read_em_fd);
+  pthread_mutex_lock(&ap->mu);
+  if (ap->done == 0) {
+    ap->done = 1;
+    pthread_cond_signal(&ap->cv);
+  }
+  pthread_mutex_unlock(&ap->mu);
+}
+
+void read_cb(void *arg, /*async_pipe*/ enum grpc_em_cb_status status) {
+  async_pipe *ap = arg;
+  ssize_t bytes_read = 0;
+
+  if (status == GRPC_CALLBACK_CANCELLED) {
+    read_shutdown_cb(arg, GRPC_CALLBACK_SUCCESS);
+    return;
+  }
+
+  do {
+    bytes_read = read(ap->fd[0], read_buf, BUF_SIZE);
+    if (bytes_read > 0) ap->bytes_read_total += bytes_read;
+  } while (bytes_read > 0);
+
+  if (bytes_read == 0) {
+    /* Note that this could just shut down directly; doing a trip through the
+       shutdown path serves only a demonstration of the API. */
+    grpc_em_fd_shutdown(&ap->read_em_fd);
+    grpc_em_fd_notify_on_read(&ap->read_em_fd, read_cb, ap, gpr_inf_future);
+  } else if (bytes_read == -1) {
+    if (errno == EAGAIN) {
+      grpc_em_fd_notify_on_read(&ap->read_em_fd, read_cb, ap, gpr_inf_future);
+    } else {
+      GPR_ASSERT(0 && strcat("unknown errno: ", strerror(errno)));
+    }
+  }
+}
+
+void dummy_cb(void *arg, /*async_pipe*/ enum grpc_em_cb_status status) {}
+
+void async_pipe_init(async_pipe *ap) {
+  int i;
+
+  ap->num_write = 0;
+  ap->bytes_written_total = 0;
+  ap->bytes_read_total = 0;
+
+  pthread_mutex_init(&ap->mu, NULL);
+  pthread_cond_init(&ap->cv, NULL);
+  ap->done = 0;
+
+  GPR_ASSERT(0 == pipe(ap->fd));
+  for (i = 0; i < 2; i++) {
+    int flags = fcntl(ap->fd[i], F_GETFL, 0);
+    GPR_ASSERT(fcntl(ap->fd[i], F_SETFL, flags | O_NONBLOCK) == 0);
+    GPR_ASSERT(fcntl(ap->fd[i], F_SETPIPE_SZ, 4096) == 4096);
+  }
+
+  grpc_em_init(&ap->em);
+  grpc_em_fd_init(&ap->read_em_fd, &ap->em, ap->fd[0]);
+  grpc_em_fd_init(&ap->write_em_fd, &ap->em, ap->fd[1]);
+}
+
+static void async_pipe_start(async_pipe *ap) {
+  grpc_em_fd_notify_on_read(&ap->read_em_fd, read_cb, ap, gpr_inf_future);
+  grpc_em_fd_notify_on_write(&ap->write_em_fd, write_cb, ap, gpr_inf_future);
+}
+
+static void async_pipe_wait_destroy(async_pipe *ap) {
+  pthread_mutex_lock(&ap->mu);
+  while (!ap->done) pthread_cond_wait(&ap->cv, &ap->mu);
+  pthread_mutex_unlock(&ap->mu);
+  pthread_mutex_destroy(&ap->mu);
+  pthread_cond_destroy(&ap->cv);
+
+  grpc_em_destroy(&ap->em);
+}
+
+int main(int argc, char **argv) {
+  async_pipe ap;
+  grpc_test_init(argc, argv);
+  async_pipe_init(&ap);
+  async_pipe_start(&ap);
+  async_pipe_wait_destroy(&ap);
+  GPR_ASSERT(ap.bytes_read_total == ap.bytes_written_total);
+  gpr_log(GPR_INFO, "read total bytes %d", ap.bytes_read_total);
+  return 0;
+}
diff --git a/test/core/eventmanager/em_test.c b/test/core/eventmanager/em_test.c
new file mode 100644
index 0000000000..2bcfe86c3b
--- /dev/null
+++ b/test/core/eventmanager/em_test.c
@@ -0,0 +1,725 @@
+/*
+ *
+ * Copyright 2014, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+/* Test gRPC event manager with a simple TCP upload server and client. */
+#include "src/core/eventmanager/em.h"
+
+#include <ctype.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <netinet/in.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/time.h>
+#include <unistd.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/sync.h>
+#include <grpc/support/time.h>
+#include "test/core/util/test_config.h"
+
+/* buffer size used to send and receive data.
+   1024 is the minimal value to set TCP send and receive buffer. */
+#define BUF_SIZE 1024
+
+/* Create a test socket with the right properties for testing.
+   port is the TCP port to listen or connect to.
+   Return a socket FD and sockaddr_in. */
+static void create_test_socket(int port, int *socket_fd,
+                               struct sockaddr_in *sin) {
+  int fd;
+  int one = 1;
+  int buf_size = BUF_SIZE;
+  int flags;
+
+  fd = socket(AF_INET, SOCK_STREAM, 0);
+  setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
+  /* Reset the size of socket send buffer to the minimal value to facilitate
+     buffer filling up and triggering notify_on_write  */
+  GPR_ASSERT(
+      setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &buf_size, sizeof(buf_size)) != -1);
+  GPR_ASSERT(
+      setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &buf_size, sizeof(buf_size)) != -1);
+  /* Make fd non-blocking */
+  flags = fcntl(fd, F_GETFL, 0);
+  GPR_ASSERT(fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0);
+  *socket_fd = fd;
+
+  /* Use local address for test */
+  sin->sin_family = AF_INET;
+  sin->sin_addr.s_addr = 0;
+  sin->sin_port = htons(port);
+}
+
+/* Dummy gRPC callback */
+void no_op_cb(void *arg, enum grpc_em_cb_status status) {}
+
+/* =======An upload server to test notify_on_read===========
+   The server simply reads and counts a stream of bytes. */
+
+/* An upload server. */
+typedef struct {
+  grpc_em em;               /* event manger used by the sever */
+  grpc_em_fd em_fd;         /* listening fd */
+  ssize_t read_bytes_total; /* total number of received bytes */
+  gpr_mu mu;                /* protect done and done_cv */
+  gpr_cv done_cv;           /* signaled when a server finishes serving */
+  int done;                 /* set to 1 when a server finishes serving */
+} server;
+
+static void server_init(server *sv) {
+  GPR_ASSERT(grpc_em_init(&sv->em) == GRPC_EM_OK);
+  sv->read_bytes_total = 0;
+  gpr_mu_init(&sv->mu);
+  gpr_cv_init(&sv->done_cv);
+  sv->done = 0;
+}
+
+/* An upload session.
+   Created when a new upload request arrives in the server. */
+typedef struct {
+  server *sv;              /* not owned by a single session */
+  grpc_em_fd em_fd;        /* fd to read upload bytes */
+  char read_buf[BUF_SIZE]; /* buffer to store upload bytes */
+} session;
+
+/* Called when an upload session can be safely shutdown.
+   Close session FD and start to shutdown listen FD. */
+static void session_shutdown_cb(void *arg, /*session*/
+                                enum grpc_em_cb_status status) {
+  session *se = arg;
+  server *sv = se->sv;
+  grpc_em_fd_destroy(&se->em_fd);
+  gpr_free(se);
+  /* Start to shutdown listen fd. */
+  grpc_em_fd_shutdown(&sv->em_fd);
+}
+
+/* Called when data become readable in a session. */
+static void session_read_cb(void *arg, /*session*/
+                            enum grpc_em_cb_status status) {
+  session *se = arg;
+  int fd = grpc_em_fd_get(&se->em_fd);
+
+  ssize_t read_once = 0;
+  ssize_t read_total = 0;
+
+  if (status == GRPC_CALLBACK_CANCELLED) {
+    close(fd);
+    session_shutdown_cb(arg, GRPC_CALLBACK_SUCCESS);
+    return;
+  }
+
+  do {
+    read_once = read(fd, se->read_buf, BUF_SIZE);
+    if (read_once > 0) read_total += read_once;
+  } while (read_once > 0);
+  se->sv->read_bytes_total += read_total;
+
+  /* read() returns 0 to indicate the TCP connection was closed by the client.
+     read(fd, read_buf, 0) also returns 0 which should never be called as such.
+     It is possible to read nothing due to spurious edge event or data has
+     been drained, In such a case, read() returns -1 and set errno to EAGAIN. */
+  if (read_once == 0) {
+    grpc_em_fd_shutdown(&se->em_fd);
+    grpc_em_fd_notify_on_read(&se->em_fd, session_read_cb, se, gpr_inf_future);
+  } else if (read_once == -1) {
+    if (errno == EAGAIN) {
+      /* An edge triggered event is cached in the kernel until next poll.
+         In the current single thread implementation, session_read_cb is called
+         in the polling thread, such that polling only happens after this
+         callback, and will catch read edge event if data is available again
+         before notify_on_read.
+         TODO(chenw): in multi-threaded version, callback and polling can be
+         run in different threads. polling may catch a persist read edge event
+         before notify_on_read is called.  */
+      GPR_ASSERT(grpc_em_fd_notify_on_read(&se->em_fd, session_read_cb, se,
+                                           gpr_inf_future) == GRPC_EM_OK);
+    } else {
+      gpr_log(GPR_ERROR, "Unhandled read error %s", strerror(errno));
+      GPR_ASSERT(0);
+    }
+  }
+}
+
+/* Called when the listen FD can be safely shutdown.
+   Close listen FD and signal that server can be shutdown. */
+static void listen_shutdown_cb(void *arg /*server*/,
+                               enum grpc_em_cb_status status) {
+  server *sv = arg;
+
+  close(grpc_em_fd_get(&sv->em_fd));
+  grpc_em_fd_destroy(&sv->em_fd);
+
+  gpr_mu_lock(&sv->mu);
+  sv->done = 1;
+  gpr_cv_signal(&sv->done_cv);
+  gpr_mu_unlock(&sv->mu);
+}
+
+/* Called when a new TCP connection request arrives in the listening port. */
+static void listen_cb(void *arg, /*=sv_arg*/
+                      enum grpc_em_cb_status status) {
+  server *sv = arg;
+  int fd;
+  int flags;
+  session *se;
+  struct sockaddr_storage ss;
+  socklen_t slen = sizeof(ss);
+  struct grpc_em_fd *listen_em_fd = &sv->em_fd;
+
+  if (status == GRPC_CALLBACK_CANCELLED) {
+    listen_shutdown_cb(arg, GRPC_CALLBACK_SUCCESS);
+    return;
+  }
+
+  fd = accept(grpc_em_fd_get(listen_em_fd), (struct sockaddr *)&ss, &slen);
+  GPR_ASSERT(fd >= 0);
+  GPR_ASSERT(fd < FD_SETSIZE);
+  flags = fcntl(fd, F_GETFL, 0);
+  fcntl(fd, F_SETFL, flags | O_NONBLOCK);
+  se = gpr_malloc(sizeof(*se));
+  se->sv = sv;
+  GPR_ASSERT(grpc_em_fd_init(&se->em_fd, &sv->em, fd) == GRPC_EM_OK);
+  GPR_ASSERT(grpc_em_fd_notify_on_read(&se->em_fd, session_read_cb, se,
+                                       gpr_inf_future) == GRPC_EM_OK);
+
+  GPR_ASSERT(grpc_em_fd_notify_on_read(listen_em_fd, listen_cb, sv,
+                                       gpr_inf_future) == GRPC_EM_OK);
+}
+
+/* Max number of connections pending to be accepted by listen(). */
+#define MAX_NUM_FD 1024
+
+/* Start a test server, return the TCP listening port bound to listen_fd.
+   listen_cb() is registered to be interested in reading from listen_fd.
+   When connection request arrives, listen_cb() is called to accept the
+   connection request. */
+static int server_start(server *sv) {
+  int port = 0;
+  int fd;
+  struct sockaddr_in sin;
+  socklen_t addr_len;
+
+  create_test_socket(port, &fd, &sin);
+  addr_len = sizeof(sin);
+  GPR_ASSERT(bind(fd, (struct sockaddr *)&sin, addr_len) == 0);
+  GPR_ASSERT(getsockname(fd, (struct sockaddr *)&sin, &addr_len) == GRPC_EM_OK);
+  port = ntohs(sin.sin_port);
+  GPR_ASSERT(listen(fd, MAX_NUM_FD) == 0);
+
+  GPR_ASSERT(grpc_em_fd_init(&sv->em_fd, &sv->em, fd) == GRPC_EM_OK);
+  /* Register to be interested in reading from listen_fd. */
+  GPR_ASSERT(grpc_em_fd_notify_on_read(&sv->em_fd, listen_cb, sv,
+                                       gpr_inf_future) == GRPC_EM_OK);
+
+  return port;
+}
+
+/* Wait and shutdown a sever. */
+static void server_wait_and_shutdown(server *sv) {
+  gpr_mu_lock(&sv->mu);
+  while (!sv->done) gpr_cv_wait(&sv->done_cv, &sv->mu, gpr_inf_future);
+  gpr_mu_unlock(&sv->mu);
+
+  gpr_mu_destroy(&sv->mu);
+  gpr_cv_destroy(&sv->done_cv);
+
+  GPR_ASSERT(grpc_em_destroy(&sv->em) == GRPC_EM_OK);
+}
+
+/* ===An upload client to test notify_on_write=== */
+
+/* Client write buffer size */
+#define CLIENT_WRITE_BUF_SIZE 10
+/* Total number of times that the client fills up the write buffer */
+#define CLIENT_TOTAL_WRITE_CNT 3
+
+/* An upload client. */
+typedef struct {
+  grpc_em em;
+  grpc_em_fd em_fd;
+  char write_buf[CLIENT_WRITE_BUF_SIZE];
+  ssize_t write_bytes_total;
+  /* Number of times that the client fills up the write buffer and calls
+     notify_on_write to schedule another write. */
+  int client_write_cnt;
+
+  gpr_mu mu;      /* protect done and done_cv */
+  gpr_cv done_cv; /* signaled when a client finishes sending */
+  int done;       /* set to 1 when a client finishes sending */
+} client;
+
+static void client_init(client *cl) {
+  GPR_ASSERT(grpc_em_init(&cl->em) == GRPC_EM_OK);
+  memset(cl->write_buf, 0, sizeof(cl->write_buf));
+  cl->write_bytes_total = 0;
+  cl->client_write_cnt = 0;
+  gpr_mu_init(&cl->mu);
+  gpr_cv_init(&cl->done_cv);
+  cl->done = 0;
+}
+
+/* Called when a client upload session is ready to shutdown. */
+static void client_session_shutdown_cb(void *arg /*client*/,
+                                       enum grpc_em_cb_status status) {
+  client *cl = arg;
+  grpc_em_fd_destroy(&cl->em_fd);
+  gpr_mu_lock(&cl->mu);
+  cl->done = 1;
+  gpr_cv_signal(&cl->done_cv);
+  gpr_mu_unlock(&cl->mu);
+}
+
+/* Write as much as possible, then register notify_on_write. */
+static void client_session_write(void *arg, /*client*/
+                                 enum grpc_em_cb_status status) {
+  client *cl = arg;
+  int fd = grpc_em_fd_get(&cl->em_fd);
+  ssize_t write_once = 0;
+
+  if (status == GRPC_CALLBACK_CANCELLED) {
+    client_session_shutdown_cb(arg, GRPC_CALLBACK_SUCCESS);
+    return;
+  }
+
+  do {
+    write_once = write(fd, cl->write_buf, CLIENT_WRITE_BUF_SIZE);
+    if (write_once > 0) cl->write_bytes_total += write_once;
+  } while (write_once > 0);
+
+  if (errno == EAGAIN) {
+    gpr_mu_lock(&cl->mu);
+    if (cl->client_write_cnt < CLIENT_TOTAL_WRITE_CNT) {
+      GPR_ASSERT(grpc_em_fd_notify_on_write(&cl->em_fd, client_session_write,
+                                            cl, gpr_inf_future) == GRPC_EM_OK);
+      cl->client_write_cnt++;
+    } else {
+      close(fd);
+      grpc_em_fd_shutdown(&cl->em_fd);
+      grpc_em_fd_notify_on_write(&cl->em_fd, client_session_write, cl,
+                                 gpr_inf_future);
+    }
+    gpr_mu_unlock(&cl->mu);
+  } else {
+    gpr_log(GPR_ERROR, "unknown errno %s", strerror(errno));
+    GPR_ASSERT(0);
+  }
+}
+
+/* Start a client to send a stream of bytes. */
+static void client_start(client *cl, int port) {
+  int fd;
+  struct sockaddr_in sin;
+  create_test_socket(port, &fd, &sin);
+  if (connect(fd, (struct sockaddr *)&sin, sizeof(sin)) == -1 &&
+      errno != EINPROGRESS) {
+    gpr_log(GPR_ERROR, "Failed to connect to the server");
+    GPR_ASSERT(0);
+  }
+
+  GPR_ASSERT(grpc_em_fd_init(&cl->em_fd, &cl->em, fd) == GRPC_EM_OK);
+
+  client_session_write(cl, GRPC_CALLBACK_SUCCESS);
+}
+
+/* Wait for the signal to shutdown a client. */
+static void client_wait_and_shutdown(client *cl) {
+  gpr_mu_lock(&cl->mu);
+  while (!cl->done) gpr_cv_wait(&cl->done_cv, &cl->mu, gpr_inf_future);
+  gpr_mu_unlock(&cl->mu);
+
+  gpr_mu_destroy(&cl->mu);
+  gpr_cv_destroy(&cl->done_cv);
+
+  GPR_ASSERT(grpc_em_destroy(&cl->em) == GRPC_EM_OK);
+}
+
+/* Test grpc_em_fd. Start an upload server and client, upload a stream of
+   bytes from the client to the server, and verify that the total number of
+   sent bytes is equal to the total number of received bytes. */
+static void test_grpc_em_fd() {
+  server sv;
+  client cl;
+  int port;
+
+  server_init(&sv);
+  port = server_start(&sv);
+  client_init(&cl);
+  client_start(&cl, port);
+  client_wait_and_shutdown(&cl);
+  server_wait_and_shutdown(&sv);
+  GPR_ASSERT(sv.read_bytes_total == cl.write_bytes_total);
+  gpr_log(GPR_INFO, "Total read bytes %d", sv.read_bytes_total);
+}
+
+typedef struct fd_change_data {
+  gpr_mu mu;
+  gpr_cv cv;
+  void (*cb_that_ran)(void *, enum grpc_em_cb_status);
+} fd_change_data;
+
+void init_change_data(fd_change_data *fdc) {
+  gpr_mu_init(&fdc->mu);
+  gpr_cv_init(&fdc->cv);
+  fdc->cb_that_ran = NULL;
+}
+
+void destroy_change_data(fd_change_data *fdc) {
+  gpr_mu_destroy(&fdc->mu);
+  gpr_cv_destroy(&fdc->cv);
+}
+
+static void first_read_callback(void *arg /* fd_change_data */,
+                                enum grpc_em_cb_status status) {
+  fd_change_data *fdc = arg;
+
+  gpr_mu_lock(&fdc->mu);
+  fdc->cb_that_ran = first_read_callback;
+  gpr_cv_signal(&fdc->cv);
+  gpr_mu_unlock(&fdc->mu);
+}
+
+static void second_read_callback(void *arg /* fd_change_data */,
+                                 enum grpc_em_cb_status status) {
+  fd_change_data *fdc = arg;
+
+  gpr_mu_lock(&fdc->mu);
+  fdc->cb_that_ran = second_read_callback;
+  gpr_cv_signal(&fdc->cv);
+  gpr_mu_unlock(&fdc->mu);
+}
+
+/* Test that changing the callback we use for notify_on_read actually works.
+   Note that we have two different but almost identical callbacks above -- the
+   point is to have two different function pointers and two different data
+   pointers and make sure that changing both really works. */
+static void test_grpc_em_fd_change() {
+  grpc_em em;
+  grpc_em_fd em_fd;
+  fd_change_data a, b;
+  int flags;
+  int sv[2];
+  char data;
+  int result;
+
+  init_change_data(&a);
+  init_change_data(&b);
+
+  GPR_ASSERT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0);
+  flags = fcntl(sv[0], F_GETFL, 0);
+  GPR_ASSERT(fcntl(sv[0], F_SETFL, flags | O_NONBLOCK) == 0);
+  flags = fcntl(sv[1], F_GETFL, 0);
+  GPR_ASSERT(fcntl(sv[1], F_SETFL, flags | O_NONBLOCK) == 0);
+
+  grpc_em_init(&em);
+  grpc_em_fd_init(&em_fd, &em, sv[0]);
+
+  /* Register the first callback, then make its FD readable */
+  grpc_em_fd_notify_on_read(&em_fd, first_read_callback, &a, gpr_inf_future);
+  data = 0;
+  result = write(sv[1], &data, 1);
+  GPR_ASSERT(result == 1);
+
+  /* And now wait for it to run. */
+  gpr_mu_lock(&a.mu);
+  while (a.cb_that_ran == NULL) {
+    gpr_cv_wait(&a.cv, &a.mu, gpr_inf_future);
+  }
+  GPR_ASSERT(a.cb_that_ran == first_read_callback);
+  gpr_mu_unlock(&a.mu);
+
+  /* And drain the socket so we can generate a new read edge */
+  result = read(sv[0], &data, 1);
+  GPR_ASSERT(result == 1);
+
+  /* Now register a second callback with distinct change data, and do the same
+     thing again. */
+  grpc_em_fd_notify_on_read(&em_fd, second_read_callback, &b, gpr_inf_future);
+  data = 0;
+  result = write(sv[1], &data, 1);
+  GPR_ASSERT(result == 1);
+
+  gpr_mu_lock(&b.mu);
+  while (b.cb_that_ran == NULL) {
+    gpr_cv_wait(&b.cv, &b.mu, gpr_inf_future);
+  }
+  /* Except now we verify that second_read_callback ran instead */
+  GPR_ASSERT(b.cb_that_ran == second_read_callback);
+  gpr_mu_unlock(&b.mu);
+
+  grpc_em_fd_destroy(&em_fd);
+  grpc_em_destroy(&em);
+  destroy_change_data(&a);
+  destroy_change_data(&b);
+  close(sv[0]);
+  close(sv[1]);
+}
+
+void timeout_callback(void *arg, enum grpc_em_cb_status status) {
+  if (status == GRPC_CALLBACK_TIMED_OUT) {
+    gpr_event_set(arg, (void *)1);
+  } else {
+    gpr_event_set(arg, (void *)2);
+  }
+}
+
+void test_grpc_em_fd_notify_timeout() {
+  grpc_em em;
+  grpc_em_fd em_fd;
+  gpr_event ev;
+  int flags;
+  int sv[2];
+  gpr_timespec timeout;
+  gpr_timespec deadline;
+
+  gpr_event_init(&ev);
+
+  GPR_ASSERT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0);
+  flags = fcntl(sv[0], F_GETFL, 0);
+  GPR_ASSERT(fcntl(sv[0], F_SETFL, flags | O_NONBLOCK) == 0);
+  flags = fcntl(sv[1], F_GETFL, 0);
+  GPR_ASSERT(fcntl(sv[1], F_SETFL, flags | O_NONBLOCK) == 0);
+
+  grpc_em_init(&em);
+  grpc_em_fd_init(&em_fd, &em, sv[0]);
+
+  timeout = gpr_time_from_micros(1000000);
+  deadline = gpr_time_add(gpr_now(), timeout);
+
+  grpc_em_fd_notify_on_read(&em_fd, timeout_callback, &ev, deadline);
+
+  GPR_ASSERT(gpr_event_wait(&ev, gpr_time_add(deadline, timeout)));
+
+  GPR_ASSERT(gpr_event_get(&ev) == (void *)1);
+  grpc_em_fd_destroy(&em_fd);
+  grpc_em_destroy(&em);
+  close(sv[0]);
+  close(sv[1]);
+}
+
+typedef struct {
+  grpc_em *em;
+  gpr_cv cv;
+  gpr_mu mu;
+  int counter;
+  int done_success_ctr;
+  int done_cancel_ctr;
+  int done;
+  gpr_event fcb_arg;
+  grpc_em_cb_status status;
+} alarm_arg;
+
+static void followup_cb(void *arg, grpc_em_cb_status status) {
+  gpr_event_set((gpr_event *)arg, arg);
+}
+
+/* Called when an alarm expires. */
+static void alarm_cb(void *arg /* alarm_arg */, grpc_em_cb_status status) {
+  alarm_arg *a = arg;
+  gpr_mu_lock(&a->mu);
+  if (status == GRPC_CALLBACK_SUCCESS) {
+    a->counter++;
+    a->done_success_ctr++;
+  } else if (status == GRPC_CALLBACK_CANCELLED) {
+    a->done_cancel_ctr++;
+  } else {
+    GPR_ASSERT(0);
+  }
+  a->done = 1;
+  a->status = status;
+  gpr_cv_signal(&a->cv);
+  gpr_mu_unlock(&a->mu);
+  grpc_em_add_callback(a->em, followup_cb, &a->fcb_arg);
+}
+
+/* Test grpc_em_alarm add and cancel. */
+static void test_grpc_em_alarm() {
+  struct grpc_em em;
+  struct grpc_em_alarm alarm;
+  struct grpc_em_alarm alarm_to_cancel;
+  gpr_timespec tv0 = {0, 1};
+  /* Timeout on the alarm cond. var, so make big enough to absorb time
+     deviations. Otherwise, operations after wait will not be properly ordered
+   */
+  gpr_timespec tv1 = gpr_time_from_micros(200000);
+  gpr_timespec tv2 = {0, 1};
+  gpr_timespec alarm_deadline;
+  gpr_timespec followup_deadline;
+
+  alarm_arg *cancel_arg = NULL;
+  alarm_arg arg;
+  alarm_arg arg2;
+  void *fdone;
+
+  GPR_ASSERT(grpc_em_init(&em) == GRPC_EM_OK);
+
+  arg.em = &em;
+  arg.counter = 0;
+  arg.status = GRPC_CALLBACK_DO_NOT_USE;
+  arg.done_success_ctr = 0;
+  arg.done_cancel_ctr = 0;
+  arg.done = 0;
+  gpr_mu_init(&arg.mu);
+  gpr_cv_init(&arg.cv);
+  gpr_event_init(&arg.fcb_arg);
+
+  GPR_ASSERT(grpc_em_alarm_init(&alarm, &em, alarm_cb, &arg) == GRPC_EM_OK);
+  GPR_ASSERT(grpc_em_alarm_add(&alarm, gpr_time_add(tv0, gpr_now())) ==
+             GRPC_EM_OK);
+
+  alarm_deadline = gpr_time_add(gpr_now(), tv1);
+  gpr_mu_lock(&arg.mu);
+  while (arg.done == 0) {
+    gpr_cv_wait(&arg.cv, &arg.mu, alarm_deadline);
+  }
+  gpr_mu_unlock(&arg.mu);
+
+  followup_deadline = gpr_time_add(gpr_now(), tv1);
+  fdone = gpr_event_wait(&arg.fcb_arg, followup_deadline);
+
+  if (arg.counter != 1) {
+    gpr_log(GPR_ERROR, "Alarm callback not called");
+    GPR_ASSERT(0);
+  } else if (arg.done_success_ctr != 1) {
+    gpr_log(GPR_ERROR, "Alarm done callback not called with success");
+    GPR_ASSERT(0);
+  } else if (arg.done_cancel_ctr != 0) {
+    gpr_log(GPR_ERROR, "Alarm done callback called with cancel");
+    GPR_ASSERT(0);
+  } else if (arg.status == GRPC_CALLBACK_DO_NOT_USE) {
+    gpr_log(GPR_ERROR, "Alarm callback without status");
+    GPR_ASSERT(0);
+  } else {
+    gpr_log(GPR_INFO, "Alarm callback called successfully");
+  }
+
+  if (fdone != (void *)&arg.fcb_arg) {
+    gpr_log(GPR_ERROR, "Followup callback #1 not invoked properly %p %p", fdone,
+            &arg.fcb_arg);
+    GPR_ASSERT(0);
+  }
+  gpr_cv_destroy(&arg.cv);
+  gpr_mu_destroy(&arg.mu);
+
+  arg2.em = &em;
+  arg2.counter = 0;
+  arg2.status = GRPC_CALLBACK_DO_NOT_USE;
+  arg2.done_success_ctr = 0;
+  arg2.done_cancel_ctr = 0;
+  arg2.done = 0;
+  gpr_mu_init(&arg2.mu);
+  gpr_cv_init(&arg2.cv);
+  gpr_event_init(&arg2.fcb_arg);
+
+  GPR_ASSERT(grpc_em_alarm_init(&alarm_to_cancel, &em, alarm_cb, &arg2) ==
+             GRPC_EM_OK);
+  GPR_ASSERT(grpc_em_alarm_add(&alarm_to_cancel,
+                               gpr_time_add(tv2, gpr_now())) == GRPC_EM_OK);
+  switch (grpc_em_alarm_cancel(&alarm_to_cancel, (void **)&cancel_arg)) {
+    case GRPC_EM_OK:
+      gpr_log(GPR_INFO, "Alarm cancel succeeded");
+      break;
+    case GRPC_EM_ERROR:
+      gpr_log(GPR_ERROR, "Alarm cancel failed");
+      GPR_ASSERT(0);
+      break;
+    case GRPC_EM_INVALID_ARGUMENTS:
+      gpr_log(GPR_ERROR, "Alarm cancel failed with bad response code");
+      gpr_log(GPR_ERROR, "Current value of triggered is %d\n",
+              (int)alarm_to_cancel.triggered);
+      GPR_ASSERT(0);
+      break;
+  }
+
+  alarm_deadline = gpr_time_add(gpr_now(), tv1);
+  gpr_mu_lock(&arg2.mu);
+  while (arg2.done == 0) {
+    gpr_cv_wait(&arg2.cv, &arg2.mu, alarm_deadline);
+  }
+  gpr_mu_unlock(&arg2.mu);
+
+  followup_deadline = gpr_time_add(gpr_now(), tv1);
+  fdone = gpr_event_wait(&arg2.fcb_arg, followup_deadline);
+
+  if (arg2.counter != arg2.done_success_ctr) {
+    gpr_log(GPR_ERROR, "Alarm callback called but didn't lead to done success");
+    GPR_ASSERT(0);
+  } else if (arg2.done_success_ctr && arg2.done_cancel_ctr) {
+    gpr_log(GPR_ERROR, "Alarm done callback called with success and cancel");
+    GPR_ASSERT(0);
+  } else if (arg2.done_cancel_ctr + arg2.done_success_ctr != 1) {
+    gpr_log(GPR_ERROR, "Alarm done callback called incorrect number of times");
+    GPR_ASSERT(0);
+  } else if (arg2.status == GRPC_CALLBACK_DO_NOT_USE) {
+    gpr_log(GPR_ERROR, "Alarm callback without status");
+    GPR_ASSERT(0);
+  } else if (arg2.done_success_ctr) {
+    gpr_log(GPR_INFO, "Alarm callback executed before cancel");
+    gpr_log(GPR_INFO, "Current value of triggered is %d\n",
+            (int)alarm_to_cancel.triggered);
+  } else if (arg2.done_cancel_ctr) {
+    gpr_log(GPR_INFO, "Alarm callback canceled");
+    gpr_log(GPR_INFO, "Current value of triggered is %d\n",
+            (int)alarm_to_cancel.triggered);
+  } else {
+    gpr_log(GPR_ERROR, "Alarm cancel test should not be here");
+    GPR_ASSERT(0);
+  }
+
+  if (cancel_arg != &arg2) {
+    gpr_log(GPR_ERROR, "Alarm cancel arg address wrong");
+    GPR_ASSERT(0);
+  }
+  if (fdone != (void *)&arg2.fcb_arg) {
+    gpr_log(GPR_ERROR, "Followup callback #2 not invoked properly %p %p", fdone,
+            &arg2.fcb_arg);
+    GPR_ASSERT(0);
+  }
+  gpr_cv_destroy(&arg2.cv);
+  gpr_mu_destroy(&arg2.mu);
+
+  GPR_ASSERT(grpc_em_destroy(&em) == GRPC_EM_OK);
+}
+
+int main(int argc, char **argv) {
+  grpc_test_init(argc, argv);
+  test_grpc_em_alarm();
+  test_grpc_em_fd();
+  test_grpc_em_fd_change();
+  test_grpc_em_fd_notify_timeout();
+  return 0;
+}