Fixed a typo that was causing a race condition error.
[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 __write_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 rc= MEMCACHED_SOME_ERRORS;
249 continue;
250 }
251 WATCHPOINT_ASSERT(instance->cursor_active == 0);
252 memcached_server_response_increment(instance);
253 WATCHPOINT_ASSERT(instance->cursor_active == 1);
254 }
255 else
256 {
257 if ((memcached_io_writev(instance, (vector + 1), 3, false)) == -1)
258 {
259 memcached_server_response_reset(instance);
260 rc= MEMCACHED_SOME_ERRORS;
261 continue;
262 }
263 }
264 }
265
266 /*
267 Should we muddle on if some servers are dead?
268 */
269 for (uint32_t x= 0; x < memcached_server_count(ptr); x++)
270 {
271 memcached_server_write_instance_st instance=
272 memcached_server_instance_fetch(ptr, x);
273
274 if (memcached_server_response_count(instance))
275 {
276 /* We need to do something about non-connnected hosts in the future */
277 if ((memcached_io_write(instance, "\r\n", 2, true)) == -1)
278 {
279 rc= MEMCACHED_SOME_ERRORS;
280 }
281 }
282 }
283
284 LIBMEMCACHED_MEMCACHED_MGET_END();
285 return rc;
286 }
287
288 memcached_return_t memcached_mget_by_key(memcached_st *ptr,
289 const char *master_key,
290 size_t master_key_length,
291 const char * const *keys,
292 const size_t *key_length,
293 size_t number_of_keys)
294 {
295 return memcached_mget_by_key_real(ptr, master_key, master_key_length, keys,
296 key_length, number_of_keys, true);
297 }
298
299 memcached_return_t memcached_mget_execute(memcached_st *ptr,
300 const char * const *keys,
301 const size_t *key_length,
302 size_t number_of_keys,
303 memcached_execute_fn *callback,
304 void *context,
305 unsigned int number_of_callbacks)
306 {
307 return memcached_mget_execute_by_key(ptr, NULL, 0, keys, key_length,
308 number_of_keys, callback,
309 context, number_of_callbacks);
310 }
311
312 memcached_return_t memcached_mget_execute_by_key(memcached_st *ptr,
313 const char *master_key,
314 size_t master_key_length,
315 const char * const *keys,
316 const size_t *key_length,
317 size_t number_of_keys,
318 memcached_execute_fn *callback,
319 void *context,
320 unsigned int number_of_callbacks)
321 {
322 if ((ptr->flags.binary_protocol) == 0)
323 return MEMCACHED_NOT_SUPPORTED;
324
325 memcached_return_t rc;
326 memcached_callback_st *original_callbacks= ptr->callbacks;
327 memcached_callback_st cb= {
328 .callback= callback,
329 .context= context,
330 .number_of_callback= number_of_callbacks
331 };
332
333 ptr->callbacks= &cb;
334 rc= memcached_mget_by_key(ptr, master_key, master_key_length, keys,
335 key_length, number_of_keys);
336 ptr->callbacks= original_callbacks;
337 return rc;
338 }
339
340 static memcached_return_t simple_binary_mget(memcached_st *ptr,
341 uint32_t master_server_key,
342 bool is_master_key_set,
343 const char * const *keys,
344 const size_t *key_length,
345 size_t number_of_keys, bool mget_mode)
346 {
347 memcached_return_t rc= MEMCACHED_NOTFOUND;
348
349 int flush= number_of_keys == 1;
350
351 /*
352 If a server fails we warn about errors and start all over with sending keys
353 to the server.
354 */
355 for (uint32_t x= 0; x < number_of_keys; ++x)
356 {
357 uint32_t server_key;
358 memcached_server_write_instance_st instance;
359
360 if (is_master_key_set)
361 {
362 server_key= master_server_key;
363 }
364 else
365 {
366 server_key= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
367 }
368
369 instance= memcached_server_instance_fetch(ptr, server_key);
370
371 if (memcached_server_response_count(instance) == 0)
372 {
373 rc= memcached_connect(instance);
374 if (rc != MEMCACHED_SUCCESS)
375 continue;
376 }
377
378 protocol_binary_request_getk request= {.bytes= {0}};
379 request.message.header.request.magic= PROTOCOL_BINARY_REQ;
380 if (mget_mode)
381 request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETKQ;
382 else
383 request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
384
385 memcached_return_t vk;
386 vk= memcached_validate_key_length(key_length[x],
387 ptr->flags.binary_protocol);
388 unlikely (vk != MEMCACHED_SUCCESS)
389 {
390 if (x > 0)
391 {
392 memcached_io_reset(instance);
393 }
394
395 return vk;
396 }
397
398 request.message.header.request.keylen= htons((uint16_t)(key_length[x] + ptr->prefix_key_length));
399 request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
400 request.message.header.request.bodylen= htonl((uint32_t)( key_length[x] + ptr->prefix_key_length));
401
402 struct __write_vector_st vector[]=
403 {
404 { .length= sizeof(request.bytes), .buffer= request.bytes },
405 { .length= ptr->prefix_key_length, .buffer= ptr->prefix_key },
406 { .length= key_length[x], .buffer= keys[x] }
407 };
408
409 if (memcached_io_writev(instance, vector, 3, flush) == -1)
410 {
411 memcached_server_response_reset(instance);
412 rc= MEMCACHED_SOME_ERRORS;
413 continue;
414 }
415
416 /* We just want one pending response per server */
417 memcached_server_response_reset(instance);
418 memcached_server_response_increment(instance);
419 if ((x > 0 && x == ptr->io_key_prefetch) && memcached_flush_buffers(ptr) != MEMCACHED_SUCCESS)
420 {
421 rc= MEMCACHED_SOME_ERRORS;
422 }
423 }
424
425 if (mget_mode)
426 {
427 /*
428 * Send a noop command to flush the buffers
429 */
430 protocol_binary_request_noop request= {.bytes= {0}};
431 request.message.header.request.magic= PROTOCOL_BINARY_REQ;
432 request.message.header.request.opcode= PROTOCOL_BINARY_CMD_NOOP;
433 request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
434
435 for (uint32_t x= 0; x < memcached_server_count(ptr); ++x)
436 {
437 memcached_server_write_instance_st instance=
438 memcached_server_instance_fetch(ptr, x);
439
440 if (memcached_server_response_count(instance))
441 {
442 if (memcached_io_write(instance, NULL, 0, true) == -1)
443 {
444 memcached_server_response_reset(instance);
445 memcached_io_reset(instance);
446 rc= MEMCACHED_SOME_ERRORS;
447 }
448
449 if (memcached_io_write(instance, request.bytes,
450 sizeof(request.bytes), true) == -1)
451 {
452 memcached_server_response_reset(instance);
453 memcached_io_reset(instance);
454 rc= MEMCACHED_SOME_ERRORS;
455 }
456 }
457 }
458 }
459
460
461 return rc;
462 }
463
464 static memcached_return_t replication_binary_mget(memcached_st *ptr,
465 uint32_t* hash,
466 bool* dead_servers,
467 const char *const *keys,
468 const size_t *key_length,
469 size_t number_of_keys)
470 {
471 memcached_return_t rc= MEMCACHED_NOTFOUND;
472 uint32_t start= 0;
473 uint64_t randomize_read= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_RANDOMIZE_REPLICA_READ);
474
475 if (randomize_read)
476 start= (uint32_t)random() % (uint32_t)(ptr->number_of_replicas + 1);
477
478 /* Loop for each replica */
479 for (uint32_t replica= 0; replica <= ptr->number_of_replicas; ++replica)
480 {
481 bool success= true;
482
483 for (uint32_t x= 0; x < number_of_keys; ++x)
484 {
485 memcached_server_write_instance_st instance;
486
487 if (hash[x] == memcached_server_count(ptr))
488 continue; /* Already successfully sent */
489
490 uint32_t server= hash[x] + replica;
491
492 /* In case of randomized reads */
493 if (randomize_read && ((server + start) <= (hash[x] + ptr->number_of_replicas)))
494 server += start;
495
496 while (server >= memcached_server_count(ptr))
497 server -= memcached_server_count(ptr);
498
499 if (dead_servers[server])
500 continue;
501
502 instance= memcached_server_instance_fetch(ptr, server);
503
504 if (memcached_server_response_count(instance) == 0)
505 {
506 rc= memcached_connect(instance);
507 if (rc != MEMCACHED_SUCCESS)
508 {
509 memcached_io_reset(instance);
510 dead_servers[server]= true;
511 success= false;
512 continue;
513 }
514 }
515
516 protocol_binary_request_getk request= {
517 .message.header.request= {
518 .magic= PROTOCOL_BINARY_REQ,
519 .opcode= PROTOCOL_BINARY_CMD_GETK,
520 .keylen= htons((uint16_t)(key_length[x] + ptr->prefix_key_length)),
521 .datatype= PROTOCOL_BINARY_RAW_BYTES,
522 .bodylen= htonl((uint32_t)(key_length[x] + ptr->prefix_key_length))
523 }
524 };
525
526 /*
527 * We need to disable buffering to actually know that the request was
528 * successfully sent to the server (so that we should expect a result
529 * back). It would be nice to do this in buffered mode, but then it
530 * would be complex to handle all error situations if we got to send
531 * some of the messages, and then we failed on writing out some others
532 * and we used the callback interface from memcached_mget_execute so
533 * that we might have processed some of the responses etc. For now,
534 * just make sure we work _correctly_
535 */
536 struct __write_vector_st vector[]=
537 {
538 { .length= sizeof(request.bytes), .buffer= request.bytes },
539 { .length= ptr->prefix_key_length, .buffer= ptr->prefix_key },
540 { .length= key_length[x], .buffer= keys[x] }
541 };
542
543 if (memcached_io_writev(instance, vector, 3, true) == -1)
544 {
545 memcached_io_reset(instance);
546 dead_servers[server]= true;
547 success= false;
548 continue;
549 }
550
551 memcached_server_response_increment(instance);
552 hash[x]= memcached_server_count(ptr);
553 }
554
555 if (success)
556 break;
557 }
558
559 return rc;
560 }
561
562 static memcached_return_t binary_mget_by_key(memcached_st *ptr,
563 uint32_t master_server_key,
564 bool is_master_key_set,
565 const char * const *keys,
566 const size_t *key_length,
567 size_t number_of_keys,
568 bool mget_mode)
569 {
570 memcached_return_t rc;
571
572 if (ptr->number_of_replicas == 0)
573 {
574 rc= simple_binary_mget(ptr, master_server_key, is_master_key_set,
575 keys, key_length, number_of_keys, mget_mode);
576 }
577 else
578 {
579 uint32_t* hash;
580 bool* dead_servers;
581
582 hash= libmemcached_malloc(ptr, sizeof(uint32_t) * number_of_keys);
583 dead_servers= libmemcached_calloc(ptr, memcached_server_count(ptr), sizeof(bool));
584
585 if (hash == NULL || dead_servers == NULL)
586 {
587 libmemcached_free(ptr, hash);
588 libmemcached_free(ptr, dead_servers);
589 return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
590 }
591
592 if (is_master_key_set)
593 {
594 for (size_t x= 0; x < number_of_keys; x++)
595 {
596 hash[x]= master_server_key;
597 }
598 }
599 else
600 {
601 for (size_t x= 0; x < number_of_keys; x++)
602 {
603 hash[x]= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
604 }
605 }
606
607 rc= replication_binary_mget(ptr, hash, dead_servers, keys,
608 key_length, number_of_keys);
609
610 libmemcached_free(ptr, hash);
611 libmemcached_free(ptr, dead_servers);
612
613 return MEMCACHED_SUCCESS;
614 }
615
616 return rc;
617 }