First merge of Trond's patches (cherry picking).
[m6w6/libmemcached] / libmemcached / memcached_get.c
index 7ee714fe2e3ead8dc5f887b2f4d5bdc04c36903c..27a1410e51c4f7fbc6d3aacabce7cb412f002ba1 100644 (file)
@@ -1,53 +1,63 @@
+/* LibMemcached
+ * Copyright (C) 2006-2009 Brian Aker 
+ * All rights reserved.
+ *
+ * Use and distribution licensed under the BSD license.  See
+ * the COPYING file in the parent directory for full text.
+ *
+ * Summary: Get functions for libmemcached
+ *
+ */
+
 #include "common.h"
 #include "memcached_io.h"
 
-/* 
+/*
   What happens if no servers exist?
 */
-char *memcached_get(memcached_st *ptr, const char *key, 
-                    size_t key_length, 
-                    size_t *value_length, 
+char *memcached_get(memcached_st *ptr, const char *key,
+                    size_t key_length,
+                    size_t *value_length,
                     uint32_t *flags,
-                    memcached_return *error)
+                    memcached_return_t *error)
 {
-  return memcached_get_by_key(ptr, NULL, 0, key, key_length, value_length, 
+  return memcached_get_by_key(ptr, NULL, 0, key, key_length, value_length,
                               flags, error);
 }
 
-static memcached_return memcached_mget_by_key_real(memcached_st *ptr, 
-                                       const char *master_key, 
-                                       size_t master_key_length,
-                                       const char **keys, 
-                                       size_t *key_length, 
-                                       size_t number_of_keys,
-                                      bool mget_mode);
-
-char *memcached_get_by_key(memcached_st *ptr, 
-                           const char *master_key, 
-                           size_t master_key_length, 
+static memcached_return_t memcached_mget_by_key_real(memcached_st *ptr,
+                                                     const char *master_key,
+                                                     size_t master_key_length,
+                                                     const char * const *keys,
+                                                     const size_t *key_length,
+                                                     size_t number_of_keys,
+                                                     bool mget_mode);
+
+char *memcached_get_by_key(memcached_st *ptr,
+                           const char *master_key,
+                           size_t master_key_length,
                            const char *key, size_t key_length,
-                           size_t *value_length, 
+                           size_t *value_length,
                            uint32_t *flags,
-                           memcached_return *error)
+                           memcached_return_t *error)
 {
   char *value;
   size_t dummy_length;
   uint32_t dummy_flags;
-  memcached_return dummy_error;
+  memcached_return_t dummy_error;
 
-  unlikely (ptr->flags & MEM_USE_UDP)
+  unlikely (ptr->flags.use_udp)
   {
     *error= MEMCACHED_NOT_SUPPORTED;
     return NULL;
   }
 
   /* Request the key */
-  *error= memcached_mget_by_key_real(ptr, 
-                                master_key, 
-                                master_key_length, 
-                                (const char **)&key, &key_length, 1, false);
+  *error= memcached_mget_by_key_real(ptr, master_key, master_key_length,
+                                     (const char * const *)&key,
+                                     &key_length, 1, false);
 
-  value= memcached_fetch(ptr, NULL, NULL, 
+  value= memcached_fetch(ptr, NULL, NULL,
                          value_length, flags, error);
   /* This is for historical reasons */
   if (*error == MEMCACHED_END)
@@ -57,11 +67,11 @@ char *memcached_get_by_key(memcached_st *ptr,
   {
     if (ptr->get_key_failure && *error == MEMCACHED_NOTFOUND)
     {
-      memcached_return rc;
+      memcached_return_t rc;
 
       memcached_result_reset(&ptr->result);
       rc= ptr->get_key_failure(ptr, key, key_length, &ptr->result);
-      
+
       /* On all failure drop to returning NULL */
       if (rc == MEMCACHED_SUCCESS || rc == MEMCACHED_BUFFERED)
       {
@@ -72,7 +82,7 @@ char *memcached_get_by_key(memcached_st *ptr,
           if (latch == 0)
             memcached_behavior_set(ptr, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS, 1);
 
-          rc= memcached_set(ptr, key, key_length, 
+          rc= memcached_set(ptr, key, key_length,
                             memcached_result_value(&ptr->result),
                             memcached_result_length(&ptr->result),
                             0, memcached_result_flags(&ptr->result));
@@ -82,7 +92,7 @@ char *memcached_get_by_key(memcached_st *ptr,
         }
         else
         {
-          rc= memcached_set(ptr, key, key_length, 
+          rc= memcached_set(ptr, key, key_length,
                             memcached_result_value(&ptr->result),
                             memcached_result_length(&ptr->result),
                             0, memcached_result_flags(&ptr->result));
@@ -101,44 +111,46 @@ char *memcached_get_by_key(memcached_st *ptr,
     return NULL;
   }
 
-  (void)memcached_fetch(ptr, NULL, NULL, 
-                        &dummy_length, &dummy_flags, 
+  (void)memcached_fetch(ptr, NULL, NULL,
+                        &dummy_length, &dummy_flags,
                         &dummy_error);
   WATCHPOINT_ASSERT(dummy_length == 0);
 
   return value;
 }
 
-memcached_return memcached_mget(memcached_st *ptr, 
-                                const char **keys, size_t *key_length, 
-                                size_t number_of_keys)
+memcached_return_t memcached_mget(memcached_st *ptr,
+                                  const char * const *keys,
+                                  const size_t *key_length,
+                                  size_t number_of_keys)
 {
   return memcached_mget_by_key(ptr, NULL, 0, keys, key_length, number_of_keys);
 }
 
-static memcached_return binary_mget_by_key(memcached_st *ptr,
-                                           unsigned int master_server_key,
-                                           bool is_master_key_set,
-                                           const char **keys, size_t *key_length,
-                                           size_t number_of_keys,
-                                          bool mget_mode);
-
-static memcached_return memcached_mget_by_key_real(memcached_st *ptr, 
-                                       const char *master_key, 
-                                       size_t master_key_length,
-                                       const char **keys, 
-                                       size_t *key_length, 
-                                       size_t number_of_keys,
-                                      bool mget_mode)
+static memcached_return_t binary_mget_by_key(memcached_st *ptr,
+                                             unsigned int master_server_key,
+                                             bool is_master_key_set,
+                                             const char * const *keys,
+                                             const size_t *key_length,
+                                             size_t number_of_keys,
+                                             bool mget_mode);
+
+static memcached_return_t memcached_mget_by_key_real(memcached_st *ptr,
+                                                     const char *master_key,
+                                                     size_t master_key_length,
+                                                     const char * const *keys,
+                                                     const size_t *key_length,
+                                                     size_t number_of_keys,
+                                                     bool mget_mode)
 {
   unsigned int x;
-  memcached_return rc= MEMCACHED_NOTFOUND;
+  memcached_return_t rc= MEMCACHED_NOTFOUND;
   const char *get_command= "get ";
   uint8_t get_command_length= 4;
   unsigned int master_server_key= (unsigned int)-1; /* 0 is a valid server id! */
   bool is_master_key_set= false;
 
-   unlikely (ptr->flags & MEM_USE_UDP)
+  unlikely (ptr->flags.use_udp)
     return MEMCACHED_NOT_SUPPORTED;
 
   LIBMEMCACHED_MEMCACHED_MGET_START();
@@ -150,18 +162,18 @@ static memcached_return memcached_mget_by_key_real(memcached_st *ptr,
   if (ptr->number_of_hosts == 0)
     return MEMCACHED_NO_SERVERS;
 
-  if ((ptr->flags & MEM_VERIFY_KEY) && (memcached_key_test(keys, key_length, number_of_keys) == MEMCACHED_BAD_KEY_PROVIDED))
+  if (ptr->flags.verify_key && (memcached_key_test(keys, key_length, number_of_keys) == MEMCACHED_BAD_KEY_PROVIDED))
     return MEMCACHED_BAD_KEY_PROVIDED;
 
   if (master_key && master_key_length)
   {
-    if ((ptr->flags & MEM_VERIFY_KEY) && (memcached_key_test((const char **)&master_key, &master_key_length, 1) == MEMCACHED_BAD_KEY_PROVIDED))
+    if (ptr->flags.verify_key && (memcached_key_test((const char * const *)&master_key, &master_key_length, 1) == MEMCACHED_BAD_KEY_PROVIDED))
       return MEMCACHED_BAD_KEY_PROVIDED;
     master_server_key= memcached_generate_hash(ptr, master_key, master_key_length);
     is_master_key_set= true;
   }
 
-  /* 
+  /*
     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.
 
@@ -173,25 +185,25 @@ static memcached_return memcached_mget_by_key_real(memcached_st *ptr,
     {
       char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
 
-      if (ptr->flags & MEM_NO_BLOCK)
+      if (ptr->flags.no_block)
         (void)memcached_io_write(&ptr->hosts[x], NULL, 0, 1);
 
       while(memcached_server_response_count(&ptr->hosts[x]))
         (void)memcached_response(&ptr->hosts[x], buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, &ptr->result);
     }
   }
-  
-  if (ptr->flags & MEM_BINARY_PROTOCOL)
-    return binary_mget_by_key(ptr, master_server_key, is_master_key_set, keys, 
+
+  if (ptr->flags.binary_protocol)
+    return binary_mget_by_key(ptr, master_server_key, is_master_key_set, keys,
                               key_length, number_of_keys, mget_mode);
 
-  if (ptr->flags & MEM_SUPPORT_CAS)
+  if (ptr->flags.support_cas)
   {
     get_command= "gets ";
     get_command_length= 5;
   }
 
-  /* 
+  /*
     If a server fails we warn about errors and start all over with sending keys
     to the server.
   */
@@ -266,33 +278,75 @@ static memcached_return memcached_mget_by_key_real(memcached_st *ptr,
   return rc;
 }
 
-memcached_return memcached_mget_by_key(memcached_st *ptr, 
-                                       const char *master_key, 
-                                       size_t master_key_length,
-                                       const char **keys, 
-                                       size_t *key_length, 
-                                       size_t number_of_keys)
+memcached_return_t memcached_mget_by_key(memcached_st *ptr,
+                                         const char *master_key,
+                                         size_t master_key_length,
+                                         const char * const *keys,
+                                         const size_t *key_length,
+                                         size_t number_of_keys)
 {
-  return memcached_mget_by_key_real(ptr, master_key, master_key_length, keys, 
+  return memcached_mget_by_key_real(ptr, master_key, master_key_length, keys,
                                     key_length, number_of_keys, true);
 }
 
-static memcached_return simple_binary_mget(memcached_st *ptr,
-                                           unsigned int master_server_key,
-                                           bool is_master_key_set,
-                                           const char **keys, size_t *key_length, 
-                                           size_t number_of_keys, bool mget_mode)
+memcached_return_t memcached_mget_execute(memcached_st *ptr,
+                                          const char * const *keys,
+                                          const size_t *key_length,
+                                          size_t number_of_keys,
+                                          memcached_execute_fn *callback,
+                                          void *context,
+                                          unsigned int number_of_callbacks)
+{
+  return memcached_mget_execute_by_key(ptr, NULL, 0, keys, key_length,
+                                       number_of_keys, callback,
+                                       context, number_of_callbacks);
+}
+
+memcached_return_t memcached_mget_execute_by_key(memcached_st *ptr,
+                                                 const char *master_key,
+                                                 size_t master_key_length,
+                                                 const char * const *keys,
+                                                 const size_t *key_length,
+                                                 size_t number_of_keys,
+                                                 memcached_execute_fn *callback,
+                                                 void *context,
+                                                 unsigned int number_of_callbacks)
 {
-  memcached_return rc= MEMCACHED_NOTFOUND;
+  if ((ptr->flags.binary_protocol) == 0)
+    return MEMCACHED_NOT_SUPPORTED;
+
+  memcached_return_t rc;
+  memcached_callback_st *original_callbacks= ptr->callbacks;
+  memcached_callback_st cb= {
+    .callback= callback,
+    .context= context,
+    .number_of_callback= number_of_callbacks
+  };
+
+  ptr->callbacks= &cb;
+  rc= memcached_mget_by_key(ptr, master_key, master_key_length, keys,
+                            key_length, number_of_keys);
+  ptr->callbacks= original_callbacks;
+  return rc;
+}
+
+static memcached_return_t simple_binary_mget(memcached_st *ptr,
+                                             unsigned int master_server_key,
+                                             bool is_master_key_set,
+                                             const char * const *keys,
+                                             const size_t *key_length,
+                                             size_t number_of_keys, bool mget_mode)
+{
+  memcached_return_t rc= MEMCACHED_NOTFOUND;
   uint32_t x;
 
   int flush= number_of_keys == 1;
 
-  /* 
+  /*
     If a server fails we warn about errors and start all over with sending keys
     to the server.
   */
-  for (x= 0; x < number_of_keys; x++) 
+  for (x= 0; x < number_of_keys; x++)
   {
     unsigned int server_key;
 
@@ -301,13 +355,13 @@ static memcached_return simple_binary_mget(memcached_st *ptr,
     else
       server_key= memcached_generate_hash(ptr, keys[x], key_length[x]);
 
-    if (memcached_server_response_count(&ptr->hosts[server_key]) == 0) 
+    if (memcached_server_response_count(&ptr->hosts[server_key]) == 0)
     {
       rc= memcached_connect(&ptr->hosts[server_key]);
-      if (rc != MEMCACHED_SUCCESS) 
+      if (rc != MEMCACHED_SUCCESS)
         continue;
     }
-     
+
     protocol_binary_request_getk request= {.bytes= {0}};
     request.message.header.request.magic= PROTOCOL_BINARY_REQ;
     if (mget_mode)
@@ -315,9 +369,9 @@ static memcached_return simple_binary_mget(memcached_st *ptr,
     else
       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
 
-    memcached_return vk;
+    memcached_return_t vk;
     vk= memcached_validate_key_length(key_length[x],
-                                      ptr->flags & MEM_BINARY_PROTOCOL);
+                                      ptr->flags.binary_protocol);
     unlikely (vk != MEMCACHED_SUCCESS)
     {
       if (x > 0)
@@ -328,79 +382,89 @@ static memcached_return simple_binary_mget(memcached_st *ptr,
     request.message.header.request.keylen= htons((uint16_t)key_length[x]);
     request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
     request.message.header.request.bodylen= htonl((uint32_t) key_length[x]);
-    
+
     if ((memcached_io_write(&ptr->hosts[server_key], request.bytes,
                             sizeof(request.bytes), 0) == -1) ||
-        (memcached_io_write(&ptr->hosts[server_key], keys[x], 
-                            key_length[x], (char) flush) == -1)) 
+        (memcached_io_write(&ptr->hosts[server_key], keys[x],
+                            key_length[x], (char) flush) == -1))
     {
       memcached_server_response_reset(&ptr->hosts[server_key]);
       rc= MEMCACHED_SOME_ERRORS;
       continue;
     }
-    
+
     /* We just want one pending response per server */
-    memcached_server_response_reset(&ptr->hosts[server_key]); 
-    memcached_server_response_increment(&ptr->hosts[server_key]);    
+    memcached_server_response_reset(&ptr->hosts[server_key]);
+    memcached_server_response_increment(&ptr->hosts[server_key]);
     if ((x > 0 && x == ptr->io_key_prefetch) &&
         memcached_flush_buffers(ptr) != MEMCACHED_SUCCESS)
       rc= MEMCACHED_SOME_ERRORS;
   }
 
-  if (mget_mode) 
+  if (mget_mode)
   {
     /*
      * Send a noop command to flush the buffers
-     */
+   */
     protocol_binary_request_noop request= {.bytes= {0}};
     request.message.header.request.magic= PROTOCOL_BINARY_REQ;
     request.message.header.request.opcode= PROTOCOL_BINARY_CMD_NOOP;
     request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
-    
+
     for (x= 0; x < ptr->number_of_hosts; x++)
-      if (memcached_server_response_count(&ptr->hosts[x])) 
+      if (memcached_server_response_count(&ptr->hosts[x]))
       {
-        if (memcached_io_write(&ptr->hosts[x], NULL, 0, 1) == -1) 
+        if (memcached_io_write(&ptr->hosts[x], NULL, 0, 1) == -1)
         {
           memcached_server_response_reset(&ptr->hosts[x]);
           memcached_io_reset(&ptr->hosts[x]);
           rc= MEMCACHED_SOME_ERRORS;
         }
 
-        if (memcached_io_write(&ptr->hosts[x], request.bytes, 
-                              sizeof(request.bytes), 1) == -1) 
+        if (memcached_io_write(&ptr->hosts[x], request.bytes,
+                               sizeof(request.bytes), 1) == -1)
         {
           memcached_server_response_reset(&ptr->hosts[x]);
           memcached_io_reset(&ptr->hosts[x]);
           rc= MEMCACHED_SOME_ERRORS;
         }
       }
-    }
+  }
 
 
   return rc;
 }
 
-static memcached_return replication_binary_mget(memcached_st *ptr,
-                                             uint32_t* hash, bool* dead_servers,
-                                             const char **keys, size_t *key_length,
-                                             size_t number_of_keys, bool mget_mode)
+static memcached_return_t replication_binary_mget(memcached_st *ptr,
+                                                  uint32_t* hash,
+                                                  bool* dead_servers,
+                                                  const char *const *keys,
+                                                  const size_t *key_length,
+                                                  size_t number_of_keys)
 {
-  memcached_return rc= MEMCACHED_NOTFOUND;
-  uint32_t x;
+  memcached_return_t rc= MEMCACHED_NOTFOUND;
+  uint32_t x, start= 0;
+  uint64_t randomize_read= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_RANDOMIZE_REPLICA_READ);
 
-  int flush= number_of_keys == 1;
+  if (randomize_read)
+    start= (uint32_t)random() % (uint32_t)(ptr->number_of_replicas + 1);
 
+  /* Loop for each replica */
   for (uint32_t replica= 0; replica <= ptr->number_of_replicas; ++replica)
   {
-    bool success= true;    
-    
+    bool success= true;
+
     for (x= 0; x < number_of_keys; ++x)
     {
       if (hash[x] == ptr->number_of_hosts)
         continue; /* Already successfully sent */
 
       uint32_t server= hash[x] + replica;
+
+      /* In case of randomized reads */
+      if (randomize_read && ((server + start) <= (hash[x] + ptr->number_of_replicas)))
+        server += start;
+
       while (server >= ptr->number_of_hosts)
         server -= ptr->number_of_hosts;
 
@@ -419,58 +483,39 @@ static memcached_return replication_binary_mget(memcached_st *ptr,
         }
       }
 
-      protocol_binary_request_getk request= {.bytes= {0}};
-      request.message.header.request.magic= PROTOCOL_BINARY_REQ;
-      if (mget_mode)
-        request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETKQ;
-      else
-        request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
-
-      request.message.header.request.keylen= htons((uint16_t)key_length[x]);
-      request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
-      request.message.header.request.bodylen= htonl((uint32_t) key_length[x]);
+      protocol_binary_request_getk request= {
+        .message.header.request= {
+          .magic= PROTOCOL_BINARY_REQ,
+          .opcode= PROTOCOL_BINARY_CMD_GETK,
+          .keylen= htons((uint16_t)key_length[x]),
+          .datatype= PROTOCOL_BINARY_RAW_BYTES,
+          .bodylen= htonl((uint32_t)key_length[x])
+        }
+      };
 
+      /*
+       * We need to disable buffering to actually know that the request was
+       * successfully sent to the server (so that we should expect a result
+       * back). It would be nice to do this in buffered mode, but then it
+       * would be complex to handle all error situations if we got to send
+       * some of the messages, and then we failed on writing out some others
+       * and we used the callback interface from memcached_mget_execute so
+       * that we might have processed some of the responses etc. For now,
+       * just make sure we work _correctly_
+     */
       if ((memcached_io_write(&ptr->hosts[server], request.bytes,
                               sizeof(request.bytes), 0) == -1) ||
           (memcached_io_write(&ptr->hosts[server], keys[x],
-                              key_length[x], (char) flush) == -1))
+                              key_length[x], 1) == -1))
       {
         memcached_io_reset(&ptr->hosts[server]);
         dead_servers[server]= true;
         success= false;
         continue;
       }
-      /* we just want one pending response per server */
-      memcached_server_response_reset(&ptr->hosts[server]); 
-      memcached_server_response_increment(&ptr->hosts[server]);
-    }
 
-    if (mget_mode)
-    {
-      /*
-       * Send a noop command to flush the buffers
-       */
-      protocol_binary_request_noop request= {.bytes= {0}};
-      request.message.header.request.magic= PROTOCOL_BINARY_REQ;
-      request.message.header.request.opcode= PROTOCOL_BINARY_CMD_NOOP;
-      request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
-
-      for (x= 0; x < ptr->number_of_hosts; x++)
-        if (memcached_server_response_count(&ptr->hosts[x]))
-        {
-          if (memcached_io_write(&ptr->hosts[x], request.bytes,
-                                 sizeof(request.bytes), 1) == -1)
-          {
-            memcached_io_reset(&ptr->hosts[x]);
-            dead_servers[x]= true;
-            success= false;
-          }
-
-          /* mark all of the messages bound for this server as sent! */
-          for (x= 0; x < number_of_keys; ++x)
-            if (hash[x] == x)
-              hash[x]= ptr->number_of_hosts;
-        }
+      memcached_server_response_increment(&ptr->hosts[server]);
+      hash[x]= ptr->number_of_hosts;
     }
 
     if (success)
@@ -480,20 +525,22 @@ static memcached_return replication_binary_mget(memcached_st *ptr,
   return rc;
 }
 
-static memcached_return binary_mget_by_key(memcached_st *ptr,
-                                           unsigned int master_server_key,
-                                           bool is_master_key_set,
-                                           const char **keys, size_t *key_length,
-                                           size_t number_of_keys, bool mget_mode)
+static memcached_return_t binary_mget_by_key(memcached_st *ptr,
+                                             unsigned int master_server_key,
+                                             bool is_master_key_set,
+                                             const char * const *keys,
+                                             const size_t *key_length,
+                                             size_t number_of_keys,
+                                             bool mget_mode)
 {
-  memcached_return rc;
+  memcached_return_t rc;
 
-  if (ptr->number_of_replicas == 0) 
+  if (ptr->number_of_replicas == 0)
   {
     rc= simple_binary_mget(ptr, master_server_key, is_master_key_set,
                            keys, key_length, number_of_keys, mget_mode);
-  } 
-  else 
+  }
+  else
   {
     uint32_t* hash;
     bool* dead_servers;
@@ -515,8 +562,8 @@ static memcached_return binary_mget_by_key(memcached_st *ptr,
       for (unsigned int x= 0; x < number_of_keys; x++)
         hash[x]= memcached_generate_hash(ptr, keys[x], key_length[x]);
 
-    rc= replication_binary_mget(ptr, hash, dead_servers, keys, 
-                                key_length, number_of_keys, mget_mode);
+    rc= replication_binary_mget(ptr, hash, dead_servers, keys,
+                                key_length, number_of_keys);
 
     ptr->call_free(ptr, hash);
     ptr->call_free(ptr, dead_servers);