projects / m6w6 / libmemcached / blob
? search:


1d75eb9a3aa143d21c4406aa30f73bf8849181e4
[m6w6/libmemcached] / libmemcached / storage.cc
1 /* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
2 *
3 * Libmemcached library
4 *
5 * Copyright (C) 2011 Data Differential, http://datadifferential.com/
6 * Copyright (C) 2006-2009 Brian Aker All rights reserved.
7 *
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions are
10 * met:
11 *
12 * * Redistributions of source code must retain the above copyright
13 * notice, this list of conditions and the following disclaimer.
14 *
15 * * Redistributions in binary form must reproduce the above
16 * copyright notice, this list of conditions and the following disclaimer
17 * in the documentation and/or other materials provided with the
18 * distribution.
19 *
20 * * The names of its contributors may not be used to endorse or
21 * promote products derived from this software without specific prior
22 * written permission.
23 *
24 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
25 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
26 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
27 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
28 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
29 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
30 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
31 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
32 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
33 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
34 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
35 *
36 */
37
38
39 #include <libmemcached/common.h>
40
41 enum memcached_storage_action_t {
42 SET_OP,
43 REPLACE_OP,
44 ADD_OP,
45 PREPEND_OP,
46 APPEND_OP,
47 CAS_OP
48 };
49
50 /* Inline this */
51 static inline const char *storage_op_string(memcached_storage_action_t verb)
52 {
53 switch (verb)
54 {
55 case REPLACE_OP:
56 return "replace ";
57
58 case ADD_OP:
59 return "add ";
60
61 case PREPEND_OP:
62 return "prepend ";
63
64 case APPEND_OP:
65 return "append ";
66
67 case CAS_OP:
68 return "cas ";
69
70 case SET_OP:
71 break;
72 }
73
74 return "set ";
75 }
76
77 static inline uint8_t get_com_code(const memcached_storage_action_t verb, const bool reply)
78 {
79 if (reply == false)
80 {
81 switch (verb)
82 {
83 case SET_OP:
84 return PROTOCOL_BINARY_CMD_SETQ;
85
86 case ADD_OP:
87 return PROTOCOL_BINARY_CMD_ADDQ;
88
89 case CAS_OP: /* FALLTHROUGH */
90 case REPLACE_OP:
91 return PROTOCOL_BINARY_CMD_REPLACEQ;
92
93 case APPEND_OP:
94 return PROTOCOL_BINARY_CMD_APPENDQ;
95
96 case PREPEND_OP:
97 return PROTOCOL_BINARY_CMD_PREPENDQ;
98 }
99 }
100
101 switch (verb)
102 {
103 case SET_OP:
104 break;
105
106 case ADD_OP:
107 return PROTOCOL_BINARY_CMD_ADD;
108
109 case CAS_OP: /* FALLTHROUGH */
110 case REPLACE_OP:
111 return PROTOCOL_BINARY_CMD_REPLACE;
112
113 case APPEND_OP:
114 return PROTOCOL_BINARY_CMD_APPEND;
115
116 case PREPEND_OP:
117 return PROTOCOL_BINARY_CMD_PREPEND;
118 }
119
120 return PROTOCOL_BINARY_CMD_SET;
121 }
122
123 static memcached_return_t memcached_send_binary(memcached_st *ptr,
124 memcached_server_write_instance_st server,
125 uint32_t server_key,
126 const char *key,
127 const size_t key_length,
128 const char *value,
129 const size_t value_length,
130 const time_t expiration,
131 const uint32_t flags,
132 const uint64_t cas,
133 const bool flush,
134 const bool reply,
135 memcached_storage_action_t verb)
136 {
137 protocol_binary_request_set request= {};
138 size_t send_length= sizeof(request.bytes);
139
140 request.message.header.request.magic= PROTOCOL_BINARY_REQ;
141 request.message.header.request.opcode= get_com_code(verb, reply);
142 request.message.header.request.keylen= htons((uint16_t)(key_length + memcached_array_size(ptr->_namespace)));
143 request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
144 if (verb == APPEND_OP or verb == PREPEND_OP)
145 {
146 send_length -= 8; /* append & prepend does not contain extras! */
147 }
148 else
149 {
150 request.message.header.request.extlen= 8;
151 request.message.body.flags= htonl(flags);
152 request.message.body.expiration= htonl((uint32_t)expiration);
153 }
154
155 request.message.header.request.bodylen= htonl((uint32_t) (key_length + memcached_array_size(ptr->_namespace) + value_length +
156 request.message.header.request.extlen));
157
158 if (cas)
159 {
160 request.message.header.request.cas= memcached_htonll(cas);
161 }
162
163 libmemcached_io_vector_st vector[]=
164 {
165 { NULL, 0 },
166 { request.bytes, send_length },
167 { memcached_array_string(ptr->_namespace), memcached_array_size(ptr->_namespace) },
168 { key, key_length },
169 { value, value_length }
170 };
171
172 /* write the header */
173 memcached_return_t rc;
174 if ((rc= memcached_vdo(server, vector, 5, flush)) != MEMCACHED_SUCCESS)
175 {
176 memcached_io_reset(server);
177
178 #if 0
179 if (memcached_has_error(ptr))
180 {
181 memcached_set_error(*server, rc, MEMCACHED_AT);
182 }
183 #endif
184
185 return MEMCACHED_WRITE_FAILURE;
186 }
187
188 if (verb == SET_OP and ptr->number_of_replicas > 0)
189 {
190 request.message.header.request.opcode= PROTOCOL_BINARY_CMD_SETQ;
191 WATCHPOINT_STRING("replicating");
192
193 for (uint32_t x= 0; x < ptr->number_of_replicas; x++)
194 {
195 ++server_key;
196 if (server_key == memcached_server_count(ptr))
197 {
198 server_key= 0;
199 }
200
201 memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, server_key);
202
203 if (memcached_vdo(instance, vector, 5, false) != MEMCACHED_SUCCESS)
204 {
205 memcached_io_reset(instance);
206 }
207 else
208 {
209 memcached_server_response_decrement(instance);
210 }
211 }
212 }
213
214 if (flush == false)
215 {
216 return MEMCACHED_BUFFERED;
217 }
218
219 // No reply always assumes success
220 if (reply == false)
221 {
222 return MEMCACHED_SUCCESS;
223 }
224
225 return memcached_response(server, NULL, 0, NULL);
226 }
227
228 static memcached_return_t memcached_send_ascii(memcached_st *ptr,
229 memcached_server_write_instance_st instance,
230 const char *key,
231 const size_t key_length,
232 const char *value,
233 const size_t value_length,
234 const time_t expiration,
235 const uint32_t flags,
236 const uint64_t cas,
237 const bool flush,
238 const bool reply,
239 const memcached_storage_action_t verb)
240 {
241 char flags_buffer[MEMCACHED_MAXIMUM_INTEGER_DISPLAY_LENGTH +1];
242 int flags_buffer_length= snprintf(flags_buffer, sizeof(flags_buffer), " %u", flags);
243 if (size_t(flags_buffer_length) >= sizeof(flags_buffer) or flags_buffer_length < 0)
244 {
245 return memcached_set_error(*instance, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT,
246 memcached_literal_param("snprintf(MEMCACHED_MAXIMUM_INTEGER_DISPLAY_LENGTH)"));
247 }
248
249 char expiration_buffer[MEMCACHED_MAXIMUM_INTEGER_DISPLAY_LENGTH +1];
250 int expiration_buffer_length= snprintf(expiration_buffer, sizeof(expiration_buffer), " %llu", (unsigned long long)expiration);
251 if (size_t(expiration_buffer_length) >= sizeof(expiration_buffer) or expiration_buffer_length < 0)
252 {
253 return memcached_set_error(*instance, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT,
254 memcached_literal_param("snprintf(MEMCACHED_MAXIMUM_INTEGER_DISPLAY_LENGTH)"));
255 }
256
257 char value_buffer[MEMCACHED_MAXIMUM_INTEGER_DISPLAY_LENGTH +1];
258 int value_buffer_length= snprintf(value_buffer, sizeof(value_buffer), " %llu", (unsigned long long)value_length);
259 if (size_t(value_buffer_length) >= sizeof(value_buffer) or value_buffer_length < 0)
260 {
261 return memcached_set_error(*instance, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT,
262 memcached_literal_param("snprintf(MEMCACHED_MAXIMUM_INTEGER_DISPLAY_LENGTH)"));
263 }
264
265 char cas_buffer[MEMCACHED_MAXIMUM_INTEGER_DISPLAY_LENGTH +1];
266 int cas_buffer_length= 0;
267 if (cas)
268 {
269 cas_buffer_length= snprintf(cas_buffer, sizeof(cas_buffer), " %llu", (unsigned long long)cas);
270 if (size_t(cas_buffer_length) >= sizeof(cas_buffer) or cas_buffer_length < 0)
271 {
272 return memcached_set_error(*instance, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT,
273 memcached_literal_param("snprintf(MEMCACHED_MAXIMUM_INTEGER_DISPLAY_LENGTH)"));
274 }
275 }
276
277 libmemcached_io_vector_st vector[]=
278 {
279 { NULL, 0 },
280 { storage_op_string(verb), strlen(storage_op_string(verb))},
281 { memcached_array_string(ptr->_namespace), memcached_array_size(ptr->_namespace) },
282 { key, key_length },
283 { flags_buffer, flags_buffer_length },
284 { expiration_buffer, expiration_buffer_length },
285 { value_buffer, value_buffer_length },
286 { cas_buffer, cas_buffer_length },
287 { " noreply", reply ? 0 : memcached_literal_param_size(" noreply") },
288 { memcached_literal_param("\r\n") },
289 { value, value_length },
290 { memcached_literal_param("\r\n") }
291 };
292
293 /* Send command header */
294 memcached_return_t rc= memcached_vdo(instance, vector, 12, flush);
295
296 // If we should not reply, return with MEMCACHED_SUCCESS, unless error
297 if (reply == false)
298 {
299 return memcached_success(rc) ? MEMCACHED_SUCCESS : rc;
300 }
301
302 if (flush == false)
303 {
304 return memcached_success(rc) ? MEMCACHED_BUFFERED : rc;
305 }
306
307 if (rc == MEMCACHED_SUCCESS)
308 {
309 char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
310 rc= memcached_response(instance, buffer, sizeof(buffer), NULL);
311
312 if (rc == MEMCACHED_STORED)
313 {
314 return MEMCACHED_SUCCESS;
315 }
316 }
317
318 if (rc == MEMCACHED_WRITE_FAILURE)
319 {
320 memcached_io_reset(instance);
321 }
322
323 assert(memcached_failed(rc));
324 #if 0
325 if (memcached_has_error(ptr) == false)
326 {
327 return memcached_set_error(*ptr, rc, MEMCACHED_AT);
328 }
329 #endif
330
331 return rc;
332 }
333
334 static inline memcached_return_t memcached_send(memcached_st *ptr,
335 const char *group_key, size_t group_key_length,
336 const char *key, size_t key_length,
337 const char *value, size_t value_length,
338 const time_t expiration,
339 const uint32_t flags,
340 const uint64_t cas,
341 memcached_storage_action_t verb)
342 {
343 memcached_return_t rc;
344 if (memcached_failed(rc= initialize_query(ptr, true)))
345 {
346 return rc;
347 }
348
349 if (memcached_failed(memcached_key_test(*ptr, (const char **)&key, &key_length, 1)))
350 {
351 return memcached_last_error(ptr);
352 }
353
354 uint32_t server_key= memcached_generate_hash_with_redistribution(ptr, group_key, group_key_length);
355 memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, server_key);
356
357 WATCHPOINT_SET(instance->io_wait_count.read= 0);
358 WATCHPOINT_SET(instance->io_wait_count.write= 0);
359
360
361 bool flush= true;
362 if (memcached_is_buffering(instance->root) and verb == SET_OP)
363 {
364 flush= false;
365 }
366
367 bool reply= memcached_is_replying(ptr);
368
369 if (memcached_is_binary(ptr))
370 {
371 return memcached_send_binary(ptr, instance, server_key,
372 key, key_length,
373 value, value_length, expiration,
374 flags, cas, flush, reply, verb);
375 }
376
377 return memcached_send_ascii(ptr, instance,
378 key, key_length,
379 value, value_length, expiration,
380 flags, cas, flush, reply, verb);
381 }
382
383
384 memcached_return_t memcached_set(memcached_st *ptr, const char *key, size_t key_length,
385 const char *value, size_t value_length,
386 time_t expiration,
387 uint32_t flags)
388 {
389 memcached_return_t rc;
390 LIBMEMCACHED_MEMCACHED_SET_START();
391 rc= memcached_send(ptr, key, key_length,
392 key, key_length, value, value_length,
393 expiration, flags, 0, SET_OP);
394 LIBMEMCACHED_MEMCACHED_SET_END();
395 return rc;
396 }
397
398 memcached_return_t memcached_add(memcached_st *ptr,
399 const char *key, size_t key_length,
400 const char *value, size_t value_length,
401 time_t expiration,
402 uint32_t flags)
403 {
404 memcached_return_t rc;
405 LIBMEMCACHED_MEMCACHED_ADD_START();
406 rc= memcached_send(ptr, key, key_length,
407 key, key_length, value, value_length,
408 expiration, flags, 0, ADD_OP);
409
410 LIBMEMCACHED_MEMCACHED_ADD_END();
411 return rc;
412 }
413
414 memcached_return_t memcached_replace(memcached_st *ptr,
415 const char *key, size_t key_length,
416 const char *value, size_t value_length,
417 time_t expiration,
418 uint32_t flags)
419 {
420 memcached_return_t rc;
421 LIBMEMCACHED_MEMCACHED_REPLACE_START();
422 rc= memcached_send(ptr, key, key_length,
423 key, key_length, value, value_length,
424 expiration, flags, 0, REPLACE_OP);
425 LIBMEMCACHED_MEMCACHED_REPLACE_END();
426 return rc;
427 }
428
429 memcached_return_t memcached_prepend(memcached_st *ptr,
430 const char *key, size_t key_length,
431 const char *value, size_t value_length,
432 time_t expiration,
433 uint32_t flags)
434 {
435 memcached_return_t rc;
436 rc= memcached_send(ptr, key, key_length,
437 key, key_length, value, value_length,
438 expiration, flags, 0, PREPEND_OP);
439 return rc;
440 }
441
442 memcached_return_t memcached_append(memcached_st *ptr,
443 const char *key, size_t key_length,
444 const char *value, size_t value_length,
445 time_t expiration,
446 uint32_t flags)
447 {
448 memcached_return_t rc;
449 rc= memcached_send(ptr, key, key_length,
450 key, key_length, value, value_length,
451 expiration, flags, 0, APPEND_OP);
452 return rc;
453 }
454
455 memcached_return_t memcached_cas(memcached_st *ptr,
456 const char *key, size_t key_length,
457 const char *value, size_t value_length,
458 time_t expiration,
459 uint32_t flags,
460 uint64_t cas)
461 {
462 memcached_return_t rc;
463 rc= memcached_send(ptr, key, key_length,
464 key, key_length, value, value_length,
465 expiration, flags, cas, CAS_OP);
466 return rc;
467 }
468
469 memcached_return_t memcached_set_by_key(memcached_st *ptr,
470 const char *group_key,
471 size_t group_key_length,
472 const char *key, size_t key_length,
473 const char *value, size_t value_length,
474 time_t expiration,
475 uint32_t flags)
476 {
477 memcached_return_t rc;
478 LIBMEMCACHED_MEMCACHED_SET_START();
479 rc= memcached_send(ptr, group_key, group_key_length,
480 key, key_length, value, value_length,
481 expiration, flags, 0, SET_OP);
482 LIBMEMCACHED_MEMCACHED_SET_END();
483 return rc;
484 }
485
486 memcached_return_t memcached_add_by_key(memcached_st *ptr,
487 const char *group_key, size_t group_key_length,
488 const char *key, size_t key_length,
489 const char *value, size_t value_length,
490 time_t expiration,
491 uint32_t flags)
492 {
493 memcached_return_t rc;
494 LIBMEMCACHED_MEMCACHED_ADD_START();
495 rc= memcached_send(ptr, group_key, group_key_length,
496 key, key_length, value, value_length,
497 expiration, flags, 0, ADD_OP);
498 LIBMEMCACHED_MEMCACHED_ADD_END();
499 return rc;
500 }
501
502 memcached_return_t memcached_replace_by_key(memcached_st *ptr,
503 const char *group_key, size_t group_key_length,
504 const char *key, size_t key_length,
505 const char *value, size_t value_length,
506 time_t expiration,
507 uint32_t flags)
508 {
509 memcached_return_t rc;
510 LIBMEMCACHED_MEMCACHED_REPLACE_START();
511 rc= memcached_send(ptr, group_key, group_key_length,
512 key, key_length, value, value_length,
513 expiration, flags, 0, REPLACE_OP);
514 LIBMEMCACHED_MEMCACHED_REPLACE_END();
515 return rc;
516 }
517
518 memcached_return_t memcached_prepend_by_key(memcached_st *ptr,
519 const char *group_key, size_t group_key_length,
520 const char *key, size_t key_length,
521 const char *value, size_t value_length,
522 time_t expiration,
523 uint32_t flags)
524 {
525 return memcached_send(ptr, group_key, group_key_length,
526 key, key_length, value, value_length,
527 expiration, flags, 0, PREPEND_OP);
528 }
529
530 memcached_return_t memcached_append_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 return memcached_send(ptr, group_key, group_key_length,
538 key, key_length, value, value_length,
539 expiration, flags, 0, APPEND_OP);
540 }
541
542 memcached_return_t memcached_cas_by_key(memcached_st *ptr,
543 const char *group_key, size_t group_key_length,
544 const char *key, size_t key_length,
545 const char *value, size_t value_length,
546 time_t expiration,
547 uint32_t flags,
548 uint64_t cas)
549 {
550 return memcached_send(ptr, group_key, group_key_length,
551 key, key_length, value, value_length,
552 expiration, flags, cas, CAS_OP);
553 }
554
Resurrection of libmemcached https://launchpad.net/libmemcached