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