Merge lp:~jc.redoutey/libmemcached/io_flush into lp:~tangent-org/libmemcached/trunk

[JC Redoutey]

[JC Redoutey (jc.redoutey)]

the test case does not work as good as I hoped, I guess because network calls are in fact local. But anyway, io_flush needs to somehow support recursive calls, so here is a version proposal.

[lp:~jc.redoutey/libmemcached/io_flush]

624. By Jean-Charles Redoutey <jc@Jayce2009lnx> on 2009-12-02

Removed wrongly introduced non portable call

[Brian Aker (brianaker)]

Anything still relevant in this?

Unmerged revisions

624. By Jean-Charles Redoutey <jc@Jayce2009lnx> on 2009-12-02

Removed wrongly introduced non portable call

623. By Jean-Charles Redoutey <jc@Jayce2009lnx> on 2009-11-30

Made io_flush reentrant, or at least enough for existing cases

=== modified file 'libmemcached/memcached_io.c'
--- libmemcached/memcached_io.c	2009-10-14 11:40:42 +0000
+++ libmemcached/memcached_io.c	2009-12-02 17:36:12 +0000
@@ -424,14 +424,13 @@
   ssize_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;
 
   WATCHPOINT_ASSERT(ptr->fd != -1);
 
   // UDP Sanity check, make sure that we are not sending somthing too big
-  if (ptr->type == MEMCACHED_CONNECTION_UDP && write_length > MAX_UDP_DATAGRAM_LENGTH)
+   if (ptr->type == MEMCACHED_CONNECTION_UDP && ptr->write_buffer_offset > MAX_UDP_DATAGRAM_LENGTH)
     return -1;
 
   if (ptr->write_buffer_offset == 0 || (ptr->type == MEMCACHED_CONNECTION_UDP
@@ -439,24 +438,32 @@
     return 0;
 
   /* Looking for memory overflows */
-#if defined(DEBUG)
-  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));
-#endif
-
-  return_length= 0;
-  while (write_length)
-  {
-    WATCHPOINT_ASSERT(ptr->fd != -1);
-    WATCHPOINT_ASSERT(write_length > 0);
-    sent_length= 0;
-    if (ptr->type == MEMCACHED_CONNECTION_UDP)
-      increment_udp_message_id(ptr);
-    sent_length= write(ptr->fd, local_write_ptr, write_length);
-
-    if (sent_length == -1)
-    {
+ #if defined(DEBUG)
+    if (ptr->write_buffer_offset == MEMCACHED_MAX_BUFFER)
+     WATCHPOINT_ASSERT(ptr->write_buffer == local_write_ptr);
+    WATCHPOINT_ASSERT((ptr->write_buffer + MEMCACHED_MAX_BUFFER) >= (local_write_ptr + ptr->write_buffer_offset));
+ #endif
+
+   return_length= 0;
+   while (ptr->write_buffer_offset)
+   {
+     WATCHPOINT_ASSERT(ptr->fd != -1);
+     WATCHPOINT_ASSERT(ptr->write_buffer_offset > 0);
+     if (ptr->type == MEMCACHED_CONNECTION_UDP)
+       increment_udp_message_id(ptr);
+
+      sent_length= write(ptr->fd, local_write_ptr + ptr->write_buffer_start_offset, ptr->write_buffer_offset);
+
+      if(sent_length > 0)
+      {
+          ptr->io_bytes_sent += (uint32_t) sent_length;
+          ptr->write_buffer_start_offset += sent_length;
+          ptr->write_buffer_offset -= (uint32_t) sent_length;
+          return_length+= (uint32_t) sent_length;
+      }
+
+     if (sent_length == -1)
+     {
       ptr->cached_errno= errno;
       switch (errno)
       {
@@ -490,23 +497,18 @@
       }
     }
 
-    if (ptr->type == MEMCACHED_CONNECTION_UDP &&
-        (size_t)sent_length != write_length)
-    {
-      memcached_quit_server(ptr, 1);
-      return -1;
-    }
-
-    ptr->io_bytes_sent += (uint32_t) sent_length;
-
-    local_write_ptr+= sent_length;
-    write_length-= (uint32_t) sent_length;
-    return_length+= (uint32_t) sent_length;
-  }
-
-  WATCHPOINT_ASSERT(write_length == 0);
-  // Need to study this assert() WATCHPOINT_ASSERT(return_length ==
-  // ptr->write_buffer_offset);
+     if (ptr->type == MEMCACHED_CONNECTION_UDP &&
+          (size_t)sent_length != ptr->write_buffer_offset)
+     {
+       memcached_quit_server(ptr, 1);
+       return -1;
+     }
+
+   }
+
+    WATCHPOINT_ASSERT(ptr->write_buffer_offset == 0);
+   // Need to study this assert() WATCHPOINT_ASSERT(return_length ==
+   // ptr->write_buffer_offset);
 
   // if we are a udp server, the begining of the buffer is reserverd for
   // the upd frame header
@@ -515,6 +517,7 @@
   else
     ptr->write_buffer_offset= 0;
 
+  ptr->write_buffer_start_offset = 0;
   return (ssize_t) return_length;
 }
 

=== modified file 'libmemcached/memcached_server.h'
--- libmemcached/memcached_server.h	2009-10-10 11:57:03 +0000
+++ libmemcached/memcached_server.h	2009-12-02 17:36:12 +0000
@@ -33,6 +33,7 @@
   size_t read_buffer_length;
   size_t read_data_length;
   size_t write_buffer_offset;
+  size_t write_buffer_start_offset;
   struct addrinfo *address_info;
   time_t next_retry;
   memcached_st *root;

=== modified file 'tests/function.c'
--- tests/function.c	2009-11-27 17:05:51 +0000
+++ tests/function.c	2009-12-02 17:36:12 +0000
@@ -3922,7 +3922,7 @@
 }
 
 #ifdef HAVE_LIBMEMCACHEDUTIL
-static void* connection_release(void *arg) 
+static void* connection_release(void *arg)
 {
   struct {
     memcached_pool_st* pool;
@@ -4169,7 +4169,7 @@
     rc= memcached_set(memc, keys[x], len[x], "1", 1, 0, 0);
     test_truth(rc == MEMCACHED_SUCCESS);
   }
-  
+
   memcached_quit(memc);
 
   for (int x=0; x< 7; ++x) {
@@ -5391,6 +5391,50 @@
   return TEST_SUCCESS;
 }
 
+
+
+/*
+ * Test that ensures that buffered set to not trigger problems during io_flush
+ */
+static test_return_t regression_bug_490520(memcached_st *memc)
+{
+
+  memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_NO_BLOCK,1);
+  memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS,1);
+  memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_POLL_TIMEOUT,1000);
+  memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_SERVER_FAILURE_LIMIT,1);
+  memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_RETRY_TIMEOUT,3600);
+
+  uint32_t number_of_hosts= memc->number_of_hosts;
+  memc->number_of_hosts= 1;
+  int max_keys= 200480;
+
+
+  char **keys= calloc((size_t)max_keys, sizeof(char*));
+  size_t *key_length=calloc((size_t)max_keys, sizeof(size_t));
+
+  /* First add all of the items.. */
+  char blob[3333] = {0};
+  memcached_return rc;
+  for (int x= 0; x < max_keys; ++x)
+  {
+    char k[251];
+    key_length[x]= (size_t)snprintf(k, sizeof(k), "0200%u", x);
+    keys[x]= strdup(k);
+    assert(keys[x] != NULL);
+    rc= memcached_set(memc, keys[x], key_length[x], blob, sizeof(blob), 0, 0);
+    assert(rc == MEMCACHED_SUCCESS || rc == MEMCACHED_BUFFERED);
+  }
+
+
+  memc->number_of_hosts= number_of_hosts;
+  return TEST_SUCCESS;
+}
+
+
+
+
+
 test_st udp_setup_server_tests[] ={
   {"set_udp_behavior_test", 0, set_udp_behavior_test},
   {"add_tcp_server_udp_client_test", 0, add_tcp_server_udp_client_test},
@@ -5567,6 +5611,7 @@
   {"lp:442914", 1, regression_bug_442914 },
   {"lp:447342", 1, regression_bug_447342 },
   {"lp:463297", 1, regression_bug_463297 },
+  {"lp:490520", 1, regression_bug_490520 },
   {0, 0, 0}
 };