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