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