Refactor of async code. poll() is now only called when needed.

[awesomized/libmemcached] / lib / memcached_io.c

diff --git a/lib/memcached_io.c b/lib/memcached_io.c
index 06ac2e9b51862f3929429d754ea12ba30b272075..daf64755c69ce171234c4c026aee925009aad9eb 100644 (file)
--- a/lib/memcached_io.c
+++ b/lib/memcached_io.c
@@ -7,12 +7,22 @@
 #include <sys/select.h>
 #include <poll.h>
 
-static int io_wait(memcached_st *ptr, unsigned int server_key, unsigned read_or_write)
+typedef enum {
+  MEM_READ,
+  MEM_WRITE,
+} memc_read_or_write;
+
+static ssize_t io_flush(memcached_st *ptr, unsigned int server_key, 
+                                  memcached_return *error);
+
+static memcached_return io_wait(memcached_st *ptr, unsigned int server_key, 
+                                memc_read_or_write read_or_write)
 {
   struct pollfd fds[1];
   short flags= 0;
+  int error;
 
-  if (read_or_write)
+  if (read_or_write == MEM_WRITE) /* write */
     flags= POLLOUT |  POLLERR;
   else
     flags= POLLIN | POLLERR;
@@ -21,10 +31,22 @@ static int io_wait(memcached_st *ptr, unsigned int server_key, unsigned read_or_
   fds[0].fd= ptr->hosts[server_key].fd;
   fds[0].events= flags;
 
-  if (poll(fds, 1, -1) < 0)
-    return MEMCACHED_FAILURE;
+  error= poll(fds, 1, ptr->poll_timeout);
+
+  if (error == 1)
+    return MEMCACHED_SUCCESS;
+  else if (error == 0)
+  {
+    WATCHPOINT_NUMBER(read_or_write);
+    return MEMCACHED_TIMEOUT;
+  }
+
+  WATCHPOINT;
+  /* Imposssible for anything other then -1 */
+  WATCHPOINT_ASSERT(error == -1);
+  memcached_quit_server(ptr, server_key, 1);
+  return MEMCACHED_FAILURE;
 
-  return MEMCACHED_SUCCESS;
 }
 
 ssize_t memcached_io_read(memcached_st *ptr, unsigned  int server_key,
@@ -42,15 +64,6 @@ ssize_t memcached_io_read(memcached_st *ptr, unsigned  int server_key,
 
       while (1)
       {
-        if (ptr->flags & MEM_NO_BLOCK)
-        {
-          memcached_return rc;
-
-          rc= io_wait(ptr, server_key, 0);
-          if (rc != MEMCACHED_SUCCESS)
-            return -1;
-        }
-
         data_read= read(ptr->hosts[server_key].fd, 
                         ptr->hosts[server_key].read_buffer, 
                         MEMCACHED_MAX_BUFFER);
@@ -59,9 +72,19 @@ ssize_t memcached_io_read(memcached_st *ptr, unsigned  int server_key,
           switch (errno)
           {
           case EAGAIN:
-            break;
+            {
+              memcached_return rc;
+              rc= io_wait(ptr, server_key, MEM_READ);
+
+              if (rc == MEMCACHED_SUCCESS)
+                continue;
+
+              memcached_quit_server(ptr, server_key, 1);
+              return -1;
+            }
           default:
             {
+              memcached_quit_server(ptr, server_key, 1);
               ptr->cached_errno= errno;
               return -1;
             }
@@ -102,9 +125,12 @@ ssize_t memcached_io_write(memcached_st *ptr, unsigned int server_key,
 
     if (ptr->hosts[server_key].write_buffer_offset == MEMCACHED_MAX_BUFFER)
     {
+      memcached_return rc;
       size_t sent_length;
 
-      sent_length= memcached_io_flush(ptr, server_key);
+      sent_length= io_flush(ptr, server_key, &rc);
+      if (sent_length == -1)
+        return -1;
 
       WATCHPOINT_ASSERT(sent_length == MEMCACHED_MAX_BUFFER);
       ptr->hosts[server_key].write_ptr= ptr->hosts[server_key].write_buffer;
@@ -114,20 +140,57 @@ ssize_t memcached_io_write(memcached_st *ptr, unsigned int server_key,
 
   if (with_flush)
   {
-    if (memcached_io_flush(ptr, server_key) == -1)
+    memcached_return rc;
+    if (io_flush(ptr, server_key, &rc) == -1)
       return -1;
   }
 
   return length;
 }
 
-ssize_t memcached_io_flush(memcached_st *ptr, unsigned int server_key)
+memcached_return memcached_io_close(memcached_st *ptr, unsigned int server_key)
+{
+  memcached_return rc;
+
+  rc= MEMCACHED_SUCCESS;
+  if (ptr->flags & MEM_NO_BLOCK)
+  {
+    int error;
+    struct pollfd fds[1];
+    short flags= 0;
+
+    flags= POLLHUP |  POLLERR;
+
+    memset(&fds, 0, sizeof(struct pollfd));
+    fds[0].fd= ptr->hosts[server_key].fd;
+    fds[0].events= flags;
+    fds[0].revents= 0;
+
+    error= poll(fds, 1, ptr->poll_timeout == -1 ? 100 : ptr->poll_timeout);
+
+    if (error == -1)
+    {
+      memcached_quit_server(ptr, server_key, 1);
+      return MEMCACHED_FAILURE;
+    }
+    else if (error == 0)
+      return MEMCACHED_FAILURE; /* Timeout occurred */
+  }
+
+  close(ptr->hosts[server_key].fd);
+
+  return rc;
+}
+
+static ssize_t io_flush(memcached_st *ptr, unsigned int server_key, 
+                                  memcached_return *error)
 {
   size_t sent_length;
   size_t return_length;
   char *write_ptr= ptr->hosts[server_key].write_buffer;
   size_t write_length= ptr->hosts[server_key].write_buffer_offset;
-  unsigned int loop= 1;
+
+  *error= MEMCACHED_SUCCESS;
 
   if (ptr->hosts[server_key].write_buffer_offset == 0)
     return 0;
@@ -135,21 +198,12 @@ ssize_t memcached_io_flush(memcached_st *ptr, unsigned int server_key)
   return_length= 0;
   while (write_length)
   {
-    if (ptr->flags & MEM_NO_BLOCK)
-    {
-      memcached_return rc;
-
-      rc= io_wait(ptr, server_key, 1);
-      if (rc != MEMCACHED_SUCCESS)
-        return -1;
-    }
-
     sent_length= 0;
     if (ptr->hosts[server_key].type == MEMCACHED_CONNECTION_UDP)
     {
       sent_length= sendto(ptr->hosts[server_key].fd, 
                           write_ptr, write_length, 0, 
-                          (struct sockaddr *)&ptr->hosts[server_key].servAddr, 
+                          (struct sockaddr *)&ptr->hosts[server_key].address_info->ai_addr, 
                           sizeof(struct sockaddr));
     }
     else
@@ -160,17 +214,22 @@ ssize_t memcached_io_flush(memcached_st *ptr, unsigned int server_key)
         switch (errno)
         {
         case ENOBUFS:
-        case EAGAIN:
-          WATCHPOINT;
           continue;
-          if (loop < 100)
+        case EAGAIN:
           {
-            loop++;
-            break;
+            memcached_return rc;
+            rc= io_wait(ptr, server_key, MEM_WRITE);
+
+            if (rc == MEMCACHED_SUCCESS)
+              continue;
+
+            memcached_quit_server(ptr, server_key, 1);
+            return -1;
           }
-          /* Yes, we want to fall through */
         default:
+          memcached_quit_server(ptr, server_key, 1);
           ptr->cached_errno= errno;
+          *error= MEMCACHED_ERRNO;
           return -1;
         }
       }