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,..
21 } 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 *master_key
, size_t master_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 memcached_return_t rc
;
71 char buffer
[MEMCACHED_DEFAULT_COMMAND_SIZE
];
73 memcached_server_write_instance_st instance
;
75 WATCHPOINT_ASSERT(!(value
== NULL
&& value_length
> 0));
77 rc
= memcached_validate_key_length(key_length
, ptr
->flags
.binary_protocol
);
78 unlikely (rc
!= MEMCACHED_SUCCESS
)
81 unlikely (memcached_server_count(ptr
) == 0)
82 return MEMCACHED_NO_SERVERS
;
84 if (ptr
->flags
.verify_key
&& (memcached_key_test((const char **)&key
, &key_length
, 1) == MEMCACHED_BAD_KEY_PROVIDED
))
85 return MEMCACHED_BAD_KEY_PROVIDED
;
87 server_key
= memcached_generate_hash_with_redistribution(ptr
, master_key
, master_key_length
);
88 instance
= memcached_server_instance_fetch(ptr
, server_key
);
90 WATCHPOINT_SET(instance
->io_wait_count
.read
= 0);
91 WATCHPOINT_SET(instance
->io_wait_count
.write
= 0);
93 if (ptr
->flags
.binary_protocol
)
95 rc
= memcached_send_binary(ptr
, instance
, server_key
,
97 value
, value_length
, expiration
,
99 WATCHPOINT_IF_LABELED_NUMBER(instance
->io_wait_count
.read
> 2, "read IO_WAIT", instance
->io_wait_count
.read
);
100 WATCHPOINT_IF_LABELED_NUMBER(instance
->io_wait_count
.write
> 2, "write_IO_WAIT", instance
->io_wait_count
.write
);
108 check_length
= snprintf(buffer
, MEMCACHED_DEFAULT_COMMAND_SIZE
,
109 "%s %.*s%.*s %u %llu %lu %llu%s\r\n",
110 storage_op_string(verb
),
111 (int)ptr
->prefix_key_length
,
113 (int)key_length
, key
, flags
,
114 (unsigned long long)expiration
, (unsigned long)value_length
,
115 (unsigned long long)cas
,
116 (ptr
->flags
.no_reply
) ? " noreply" : "");
117 if (check_length
>= MEMCACHED_DEFAULT_COMMAND_SIZE
|| check_length
< 0)
119 rc
= MEMCACHED_WRITE_FAILURE
;
120 memcached_io_reset(instance
);
124 write_length
= check_length
;
128 char *buffer_ptr
= buffer
;
129 const char *command
= storage_op_string(verb
);
131 /* Copy in the command, no space needed, we handle that in the command function*/
132 memcpy(buffer_ptr
, command
, strlen(command
));
134 /* Copy in the key prefix, switch to the buffer_ptr */
135 buffer_ptr
= memcpy((buffer_ptr
+ strlen(command
)), ptr
->prefix_key
, ptr
->prefix_key_length
);
137 /* Copy in the key, adjust point if a key prefix was used. */
138 buffer_ptr
= memcpy(buffer_ptr
+ (ptr
->prefix_key_length
? ptr
->prefix_key_length
: 0),
140 buffer_ptr
+= key_length
;
144 write_length
= (size_t)(buffer_ptr
- buffer
);
146 check_length
= snprintf(buffer_ptr
, MEMCACHED_DEFAULT_COMMAND_SIZE
-(size_t)(buffer_ptr
- buffer
),
149 (unsigned long long)expiration
, (unsigned long)value_length
,
150 ptr
->flags
.no_reply
? " noreply" : "");
151 if ((size_t)check_length
>= MEMCACHED_DEFAULT_COMMAND_SIZE
-(size_t)(buffer_ptr
- buffer
) || check_length
< 0)
153 rc
= MEMCACHED_WRITE_FAILURE
;
154 memcached_io_reset(instance
);
159 write_length
+= (size_t)check_length
;
160 WATCHPOINT_ASSERT(write_length
< MEMCACHED_DEFAULT_COMMAND_SIZE
);
163 if (ptr
->flags
.use_udp
&& ptr
->flags
.buffer_requests
)
165 size_t cmd_size
= write_length
+ value_length
+ 2;
166 if (cmd_size
> MAX_UDP_DATAGRAM_LENGTH
- UDP_DATAGRAM_HEADER_LENGTH
)
167 return MEMCACHED_WRITE_FAILURE
;
168 if (cmd_size
+ instance
->write_buffer_offset
> MAX_UDP_DATAGRAM_LENGTH
)
169 memcached_io_write(instance
, NULL
, 0, true);
172 if (write_length
>= MEMCACHED_DEFAULT_COMMAND_SIZE
)
174 rc
= MEMCACHED_WRITE_FAILURE
;
178 struct libmemcached_io_vector_st vector
[]=
180 { .length
= write_length
, .buffer
= buffer
},
181 { .length
= value_length
, .buffer
= value
},
182 { .length
= 2, .buffer
= "\r\n" }
185 if (ptr
->flags
.buffer_requests
&& verb
== SET_OP
)
194 /* Send command header */
195 rc
= memcached_vdo(instance
, vector
, 3, to_write
);
196 if (rc
== MEMCACHED_SUCCESS
)
199 if (ptr
->flags
.no_reply
)
201 rc
= (to_write
== false) ? MEMCACHED_BUFFERED
: MEMCACHED_SUCCESS
;
203 else if (to_write
== false)
205 rc
= MEMCACHED_BUFFERED
;
209 rc
= memcached_response(instance
, buffer
, MEMCACHED_DEFAULT_COMMAND_SIZE
, NULL
);
211 if (rc
== MEMCACHED_STORED
)
212 rc
= MEMCACHED_SUCCESS
;
217 if (rc
== MEMCACHED_WRITE_FAILURE
)
218 memcached_io_reset(instance
);
221 WATCHPOINT_IF_LABELED_NUMBER(instance
->io_wait_count
.read
> 2, "read IO_WAIT", instance
->io_wait_count
.read
);
222 WATCHPOINT_IF_LABELED_NUMBER(instance
->io_wait_count
.write
> 2, "write_IO_WAIT", instance
->io_wait_count
.write
);
228 memcached_return_t
memcached_set(memcached_st
*ptr
, const char *key
, size_t key_length
,
229 const char *value
, size_t value_length
,
233 memcached_return_t rc
;
234 LIBMEMCACHED_MEMCACHED_SET_START();
235 rc
= memcached_send(ptr
, key
, key_length
,
236 key
, key_length
, value
, value_length
,
237 expiration
, flags
, 0, SET_OP
);
238 LIBMEMCACHED_MEMCACHED_SET_END();
242 memcached_return_t
memcached_add(memcached_st
*ptr
,
243 const char *key
, size_t key_length
,
244 const char *value
, size_t value_length
,
248 memcached_return_t rc
;
249 LIBMEMCACHED_MEMCACHED_ADD_START();
250 rc
= memcached_send(ptr
, key
, key_length
,
251 key
, key_length
, value
, value_length
,
252 expiration
, flags
, 0, ADD_OP
);
253 LIBMEMCACHED_MEMCACHED_ADD_END();
257 memcached_return_t
memcached_replace(memcached_st
*ptr
,
258 const char *key
, size_t key_length
,
259 const char *value
, size_t value_length
,
263 memcached_return_t rc
;
264 LIBMEMCACHED_MEMCACHED_REPLACE_START();
265 rc
= memcached_send(ptr
, key
, key_length
,
266 key
, key_length
, value
, value_length
,
267 expiration
, flags
, 0, REPLACE_OP
);
268 LIBMEMCACHED_MEMCACHED_REPLACE_END();
272 memcached_return_t
memcached_prepend(memcached_st
*ptr
,
273 const char *key
, size_t key_length
,
274 const char *value
, size_t value_length
,
278 memcached_return_t rc
;
279 rc
= memcached_send(ptr
, key
, key_length
,
280 key
, key_length
, value
, value_length
,
281 expiration
, flags
, 0, PREPEND_OP
);
285 memcached_return_t
memcached_append(memcached_st
*ptr
,
286 const char *key
, size_t key_length
,
287 const char *value
, size_t value_length
,
291 memcached_return_t rc
;
292 rc
= memcached_send(ptr
, key
, key_length
,
293 key
, key_length
, value
, value_length
,
294 expiration
, flags
, 0, APPEND_OP
);
298 memcached_return_t
memcached_cas(memcached_st
*ptr
,
299 const char *key
, size_t key_length
,
300 const char *value
, size_t value_length
,
305 memcached_return_t rc
;
306 rc
= memcached_send(ptr
, key
, key_length
,
307 key
, key_length
, value
, value_length
,
308 expiration
, flags
, cas
, CAS_OP
);
312 memcached_return_t
memcached_set_by_key(memcached_st
*ptr
,
313 const char *master_key
__attribute__((unused
)),
314 size_t master_key_length
__attribute__((unused
)),
315 const char *key
, size_t key_length
,
316 const char *value
, size_t value_length
,
320 memcached_return_t rc
;
321 LIBMEMCACHED_MEMCACHED_SET_START();
322 rc
= memcached_send(ptr
, master_key
, master_key_length
,
323 key
, key_length
, value
, value_length
,
324 expiration
, flags
, 0, SET_OP
);
325 LIBMEMCACHED_MEMCACHED_SET_END();
329 memcached_return_t
memcached_add_by_key(memcached_st
*ptr
,
330 const char *master_key
, size_t master_key_length
,
331 const char *key
, size_t key_length
,
332 const char *value
, size_t value_length
,
336 memcached_return_t rc
;
337 LIBMEMCACHED_MEMCACHED_ADD_START();
338 rc
= memcached_send(ptr
, master_key
, master_key_length
,
339 key
, key_length
, value
, value_length
,
340 expiration
, flags
, 0, ADD_OP
);
341 LIBMEMCACHED_MEMCACHED_ADD_END();
345 memcached_return_t
memcached_replace_by_key(memcached_st
*ptr
,
346 const char *master_key
, size_t master_key_length
,
347 const char *key
, size_t key_length
,
348 const char *value
, size_t value_length
,
352 memcached_return_t rc
;
353 LIBMEMCACHED_MEMCACHED_REPLACE_START();
354 rc
= memcached_send(ptr
, master_key
, master_key_length
,
355 key
, key_length
, value
, value_length
,
356 expiration
, flags
, 0, REPLACE_OP
);
357 LIBMEMCACHED_MEMCACHED_REPLACE_END();
361 memcached_return_t
memcached_prepend_by_key(memcached_st
*ptr
,
362 const char *master_key
, size_t master_key_length
,
363 const char *key
, size_t key_length
,
364 const char *value
, size_t value_length
,
368 memcached_return_t rc
;
369 rc
= memcached_send(ptr
, master_key
, master_key_length
,
370 key
, key_length
, value
, value_length
,
371 expiration
, flags
, 0, PREPEND_OP
);
375 memcached_return_t
memcached_append_by_key(memcached_st
*ptr
,
376 const char *master_key
, size_t master_key_length
,
377 const char *key
, size_t key_length
,
378 const char *value
, size_t value_length
,
382 memcached_return_t rc
;
383 rc
= memcached_send(ptr
, master_key
, master_key_length
,
384 key
, key_length
, value
, value_length
,
385 expiration
, flags
, 0, APPEND_OP
);
389 memcached_return_t
memcached_cas_by_key(memcached_st
*ptr
,
390 const char *master_key
, size_t master_key_length
,
391 const char *key
, size_t key_length
,
392 const char *value
, size_t value_length
,
397 memcached_return_t rc
;
398 rc
= memcached_send(ptr
, master_key
, master_key_length
,
399 key
, key_length
, value
, value_length
,
400 expiration
, flags
, cas
, CAS_OP
);
404 static inline uint8_t get_com_code(memcached_storage_action_t verb
, bool noreply
)
406 /* 0 isn't a value we want, but GCC 4.2 seems to think ret can otherwise
407 * be used uninitialized in this function. FAIL */
414 ret
=PROTOCOL_BINARY_CMD_SETQ
;
417 ret
=PROTOCOL_BINARY_CMD_ADDQ
;
419 case CAS_OP
: /* FALLTHROUGH */
421 ret
=PROTOCOL_BINARY_CMD_REPLACEQ
;
424 ret
=PROTOCOL_BINARY_CMD_APPENDQ
;
427 ret
=PROTOCOL_BINARY_CMD_PREPENDQ
;
430 WATCHPOINT_ASSERT(verb
);
437 ret
=PROTOCOL_BINARY_CMD_SET
;
440 ret
=PROTOCOL_BINARY_CMD_ADD
;
442 case CAS_OP
: /* FALLTHROUGH */
444 ret
=PROTOCOL_BINARY_CMD_REPLACE
;
447 ret
=PROTOCOL_BINARY_CMD_APPEND
;
450 ret
=PROTOCOL_BINARY_CMD_PREPEND
;
453 WATCHPOINT_ASSERT(verb
);
462 static memcached_return_t
memcached_send_binary(memcached_st
*ptr
,
463 memcached_server_write_instance_st server
,
472 memcached_storage_action_t verb
)
475 protocol_binary_request_set request
= {.bytes
= {0}};
476 size_t send_length
= sizeof(request
.bytes
);
478 bool noreply
= server
->root
->flags
.no_reply
;
480 request
.message
.header
.request
.magic
= PROTOCOL_BINARY_REQ
;
481 request
.message
.header
.request
.opcode
= get_com_code(verb
, noreply
);
482 request
.message
.header
.request
.keylen
= htons((uint16_t)(key_length
+ ptr
->prefix_key_length
));
483 request
.message
.header
.request
.datatype
= PROTOCOL_BINARY_RAW_BYTES
;
484 if (verb
== APPEND_OP
|| verb
== PREPEND_OP
)
485 send_length
-= 8; /* append & prepend does not contain extras! */
488 request
.message
.header
.request
.extlen
= 8;
489 request
.message
.body
.flags
= htonl(flags
);
490 request
.message
.body
.expiration
= htonl((uint32_t)expiration
);
493 request
.message
.header
.request
.bodylen
= htonl((uint32_t) (key_length
+ ptr
->prefix_key_length
+ value_length
+
494 request
.message
.header
.request
.extlen
));
497 request
.message
.header
.request
.cas
= htonll(cas
);
499 flush
= (bool) ((server
->root
->flags
.buffer_requests
&& verb
== SET_OP
) ? 0 : 1);
501 if (server
->root
->flags
.use_udp
&& ! flush
)
503 size_t cmd_size
= send_length
+ key_length
+ value_length
;
505 if (cmd_size
> MAX_UDP_DATAGRAM_LENGTH
- UDP_DATAGRAM_HEADER_LENGTH
)
507 return MEMCACHED_WRITE_FAILURE
;
509 if (cmd_size
+ server
->write_buffer_offset
> MAX_UDP_DATAGRAM_LENGTH
)
511 memcached_io_write(server
, NULL
, 0, true);
515 struct libmemcached_io_vector_st vector
[]=
517 { .length
= send_length
, .buffer
= request
.bytes
},
518 { .length
= ptr
->prefix_key_length
, .buffer
= ptr
->prefix_key
},
519 { .length
= key_length
, .buffer
= key
},
520 { .length
= value_length
, .buffer
= value
}
523 /* write the header */
524 memcached_return_t rc
;
525 if ((rc
= memcached_vdo(server
, vector
, 4, flush
)) != MEMCACHED_SUCCESS
)
527 memcached_io_reset(server
);
528 return (rc
== MEMCACHED_SUCCESS
) ? MEMCACHED_WRITE_FAILURE
: rc
;
531 if (verb
== SET_OP
&& ptr
->number_of_replicas
> 0)
533 request
.message
.header
.request
.opcode
= PROTOCOL_BINARY_CMD_SETQ
;
534 WATCHPOINT_STRING("replicating");
536 for (uint32_t x
= 0; x
< ptr
->number_of_replicas
; x
++)
538 memcached_server_write_instance_st instance
;
541 if (server_key
== memcached_server_count(ptr
))
544 instance
= memcached_server_instance_fetch(ptr
, server_key
);
546 if (memcached_vdo(instance
, vector
, 4, false) != MEMCACHED_SUCCESS
)
548 memcached_io_reset(instance
);
552 memcached_server_response_decrement(instance
);
559 return MEMCACHED_BUFFERED
;
564 return MEMCACHED_SUCCESS
;
567 return memcached_response(server
, NULL
, 0, NULL
);