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