unix: fix test-callback-stack

oio-unix.c

@@ -388,6 +388,35 @@ void oio__next(EV_P_ ev_idle* watcher, int revents) {
 }
 
 
+static void oio__drain(oio_handle* handle) {
+  assert(!oio_write_queue_head(handle));
+  assert(handle->write_queue_size == 0);
+
+  ev_io_stop(EV_DEFAULT_ &handle->write_watcher);
+
+  /* Shutdown? */
+  if (oio_flag_is_set(handle, OIO_SHUTTING) &&
+      !oio_flag_is_set(handle, OIO_CLOSING) &&
+      !oio_flag_is_set(handle, OIO_SHUT)) {
+    assert(handle->shutdown_req);
+
+    oio_req* req = handle->shutdown_req;
+    oio_shutdown_cb cb = req->cb;
+
+    if (shutdown(handle->fd, SHUT_WR)) {
+      /* Error. Nothing we can do, close the handle. */
+      oio_err_new(handle, errno);
+      oio_close(handle);
+      if (cb) cb(req, -1);
+    } else {
+      oio_err_new(handle, 0);
+      oio_flag_set(handle, OIO_SHUT);
+      if (cb) cb(req, 0);
+    }
+  }
+}
+
+
 void oio__write(oio_handle* handle) {
   assert(handle->fd >= 0);
 
@@ -396,8 +425,8 @@ void oio__write(oio_handle* handle) {
   /* Get the request at the head of the queue. */
   oio_req* req = oio_write_queue_head(handle);
   if (!req) {
-    /* This probably shouldn't happen. Maybe assert(0) here. */
-    ev_io_stop(EV_DEFAULT_ &handle->write_watcher);
+    assert(handle->write_queue_size == 0);
+    oio__drain(handle);
     return;
   }
 
@@ -477,28 +506,7 @@ void oio__write(oio_handle* handle) {
             assert(handle->write_queue_size > 0);
           } else {
             /* Write queue drained. */
-            assert(handle->write_queue_size == 0);
-            ev_io_stop(EV_DEFAULT_ &handle->write_watcher);
-
-            if (oio_flag_is_set(handle, OIO_SHUTTING) &&
-                !oio_flag_is_set(handle, OIO_CLOSING) &&
-                !oio_flag_is_set(handle, OIO_SHUT)) {
-              assert(handle->shutdown_req);
-
-              req = handle->shutdown_req;
-              oio_shutdown_cb cb = req->cb;
-
-              if (shutdown(handle->fd, SHUT_WR)) {
-                /* Error. Nothing we can do, close the handle. */
-                oio_err_new(handle, errno);
-                oio_close(handle);
-                if (cb) cb(req, -1);
-              } else {
-                oio_err_new(handle, 0);
-                oio_flag_set(handle, OIO_SHUT);
-                if (cb) cb(req, 0);
-              }
-            }
+            oio__drain(handle);
           }
 
           return;
@@ -577,6 +585,8 @@ int oio_shutdown(oio_req* req) {
 
   oio_flag_set(handle, OIO_SHUTTING);
 
+  ev_io_start(EV_DEFAULT_UC_ &handle->write_watcher);
+
   return 0;
 }
 

test/test-callback-stack.c

@@ -61,7 +61,14 @@ void shutdown_cb(oio_req* req, int status) {
 void read_cb(oio_handle* handle, int nread, oio_buf buf) {
   ASSERT(nested == 0 && "read_cb must be called from a fresh stack");
 
-  if (nread == -1) {
+  printf("Read. nread == %d\n", nread);
+  free(buf.base);
+
+  if (nread == 0) {
+    ASSERT(oio_last_error().code == OIO_EAGAIN);
+    return;
+
+  } else if (nread == -1) {
     ASSERT(oio_last_error().code == OIO_EOF);
 
     nested++;
@@ -74,7 +81,6 @@ void read_cb(oio_handle* handle, int nread, oio_buf buf) {
   }
 
   bytes_received += nread;
-  free(buf.base);
 
   /* We call shutdown here because when bytes_received == sizeof MESSAGE */
   /* there will be no more data sent nor received, so here it would be */
@@ -83,6 +89,9 @@ void read_cb(oio_handle* handle, int nread, oio_buf buf) {
   if (bytes_received == sizeof MESSAGE) {
     nested++;
     oio_req_init(&shutdown_req, handle, shutdown_cb);
+
+    puts("Shutdown");
+
     if (oio_shutdown(&shutdown_req)) {
       FATAL("oio_shutdown failed");
     }
@@ -95,6 +104,8 @@ void timeout_cb(oio_req* req, int64_t skew, int status) {
   ASSERT(status == 0);
   ASSERT(nested == 0 && "timeout_cb must be called from a fresh stack");
 
+  puts("Timeout complete. Now read data...");
+
   nested++;
   if (oio_read_start(&client, read_cb)) {
     FATAL("oio_read_start failed");
@@ -109,6 +120,8 @@ void write_cb(oio_req* req, int status) {
   ASSERT(status == 0);
   ASSERT(nested == 0 && "write_cb must be called from a fresh stack");
 
+  puts("Data written. 500ms timeout...");
+
   /* After the data has been sent, we're going to wait for a while, then */
   /* start reading. This makes us certain that the message has been echoed */
   /* back to our receive buffer when we start reading. This maximizes the */
@@ -127,6 +140,8 @@ void write_cb(oio_req* req, int status) {
 void connect_cb(oio_req* req, int status) {
   oio_buf buf;
 
+  puts("Connected. Write some data to echo server...");
+
   ASSERT(status == 0);
   ASSERT(nested == 0 && "connect_cb must be called from a fresh stack");
 
@@ -165,6 +180,8 @@ TEST_IMPL(callback_stack) {
     FATAL("oio_tcp_init failed");
   }
 
+  puts("Connecting...");
+
   nested++;
   oio_req_init(&connect_req, &client, connect_cb);
   if (oio_connect(&connect_req, (struct sockaddr*) &addr)) {