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