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 *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                                         memcached_execute_function *callback,
 288                                         void *context,
 289                                         unsigned int number_of_callbacks)
 290 {
 291   if ((ptr->flags & MEM_BINARY_PROTOCOL) == 0)
 292     return MEMCACHED_NOT_SUPPORTED;
 293
 294   memcached_return rc;
 295   memcached_callback_st *original_callbacks= ptr->callbacks;
 296   memcached_callback_st cb= {
 297     .callback= callback,
 298     .context= context,
 299     .number_of_callback= 1
 300   };
 301
 302   ptr->callbacks= &cb;
 303   rc= memcached_mget_by_key(ptr, master_key, master_key_length, keys,
 304                             key_length, number_of_keys);
 305   ptr->callbacks= original_callbacks;
 306   return rc;
 307 }
 308
 309 static memcached_return simple_binary_mget(memcached_st *ptr,
 310                                            unsigned int master_server_key,
 311                                            bool is_master_key_set,
 312                                            const char * const *keys,
 313                                            const size_t *key_length,
 314                                            size_t number_of_keys, bool mget_mode)
 315 {
 316   memcached_return rc= MEMCACHED_NOTFOUND;
 317   uint32_t x;
 318
 319   int flush= number_of_keys == 1;
 320
 321   /*
 322     If a server fails we warn about errors and start all over with sending keys
 323     to the server.
 324   */
 325   for (x= 0; x < number_of_keys; x++)
 326   {
 327     unsigned int server_key;
 328
 329     if (is_master_key_set)
 330       server_key= master_server_key;
 331     else
 332       server_key= memcached_generate_hash(ptr, keys[x], key_length[x]);
 333
 334     if (memcached_server_response_count(&ptr->hosts[server_key]) == 0)
 335     {
 336       rc= memcached_connect(&ptr->hosts[server_key]);
 337       if (rc != MEMCACHED_SUCCESS)
 338         continue;
 339     }
 340
 341     protocol_binary_request_getk request= {.bytes= {0}};
 342     request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 343     if (mget_mode)
 344       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETKQ;
 345     else
 346       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
 347
 348     memcached_return vk;
 349     vk= memcached_validate_key_length(key_length[x],
 350                                       ptr->flags & MEM_BINARY_PROTOCOL);
 351     unlikely (vk != MEMCACHED_SUCCESS)
 352     {
 353       if (x > 0)
 354         memcached_io_reset(&ptr->hosts[server_key]);
 355       return vk;
 356     }
 357
 358     request.message.header.request.keylen= htons((uint16_t)key_length[x]);
 359     request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 360     request.message.header.request.bodylen= htonl((uint32_t) key_length[x]);
 361
 362     if ((memcached_io_write(&ptr->hosts[server_key], request.bytes,
 363                             sizeof(request.bytes), 0) == -1) ||
 364         (memcached_io_write(&ptr->hosts[server_key], keys[x],
 365                             key_length[x], (char) flush) == -1))
 366     {
 367       memcached_server_response_reset(&ptr->hosts[server_key]);
 368       rc= MEMCACHED_SOME_ERRORS;
 369       continue;
 370     }
 371
 372     /* We just want one pending response per server */
 373     memcached_server_response_reset(&ptr->hosts[server_key]);
 374     memcached_server_response_increment(&ptr->hosts[server_key]);
 375     if ((x > 0 && x == ptr->io_key_prefetch) &&
 376         memcached_flush_buffers(ptr) != MEMCACHED_SUCCESS)
 377       rc= MEMCACHED_SOME_ERRORS;
 378   }
 379
 380   if (mget_mode)
 381   {
 382     /*
 383      * Send a noop command to flush the buffers
 384      */
 385     protocol_binary_request_noop request= {.bytes= {0}};
 386     request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 387     request.message.header.request.opcode= PROTOCOL_BINARY_CMD_NOOP;
 388     request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 389
 390     for (x= 0; x < ptr->number_of_hosts; x++)
 391       if (memcached_server_response_count(&ptr->hosts[x]))
 392       {
 393         if (memcached_io_write(&ptr->hosts[x], NULL, 0, 1) == -1)
 394         {
 395           memcached_server_response_reset(&ptr->hosts[x]);
 396           memcached_io_reset(&ptr->hosts[x]);
 397           rc= MEMCACHED_SOME_ERRORS;
 398         }
 399
 400         if (memcached_io_write(&ptr->hosts[x], request.bytes,
 401                                sizeof(request.bytes), 1) == -1)
 402         {
 403           memcached_server_response_reset(&ptr->hosts[x]);
 404           memcached_io_reset(&ptr->hosts[x]);
 405           rc= MEMCACHED_SOME_ERRORS;
 406         }
 407       }
 408     }
 409
 410
 411   return rc;
 412 }
 413
 414 static memcached_return replication_binary_mget(memcached_st *ptr,
 415                                                 uint32_t* hash,
 416                                                 bool* dead_servers,
 417                                                 const char *const *keys,
 418                                                 const size_t *key_length,
 419                                                 size_t number_of_keys,
 420                                                 bool mget_mode)
 421 {
 422   memcached_return rc= MEMCACHED_NOTFOUND;
 423   uint32_t x;
 424
 425   int flush= number_of_keys == 1;
 426
 427   for (uint32_t replica= 0; replica <= ptr->number_of_replicas; ++replica)
 428   {
 429     bool success= true;
 430
 431     for (x= 0; x < number_of_keys; ++x)
 432     {
 433       if (hash[x] == ptr->number_of_hosts)
 434         continue; /* Already successfully sent */
 435
 436       uint32_t server= hash[x] + replica;
 437       while (server >= ptr->number_of_hosts)
 438         server -= ptr->number_of_hosts;
 439
 440       if (dead_servers[server])
 441         continue;
 442
 443       if (memcached_server_response_count(&ptr->hosts[server]) == 0)
 444       {
 445         rc= memcached_connect(&ptr->hosts[server]);
 446         if (rc != MEMCACHED_SUCCESS)
 447         {
 448           memcached_io_reset(&ptr->hosts[server]);
 449           dead_servers[server]= true;
 450           success= false;
 451           continue;
 452         }
 453       }
 454
 455       protocol_binary_request_getk request= {.bytes= {0}};
 456       request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 457       if (mget_mode)
 458         request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETKQ;
 459       else
 460         request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
 461
 462       request.message.header.request.keylen= htons((uint16_t)key_length[x]);
 463       request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 464       request.message.header.request.bodylen= htonl((uint32_t) key_length[x]);
 465
 466       if ((memcached_io_write(&ptr->hosts[server], request.bytes,
 467                               sizeof(request.bytes), 0) == -1) ||
 468           (memcached_io_write(&ptr->hosts[server], keys[x],
 469                               key_length[x], (char) flush) == -1))
 470       {
 471         memcached_io_reset(&ptr->hosts[server]);
 472         dead_servers[server]= true;
 473         success= false;
 474         continue;
 475       }
 476       /* we just want one pending response per server */
 477       memcached_server_response_reset(&ptr->hosts[server]);
 478       memcached_server_response_increment(&ptr->hosts[server]);
 479     }
 480
 481     if (mget_mode)
 482     {
 483       /*
 484        * Send a noop command to flush the buffers
 485        */
 486       protocol_binary_request_noop request= {.bytes= {0}};
 487       request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 488       request.message.header.request.opcode= PROTOCOL_BINARY_CMD_NOOP;
 489       request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 490
 491       for (x= 0; x < ptr->number_of_hosts; x++)
 492         if (memcached_server_response_count(&ptr->hosts[x]))
 493         {
 494           if (memcached_io_write(&ptr->hosts[x], request.bytes,
 495                                  sizeof(request.bytes), 1) == -1)
 496           {
 497             memcached_io_reset(&ptr->hosts[x]);
 498             dead_servers[x]= true;
 499             success= false;
 500           }
 501
 502           /* mark all of the messages bound for this server as sent! */
 503           for (x= 0; x < number_of_keys; ++x)
 504             if (hash[x] == x)
 505               hash[x]= ptr->number_of_hosts;
 506         }
 507     }
 508
 509     if (success)
 510       break;
 511   }
 512
 513   return rc;
 514 }
 515
 516 static memcached_return binary_mget_by_key(memcached_st *ptr,
 517                                            unsigned int master_server_key,
 518                                            bool is_master_key_set,
 519                                            const char * const *keys,
 520                                            const size_t *key_length,
 521                                            size_t number_of_keys,
 522                                            bool mget_mode)
 523 {
 524   memcached_return rc;
 525
 526   if (ptr->number_of_replicas == 0)
 527   {
 528     rc= simple_binary_mget(ptr, master_server_key, is_master_key_set,
 529                            keys, key_length, number_of_keys, mget_mode);
 530   }
 531   else
 532   {
 533     uint32_t* hash;
 534     bool* dead_servers;
 535
 536     hash= ptr->call_malloc(ptr, sizeof(uint32_t) * number_of_keys);
 537     dead_servers= ptr->call_calloc(ptr, ptr->number_of_hosts, sizeof(bool));
 538
 539     if (hash == NULL || dead_servers == NULL)
 540     {
 541       ptr->call_free(ptr, hash);
 542       ptr->call_free(ptr, dead_servers);
 543       return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
 544     }
 545
 546     if (is_master_key_set)
 547       for (unsigned int x= 0; x < number_of_keys; x++)
 548         hash[x]= master_server_key;
 549     else
 550       for (unsigned int x= 0; x < number_of_keys; x++)
 551         hash[x]= memcached_generate_hash(ptr, keys[x], key_length[x]);
 552
 553     rc= replication_binary_mget(ptr, hash, dead_servers, keys,
 554                                 key_length, number_of_keys, mget_mode);
 555
 556     ptr->call_free(ptr, hash);
 557     ptr->call_free(ptr, dead_servers);
 558
 559     return MEMCACHED_SUCCESS;
 560   }
 561
 562   return rc;
 563 }