Merge in work to test return types.
[m6w6/libmemcached] / libmemcached / get.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 #include <libmemcached/common.h>
39
40 /*
41 What happens if no servers exist?
42 */
43 char *memcached_get(memcached_st *ptr, const char *key,
44 size_t key_length,
45 size_t *value_length,
46 uint32_t *flags,
47 memcached_return_t *error)
48 {
49 return memcached_get_by_key(ptr, NULL, 0, key, key_length, value_length,
50 flags, error);
51 }
52
53 static memcached_return_t memcached_mget_by_key_real(memcached_st *ptr,
54 const char *group_key,
55 size_t group_key_length,
56 const char * const *keys,
57 const size_t *key_length,
58 size_t number_of_keys,
59 bool mget_mode);
60
61 char *memcached_get_by_key(memcached_st *ptr,
62 const char *group_key,
63 size_t group_key_length,
64 const char *key, size_t key_length,
65 size_t *value_length,
66 uint32_t *flags,
67 memcached_return_t *error)
68 {
69 memcached_return_t unused;
70 if (error == NULL)
71 {
72 error= &unused;
73 }
74
75 if (ptr->flags.use_udp)
76 {
77 if (value_length)
78 {
79 *value_length= 0;
80 }
81
82 *error= memcached_set_error(*ptr, MEMCACHED_NOT_SUPPORTED, MEMCACHED_AT);
83 return NULL;
84 }
85
86 uint64_t query_id= ptr->query_id;
87 (void)query_id;
88
89 /* Request the key */
90 *error= memcached_mget_by_key_real(ptr, group_key, group_key_length,
91 (const char * const *)&key, &key_length,
92 1, false);
93 assert_msg(ptr->query_id == query_id +1, "Programmer error, the query_id was not incremented.");
94
95
96 if (memcached_failed(*error))
97 {
98 if (memcached_has_current_error(*ptr)) // Find the most accurate error
99 {
100 *error= memcached_last_error(ptr);
101 }
102
103 if (value_length)
104 *value_length= 0;
105
106 return NULL;
107 }
108
109 char *value= memcached_fetch(ptr, NULL, NULL,
110 value_length, flags, error);
111 assert_msg(ptr->query_id == query_id +1, "Programmer error, the query_id was not incremented.");
112
113 /* This is for historical reasons */
114 if (*error == MEMCACHED_END)
115 *error= MEMCACHED_NOTFOUND;
116
117 if (value == NULL)
118 {
119 if (ptr->get_key_failure && *error == MEMCACHED_NOTFOUND)
120 {
121 memcached_result_reset(&ptr->result);
122 memcached_return_t rc= ptr->get_key_failure(ptr, key, key_length, &ptr->result);
123
124 /* On all failure drop to returning NULL */
125 if (rc == MEMCACHED_SUCCESS || rc == MEMCACHED_BUFFERED)
126 {
127 if (rc == MEMCACHED_BUFFERED)
128 {
129 uint64_t latch; /* We use latch to track the state of the original socket */
130 latch= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS);
131 if (latch == 0)
132 memcached_behavior_set(ptr, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS, 1);
133
134 rc= memcached_set(ptr, key, key_length,
135 (memcached_result_value(&ptr->result)),
136 (memcached_result_length(&ptr->result)),
137 0,
138 (memcached_result_flags(&ptr->result)));
139
140 if (rc == MEMCACHED_BUFFERED && latch == 0)
141 {
142 memcached_behavior_set(ptr, MEMCACHED_BEHAVIOR_BUFFER_REQUESTS, 0);
143 }
144 }
145 else
146 {
147 rc= memcached_set(ptr, key, key_length,
148 (memcached_result_value(&ptr->result)),
149 (memcached_result_length(&ptr->result)),
150 0,
151 (memcached_result_flags(&ptr->result)));
152 }
153
154 if (rc == MEMCACHED_SUCCESS || rc == MEMCACHED_BUFFERED)
155 {
156 *error= rc;
157 *value_length= memcached_result_length(&ptr->result);
158 *flags= memcached_result_flags(&ptr->result);
159 return memcached_string_take_value(&ptr->result.value);
160 }
161 }
162 }
163 assert_msg(ptr->query_id == query_id +1, "Programmer error, the query_id was not incremented.");
164
165 return NULL;
166 }
167
168 size_t dummy_length;
169 uint32_t dummy_flags;
170 memcached_return_t dummy_error;
171
172 char *dummy_value= memcached_fetch(ptr, NULL, NULL,
173 &dummy_length, &dummy_flags,
174 &dummy_error);
175 assert_msg(dummy_value == 0, "memcached_fetch() returned additional values beyond the single get it expected");
176 assert_msg(dummy_length == 0, "memcached_fetch() returned additional values beyond the single get it expected");
177 assert_msg(ptr->query_id == query_id +1, "Programmer error, the query_id was not incremented.");
178
179 return value;
180 }
181
182 memcached_return_t memcached_mget(memcached_st *ptr,
183 const char * const *keys,
184 const size_t *key_length,
185 size_t number_of_keys)
186 {
187 return memcached_mget_by_key(ptr, NULL, 0, keys, key_length, number_of_keys);
188 }
189
190 static memcached_return_t binary_mget_by_key(memcached_st *ptr,
191 uint32_t master_server_key,
192 bool is_group_key_set,
193 const char * const *keys,
194 const size_t *key_length,
195 size_t number_of_keys,
196 bool mget_mode);
197
198 static memcached_return_t memcached_mget_by_key_real(memcached_st *ptr,
199 const char *group_key,
200 size_t group_key_length,
201 const char * const *keys,
202 const size_t *key_length,
203 size_t number_of_keys,
204 bool mget_mode)
205 {
206 bool failures_occured_in_sending= false;
207 const char *get_command= "get ";
208 uint8_t get_command_length= 4;
209 unsigned int master_server_key= (unsigned int)-1; /* 0 is a valid server id! */
210
211 memcached_return_t rc;
212 if (memcached_failed(rc= initialize_query(ptr)))
213 {
214 return rc;
215 }
216
217 if (ptr->flags.use_udp)
218 {
219 return memcached_set_error(*ptr, MEMCACHED_NOT_SUPPORTED, MEMCACHED_AT);
220 }
221
222 LIBMEMCACHED_MEMCACHED_MGET_START();
223
224 if (number_of_keys == 0)
225 {
226 return memcached_set_error(*ptr, MEMCACHED_NOTFOUND, MEMCACHED_AT, memcached_literal_param("number_of_keys was zero"));
227 }
228
229 if (memcached_failed(memcached_key_test(*ptr, keys, key_length, number_of_keys)))
230 {
231 return memcached_set_error(*ptr, MEMCACHED_BAD_KEY_PROVIDED, MEMCACHED_AT, memcached_literal_param("A bad key value was provided"));
232 }
233
234 bool is_group_key_set= false;
235 if (group_key and group_key_length)
236 {
237 if (memcached_failed(memcached_key_test(*ptr, (const char * const *)&group_key, &group_key_length, 1)))
238 {
239 return memcached_set_error(*ptr, MEMCACHED_BAD_KEY_PROVIDED, MEMCACHED_AT, memcached_literal_param("A bad group key was provided."));
240 }
241
242 master_server_key= memcached_generate_hash_with_redistribution(ptr, group_key, group_key_length);
243 is_group_key_set= true;
244 }
245
246 /*
247 Here is where we pay for the non-block API. We need to remove any data sitting
248 in the queue before we start our get.
249
250 It might be optimum to bounce the connection if count > some number.
251 */
252 for (uint32_t x= 0; x < memcached_server_count(ptr); x++)
253 {
254 memcached_server_write_instance_st instance=
255 memcached_server_instance_fetch(ptr, x);
256
257 if (memcached_server_response_count(instance))
258 {
259 char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
260
261 if (ptr->flags.no_block)
262 (void)memcached_io_write(instance, NULL, 0, true);
263
264 while(memcached_server_response_count(instance))
265 (void)memcached_response(instance, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, &ptr->result);
266 }
267 }
268
269 if (ptr->flags.binary_protocol)
270 {
271 return binary_mget_by_key(ptr, master_server_key, is_group_key_set, keys,
272 key_length, number_of_keys, mget_mode);
273 }
274
275 if (ptr->flags.support_cas)
276 {
277 get_command= "gets ";
278 get_command_length= 5;
279 }
280
281 /*
282 If a server fails we warn about errors and start all over with sending keys
283 to the server.
284 */
285 WATCHPOINT_ASSERT(rc == MEMCACHED_SUCCESS);
286 size_t hosts_connected= 0;
287 for (uint32_t x= 0; x < number_of_keys; x++)
288 {
289 memcached_server_write_instance_st instance;
290 uint32_t server_key;
291
292 if (is_group_key_set)
293 {
294 server_key= master_server_key;
295 }
296 else
297 {
298 server_key= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
299 }
300
301 instance= memcached_server_instance_fetch(ptr, server_key);
302
303 struct libmemcached_io_vector_st vector[]=
304 {
305 { get_command, get_command_length },
306 { memcached_array_string(ptr->_namespace), memcached_array_size(ptr->_namespace) },
307 { keys[x], key_length[x] },
308 { memcached_literal_param(" ") }
309 };
310
311
312 if (memcached_server_response_count(instance) == 0)
313 {
314 rc= memcached_connect(instance);
315
316 if (memcached_failed(rc))
317 {
318 memcached_set_error(*instance, rc, MEMCACHED_AT);
319 continue;
320 }
321 hosts_connected++;
322
323 if ((memcached_io_writev(instance, vector, 4, false)) == -1)
324 {
325 failures_occured_in_sending= true;
326 continue;
327 }
328 WATCHPOINT_ASSERT(instance->cursor_active == 0);
329 memcached_server_response_increment(instance);
330 WATCHPOINT_ASSERT(instance->cursor_active == 1);
331 }
332 else
333 {
334 if ((memcached_io_writev(instance, (vector + 1), 3, false)) == -1)
335 {
336 memcached_server_response_reset(instance);
337 failures_occured_in_sending= true;
338 continue;
339 }
340 }
341 }
342
343 if (hosts_connected == 0)
344 {
345 LIBMEMCACHED_MEMCACHED_MGET_END();
346
347 if (memcached_failed(rc))
348 return rc;
349
350 return memcached_set_error(*ptr, MEMCACHED_NO_SERVERS, MEMCACHED_AT);
351 }
352
353
354 /*
355 Should we muddle on if some servers are dead?
356 */
357 bool success_happened= false;
358 for (uint32_t x= 0; x < memcached_server_count(ptr); x++)
359 {
360 memcached_server_write_instance_st instance=
361 memcached_server_instance_fetch(ptr, x);
362
363 if (memcached_server_response_count(instance))
364 {
365 /* We need to do something about non-connnected hosts in the future */
366 if ((memcached_io_write(instance, "\r\n", 2, true)) == -1)
367 {
368 failures_occured_in_sending= true;
369 }
370 else
371 {
372 success_happened= true;
373 }
374 }
375 }
376
377 LIBMEMCACHED_MEMCACHED_MGET_END();
378
379 if (failures_occured_in_sending && success_happened)
380 {
381 return MEMCACHED_SOME_ERRORS;
382 }
383
384 if (success_happened)
385 return MEMCACHED_SUCCESS;
386
387 return MEMCACHED_FAILURE; // Complete failure occurred
388 }
389
390 memcached_return_t memcached_mget_by_key(memcached_st *ptr,
391 const char *group_key,
392 size_t group_key_length,
393 const char * const *keys,
394 const size_t *key_length,
395 size_t number_of_keys)
396 {
397 return memcached_mget_by_key_real(ptr, group_key, group_key_length, keys,
398 key_length, number_of_keys, true);
399 }
400
401 memcached_return_t memcached_mget_execute(memcached_st *ptr,
402 const char * const *keys,
403 const size_t *key_length,
404 size_t number_of_keys,
405 memcached_execute_fn *callback,
406 void *context,
407 unsigned int number_of_callbacks)
408 {
409 return memcached_mget_execute_by_key(ptr, NULL, 0, keys, key_length,
410 number_of_keys, callback,
411 context, number_of_callbacks);
412 }
413
414 memcached_return_t memcached_mget_execute_by_key(memcached_st *ptr,
415 const char *group_key,
416 size_t group_key_length,
417 const char * const *keys,
418 const size_t *key_length,
419 size_t number_of_keys,
420 memcached_execute_fn *callback,
421 void *context,
422 unsigned int number_of_callbacks)
423 {
424 if ((ptr->flags.binary_protocol) == 0)
425 {
426 return MEMCACHED_NOT_SUPPORTED;
427 }
428
429 memcached_return_t rc;
430 memcached_callback_st *original_callbacks= ptr->callbacks;
431 memcached_callback_st cb= {
432 callback,
433 context,
434 number_of_callbacks
435 };
436
437 ptr->callbacks= &cb;
438 rc= memcached_mget_by_key(ptr, group_key, group_key_length, keys,
439 key_length, number_of_keys);
440 ptr->callbacks= original_callbacks;
441 return rc;
442 }
443
444 static memcached_return_t simple_binary_mget(memcached_st *ptr,
445 uint32_t master_server_key,
446 bool is_group_key_set,
447 const char * const *keys,
448 const size_t *key_length,
449 size_t number_of_keys, bool mget_mode)
450 {
451 memcached_return_t rc= MEMCACHED_NOTFOUND;
452
453 bool flush= (number_of_keys == 1);
454
455 /*
456 If a server fails we warn about errors and start all over with sending keys
457 to the server.
458 */
459 for (uint32_t x= 0; x < number_of_keys; ++x)
460 {
461 uint32_t server_key;
462
463 if (is_group_key_set)
464 {
465 server_key= master_server_key;
466 }
467 else
468 {
469 server_key= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
470 }
471
472 memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, server_key);
473
474 if (memcached_server_response_count(instance) == 0)
475 {
476 rc= memcached_connect(instance);
477 if (memcached_failed(rc))
478 {
479 continue;
480 }
481 }
482
483 protocol_binary_request_getk request= { }; //= {.bytes= {0}};
484 request.message.header.request.magic= PROTOCOL_BINARY_REQ;
485 if (mget_mode)
486 request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETKQ;
487 else
488 request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
489
490 memcached_return_t vk;
491 vk= memcached_validate_key_length(key_length[x],
492 ptr->flags.binary_protocol);
493 unlikely (vk != MEMCACHED_SUCCESS)
494 {
495 if (x > 0)
496 {
497 memcached_io_reset(instance);
498 }
499
500 return vk;
501 }
502
503 request.message.header.request.keylen= htons((uint16_t)(key_length[x] + memcached_array_size(ptr->_namespace)));
504 request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
505 request.message.header.request.bodylen= htonl((uint32_t)( key_length[x] + memcached_array_size(ptr->_namespace)));
506
507 struct libmemcached_io_vector_st vector[]=
508 {
509 { request.bytes, sizeof(request.bytes) },
510 { memcached_array_string(ptr->_namespace), memcached_array_size(ptr->_namespace) },
511 { keys[x], key_length[x] }
512 };
513
514 if (memcached_io_writev(instance, vector, 3, flush) == -1)
515 {
516 memcached_server_response_reset(instance);
517 rc= MEMCACHED_SOME_ERRORS;
518 continue;
519 }
520
521 /* We just want one pending response per server */
522 memcached_server_response_reset(instance);
523 memcached_server_response_increment(instance);
524 if ((x > 0 && x == ptr->io_key_prefetch) && memcached_flush_buffers(ptr) != MEMCACHED_SUCCESS)
525 {
526 rc= MEMCACHED_SOME_ERRORS;
527 }
528 }
529
530 if (mget_mode)
531 {
532 /*
533 Send a noop command to flush the buffers
534 */
535 protocol_binary_request_noop request= {}; //= {.bytes= {0}};
536 request.message.header.request.magic= PROTOCOL_BINARY_REQ;
537 request.message.header.request.opcode= PROTOCOL_BINARY_CMD_NOOP;
538 request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
539
540 for (uint32_t x= 0; x < memcached_server_count(ptr); ++x)
541 {
542 memcached_server_write_instance_st instance=
543 memcached_server_instance_fetch(ptr, x);
544
545 if (memcached_server_response_count(instance))
546 {
547 if (memcached_io_write(instance, NULL, 0, true) == -1)
548 {
549 memcached_server_response_reset(instance);
550 memcached_io_reset(instance);
551 rc= MEMCACHED_SOME_ERRORS;
552 }
553
554 if (memcached_io_write(instance, request.bytes,
555 sizeof(request.bytes), true) == -1)
556 {
557 memcached_server_response_reset(instance);
558 memcached_io_reset(instance);
559 rc= MEMCACHED_SOME_ERRORS;
560 }
561 }
562 }
563 }
564
565
566 return rc;
567 }
568
569 static memcached_return_t replication_binary_mget(memcached_st *ptr,
570 uint32_t* hash,
571 bool* dead_servers,
572 const char *const *keys,
573 const size_t *key_length,
574 size_t number_of_keys)
575 {
576 memcached_return_t rc= MEMCACHED_NOTFOUND;
577 uint32_t start= 0;
578 uint64_t randomize_read= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_RANDOMIZE_REPLICA_READ);
579
580 if (randomize_read)
581 start= (uint32_t)random() % (uint32_t)(ptr->number_of_replicas + 1);
582
583 /* Loop for each replica */
584 for (uint32_t replica= 0; replica <= ptr->number_of_replicas; ++replica)
585 {
586 bool success= true;
587
588 for (uint32_t x= 0; x < number_of_keys; ++x)
589 {
590 if (hash[x] == memcached_server_count(ptr))
591 continue; /* Already successfully sent */
592
593 uint32_t server= hash[x] + replica;
594
595 /* In case of randomized reads */
596 if (randomize_read && ((server + start) <= (hash[x] + ptr->number_of_replicas)))
597 server += start;
598
599 while (server >= memcached_server_count(ptr))
600 {
601 server -= memcached_server_count(ptr);
602 }
603
604 if (dead_servers[server])
605 {
606 continue;
607 }
608
609 memcached_server_write_instance_st instance= memcached_server_instance_fetch(ptr, server);
610
611 if (memcached_server_response_count(instance) == 0)
612 {
613 rc= memcached_connect(instance);
614
615 if (memcached_failed(rc))
616 {
617 memcached_io_reset(instance);
618 dead_servers[server]= true;
619 success= false;
620 continue;
621 }
622 }
623
624 protocol_binary_request_getk request= {};
625 request.message.header.request.magic= PROTOCOL_BINARY_REQ;
626 request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
627 request.message.header.request.keylen= htons((uint16_t)(key_length[x] + memcached_array_size(ptr->_namespace)));
628 request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
629 request.message.header.request.bodylen= htonl((uint32_t)(key_length[x] + memcached_array_size(ptr->_namespace)));
630
631 /*
632 * We need to disable buffering to actually know that the request was
633 * successfully sent to the server (so that we should expect a result
634 * back). It would be nice to do this in buffered mode, but then it
635 * would be complex to handle all error situations if we got to send
636 * some of the messages, and then we failed on writing out some others
637 * and we used the callback interface from memcached_mget_execute so
638 * that we might have processed some of the responses etc. For now,
639 * just make sure we work _correctly_
640 */
641 struct libmemcached_io_vector_st vector[]=
642 {
643 { request.bytes, sizeof(request.bytes) },
644 { memcached_array_string(ptr->_namespace), memcached_array_size(ptr->_namespace) },
645 { keys[x], key_length[x] }
646 };
647
648 if (memcached_io_writev(instance, vector, 3, true) == -1)
649 {
650 memcached_io_reset(instance);
651 dead_servers[server]= true;
652 success= false;
653 continue;
654 }
655
656 memcached_server_response_increment(instance);
657 hash[x]= memcached_server_count(ptr);
658 }
659
660 if (success)
661 {
662 break;
663 }
664 }
665
666 return rc;
667 }
668
669 static memcached_return_t binary_mget_by_key(memcached_st *ptr,
670 uint32_t master_server_key,
671 bool is_group_key_set,
672 const char * const *keys,
673 const size_t *key_length,
674 size_t number_of_keys,
675 bool mget_mode)
676 {
677 if (ptr->number_of_replicas == 0)
678 {
679 return simple_binary_mget(ptr, master_server_key, is_group_key_set,
680 keys, key_length, number_of_keys, mget_mode);
681 }
682
683 uint32_t* hash= libmemcached_xvalloc(ptr, number_of_keys, uint32_t);
684 bool* dead_servers= libmemcached_xcalloc(ptr, memcached_server_count(ptr), bool);
685
686 if (hash == NULL || dead_servers == NULL)
687 {
688 libmemcached_free(ptr, hash);
689 libmemcached_free(ptr, dead_servers);
690 return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
691 }
692
693 if (is_group_key_set)
694 {
695 for (size_t x= 0; x < number_of_keys; x++)
696 {
697 hash[x]= master_server_key;
698 }
699 }
700 else
701 {
702 for (size_t x= 0; x < number_of_keys; x++)
703 {
704 hash[x]= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
705 }
706 }
707
708 memcached_return_t rc= replication_binary_mget(ptr, hash, dead_servers, keys,
709 key_length, number_of_keys);
710
711 WATCHPOINT_IFERROR(rc);
712 libmemcached_free(ptr, hash);
713 libmemcached_free(ptr, dead_servers);
714
715 return MEMCACHED_SUCCESS;
716 }