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   memcached_return_t rc;
 359   if (memcached_failed(rc= initialize_query(ptr)))
 360   {
 361     return rc;
 362   }
 363
 364   if (memcached_failed(rc= memcached_validate_key_length(key_length, ptr->flags.binary_protocol)))
 365   {
 366     return rc;
 367   }
 368
 369   if (memcached_failed(memcached_key_test(*ptr, (const char **)&key, &key_length, 1)))
 370   {
 371     return MEMCACHED_BAD_KEY_PROVIDED;
 372   }
 373
 374   uint32_t server_key= memcached_generate_hash_with_redistribution(ptr, group_key, group_key_length);
 375   memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, server_key);
 376
 377   WATCHPOINT_SET(instance->io_wait_count.read= 0);
 378   WATCHPOINT_SET(instance->io_wait_count.write= 0);
 379
 380   if (ptr->flags.binary_protocol)
 381   {
 382     rc= memcached_send_binary(ptr, instance, server_key,
 383                               key, key_length,
 384                               value, value_length, expiration,
 385                               flags, cas, verb);
 386   }
 387   else
 388   {
 389     rc= memcached_send_ascii(ptr, instance,
 390                              key, key_length,
 391                              value, value_length, expiration,
 392                              flags, cas, verb);
 393   }
 394
 395   return rc;
 396 }
 397
 398
 399 memcached_return_t memcached_set(memcached_st *ptr, const char *key, size_t key_length,
 400                                  const char *value, size_t value_length,
 401                                  time_t expiration,
 402                                  uint32_t flags)
 403 {
 404   memcached_return_t rc;
 405   LIBMEMCACHED_MEMCACHED_SET_START();
 406   rc= memcached_send(ptr, key, key_length,
 407                      key, key_length, value, value_length,
 408                      expiration, flags, 0, SET_OP);
 409   LIBMEMCACHED_MEMCACHED_SET_END();
 410   return rc;
 411 }
 412
 413 memcached_return_t memcached_add(memcached_st *ptr,
 414                                  const char *key, size_t key_length,
 415                                  const char *value, size_t value_length,
 416                                  time_t expiration,
 417                                  uint32_t flags)
 418 {
 419   memcached_return_t rc;
 420   LIBMEMCACHED_MEMCACHED_ADD_START();
 421   rc= memcached_send(ptr, key, key_length,
 422                      key, key_length, value, value_length,
 423                      expiration, flags, 0, ADD_OP);
 424   LIBMEMCACHED_MEMCACHED_ADD_END();
 425   return rc;
 426 }
 427
 428 memcached_return_t memcached_replace(memcached_st *ptr,
 429                                      const char *key, size_t key_length,
 430                                      const char *value, size_t value_length,
 431                                      time_t expiration,
 432                                      uint32_t flags)
 433 {
 434   memcached_return_t rc;
 435   LIBMEMCACHED_MEMCACHED_REPLACE_START();
 436   rc= memcached_send(ptr, key, key_length,
 437                      key, key_length, value, value_length,
 438                      expiration, flags, 0, REPLACE_OP);
 439   LIBMEMCACHED_MEMCACHED_REPLACE_END();
 440   return rc;
 441 }
 442
 443 memcached_return_t memcached_prepend(memcached_st *ptr,
 444                                      const char *key, size_t key_length,
 445                                      const char *value, size_t value_length,
 446                                      time_t expiration,
 447                                      uint32_t flags)
 448 {
 449   memcached_return_t rc;
 450   rc= memcached_send(ptr, key, key_length,
 451                      key, key_length, value, value_length,
 452                      expiration, flags, 0, PREPEND_OP);
 453   return rc;
 454 }
 455
 456 memcached_return_t memcached_append(memcached_st *ptr,
 457                                     const char *key, size_t key_length,
 458                                     const char *value, size_t value_length,
 459                                     time_t expiration,
 460                                     uint32_t flags)
 461 {
 462   memcached_return_t rc;
 463   rc= memcached_send(ptr, key, key_length,
 464                      key, key_length, value, value_length,
 465                      expiration, flags, 0, APPEND_OP);
 466   return rc;
 467 }
 468
 469 memcached_return_t memcached_cas(memcached_st *ptr,
 470                                  const char *key, size_t key_length,
 471                                  const char *value, size_t value_length,
 472                                  time_t expiration,
 473                                  uint32_t flags,
 474                                  uint64_t cas)
 475 {
 476   memcached_return_t rc;
 477   rc= memcached_send(ptr, key, key_length,
 478                      key, key_length, value, value_length,
 479                      expiration, flags, cas, CAS_OP);
 480   return rc;
 481 }
 482
 483 memcached_return_t memcached_set_by_key(memcached_st *ptr,
 484                                         const char *group_key,
 485                                         size_t group_key_length,
 486                                         const char *key, size_t key_length,
 487                                         const char *value, size_t value_length,
 488                                         time_t expiration,
 489                                         uint32_t flags)
 490 {
 491   memcached_return_t rc;
 492   LIBMEMCACHED_MEMCACHED_SET_START();
 493   rc= memcached_send(ptr, group_key, group_key_length,
 494                      key, key_length, value, value_length,
 495                      expiration, flags, 0, SET_OP);
 496   LIBMEMCACHED_MEMCACHED_SET_END();
 497   return rc;
 498 }
 499
 500 memcached_return_t memcached_add_by_key(memcached_st *ptr,
 501                                         const char *group_key, size_t group_key_length,
 502                                         const char *key, size_t key_length,
 503                                         const char *value, size_t value_length,
 504                                         time_t expiration,
 505                                         uint32_t flags)
 506 {
 507   memcached_return_t rc;
 508   LIBMEMCACHED_MEMCACHED_ADD_START();
 509   rc= memcached_send(ptr, group_key, group_key_length,
 510                      key, key_length, value, value_length,
 511                      expiration, flags, 0, ADD_OP);
 512   LIBMEMCACHED_MEMCACHED_ADD_END();
 513   return rc;
 514 }
 515
 516 memcached_return_t memcached_replace_by_key(memcached_st *ptr,
 517                                             const char *group_key, size_t group_key_length,
 518                                             const char *key, size_t key_length,
 519                                             const char *value, size_t value_length,
 520                                             time_t expiration,
 521                                             uint32_t flags)
 522 {
 523   memcached_return_t rc;
 524   LIBMEMCACHED_MEMCACHED_REPLACE_START();
 525   rc= memcached_send(ptr, group_key, group_key_length,
 526                      key, key_length, value, value_length,
 527                      expiration, flags, 0, REPLACE_OP);
 528   LIBMEMCACHED_MEMCACHED_REPLACE_END();
 529   return rc;
 530 }
 531
 532 memcached_return_t memcached_prepend_by_key(memcached_st *ptr,
 533                                             const char *group_key, size_t group_key_length,
 534                                             const char *key, size_t key_length,
 535                                             const char *value, size_t value_length,
 536                                             time_t expiration,
 537                                             uint32_t flags)
 538 {
 539   memcached_return_t rc;
 540   rc= memcached_send(ptr, group_key, group_key_length,
 541                      key, key_length, value, value_length,
 542                      expiration, flags, 0, PREPEND_OP);
 543   return rc;
 544 }
 545
 546 memcached_return_t memcached_append_by_key(memcached_st *ptr,
 547                                            const char *group_key, size_t group_key_length,
 548                                            const char *key, size_t key_length,
 549                                            const char *value, size_t value_length,
 550                                            time_t expiration,
 551                                            uint32_t flags)
 552 {
 553   memcached_return_t rc;
 554   rc= memcached_send(ptr, group_key, group_key_length,
 555                      key, key_length, value, value_length,
 556                      expiration, flags, 0, APPEND_OP);
 557   return rc;
 558 }
 559
 560 memcached_return_t memcached_cas_by_key(memcached_st *ptr,
 561                                         const char *group_key, size_t group_key_length,
 562                                         const char *key, size_t key_length,
 563                                         const char *value, size_t value_length,
 564                                         time_t expiration,
 565                                         uint32_t flags,
 566                                         uint64_t cas)
 567 {
 568   memcached_return_t rc;
 569   rc= memcached_send(ptr, group_key, group_key_length,
 570                      key, key_length, value, value_length,
 571                      expiration, flags, cas, CAS_OP);
 572   return rc;
 573 }
 574