Merge in support for prefix keys in the binary protoocol.
[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 uint32_t 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_with_redistribution(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_write_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, true);
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_write_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_with_redistribution(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, false)) == -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, false)) == -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], false)) == -1)
258 {
259 memcached_server_response_reset(instance);
260 rc= MEMCACHED_SOME_ERRORS;
261 continue;
262 }
263
264 if ((memcached_io_write(instance, " ", 1, false)) == -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_write_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, true)) == -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_write_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_with_redistribution(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] + ptr->prefix_key_length));
406 request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
407 request.message.header.request.bodylen= htonl((uint32_t)( key_length[x] + ptr->prefix_key_length));
408
409 if (memcached_io_write(instance, request.bytes, sizeof(request.bytes), false) == -1)
410 {
411 memcached_server_response_reset(instance);
412 rc= MEMCACHED_SOME_ERRORS;
413 continue;
414 }
415
416 if (ptr->prefix_key_length)
417 {
418 if (memcached_io_write(instance, ptr->prefix_key, ptr->prefix_key_length, false) == -1)
419 {
420 memcached_server_response_reset(instance);
421 rc= MEMCACHED_SOME_ERRORS;
422 continue;
423 }
424 }
425
426 if (memcached_io_write(instance, keys[x], key_length[x], flush) == -1)
427 {
428 memcached_server_response_reset(instance);
429 rc= MEMCACHED_SOME_ERRORS;
430 continue;
431 }
432
433 /* We just want one pending response per server */
434 memcached_server_response_reset(instance);
435 memcached_server_response_increment(instance);
436 if ((x > 0 && x == ptr->io_key_prefetch) && memcached_flush_buffers(ptr) != MEMCACHED_SUCCESS)
437 {
438 rc= MEMCACHED_SOME_ERRORS;
439 }
440 }
441
442 if (mget_mode)
443 {
444 /*
445 * Send a noop command to flush the buffers
446 */
447 protocol_binary_request_noop request= {.bytes= {0}};
448 request.message.header.request.magic= PROTOCOL_BINARY_REQ;
449 request.message.header.request.opcode= PROTOCOL_BINARY_CMD_NOOP;
450 request.message.header.request.datatype= PROTOCOL_BINARY_RAW_BYTES;
451
452 for (x= 0; x < memcached_server_count(ptr); x++)
453 {
454 memcached_server_write_instance_st instance=
455 memcached_server_instance_fetch(ptr, x);
456
457 if (memcached_server_response_count(instance))
458 {
459 if (memcached_io_write(instance, NULL, 0, true) == -1)
460 {
461 memcached_server_response_reset(instance);
462 memcached_io_reset(instance);
463 rc= MEMCACHED_SOME_ERRORS;
464 }
465
466 if (memcached_io_write(instance, request.bytes,
467 sizeof(request.bytes), true) == -1)
468 {
469 memcached_server_response_reset(instance);
470 memcached_io_reset(instance);
471 rc= MEMCACHED_SOME_ERRORS;
472 }
473 }
474 }
475 }
476
477
478 return rc;
479 }
480
481 static memcached_return_t replication_binary_mget(memcached_st *ptr,
482 uint32_t* hash,
483 bool* dead_servers,
484 const char *const *keys,
485 const size_t *key_length,
486 size_t number_of_keys)
487 {
488 memcached_return_t rc= MEMCACHED_NOTFOUND;
489 uint32_t x, start= 0;
490 uint64_t randomize_read= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_RANDOMIZE_REPLICA_READ);
491
492 if (randomize_read)
493 start= (uint32_t)random() % (uint32_t)(ptr->number_of_replicas + 1);
494
495 /* Loop for each replica */
496 for (uint32_t replica= 0; replica <= ptr->number_of_replicas; ++replica)
497 {
498 bool success= true;
499
500 for (x= 0; x < number_of_keys; ++x)
501 {
502 memcached_server_write_instance_st instance;
503
504 if (hash[x] == memcached_server_count(ptr))
505 continue; /* Already successfully sent */
506
507 uint32_t server= hash[x] + replica;
508
509 /* In case of randomized reads */
510 if (randomize_read && ((server + start) <= (hash[x] + ptr->number_of_replicas)))
511 server += start;
512
513 while (server >= memcached_server_count(ptr))
514 server -= memcached_server_count(ptr);
515
516 if (dead_servers[server])
517 continue;
518
519 instance= memcached_server_instance_fetch(ptr, server);
520
521 if (memcached_server_response_count(instance) == 0)
522 {
523 rc= memcached_connect(instance);
524 if (rc != MEMCACHED_SUCCESS)
525 {
526 memcached_io_reset(instance);
527 dead_servers[server]= true;
528 success= false;
529 continue;
530 }
531 }
532
533 protocol_binary_request_getk request= {
534 .message.header.request= {
535 .magic= PROTOCOL_BINARY_REQ,
536 .opcode= PROTOCOL_BINARY_CMD_GETK,
537 .keylen= htons((uint16_t)(key_length[x] + ptr->prefix_key_length)),
538 .datatype= PROTOCOL_BINARY_RAW_BYTES,
539 .bodylen= htonl((uint32_t)(key_length[x] + ptr->prefix_key_length))
540 }
541 };
542
543 /*
544 * We need to disable buffering to actually know that the request was
545 * successfully sent to the server (so that we should expect a result
546 * back). It would be nice to do this in buffered mode, but then it
547 * would be complex to handle all error situations if we got to send
548 * some of the messages, and then we failed on writing out some others
549 * and we used the callback interface from memcached_mget_execute so
550 * that we might have processed some of the responses etc. For now,
551 * just make sure we work _correctly_
552 */
553 if ((memcached_io_write(instance, request.bytes, sizeof(request.bytes), false) == -1) ||
554 (memcached_io_write(instance, ptr->prefix_key, ptr->prefix_key_length, false) == -1) ||
555 (memcached_io_write(instance, keys[x], key_length[x], true) == -1))
556 {
557 memcached_io_reset(instance);
558 dead_servers[server]= true;
559 success= false;
560 continue;
561 }
562
563 memcached_server_response_increment(instance);
564 hash[x]= memcached_server_count(ptr);
565 }
566
567 if (success)
568 break;
569 }
570
571 return rc;
572 }
573
574 static memcached_return_t binary_mget_by_key(memcached_st *ptr,
575 uint32_t master_server_key,
576 bool is_master_key_set,
577 const char * const *keys,
578 const size_t *key_length,
579 size_t number_of_keys,
580 bool mget_mode)
581 {
582 memcached_return_t rc;
583
584 if (ptr->number_of_replicas == 0)
585 {
586 rc= simple_binary_mget(ptr, master_server_key, is_master_key_set,
587 keys, key_length, number_of_keys, mget_mode);
588 }
589 else
590 {
591 uint32_t* hash;
592 bool* dead_servers;
593
594 hash= libmemcached_malloc(ptr, sizeof(uint32_t) * number_of_keys);
595 dead_servers= libmemcached_calloc(ptr, memcached_server_count(ptr), sizeof(bool));
596
597 if (hash == NULL || dead_servers == NULL)
598 {
599 libmemcached_free(ptr, hash);
600 libmemcached_free(ptr, dead_servers);
601 return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
602 }
603
604 if (is_master_key_set)
605 {
606 for (size_t x= 0; x < number_of_keys; x++)
607 {
608 hash[x]= master_server_key;
609 }
610 }
611 else
612 {
613 for (size_t x= 0; x < number_of_keys; x++)
614 {
615 hash[x]= memcached_generate_hash_with_redistribution(ptr, keys[x], key_length[x]);
616 }
617 }
618
619 rc= replication_binary_mget(ptr, hash, dead_servers, keys,
620 key_length, number_of_keys);
621
622 libmemcached_free(ptr, hash);
623 libmemcached_free(ptr, dead_servers);
624
625 return MEMCACHED_SUCCESS;
626 }
627
628 return rc;
629 }