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   {
  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     if (memcached_failed(memcached_key_test(*ptr, (const char * const *)&group_key, &group_key_length, 1)))
 241     {
 242       return memcached_set_error(*ptr, MEMCACHED_BAD_KEY_PROVIDED, MEMCACHED_AT, memcached_literal_param("A bad group key was provided."));
 243     }
 244
 245     master_server_key= memcached_generate_hash_with_redistribution(ptr, group_key, group_key_length);
 246     is_group_key_set= true;
 247   }
 248
 249   /*
 250     Here is where we pay for the non-block API. We need to remove any data sitting
 251     in the queue before we start our get.
 252
 253     It might be optimum to bounce the connection if count > some number.
 254   */
 255   for (uint32_t x= 0; x < memcached_server_count(ptr); x++)
 256   {
 257     memcached_server_write_instance_st instance=
 258       memcached_server_instance_fetch(ptr, x);
 259
 260     if (memcached_server_response_count(instance))
 261     {
 262       char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
 263
 264       if (ptr->flags.no_block)
 265       {
 266         memcached_io_write(instance);
 267       }
 268
 269       while(memcached_server_response_count(instance))
 270       {
 271         (void)memcached_response(instance, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, &ptr->result);
 272       }
 273     }
 274   }
 275
 276   if (ptr->flags.binary_protocol)
 277   {
 278     return binary_mget_by_key(ptr, master_server_key, is_group_key_set, keys,
 279                               key_length, number_of_keys, mget_mode);
 280   }
 281
 282   if (ptr->flags.support_cas)
 283   {
 284     get_command= "gets ";
 285     get_command_length= 5;
 286   }
 287
 288   /*
 289     If a server fails we warn about errors and start all over with sending keys
 290     to the server.
 291   */
 292   WATCHPOINT_ASSERT(rc == MEMCACHED_SUCCESS);
 293   size_t hosts_connected= 0;
 294   for (uint32_t x= 0; x < number_of_keys; x++)
 295   {
 296     memcached_server_write_instance_st instance;
 297     uint32_t server_key;
 298
 299     if (is_group_key_set)
 300     {
 301       server_key= master_server_key;
 302     }
 303     else
 304     {
 305       server_key= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
 306     }
 307
 308     instance= memcached_server_instance_fetch(ptr, server_key);
 309
 310     libmemcached_io_vector_st vector[]=
 311     {
 312       { get_command, get_command_length },
 313       { memcached_array_string(ptr->_namespace), memcached_array_size(ptr->_namespace) },
 314       { keys[x], key_length[x] },
 315       { memcached_literal_param(" ") }
 316     };
 317
 318
 319     if (memcached_server_response_count(instance) == 0)
 320     {
 321       rc= memcached_connect(instance);
 322
 323       if (memcached_failed(rc))
 324       {
 325         memcached_set_error(*instance, rc, MEMCACHED_AT);
 326         continue;
 327       }
 328       hosts_connected++;
 329
 330       if ((memcached_io_writev(instance, vector, 4, false)) == -1)
 331       {
 332         failures_occured_in_sending= true;
 333         continue;
 334       }
 335       WATCHPOINT_ASSERT(instance->cursor_active == 0);
 336       memcached_server_response_increment(instance);
 337       WATCHPOINT_ASSERT(instance->cursor_active == 1);
 338     }
 339     else
 340     {
 341       if ((memcached_io_writev(instance, (vector + 1), 3, false)) == -1)
 342       {
 343         memcached_server_response_reset(instance);
 344         failures_occured_in_sending= true;
 345         continue;
 346       }
 347     }
 348   }
 349
 350   if (hosts_connected == 0)
 351   {
 352     LIBMEMCACHED_MEMCACHED_MGET_END();
 353
 354     if (memcached_failed(rc))
 355     {
 356       return rc;
 357     }
 358
 359     return memcached_set_error(*ptr, MEMCACHED_NO_SERVERS, MEMCACHED_AT);
 360   }
 361
 362
 363   /*
 364     Should we muddle on if some servers are dead?
 365   */
 366   bool success_happened= false;
 367   for (uint32_t x= 0; x < memcached_server_count(ptr); x++)
 368   {
 369     memcached_server_write_instance_st instance=
 370       memcached_server_instance_fetch(ptr, x);
 371
 372     if (memcached_server_response_count(instance))
 373     {
 374       /* We need to do something about non-connnected hosts in the future */
 375       if ((memcached_io_write(instance, "\r\n", 2, true)) == -1)
 376       {
 377         failures_occured_in_sending= true;
 378       }
 379       else
 380       {
 381         success_happened= true;
 382       }
 383     }
 384   }
 385
 386   LIBMEMCACHED_MEMCACHED_MGET_END();
 387
 388   if (failures_occured_in_sending && success_happened)
 389   {
 390     return MEMCACHED_SOME_ERRORS;
 391   }
 392
 393   if (success_happened)
 394     return MEMCACHED_SUCCESS;
 395
 396   return MEMCACHED_FAILURE; // Complete failure occurred
 397 }
 398
 399 memcached_return_t memcached_mget_by_key(memcached_st *ptr,
 400                                          const char *group_key,
 401                                          size_t group_key_length,
 402                                          const char * const *keys,
 403                                          const size_t *key_length,
 404                                          size_t number_of_keys)
 405 {
 406   return memcached_mget_by_key_real(ptr, group_key, group_key_length, keys,
 407                                     key_length, number_of_keys, true);
 408 }
 409
 410 memcached_return_t memcached_mget_execute(memcached_st *ptr,
 411                                           const char * const *keys,
 412                                           const size_t *key_length,
 413                                           size_t number_of_keys,
 414                                           memcached_execute_fn *callback,
 415                                           void *context,
 416                                           unsigned int number_of_callbacks)
 417 {
 418   return memcached_mget_execute_by_key(ptr, NULL, 0, keys, key_length,
 419                                        number_of_keys, callback,
 420                                        context, number_of_callbacks);
 421 }
 422
 423 memcached_return_t memcached_mget_execute_by_key(memcached_st *ptr,
 424                                                  const char *group_key,
 425                                                  size_t group_key_length,
 426                                                  const char * const *keys,
 427                                                  const size_t *key_length,
 428                                                  size_t number_of_keys,
 429                                                  memcached_execute_fn *callback,
 430                                                  void *context,
 431                                                  unsigned int number_of_callbacks)
 432 {
 433   memcached_return_t rc;
 434   if (memcached_failed(rc= initialize_query(ptr, false)))
 435   {
 436     return rc;
 437   }
 438
 439   if (memcached_is_udp(ptr))
 440   {
 441     return memcached_set_error(*ptr, MEMCACHED_NOT_SUPPORTED, MEMCACHED_AT);
 442   }
 443
 444   if (memcached_is_binary(ptr) == false)
 445   {
 446     return memcached_set_error(*ptr, MEMCACHED_NOT_SUPPORTED, MEMCACHED_AT,
 447                                memcached_literal_param("ASCII protocol is not supported for memcached_mget_execute_by_key()"));
 448   }
 449
 450   memcached_callback_st *original_callbacks= ptr->callbacks;
 451   memcached_callback_st cb= {
 452     callback,
 453     context,
 454     number_of_callbacks
 455   };
 456
 457   ptr->callbacks= &cb;
 458   rc= memcached_mget_by_key(ptr, group_key, group_key_length, keys,
 459                             key_length, number_of_keys);
 460   ptr->callbacks= original_callbacks;
 461   return rc;
 462 }
 463
 464 static memcached_return_t simple_binary_mget(memcached_st *ptr,
 465                                              uint32_t master_server_key,
 466                                              bool is_group_key_set,
 467                                              const char * const *keys,
 468                                              const size_t *key_length,
 469                                              size_t number_of_keys, bool mget_mode)
 470 {
 471   memcached_return_t rc= MEMCACHED_NOTFOUND;
 472
 473   bool flush= (number_of_keys == 1);
 474
 475   /*
 476     If a server fails we warn about errors and start all over with sending keys
 477     to the server.
 478   */
 479   for (uint32_t x= 0; x < number_of_keys; ++x)
 480   {
 481     uint32_t server_key;
 482
 483     if (is_group_key_set)
 484     {
 485       server_key= master_server_key;
 486     }
 487     else
 488     {
 489       server_key= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
 490     }
 491
 492     memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, server_key);
 493
 494     if (memcached_server_response_count(instance) == 0)
 495     {
 496       rc= memcached_connect(instance);
 497       if (memcached_failed(rc))
 498       {
 499         continue;
 500       }
 501     }
 502
 503     protocol_binary_request_getk request= { }; //= {.bytes= {0}};
 504     request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 505     if (mget_mode)
 506     {
 507       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETKQ;
 508     }
 509     else
 510     {
 511       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
 512     }
 513
 514     memcached_return_t vk;
 515     vk= memcached_validate_key_length(key_length[x],
 516                                       ptr->flags.binary_protocol);
 517     if (vk != MEMCACHED_SUCCESS)
 518     {
 519       if (x > 0)
 520       {
 521         memcached_io_reset(instance);
 522       }
 523
 524       return vk;
 525     }
 526
 527     request.message.header.request.keylen= htons((uint16_t)(key_length[x] + memcached_array_size(ptr->_namespace)));
 528     request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 529     request.message.header.request.bodylen= htonl((uint32_t)( key_length[x] + memcached_array_size(ptr->_namespace)));
 530
 531     libmemcached_io_vector_st vector[]=
 532     {
 533       { request.bytes, sizeof(request.bytes) },
 534       { memcached_array_string(ptr->_namespace), memcached_array_size(ptr->_namespace) },
 535       { keys[x], key_length[x] }
 536     };
 537
 538     if (memcached_io_writev(instance, vector, 3, flush) == -1)
 539     {
 540       memcached_server_response_reset(instance);
 541       rc= MEMCACHED_SOME_ERRORS;
 542       continue;
 543     }
 544
 545     /* We just want one pending response per server */
 546     memcached_server_response_reset(instance);
 547     memcached_server_response_increment(instance);
 548     if ((x > 0 && x == ptr->io_key_prefetch) && memcached_flush_buffers(ptr) != MEMCACHED_SUCCESS)
 549     {
 550       rc= MEMCACHED_SOME_ERRORS;
 551     }
 552   }
 553
 554   if (mget_mode)
 555   {
 556     /*
 557       Send a noop command to flush the buffers
 558     */
 559     protocol_binary_request_noop request= {}; //= {.bytes= {0}};
 560     request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 561     request.message.header.request.opcode= PROTOCOL_BINARY_CMD_NOOP;
 562     request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 563
 564     for (uint32_t x= 0; x < memcached_server_count(ptr); ++x)
 565     {
 566       memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, x);
 567
 568       if (memcached_server_response_count(instance))
 569       {
 570         if (memcached_io_write(instance) == false)
 571         {
 572           memcached_server_response_reset(instance);
 573           memcached_io_reset(instance);
 574           rc= MEMCACHED_SOME_ERRORS;
 575         }
 576
 577         if (memcached_io_write(instance, request.bytes,
 578                                sizeof(request.bytes), true) == -1)
 579         {
 580           memcached_server_response_reset(instance);
 581           memcached_io_reset(instance);
 582           rc= MEMCACHED_SOME_ERRORS;
 583         }
 584       }
 585     }
 586   }
 587
 588
 589   return rc;
 590 }
 591
 592 static memcached_return_t replication_binary_mget(memcached_st *ptr,
 593                                                   uint32_t* hash,
 594                                                   bool* dead_servers,
 595                                                   const char *const *keys,
 596                                                   const size_t *key_length,
 597                                                   size_t number_of_keys)
 598 {
 599   memcached_return_t rc= MEMCACHED_NOTFOUND;
 600   uint32_t start= 0;
 601   uint64_t randomize_read= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_RANDOMIZE_REPLICA_READ);
 602
 603   if (randomize_read)
 604     start= (uint32_t)random() % (uint32_t)(ptr->number_of_replicas + 1);
 605
 606   /* Loop for each replica */
 607   for (uint32_t replica= 0; replica <= ptr->number_of_replicas; ++replica)
 608   {
 609     bool success= true;
 610
 611     for (uint32_t x= 0; x < number_of_keys; ++x)
 612     {
 613       if (hash[x] == memcached_server_count(ptr))
 614         continue; /* Already successfully sent */
 615
 616       uint32_t server= hash[x] + replica;
 617
 618       /* In case of randomized reads */
 619       if (randomize_read && ((server + start) <= (hash[x] + ptr->number_of_replicas)))
 620         server += start;
 621
 622       while (server >= memcached_server_count(ptr))
 623       {
 624         server -= memcached_server_count(ptr);
 625       }
 626
 627       if (dead_servers[server])
 628       {
 629         continue;
 630       }
 631
 632       memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, server);
 633
 634       if (memcached_server_response_count(instance) == 0)
 635       {
 636         rc= memcached_connect(instance);
 637
 638         if (memcached_failed(rc))
 639         {
 640           memcached_io_reset(instance);
 641           dead_servers[server]= true;
 642           success= false;
 643           continue;
 644         }
 645       }
 646
 647       protocol_binary_request_getk request= {};
 648       request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 649       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
 650       request.message.header.request.keylen= htons((uint16_t)(key_length[x] + memcached_array_size(ptr->_namespace)));
 651       request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 652       request.message.header.request.bodylen= htonl((uint32_t)(key_length[x] + memcached_array_size(ptr->_namespace)));
 653
 654       /*
 655        * We need to disable buffering to actually know that the request was
 656        * successfully sent to the server (so that we should expect a result
 657        * back). It would be nice to do this in buffered mode, but then it
 658        * would be complex to handle all error situations if we got to send
 659        * some of the messages, and then we failed on writing out some others
 660        * and we used the callback interface from memcached_mget_execute so
 661        * that we might have processed some of the responses etc. For now,
 662        * just make sure we work _correctly_
 663      */
 664       libmemcached_io_vector_st vector[]=
 665       {
 666         { request.bytes, sizeof(request.bytes) },
 667         { memcached_array_string(ptr->_namespace), memcached_array_size(ptr->_namespace) },
 668         { keys[x], key_length[x] }
 669       };
 670
 671       if (memcached_io_writev(instance, vector, 3, true) == -1)
 672       {
 673         memcached_io_reset(instance);
 674         dead_servers[server]= true;
 675         success= false;
 676         continue;
 677       }
 678
 679       memcached_server_response_increment(instance);
 680       hash[x]= memcached_server_count(ptr);
 681     }
 682
 683     if (success)
 684     {
 685       break;
 686     }
 687   }
 688
 689   return rc;
 690 }
 691
 692 static memcached_return_t binary_mget_by_key(memcached_st *ptr,
 693                                              uint32_t master_server_key,
 694                                              bool is_group_key_set,
 695                                              const char * const *keys,
 696                                              const size_t *key_length,
 697                                              size_t number_of_keys,
 698                                              bool mget_mode)
 699 {
 700   if (ptr->number_of_replicas == 0)
 701   {
 702     return simple_binary_mget(ptr, master_server_key, is_group_key_set,
 703                               keys, key_length, number_of_keys, mget_mode);
 704   }
 705
 706   uint32_t* hash= libmemcached_xvalloc(ptr, number_of_keys, uint32_t);
 707   bool* dead_servers= libmemcached_xcalloc(ptr, memcached_server_count(ptr), bool);
 708
 709   if (hash == NULL || dead_servers == NULL)
 710   {
 711     libmemcached_free(ptr, hash);
 712     libmemcached_free(ptr, dead_servers);
 713     return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
 714   }
 715
 716   if (is_group_key_set)
 717   {
 718     for (size_t x= 0; x < number_of_keys; x++)
 719     {
 720       hash[x]= master_server_key;
 721     }
 722   }
 723   else
 724   {
 725     for (size_t x= 0; x < number_of_keys; x++)
 726     {
 727       hash[x]= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
 728     }
 729   }
 730
 731   memcached_return_t rc= replication_binary_mget(ptr, hash, dead_servers, keys,
 732                                                  key_length, number_of_keys);
 733
 734   WATCHPOINT_IFERROR(rc);
 735   libmemcached_free(ptr, hash);
 736   libmemcached_free(ptr, dead_servers);
 737
 738   return MEMCACHED_SUCCESS;
 739 }