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

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