git.m6w6.name Git - m6w6/libmemcached/blob - libmemcached/memcached_get.c

   1 /* LibMemcached
   2  * Copyright (C) 2006-2009 Brian Aker
   3  * All rights reserved.
   4  *
   5  * Use and distribution licensed under the BSD license.  See
   6  * the COPYING file in the parent directory for full text.
   7  *
   8  * Summary: Get functions for libmemcached
   9  *
  10  */
  11
  12 #include "common.h"
  13 #include "memcached_io.h"
  14
  15 /*
  16   What happens if no servers exist?
  17 */
  18 char *memcached_get(memcached_st *ptr, const char *key,
  19                     size_t key_length,
  20                     size_t *value_length,
  21                     uint32_t *flags,
  22                     memcached_return_t *error)
  23 {
  24   return memcached_get_by_key(ptr, NULL, 0, key, key_length, value_length,
  25                               flags, error);
  26 }
  27
  28 static memcached_return_t memcached_mget_by_key_real(memcached_st *ptr,
  29                                                      const char *master_key,
  30                                                      size_t master_key_length,
  31                                                      const char * const *keys,
  32                                                      const size_t *key_length,
  33                                                      size_t number_of_keys,
  34                                                      bool mget_mode);
  35
  36 char *memcached_get_by_key(memcached_st *ptr,
  37                            const char *master_key,
  38                            size_t master_key_length,
  39                            const char *key, size_t key_length,
  40                            size_t *value_length,
  41                            uint32_t *flags,
  42                            memcached_return_t *error)
  43 {
  44   char *value;
  45   size_t dummy_length;
  46   uint32_t dummy_flags;
  47   memcached_return_t dummy_error;
  48
  49   unlikely (ptr->flags.use_udp)
  50   {
  51     *error= MEMCACHED_NOT_SUPPORTED;
  52     return NULL;
  53   }
  54
  55   /* Request the key */
  56   *error= memcached_mget_by_key_real(ptr, master_key, master_key_length,
  57                                      (const char * const *)&key,
  58                                      &key_length, 1, false);
  59
  60   value= memcached_fetch(ptr, NULL, NULL,
  61                          value_length, flags, error);
  62   /* This is for historical reasons */
  63   if (*error == MEMCACHED_END)
  64     *error= MEMCACHED_NOTFOUND;
  65
  66   if (value == NULL)
  67   {
  68     if (ptr->get_key_failure && *error == MEMCACHED_NOTFOUND)
  69     {
  70       memcached_return_t rc;
  71
  72       memcached_result_reset(&ptr->result);
  73       rc= ptr->get_key_failure(ptr, key, key_length, &ptr->result);
  74
  75       /* On all failure drop to returning NULL */
  76       if (rc == MEMCACHED_SUCCESS || rc == MEMCACHED_BUFFERED)
  77       {
  78         if (rc == MEMCACHED_BUFFERED)
  79         {
  80           uint64_t latch; /* We use latch to track the state of the original socket */
  81           latch= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS);
  82           if (latch == 0)
  83             memcached_behavior_set(ptr, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS, 1);
  84
  85           rc= memcached_set(ptr, key, key_length,
  86                             memcached_result_value(&ptr->result),
  87                             memcached_result_length(&ptr->result),
  88                             0, memcached_result_flags(&ptr->result));
  89
  90           if (rc == MEMCACHED_BUFFERED && latch == 0)
  91             memcached_behavior_set(ptr, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS, 0);
  92         }
  93         else
  94         {
  95           rc= memcached_set(ptr, key, key_length,
  96                             memcached_result_value(&ptr->result),
  97                             memcached_result_length(&ptr->result),
  98                             0, memcached_result_flags(&ptr->result));
  99         }
 100
 101         if (rc == MEMCACHED_SUCCESS || rc == MEMCACHED_BUFFERED)
 102         {
 103           *error= rc;
 104           *value_length= memcached_result_length(&ptr->result);
 105           *flags= memcached_result_flags(&ptr->result);
 106           return memcached_string_c_copy(&ptr->result.value);
 107         }
 108       }
 109     }
 110
 111     return NULL;
 112   }
 113
 114   (void)memcached_fetch(ptr, NULL, NULL,
 115                         &dummy_length, &dummy_flags,
 116                         &dummy_error);
 117   WATCHPOINT_ASSERT(dummy_length == 0);
 118
 119   return value;
 120 }
 121
 122 memcached_return_t memcached_mget(memcached_st *ptr,
 123                                   const char * const *keys,
 124                                   const size_t *key_length,
 125                                   size_t number_of_keys)
 126 {
 127   return memcached_mget_by_key(ptr, NULL, 0, keys, key_length, number_of_keys);
 128 }
 129
 130 static memcached_return_t binary_mget_by_key(memcached_st *ptr,
 131                                              unsigned int master_server_key,
 132                                              bool is_master_key_set,
 133                                              const char * const *keys,
 134                                              const size_t *key_length,
 135                                              size_t number_of_keys,
 136                                              bool mget_mode);
 137
 138 static memcached_return_t memcached_mget_by_key_real(memcached_st *ptr,
 139                                                      const char *master_key,
 140                                                      size_t master_key_length,
 141                                                      const char * const *keys,
 142                                                      const size_t *key_length,
 143                                                      size_t number_of_keys,
 144                                                      bool mget_mode)
 145 {
 146   unsigned int x;
 147   memcached_return_t rc= MEMCACHED_NOTFOUND;
 148   const char *get_command= "get ";
 149   uint8_t get_command_length= 4;
 150   unsigned int master_server_key= (unsigned int)-1; /* 0 is a valid server id! */
 151   bool is_master_key_set= false;
 152
 153   unlikely (ptr->flags.use_udp)
 154     return MEMCACHED_NOT_SUPPORTED;
 155
 156   LIBMEMCACHED_MEMCACHED_MGET_START();
 157   ptr->cursor_server= 0;
 158
 159   if (number_of_keys == 0)
 160     return MEMCACHED_NOTFOUND;
 161
 162   if (ptr->number_of_hosts == 0)
 163     return MEMCACHED_NO_SERVERS;
 164
 165   if (ptr->flags.verify_key && (memcached_key_test(keys, key_length, number_of_keys) == MEMCACHED_BAD_KEY_PROVIDED))
 166     return MEMCACHED_BAD_KEY_PROVIDED;
 167
 168   if (master_key && master_key_length)
 169   {
 170     if (ptr->flags.verify_key && (memcached_key_test((const char * const *)&master_key, &master_key_length, 1) == MEMCACHED_BAD_KEY_PROVIDED))
 171       return MEMCACHED_BAD_KEY_PROVIDED;
 172     master_server_key= memcached_generate_hash(ptr, master_key, master_key_length);
 173     is_master_key_set= true;
 174   }
 175
 176   /*
 177     Here is where we pay for the non-block API. We need to remove any data sitting
 178     in the queue before we start our get.
 179
 180     It might be optimum to bounce the connection if count > some number.
 181   */
 182   for (x= 0; x < ptr->number_of_hosts; x++)
 183   {
 184     if (memcached_server_response_count(&ptr->hosts[x]))
 185     {
 186       char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
 187
 188       if (ptr->flags.no_block)
 189         (void)memcached_io_write(&ptr->hosts[x], NULL, 0, 1);
 190
 191       while(memcached_server_response_count(&ptr->hosts[x]))
 192         (void)memcached_response(&ptr->hosts[x], buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, &ptr->result);
 193     }
 194   }
 195
 196   if (ptr->flags.binary_protocol)
 197     return binary_mget_by_key(ptr, master_server_key, is_master_key_set, keys,
 198                               key_length, number_of_keys, mget_mode);
 199
 200   if (ptr->flags.support_cas)
 201   {
 202     get_command= "gets ";
 203     get_command_length= 5;
 204   }
 205
 206   /*
 207     If a server fails we warn about errors and start all over with sending keys
 208     to the server.
 209   */
 210   for (x= 0; x < number_of_keys; x++)
 211   {
 212     unsigned int server_key;
 213
 214     if (is_master_key_set)
 215       server_key= master_server_key;
 216     else
 217       server_key= memcached_generate_hash(ptr, keys[x], key_length[x]);
 218
 219     if (memcached_server_response_count(&ptr->hosts[server_key]) == 0)
 220     {
 221       rc= memcached_connect(&ptr->hosts[server_key]);
 222
 223       if (rc != MEMCACHED_SUCCESS)
 224         continue;
 225
 226       if ((memcached_io_write(&ptr->hosts[server_key], get_command, get_command_length, 0)) == -1)
 227       {
 228         rc= MEMCACHED_SOME_ERRORS;
 229         continue;
 230       }
 231       WATCHPOINT_ASSERT(ptr->hosts[server_key].cursor_active == 0);
 232       memcached_server_response_increment(&ptr->hosts[server_key]);
 233       WATCHPOINT_ASSERT(ptr->hosts[server_key].cursor_active == 1);
 234     }
 235
 236     /* Only called when we have a prefix key */
 237     if (ptr->prefix_key[0] != 0)
 238     {
 239       if ((memcached_io_write(&ptr->hosts[server_key], ptr->prefix_key, ptr->prefix_key_length, 0)) == -1)
 240       {
 241         memcached_server_response_reset(&ptr->hosts[server_key]);
 242         rc= MEMCACHED_SOME_ERRORS;
 243         continue;
 244       }
 245     }
 246
 247     if ((memcached_io_write(&ptr->hosts[server_key], keys[x], key_length[x], 0)) == -1)
 248     {
 249       memcached_server_response_reset(&ptr->hosts[server_key]);
 250       rc= MEMCACHED_SOME_ERRORS;
 251       continue;
 252     }
 253
 254     if ((memcached_io_write(&ptr->hosts[server_key], " ", 1, 0)) == -1)
 255     {
 256       memcached_server_response_reset(&ptr->hosts[server_key]);
 257       rc= MEMCACHED_SOME_ERRORS;
 258       continue;
 259     }
 260   }
 261
 262   /*
 263     Should we muddle on if some servers are dead?
 264   */
 265   for (x= 0; x < ptr->number_of_hosts; x++)
 266   {
 267     if (memcached_server_response_count(&ptr->hosts[x]))
 268     {
 269       /* We need to do something about non-connnected hosts in the future */
 270       if ((memcached_io_write(&ptr->hosts[x], "\r\n", 2, 1)) == -1)
 271       {
 272         rc= MEMCACHED_SOME_ERRORS;
 273       }
 274     }
 275   }
 276
 277   LIBMEMCACHED_MEMCACHED_MGET_END();
 278   return rc;
 279 }
 280
 281 memcached_return_t memcached_mget_by_key(memcached_st *ptr,
 282                                          const char *master_key,
 283                                          size_t master_key_length,
 284                                          const char * const *keys,
 285                                          const size_t *key_length,
 286                                          size_t number_of_keys)
 287 {
 288   return memcached_mget_by_key_real(ptr, master_key, master_key_length, keys,
 289                                     key_length, number_of_keys, true);
 290 }
 291
 292 memcached_return_t memcached_mget_execute(memcached_st *ptr,
 293                                           const char * const *keys,
 294                                           const size_t *key_length,
 295                                           size_t number_of_keys,
 296                                           memcached_execute_fn *callback,
 297                                           void *context,
 298                                           unsigned int number_of_callbacks)
 299 {
 300   return memcached_mget_execute_by_key(ptr, NULL, 0, keys, key_length,
 301                                        number_of_keys, callback,
 302                                        context, number_of_callbacks);
 303 }
 304
 305 memcached_return_t memcached_mget_execute_by_key(memcached_st *ptr,
 306                                                  const char *master_key,
 307                                                  size_t master_key_length,
 308                                                  const char * const *keys,
 309                                                  const size_t *key_length,
 310                                                  size_t number_of_keys,
 311                                                  memcached_execute_fn *callback,
 312                                                  void *context,
 313                                                  unsigned int number_of_callbacks)
 314 {
 315   if ((ptr->flags.binary_protocol) == 0)
 316     return MEMCACHED_NOT_SUPPORTED;
 317
 318   memcached_return_t rc;
 319   memcached_callback_st *original_callbacks= ptr->callbacks;
 320   memcached_callback_st cb= {
 321     .callback= callback,
 322     .context= context,
 323     .number_of_callback= number_of_callbacks
 324   };
 325
 326   ptr->callbacks= &cb;
 327   rc= memcached_mget_by_key(ptr, master_key, master_key_length, keys,
 328                             key_length, number_of_keys);
 329   ptr->callbacks= original_callbacks;
 330   return rc;
 331 }
 332
 333 static memcached_return_t simple_binary_mget(memcached_st *ptr,
 334                                              unsigned int master_server_key,
 335                                              bool is_master_key_set,
 336                                              const char * const *keys,
 337                                              const size_t *key_length,
 338                                              size_t number_of_keys, bool mget_mode)
 339 {
 340   memcached_return_t rc= MEMCACHED_NOTFOUND;
 341   uint32_t x;
 342
 343   int flush= number_of_keys == 1;
 344
 345   /*
 346     If a server fails we warn about errors and start all over with sending keys
 347     to the server.
 348   */
 349   for (x= 0; x < number_of_keys; x++)
 350   {
 351     unsigned int server_key;
 352
 353     if (is_master_key_set)
 354       server_key= master_server_key;
 355     else
 356       server_key= memcached_generate_hash(ptr, keys[x], key_length[x]);
 357
 358     if (memcached_server_response_count(&ptr->hosts[server_key]) == 0)
 359     {
 360       rc= memcached_connect(&ptr->hosts[server_key]);
 361       if (rc != MEMCACHED_SUCCESS)
 362         continue;
 363     }
 364
 365     protocol_binary_request_getk request= {.bytes= {0}};
 366     request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 367     if (mget_mode)
 368       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETKQ;
 369     else
 370       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
 371
 372     memcached_return_t vk;
 373     vk= memcached_validate_key_length(key_length[x],
 374                                       ptr->flags.binary_protocol);
 375     unlikely (vk != MEMCACHED_SUCCESS)
 376     {
 377       if (x > 0)
 378         memcached_io_reset(&ptr->hosts[server_key]);
 379       return vk;
 380     }
 381
 382     request.message.header.request.keylen= htons((uint16_t)key_length[x]);
 383     request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 384     request.message.header.request.bodylen= htonl((uint32_t) key_length[x]);
 385
 386     if ((memcached_io_write(&ptr->hosts[server_key], request.bytes,
 387                             sizeof(request.bytes), 0) == -1) ||
 388         (memcached_io_write(&ptr->hosts[server_key], keys[x],
 389                             key_length[x], (char) flush) == -1))
 390     {
 391       memcached_server_response_reset(&ptr->hosts[server_key]);
 392       rc= MEMCACHED_SOME_ERRORS;
 393       continue;
 394     }
 395
 396     /* We just want one pending response per server */
 397     memcached_server_response_reset(&ptr->hosts[server_key]);
 398     memcached_server_response_increment(&ptr->hosts[server_key]);
 399     if ((x > 0 && x == ptr->io_key_prefetch) &&
 400         memcached_flush_buffers(ptr) != MEMCACHED_SUCCESS)
 401       rc= MEMCACHED_SOME_ERRORS;
 402   }
 403
 404   if (mget_mode)
 405   {
 406     /*
 407      * Send a noop command to flush the buffers
 408    */
 409     protocol_binary_request_noop request= {.bytes= {0}};
 410     request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 411     request.message.header.request.opcode= PROTOCOL_BINARY_CMD_NOOP;
 412     request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 413
 414     for (x= 0; x < ptr->number_of_hosts; x++)
 415       if (memcached_server_response_count(&ptr->hosts[x]))
 416       {
 417         if (memcached_io_write(&ptr->hosts[x], NULL, 0, 1) == -1)
 418         {
 419           memcached_server_response_reset(&ptr->hosts[x]);
 420           memcached_io_reset(&ptr->hosts[x]);
 421           rc= MEMCACHED_SOME_ERRORS;
 422         }
 423
 424         if (memcached_io_write(&ptr->hosts[x], request.bytes,
 425                                sizeof(request.bytes), 1) == -1)
 426         {
 427           memcached_server_response_reset(&ptr->hosts[x]);
 428           memcached_io_reset(&ptr->hosts[x]);
 429           rc= MEMCACHED_SOME_ERRORS;
 430         }
 431       }
 432   }
 433
 434
 435   return rc;
 436 }
 437
 438 static memcached_return_t replication_binary_mget(memcached_st *ptr,
 439                                                   uint32_t* hash,
 440                                                   bool* dead_servers,
 441                                                   const char *const *keys,
 442                                                   const size_t *key_length,
 443                                                   size_t number_of_keys)
 444 {
 445   memcached_return_t rc= MEMCACHED_NOTFOUND;
 446   uint32_t x, start= 0;
 447   uint64_t randomize_read= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_RANDOMIZE_REPLICA_READ);
 448
 449   if (randomize_read)
 450     start= (uint32_t)random() % (uint32_t)(ptr->number_of_replicas + 1);
 451
 452   /* Loop for each replica */
 453   for (uint32_t replica= 0; replica <= ptr->number_of_replicas; ++replica)
 454   {
 455     bool success= true;
 456
 457     for (x= 0; x < number_of_keys; ++x)
 458     {
 459       if (hash[x] == ptr->number_of_hosts)
 460         continue; /* Already successfully sent */
 461
 462       uint32_t server= hash[x] + replica;
 463
 464       /* In case of randomized reads */
 465       if (randomize_read && ((server + start) <= (hash[x] + ptr->number_of_replicas)))
 466         server += start;
 467
 468       while (server >= ptr->number_of_hosts)
 469         server -= ptr->number_of_hosts;
 470
 471       if (dead_servers[server])
 472         continue;
 473
 474       if (memcached_server_response_count(&ptr->hosts[server]) == 0)
 475       {
 476         rc= memcached_connect(&ptr->hosts[server]);
 477         if (rc != MEMCACHED_SUCCESS)
 478         {
 479           memcached_io_reset(&ptr->hosts[server]);
 480           dead_servers[server]= true;
 481           success= false;
 482           continue;
 483         }
 484       }
 485
 486       protocol_binary_request_getk request= {
 487         .message.header.request= {
 488           .magic= PROTOCOL_BINARY_REQ,
 489           .opcode= PROTOCOL_BINARY_CMD_GETK,
 490           .keylen= htons((uint16_t)key_length[x]),
 491           .datatype= PROTOCOL_BINARY_RAW_BYTES,
 492           .bodylen= htonl((uint32_t)key_length[x])
 493         }
 494       };
 495
 496       /*
 497        * We need to disable buffering to actually know that the request was
 498        * successfully sent to the server (so that we should expect a result
 499        * back). It would be nice to do this in buffered mode, but then it
 500        * would be complex to handle all error situations if we got to send
 501        * some of the messages, and then we failed on writing out some others
 502        * and we used the callback interface from memcached_mget_execute so
 503        * that we might have processed some of the responses etc. For now,
 504        * just make sure we work _correctly_
 505      */
 506       if ((memcached_io_write(&ptr->hosts[server], request.bytes,
 507                               sizeof(request.bytes), 0) == -1) ||
 508           (memcached_io_write(&ptr->hosts[server], keys[x],
 509                               key_length[x], 1) == -1))
 510       {
 511         memcached_io_reset(&ptr->hosts[server]);
 512         dead_servers[server]= true;
 513         success= false;
 514         continue;
 515       }
 516
 517       memcached_server_response_increment(&ptr->hosts[server]);
 518       hash[x]= ptr->number_of_hosts;
 519     }
 520
 521     if (success)
 522       break;
 523   }
 524
 525   return rc;
 526 }
 527
 528 static memcached_return_t binary_mget_by_key(memcached_st *ptr,
 529                                              unsigned int master_server_key,
 530                                              bool is_master_key_set,
 531                                              const char * const *keys,
 532                                              const size_t *key_length,
 533                                              size_t number_of_keys,
 534                                              bool mget_mode)
 535 {
 536   memcached_return_t rc;
 537
 538   if (ptr->number_of_replicas == 0)
 539   {
 540     rc= simple_binary_mget(ptr, master_server_key, is_master_key_set,
 541                            keys, key_length, number_of_keys, mget_mode);
 542   }
 543   else
 544   {
 545     uint32_t* hash;
 546     bool* dead_servers;
 547
 548     hash= ptr->call_malloc(ptr, sizeof(uint32_t) * number_of_keys);
 549     dead_servers= ptr->call_calloc(ptr, ptr->number_of_hosts, sizeof(bool));
 550
 551     if (hash == NULL || dead_servers == NULL)
 552     {
 553       ptr->call_free(ptr, hash);
 554       ptr->call_free(ptr, dead_servers);
 555       return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
 556     }
 557
 558     if (is_master_key_set)
 559       for (unsigned int x= 0; x < number_of_keys; x++)
 560         hash[x]= master_server_key;
 561     else
 562       for (unsigned int x= 0; x < number_of_keys; x++)
 563         hash[x]= memcached_generate_hash(ptr, keys[x], key_length[x]);
 564
 565     rc= replication_binary_mget(ptr, hash, dead_servers, keys,
 566                                 key_length, number_of_keys);
 567
 568     ptr->call_free(ptr, hash);
 569     ptr->call_free(ptr, dead_servers);
 570
 571     return MEMCACHED_SUCCESS;
 572   }
 573
 574   return rc;
 575 }