Small style cleanup.
[m6w6/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 unsigned int x;
148 memcached_return_t rc= MEMCACHED_NOTFOUND;
149 const char *get_command= "get ";
150 uint8_t get_command_length= 4;
151 unsigned int master_server_key= (unsigned int)-1; /* 0 is a valid server id! */
152 bool is_master_key_set= false;
153
154 unlikely (ptr->flags.use_udp)
155 return MEMCACHED_NOT_SUPPORTED;
156
157 LIBMEMCACHED_MEMCACHED_MGET_START();
158
159 if (number_of_keys == 0)
160 return MEMCACHED_NOTFOUND;
161
162 if (memcached_server_count(ptr) == 0)
163 return MEMCACHED_NO_SERVERS;
164
165 if (ptr->flags.verify_key && (memcached_key_test(keys, key_length, number_of_keys) == MEMCACHED_BAD_KEY_PROVIDED))
166 return MEMCACHED_BAD_KEY_PROVIDED;
167
168 if (master_key && master_key_length)
169 {
170 if (ptr->flags.verify_key && (memcached_key_test((const char * const *)&master_key, &master_key_length, 1) == MEMCACHED_BAD_KEY_PROVIDED))
171 return MEMCACHED_BAD_KEY_PROVIDED;
172 master_server_key= memcached_generate_hash(ptr, master_key, master_key_length);
173 is_master_key_set= true;
174 }
175
176 /*
177 Here is where we pay for the non-block API. We need to remove any data sitting
178 in the queue before we start our get.
179
180 It might be optimum to bounce the connection if count > some number.
181 */
182 for (x= 0; x < memcached_server_count(ptr); x++)
183 {
184 memcached_server_instance_st *instance=
185 memcached_server_instance_fetch(ptr, x);
186
187 if (memcached_server_response_count(instance))
188 {
189 char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
190
191 if (ptr->flags.no_block)
192 (void)memcached_io_write(instance, NULL, 0, 1);
193
194 while(memcached_server_response_count(instance))
195 (void)memcached_response(instance, buffer, MEMCACHED_DEFAULT_COMMAND_SIZE, &ptr->result);
196 }
197 }
198
199 if (ptr->flags.binary_protocol)
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 if (ptr->flags.support_cas)
204 {
205 get_command= "gets ";
206 get_command_length= 5;
207 }
208
209 /*
210 If a server fails we warn about errors and start all over with sending keys
211 to the server.
212 */
213 for (x= 0; x < number_of_keys; x++)
214 {
215 memcached_server_instance_st *instance;
216 uint32_t server_key;
217
218 if (is_master_key_set)
219 {
220 server_key= master_server_key;
221 }
222 else
223 {
224 server_key= memcached_generate_hash(ptr, keys[x], key_length[x]);
225 }
226
227 instance= memcached_server_instance_fetch(ptr, server_key);
228
229 if (memcached_server_response_count(instance) == 0)
230 {
231 rc= memcached_connect(instance);
232
233 if (rc != MEMCACHED_SUCCESS)
234 continue;
235
236 if ((memcached_io_write(instance, get_command, get_command_length, 0)) == -1)
237 {
238 rc= MEMCACHED_SOME_ERRORS;
239 continue;
240 }
241 WATCHPOINT_ASSERT(instance->cursor_active == 0);
242 memcached_server_response_increment(instance);
243 WATCHPOINT_ASSERT(instance->cursor_active == 1);
244 }
245
246 /* Only called when we have a prefix key */
247 if (ptr->prefix_key_length)
248 {
249 if ((memcached_io_write(instance, ptr->prefix_key, ptr->prefix_key_length, 0)) == -1)
250 {
251 memcached_server_response_reset(instance);
252 rc= MEMCACHED_SOME_ERRORS;
253 continue;
254 }
255 }
256
257 if ((memcached_io_write(instance, keys[x], key_length[x], 0)) == -1)
258 {
259 memcached_server_response_reset(instance);
260 rc= MEMCACHED_SOME_ERRORS;
261 continue;
262 }
263
264 if ((memcached_io_write(instance, " ", 1, 0)) == -1)
265 {
266 memcached_server_response_reset(instance);
267 rc= MEMCACHED_SOME_ERRORS;
268 continue;
269 }
270 }
271
272 /*
273 Should we muddle on if some servers are dead?
274 */
275 for (x= 0; x < memcached_server_count(ptr); x++)
276 {
277 memcached_server_instance_st *instance=
278 memcached_server_instance_fetch(ptr, x);
279
280 if (memcached_server_response_count(instance))
281 {
282 /* We need to do something about non-connnected hosts in the future */
283 if ((memcached_io_write(instance, "\r\n", 2, 1)) == -1)
284 {
285 rc= MEMCACHED_SOME_ERRORS;
286 }
287 }
288 }
289
290 LIBMEMCACHED_MEMCACHED_MGET_END();
291 return rc;
292 }
293
294 memcached_return_t memcached_mget_by_key(memcached_st *ptr,
295 const char *master_key,
296 size_t master_key_length,
297 const char * const *keys,
298 const size_t *key_length,
299 size_t number_of_keys)
300 {
301 return memcached_mget_by_key_real(ptr, master_key, master_key_length, keys,
302 key_length, number_of_keys, true);
303 }
304
305 memcached_return_t memcached_mget_execute(memcached_st *ptr,
306 const char * const *keys,
307 const size_t *key_length,
308 size_t number_of_keys,
309 memcached_execute_fn *callback,
310 void *context,
311 unsigned int number_of_callbacks)
312 {
313 return memcached_mget_execute_by_key(ptr, NULL, 0, keys, key_length,
314 number_of_keys, callback,
315 context, number_of_callbacks);
316 }
317
318 memcached_return_t memcached_mget_execute_by_key(memcached_st *ptr,
319 const char *master_key,
320 size_t master_key_length,
321 const char * const *keys,
322 const size_t *key_length,
323 size_t number_of_keys,
324 memcached_execute_fn *callback,
325 void *context,
326 unsigned int number_of_callbacks)
327 {
328 if ((ptr->flags.binary_protocol) == 0)
329 return MEMCACHED_NOT_SUPPORTED;
330
331 memcached_return_t rc;
332 memcached_callback_st *original_callbacks= ptr->callbacks;
333 memcached_callback_st cb= {
334 .callback= callback,
335 .context= context,
336 .number_of_callback= number_of_callbacks
337 };
338
339 ptr->callbacks= &cb;
340 rc= memcached_mget_by_key(ptr, master_key, master_key_length, keys,
341 key_length, number_of_keys);
342 ptr->callbacks= original_callbacks;
343 return rc;
344 }
345
346 static memcached_return_t simple_binary_mget(memcached_st *ptr,
347 uint32_t master_server_key,
348 bool is_master_key_set,
349 const char * const *keys,
350 const size_t *key_length,
351 size_t number_of_keys, bool mget_mode)
352 {
353 memcached_return_t rc= MEMCACHED_NOTFOUND;
354 uint32_t x;
355
356 int flush= number_of_keys == 1;
357
358 /*
359 If a server fails we warn about errors and start all over with sending keys
360 to the server.
361 */
362 for (x= 0; x < number_of_keys; x++)
363 {
364 uint32_t server_key;
365 memcached_server_instance_st *instance;
366
367 if (is_master_key_set)
368 {
369 server_key= master_server_key;
370 }
371 else
372 {
373 server_key= memcached_generate_hash(ptr, keys[x], key_length[x]);
374 }
375
376 instance= memcached_server_instance_fetch(ptr, server_key);
377
378 if (memcached_server_response_count(instance) == 0)
379 {
380 rc= memcached_connect(instance);
381 if (rc != MEMCACHED_SUCCESS)
382 continue;
383 }
384
385 protocol_binary_request_getk request= {.bytes= {0}};
386 request.message.header.request.magic= PROTOCOL_BINARY_REQ;
387 if (mget_mode)
388 request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETKQ;
389 else
390 request.message.header.request.opcode= PROTOCOL_BINARY_CMD_GETK;
391
392 memcached_return_t vk;
393 vk= memcached_validate_key_length(key_length[x],
394 ptr->flags.binary_protocol);
395 unlikely (vk != MEMCACHED_SUCCESS)
396 {
397 if (x > 0)
398 {
399 memcached_io_reset(instance);
400 }
401
402 return vk;
403 }
404
405 request.message.header.request.keylen= htons((uint16_t)key_length[x]);
406 request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
407 request.message.header.request.bodylen= htonl((uint32_t) key_length[x]);
408
409 if ((memcached_io_write(instance, request.bytes,
410 sizeof(request.bytes), 0) == -1) ||
411 (memcached_io_write(instance, keys[x],
412 key_length[x], (char) flush) == -1))
413 {
414 memcached_server_response_reset(instance);
415 rc= MEMCACHED_SOME_ERRORS;
416 continue;
417 }
418
419 /* We just want one pending response per server */
420 memcached_server_response_reset(instance);
421 memcached_server_response_increment(instance);
422 if ((x > 0 && x == ptr->io_key_prefetch) &&
423 memcached_flush_buffers(ptr) != MEMCACHED_SUCCESS)
424 rc= MEMCACHED_SOME_ERRORS;
425 }
426
427 if (mget_mode)
428 {
429 /*
430 * Send a noop command to flush the buffers
431 */
432 protocol_binary_request_noop request= {.bytes= {0}};
433 request.message.header.request.magic= PROTOCOL_BINARY_REQ;
434 request.message.header.request.opcode= PROTOCOL_BINARY_CMD_NOOP;
435 request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
436
437 for (x= 0; x < memcached_server_count(ptr); x++)
438 {
439 memcached_server_instance_st *instance=
440 memcached_server_instance_fetch(ptr, x);
441
442 if (memcached_server_response_count(instance))
443 {
444 if (memcached_io_write(instance, NULL, 0, 1) == -1)
445 {
446 memcached_server_response_reset(instance);
447 memcached_io_reset(instance);
448 rc= MEMCACHED_SOME_ERRORS;
449 }
450
451 if (memcached_io_write(instance, request.bytes,
452 sizeof(request.bytes), 1) == -1)
453 {
454 memcached_server_response_reset(instance);
455 memcached_io_reset(instance);
456 rc= MEMCACHED_SOME_ERRORS;
457 }
458 }
459 }
460 }
461
462
463 return rc;
464 }
465
466 static memcached_return_t replication_binary_mget(memcached_st *ptr,
467 uint32_t* hash,
468 bool* dead_servers,
469 const char *const *keys,
470 const size_t *key_length,
471 size_t number_of_keys)
472 {
473 memcached_return_t rc= MEMCACHED_NOTFOUND;
474 uint32_t x, start= 0;
475 uint64_t randomize_read= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_RANDOMIZE_REPLICA_READ);
476
477 if (randomize_read)
478 start= (uint32_t)random() % (uint32_t)(ptr->number_of_replicas + 1);
479
480 /* Loop for each replica */
481 for (uint32_t replica= 0; replica <= ptr->number_of_replicas; ++replica)
482 {
483 bool success= true;
484
485 for (x= 0; x < number_of_keys; ++x)
486 {
487 memcached_server_instance_st *instance;
488
489 if (hash[x] == memcached_server_count(ptr))
490 continue; /* Already successfully sent */
491
492 uint32_t server= hash[x] + replica;
493
494 /* In case of randomized reads */
495 if (randomize_read && ((server + start) <= (hash[x] + ptr->number_of_replicas)))
496 server += start;
497
498 while (server >= memcached_server_count(ptr))
499 server -= memcached_server_count(ptr);
500
501 if (dead_servers[server])
502 continue;
503
504 instance= memcached_server_instance_fetch(ptr, server);
505
506 if (memcached_server_response_count(instance) == 0)
507 {
508 rc= memcached_connect(instance);
509 if (rc != MEMCACHED_SUCCESS)
510 {
511 memcached_io_reset(instance);
512 dead_servers[server]= true;
513 success= false;
514 continue;
515 }
516 }
517
518 protocol_binary_request_getk request= {
519 .message.header.request= {
520 .magic= PROTOCOL_BINARY_REQ,
521 .opcode= PROTOCOL_BINARY_CMD_GETK,
522 .keylen= htons((uint16_t)key_length[x]),
523 .datatype= PROTOCOL_BINARY_RAW_BYTES,
524 .bodylen= htonl((uint32_t)key_length[x])
525 }
526 };
527
528 /*
529 * We need to disable buffering to actually know that the request was
530 * successfully sent to the server (so that we should expect a result
531 * back). It would be nice to do this in buffered mode, but then it
532 * would be complex to handle all error situations if we got to send
533 * some of the messages, and then we failed on writing out some others
534 * and we used the callback interface from memcached_mget_execute so
535 * that we might have processed some of the responses etc. For now,
536 * just make sure we work _correctly_
537 */
538 if ((memcached_io_write(instance, request.bytes,
539 sizeof(request.bytes), 0) == -1) ||
540 (memcached_io_write(instance, keys[x],
541 key_length[x], 1) == -1))
542 {
543 memcached_io_reset(instance);
544 dead_servers[server]= true;
545 success= false;
546 continue;
547 }
548
549 memcached_server_response_increment(instance);
550 hash[x]= memcached_server_count(ptr);
551 }
552
553 if (success)
554 break;
555 }
556
557 return rc;
558 }
559
560 static memcached_return_t binary_mget_by_key(memcached_st *ptr,
561 uint32_t master_server_key,
562 bool is_master_key_set,
563 const char * const *keys,
564 const size_t *key_length,
565 size_t number_of_keys,
566 bool mget_mode)
567 {
568 memcached_return_t rc;
569
570 if (ptr->number_of_replicas == 0)
571 {
572 rc= simple_binary_mget(ptr, master_server_key, is_master_key_set,
573 keys, key_length, number_of_keys, mget_mode);
574 }
575 else
576 {
577 uint32_t* hash;
578 bool* dead_servers;
579
580 hash= ptr->call_malloc(ptr, sizeof(uint32_t) * number_of_keys);
581 dead_servers= ptr->call_calloc(ptr, memcached_server_count(ptr), sizeof(bool));
582
583 if (hash == NULL || dead_servers == NULL)
584 {
585 ptr->call_free(ptr, hash);
586 ptr->call_free(ptr, dead_servers);
587 return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
588 }
589
590 if (is_master_key_set)
591 for (unsigned int x= 0; x < number_of_keys; x++)
592 hash[x]= master_server_key;
593 else
594 for (unsigned int x= 0; x < number_of_keys; x++)
595 hash[x]= memcached_generate_hash(ptr, keys[x], key_length[x]);
596
597 rc= replication_binary_mget(ptr, hash, dead_servers, keys,
598 key_length, number_of_keys);
599
600 ptr->call_free(ptr, hash);
601 ptr->call_free(ptr, dead_servers);
602
603 return MEMCACHED_SUCCESS;
604 }
605
606 return rc;
607 }