git.m6w6.name Git - m6w6/libmemcached/blob - libmemcached/get.cc

   1 /*  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
   2  *
   3  *  Libmemcached library
   4  *
   5  *  Copyright (C) 2011 Data Differential, http://datadifferential.com/
   6  *  Copyright (C) 2006-2009 Brian Aker All rights reserved.
   7  *
   8  *  Redistribution and use in source and binary forms, with or without
   9  *  modification, are permitted provided that the following conditions are
  10  *  met:
  11  *
  12  *      * Redistributions of source code must retain the above copyright
  13  *  notice, this list of conditions and the following disclaimer.
  14  *
  15  *      * Redistributions in binary form must reproduce the above
  16  *  copyright notice, this list of conditions and the following disclaimer
  17  *  in the documentation and/or other materials provided with the
  18  *  distribution.
  19  *
  20  *      * The names of its contributors may not be used to endorse or
  21  *  promote products derived from this software without specific prior
  22  *  written permission.
  23  *
  24  *  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  25  *  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  26  *  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  27  *  A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  28  *  OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  29  *  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  30  *  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  31  *  DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  32  *  THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  33  *  (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  34  *  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  35  *
  36  */
  37
  38 #include <libmemcached/common.h>
  39
  40 /*
  41   What happens if no servers exist?
  42 */
  43 char *memcached_get(memcached_st *ptr, const char *key,
  44                     size_t key_length,
  45                     size_t *value_length,
  46                     uint32_t *flags,
  47                     memcached_return_t *error)
  48 {
  49   return memcached_get_by_key(ptr, NULL, 0, key, key_length, value_length,
  50                               flags, error);
  51 }
  52
  53 static memcached_return_t memcached_mget_by_key_real(memcached_st *ptr,
  54                                                      const char *group_key,
  55                                                      size_t group_key_length,
  56                                                      const char * const *keys,
  57                                                      const size_t *key_length,
  58                                                      size_t number_of_keys,
  59                                                      bool mget_mode);
  60
  61 char *memcached_get_by_key(memcached_st *ptr,
  62                            const char *group_key,
  63                            size_t group_key_length,
  64                            const char *key, size_t key_length,
  65                            size_t *value_length,
  66                            uint32_t *flags,
  67                            memcached_return_t *error)
  68 {
  69   memcached_return_t unused;
  70   if (error == NULL)
  71   {
  72     error= &unused;
  73   }
  74
  75   uint64_t query_id= 0;
  76   if (ptr)
  77   {
  78     query_id= ptr->query_id;
  79   }
  80
  81   /* Request the key */
  82   *error= memcached_mget_by_key_real(ptr, group_key, group_key_length,
  83                                      (const char * const *)&key, &key_length,
  84                                      1, false);
  85   if (ptr)
  86   {
  87     assert_msg(ptr->query_id == query_id +1, "Programmer error, the query_id was not incremented.");
  88   }
  89
  90   if (memcached_failed(*error))
  91   {
  92     if (ptr)
  93     {
  94       if (memcached_has_current_error(*ptr)) // Find the most accurate error
  95       {
  96         *error= memcached_last_error(ptr);
  97       }
  98     }
  99
 100     if (value_length)
 101     {
 102       *value_length= 0;
 103     }
 104
 105     return NULL;
 106   }
 107
 108   char *value= memcached_fetch(ptr, NULL, NULL,
 109                                value_length, flags, error);
 110   assert_msg(ptr->query_id == query_id +1, "Programmer error, the query_id was not incremented.");
 111
 112   /* This is for historical reasons */
 113   if (*error == MEMCACHED_END)
 114   {
 115     *error= MEMCACHED_NOTFOUND;
 116   }
 117
 118   if (value == NULL)
 119   {
 120     if (ptr->get_key_failure && *error == MEMCACHED_NOTFOUND)
 121     {
 122       memcached_result_reset(&ptr->result);
 123       memcached_return_t rc= ptr->get_key_failure(ptr, key, key_length, &ptr->result);
 124
 125       /* On all failure drop to returning NULL */
 126       if (rc == MEMCACHED_SUCCESS or rc == MEMCACHED_BUFFERED)
 127       {
 128         if (rc == MEMCACHED_BUFFERED)
 129         {
 130           uint64_t latch; /* We use latch to track the state of the original socket */
 131           latch= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS);
 132           if (latch == 0)
 133           {
 134             memcached_behavior_set(ptr, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS, 1);
 135           }
 136
 137           rc= memcached_set(ptr, key, key_length,
 138                             (memcached_result_value(&ptr->result)),
 139                             (memcached_result_length(&ptr->result)),
 140                             0,
 141                             (memcached_result_flags(&ptr->result)));
 142
 143           if (rc == MEMCACHED_BUFFERED && latch == 0)
 144           {
 145             memcached_behavior_set(ptr, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS, 0);
 146           }
 147         }
 148         else
 149         {
 150           rc= memcached_set(ptr, key, key_length,
 151                             (memcached_result_value(&ptr->result)),
 152                             (memcached_result_length(&ptr->result)),
 153                             0,
 154                             (memcached_result_flags(&ptr->result)));
 155         }
 156
 157         if (rc == MEMCACHED_SUCCESS || rc == MEMCACHED_BUFFERED)
 158         {
 159           *error= rc;
 160           *value_length= memcached_result_length(&ptr->result);
 161           *flags= memcached_result_flags(&ptr->result);
 162           return memcached_string_take_value(&ptr->result.value);
 163         }
 164       }
 165     }
 166     assert_msg(ptr->query_id == query_id +1, "Programmer error, the query_id was not incremented.");
 167
 168     return NULL;
 169   }
 170
 171   size_t dummy_length;
 172   uint32_t dummy_flags;
 173   memcached_return_t dummy_error;
 174
 175   char *dummy_value= memcached_fetch(ptr, NULL, NULL,
 176                                      &dummy_length, &dummy_flags,
 177                                      &dummy_error);
 178   assert_msg(dummy_value == 0, "memcached_fetch() returned additional values beyond the single get it expected");
 179   assert_msg(dummy_length == 0, "memcached_fetch() returned additional values beyond the single get it expected");
 180   assert_msg(ptr->query_id == query_id +1, "Programmer error, the query_id was not incremented.");
 181
 182   return value;
 183 }
 184
 185 memcached_return_t memcached_mget(memcached_st *ptr,
 186                                   const char * const *keys,
 187                                   const size_t *key_length,
 188                                   size_t number_of_keys)
 189 {
 190   return memcached_mget_by_key(ptr, NULL, 0, keys, key_length, number_of_keys);
 191 }
 192
 193 static memcached_return_t binary_mget_by_key(memcached_st *ptr,
 194                                              uint32_t master_server_key,
 195                                              bool is_group_key_set,
 196                                              const char * const *keys,
 197                                              const size_t *key_length,
 198                                              size_t number_of_keys,
 199                                              bool mget_mode);
 200
 201 static memcached_return_t memcached_mget_by_key_real(memcached_st *ptr,
 202                                                      const char *group_key,
 203                                                      size_t group_key_length,
 204                                                      const char * const *keys,
 205                                                      const size_t *key_length,
 206                                                      size_t number_of_keys,
 207                                                      bool mget_mode)
 208 {
 209   bool failures_occured_in_sending= false;
 210   const char *get_command= "get ";
 211   uint8_t get_command_length= 4;
 212   unsigned int master_server_key= (unsigned int)-1; /* 0 is a valid server id! */
 213
 214   memcached_return_t rc;
 215   if (memcached_failed(rc= initialize_query(ptr, true)))
 216   {
 217     return rc;
 218   }
 219
 220   if (memcached_is_udp(ptr))
 221   {
 222     return memcached_set_error(*ptr, MEMCACHED_NOT_SUPPORTED, MEMCACHED_AT);
 223   }
 224
 225   LIBMEMCACHED_MEMCACHED_MGET_START();
 226
 227   if (number_of_keys == 0)
 228   {
 229     return memcached_set_error(*ptr, MEMCACHED_NOTFOUND, MEMCACHED_AT, memcached_literal_param("number_of_keys was zero"));
 230   }
 231
 232   if (memcached_failed(memcached_key_test(*ptr, keys, key_length, number_of_keys)))
 233   {
 234     return memcached_last_error(ptr);
 235   }
 236
 237   bool is_group_key_set= false;
 238   if (group_key and group_key_length)
 239   {
 240     master_server_key= memcached_generate_hash_with_redistribution(ptr, group_key, group_key_length);
 241     is_group_key_set= true;
 242   }
 243
 244   /*
 245     Here is where we pay for the non-block API. We need to remove any data sitting
 246     in the queue before we start our get.
 247
 248     It might be optimum to bounce the connection if count > some number.
 249   */
 250   for (uint32_t x= 0; x < memcached_server_count(ptr); x++)
 251   {
 252     memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, x);
 253
 254     if (memcached_server_response_count(instance))
 255     {
 256       char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
 257
 258       if (ptr->flags.no_block)
 259       {
 260         memcached_io_write(instance);
 261       }
 262
 263       while(memcached_server_response_count(instance))
 264       {
 265         (void)memcached_response(instance, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, &ptr->result);
 266       }
 267     }
 268   }
 269
 270   if (memcached_is_binary(ptr))
 271   {
 272     return binary_mget_by_key(ptr, master_server_key, is_group_key_set, keys,
 273                               key_length, number_of_keys, mget_mode);
 274   }
 275
 276   if (ptr->flags.support_cas)
 277   {
 278     get_command= "gets ";
 279     get_command_length= 5;
 280   }
 281
 282   /*
 283     If a server fails we warn about errors and start all over with sending keys
 284     to the server.
 285   */
 286   WATCHPOINT_ASSERT(rc == MEMCACHED_SUCCESS);
 287   size_t hosts_connected= 0;
 288   for (uint32_t x= 0; x < number_of_keys; x++)
 289   {
 290     memcached_server_write_instance_st instance;
 291     uint32_t server_key;
 292
 293     if (is_group_key_set)
 294     {
 295       server_key= master_server_key;
 296     }
 297     else
 298     {
 299       server_key= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
 300     }
 301
 302     instance= memcached_server_instance_fetch(ptr, server_key);
 303
 304     libmemcached_io_vector_st vector[]=
 305     {
 306       { get_command, get_command_length },
 307       { memcached_array_string(ptr->_namespace), memcached_array_size(ptr->_namespace) },
 308       { keys[x], key_length[x] },
 309       { memcached_literal_param(" ") }
 310     };
 311
 312
 313     if (memcached_server_response_count(instance) == 0)
 314     {
 315       rc= memcached_connect(instance);
 316
 317       if (memcached_failed(rc))
 318       {
 319         memcached_set_error(*instance, rc, MEMCACHED_AT);
 320         continue;
 321       }
 322       hosts_connected++;
 323
 324       if ((memcached_io_writev(instance, vector, 4, false)) == -1)
 325       {
 326         failures_occured_in_sending= true;
 327         continue;
 328       }
 329       WATCHPOINT_ASSERT(instance->cursor_active == 0);
 330       memcached_server_response_increment(instance);
 331       WATCHPOINT_ASSERT(instance->cursor_active == 1);
 332     }
 333     else
 334     {
 335       if ((memcached_io_writev(instance, (vector + 1), 3, false)) == -1)
 336       {
 337         memcached_server_response_reset(instance);
 338         failures_occured_in_sending= true;
 339         continue;
 340       }
 341     }
 342   }
 343
 344   if (hosts_connected == 0)
 345   {
 346     LIBMEMCACHED_MEMCACHED_MGET_END();
 347
 348     if (memcached_failed(rc))
 349     {
 350       return rc;
 351     }
 352
 353     return memcached_set_error(*ptr, MEMCACHED_NO_SERVERS, MEMCACHED_AT);
 354   }
 355
 356
 357   /*
 358     Should we muddle on if some servers are dead?
 359   */
 360   bool success_happened= false;
 361   for (uint32_t x= 0; x < memcached_server_count(ptr); x++)
 362   {
 363     memcached_server_write_instance_st instance=
 364       memcached_server_instance_fetch(ptr, x);
 365
 366     if (memcached_server_response_count(instance))
 367     {
 368       /* We need to do something about non-connnected hosts in the future */
 369       if ((memcached_io_write(instance, "\r\n", 2, true)) == -1)
 370       {
 371         failures_occured_in_sending= true;
 372       }
 373       else
 374       {
 375         success_happened= true;
 376       }
 377     }
 378   }
 379
 380   LIBMEMCACHED_MEMCACHED_MGET_END();
 381
 382   if (failures_occured_in_sending && success_happened)
 383   {
 384     return MEMCACHED_SOME_ERRORS;
 385   }
 386
 387   if (success_happened)
 388     return MEMCACHED_SUCCESS;
 389
 390   return MEMCACHED_FAILURE; // Complete failure occurred
 391 }
 392
 393 memcached_return_t memcached_mget_by_key(memcached_st *ptr,
 394                                          const char *group_key,
 395                                          size_t group_key_length,
 396                                          const char * const *keys,
 397                                          const size_t *key_length,
 398                                          size_t number_of_keys)
 399 {
 400   return memcached_mget_by_key_real(ptr, group_key, group_key_length, keys,
 401                                     key_length, number_of_keys, true);
 402 }
 403
 404 memcached_return_t memcached_mget_execute(memcached_st *ptr,
 405                                           const char * const *keys,
 406                                           const size_t *key_length,
 407                                           size_t number_of_keys,
 408                                           memcached_execute_fn *callback,
 409                                           void *context,
 410                                           unsigned int number_of_callbacks)
 411 {
 412   return memcached_mget_execute_by_key(ptr, NULL, 0, keys, key_length,
 413                                        number_of_keys, callback,
 414                                        context, number_of_callbacks);
 415 }
 416
 417 memcached_return_t memcached_mget_execute_by_key(memcached_st *ptr,
 418                                                  const char *group_key,
 419                                                  size_t group_key_length,
 420                                                  const char * const *keys,
 421                                                  const size_t *key_length,
 422                                                  size_t number_of_keys,
 423                                                  memcached_execute_fn *callback,
 424                                                  void *context,
 425                                                  unsigned int number_of_callbacks)
 426 {
 427   memcached_return_t rc;
 428   if (memcached_failed(rc= initialize_query(ptr, false)))
 429   {
 430     return rc;
 431   }
 432
 433   if (memcached_is_udp(ptr))
 434   {
 435     return memcached_set_error(*ptr, MEMCACHED_NOT_SUPPORTED, MEMCACHED_AT);
 436   }
 437
 438   if (memcached_is_binary(ptr) == false)
 439   {
 440     return memcached_set_error(*ptr, MEMCACHED_NOT_SUPPORTED, MEMCACHED_AT,
 441                                memcached_literal_param("ASCII protocol is not supported for memcached_mget_execute_by_key()"));
 442   }
 443
 444   memcached_callback_st *original_callbacks= ptr->callbacks;
 445   memcached_callback_st cb= {
 446     callback,
 447     context,
 448     number_of_callbacks
 449   };
 450
 451   ptr->callbacks= &cb;
 452   rc= memcached_mget_by_key(ptr, group_key, group_key_length, keys,
 453                             key_length, number_of_keys);
 454   ptr->callbacks= original_callbacks;
 455   return rc;
 456 }
 457
 458 static memcached_return_t simple_binary_mget(memcached_st *ptr,
 459                                              uint32_t master_server_key,
 460                                              bool is_group_key_set,
 461                                              const char * const *keys,
 462                                              const size_t *key_length,
 463                                              size_t number_of_keys, bool mget_mode)
 464 {
 465   memcached_return_t rc= MEMCACHED_NOTFOUND;
 466
 467   bool flush= (number_of_keys == 1);
 468
 469   /*
 470     If a server fails we warn about errors and start all over with sending keys
 471     to the server.
 472   */
 473   for (uint32_t x= 0; x < number_of_keys; ++x)
 474   {
 475     uint32_t server_key;
 476
 477     if (is_group_key_set)
 478     {
 479       server_key= master_server_key;
 480     }
 481     else
 482     {
 483       server_key= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
 484     }
 485
 486     memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, server_key);
 487
 488     if (memcached_server_response_count(instance) == 0)
 489     {
 490       rc= memcached_connect(instance);
 491       if (memcached_failed(rc))
 492       {
 493         continue;
 494       }
 495     }
 496
 497     protocol_binary_request_getk request= { }; //= {.bytes= {0}};
 498     request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 499     if (mget_mode)
 500     {
 501       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETKQ;
 502     }
 503     else
 504     {
 505       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
 506     }
 507
 508     memcached_return_t vk;
 509     vk= memcached_validate_key_length(key_length[x],
 510                                       ptr->flags.binary_protocol);
 511     if (vk != MEMCACHED_SUCCESS)
 512     {
 513       if (x > 0)
 514       {
 515         memcached_io_reset(instance);
 516       }
 517
 518       return vk;
 519     }
 520
 521     request.message.header.request.keylen= htons((uint16_t)(key_length[x] + memcached_array_size(ptr->_namespace)));
 522     request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 523     request.message.header.request.bodylen= htonl((uint32_t)( key_length[x] + memcached_array_size(ptr->_namespace)));
 524
 525     libmemcached_io_vector_st vector[]=
 526     {
 527       { request.bytes, sizeof(request.bytes) },
 528       { memcached_array_string(ptr->_namespace), memcached_array_size(ptr->_namespace) },
 529       { keys[x], key_length[x] }
 530     };
 531
 532     if (memcached_io_writev(instance, vector, 3, flush) == -1)
 533     {
 534       memcached_server_response_reset(instance);
 535       rc= MEMCACHED_SOME_ERRORS;
 536       continue;
 537     }
 538
 539     /* We just want one pending response per server */
 540     memcached_server_response_reset(instance);
 541     memcached_server_response_increment(instance);
 542     if ((x > 0 && x == ptr->io_key_prefetch) && memcached_flush_buffers(ptr) != MEMCACHED_SUCCESS)
 543     {
 544       rc= MEMCACHED_SOME_ERRORS;
 545     }
 546   }
 547
 548   if (mget_mode)
 549   {
 550     /*
 551       Send a noop command to flush the buffers
 552     */
 553     protocol_binary_request_noop request= {}; //= {.bytes= {0}};
 554     request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 555     request.message.header.request.opcode= PROTOCOL_BINARY_CMD_NOOP;
 556     request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 557
 558     for (uint32_t x= 0; x < memcached_server_count(ptr); ++x)
 559     {
 560       memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, x);
 561
 562       if (memcached_server_response_count(instance))
 563       {
 564         if (memcached_io_write(instance) == false)
 565         {
 566           memcached_server_response_reset(instance);
 567           memcached_io_reset(instance);
 568           rc= MEMCACHED_SOME_ERRORS;
 569         }
 570
 571         if (memcached_io_write(instance, request.bytes,
 572                                sizeof(request.bytes), true) == -1)
 573         {
 574           memcached_server_response_reset(instance);
 575           memcached_io_reset(instance);
 576           rc= MEMCACHED_SOME_ERRORS;
 577         }
 578       }
 579     }
 580   }
 581
 582
 583   return rc;
 584 }
 585
 586 static memcached_return_t replication_binary_mget(memcached_st *ptr,
 587                                                   uint32_t* hash,
 588                                                   bool* dead_servers,
 589                                                   const char *const *keys,
 590                                                   const size_t *key_length,
 591                                                   size_t number_of_keys)
 592 {
 593   memcached_return_t rc= MEMCACHED_NOTFOUND;
 594   uint32_t start= 0;
 595   uint64_t randomize_read= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_RANDOMIZE_REPLICA_READ);
 596
 597   if (randomize_read)
 598     start= (uint32_t)random() % (uint32_t)(ptr->number_of_replicas + 1);
 599
 600   /* Loop for each replica */
 601   for (uint32_t replica= 0; replica <= ptr->number_of_replicas; ++replica)
 602   {
 603     bool success= true;
 604
 605     for (uint32_t x= 0; x < number_of_keys; ++x)
 606     {
 607       if (hash[x] == memcached_server_count(ptr))
 608         continue; /* Already successfully sent */
 609
 610       uint32_t server= hash[x] + replica;
 611
 612       /* In case of randomized reads */
 613       if (randomize_read && ((server + start) <= (hash[x] + ptr->number_of_replicas)))
 614         server += start;
 615
 616       while (server >= memcached_server_count(ptr))
 617       {
 618         server -= memcached_server_count(ptr);
 619       }
 620
 621       if (dead_servers[server])
 622       {
 623         continue;
 624       }
 625
 626       memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, server);
 627
 628       if (memcached_server_response_count(instance) == 0)
 629       {
 630         rc= memcached_connect(instance);
 631
 632         if (memcached_failed(rc))
 633         {
 634           memcached_io_reset(instance);
 635           dead_servers[server]= true;
 636           success= false;
 637           continue;
 638         }
 639       }
 640
 641       protocol_binary_request_getk request= {};
 642       request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 643       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
 644       request.message.header.request.keylen= htons((uint16_t)(key_length[x] + memcached_array_size(ptr->_namespace)));
 645       request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 646       request.message.header.request.bodylen= htonl((uint32_t)(key_length[x] + memcached_array_size(ptr->_namespace)));
 647
 648       /*
 649        * We need to disable buffering to actually know that the request was
 650        * successfully sent to the server (so that we should expect a result
 651        * back). It would be nice to do this in buffered mode, but then it
 652        * would be complex to handle all error situations if we got to send
 653        * some of the messages, and then we failed on writing out some others
 654        * and we used the callback interface from memcached_mget_execute so
 655        * that we might have processed some of the responses etc. For now,
 656        * just make sure we work _correctly_
 657      */
 658       libmemcached_io_vector_st vector[]=
 659       {
 660         { request.bytes, sizeof(request.bytes) },
 661         { memcached_array_string(ptr->_namespace), memcached_array_size(ptr->_namespace) },
 662         { keys[x], key_length[x] }
 663       };
 664
 665       if (memcached_io_writev(instance, vector, 3, true) == -1)
 666       {
 667         memcached_io_reset(instance);
 668         dead_servers[server]= true;
 669         success= false;
 670         continue;
 671       }
 672
 673       memcached_server_response_increment(instance);
 674       hash[x]= memcached_server_count(ptr);
 675     }
 676
 677     if (success)
 678     {
 679       break;
 680     }
 681   }
 682
 683   return rc;
 684 }
 685
 686 static memcached_return_t binary_mget_by_key(memcached_st *ptr,
 687                                              uint32_t master_server_key,
 688                                              bool is_group_key_set,
 689                                              const char * const *keys,
 690                                              const size_t *key_length,
 691                                              size_t number_of_keys,
 692                                              bool mget_mode)
 693 {
 694   if (ptr->number_of_replicas == 0)
 695   {
 696     return simple_binary_mget(ptr, master_server_key, is_group_key_set,
 697                               keys, key_length, number_of_keys, mget_mode);
 698   }
 699
 700   uint32_t* hash= libmemcached_xvalloc(ptr, number_of_keys, uint32_t);
 701   bool* dead_servers= libmemcached_xcalloc(ptr, memcached_server_count(ptr), bool);
 702
 703   if (hash == NULL || dead_servers == NULL)
 704   {
 705     libmemcached_free(ptr, hash);
 706     libmemcached_free(ptr, dead_servers);
 707     return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
 708   }
 709
 710   if (is_group_key_set)
 711   {
 712     for (size_t x= 0; x < number_of_keys; x++)
 713     {
 714       hash[x]= master_server_key;
 715     }
 716   }
 717   else
 718   {
 719     for (size_t x= 0; x < number_of_keys; x++)
 720     {
 721       hash[x]= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
 722     }
 723   }
 724
 725   memcached_return_t rc= replication_binary_mget(ptr, hash, dead_servers, keys,
 726                                                  key_length, number_of_keys);
 727
 728   WATCHPOINT_IFERROR(rc);
 729   libmemcached_free(ptr, hash);
 730   libmemcached_free(ptr, dead_servers);
 731
 732   return MEMCACHED_SUCCESS;
 733 }