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
)
130 send_length
-= 8; /* append & prepend does not contain extras! */
134 request
.message
.header
.request
.extlen
= 8;
135 request
.message
.body
.flags
= htonl(flags
);
136 request
.message
.body
.expiration
= htonl((uint32_t)expiration
);
139 request
.message
.header
.request
.bodylen
= htonl((uint32_t) (key_length
+ memcached_array_size(ptr
->_namespace
) + value_length
+
140 request
.message
.header
.request
.extlen
));
144 request
.message
.header
.request
.cas
= memcached_htonll(cas
);
147 flush
= (bool) ((server
->root
->flags
.buffer_requests
&& verb
== SET_OP
) ? 0 : 1);
149 if (server
->root
->flags
.use_udp
&& ! flush
)
151 size_t cmd_size
= send_length
+ key_length
+ value_length
;
153 if (cmd_size
> MAX_UDP_DATAGRAM_LENGTH
- UDP_DATAGRAM_HEADER_LENGTH
)
155 return MEMCACHED_WRITE_FAILURE
;
157 if (cmd_size
+ server
->write_buffer_offset
> MAX_UDP_DATAGRAM_LENGTH
)
159 memcached_io_write(server
, NULL
, 0, true);
163 struct libmemcached_io_vector_st vector
[]=
165 { request
.bytes
, send_length
},
166 { memcached_array_string(ptr
->_namespace
), memcached_array_size(ptr
->_namespace
) },
168 { value
, value_length
}
171 /* write the header */
172 memcached_return_t rc
;
173 if ((rc
= memcached_vdo(server
, vector
, 4, flush
)) != MEMCACHED_SUCCESS
)
175 memcached_io_reset(server
);
176 return (rc
== MEMCACHED_SUCCESS
) ? MEMCACHED_WRITE_FAILURE
: rc
;
179 if (verb
== SET_OP
&& ptr
->number_of_replicas
> 0)
181 request
.message
.header
.request
.opcode
= PROTOCOL_BINARY_CMD_SETQ
;
182 WATCHPOINT_STRING("replicating");
184 for (uint32_t x
= 0; x
< ptr
->number_of_replicas
; x
++)
186 memcached_server_write_instance_st instance
;
189 if (server_key
== memcached_server_count(ptr
))
192 instance
= memcached_server_instance_fetch(ptr
, server_key
);
194 if (memcached_vdo(instance
, vector
, 4, false) != MEMCACHED_SUCCESS
)
196 memcached_io_reset(instance
);
200 memcached_server_response_decrement(instance
);
207 return MEMCACHED_BUFFERED
;
212 return MEMCACHED_SUCCESS
;
215 return memcached_response(server
, NULL
, 0, NULL
);
218 static memcached_return_t
memcached_send_ascii(memcached_st
*ptr
,
219 memcached_server_write_instance_st instance
,
227 memcached_storage_action_t verb
)
231 char buffer
[MEMCACHED_DEFAULT_COMMAND_SIZE
];
232 memcached_return_t rc
;
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)
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
);
253 write_length
= check_length
;
257 char *buffer_ptr
= buffer
;
258 const char *command
= storage_op_string(verb
);
260 /* Copy in the command, no space needed, we handle that in the command function*/
261 memcpy(buffer_ptr
, command
, strlen(command
));
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
));
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
),
269 buffer_ptr
+= key_length
;
273 write_length
= (size_t)(buffer_ptr
- buffer
);
274 int check_length
= snprintf(buffer_ptr
, MEMCACHED_DEFAULT_COMMAND_SIZE
-(size_t)(buffer_ptr
- buffer
),
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)
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
);
288 write_length
+= (size_t)check_length
;
289 WATCHPOINT_ASSERT(write_length
< MEMCACHED_DEFAULT_COMMAND_SIZE
);
292 if (ptr
->flags
.use_udp
&& ptr
->flags
.buffer_requests
)
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
);
298 if (cmd_size
+ instance
->write_buffer_offset
> MAX_UDP_DATAGRAM_LENGTH
)
299 memcached_io_write(instance
, NULL
, 0, true);
302 if (write_length
>= MEMCACHED_DEFAULT_COMMAND_SIZE
)
304 rc
= memcached_set_error(*ptr
, MEMCACHED_WRITE_FAILURE
, MEMCACHED_AT
);
308 struct libmemcached_io_vector_st vector
[]=
310 { buffer
, write_length
},
311 { value
, value_length
},
312 { memcached_literal_param("\r\n") }
315 if (ptr
->flags
.buffer_requests
&& verb
== SET_OP
)
324 /* Send command header */
325 rc
= memcached_vdo(instance
, vector
, 3, to_write
);
326 if (rc
== MEMCACHED_SUCCESS
)
329 if (ptr
->flags
.no_reply
)
331 rc
= (to_write
== false) ? MEMCACHED_BUFFERED
: MEMCACHED_SUCCESS
;
333 else if (to_write
== false)
335 rc
= MEMCACHED_BUFFERED
;
339 rc
= memcached_response(instance
, buffer
, MEMCACHED_DEFAULT_COMMAND_SIZE
, NULL
);
341 if (rc
== MEMCACHED_STORED
)
342 rc
= MEMCACHED_SUCCESS
;
347 if (rc
== MEMCACHED_WRITE_FAILURE
)
348 memcached_io_reset(instance
);
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
,
360 memcached_storage_action_t verb
)
362 memcached_return_t rc
;
363 if (memcached_failed(rc
= initialize_query(ptr
)))
368 if (memcached_failed(rc
= memcached_validate_key_length(key_length
, ptr
->flags
.binary_protocol
)))
373 if (memcached_failed(memcached_key_test(*ptr
, (const char **)&key
, &key_length
, 1)))
375 return MEMCACHED_BAD_KEY_PROVIDED
;
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
);
381 WATCHPOINT_SET(instance
->io_wait_count
.read
= 0);
382 WATCHPOINT_SET(instance
->io_wait_count
.write
= 0);
384 if (ptr
->flags
.binary_protocol
)
386 rc
= memcached_send_binary(ptr
, instance
, server_key
,
388 value
, value_length
, expiration
,
393 rc
= memcached_send_ascii(ptr
, instance
,
395 value
, value_length
, expiration
,
403 memcached_return_t
memcached_set(memcached_st
*ptr
, const char *key
, size_t key_length
,
404 const char *value
, size_t value_length
,
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();
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
,
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();
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
,
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();
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
,
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
);
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
,
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
);
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
,
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
);
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
,
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();
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
,
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();
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
,
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();
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
,
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
);
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
,
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
);
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
,
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
);