Partial encapsulation of memcached_st->hosts
[m6w6/libmemcached] / libmemcached / storage.c
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 "common.h"
13
14 typedef enum {
15 SET_OP,
16 REPLACE_OP,
17 ADD_OP,
18 PREPEND_OP,
19 APPEND_OP,
20 CAS_OP,
21 } memcached_storage_action_t;
22
23 /* Inline this */
24 static inline const char *storage_op_string(memcached_storage_action_t verb)
25 {
26 switch (verb)
27 {
28 case SET_OP:
29 return "set ";
30 case REPLACE_OP:
31 return "replace ";
32 case ADD_OP:
33 return "add ";
34 case PREPEND_OP:
35 return "prepend ";
36 case APPEND_OP:
37 return "append ";
38 case CAS_OP:
39 return "cas ";
40 default:
41 return "tosserror"; /* This is impossible, fixes issue for compiler warning in VisualStudio */
42 }
43
44 /* NOTREACHED */
45 }
46
47 static memcached_return_t memcached_send_binary(memcached_st *ptr,
48 const char *master_key,
49 size_t master_key_length,
50 const char *key,
51 size_t key_length,
52 const char *value,
53 size_t value_length,
54 time_t expiration,
55 uint32_t flags,
56 uint64_t cas,
57 memcached_storage_action_t verb);
58
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,
63 time_t expiration,
64 uint32_t flags,
65 uint64_t cas,
66 memcached_storage_action_t verb)
67 {
68 char to_write;
69 size_t write_length;
70 memcached_return_t rc;
71 char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
72 uint32_t server_key;
73 memcached_server_instance_st *instance;
74
75 WATCHPOINT_ASSERT(!(value == NULL && value_length > 0));
76
77 rc= memcached_validate_key_length(key_length, ptr->flags.binary_protocol);
78 unlikely (rc != MEMCACHED_SUCCESS)
79 return rc;
80
81 unlikely (memcached_server_count(ptr) == 0)
82 return MEMCACHED_NO_SERVERS;
83
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;
86
87 if (ptr->flags.binary_protocol)
88 {
89 return memcached_send_binary(ptr, master_key, master_key_length,
90 key, key_length,
91 value, value_length, expiration,
92 flags, cas, verb);
93 }
94
95 server_key= memcached_generate_hash(ptr, master_key, master_key_length);
96 instance= memcached_server_instance_fetch(ptr, server_key);
97
98 if (cas)
99 {
100 write_length= (size_t) snprintf(buffer, MEMCACHED_DEFAULT_COMMAND_SIZE,
101 "%s %s%.*s %u %llu %zu %llu%s\r\n",
102 storage_op_string(verb),
103 ptr->prefix_key,
104 (int)key_length, key, flags,
105 (unsigned long long)expiration, value_length,
106 (unsigned long long)cas,
107 (ptr->flags.no_reply) ? " noreply" : "");
108 }
109 else
110 {
111 char *buffer_ptr= buffer;
112 const char *command= storage_op_string(verb);
113
114 /* Copy in the command, no space needed, we handle that in the command function*/
115 memcpy(buffer_ptr, command, strlen(command));
116
117 /* Copy in the key prefix, switch to the buffer_ptr */
118 buffer_ptr= memcpy((buffer_ptr + strlen(command)), ptr->prefix_key, strlen(ptr->prefix_key));
119
120 /* Copy in the key, adjust point if a key prefix was used. */
121 buffer_ptr= memcpy(buffer_ptr + (ptr->prefix_key ? strlen(ptr->prefix_key) : 0),
122 key, key_length);
123 buffer_ptr+= key_length;
124 buffer_ptr[0]= ' ';
125 buffer_ptr++;
126
127 write_length= (size_t)(buffer_ptr - buffer);
128 write_length+= (size_t) snprintf(buffer_ptr, MEMCACHED_DEFAULT_COMMAND_SIZE,
129 "%u %llu %zu%s\r\n",
130 flags,
131 (unsigned long long)expiration, value_length,
132 ptr->flags.no_reply ? " noreply" : "");
133 }
134
135 if (ptr->flags.use_udp && ptr->flags.buffer_requests)
136 {
137 size_t cmd_size= write_length + value_length + 2;
138 if (cmd_size > MAX_UDP_DATAGRAM_LENGTH - UDP_DATAGRAM_HEADER_LENGTH)
139 return MEMCACHED_WRITE_FAILURE;
140 if (cmd_size + instance->write_buffer_offset > MAX_UDP_DATAGRAM_LENGTH)
141 memcached_io_write(instance, NULL, 0, 1);
142 }
143
144 if (write_length >= MEMCACHED_DEFAULT_COMMAND_SIZE)
145 {
146 rc= MEMCACHED_WRITE_FAILURE;
147 goto error;
148 }
149
150 /* Send command header */
151 rc= memcached_do(instance, buffer, write_length, 0);
152 if (rc != MEMCACHED_SUCCESS)
153 goto error;
154
155 /* Send command body */
156 if (memcached_io_write(instance, value, value_length, 0) == -1)
157 {
158 rc= MEMCACHED_WRITE_FAILURE;
159 goto error;
160 }
161
162 if (ptr->flags.buffer_requests && verb == SET_OP)
163 {
164 to_write= 0;
165 }
166 else
167 {
168 to_write= 1;
169 }
170
171 if (memcached_io_write(instance, "\r\n", 2, to_write) == -1)
172 {
173 rc= MEMCACHED_WRITE_FAILURE;
174 goto error;
175 }
176
177 if (ptr->flags.no_reply)
178 return (to_write == 0) ? MEMCACHED_BUFFERED : MEMCACHED_SUCCESS;
179
180 if (to_write == 0)
181 return MEMCACHED_BUFFERED;
182
183 rc= memcached_response(instance, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, NULL);
184
185 if (rc == MEMCACHED_STORED)
186 return MEMCACHED_SUCCESS;
187 else
188 return rc;
189
190 error:
191 memcached_io_reset(instance);
192
193 return rc;
194 }
195
196
197 memcached_return_t memcached_set(memcached_st *ptr, const char *key, size_t key_length,
198 const char *value, size_t value_length,
199 time_t expiration,
200 uint32_t flags)
201 {
202 memcached_return_t rc;
203 LIBMEMCACHED_MEMCACHED_SET_START();
204 rc= memcached_send(ptr, key, key_length,
205 key, key_length, value, value_length,
206 expiration, flags, 0, SET_OP);
207 LIBMEMCACHED_MEMCACHED_SET_END();
208 return rc;
209 }
210
211 memcached_return_t memcached_add(memcached_st *ptr,
212 const char *key, size_t key_length,
213 const char *value, size_t value_length,
214 time_t expiration,
215 uint32_t flags)
216 {
217 memcached_return_t rc;
218 LIBMEMCACHED_MEMCACHED_ADD_START();
219 rc= memcached_send(ptr, key, key_length,
220 key, key_length, value, value_length,
221 expiration, flags, 0, ADD_OP);
222 LIBMEMCACHED_MEMCACHED_ADD_END();
223 return rc;
224 }
225
226 memcached_return_t memcached_replace(memcached_st *ptr,
227 const char *key, size_t key_length,
228 const char *value, size_t value_length,
229 time_t expiration,
230 uint32_t flags)
231 {
232 memcached_return_t rc;
233 LIBMEMCACHED_MEMCACHED_REPLACE_START();
234 rc= memcached_send(ptr, key, key_length,
235 key, key_length, value, value_length,
236 expiration, flags, 0, REPLACE_OP);
237 LIBMEMCACHED_MEMCACHED_REPLACE_END();
238 return rc;
239 }
240
241 memcached_return_t memcached_prepend(memcached_st *ptr,
242 const char *key, size_t key_length,
243 const char *value, size_t value_length,
244 time_t expiration,
245 uint32_t flags)
246 {
247 memcached_return_t rc;
248 rc= memcached_send(ptr, key, key_length,
249 key, key_length, value, value_length,
250 expiration, flags, 0, PREPEND_OP);
251 return rc;
252 }
253
254 memcached_return_t memcached_append(memcached_st *ptr,
255 const char *key, size_t key_length,
256 const char *value, size_t value_length,
257 time_t expiration,
258 uint32_t flags)
259 {
260 memcached_return_t rc;
261 rc= memcached_send(ptr, key, key_length,
262 key, key_length, value, value_length,
263 expiration, flags, 0, APPEND_OP);
264 return rc;
265 }
266
267 memcached_return_t memcached_cas(memcached_st *ptr,
268 const char *key, size_t key_length,
269 const char *value, size_t value_length,
270 time_t expiration,
271 uint32_t flags,
272 uint64_t cas)
273 {
274 memcached_return_t rc;
275 rc= memcached_send(ptr, key, key_length,
276 key, key_length, value, value_length,
277 expiration, flags, cas, CAS_OP);
278 return rc;
279 }
280
281 memcached_return_t memcached_set_by_key(memcached_st *ptr,
282 const char *master_key __attribute__((unused)),
283 size_t master_key_length __attribute__((unused)),
284 const char *key, size_t key_length,
285 const char *value, size_t value_length,
286 time_t expiration,
287 uint32_t flags)
288 {
289 memcached_return_t rc;
290 LIBMEMCACHED_MEMCACHED_SET_START();
291 rc= memcached_send(ptr, master_key, master_key_length,
292 key, key_length, value, value_length,
293 expiration, flags, 0, SET_OP);
294 LIBMEMCACHED_MEMCACHED_SET_END();
295 return rc;
296 }
297
298 memcached_return_t memcached_add_by_key(memcached_st *ptr,
299 const char *master_key, size_t master_key_length,
300 const char *key, size_t key_length,
301 const char *value, size_t value_length,
302 time_t expiration,
303 uint32_t flags)
304 {
305 memcached_return_t rc;
306 LIBMEMCACHED_MEMCACHED_ADD_START();
307 rc= memcached_send(ptr, master_key, master_key_length,
308 key, key_length, value, value_length,
309 expiration, flags, 0, ADD_OP);
310 LIBMEMCACHED_MEMCACHED_ADD_END();
311 return rc;
312 }
313
314 memcached_return_t memcached_replace_by_key(memcached_st *ptr,
315 const char *master_key, size_t master_key_length,
316 const char *key, size_t key_length,
317 const char *value, size_t value_length,
318 time_t expiration,
319 uint32_t flags)
320 {
321 memcached_return_t rc;
322 LIBMEMCACHED_MEMCACHED_REPLACE_START();
323 rc= memcached_send(ptr, master_key, master_key_length,
324 key, key_length, value, value_length,
325 expiration, flags, 0, REPLACE_OP);
326 LIBMEMCACHED_MEMCACHED_REPLACE_END();
327 return rc;
328 }
329
330 memcached_return_t memcached_prepend_by_key(memcached_st *ptr,
331 const char *master_key, size_t master_key_length,
332 const char *key, size_t key_length,
333 const char *value, size_t value_length,
334 time_t expiration,
335 uint32_t flags)
336 {
337 memcached_return_t rc;
338 rc= memcached_send(ptr, master_key, master_key_length,
339 key, key_length, value, value_length,
340 expiration, flags, 0, PREPEND_OP);
341 return rc;
342 }
343
344 memcached_return_t memcached_append_by_key(memcached_st *ptr,
345 const char *master_key, size_t master_key_length,
346 const char *key, size_t key_length,
347 const char *value, size_t value_length,
348 time_t expiration,
349 uint32_t flags)
350 {
351 memcached_return_t rc;
352 rc= memcached_send(ptr, master_key, master_key_length,
353 key, key_length, value, value_length,
354 expiration, flags, 0, APPEND_OP);
355 return rc;
356 }
357
358 memcached_return_t memcached_cas_by_key(memcached_st *ptr,
359 const char *master_key, size_t master_key_length,
360 const char *key, size_t key_length,
361 const char *value, size_t value_length,
362 time_t expiration,
363 uint32_t flags,
364 uint64_t cas)
365 {
366 memcached_return_t rc;
367 rc= memcached_send(ptr, master_key, master_key_length,
368 key, key_length, value, value_length,
369 expiration, flags, cas, CAS_OP);
370 return rc;
371 }
372
373 static inline uint8_t get_com_code(memcached_storage_action_t verb, bool noreply)
374 {
375 /* 0 isn't a value we want, but GCC 4.2 seems to think ret can otherwise
376 * be used uninitialized in this function. FAIL */
377 uint8_t ret= 0;
378
379 if (noreply)
380 switch (verb)
381 {
382 case SET_OP:
383 ret=PROTOCOL_BINARY_CMD_SETQ;
384 break;
385 case ADD_OP:
386 ret=PROTOCOL_BINARY_CMD_ADDQ;
387 break;
388 case CAS_OP: /* FALLTHROUGH */
389 case REPLACE_OP:
390 ret=PROTOCOL_BINARY_CMD_REPLACEQ;
391 break;
392 case APPEND_OP:
393 ret=PROTOCOL_BINARY_CMD_APPENDQ;
394 break;
395 case PREPEND_OP:
396 ret=PROTOCOL_BINARY_CMD_PREPENDQ;
397 break;
398 default:
399 WATCHPOINT_ASSERT(verb);
400 break;
401 }
402 else
403 switch (verb)
404 {
405 case SET_OP:
406 ret=PROTOCOL_BINARY_CMD_SET;
407 break;
408 case ADD_OP:
409 ret=PROTOCOL_BINARY_CMD_ADD;
410 break;
411 case CAS_OP: /* FALLTHROUGH */
412 case REPLACE_OP:
413 ret=PROTOCOL_BINARY_CMD_REPLACE;
414 break;
415 case APPEND_OP:
416 ret=PROTOCOL_BINARY_CMD_APPEND;
417 break;
418 case PREPEND_OP:
419 ret=PROTOCOL_BINARY_CMD_PREPEND;
420 break;
421 default:
422 WATCHPOINT_ASSERT(verb);
423 break;
424 }
425
426 return ret;
427 }
428
429
430
431 static memcached_return_t memcached_send_binary(memcached_st *ptr,
432 const char *master_key,
433 size_t master_key_length,
434 const char *key,
435 size_t key_length,
436 const char *value,
437 size_t value_length,
438 time_t expiration,
439 uint32_t flags,
440 uint64_t cas,
441 memcached_storage_action_t verb)
442 {
443 uint8_t flush;
444 protocol_binary_request_set request= {.bytes= {0}};
445 size_t send_length= sizeof(request.bytes);
446 uint32_t server_key= memcached_generate_hash(ptr, master_key,
447 master_key_length);
448
449 memcached_server_instance_st *server=
450 memcached_server_instance_fetch(ptr, server_key);
451
452 bool noreply= server->root->flags.no_reply;
453
454 request.message.header.request.magic= PROTOCOL_BINARY_REQ;
455 request.message.header.request.opcode= get_com_code(verb, noreply);
456 request.message.header.request.keylen= htons((uint16_t)key_length);
457 request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
458 if (verb == APPEND_OP || verb == PREPEND_OP)
459 send_length -= 8; /* append & prepend does not contain extras! */
460 else
461 {
462 request.message.header.request.extlen= 8;
463 request.message.body.flags= htonl(flags);
464 request.message.body.expiration= htonl((uint32_t)expiration);
465 }
466
467 request.message.header.request.bodylen= htonl((uint32_t) (key_length + value_length +
468 request.message.header.request.extlen));
469
470 if (cas)
471 request.message.header.request.cas= htonll(cas);
472
473 flush= (uint8_t) ((server->root->flags.buffer_requests && verb == SET_OP) ? 0 : 1);
474
475 if (server->root->flags.use_udp && !flush)
476 {
477 size_t cmd_size= send_length + key_length + value_length;
478
479 if (cmd_size > MAX_UDP_DATAGRAM_LENGTH - UDP_DATAGRAM_HEADER_LENGTH)
480 {
481 return MEMCACHED_WRITE_FAILURE;
482 }
483 if (cmd_size + server->write_buffer_offset > MAX_UDP_DATAGRAM_LENGTH)
484 {
485 memcached_io_write(server, NULL, 0, 1);
486 }
487 }
488
489 /* write the header */
490 if ((memcached_do(server, (const char*)request.bytes, send_length, 0) != MEMCACHED_SUCCESS) ||
491 (memcached_io_write(server, key, key_length, 0) == -1) ||
492 (memcached_io_write(server, value, value_length, (char) flush) == -1))
493 {
494 memcached_io_reset(server);
495 return MEMCACHED_WRITE_FAILURE;
496 }
497
498 unlikely (verb == SET_OP && ptr->number_of_replicas > 0)
499 {
500 request.message.header.request.opcode= PROTOCOL_BINARY_CMD_SETQ;
501
502 for (uint32_t x= 0; x < ptr->number_of_replicas; x++)
503 {
504 memcached_server_instance_st *instance;
505
506 ++server_key;
507 if (server_key == memcached_server_count(ptr))
508 server_key= 0;
509
510 instance= memcached_server_instance_fetch(ptr, server_key);
511
512 if ((memcached_do(instance, (const char*)request.bytes,
513 send_length, 0) != MEMCACHED_SUCCESS) ||
514 (memcached_io_write(instance, key, key_length, 0) == -1) ||
515 (memcached_io_write(instance, value, value_length, (char) flush) == -1))
516 {
517 memcached_io_reset(instance);
518 }
519 else
520 {
521 memcached_server_response_decrement(instance);
522 }
523 }
524 }
525
526 if (flush == 0)
527 {
528 return MEMCACHED_BUFFERED;
529 }
530
531 if (noreply)
532 {
533 return MEMCACHED_SUCCESS;
534 }
535
536 return memcached_response(server, NULL, 0, NULL);
537 }
538