X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=lib%2Fmemcached_io.c;h=22c54e420da76f51f888a3c4eef6b0480e3c892c;hb=69b905e54a2f5ad5038a3f75c269bbbb4c3e610f;hp=9277b6ba3a883a387829acc9130046843f3756f4;hpb=d3ef73cd144dfdbcaff41d3373e4458fb351f7af;p=m6w6%2Flibmemcached

diff --git a/lib/memcached_io.c b/lib/memcached_io.c
index 9277b6ba..22c54e42 100644
--- a/lib/memcached_io.c
+++ b/lib/memcached_io.c
@@ -5,151 +5,286 @@
 #include "common.h"
 #include "memcached_io.h"
 #include <sys/select.h>
+#include <poll.h>
 
-ssize_t memcached_io_read(memcached_st *ptr, unsigned  int server_key,
-                          char *buffer, size_t length)
+typedef enum {
+  MEM_READ,
+  MEM_WRITE,
+} memc_read_or_write;
+
+static ssize_t io_flush(memcached_server_st *ptr, memcached_return *error);
+
+static memcached_return io_wait(memcached_server_st *ptr,
+                                memc_read_or_write read_or_write)
 {
-  size_t x;
-  char *buffer_ptr;
+  struct pollfd fds[1];
+  short flags= 0;
+  int error;
 
-  buffer_ptr= buffer;
+  if (read_or_write == MEM_WRITE) /* write */
+    flags= POLLOUT |  POLLERR;
+  else
+    flags= POLLIN | POLLERR;
 
-  for (x= 0, buffer_ptr= buffer; 
-       x < length; x++)
+  memset(&fds, 0, sizeof(struct pollfd));
+  fds[0].fd= ptr->fd;
+  fds[0].events= flags;
+
+  error= poll(fds, 1, ptr->root->poll_timeout);
+
+  if (error == 1)
+    return MEMCACHED_SUCCESS;
+  else if (error == 0)
   {
-    if (!ptr->read_buffer_length)
-    {
-      if (length > 1)
-      {
+    return MEMCACHED_TIMEOUT;
+  }
 
-        size_t data_read;
-        data_read= recv(ptr->hosts[server_key].fd, 
-                        buffer_ptr, 
-                        length - x, 0);
-        if (data_read == -1)
-        {
-          return -1;
-        }
-        if (data_read == 0)
-          return x;
+  /* Imposssible for anything other then -1 */
+  WATCHPOINT_ASSERT(error == -1);
+  memcached_quit_server(ptr, 1);
 
-        data_read+= x;
+  return MEMCACHED_FAILURE;
 
-        return data_read;
-      }
-      else
-      {
-        size_t data_read;
-try_again:
+}
 
-        if (ptr->flags & MEM_NO_BLOCK)
-        {
-          struct timeval local_tv;
-          fd_set set;
+#ifdef UNUSED
+void memcached_io_preread(memcached_st *ptr)
+{
+  unsigned int x;
 
-          memset(&local_tv, 0, sizeof(struct timeval));
+  return;
 
-          local_tv.tv_sec= 0;
-          local_tv.tv_usec= 300;
+  for (x= 0; x < ptr->number_of_hosts; x++)
+  {
+    if (memcached_server_response_count(ptr, x) &&
+        ptr->hosts[x].read_data_length < MEMCACHED_MAX_BUFFER )
+    {
+      size_t data_read;
 
-          FD_ZERO(&set);
-          FD_SET(ptr->hosts[server_key].fd, &set);
+      data_read= read(ptr->hosts[x].fd,
+                      ptr->hosts[x].read_ptr + ptr->hosts[x].read_data_length,
+                      MEMCACHED_MAX_BUFFER - ptr->hosts[x].read_data_length);
+      if (data_read == -1)
+        continue;
 
-          select(1, &set, NULL, NULL, &local_tv);
-        }
+      ptr->hosts[x].read_buffer_length+= data_read;
+      ptr->hosts[x].read_data_length+= data_read;
+    }
+  }
+}
+#endif
+
+ssize_t memcached_io_read(memcached_server_st *ptr,
+                          char *buffer, size_t length)
+{
+  char *buffer_ptr;
 
-        data_read= recv(ptr->hosts[server_key].fd, 
+  buffer_ptr= buffer;
+
+  while (length)
+  {
+    uint8_t found_eof= 0;
+    if (!ptr->read_buffer_length)
+    {
+      size_t data_read;
+
+      while (1)
+      {
+        data_read= read(ptr->fd, 
                         ptr->read_buffer, 
-                        MEMCACHED_MAX_BUFFER, 0);
-        if (data_read == -1)
+                        MEMCACHED_MAX_BUFFER);
+        if (data_read)
+          break;
+        else if (data_read == -1)
         {
-          if (errno == EAGAIN)
-            goto try_again;
-          return -1;
+          ptr->cached_errno= errno;
+          switch (errno)
+          {
+          case EAGAIN:
+            {
+              memcached_return rc;
+
+              rc= io_wait(ptr, MEM_READ);
+
+              if (rc == MEMCACHED_SUCCESS)
+                continue;
+            }
+          /* fall trough */
+          default:
+            {
+              memcached_quit_server(ptr, 1);
+              return -1;
+            }
+          }
+        }
+        else
+        {
+          WATCHPOINT_ASSERT(0);
+          found_eof= 1;
+          break;
         }
-        ptr->read_buffer_length= data_read;
-        ptr->read_ptr= ptr->read_buffer;
       }
 
-      if (ptr->read_buffer_length == -1)
-        return -1;
-      if (ptr->read_buffer_length == 0)
-        return x;
+      ptr->read_data_length= data_read;
+      ptr->read_buffer_length= data_read;
+      ptr->read_ptr= ptr->read_buffer;
+    }
+
+    if (length > 1)
+    {
+      size_t difference;
+
+      difference= (length > ptr->read_buffer_length) ? ptr->read_buffer_length : length;
+
+      memcpy(buffer_ptr, ptr->read_ptr, difference);
+      length -= difference;
+      ptr->read_ptr+= difference;
+      ptr->read_buffer_length-= difference;
+      buffer_ptr+= difference;
     }
-    *buffer_ptr= *ptr->read_ptr;
-    buffer_ptr++;
-    ptr->read_ptr++;
-    ptr->read_buffer_length--;
+    else
+    {
+      *buffer_ptr= *ptr->read_ptr;
+      ptr->read_ptr++;
+      ptr->read_buffer_length--;
+      buffer_ptr++;
+      break;
+    }
+
+    if (found_eof)
+      break;
   }
 
-  return length;
+  return (size_t)(buffer_ptr - buffer);
 }
 
-ssize_t memcached_io_write(memcached_st *ptr, unsigned int server_key,
-                        char *buffer, size_t length, char with_flush)
+ssize_t memcached_io_write(memcached_server_st *ptr,
+                           char *buffer, size_t length, char with_flush)
 {
-  unsigned long long x;
+  size_t original_length;
+  char* buffer_ptr;
 
-  for (x= 0; x < length; x++)
+  original_length= length;
+  buffer_ptr= buffer;
+
+  while (length)
   {
-    ptr->write_buffer[ptr->write_buffer_offset]= buffer[x];
-    ptr->write_buffer_offset++;
+    char *write_ptr;
+    size_t should_write;
+
+    should_write= MEMCACHED_MAX_BUFFER - ptr->write_buffer_offset;
+    write_ptr= ptr->write_buffer + ptr->write_buffer_offset;
+
+    should_write= (should_write < length) ? should_write : length;
+
+    memcpy(write_ptr, buffer_ptr, should_write);
+    ptr->write_buffer_offset+= should_write;
+    buffer_ptr+= should_write;
+    length-= should_write;
 
     if (ptr->write_buffer_offset == MEMCACHED_MAX_BUFFER)
     {
-      size_t sent_length;
+      memcached_return rc;
+      ssize_t sent_length;
 
-      sent_length= memcached_io_flush(ptr, server_key);
+      sent_length= io_flush(ptr, &rc);
+      if (sent_length == -1)
+        return -1;
 
-      assert(sent_length == MEMCACHED_MAX_BUFFER);
-      ptr->write_buffer_offset= 0;
+      WATCHPOINT_ASSERT(sent_length == MEMCACHED_MAX_BUFFER);
     }
   }
 
   if (with_flush)
-    memcached_io_flush(ptr, server_key);
+  {
+    memcached_return rc;
+    if (io_flush(ptr, &rc) == -1)
+      return -1;
+  }
 
-  return length;
+  return original_length;
 }
 
-ssize_t memcached_io_flush(memcached_st *ptr, unsigned int server_key)
+memcached_return memcached_io_close(memcached_server_st *ptr)
+{
+  close(ptr->fd);
+
+  return MEMCACHED_SUCCESS;
+}
+
+static ssize_t io_flush(memcached_server_st *ptr,
+                                  memcached_return *error)
 {
   size_t sent_length;
+  size_t return_length;
+  char *local_write_ptr= ptr->write_buffer;
+  size_t write_length= ptr->write_buffer_offset;
+
+  *error= MEMCACHED_SUCCESS;
 
   if (ptr->write_buffer_offset == 0)
     return 0;
 
-  if (ptr->flags & MEM_NO_BLOCK)
+  /* Looking for memory overflows */
+  if (write_length == MEMCACHED_MAX_BUFFER)
+    WATCHPOINT_ASSERT(ptr->write_buffer == local_write_ptr);
+  WATCHPOINT_ASSERT((ptr->write_buffer + MEMCACHED_MAX_BUFFER) >= (local_write_ptr + write_length));
+  return_length= 0;
+  while (write_length)
   {
-    struct timeval local_tv;
-    fd_set set;
-
-    local_tv.tv_sec= 0;
-    local_tv.tv_usec= 300;
-
-    FD_ZERO(&set);
-    FD_SET(ptr->hosts[server_key].fd, &set);
+    sent_length= 0;
+    if (ptr->type == MEMCACHED_CONNECTION_UDP)
+    {
+      sent_length= sendto(ptr->fd, local_write_ptr, write_length, 0, 
+                          (struct sockaddr *)&ptr->address_info->ai_addr, 
+                          sizeof(struct sockaddr));
+    }
+    else
+    {
+      if ((ssize_t)(sent_length= write(ptr->fd, local_write_ptr, 
+                                       write_length)) == -1)
+      {
+        switch (errno)
+        {
+        case ENOBUFS:
+          continue;
+        case EAGAIN:
+          {
+            memcached_return rc;
+            rc= io_wait(ptr, MEM_WRITE);
+
+            if (rc == MEMCACHED_SUCCESS)
+              continue;
+
+            memcached_quit_server(ptr, 1);
+            return -1;
+          }
+        default:
+          memcached_quit_server(ptr, 1);
+          ptr->cached_errno= errno;
+          *error= MEMCACHED_ERRNO;
+          return -1;
+        }
+      }
+    }
 
-    select(1, NULL, &set, NULL, &local_tv);
-  }
-  if ((sent_length= send(ptr->hosts[server_key].fd, ptr->write_buffer, 
-                         ptr->write_buffer_offset, 0)) == -1)
-  {
-    return -1;
+    local_write_ptr+= sent_length;
+    write_length-= sent_length;
+    return_length+= sent_length;
   }
 
-  assert(sent_length == ptr->write_buffer_offset);
-
+  WATCHPOINT_ASSERT(write_length == 0);
+  WATCHPOINT_ASSERT(return_length == ptr->write_buffer_offset);
   ptr->write_buffer_offset= 0;
 
-  return sent_length;
+  return return_length;
 }
 
 /* 
   Eventually we will just kill off the server with the problem.
 */
-void memcached_io_reset(memcached_st *ptr, unsigned int server_key)
+void memcached_io_reset(memcached_server_st *ptr)
 {
   ptr->write_buffer_offset= 0;
-  memcached_quit(ptr);
+  memcached_quit_server(ptr, 0);
 }