git.m6w6.name Git - awesomized/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     error= &unused;
  72
  73   unlikely (ptr->flags.use_udp)
  74   {
  75     if (value_length)
  76       *value_length= 0;
  77
  78     *error= memcached_set_error(*ptr, MEMCACHED_NOT_SUPPORTED, MEMCACHED_AT);
  79     return NULL;
  80   }
  81
  82   uint64_t query_id= ptr->query_id;
  83   (void)query_id;
  84
  85   /* Request the key */
  86   *error= memcached_mget_by_key_real(ptr, group_key, group_key_length,
  87                                      (const char * const *)&key, &key_length,
  88                                      1, false);
  89   assert_msg(ptr->query_id == query_id +1, "Programmer error, the query_id was not incremented.");
  90
  91
  92   if (memcached_failed(*error))
  93   {
  94     if (memcached_has_current_error(*ptr)) // Find the most accurate error
  95     {
  96       *error= memcached_last_error(ptr);
  97     }
  98
  99     if (value_length)
 100       *value_length= 0;
 101
 102     return NULL;
 103   }
 104
 105   char *value= memcached_fetch(ptr, NULL, NULL,
 106                                value_length, flags, error);
 107   assert_msg(ptr->query_id == query_id +1, "Programmer error, the query_id was not incremented.");
 108
 109   /* This is for historical reasons */
 110   if (*error == MEMCACHED_END)
 111     *error= MEMCACHED_NOTFOUND;
 112
 113   if (value == NULL)
 114   {
 115     if (ptr->get_key_failure && *error == MEMCACHED_NOTFOUND)
 116     {
 117       memcached_result_reset(&ptr->result);
 118       memcached_return_t rc= ptr->get_key_failure(ptr, key, key_length, &ptr->result);
 119
 120       /* On all failure drop to returning NULL */
 121       if (rc == MEMCACHED_SUCCESS || rc == MEMCACHED_BUFFERED)
 122       {
 123         if (rc == MEMCACHED_BUFFERED)
 124         {
 125           uint64_t latch; /* We use latch to track the state of the original socket */
 126           latch= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS);
 127           if (latch == 0)
 128             memcached_behavior_set(ptr, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS, 1);
 129
 130           rc= memcached_set(ptr, key, key_length,
 131                             (memcached_result_value(&ptr->result)),
 132                             (memcached_result_length(&ptr->result)),
 133                             0,
 134                             (memcached_result_flags(&ptr->result)));
 135
 136           if (rc == MEMCACHED_BUFFERED && latch == 0)
 137           {
 138             memcached_behavior_set(ptr, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS, 0);
 139           }
 140         }
 141         else
 142         {
 143           rc= memcached_set(ptr, key, key_length,
 144                             (memcached_result_value(&ptr->result)),
 145                             (memcached_result_length(&ptr->result)),
 146                             0,
 147                             (memcached_result_flags(&ptr->result)));
 148         }
 149
 150         if (rc == MEMCACHED_SUCCESS || rc == MEMCACHED_BUFFERED)
 151         {
 152           *error= rc;
 153           *value_length= memcached_result_length(&ptr->result);
 154           *flags= memcached_result_flags(&ptr->result);
 155           return memcached_string_take_value(&ptr->result.value);
 156         }
 157       }
 158     }
 159     assert_msg(ptr->query_id == query_id +1, "Programmer error, the query_id was not incremented.");
 160
 161     return NULL;
 162   }
 163
 164   size_t dummy_length;
 165   uint32_t dummy_flags;
 166   memcached_return_t dummy_error;
 167
 168   char *dummy_value= memcached_fetch(ptr, NULL, NULL,
 169                                      &dummy_length, &dummy_flags,
 170                                      &dummy_error);
 171   assert_msg(dummy_value == 0, "memcached_fetch() returned additional values beyond the single get it expected");
 172   assert_msg(dummy_length == 0, "memcached_fetch() returned additional values beyond the single get it expected");
 173   assert_msg(ptr->query_id == query_id +1, "Programmer error, the query_id was not incremented.");
 174
 175   return value;
 176 }
 177
 178 memcached_return_t memcached_mget(memcached_st *ptr,
 179                                   const char * const *keys,
 180                                   const size_t *key_length,
 181                                   size_t number_of_keys)
 182 {
 183   return memcached_mget_by_key(ptr, NULL, 0, keys, key_length, number_of_keys);
 184 }
 185
 186 static memcached_return_t binary_mget_by_key(memcached_st *ptr,
 187                                              uint32_t master_server_key,
 188                                              bool is_group_key_set,
 189                                              const char * const *keys,
 190                                              const size_t *key_length,
 191                                              size_t number_of_keys,
 192                                              bool mget_mode);
 193
 194 static memcached_return_t memcached_mget_by_key_real(memcached_st *ptr,
 195                                                      const char *group_key,
 196                                                      size_t group_key_length,
 197                                                      const char * const *keys,
 198                                                      const size_t *key_length,
 199                                                      size_t number_of_keys,
 200                                                      bool mget_mode)
 201 {
 202   bool failures_occured_in_sending= false;
 203   const char *get_command= "get ";
 204   uint8_t get_command_length= 4;
 205   unsigned int master_server_key= (unsigned int)-1; /* 0 is a valid server id! */
 206
 207   memcached_return_t rc;
 208   if (memcached_failed(rc= initialize_query(ptr)))
 209   {
 210     return rc;
 211   }
 212
 213   unlikely (ptr->flags.use_udp)
 214   {
 215     return memcached_set_error(*ptr, MEMCACHED_NOT_SUPPORTED, MEMCACHED_AT);
 216   }
 217
 218   LIBMEMCACHED_MEMCACHED_MGET_START();
 219
 220   if (number_of_keys == 0)
 221   {
 222     return memcached_set_error(*ptr, MEMCACHED_NOTFOUND, MEMCACHED_AT, memcached_literal_param("number_of_keys was zero"));
 223   }
 224
 225   if (memcached_failed(memcached_key_test(*ptr, keys, key_length, number_of_keys)))
 226   {
 227     return memcached_set_error(*ptr, MEMCACHED_BAD_KEY_PROVIDED, MEMCACHED_AT, memcached_literal_param("A bad key value was provided"));
 228   }
 229
 230   bool is_group_key_set= false;
 231   if (group_key && group_key_length)
 232   {
 233     if (memcached_failed(memcached_key_test(*ptr, (const char * const *)&group_key, &group_key_length, 1)))
 234     {
 235       return memcached_set_error(*ptr, MEMCACHED_BAD_KEY_PROVIDED, MEMCACHED_AT, memcached_literal_param("A bad group key was provided."));
 236     }
 237
 238     master_server_key= memcached_generate_hash_with_redistribution(ptr, group_key, group_key_length);
 239     is_group_key_set= true;
 240   }
 241
 242   /*
 243     Here is where we pay for the non-block API. We need to remove any data sitting
 244     in the queue before we start our get.
 245
 246     It might be optimum to bounce the connection if count > some number.
 247   */
 248   for (uint32_t x= 0; x < memcached_server_count(ptr); x++)
 249   {
 250     memcached_server_write_instance_st instance=
 251       memcached_server_instance_fetch(ptr, x);
 252
 253     if (memcached_server_response_count(instance))
 254     {
 255       char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
 256
 257       if (ptr->flags.no_block)
 258         (void)memcached_io_write(instance, NULL, 0, true);
 259
 260       while(memcached_server_response_count(instance))
 261         (void)memcached_response(instance, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, &ptr->result);
 262     }
 263   }
 264
 265   if (ptr->flags.binary_protocol)
 266   {
 267     return binary_mget_by_key(ptr, master_server_key, is_group_key_set, keys,
 268                               key_length, number_of_keys, mget_mode);
 269   }
 270
 271   if (ptr->flags.support_cas)
 272   {
 273     get_command= "gets ";
 274     get_command_length= 5;
 275   }
 276
 277   /*
 278     If a server fails we warn about errors and start all over with sending keys
 279     to the server.
 280   */
 281   WATCHPOINT_ASSERT(rc == MEMCACHED_SUCCESS);
 282   size_t hosts_connected= 0;
 283   for (uint32_t x= 0; x < number_of_keys; x++)
 284   {
 285     memcached_server_write_instance_st instance;
 286     uint32_t server_key;
 287
 288     if (is_group_key_set)
 289     {
 290       server_key= master_server_key;
 291     }
 292     else
 293     {
 294       server_key= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
 295     }
 296
 297     instance= memcached_server_instance_fetch(ptr, server_key);
 298
 299     struct libmemcached_io_vector_st vector[]=
 300     {
 301       { get_command_length, get_command },
 302       { memcached_array_size(ptr->_namespace), memcached_array_string(ptr->_namespace) },
 303       { key_length[x], keys[x] },
 304       { 1, " " }
 305     };
 306
 307
 308     if (memcached_server_response_count(instance) == 0)
 309     {
 310       rc= memcached_connect(instance);
 311
 312       if (memcached_failed(rc))
 313       {
 314         memcached_set_error(*instance, rc, MEMCACHED_AT);
 315         continue;
 316       }
 317       hosts_connected++;
 318
 319       if ((memcached_io_writev(instance, vector, 4, false)) == -1)
 320       {
 321         failures_occured_in_sending= true;
 322         continue;
 323       }
 324       WATCHPOINT_ASSERT(instance->cursor_active == 0);
 325       memcached_server_response_increment(instance);
 326       WATCHPOINT_ASSERT(instance->cursor_active == 1);
 327     }
 328     else
 329     {
 330       if ((memcached_io_writev(instance, (vector + 1), 3, false)) == -1)
 331       {
 332         memcached_server_response_reset(instance);
 333         failures_occured_in_sending= true;
 334         continue;
 335       }
 336     }
 337   }
 338
 339   if (hosts_connected == 0)
 340   {
 341     LIBMEMCACHED_MEMCACHED_MGET_END();
 342
 343     if (memcached_failed(rc))
 344       return rc;
 345
 346     return memcached_set_error(*ptr, MEMCACHED_NO_SERVERS, MEMCACHED_AT);
 347   }
 348
 349
 350   /*
 351     Should we muddle on if some servers are dead?
 352   */
 353   bool success_happened= false;
 354   for (uint32_t x= 0; x < memcached_server_count(ptr); x++)
 355   {
 356     memcached_server_write_instance_st instance=
 357       memcached_server_instance_fetch(ptr, x);
 358
 359     if (memcached_server_response_count(instance))
 360     {
 361       /* We need to do something about non-connnected hosts in the future */
 362       if ((memcached_io_write(instance, "\r\n", 2, true)) == -1)
 363       {
 364         failures_occured_in_sending= true;
 365       }
 366       else
 367       {
 368         success_happened= true;
 369       }
 370     }
 371   }
 372
 373   LIBMEMCACHED_MEMCACHED_MGET_END();
 374
 375   if (failures_occured_in_sending && success_happened)
 376   {
 377     return MEMCACHED_SOME_ERRORS;
 378   }
 379
 380   if (success_happened)
 381     return MEMCACHED_SUCCESS;
 382
 383   return MEMCACHED_FAILURE; // Complete failure occurred
 384 }
 385
 386 memcached_return_t memcached_mget_by_key(memcached_st *ptr,
 387                                          const char *group_key,
 388                                          size_t group_key_length,
 389                                          const char * const *keys,
 390                                          const size_t *key_length,
 391                                          size_t number_of_keys)
 392 {
 393   return memcached_mget_by_key_real(ptr, group_key, group_key_length, keys,
 394                                     key_length, number_of_keys, true);
 395 }
 396
 397 memcached_return_t memcached_mget_execute(memcached_st *ptr,
 398                                           const char * const *keys,
 399                                           const size_t *key_length,
 400                                           size_t number_of_keys,
 401                                           memcached_execute_fn *callback,
 402                                           void *context,
 403                                           unsigned int number_of_callbacks)
 404 {
 405   return memcached_mget_execute_by_key(ptr, NULL, 0, keys, key_length,
 406                                        number_of_keys, callback,
 407                                        context, number_of_callbacks);
 408 }
 409
 410 memcached_return_t memcached_mget_execute_by_key(memcached_st *ptr,
 411                                                  const char *group_key,
 412                                                  size_t group_key_length,
 413                                                  const char * const *keys,
 414                                                  const size_t *key_length,
 415                                                  size_t number_of_keys,
 416                                                  memcached_execute_fn *callback,
 417                                                  void *context,
 418                                                  unsigned int number_of_callbacks)
 419 {
 420   if ((ptr->flags.binary_protocol) == 0)
 421     return MEMCACHED_NOT_SUPPORTED;
 422
 423   memcached_return_t rc;
 424   memcached_callback_st *original_callbacks= ptr->callbacks;
 425   memcached_callback_st cb= {
 426     callback,
 427     context,
 428     number_of_callbacks
 429   };
 430
 431   ptr->callbacks= &cb;
 432   rc= memcached_mget_by_key(ptr, group_key, group_key_length, keys,
 433                             key_length, number_of_keys);
 434   ptr->callbacks= original_callbacks;
 435   return rc;
 436 }
 437
 438 static memcached_return_t simple_binary_mget(memcached_st *ptr,
 439                                              uint32_t master_server_key,
 440                                              bool is_group_key_set,
 441                                              const char * const *keys,
 442                                              const size_t *key_length,
 443                                              size_t number_of_keys, bool mget_mode)
 444 {
 445   memcached_return_t rc= MEMCACHED_NOTFOUND;
 446
 447   bool flush= (number_of_keys == 1);
 448
 449   /*
 450     If a server fails we warn about errors and start all over with sending keys
 451     to the server.
 452   */
 453   for (uint32_t x= 0; x < number_of_keys; ++x)
 454   {
 455     uint32_t server_key;
 456
 457     if (is_group_key_set)
 458     {
 459       server_key= master_server_key;
 460     }
 461     else
 462     {
 463       server_key= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
 464     }
 465
 466     memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, server_key);
 467
 468     if (memcached_server_response_count(instance) == 0)
 469     {
 470       rc= memcached_connect(instance);
 471       if (memcached_failed(rc))
 472       {
 473         continue;
 474       }
 475     }
 476
 477     protocol_binary_request_getk request= { }; //= {.bytes= {0}};
 478     request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 479     if (mget_mode)
 480       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETKQ;
 481     else
 482       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
 483
 484     memcached_return_t vk;
 485     vk= memcached_validate_key_length(key_length[x],
 486                                       ptr->flags.binary_protocol);
 487     unlikely (vk != MEMCACHED_SUCCESS)
 488     {
 489       if (x > 0)
 490       {
 491         memcached_io_reset(instance);
 492       }
 493
 494       return vk;
 495     }
 496
 497     request.message.header.request.keylen= htons((uint16_t)(key_length[x] + memcached_array_size(ptr->_namespace)));
 498     request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 499     request.message.header.request.bodylen= htonl((uint32_t)( key_length[x] + memcached_array_size(ptr->_namespace)));
 500
 501     struct libmemcached_io_vector_st vector[]=
 502     {
 503       { sizeof(request.bytes), request.bytes },
 504       { memcached_array_size(ptr->_namespace), memcached_array_string(ptr->_namespace) },
 505       { key_length[x], keys[x] }
 506     };
 507
 508     if (memcached_io_writev(instance, vector, 3, flush) == -1)
 509     {
 510       memcached_server_response_reset(instance);
 511       rc= MEMCACHED_SOME_ERRORS;
 512       continue;
 513     }
 514
 515     /* We just want one pending response per server */
 516     memcached_server_response_reset(instance);
 517     memcached_server_response_increment(instance);
 518     if ((x > 0 && x == ptr->io_key_prefetch) && memcached_flush_buffers(ptr) != MEMCACHED_SUCCESS)
 519     {
 520       rc= MEMCACHED_SOME_ERRORS;
 521     }
 522   }
 523
 524   if (mget_mode)
 525   {
 526     /*
 527       Send a noop command to flush the buffers
 528     */
 529     protocol_binary_request_noop request= {}; //= {.bytes= {0}};
 530     request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 531     request.message.header.request.opcode= PROTOCOL_BINARY_CMD_NOOP;
 532     request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 533
 534     for (uint32_t x= 0; x < memcached_server_count(ptr); ++x)
 535     {
 536       memcached_server_write_instance_st instance=
 537         memcached_server_instance_fetch(ptr, x);
 538
 539       if (memcached_server_response_count(instance))
 540       {
 541         if (memcached_io_write(instance, NULL, 0, true) == -1)
 542         {
 543           memcached_server_response_reset(instance);
 544           memcached_io_reset(instance);
 545           rc= MEMCACHED_SOME_ERRORS;
 546         }
 547
 548         if (memcached_io_write(instance, request.bytes,
 549                                sizeof(request.bytes), true) == -1)
 550         {
 551           memcached_server_response_reset(instance);
 552           memcached_io_reset(instance);
 553           rc= MEMCACHED_SOME_ERRORS;
 554         }
 555       }
 556     }
 557   }
 558
 559
 560   return rc;
 561 }
 562
 563 static memcached_return_t replication_binary_mget(memcached_st *ptr,
 564                                                   uint32_t* hash,
 565                                                   bool* dead_servers,
 566                                                   const char *const *keys,
 567                                                   const size_t *key_length,
 568                                                   size_t number_of_keys)
 569 {
 570   memcached_return_t rc= MEMCACHED_NOTFOUND;
 571   uint32_t start= 0;
 572   uint64_t randomize_read= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_RANDOMIZE_REPLICA_READ);
 573
 574   if (randomize_read)
 575     start= (uint32_t)random() % (uint32_t)(ptr->number_of_replicas + 1);
 576
 577   /* Loop for each replica */
 578   for (uint32_t replica= 0; replica <= ptr->number_of_replicas; ++replica)
 579   {
 580     bool success= true;
 581
 582     for (uint32_t x= 0; x < number_of_keys; ++x)
 583     {
 584       if (hash[x] == memcached_server_count(ptr))
 585         continue; /* Already successfully sent */
 586
 587       uint32_t server= hash[x] + replica;
 588
 589       /* In case of randomized reads */
 590       if (randomize_read && ((server + start) <= (hash[x] + ptr->number_of_replicas)))
 591         server += start;
 592
 593       while (server >= memcached_server_count(ptr))
 594       {
 595         server -= memcached_server_count(ptr);
 596       }
 597
 598       if (dead_servers[server])
 599       {
 600         continue;
 601       }
 602
 603       memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, server);
 604
 605       if (memcached_server_response_count(instance) == 0)
 606       {
 607         rc= memcached_connect(instance);
 608
 609         if (memcached_failed(rc))
 610         {
 611           memcached_io_reset(instance);
 612           dead_servers[server]= true;
 613           success= false;
 614           continue;
 615         }
 616       }
 617
 618       protocol_binary_request_getk request= {};
 619       request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 620       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
 621       request.message.header.request.keylen= htons((uint16_t)(key_length[x] + memcached_array_size(ptr->_namespace)));
 622       request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 623       request.message.header.request.bodylen= htonl((uint32_t)(key_length[x] + memcached_array_size(ptr->_namespace)));
 624
 625       /*
 626        * We need to disable buffering to actually know that the request was
 627        * successfully sent to the server (so that we should expect a result
 628        * back). It would be nice to do this in buffered mode, but then it
 629        * would be complex to handle all error situations if we got to send
 630        * some of the messages, and then we failed on writing out some others
 631        * and we used the callback interface from memcached_mget_execute so
 632        * that we might have processed some of the responses etc. For now,
 633        * just make sure we work _correctly_
 634      */
 635       struct libmemcached_io_vector_st vector[]=
 636       {
 637         { sizeof(request.bytes), request.bytes },
 638         { memcached_array_size(ptr->_namespace), memcached_array_string(ptr->_namespace) },
 639         { key_length[x], keys[x] }
 640       };
 641
 642       if (memcached_io_writev(instance, vector, 3, true) == -1)
 643       {
 644         memcached_io_reset(instance);
 645         dead_servers[server]= true;
 646         success= false;
 647         continue;
 648       }
 649
 650       memcached_server_response_increment(instance);
 651       hash[x]= memcached_server_count(ptr);
 652     }
 653
 654     if (success)
 655       break;
 656   }
 657
 658   return rc;
 659 }
 660
 661 static memcached_return_t binary_mget_by_key(memcached_st *ptr,
 662                                              uint32_t master_server_key,
 663                                              bool is_group_key_set,
 664                                              const char * const *keys,
 665                                              const size_t *key_length,
 666                                              size_t number_of_keys,
 667                                              bool mget_mode)
 668 {
 669   if (ptr->number_of_replicas == 0)
 670   {
 671     return simple_binary_mget(ptr, master_server_key, is_group_key_set,
 672                               keys, key_length, number_of_keys, mget_mode);
 673   }
 674
 675   uint32_t* hash= static_cast<uint32_t*>(libmemcached_malloc(ptr, sizeof(uint32_t) * number_of_keys));
 676   bool* dead_servers= static_cast<bool*>(libmemcached_calloc(ptr, memcached_server_count(ptr), sizeof(bool)));
 677
 678   if (hash == NULL || dead_servers == NULL)
 679   {
 680     libmemcached_free(ptr, hash);
 681     libmemcached_free(ptr, dead_servers);
 682     return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
 683   }
 684
 685   if (is_group_key_set)
 686   {
 687     for (size_t x= 0; x < number_of_keys; x++)
 688     {
 689       hash[x]= master_server_key;
 690     }
 691   }
 692   else
 693   {
 694     for (size_t x= 0; x < number_of_keys; x++)
 695     {
 696       hash[x]= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
 697     }
 698   }
 699
 700   memcached_return_t rc= replication_binary_mget(ptr, hash, dead_servers, keys,
 701                                                  key_length, number_of_keys);
 702
 703   WATCHPOINT_IFERROR(rc);
 704   libmemcached_free(ptr, hash);
 705   libmemcached_free(ptr, dead_servers);
 706
 707   return MEMCACHED_SUCCESS;
 708 }