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
)
41 return "tosserror"; /* This is impossible, fixes issue for compiler warning in VisualStudio */
47 static memcached_return_t
memcached_send_binary(memcached_st
*ptr
,
48 memcached_server_write_instance_st server
,
57 memcached_storage_action_t verb
);
59 static inline memcached_return_t
memcached_send(memcached_st
*ptr
,
60 const char *group_key
, size_t group_key_length
,
61 const char *key
, size_t key_length
,
62 const char *value
, size_t value_length
,
66 memcached_storage_action_t verb
)
70 char buffer
[MEMCACHED_DEFAULT_COMMAND_SIZE
];
72 WATCHPOINT_ASSERT(!(value
== NULL
&& value_length
> 0));
74 memcached_return_t rc
;
75 if (memcached_failed(rc
= initialize_query(ptr
)))
80 if (memcached_failed(rc
= memcached_validate_key_length(key_length
, ptr
->flags
.binary_protocol
)))
83 if (ptr
->flags
.verify_key
&& (memcached_key_test((const char **)&key
, &key_length
, 1) == MEMCACHED_BAD_KEY_PROVIDED
))
84 return MEMCACHED_BAD_KEY_PROVIDED
;
86 uint32_t server_key
= memcached_generate_hash_with_redistribution(ptr
, group_key
, group_key_length
);
87 memcached_server_write_instance_st instance
= memcached_server_instance_fetch(ptr
, server_key
);
89 WATCHPOINT_SET(instance
->io_wait_count
.read
= 0);
90 WATCHPOINT_SET(instance
->io_wait_count
.write
= 0);
92 if (ptr
->flags
.binary_protocol
)
94 rc
= memcached_send_binary(ptr
, instance
, server_key
,
96 value
, value_length
, expiration
,
98 WATCHPOINT_IF_LABELED_NUMBER(instance
->io_wait_count
.read
> 2, "read IO_WAIT", instance
->io_wait_count
.read
);
99 WATCHPOINT_IF_LABELED_NUMBER(instance
->io_wait_count
.write
> 2, "write_IO_WAIT", instance
->io_wait_count
.write
);
107 check_length
= snprintf(buffer
, MEMCACHED_DEFAULT_COMMAND_SIZE
,
108 "%s %.*s%.*s %u %llu %lu %llu%s\r\n",
109 storage_op_string(verb
),
110 memcached_print_array(ptr
->prefix_key
),
111 (int)key_length
, key
, flags
,
112 (unsigned long long)expiration
, (unsigned long)value_length
,
113 (unsigned long long)cas
,
114 (ptr
->flags
.no_reply
) ? " noreply" : "");
115 if (check_length
>= MEMCACHED_DEFAULT_COMMAND_SIZE
|| check_length
< 0)
117 rc
= MEMCACHED_WRITE_FAILURE
;
118 memcached_io_reset(instance
);
122 write_length
= check_length
;
126 char *buffer_ptr
= buffer
;
127 const char *command
= storage_op_string(verb
);
129 /* Copy in the command, no space needed, we handle that in the command function*/
130 memcpy(buffer_ptr
, command
, strlen(command
));
132 /* Copy in the key prefix, switch to the buffer_ptr */
133 buffer_ptr
= (char *)memcpy((char *)(buffer_ptr
+ strlen(command
)), (char *)memcached_array_string(ptr
->prefix_key
), memcached_array_size(ptr
->prefix_key
));
135 /* Copy in the key, adjust point if a key prefix was used. */
136 buffer_ptr
= (char *)memcpy(buffer_ptr
+ memcached_array_size(ptr
->prefix_key
),
138 buffer_ptr
+= key_length
;
142 write_length
= (size_t)(buffer_ptr
- buffer
);
143 int check_length
= snprintf(buffer_ptr
, MEMCACHED_DEFAULT_COMMAND_SIZE
-(size_t)(buffer_ptr
- buffer
),
146 (unsigned long long)expiration
, (unsigned long)value_length
,
147 ptr
->flags
.no_reply
? " noreply" : "");
148 if ((size_t)check_length
>= MEMCACHED_DEFAULT_COMMAND_SIZE
-size_t(buffer_ptr
- buffer
) || check_length
< 0)
150 rc
= MEMCACHED_WRITE_FAILURE
;
151 memcached_io_reset(instance
);
156 write_length
+= (size_t)check_length
;
157 WATCHPOINT_ASSERT(write_length
< MEMCACHED_DEFAULT_COMMAND_SIZE
);
160 if (ptr
->flags
.use_udp
&& ptr
->flags
.buffer_requests
)
162 size_t cmd_size
= write_length
+ value_length
+2;
163 if (cmd_size
> MAX_UDP_DATAGRAM_LENGTH
- UDP_DATAGRAM_HEADER_LENGTH
)
164 return memcached_set_error(*ptr
, MEMCACHED_WRITE_FAILURE
, MEMCACHED_AT
);
166 if (cmd_size
+ instance
->write_buffer_offset
> MAX_UDP_DATAGRAM_LENGTH
)
167 memcached_io_write(instance
, NULL
, 0, true);
170 if (write_length
>= MEMCACHED_DEFAULT_COMMAND_SIZE
)
172 rc
= memcached_set_error(*ptr
, MEMCACHED_WRITE_FAILURE
, MEMCACHED_AT
);
176 struct libmemcached_io_vector_st vector
[]=
178 { write_length
, buffer
},
179 { value_length
, value
},
183 if (ptr
->flags
.buffer_requests
&& verb
== SET_OP
)
192 /* Send command header */
193 rc
= memcached_vdo(instance
, vector
, 3, to_write
);
194 if (rc
== MEMCACHED_SUCCESS
)
197 if (ptr
->flags
.no_reply
)
199 rc
= (to_write
== false) ? MEMCACHED_BUFFERED
: MEMCACHED_SUCCESS
;
201 else if (to_write
== false)
203 rc
= MEMCACHED_BUFFERED
;
207 rc
= memcached_response(instance
, buffer
, MEMCACHED_DEFAULT_COMMAND_SIZE
, NULL
);
209 if (rc
== MEMCACHED_STORED
)
210 rc
= MEMCACHED_SUCCESS
;
215 if (rc
== MEMCACHED_WRITE_FAILURE
)
216 memcached_io_reset(instance
);
219 WATCHPOINT_IF_LABELED_NUMBER(instance
->io_wait_count
.read
> 2, "read IO_WAIT", instance
->io_wait_count
.read
);
220 WATCHPOINT_IF_LABELED_NUMBER(instance
->io_wait_count
.write
> 2, "write_IO_WAIT", instance
->io_wait_count
.write
);
226 memcached_return_t
memcached_set(memcached_st
*ptr
, const char *key
, size_t key_length
,
227 const char *value
, size_t value_length
,
231 memcached_return_t rc
;
232 LIBMEMCACHED_MEMCACHED_SET_START();
233 rc
= memcached_send(ptr
, key
, key_length
,
234 key
, key_length
, value
, value_length
,
235 expiration
, flags
, 0, SET_OP
);
236 LIBMEMCACHED_MEMCACHED_SET_END();
240 memcached_return_t
memcached_add(memcached_st
*ptr
,
241 const char *key
, size_t key_length
,
242 const char *value
, size_t value_length
,
246 memcached_return_t rc
;
247 LIBMEMCACHED_MEMCACHED_ADD_START();
248 rc
= memcached_send(ptr
, key
, key_length
,
249 key
, key_length
, value
, value_length
,
250 expiration
, flags
, 0, ADD_OP
);
251 LIBMEMCACHED_MEMCACHED_ADD_END();
255 memcached_return_t
memcached_replace(memcached_st
*ptr
,
256 const char *key
, size_t key_length
,
257 const char *value
, size_t value_length
,
261 memcached_return_t rc
;
262 LIBMEMCACHED_MEMCACHED_REPLACE_START();
263 rc
= memcached_send(ptr
, key
, key_length
,
264 key
, key_length
, value
, value_length
,
265 expiration
, flags
, 0, REPLACE_OP
);
266 LIBMEMCACHED_MEMCACHED_REPLACE_END();
270 memcached_return_t
memcached_prepend(memcached_st
*ptr
,
271 const char *key
, size_t key_length
,
272 const char *value
, size_t value_length
,
276 memcached_return_t rc
;
277 rc
= memcached_send(ptr
, key
, key_length
,
278 key
, key_length
, value
, value_length
,
279 expiration
, flags
, 0, PREPEND_OP
);
283 memcached_return_t
memcached_append(memcached_st
*ptr
,
284 const char *key
, size_t key_length
,
285 const char *value
, size_t value_length
,
289 memcached_return_t rc
;
290 rc
= memcached_send(ptr
, key
, key_length
,
291 key
, key_length
, value
, value_length
,
292 expiration
, flags
, 0, APPEND_OP
);
296 memcached_return_t
memcached_cas(memcached_st
*ptr
,
297 const char *key
, size_t key_length
,
298 const char *value
, size_t value_length
,
303 memcached_return_t rc
;
304 rc
= memcached_send(ptr
, key
, key_length
,
305 key
, key_length
, value
, value_length
,
306 expiration
, flags
, cas
, CAS_OP
);
310 memcached_return_t
memcached_set_by_key(memcached_st
*ptr
,
311 const char *group_key
,
312 size_t group_key_length
,
313 const char *key
, size_t key_length
,
314 const char *value
, size_t value_length
,
318 memcached_return_t rc
;
319 LIBMEMCACHED_MEMCACHED_SET_START();
320 rc
= memcached_send(ptr
, group_key
, group_key_length
,
321 key
, key_length
, value
, value_length
,
322 expiration
, flags
, 0, SET_OP
);
323 LIBMEMCACHED_MEMCACHED_SET_END();
327 memcached_return_t
memcached_add_by_key(memcached_st
*ptr
,
328 const char *group_key
, size_t group_key_length
,
329 const char *key
, size_t key_length
,
330 const char *value
, size_t value_length
,
334 memcached_return_t rc
;
335 LIBMEMCACHED_MEMCACHED_ADD_START();
336 rc
= memcached_send(ptr
, group_key
, group_key_length
,
337 key
, key_length
, value
, value_length
,
338 expiration
, flags
, 0, ADD_OP
);
339 LIBMEMCACHED_MEMCACHED_ADD_END();
343 memcached_return_t
memcached_replace_by_key(memcached_st
*ptr
,
344 const char *group_key
, size_t group_key_length
,
345 const char *key
, size_t key_length
,
346 const char *value
, size_t value_length
,
350 memcached_return_t rc
;
351 LIBMEMCACHED_MEMCACHED_REPLACE_START();
352 rc
= memcached_send(ptr
, group_key
, group_key_length
,
353 key
, key_length
, value
, value_length
,
354 expiration
, flags
, 0, REPLACE_OP
);
355 LIBMEMCACHED_MEMCACHED_REPLACE_END();
359 memcached_return_t
memcached_prepend_by_key(memcached_st
*ptr
,
360 const char *group_key
, size_t group_key_length
,
361 const char *key
, size_t key_length
,
362 const char *value
, size_t value_length
,
366 memcached_return_t rc
;
367 rc
= memcached_send(ptr
, group_key
, group_key_length
,
368 key
, key_length
, value
, value_length
,
369 expiration
, flags
, 0, PREPEND_OP
);
373 memcached_return_t
memcached_append_by_key(memcached_st
*ptr
,
374 const char *group_key
, size_t group_key_length
,
375 const char *key
, size_t key_length
,
376 const char *value
, size_t value_length
,
380 memcached_return_t rc
;
381 rc
= memcached_send(ptr
, group_key
, group_key_length
,
382 key
, key_length
, value
, value_length
,
383 expiration
, flags
, 0, APPEND_OP
);
387 memcached_return_t
memcached_cas_by_key(memcached_st
*ptr
,
388 const char *group_key
, size_t group_key_length
,
389 const char *key
, size_t key_length
,
390 const char *value
, size_t value_length
,
395 memcached_return_t rc
;
396 rc
= memcached_send(ptr
, group_key
, group_key_length
,
397 key
, key_length
, value
, value_length
,
398 expiration
, flags
, cas
, CAS_OP
);
402 static inline uint8_t get_com_code(memcached_storage_action_t verb
, bool noreply
)
404 /* 0 isn't a value we want, but GCC 4.2 seems to think ret can otherwise
405 * be used uninitialized in this function. FAIL */
412 ret
=PROTOCOL_BINARY_CMD_SETQ
;
415 ret
=PROTOCOL_BINARY_CMD_ADDQ
;
417 case CAS_OP
: /* FALLTHROUGH */
419 ret
=PROTOCOL_BINARY_CMD_REPLACEQ
;
422 ret
=PROTOCOL_BINARY_CMD_APPENDQ
;
425 ret
=PROTOCOL_BINARY_CMD_PREPENDQ
;
428 WATCHPOINT_ASSERT(verb
);
435 ret
=PROTOCOL_BINARY_CMD_SET
;
438 ret
=PROTOCOL_BINARY_CMD_ADD
;
440 case CAS_OP
: /* FALLTHROUGH */
442 ret
=PROTOCOL_BINARY_CMD_REPLACE
;
445 ret
=PROTOCOL_BINARY_CMD_APPEND
;
448 ret
=PROTOCOL_BINARY_CMD_PREPEND
;
451 WATCHPOINT_ASSERT(verb
);
460 static memcached_return_t
memcached_send_binary(memcached_st
*ptr
,
461 memcached_server_write_instance_st server
,
470 memcached_storage_action_t verb
)
473 protocol_binary_request_set request
= {};
474 size_t send_length
= sizeof(request
.bytes
);
476 bool noreply
= server
->root
->flags
.no_reply
;
478 request
.message
.header
.request
.magic
= PROTOCOL_BINARY_REQ
;
479 request
.message
.header
.request
.opcode
= get_com_code(verb
, noreply
);
480 request
.message
.header
.request
.keylen
= htons((uint16_t)(key_length
+ memcached_array_size(ptr
->prefix_key
)));
481 request
.message
.header
.request
.datatype
= PROTOCOL_BINARY_RAW_BYTES
;
482 if (verb
== APPEND_OP
|| verb
== PREPEND_OP
)
483 send_length
-= 8; /* append & prepend does not contain extras! */
486 request
.message
.header
.request
.extlen
= 8;
487 request
.message
.body
.flags
= htonl(flags
);
488 request
.message
.body
.expiration
= htonl((uint32_t)expiration
);
491 request
.message
.header
.request
.bodylen
= htonl((uint32_t) (key_length
+ memcached_array_size(ptr
->prefix_key
) + value_length
+
492 request
.message
.header
.request
.extlen
));
495 request
.message
.header
.request
.cas
= memcached_htonll(cas
);
497 flush
= (bool) ((server
->root
->flags
.buffer_requests
&& verb
== SET_OP
) ? 0 : 1);
499 if (server
->root
->flags
.use_udp
&& ! flush
)
501 size_t cmd_size
= send_length
+ key_length
+ value_length
;
503 if (cmd_size
> MAX_UDP_DATAGRAM_LENGTH
- UDP_DATAGRAM_HEADER_LENGTH
)
505 return MEMCACHED_WRITE_FAILURE
;
507 if (cmd_size
+ server
->write_buffer_offset
> MAX_UDP_DATAGRAM_LENGTH
)
509 memcached_io_write(server
, NULL
, 0, true);
513 struct libmemcached_io_vector_st vector
[]=
515 { send_length
, request
.bytes
},
516 { memcached_array_size(ptr
->prefix_key
), memcached_array_string(ptr
->prefix_key
) },
518 { value_length
, value
}
521 /* write the header */
522 memcached_return_t rc
;
523 if ((rc
= memcached_vdo(server
, vector
, 4, flush
)) != MEMCACHED_SUCCESS
)
525 memcached_io_reset(server
);
526 return (rc
== MEMCACHED_SUCCESS
) ? MEMCACHED_WRITE_FAILURE
: rc
;
529 if (verb
== SET_OP
&& ptr
->number_of_replicas
> 0)
531 request
.message
.header
.request
.opcode
= PROTOCOL_BINARY_CMD_SETQ
;
532 WATCHPOINT_STRING("replicating");
534 for (uint32_t x
= 0; x
< ptr
->number_of_replicas
; x
++)
536 memcached_server_write_instance_st instance
;
539 if (server_key
== memcached_server_count(ptr
))
542 instance
= memcached_server_instance_fetch(ptr
, server_key
);
544 if (memcached_vdo(instance
, vector
, 4, false) != MEMCACHED_SUCCESS
)
546 memcached_io_reset(instance
);
550 memcached_server_response_decrement(instance
);
557 return MEMCACHED_BUFFERED
;
562 return MEMCACHED_SUCCESS
;
565 return memcached_response(server
, NULL
, 0, NULL
);