2 * Copyright (C) 2006-2009 Brian Aker
5 * Use and distribution licensed under the BSD license. See
6 * the COPYING file in the parent directory for full text.
8 * Summary: Storage related functions, aka set, replace,..
12 #include <libmemcached/common.h>
14 enum memcached_storage_action_t
{
24 static inline const char *storage_op_string(memcached_storage_action_t verb
)
50 static inline uint8_t get_com_code(memcached_storage_action_t verb
, bool noreply
)
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 */
60 ret
=PROTOCOL_BINARY_CMD_SETQ
;
63 ret
=PROTOCOL_BINARY_CMD_ADDQ
;
65 case CAS_OP
: /* FALLTHROUGH */
67 ret
=PROTOCOL_BINARY_CMD_REPLACEQ
;
70 ret
=PROTOCOL_BINARY_CMD_APPENDQ
;
73 ret
=PROTOCOL_BINARY_CMD_PREPENDQ
;
76 WATCHPOINT_ASSERT(verb
);
83 ret
=PROTOCOL_BINARY_CMD_SET
;
86 ret
=PROTOCOL_BINARY_CMD_ADD
;
88 case CAS_OP
: /* FALLTHROUGH */
90 ret
=PROTOCOL_BINARY_CMD_REPLACE
;
93 ret
=PROTOCOL_BINARY_CMD_APPEND
;
96 ret
=PROTOCOL_BINARY_CMD_PREPEND
;
99 WATCHPOINT_ASSERT(verb
);
106 static memcached_return_t
memcached_send_binary(memcached_st
*ptr
,
107 memcached_server_write_instance_st server
,
116 memcached_storage_action_t verb
)
119 protocol_binary_request_set request
= {};
120 size_t send_length
= sizeof(request
.bytes
);
122 bool noreply
= server
->root
->flags
.no_reply
;
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! */
132 request
.message
.header
.request
.extlen
= 8;
133 request
.message
.body
.flags
= htonl(flags
);
134 request
.message
.body
.expiration
= htonl((uint32_t)expiration
);
137 request
.message
.header
.request
.bodylen
= htonl((uint32_t) (key_length
+ memcached_array_size(ptr
->_namespace
) + value_length
+
138 request
.message
.header
.request
.extlen
));
141 request
.message
.header
.request
.cas
= memcached_htonll(cas
);
143 flush
= (bool) ((server
->root
->flags
.buffer_requests
&& verb
== SET_OP
) ? 0 : 1);
145 if (server
->root
->flags
.use_udp
&& ! flush
)
147 size_t cmd_size
= send_length
+ key_length
+ value_length
;
149 if (cmd_size
> MAX_UDP_DATAGRAM_LENGTH
- UDP_DATAGRAM_HEADER_LENGTH
)
151 return MEMCACHED_WRITE_FAILURE
;
153 if (cmd_size
+ server
->write_buffer_offset
> MAX_UDP_DATAGRAM_LENGTH
)
155 memcached_io_write(server
, NULL
, 0, true);
159 struct libmemcached_io_vector_st vector
[]=
161 { send_length
, request
.bytes
},
162 { memcached_array_size(ptr
->_namespace
), memcached_array_string(ptr
->_namespace
) },
164 { value_length
, value
}
167 /* write the header */
168 memcached_return_t rc
;
169 if ((rc
= memcached_vdo(server
, vector
, 4, flush
)) != MEMCACHED_SUCCESS
)
171 memcached_io_reset(server
);
172 return (rc
== MEMCACHED_SUCCESS
) ? MEMCACHED_WRITE_FAILURE
: rc
;
175 if (verb
== SET_OP
&& ptr
->number_of_replicas
> 0)
177 request
.message
.header
.request
.opcode
= PROTOCOL_BINARY_CMD_SETQ
;
178 WATCHPOINT_STRING("replicating");
180 for (uint32_t x
= 0; x
< ptr
->number_of_replicas
; x
++)
182 memcached_server_write_instance_st instance
;
185 if (server_key
== memcached_server_count(ptr
))
188 instance
= memcached_server_instance_fetch(ptr
, server_key
);
190 if (memcached_vdo(instance
, vector
, 4, false) != MEMCACHED_SUCCESS
)
192 memcached_io_reset(instance
);
196 memcached_server_response_decrement(instance
);
203 return MEMCACHED_BUFFERED
;
208 return MEMCACHED_SUCCESS
;
211 return memcached_response(server
, NULL
, 0, NULL
);
214 static memcached_return_t
memcached_send_ascii(memcached_st
*ptr
,
215 memcached_server_write_instance_st instance
,
223 memcached_storage_action_t verb
)
227 char buffer
[MEMCACHED_DEFAULT_COMMAND_SIZE
];
228 memcached_return_t rc
;
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)
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
);
249 write_length
= check_length
;
253 char *buffer_ptr
= buffer
;
254 const char *command
= storage_op_string(verb
);
256 /* Copy in the command, no space needed, we handle that in the command function*/
257 memcpy(buffer_ptr
, command
, strlen(command
));
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
));
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
),
265 buffer_ptr
+= key_length
;
269 write_length
= (size_t)(buffer_ptr
- buffer
);
270 int check_length
= snprintf(buffer_ptr
, MEMCACHED_DEFAULT_COMMAND_SIZE
-(size_t)(buffer_ptr
- buffer
),
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)
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
);
284 write_length
+= (size_t)check_length
;
285 WATCHPOINT_ASSERT(write_length
< MEMCACHED_DEFAULT_COMMAND_SIZE
);
288 if (ptr
->flags
.use_udp
&& ptr
->flags
.buffer_requests
)
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
);
294 if (cmd_size
+ instance
->write_buffer_offset
> MAX_UDP_DATAGRAM_LENGTH
)
295 memcached_io_write(instance
, NULL
, 0, true);
298 if (write_length
>= MEMCACHED_DEFAULT_COMMAND_SIZE
)
300 rc
= memcached_set_error(*ptr
, MEMCACHED_WRITE_FAILURE
, MEMCACHED_AT
);
304 struct libmemcached_io_vector_st vector
[]=
306 { write_length
, buffer
},
307 { value_length
, value
},
311 if (ptr
->flags
.buffer_requests
&& verb
== SET_OP
)
320 /* Send command header */
321 rc
= memcached_vdo(instance
, vector
, 3, to_write
);
322 if (rc
== MEMCACHED_SUCCESS
)
325 if (ptr
->flags
.no_reply
)
327 rc
= (to_write
== false) ? MEMCACHED_BUFFERED
: MEMCACHED_SUCCESS
;
329 else if (to_write
== false)
331 rc
= MEMCACHED_BUFFERED
;
335 rc
= memcached_response(instance
, buffer
, MEMCACHED_DEFAULT_COMMAND_SIZE
, NULL
);
337 if (rc
== MEMCACHED_STORED
)
338 rc
= MEMCACHED_SUCCESS
;
343 if (rc
== MEMCACHED_WRITE_FAILURE
)
344 memcached_io_reset(instance
);
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
,
356 memcached_storage_action_t verb
)
358 memcached_return_t rc
;
359 if (memcached_failed(rc
= initialize_query(ptr
)))
364 if (memcached_failed(rc
= memcached_validate_key_length(key_length
, ptr
->flags
.binary_protocol
)))
369 if (memcached_failed(memcached_key_test(*ptr
, (const char **)&key
, &key_length
, 1)))
371 return MEMCACHED_BAD_KEY_PROVIDED
;
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
);
377 WATCHPOINT_SET(instance
->io_wait_count
.read
= 0);
378 WATCHPOINT_SET(instance
->io_wait_count
.write
= 0);
380 if (ptr
->flags
.binary_protocol
)
382 rc
= memcached_send_binary(ptr
, instance
, server_key
,
384 value
, value_length
, expiration
,
389 rc
= memcached_send_ascii(ptr
, instance
,
391 value
, value_length
, expiration
,
399 memcached_return_t
memcached_set(memcached_st
*ptr
, const char *key
, size_t key_length
,
400 const char *value
, size_t value_length
,
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();
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
,
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();
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
,
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();
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
,
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
);
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
,
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
);
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
,
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
);
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
,
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();
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
,
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();
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
,
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();
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
,
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
);
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
,
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
);
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
,
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
);