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     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   WATCHPOINT_ASSERT(dummy_length == 0);
 172   WATCHPOINT_ASSERT(dummy_value == 0);
 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         continue;
 473     }
 474
 475     protocol_binary_request_getk request= { }; //= {.bytes= {0}};
 476     request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 477     if (mget_mode)
 478       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETKQ;
 479     else
 480       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
 481
 482     memcached_return_t vk;
 483     vk= memcached_validate_key_length(key_length[x],
 484                                       ptr->flags.binary_protocol);
 485     unlikely (vk != MEMCACHED_SUCCESS)
 486     {
 487       if (x > 0)
 488       {
 489         memcached_io_reset(instance);
 490       }
 491
 492       return vk;
 493     }
 494
 495     request.message.header.request.keylen= htons((uint16_t)(key_length[x] + memcached_array_size(ptr->_namespace)));
 496     request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 497     request.message.header.request.bodylen= htonl((uint32_t)( key_length[x] + memcached_array_size(ptr->_namespace)));
 498
 499     struct libmemcached_io_vector_st vector[]=
 500     {
 501       { sizeof(request.bytes), request.bytes },
 502       { memcached_array_size(ptr->_namespace), memcached_array_string(ptr->_namespace) },
 503       { key_length[x], keys[x] }
 504     };
 505
 506     if (memcached_io_writev(instance, vector, 3, flush) == -1)
 507     {
 508       memcached_server_response_reset(instance);
 509       rc= MEMCACHED_SOME_ERRORS;
 510       continue;
 511     }
 512
 513     /* We just want one pending response per server */
 514     memcached_server_response_reset(instance);
 515     memcached_server_response_increment(instance);
 516     if ((x > 0 && x == ptr->io_key_prefetch) && memcached_flush_buffers(ptr) != MEMCACHED_SUCCESS)
 517     {
 518       rc= MEMCACHED_SOME_ERRORS;
 519     }
 520   }
 521
 522   if (mget_mode)
 523   {
 524     /*
 525       Send a noop command to flush the buffers
 526     */
 527     protocol_binary_request_noop request= {}; //= {.bytes= {0}};
 528     request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 529     request.message.header.request.opcode= PROTOCOL_BINARY_CMD_NOOP;
 530     request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 531
 532     for (uint32_t x= 0; x < memcached_server_count(ptr); ++x)
 533     {
 534       memcached_server_write_instance_st instance=
 535         memcached_server_instance_fetch(ptr, x);
 536
 537       if (memcached_server_response_count(instance))
 538       {
 539         if (memcached_io_write(instance, NULL, 0, true) == -1)
 540         {
 541           memcached_server_response_reset(instance);
 542           memcached_io_reset(instance);
 543           rc= MEMCACHED_SOME_ERRORS;
 544         }
 545
 546         if (memcached_io_write(instance, request.bytes,
 547                                sizeof(request.bytes), true) == -1)
 548         {
 549           memcached_server_response_reset(instance);
 550           memcached_io_reset(instance);
 551           rc= MEMCACHED_SOME_ERRORS;
 552         }
 553       }
 554     }
 555   }
 556
 557
 558   return rc;
 559 }
 560
 561 static memcached_return_t replication_binary_mget(memcached_st *ptr,
 562                                                   uint32_t* hash,
 563                                                   bool* dead_servers,
 564                                                   const char *const *keys,
 565                                                   const size_t *key_length,
 566                                                   size_t number_of_keys)
 567 {
 568   memcached_return_t rc= MEMCACHED_NOTFOUND;
 569   uint32_t start= 0;
 570   uint64_t randomize_read= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_RANDOMIZE_REPLICA_READ);
 571
 572   if (randomize_read)
 573     start= (uint32_t)random() % (uint32_t)(ptr->number_of_replicas + 1);
 574
 575   /* Loop for each replica */
 576   for (uint32_t replica= 0; replica <= ptr->number_of_replicas; ++replica)
 577   {
 578     bool success= true;
 579
 580     for (uint32_t x= 0; x < number_of_keys; ++x)
 581     {
 582       if (hash[x] == memcached_server_count(ptr))
 583         continue; /* Already successfully sent */
 584
 585       uint32_t server= hash[x] + replica;
 586
 587       /* In case of randomized reads */
 588       if (randomize_read && ((server + start) <= (hash[x] + ptr->number_of_replicas)))
 589         server += start;
 590
 591       while (server >= memcached_server_count(ptr))
 592         server -= memcached_server_count(ptr);
 593
 594       if (dead_servers[server])
 595         continue;
 596
 597       memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, server);
 598
 599       if (memcached_server_response_count(instance) == 0)
 600       {
 601         rc= memcached_connect(instance);
 602         if (memcached_failed(rc))
 603         {
 604           memcached_io_reset(instance);
 605           dead_servers[server]= true;
 606           success= false;
 607           continue;
 608         }
 609       }
 610
 611       protocol_binary_request_getk request= {};
 612       request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 613       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
 614       request.message.header.request.keylen= htons((uint16_t)(key_length[x] + memcached_array_size(ptr->_namespace)));
 615       request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 616       request.message.header.request.bodylen= htonl((uint32_t)(key_length[x] + memcached_array_size(ptr->_namespace)));
 617
 618       /*
 619        * We need to disable buffering to actually know that the request was
 620        * successfully sent to the server (so that we should expect a result
 621        * back). It would be nice to do this in buffered mode, but then it
 622        * would be complex to handle all error situations if we got to send
 623        * some of the messages, and then we failed on writing out some others
 624        * and we used the callback interface from memcached_mget_execute so
 625        * that we might have processed some of the responses etc. For now,
 626        * just make sure we work _correctly_
 627      */
 628       struct libmemcached_io_vector_st vector[]=
 629       {
 630         { sizeof(request.bytes), request.bytes },
 631         { memcached_array_size(ptr->_namespace), memcached_array_string(ptr->_namespace) },
 632         { key_length[x], keys[x] }
 633       };
 634
 635       if (memcached_io_writev(instance, vector, 3, true) == -1)
 636       {
 637         memcached_io_reset(instance);
 638         dead_servers[server]= true;
 639         success= false;
 640         continue;
 641       }
 642
 643       memcached_server_response_increment(instance);
 644       hash[x]= memcached_server_count(ptr);
 645     }
 646
 647     if (success)
 648       break;
 649   }
 650
 651   return rc;
 652 }
 653
 654 static memcached_return_t binary_mget_by_key(memcached_st *ptr,
 655                                              uint32_t master_server_key,
 656                                              bool is_group_key_set,
 657                                              const char * const *keys,
 658                                              const size_t *key_length,
 659                                              size_t number_of_keys,
 660                                              bool mget_mode)
 661 {
 662   if (ptr->number_of_replicas == 0)
 663   {
 664     return simple_binary_mget(ptr, master_server_key, is_group_key_set,
 665                               keys, key_length, number_of_keys, mget_mode);
 666   }
 667
 668   uint32_t* hash= static_cast<uint32_t*>(libmemcached_malloc(ptr, sizeof(uint32_t) * number_of_keys));
 669   bool* dead_servers= static_cast<bool*>(libmemcached_calloc(ptr, memcached_server_count(ptr), sizeof(bool)));
 670
 671   if (hash == NULL || dead_servers == NULL)
 672   {
 673     libmemcached_free(ptr, hash);
 674     libmemcached_free(ptr, dead_servers);
 675     return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
 676   }
 677
 678   if (is_group_key_set)
 679   {
 680     for (size_t x= 0; x < number_of_keys; x++)
 681     {
 682       hash[x]= master_server_key;
 683     }
 684   }
 685   else
 686   {
 687     for (size_t x= 0; x < number_of_keys; x++)
 688     {
 689       hash[x]= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
 690     }
 691   }
 692
 693   memcached_return_t rc= replication_binary_mget(ptr, hash, dead_servers, keys,
 694                                                  key_length, number_of_keys);
 695
 696   WATCHPOINT_IFERROR(rc);
 697   libmemcached_free(ptr, hash);
 698   libmemcached_free(ptr, dead_servers);
 699
 700   return MEMCACHED_SUCCESS;
 701 }