git.m6w6.name Git - m6w6/libmemcached/blob - libmemcached/storage.cc

   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: Storage related functions, aka set, replace,..
   9  *
  10  */
  11
  12 #include <libmemcached/common.h>
  13
  14 enum memcached_storage_action_t {
  15   SET_OP,
  16   REPLACE_OP,
  17   ADD_OP,
  18   PREPEND_OP,
  19   APPEND_OP,
  20   CAS_OP
  21 };
  22
  23 /* Inline this */
  24 static inline const char *storage_op_string(memcached_storage_action_t verb)
  25 {
  26   switch (verb)
  27   {
  28   case REPLACE_OP:
  29     return "replace ";
  30
  31   case ADD_OP:
  32     return "add ";
  33
  34   case PREPEND_OP:
  35     return "prepend ";
  36
  37   case APPEND_OP:
  38     return "append ";
  39
  40   case CAS_OP:
  41     return "cas ";
  42
  43   case SET_OP:
  44     break;
  45   }
  46
  47   return "set ";
  48 }
  49
  50 static inline uint8_t get_com_code(memcached_storage_action_t verb, bool noreply)
  51 {
  52   /* 0 isn't a value we want, but GCC 4.2 seems to think ret can otherwise
  53    * be used uninitialized in this function. FAIL */
  54   uint8_t ret= 0;
  55
  56   if (noreply)
  57     switch (verb)
  58     {
  59     case SET_OP:
  60       ret=PROTOCOL_BINARY_CMD_SETQ;
  61       break;
  62     case ADD_OP:
  63       ret=PROTOCOL_BINARY_CMD_ADDQ;
  64       break;
  65     case CAS_OP: /* FALLTHROUGH */
  66     case REPLACE_OP:
  67       ret=PROTOCOL_BINARY_CMD_REPLACEQ;
  68       break;
  69     case APPEND_OP:
  70       ret=PROTOCOL_BINARY_CMD_APPENDQ;
  71       break;
  72     case PREPEND_OP:
  73       ret=PROTOCOL_BINARY_CMD_PREPENDQ;
  74       break;
  75     default:
  76       WATCHPOINT_ASSERT(verb);
  77       break;
  78     }
  79   else
  80     switch (verb)
  81     {
  82     case SET_OP:
  83       ret=PROTOCOL_BINARY_CMD_SET;
  84       break;
  85     case ADD_OP:
  86       ret=PROTOCOL_BINARY_CMD_ADD;
  87       break;
  88     case CAS_OP: /* FALLTHROUGH */
  89     case REPLACE_OP:
  90       ret=PROTOCOL_BINARY_CMD_REPLACE;
  91       break;
  92     case APPEND_OP:
  93       ret=PROTOCOL_BINARY_CMD_APPEND;
  94       break;
  95     case PREPEND_OP:
  96       ret=PROTOCOL_BINARY_CMD_PREPEND;
  97       break;
  98     default:
  99       WATCHPOINT_ASSERT(verb);
 100       break;
 101     }
 102
 103   return ret;
 104 }
 105
 106 static memcached_return_t memcached_send_binary(memcached_st *ptr,
 107                                                 memcached_server_write_instance_st server,
 108                                                 uint32_t server_key,
 109                                                 const char *key,
 110                                                 size_t key_length,
 111                                                 const char *value,
 112                                                 size_t value_length,
 113                                                 time_t expiration,
 114                                                 uint32_t flags,
 115                                                 uint64_t cas,
 116                                                 memcached_storage_action_t verb)
 117 {
 118   bool flush;
 119   protocol_binary_request_set request= {};
 120   size_t send_length= sizeof(request.bytes);
 121
 122   bool noreply= server->root->flags.no_reply;
 123
 124   request.message.header.request.magic= PROTOCOL_BINARY_REQ;
 125   request.message.header.request.opcode= get_com_code(verb, noreply);
 126   request.message.header.request.keylen= htons((uint16_t)(key_length + memcached_array_size(ptr->_namespace)));
 127   request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
 128   if (verb == APPEND_OP || verb == PREPEND_OP)
 129     send_length -= 8; /* append & prepend does not contain extras! */
 130   else
 131   {
 132     request.message.header.request.extlen= 8;
 133     request.message.body.flags= htonl(flags);
 134     request.message.body.expiration= htonl((uint32_t)expiration);
 135   }
 136
 137   request.message.header.request.bodylen= htonl((uint32_t) (key_length + memcached_array_size(ptr->_namespace) + value_length +
 138                                                             request.message.header.request.extlen));
 139
 140   if (cas)
 141     request.message.header.request.cas= memcached_htonll(cas);
 142
 143   flush= (bool) ((server->root->flags.buffer_requests && verb == SET_OP) ? 0 : 1);
 144
 145   if (server->root->flags.use_udp && ! flush)
 146   {
 147     size_t cmd_size= send_length + key_length + value_length;
 148
 149     if (cmd_size > MAX_UDP_DATAGRAM_LENGTH - UDP_DATAGRAM_HEADER_LENGTH)
 150     {
 151       return MEMCACHED_WRITE_FAILURE;
 152     }
 153     if (cmd_size + server->write_buffer_offset > MAX_UDP_DATAGRAM_LENGTH)
 154     {
 155       memcached_io_write(server, NULL, 0, true);
 156     }
 157   }
 158
 159   struct libmemcached_io_vector_st vector[]=
 160   {
 161     { send_length, request.bytes },
 162     { memcached_array_size(ptr->_namespace), memcached_array_string(ptr->_namespace) },
 163     { key_length, key },
 164     { value_length, value }
 165   };
 166
 167   /* write the header */
 168   memcached_return_t rc;
 169   if ((rc= memcached_vdo(server, vector, 4, flush)) != MEMCACHED_SUCCESS)
 170   {
 171     memcached_io_reset(server);
 172     return (rc == MEMCACHED_SUCCESS) ? MEMCACHED_WRITE_FAILURE : rc;
 173   }
 174
 175   if (verb == SET_OP && ptr->number_of_replicas > 0)
 176   {
 177     request.message.header.request.opcode= PROTOCOL_BINARY_CMD_SETQ;
 178     WATCHPOINT_STRING("replicating");
 179
 180     for (uint32_t x= 0; x < ptr->number_of_replicas; x++)
 181     {
 182       memcached_server_write_instance_st instance;
 183
 184       ++server_key;
 185       if (server_key == memcached_server_count(ptr))
 186         server_key= 0;
 187
 188       instance= memcached_server_instance_fetch(ptr, server_key);
 189
 190       if (memcached_vdo(instance, vector, 4, false) != MEMCACHED_SUCCESS)
 191       {
 192         memcached_io_reset(instance);
 193       }
 194       else
 195       {
 196         memcached_server_response_decrement(instance);
 197       }
 198     }
 199   }
 200
 201   if (flush == false)
 202   {
 203     return MEMCACHED_BUFFERED;
 204   }
 205
 206   if (noreply)
 207   {
 208     return MEMCACHED_SUCCESS;
 209   }
 210
 211   return memcached_response(server, NULL, 0, NULL);
 212 }
 213
 214 static memcached_return_t memcached_send_ascii(memcached_st *ptr,
 215                                                memcached_server_write_instance_st instance,
 216                                                const char *key,
 217                                                size_t key_length,
 218                                                const char *value,
 219                                                size_t value_length,
 220                                                time_t expiration,
 221                                                uint32_t flags,
 222                                                uint64_t cas,
 223                                                memcached_storage_action_t verb)
 224 {
 225   bool to_write;
 226   size_t write_length;
 227   char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
 228   memcached_return_t rc;
 229
 230   if (cas)
 231   {
 232     int check_length;
 233     check_length= snprintf(buffer, MEMCACHED_DEFAULT_COMMAND_SIZE,
 234                            "%s %.*s%.*s %u %llu %lu %llu%s\r\n",
 235                            storage_op_string(verb),
 236                            memcached_print_array(ptr->_namespace),
 237                            (int)key_length, key, flags,
 238                            (unsigned long long)expiration, (unsigned long)value_length,
 239                            (unsigned long long)cas,
 240                            (ptr->flags.no_reply) ? " noreply" : "");
 241     if (check_length >= MEMCACHED_DEFAULT_COMMAND_SIZE || check_length < 0)
 242     {
 243       rc= memcached_set_error(*instance, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT,
 244                               memcached_literal_param("snprintf(MEMCACHED_DEFAULT_COMMAND_SIZE)"));
 245       memcached_io_reset(instance);
 246
 247       return rc;
 248     }
 249     write_length= check_length;
 250   }
 251   else
 252   {
 253     char *buffer_ptr= buffer;
 254     const char *command= storage_op_string(verb);
 255
 256     /* Copy in the command, no space needed, we handle that in the command function*/
 257     memcpy(buffer_ptr, command, strlen(command));
 258
 259     /* Copy in the key prefix, switch to the buffer_ptr */
 260     buffer_ptr= (char *)memcpy((char *)(buffer_ptr + strlen(command)), (char *)memcached_array_string(ptr->_namespace), memcached_array_size(ptr->_namespace));
 261
 262     /* Copy in the key, adjust point if a key prefix was used. */
 263     buffer_ptr= (char *)memcpy(buffer_ptr + memcached_array_size(ptr->_namespace),
 264                                key, key_length);
 265     buffer_ptr+= key_length;
 266     buffer_ptr[0]=  ' ';
 267     buffer_ptr++;
 268
 269     write_length= (size_t)(buffer_ptr - buffer);
 270     int check_length= snprintf(buffer_ptr, MEMCACHED_DEFAULT_COMMAND_SIZE -(size_t)(buffer_ptr - buffer),
 271                                "%u %llu %lu%s\r\n",
 272                                flags,
 273                                (unsigned long long)expiration, (unsigned long)value_length,
 274                                ptr->flags.no_reply ? " noreply" : "");
 275     if ((size_t)check_length >= MEMCACHED_DEFAULT_COMMAND_SIZE -size_t(buffer_ptr - buffer) || check_length < 0)
 276     {
 277       rc= memcached_set_error(*ptr, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT,
 278                               memcached_literal_param("snprintf(MEMCACHED_DEFAULT_COMMAND_SIZE)"));
 279       memcached_io_reset(instance);
 280
 281       return rc;
 282     }
 283
 284     write_length+= (size_t)check_length;
 285     WATCHPOINT_ASSERT(write_length < MEMCACHED_DEFAULT_COMMAND_SIZE);
 286   }
 287
 288   if (ptr->flags.use_udp && ptr->flags.buffer_requests)
 289   {
 290     size_t cmd_size= write_length + value_length +2;
 291     if (cmd_size > MAX_UDP_DATAGRAM_LENGTH - UDP_DATAGRAM_HEADER_LENGTH)
 292       return memcached_set_error(*ptr, MEMCACHED_WRITE_FAILURE, MEMCACHED_AT);
 293
 294     if (cmd_size + instance->write_buffer_offset > MAX_UDP_DATAGRAM_LENGTH)
 295       memcached_io_write(instance, NULL, 0, true);
 296   }
 297
 298   if (write_length >= MEMCACHED_DEFAULT_COMMAND_SIZE)
 299   {
 300     rc= memcached_set_error(*ptr, MEMCACHED_WRITE_FAILURE, MEMCACHED_AT);
 301   }
 302   else
 303   {
 304     struct libmemcached_io_vector_st vector[]=
 305     {
 306       { write_length, buffer },
 307       { value_length, value },
 308       { 2, "\r\n" }
 309     };
 310
 311     if (ptr->flags.buffer_requests && verb == SET_OP)
 312     {
 313       to_write= false;
 314     }
 315     else
 316     {
 317       to_write= true;
 318     }
 319
 320     /* Send command header */
 321     rc=  memcached_vdo(instance, vector, 3, to_write);
 322     if (rc == MEMCACHED_SUCCESS)
 323     {
 324
 325       if (ptr->flags.no_reply)
 326       {
 327         rc= (to_write == false) ? MEMCACHED_BUFFERED : MEMCACHED_SUCCESS;
 328       }
 329       else if (to_write == false)
 330       {
 331         rc= MEMCACHED_BUFFERED;
 332       }
 333       else
 334       {
 335         rc= memcached_response(instance, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, NULL);
 336
 337         if (rc == MEMCACHED_STORED)
 338           rc= MEMCACHED_SUCCESS;
 339       }
 340     }
 341   }
 342
 343   if (rc == MEMCACHED_WRITE_FAILURE)
 344     memcached_io_reset(instance);
 345
 346   return rc;
 347 }
 348
 349 static inline memcached_return_t memcached_send(memcached_st *ptr,
 350                                                 const char *group_key, size_t group_key_length,
 351                                                 const char *key, size_t key_length,
 352                                                 const char *value, size_t value_length,
 353                                                 time_t expiration,
 354                                                 uint32_t flags,
 355                                                 uint64_t cas,
 356                                                 memcached_storage_action_t verb)
 357 {
 358   WATCHPOINT_ASSERT(!(value == NULL && value_length > 0));
 359
 360   memcached_return_t rc;
 361   if (memcached_failed(rc= initialize_query(ptr)))
 362   {
 363     return rc;
 364   }
 365
 366   if (memcached_failed(rc= memcached_validate_key_length(key_length, ptr->flags.binary_protocol)))
 367   {
 368     return rc;
 369   }
 370
 371   if (memcached_failed(memcached_key_test(*ptr, (const char **)&key, &key_length, 1)))
 372   {
 373     return MEMCACHED_BAD_KEY_PROVIDED;
 374   }
 375
 376   uint32_t server_key= memcached_generate_hash_with_redistribution(ptr, group_key, group_key_length);
 377   memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, server_key);
 378
 379   WATCHPOINT_SET(instance->io_wait_count.read= 0);
 380   WATCHPOINT_SET(instance->io_wait_count.write= 0);
 381
 382   if (ptr->flags.binary_protocol)
 383   {
 384     rc= memcached_send_binary(ptr, instance, server_key,
 385                               key, key_length,
 386                               value, value_length, expiration,
 387                               flags, cas, verb);
 388   }
 389   else
 390   {
 391     rc= memcached_send_ascii(ptr, instance,
 392                              key, key_length,
 393                              value, value_length, expiration,
 394                              flags, cas, verb);
 395   }
 396
 397   return rc;
 398 }
 399
 400
 401 memcached_return_t memcached_set(memcached_st *ptr, const char *key, size_t key_length,
 402                                  const char *value, size_t value_length,
 403                                  time_t expiration,
 404                                  uint32_t flags)
 405 {
 406   memcached_return_t rc;
 407   LIBMEMCACHED_MEMCACHED_SET_START();
 408   rc= memcached_send(ptr, key, key_length,
 409                      key, key_length, value, value_length,
 410                      expiration, flags, 0, SET_OP);
 411   LIBMEMCACHED_MEMCACHED_SET_END();
 412   return rc;
 413 }
 414
 415 memcached_return_t memcached_add(memcached_st *ptr,
 416                                  const char *key, size_t key_length,
 417                                  const char *value, size_t value_length,
 418                                  time_t expiration,
 419                                  uint32_t flags)
 420 {
 421   memcached_return_t rc;
 422   LIBMEMCACHED_MEMCACHED_ADD_START();
 423   rc= memcached_send(ptr, key, key_length,
 424                      key, key_length, value, value_length,
 425                      expiration, flags, 0, ADD_OP);
 426   LIBMEMCACHED_MEMCACHED_ADD_END();
 427   return rc;
 428 }
 429
 430 memcached_return_t memcached_replace(memcached_st *ptr,
 431                                      const char *key, size_t key_length,
 432                                      const char *value, size_t value_length,
 433                                      time_t expiration,
 434                                      uint32_t flags)
 435 {
 436   memcached_return_t rc;
 437   LIBMEMCACHED_MEMCACHED_REPLACE_START();
 438   rc= memcached_send(ptr, key, key_length,
 439                      key, key_length, value, value_length,
 440                      expiration, flags, 0, REPLACE_OP);
 441   LIBMEMCACHED_MEMCACHED_REPLACE_END();
 442   return rc;
 443 }
 444
 445 memcached_return_t memcached_prepend(memcached_st *ptr,
 446                                      const char *key, size_t key_length,
 447                                      const char *value, size_t value_length,
 448                                      time_t expiration,
 449                                      uint32_t flags)
 450 {
 451   memcached_return_t rc;
 452   rc= memcached_send(ptr, key, key_length,
 453                      key, key_length, value, value_length,
 454                      expiration, flags, 0, PREPEND_OP);
 455   return rc;
 456 }
 457
 458 memcached_return_t memcached_append(memcached_st *ptr,
 459                                     const char *key, size_t key_length,
 460                                     const char *value, size_t value_length,
 461                                     time_t expiration,
 462                                     uint32_t flags)
 463 {
 464   memcached_return_t rc;
 465   rc= memcached_send(ptr, key, key_length,
 466                      key, key_length, value, value_length,
 467                      expiration, flags, 0, APPEND_OP);
 468   return rc;
 469 }
 470
 471 memcached_return_t memcached_cas(memcached_st *ptr,
 472                                  const char *key, size_t key_length,
 473                                  const char *value, size_t value_length,
 474                                  time_t expiration,
 475                                  uint32_t flags,
 476                                  uint64_t cas)
 477 {
 478   memcached_return_t rc;
 479   rc= memcached_send(ptr, key, key_length,
 480                      key, key_length, value, value_length,
 481                      expiration, flags, cas, CAS_OP);
 482   return rc;
 483 }
 484
 485 memcached_return_t memcached_set_by_key(memcached_st *ptr,
 486                                         const char *group_key,
 487                                         size_t group_key_length,
 488                                         const char *key, size_t key_length,
 489                                         const char *value, size_t value_length,
 490                                         time_t expiration,
 491                                         uint32_t flags)
 492 {
 493   memcached_return_t rc;
 494   LIBMEMCACHED_MEMCACHED_SET_START();
 495   rc= memcached_send(ptr, group_key, group_key_length,
 496                      key, key_length, value, value_length,
 497                      expiration, flags, 0, SET_OP);
 498   LIBMEMCACHED_MEMCACHED_SET_END();
 499   return rc;
 500 }
 501
 502 memcached_return_t memcached_add_by_key(memcached_st *ptr,
 503                                         const char *group_key, size_t group_key_length,
 504                                         const char *key, size_t key_length,
 505                                         const char *value, size_t value_length,
 506                                         time_t expiration,
 507                                         uint32_t flags)
 508 {
 509   memcached_return_t rc;
 510   LIBMEMCACHED_MEMCACHED_ADD_START();
 511   rc= memcached_send(ptr, group_key, group_key_length,
 512                      key, key_length, value, value_length,
 513                      expiration, flags, 0, ADD_OP);
 514   LIBMEMCACHED_MEMCACHED_ADD_END();
 515   return rc;
 516 }
 517
 518 memcached_return_t memcached_replace_by_key(memcached_st *ptr,
 519                                             const char *group_key, size_t group_key_length,
 520                                             const char *key, size_t key_length,
 521                                             const char *value, size_t value_length,
 522                                             time_t expiration,
 523                                             uint32_t flags)
 524 {
 525   memcached_return_t rc;
 526   LIBMEMCACHED_MEMCACHED_REPLACE_START();
 527   rc= memcached_send(ptr, group_key, group_key_length,
 528                      key, key_length, value, value_length,
 529                      expiration, flags, 0, REPLACE_OP);
 530   LIBMEMCACHED_MEMCACHED_REPLACE_END();
 531   return rc;
 532 }
 533
 534 memcached_return_t memcached_prepend_by_key(memcached_st *ptr,
 535                                             const char *group_key, size_t group_key_length,
 536                                             const char *key, size_t key_length,
 537                                             const char *value, size_t value_length,
 538                                             time_t expiration,
 539                                             uint32_t flags)
 540 {
 541   memcached_return_t rc;
 542   rc= memcached_send(ptr, group_key, group_key_length,
 543                      key, key_length, value, value_length,
 544                      expiration, flags, 0, PREPEND_OP);
 545   return rc;
 546 }
 547
 548 memcached_return_t memcached_append_by_key(memcached_st *ptr,
 549                                            const char *group_key, size_t group_key_length,
 550                                            const char *key, size_t key_length,
 551                                            const char *value, size_t value_length,
 552                                            time_t expiration,
 553                                            uint32_t flags)
 554 {
 555   memcached_return_t rc;
 556   rc= memcached_send(ptr, group_key, group_key_length,
 557                      key, key_length, value, value_length,
 558                      expiration, flags, 0, APPEND_OP);
 559   return rc;
 560 }
 561
 562 memcached_return_t memcached_cas_by_key(memcached_st *ptr,
 563                                         const char *group_key, size_t group_key_length,
 564                                         const char *key, size_t key_length,
 565                                         const char *value, size_t value_length,
 566                                         time_t expiration,
 567                                         uint32_t flags,
 568                                         uint64_t cas)
 569 {
 570   memcached_return_t rc;
 571   rc= memcached_send(ptr, group_key, group_key_length,
 572                      key, key_length, value, value_length,
 573                      expiration, flags, cas, CAS_OP);
 574   return rc;
 575 }
 576