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