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