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