git.m6w6.name Git - awesomized/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   {
 130     send_length -= 8; /* append & prepend does not contain extras! */
 131   }
 132   else
 133   {
 134     request.message.header.request.extlen= 8;
 135     request.message.body.flags= htonl(flags);
 136     request.message.body.expiration= htonl((uint32_t)expiration);
 137   }
 138
 139   request.message.header.request.bodylen= htonl((uint32_t) (key_length + memcached_array_size(ptr->_namespace) + value_length +
 140                                                             request.message.header.request.extlen));
 141
 142   if (cas)
 143   {
 144     request.message.header.request.cas= memcached_htonll(cas);
 145   }
 146
 147   flush= (bool) ((server->root->flags.buffer_requests && verb == SET_OP) ? 0 : 1);
 148
 149   if (server->root->flags.use_udp && ! flush)
 150   {
 151     size_t cmd_size= send_length + key_length + value_length;
 152
 153     if (cmd_size > MAX_UDP_DATAGRAM_LENGTH - UDP_DATAGRAM_HEADER_LENGTH)
 154     {
 155       return MEMCACHED_WRITE_FAILURE;
 156     }
 157     if (cmd_size + server->write_buffer_offset > MAX_UDP_DATAGRAM_LENGTH)
 158     {
 159       memcached_io_write(server, NULL, 0, true);
 160     }
 161   }
 162
 163   struct libmemcached_io_vector_st vector[]=
 164   {
 165     { send_length, request.bytes },
 166     { memcached_array_size(ptr->_namespace), memcached_array_string(ptr->_namespace) },
 167     { key_length, key },
 168     { value_length, value }
 169   };
 170
 171   /* write the header */
 172   memcached_return_t rc;
 173   if ((rc= memcached_vdo(server, vector, 4, flush)) != MEMCACHED_SUCCESS)
 174   {
 175     memcached_io_reset(server);
 176     return (rc == MEMCACHED_SUCCESS) ? MEMCACHED_WRITE_FAILURE : rc;
 177   }
 178
 179   if (verb == SET_OP && ptr->number_of_replicas > 0)
 180   {
 181     request.message.header.request.opcode= PROTOCOL_BINARY_CMD_SETQ;
 182     WATCHPOINT_STRING("replicating");
 183
 184     for (uint32_t x= 0; x < ptr->number_of_replicas; x++)
 185     {
 186       memcached_server_write_instance_st instance;
 187
 188       ++server_key;
 189       if (server_key == memcached_server_count(ptr))
 190         server_key= 0;
 191
 192       instance= memcached_server_instance_fetch(ptr, server_key);
 193
 194       if (memcached_vdo(instance, vector, 4, false) != MEMCACHED_SUCCESS)
 195       {
 196         memcached_io_reset(instance);
 197       }
 198       else
 199       {
 200         memcached_server_response_decrement(instance);
 201       }
 202     }
 203   }
 204
 205   if (flush == false)
 206   {
 207     return MEMCACHED_BUFFERED;
 208   }
 209
 210   if (noreply)
 211   {
 212     return MEMCACHED_SUCCESS;
 213   }
 214
 215   return memcached_response(server, NULL, 0, NULL);
 216 }
 217
 218 static memcached_return_t memcached_send_ascii(memcached_st *ptr,
 219                                                memcached_server_write_instance_st instance,
 220                                                const char *key,
 221                                                size_t key_length,
 222                                                const char *value,
 223                                                size_t value_length,
 224                                                time_t expiration,
 225                                                uint32_t flags,
 226                                                uint64_t cas,
 227                                                memcached_storage_action_t verb)
 228 {
 229   bool to_write;
 230   size_t write_length;
 231   char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
 232   memcached_return_t rc;
 233
 234   if (cas)
 235   {
 236     int check_length;
 237     check_length= snprintf(buffer, MEMCACHED_DEFAULT_COMMAND_SIZE,
 238                            "%s %.*s%.*s %u %llu %lu %llu%s\r\n",
 239                            storage_op_string(verb),
 240                            memcached_print_array(ptr->_namespace),
 241                            (int)key_length, key, flags,
 242                            (unsigned long long)expiration, (unsigned long)value_length,
 243                            (unsigned long long)cas,
 244                            (ptr->flags.no_reply) ? " noreply" : "");
 245     if (check_length >= MEMCACHED_DEFAULT_COMMAND_SIZE || check_length < 0)
 246     {
 247       rc= memcached_set_error(*instance, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT,
 248                               memcached_literal_param("snprintf(MEMCACHED_DEFAULT_COMMAND_SIZE)"));
 249       memcached_io_reset(instance);
 250
 251       return rc;
 252     }
 253     write_length= check_length;
 254   }
 255   else
 256   {
 257     char *buffer_ptr= buffer;
 258     const char *command= storage_op_string(verb);
 259
 260     /* Copy in the command, no space needed, we handle that in the command function*/
 261     memcpy(buffer_ptr, command, strlen(command));
 262
 263     /* Copy in the key prefix, switch to the buffer_ptr */
 264     buffer_ptr= (char *)memcpy((char *)(buffer_ptr + strlen(command)), (char *)memcached_array_string(ptr->_namespace), memcached_array_size(ptr->_namespace));
 265
 266     /* Copy in the key, adjust point if a key prefix was used. */
 267     buffer_ptr= (char *)memcpy(buffer_ptr + memcached_array_size(ptr->_namespace),
 268                                key, key_length);
 269     buffer_ptr+= key_length;
 270     buffer_ptr[0]=  ' ';
 271     buffer_ptr++;
 272
 273     write_length= (size_t)(buffer_ptr - buffer);
 274     int check_length= snprintf(buffer_ptr, MEMCACHED_DEFAULT_COMMAND_SIZE -(size_t)(buffer_ptr - buffer),
 275                                "%u %llu %lu%s\r\n",
 276                                flags,
 277                                (unsigned long long)expiration, (unsigned long)value_length,
 278                                ptr->flags.no_reply ? " noreply" : "");
 279     if ((size_t)check_length >= MEMCACHED_DEFAULT_COMMAND_SIZE -size_t(buffer_ptr - buffer) || check_length < 0)
 280     {
 281       rc= memcached_set_error(*ptr, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT,
 282                               memcached_literal_param("snprintf(MEMCACHED_DEFAULT_COMMAND_SIZE)"));
 283       memcached_io_reset(instance);
 284
 285       return rc;
 286     }
 287
 288     write_length+= (size_t)check_length;
 289     WATCHPOINT_ASSERT(write_length < MEMCACHED_DEFAULT_COMMAND_SIZE);
 290   }
 291
 292   if (ptr->flags.use_udp && ptr->flags.buffer_requests)
 293   {
 294     size_t cmd_size= write_length + value_length +2;
 295     if (cmd_size > MAX_UDP_DATAGRAM_LENGTH - UDP_DATAGRAM_HEADER_LENGTH)
 296       return memcached_set_error(*ptr, MEMCACHED_WRITE_FAILURE, MEMCACHED_AT);
 297
 298     if (cmd_size + instance->write_buffer_offset > MAX_UDP_DATAGRAM_LENGTH)
 299       memcached_io_write(instance, NULL, 0, true);
 300   }
 301
 302   if (write_length >= MEMCACHED_DEFAULT_COMMAND_SIZE)
 303   {
 304     rc= memcached_set_error(*ptr, MEMCACHED_WRITE_FAILURE, MEMCACHED_AT);
 305   }
 306   else
 307   {
 308     struct libmemcached_io_vector_st vector[]=
 309     {
 310       { write_length, buffer },
 311       { value_length, value },
 312       { 2, "\r\n" }
 313     };
 314
 315     if (ptr->flags.buffer_requests && verb == SET_OP)
 316     {
 317       to_write= false;
 318     }
 319     else
 320     {
 321       to_write= true;
 322     }
 323
 324     /* Send command header */
 325     rc=  memcached_vdo(instance, vector, 3, to_write);
 326     if (rc == MEMCACHED_SUCCESS)
 327     {
 328
 329       if (ptr->flags.no_reply)
 330       {
 331         rc= (to_write == false) ? MEMCACHED_BUFFERED : MEMCACHED_SUCCESS;
 332       }
 333       else if (to_write == false)
 334       {
 335         rc= MEMCACHED_BUFFERED;
 336       }
 337       else
 338       {
 339         rc= memcached_response(instance, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, NULL);
 340
 341         if (rc == MEMCACHED_STORED)
 342           rc= MEMCACHED_SUCCESS;
 343       }
 344     }
 345   }
 346
 347   if (rc == MEMCACHED_WRITE_FAILURE)
 348     memcached_io_reset(instance);
 349
 350   return rc;
 351 }
 352
 353 static inline memcached_return_t memcached_send(memcached_st *ptr,
 354                                                 const char *group_key, size_t group_key_length,
 355                                                 const char *key, size_t key_length,
 356                                                 const char *value, size_t value_length,
 357                                                 time_t expiration,
 358                                                 uint32_t flags,
 359                                                 uint64_t cas,
 360                                                 memcached_storage_action_t verb)
 361 {
 362   memcached_return_t rc;
 363   if (memcached_failed(rc= initialize_query(ptr)))
 364   {
 365     return rc;
 366   }
 367
 368   if (memcached_failed(rc= memcached_validate_key_length(key_length, ptr->flags.binary_protocol)))
 369   {
 370     return rc;
 371   }
 372
 373   if (memcached_failed(memcached_key_test(*ptr, (const char **)&key, &key_length, 1)))
 374   {
 375     return MEMCACHED_BAD_KEY_PROVIDED;
 376   }
 377
 378   uint32_t server_key= memcached_generate_hash_with_redistribution(ptr, group_key, group_key_length);
 379   memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, server_key);
 380
 381   WATCHPOINT_SET(instance->io_wait_count.read= 0);
 382   WATCHPOINT_SET(instance->io_wait_count.write= 0);
 383
 384   if (ptr->flags.binary_protocol)
 385   {
 386     rc= memcached_send_binary(ptr, instance, server_key,
 387                               key, key_length,
 388                               value, value_length, expiration,
 389                               flags, cas, verb);
 390   }
 391   else
 392   {
 393     rc= memcached_send_ascii(ptr, instance,
 394                              key, key_length,
 395                              value, value_length, expiration,
 396                              flags, cas, verb);
 397   }
 398
 399   return rc;
 400 }
 401
 402
 403 memcached_return_t memcached_set(memcached_st *ptr, const char *key, size_t key_length,
 404                                  const char *value, size_t value_length,
 405                                  time_t expiration,
 406                                  uint32_t flags)
 407 {
 408   memcached_return_t rc;
 409   LIBMEMCACHED_MEMCACHED_SET_START();
 410   rc= memcached_send(ptr, key, key_length,
 411                      key, key_length, value, value_length,
 412                      expiration, flags, 0, SET_OP);
 413   LIBMEMCACHED_MEMCACHED_SET_END();
 414   return rc;
 415 }
 416
 417 memcached_return_t memcached_add(memcached_st *ptr,
 418                                  const char *key, size_t key_length,
 419                                  const char *value, size_t value_length,
 420                                  time_t expiration,
 421                                  uint32_t flags)
 422 {
 423   memcached_return_t rc;
 424   LIBMEMCACHED_MEMCACHED_ADD_START();
 425   rc= memcached_send(ptr, key, key_length,
 426                      key, key_length, value, value_length,
 427                      expiration, flags, 0, ADD_OP);
 428   LIBMEMCACHED_MEMCACHED_ADD_END();
 429   return rc;
 430 }
 431
 432 memcached_return_t memcached_replace(memcached_st *ptr,
 433                                      const char *key, size_t key_length,
 434                                      const char *value, size_t value_length,
 435                                      time_t expiration,
 436                                      uint32_t flags)
 437 {
 438   memcached_return_t rc;
 439   LIBMEMCACHED_MEMCACHED_REPLACE_START();
 440   rc= memcached_send(ptr, key, key_length,
 441                      key, key_length, value, value_length,
 442                      expiration, flags, 0, REPLACE_OP);
 443   LIBMEMCACHED_MEMCACHED_REPLACE_END();
 444   return rc;
 445 }
 446
 447 memcached_return_t memcached_prepend(memcached_st *ptr,
 448                                      const char *key, size_t key_length,
 449                                      const char *value, size_t value_length,
 450                                      time_t expiration,
 451                                      uint32_t flags)
 452 {
 453   memcached_return_t rc;
 454   rc= memcached_send(ptr, key, key_length,
 455                      key, key_length, value, value_length,
 456                      expiration, flags, 0, PREPEND_OP);
 457   return rc;
 458 }
 459
 460 memcached_return_t memcached_append(memcached_st *ptr,
 461                                     const char *key, size_t key_length,
 462                                     const char *value, size_t value_length,
 463                                     time_t expiration,
 464                                     uint32_t flags)
 465 {
 466   memcached_return_t rc;
 467   rc= memcached_send(ptr, key, key_length,
 468                      key, key_length, value, value_length,
 469                      expiration, flags, 0, APPEND_OP);
 470   return rc;
 471 }
 472
 473 memcached_return_t memcached_cas(memcached_st *ptr,
 474                                  const char *key, size_t key_length,
 475                                  const char *value, size_t value_length,
 476                                  time_t expiration,
 477                                  uint32_t flags,
 478                                  uint64_t cas)
 479 {
 480   memcached_return_t rc;
 481   rc= memcached_send(ptr, key, key_length,
 482                      key, key_length, value, value_length,
 483                      expiration, flags, cas, CAS_OP);
 484   return rc;
 485 }
 486
 487 memcached_return_t memcached_set_by_key(memcached_st *ptr,
 488                                         const char *group_key,
 489                                         size_t group_key_length,
 490                                         const char *key, size_t key_length,
 491                                         const char *value, size_t value_length,
 492                                         time_t expiration,
 493                                         uint32_t flags)
 494 {
 495   memcached_return_t rc;
 496   LIBMEMCACHED_MEMCACHED_SET_START();
 497   rc= memcached_send(ptr, group_key, group_key_length,
 498                      key, key_length, value, value_length,
 499                      expiration, flags, 0, SET_OP);
 500   LIBMEMCACHED_MEMCACHED_SET_END();
 501   return rc;
 502 }
 503
 504 memcached_return_t memcached_add_by_key(memcached_st *ptr,
 505                                         const char *group_key, size_t group_key_length,
 506                                         const char *key, size_t key_length,
 507                                         const char *value, size_t value_length,
 508                                         time_t expiration,
 509                                         uint32_t flags)
 510 {
 511   memcached_return_t rc;
 512   LIBMEMCACHED_MEMCACHED_ADD_START();
 513   rc= memcached_send(ptr, group_key, group_key_length,
 514                      key, key_length, value, value_length,
 515                      expiration, flags, 0, ADD_OP);
 516   LIBMEMCACHED_MEMCACHED_ADD_END();
 517   return rc;
 518 }
 519
 520 memcached_return_t memcached_replace_by_key(memcached_st *ptr,
 521                                             const char *group_key, size_t group_key_length,
 522                                             const char *key, size_t key_length,
 523                                             const char *value, size_t value_length,
 524                                             time_t expiration,
 525                                             uint32_t flags)
 526 {
 527   memcached_return_t rc;
 528   LIBMEMCACHED_MEMCACHED_REPLACE_START();
 529   rc= memcached_send(ptr, group_key, group_key_length,
 530                      key, key_length, value, value_length,
 531                      expiration, flags, 0, REPLACE_OP);
 532   LIBMEMCACHED_MEMCACHED_REPLACE_END();
 533   return rc;
 534 }
 535
 536 memcached_return_t memcached_prepend_by_key(memcached_st *ptr,
 537                                             const char *group_key, size_t group_key_length,
 538                                             const char *key, size_t key_length,
 539                                             const char *value, size_t value_length,
 540                                             time_t expiration,
 541                                             uint32_t flags)
 542 {
 543   memcached_return_t rc;
 544   rc= memcached_send(ptr, group_key, group_key_length,
 545                      key, key_length, value, value_length,
 546                      expiration, flags, 0, PREPEND_OP);
 547   return rc;
 548 }
 549
 550 memcached_return_t memcached_append_by_key(memcached_st *ptr,
 551                                            const char *group_key, size_t group_key_length,
 552                                            const char *key, size_t key_length,
 553                                            const char *value, size_t value_length,
 554                                            time_t expiration,
 555                                            uint32_t flags)
 556 {
 557   memcached_return_t rc;
 558   rc= memcached_send(ptr, group_key, group_key_length,
 559                      key, key_length, value, value_length,
 560                      expiration, flags, 0, APPEND_OP);
 561   return rc;
 562 }
 563
 564 memcached_return_t memcached_cas_by_key(memcached_st *ptr,
 565                                         const char *group_key, size_t group_key_length,
 566                                         const char *key, size_t key_length,
 567                                         const char *value, size_t value_length,
 568                                         time_t expiration,
 569                                         uint32_t flags,
 570                                         uint64_t cas)
 571 {
 572   memcached_return_t rc;
 573   rc= memcached_send(ptr, group_key, group_key_length,
 574                      key, key_length, value, value_length,
 575                      expiration, flags, cas, CAS_OP);
 576   return rc;
 577 }
 578