Updating for 1.0.2 release
[awesomized/libmemcached] / 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