From: Brian Aker <brian@tangent.org>
Date: Mon, 17 Dec 2007 21:16:08 +0000 (-0800)
Subject: Refactor of response code. All of it has been streamlined to allow for multi
X-Git-Tag: 0.13~26
X-Git-Url: https://git.m6w6.name/?a=commitdiff_plain;h=b1d5d8be2237f04ccc99939399ebe7efad3f684e;p=m6w6%2Flibmemcached

Refactor of response code. All of it has been streamlined to allow for multi
delete/set operations.
---

diff --git a/include/memcached.h b/include/memcached.h
index 1e2d3b3f..8e90d6ed 100644
--- a/include/memcached.h
+++ b/include/memcached.h
@@ -119,7 +119,6 @@ struct memcached_server_st {
   char hostname[MEMCACHED_MAX_HOST_LENGTH];
   unsigned int port;
   int fd;
-  unsigned int stack_responses;
   unsigned int cursor_active;
   char write_buffer[MEMCACHED_MAX_BUFFER];
   size_t write_buffer_offset;
@@ -195,7 +194,7 @@ struct memcached_st {
   int send_size;
   int recv_size;
   int32_t poll_timeout;
-  memcached_string_st result_buffer;
+  memcached_result_st result;
   memcached_hash hash;
   memcached_server_distribution distribution;
   unsigned int wheel[MEMCACHED_WHEEL_SIZE];
@@ -281,7 +280,7 @@ memcached_result_st *memcached_fetch_result(memcached_st *ptr,
 #define memcached_server_name(A,B) (B).hostname
 #define memcached_server_port(A,B) (B).port
 #define memcached_server_list(A) (A)->hosts
-#define memcached_server_response_count(A,B) (A)->hosts[B].stack_responses
+#define memcached_server_response_count(A,B) (A)->hosts[B].cursor_active
 
 memcached_return memcached_server_add_udp(memcached_st *ptr, 
                                           char *hostname,
@@ -367,17 +366,17 @@ memcached_return memcached_delete_by_key(memcached_st *ptr,
 void memcached_result_free(memcached_result_st *result);
 memcached_result_st *memcached_result_create(memcached_st *ptr, 
                                              memcached_result_st *result);
-#define memcached_result_key_value(A) A->key
-#define memcached_result_key_length(A) A->key_length
+#define memcached_result_key_value(A) (A)->key
+#define memcached_result_key_length(A) (A)->key_length
 #ifdef FIX
-#define memcached_result_value(A) memcached_string_value(A->value)
-#define memcached_result_length(A) memcached_string_length(A->value)
+#define memcached_result_value(A) memcached_string_value((A)->value)
+#define memcached_result_length(A) memcached_string_length((A)->value)
 #else
 char *memcached_result_value(memcached_result_st *ptr);
 size_t memcached_result_length(memcached_result_st *ptr);
 #endif
-#define memcached_result_flags(A) A->flags
-#define memcached_result_cas(A) A->cas
+#define memcached_result_flags(A) (A)->flags
+#define memcached_result_cas(A) (A)->cas
 
 
 #ifndef __WATCHPOINT_H__
diff --git a/lib/common.h b/lib/common.h
index c49b3b96..76c734ed 100644
--- a/lib/common.h
+++ b/lib/common.h
@@ -60,18 +60,20 @@ uint32_t hsieh_hash(char *key, size_t key_length);
 memcached_return memcached_connect(memcached_st *ptr, unsigned int server_key);
 memcached_return memcached_response(memcached_st *ptr, 
                                     char *buffer, size_t buffer_length,
+                                    memcached_result_st *result,
                                     unsigned int server_key);
 unsigned int memcached_generate_hash(memcached_st *ptr, char *key, size_t key_length);
 void memcached_quit_server(memcached_st *ptr, unsigned int server_key, uint8_t io_death);
 
-#define memcached_server_response_increment(A,B) A->hosts[B].stack_responses++
-#define memcached_server_response_decrement(A,B) A->hosts[B].stack_responses--
+#define memcached_server_response_increment(A,B) A->hosts[B].cursor_active++
+#define memcached_server_response_decrement(A,B) A->hosts[B].cursor_active--
+#define memcached_server_response_reset(A,B) A->hosts[B].cursor_active=0
 
 /* String Struct */
-#define memcached_string_length(A) (size_t)(A->end - A->string)
-#define memcached_string_set_length(A, B) A->end= A->string + B
-#define memcached_string_size(A) A->current_size
-#define memcached_string_value(A) A->string
+#define memcached_string_length(A) (size_t)((A)->end - (A)->string)
+#define memcached_string_set_length(A, B) (A)->end= (A)->string + B
+#define memcached_string_size(A) (A)->current_size
+#define memcached_string_value(A) (A)->string
 
 memcached_string_st *memcached_string_create(memcached_st *ptr, 
                                              memcached_string_st *string, 
@@ -88,9 +90,10 @@ void memcached_string_free(memcached_string_st *string);
 memcached_return memcached_do(memcached_st *ptr, unsigned int server_key, char *commmand, 
                               size_t command_length, char with_flush);
 memcached_return memcached_version(memcached_st *ptr);
-memcached_return memcached_finish_server(memcached_st *ptr, unsigned int server_key);
-void memcached_finish(memcached_st *ptr);
-
+memcached_return value_fetch(memcached_st *ptr,
+                             char *buffer,
+                             memcached_result_st *result,
+                             unsigned int server_key);
 
 
 #endif /* __COMMON_H__ */
diff --git a/lib/memcached.c b/lib/memcached.c
index 30099c39..24c3b8bc 100644
--- a/lib/memcached.c
+++ b/lib/memcached.c
@@ -5,7 +5,7 @@
 
 memcached_st *memcached_create(memcached_st *ptr)
 {
-  memcached_string_st *string_ptr;
+  memcached_result_st *result_ptr;
   if (!ptr)
   {
     ptr= (memcached_st *)malloc(sizeof(memcached_st));
@@ -20,8 +20,8 @@ memcached_st *memcached_create(memcached_st *ptr)
   {
     memset(ptr, 0, sizeof(memcached_st));
   }
-  string_ptr= memcached_string_create(ptr, &ptr->result_buffer, 0);
-  WATCHPOINT_ASSERT(string_ptr);
+  result_ptr= memcached_result_create(ptr, &ptr->result);
+  WATCHPOINT_ASSERT(result_ptr);
   ptr->poll_timeout= MEMCACHED_DEFAULT_TIMEOUT;
   ptr->distribution= MEMCACHED_DISTRIBUTION_MODULA;
 
@@ -33,7 +33,7 @@ void memcached_free(memcached_st *ptr)
   /* If we have anything open, lets close it now */
   memcached_quit(ptr);
   memcached_server_list_free(ptr->hosts);
-  memcached_string_free(&ptr->result_buffer);
+  memcached_result_free(&ptr->result);
 
   if (ptr->is_allocated == MEMCACHED_ALLOCATED)
     free(ptr);
diff --git a/lib/memcached_auto.c b/lib/memcached_auto.c
index b9b74628..aa8cde32 100644
--- a/lib/memcached_auto.c
+++ b/lib/memcached_auto.c
@@ -30,7 +30,7 @@ static memcached_return memcached_auto(memcached_st *ptr,
   if (rc != MEMCACHED_SUCCESS)
     return rc;
 
-  rc= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, server_key);
+  rc= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, NULL, server_key);
 
   /* 
     So why recheck responce? Because the protocol is brain dead :)
diff --git a/lib/memcached_connect.c b/lib/memcached_connect.c
index 98d7630f..0bb28177 100644
--- a/lib/memcached_connect.c
+++ b/lib/memcached_connect.c
@@ -77,7 +77,7 @@ static memcached_return udp_connect(memcached_st *ptr, unsigned int server_key)
   if (ptr->hosts[server_key].fd == -1)
   {
     /* Old connection junk still is in the structure */
-    WATCHPOINT_ASSERT(ptr->hosts[server_key].stack_responses == 0);
+    WATCHPOINT_ASSERT(ptr->hosts[server_key].cursor_active == 0);
 
     /*
       If we have not allocated the hosts object.
@@ -111,7 +111,7 @@ static memcached_return tcp_connect(memcached_st *ptr, unsigned int server_key)
   if (ptr->hosts[server_key].fd == -1)
   {
     /* Old connection junk still is in the structure */
-    WATCHPOINT_ASSERT(ptr->hosts[server_key].stack_responses == 0);
+    WATCHPOINT_ASSERT(ptr->hosts[server_key].cursor_active == 0);
     struct addrinfo *use;
 
     if (ptr->hosts[server_key].sockaddr_inited == MEMCACHED_NOT_ALLOCATED || 
@@ -223,7 +223,7 @@ test_connect:
       ptr->connected++;
     }
 
-    WATCHPOINT_ASSERT(ptr->hosts[server_key].stack_responses == 0);
+    WATCHPOINT_ASSERT(ptr->hosts[server_key].cursor_active == 0);
   }
 
   return MEMCACHED_SUCCESS;
diff --git a/lib/memcached_delete.c b/lib/memcached_delete.c
index f68d7340..519e52be 100644
--- a/lib/memcached_delete.c
+++ b/lib/memcached_delete.c
@@ -57,7 +57,7 @@ memcached_return memcached_delete_by_key(memcached_st *ptr,
   }
   else
   {
-    rc= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, server_key);
+    rc= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, NULL, server_key);
     if (rc == MEMCACHED_DELETED)
       rc= MEMCACHED_SUCCESS;
   }
diff --git a/lib/memcached_do.c b/lib/memcached_do.c
index a929927c..cfc00ccd 100644
--- a/lib/memcached_do.c
+++ b/lib/memcached_do.c
@@ -9,9 +9,6 @@ memcached_return memcached_do(memcached_st *ptr, unsigned int server_key, char *
   WATCHPOINT_ASSERT(command_length);
   WATCHPOINT_ASSERT(command);
 
-  if (ptr->hosts[server_key].cursor_active)
-    (void)memcached_finish_server(ptr, server_key);
-
   if ((rc= memcached_connect(ptr, server_key)) != MEMCACHED_SUCCESS)
     return rc;
 
diff --git a/lib/memcached_fetch.c b/lib/memcached_fetch.c
index b6161845..67338835 100644
--- a/lib/memcached_fetch.c
+++ b/lib/memcached_fetch.c
@@ -1,135 +1,126 @@
 #include "common.h"
 #include "memcached_io.h"
 
-static memcached_return memcached_value_fetch(memcached_st *ptr, char *key, size_t *key_length, 
-                                              memcached_string_st *value,
-                                              uint32_t *flags,
-                                              uint64_t *cas,
-                                              unsigned int server_key)
+memcached_return value_fetch(memcached_st *ptr,
+                             char *buffer,
+                             memcached_result_st *result,
+                             unsigned int server_key)
 {
   memcached_return rc;
-  char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
   char *string_ptr;
   char *end_ptr;
+  char *next_ptr;
+  size_t value_length;
 
   end_ptr= buffer + MEMCACHED_DEFAULT_COMMAND_SIZE;
 
-  if (flags)
-    *flags= 0;
+  result->key_length= 0;
+  result->flags= 0;
+  memcached_string_reset(&result->value);
 
-  memcached_string_reset(value);
+  string_ptr= buffer;
+  string_ptr+= 6; /* "VALUE " */
 
-  rc= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, server_key);
 
-  if (rc == MEMCACHED_SUCCESS)
+  /* We load the key */
   {
-    char *next_ptr;
-    size_t value_length;
+    char *key;
 
-    string_ptr= buffer;
-    string_ptr+= 6; /* "VALUE " */
+    key= result->key;
+    result->key_length= 0;
 
-    /* We load the key */
-    if (key)
+    for (; isgraph(*string_ptr); string_ptr++)
     {
-      *key_length= 0;
-
-      for (; isgraph(*string_ptr); string_ptr++)
-      {
-        *key= *string_ptr;
-        key++;
-        (*key_length)++;
-      }
+      *key= *string_ptr;
+      key++;
+      result->key_length++;
     }
-    else /* Skip characters */
-      for (; isgraph(*string_ptr); string_ptr++);
+  }
 
-    if (end_ptr == string_ptr)
-        goto read_error;
+  if (end_ptr == string_ptr)
+    goto read_error;
 
-    /* Flags fetch move past space */
-    string_ptr++;
-    if (end_ptr == string_ptr)
-        goto read_error;
-    for (next_ptr= string_ptr; isdigit(*string_ptr); string_ptr++);
-    if (flags)
-      *flags= (uint32_t)strtol(next_ptr, &string_ptr, 10);
+  /* Flags fetch move past space */
+  string_ptr++;
+  if (end_ptr == string_ptr)
+    goto read_error;
+  for (next_ptr= string_ptr; isdigit(*string_ptr); string_ptr++);
+  result->flags= (uint32_t)strtol(next_ptr, &string_ptr, 10);
 
-    if (end_ptr == string_ptr)
-        goto read_error;
+  if (end_ptr == string_ptr)
+    goto read_error;
 
-    /* Length fetch move past space*/
-    string_ptr++;
-    if (end_ptr == string_ptr)
-        goto read_error;
+  /* Length fetch move past space*/
+  string_ptr++;
+  if (end_ptr == string_ptr)
+    goto read_error;
 
-    for (next_ptr= string_ptr; isdigit(*string_ptr); string_ptr++);
-    value_length= (size_t)strtoll(next_ptr, &string_ptr, 10);
+  for (next_ptr= string_ptr; isdigit(*string_ptr); string_ptr++);
+  value_length= (size_t)strtoll(next_ptr, &string_ptr, 10);
 
-    if (end_ptr == string_ptr)
-        goto read_error;
+  if (end_ptr == string_ptr)
+    goto read_error;
 
-    /* Skip spaces */
-    if (*string_ptr == '\r')
-    {
-      /* Skip past the \r\n */
-      string_ptr+= 2;
-    }
-    else
-    {
-      string_ptr++;
-      for (next_ptr= string_ptr; isdigit(*string_ptr); string_ptr++);
-      if (cas)
-        *cas= (size_t)strtoll(next_ptr, &string_ptr, 10);
-    }
-
-    if (end_ptr < string_ptr)
-        goto read_error;
+  /* Skip spaces */
+  if (*string_ptr == '\r')
+  {
+    /* Skip past the \r\n */
+    string_ptr+= 2;
+    result->cas= 0;
+  }
+  else
+  {
+    string_ptr++;
+    for (next_ptr= string_ptr; isdigit(*string_ptr); string_ptr++);
+    result->cas= (size_t)strtoll(next_ptr, &string_ptr, 10);
+  }
 
-    if (value_length)
-    {
-      size_t read_length;
-      size_t to_read;
-      char *value_ptr;
+  if (end_ptr < string_ptr)
+    goto read_error;
 
-      /* We add two bytes so that we can walk the \r\n */
-      rc= memcached_string_check(value, value_length+2);
-      if (rc != MEMCACHED_SUCCESS)
-      {
-        value_length= 0;
-        return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
-      }
+  if (value_length)
+  {
+    size_t read_length;
+    size_t to_read;
+    char *value_ptr;
 
-      value_ptr= memcached_string_value(value);
-      read_length= 0;
-      /* 
-        We read the \r\n into the string since not doing so is more 
-        cycles then the waster of memory to do so.
+    /* We add two bytes so that we can walk the \r\n */
+    rc= memcached_string_check(&result->value, value_length+2);
+    if (rc != MEMCACHED_SUCCESS)
+    {
+      value_length= 0;
+      return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
+    }
 
-        We are null terminating through, which will most likely make
-        some people lazy about using the return length.
-      */
-      to_read= (value_length) + 2;
+    value_ptr= memcached_string_value(&result->value);
+    read_length= 0;
+    /* 
+      We read the \r\n into the string since not doing so is more 
+      cycles then the waster of memory to do so.
 
-      read_length= memcached_io_read(ptr, server_key,
-                                     value_ptr, to_read);
+      We are null terminating through, which will most likely make
+      some people lazy about using the return length.
+    */
+    to_read= (value_length) + 2;
 
-      if (read_length != (size_t)(value_length + 2))
-      {
-        goto read_error;
-      }
+    read_length= memcached_io_read(ptr, server_key,
+                                   value_ptr, to_read);
 
-      /* This next bit blows the API, but this is internal....*/
-      {
-        char *char_ptr;
-        char_ptr= memcached_string_value(value);;
-        char_ptr[value_length]= 0;
-        char_ptr[value_length + 1]= 0;
-        memcached_string_set_length(value, value_length);
-      }
+    if (read_length != (size_t)(value_length + 2))
+    {
+      goto read_error;
+    }
 
-      return MEMCACHED_SUCCESS;
+    /* This next bit blows the API, but this is internal....*/
+    {
+      char *char_ptr;
+      char_ptr= memcached_string_value(&result->value);;
+      char_ptr[value_length]= 0;
+      char_ptr[value_length + 1]= 0;
+      memcached_string_set_length(&result->value, value_length);
     }
+
+    return MEMCACHED_SUCCESS;
   }
 
   return rc;
@@ -143,34 +134,44 @@ char *memcached_fetch(memcached_st *ptr, char *key, size_t *key_length,
                     uint32_t *flags,
                     memcached_return *error)
 {
-  memcached_string_st *result_buffer;
-  result_buffer= &ptr->result_buffer;
-
-  if (ptr->flags & MEM_NO_BLOCK)
-    memcached_io_preread(ptr);
+  memcached_result_st *result_buffer= &ptr->result;
 
   while (ptr->cursor_server < ptr->number_of_hosts)
   {
-    if (!ptr->hosts[ptr->cursor_server].cursor_active)
+    char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
+
+    if (memcached_server_response_count(ptr, ptr->cursor_server) == 0)
     {
       ptr->cursor_server++;
       continue;
     }
 
-    *error = memcached_value_fetch(ptr, key, key_length, result_buffer, 
-                                   flags, NULL, ptr->cursor_server);
-    *value_length= memcached_string_length(result_buffer);
-    
+  *error= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, result_buffer, ptr->cursor_server);
+
     if (*error == MEMCACHED_END) /* END means that we move on to the next */
     {
-      ptr->hosts[ptr->cursor_server].cursor_active= 0;
+      memcached_server_response_reset(ptr, ptr->cursor_server);
       ptr->cursor_server++;
       continue;
     }
     else if (*error == MEMCACHED_SUCCESS)
-      return  memcached_string_c_copy(result_buffer);
+    {
+      *value_length= memcached_string_length(&result_buffer->value);
+    
+      if (key)
+      {
+        strncpy(key, result_buffer->key, result_buffer->key_length);
+        *key_length= result_buffer->key_length;
+      }
+      *flags= result_buffer->flags;
+
+      return  memcached_string_c_copy(&result_buffer->value);
+    }
     else
+    {
+      *value_length= 0;
       return NULL;
+    }
   }
 
   ptr->cursor_server= 0;
@@ -192,22 +193,19 @@ memcached_result_st *memcached_fetch_result(memcached_st *ptr,
 
   while (ptr->cursor_server < ptr->number_of_hosts)
   {
-    if (!ptr->hosts[ptr->cursor_server].cursor_active)
+    char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
+
+    if (memcached_server_response_count(ptr, ptr->cursor_server) == 0)
     {
       ptr->cursor_server++;
       continue;
     }
 
-    result->cas= 0; /* We do this so we do not send in any junk */
-    *error= memcached_value_fetch(ptr, result->key, &result->key_length, 
-                                       &result->value, 
-                                       &result->flags,
-                                       &result->cas,
-                                       ptr->cursor_server);
+    *error= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, result, ptr->cursor_server);
     
     if (*error == MEMCACHED_END) /* END means that we move on to the next */
     {
-      ptr->hosts[ptr->cursor_server].cursor_active= 0;
+      memcached_server_response_reset(ptr, ptr->cursor_server);
       ptr->cursor_server++;
       continue;
     }
@@ -226,34 +224,3 @@ memcached_result_st *memcached_fetch_result(memcached_st *ptr,
   ptr->cursor_server= 0;
   return NULL;
 }
-
-memcached_return memcached_finish_server(memcached_st *ptr, unsigned int server_key)
-{
-  memcached_return rc;
-  memcached_string_st *result_buffer;
-
-  result_buffer= &ptr->result_buffer;
-
-  rc= MEMCACHED_SUCCESS;
-  while (rc == MEMCACHED_SUCCESS)
-  {
-    rc= memcached_value_fetch(ptr, NULL, NULL, result_buffer, 
-                               NULL, NULL, server_key);
-  }
-  ptr->hosts[server_key].cursor_active= 0;
-
-  return rc;
-}
-
-void memcached_finish(memcached_st *ptr)
-{
-  unsigned int x;
-
-  for (x= 0; x < ptr->number_of_hosts; x++)
-  {
-    if (ptr->hosts[x].cursor_active)
-      (void)memcached_finish_server(ptr, x);
-  }
-
-  ptr->cursor_server= 0;
-}
diff --git a/lib/memcached_flush.c b/lib/memcached_flush.c
index 5fa5d960..33cc6eb6 100644
--- a/lib/memcached_flush.c
+++ b/lib/memcached_flush.c
@@ -23,7 +23,7 @@ memcached_return memcached_flush(memcached_st *ptr, time_t expiration)
     rc= memcached_do(ptr, x, buffer, send_length, 1);
 
     if (rc == MEMCACHED_SUCCESS)
-      (void)memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, x);
+      (void)memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, NULL, x);
   }
 
   LIBMEMCACHED_MEMCACHED_FLUSH_END();
diff --git a/lib/memcached_get.c b/lib/memcached_get.c
index ffb2c5f6..b4266130 100644
--- a/lib/memcached_get.c
+++ b/lib/memcached_get.c
@@ -21,6 +21,9 @@ char *memcached_get_by_key(memcached_st *ptr,
                            memcached_return *error)
 {
   char *value;
+  size_t dummy_length;
+  uint32_t dummy_flags;
+  memcached_return dummy_error;
 
   /* Request the key */
   *error= memcached_mget_by_key(ptr, 
@@ -30,7 +33,6 @@ char *memcached_get_by_key(memcached_st *ptr,
 
   value= memcached_fetch(ptr, NULL, NULL, 
                          value_length, flags, error);
-
   /* This is for historical reasons */
   if (*error == MEMCACHED_END)
     *error= MEMCACHED_NOTFOUND;
@@ -38,7 +40,9 @@ char *memcached_get_by_key(memcached_st *ptr,
   if (value == NULL)
     return NULL;
 
-  memcached_finish(ptr);
+  (void)memcached_fetch(ptr, NULL, NULL, 
+                        &dummy_length, &dummy_flags, 
+                        &dummy_error);
 
   return value;
 }
@@ -76,11 +80,29 @@ memcached_return memcached_mget_by_key(memcached_st *ptr,
     get_command_length= 5;
   }
 
-  memcached_finish(ptr);
-
   if (master_key && master_key_length)
     master_server_key= memcached_generate_hash(ptr, master_key, master_key_length);
 
+  /* 
+    Here is where we pay for the non-block API. We need to remove any data sitting
+    in the queue before we start our get.
+
+    It might be optimum to bounce the connection if count > some number.
+  */
+  for (x= 0; x < ptr->number_of_hosts; x++)
+  {
+    if (memcached_server_response_count(ptr, x))
+    {
+      char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
+
+      if (ptr->flags & MEM_NO_BLOCK)
+        (void)memcached_io_write(ptr, x, NULL, 0, 1);
+
+      while(memcached_server_response_count(ptr, x))
+        (void)memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, &ptr->result, x);
+    }
+  }
+
   /* 
     If a server fails we warn about errors and start all over with sending keys
     to the server.
@@ -94,7 +116,7 @@ memcached_return memcached_mget_by_key(memcached_st *ptr,
     else
       server_key= memcached_generate_hash(ptr, keys[x], key_length[x]);
 
-    if (ptr->hosts[server_key].cursor_active == 0)
+    if (memcached_server_response_count(ptr, server_key) == 0)
     {
       rc= memcached_connect(ptr, server_key);
 
@@ -103,19 +125,21 @@ memcached_return memcached_mget_by_key(memcached_st *ptr,
         rc= MEMCACHED_SOME_ERRORS;
         continue;
       }
-      ptr->hosts[server_key].cursor_active++;
+      WATCHPOINT_ASSERT(ptr->hosts[server_key].cursor_active == 0);
+      memcached_server_response_increment(ptr, server_key);
+      WATCHPOINT_ASSERT(ptr->hosts[server_key].cursor_active == 1);
     }
 
     if ((memcached_io_write(ptr, server_key, keys[x], key_length[x], 0)) == -1)
     {
-      ptr->hosts[server_key].cursor_active= 0;
+      memcached_server_response_reset(ptr, server_key);
       rc= MEMCACHED_SOME_ERRORS;
       continue;
     }
 
     if ((memcached_io_write(ptr, server_key, " ", 1, 0)) == -1)
     {
-      ptr->hosts[server_key].cursor_active= 0;
+      memcached_server_response_reset(ptr, server_key);
       rc= MEMCACHED_SOME_ERRORS;
       continue;
     }
@@ -126,14 +150,13 @@ memcached_return memcached_mget_by_key(memcached_st *ptr,
   */
   for (x= 0; x < ptr->number_of_hosts; x++)
   {
-    if (ptr->hosts[x].cursor_active)
+    if (memcached_server_response_count(ptr, x))
     {
-      /* We need to doo something about non-connnected hosts in the future */
+      /* We need to do something about non-connnected hosts in the future */
       if ((memcached_io_write(ptr, x, "\r\n", 2, 1)) == -1)
       {
         rc= MEMCACHED_SOME_ERRORS;
       }
-      memcached_server_response_increment(ptr, x);
     }
   }
 
diff --git a/lib/memcached_io.c b/lib/memcached_io.c
index 84928d2b..38580638 100644
--- a/lib/memcached_io.c
+++ b/lib/memcached_io.c
@@ -57,7 +57,7 @@ void memcached_io_preread(memcached_st *ptr)
 
   for (x= 0; x < ptr->number_of_hosts; x++)
   {
-    if (ptr->hosts[x].cursor_active &&
+    if (memcached_server_response_count(ptr, x) &&
         ptr->hosts[x].read_data_length < MEMCACHED_MAX_BUFFER )
     {
       size_t data_read;
diff --git a/lib/memcached_quit.c b/lib/memcached_quit.c
index 44f435c4..fdc86b6c 100644
--- a/lib/memcached_quit.c
+++ b/lib/memcached_quit.c
@@ -29,12 +29,11 @@ void memcached_quit_server(memcached_st *ptr, unsigned int server_key, uint8_t i
     }
 
     ptr->hosts[server_key].fd= -1;
-    ptr->hosts[server_key].stack_responses= 0;
-    ptr->hosts[server_key].cursor_active= 0;
     ptr->hosts[server_key].write_buffer_offset= 0;
     ptr->hosts[server_key].read_buffer_length= 0;
     ptr->hosts[server_key].read_ptr= ptr->hosts[server_key].read_buffer;
     ptr->hosts[server_key].write_ptr= ptr->hosts[server_key].write_buffer;
+    memcached_server_response_reset(ptr, server_key);
   }
 
   ptr->connected--;
@@ -43,6 +42,7 @@ void memcached_quit_server(memcached_st *ptr, unsigned int server_key, uint8_t i
 void memcached_quit(memcached_st *ptr)
 {
   unsigned int x;
+
   if (ptr->hosts == NULL || 
       ptr->number_of_hosts == 0)
     return;
diff --git a/lib/memcached_response.c b/lib/memcached_response.c
index a695e9d8..6ec120c8 100644
--- a/lib/memcached_response.c
+++ b/lib/memcached_response.c
@@ -10,6 +10,7 @@
 
 memcached_return memcached_response(memcached_st *ptr, 
                                     char *buffer, size_t buffer_length,
+                                    memcached_result_st *result,
                                     unsigned int server_key)
 {
   unsigned int x;
@@ -20,6 +21,10 @@ memcached_return memcached_response(memcached_st *ptr,
 
   send_length= 0;
 
+  /* We may have old commands in the buffer not set, first purge */
+  if (ptr->flags & MEM_NO_BLOCK)
+    (void)memcached_io_write(ptr, server_key, NULL, 0, 1);
+
   max_messages= memcached_server_response_count(ptr, server_key);
   for (x= 0; x <  max_messages; x++)
   {
@@ -58,9 +63,16 @@ memcached_return memcached_response(memcached_st *ptr,
   case 'V': /* VALUE || VERSION */
     if (buffer[1] == 'A') /* VALUE */
     {
+      memcached_return rc;
+
       /* We add back in one because we will need to search for END */
       memcached_server_response_increment(ptr, server_key);
-      return MEMCACHED_SUCCESS;
+      if (result)
+        rc= value_fetch(ptr, buffer, result, server_key);
+      else
+        rc= value_fetch(ptr, buffer, &ptr->result, server_key);
+
+      return rc;
     }
     else if (buffer[1] == 'E') /* VERSION */
     {
diff --git a/lib/memcached_stats.c b/lib/memcached_stats.c
index decf22b6..a20426a6 100644
--- a/lib/memcached_stats.c
+++ b/lib/memcached_stats.c
@@ -221,7 +221,7 @@ static memcached_return memcached_stats_fetch(memcached_st *ptr,
 
   while (1)
   {
-    rc= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, server_key);
+    rc= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, NULL, server_key);
 
     if (rc == MEMCACHED_STAT)
     {
diff --git a/lib/memcached_storage.c b/lib/memcached_storage.c
index 5f694135..4c36edc6 100644
--- a/lib/memcached_storage.c
+++ b/lib/memcached_storage.c
@@ -110,7 +110,7 @@ static inline memcached_return memcached_send(memcached_st *ptr,
   if (to_write == 0)
     rc= MEMCACHED_SUCCESS;
   else
-    rc= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, server_key);
+    rc= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, NULL, server_key);
 
   if (rc == MEMCACHED_STORED)
     return MEMCACHED_SUCCESS;
diff --git a/lib/memcached_verbosity.c b/lib/memcached_verbosity.c
index 1ee9d070..44c28012 100644
--- a/lib/memcached_verbosity.c
+++ b/lib/memcached_verbosity.c
@@ -24,7 +24,7 @@ memcached_return memcached_verbosity(memcached_st *ptr, unsigned int verbosity)
       continue;
     }
 
-    rrc= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, x);
+    rrc= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, NULL, x);
     if (rrc != MEMCACHED_SUCCESS)
       rc= MEMCACHED_SOME_ERRORS;
   }
diff --git a/lib/memcached_version.c b/lib/memcached_version.c
index a463dfc4..2863b76b 100644
--- a/lib/memcached_version.c
+++ b/lib/memcached_version.c
@@ -23,7 +23,7 @@ memcached_return memcached_version(memcached_st *ptr)
       continue;
     }
 
-    rrc= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, x);
+    rrc= memcached_response(ptr, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, NULL, x);
     if (rrc != MEMCACHED_SUCCESS)
       rc= MEMCACHED_SOME_ERRORS;
 
diff --git a/tests/test.c b/tests/test.c
index f8cde94d..60db54cb 100644
--- a/tests/test.c
+++ b/tests/test.c
@@ -107,7 +107,6 @@ int main(int argc, char *argv[])
   for (x= 0; x < memcached_server_list_count(servers); x++)
   {
     printf("\t%s : %u\n", servers[x].hostname, servers[x].port);
-    assert(servers[x].stack_responses == 0);
     assert(servers[x].fd == -1);
     assert(servers[x].cursor_active == 0);
   }
@@ -151,7 +150,6 @@ int main(int argc, char *argv[])
 
       for (loop= 0; loop < memcached_server_list_count(servers); loop++)
       {
-        assert(memc->hosts[loop].stack_responses == 0);
         assert(memc->hosts[loop].fd == -1);
         assert(memc->hosts[loop].cursor_active == 0);
       }