projects / m6w6 / libmemcached / blob
? search:


f4375ae87bea7c318859d550fcd8b19b0991615b
[m6w6/libmemcached] / contrib / bin / memaslap / ms_conn.c
1 /*
2 +--------------------------------------------------------------------+
3 | libmemcached - C/C++ Client Library for memcached |
4 +--------------------------------------------------------------------+
5 | Redistribution and use in source and binary forms, with or without |
6 | modification, are permitted under the terms of the BSD license. |
7 | You should have received a copy of the license in a bundled file |
8 | named LICENSE; in case you did not receive a copy you can review |
9 | the terms online at: https://opensource.org/licenses/BSD-3-Clause |
10 +--------------------------------------------------------------------+
11 | Copyright (c) 2006-2014 Brian Aker https://datadifferential.com/ |
12 | Copyright (c) 2020 Michael Wallner <mike@php.net> |
13 +--------------------------------------------------------------------+
14 */
15
16 #include "mem_config.h"
17
18 #include <stdio.h>
19 #include <inttypes.h>
20 #include <limits.h>
21 #include <sys/uio.h>
22 #include <event.h>
23 #include <fcntl.h>
24 #include <netinet/tcp.h>
25 #include <netinet/in.h>
26
27 #if defined(HAVE_ARPA_INET_H)
28 # include <arpa/inet.h>
29 #endif
30
31 #if defined(HAVE_SYS_TIME_H)
32 # include <sys/time.h>
33 #endif
34
35 #if defined(HAVE_TIME_H)
36 # include <time.h>
37 #endif
38
39 #include "ms_setting.h"
40 #include "ms_thread.h"
41 #include "ms_atomic.h"
42
43 #ifdef linux
44 /* /usr/include/netinet/in.h defines macros from ntohs() to _bswap_nn to
45 * optimize the conversion functions, but the prototypes generate warnings
46 * from gcc. The conversion methods isn't the bottleneck for my app, so
47 * just remove the warnings by undef'ing the optimization ..
48 */
49 # undef ntohs
50 # undef ntohl
51 # undef htons
52 # undef htonl
53 #endif
54
55 /* for network write */
56 #define TRANSMIT_COMPLETE 0
57 #define TRANSMIT_INCOMPLETE 1
58 #define TRANSMIT_SOFT_ERROR 2
59 #define TRANSMIT_HARD_ERROR 3
60
61 /* for generating key */
62 #define KEY_PREFIX_BASE 0x1010101010101010 /* not include ' ' '\r' '\n' '\0' */
63 #define KEY_PREFIX_MASK 0x1010101010101010
64
65 /* For parse the value length return by server */
66 #define KEY_TOKEN 1
67 #define VALUELEN_TOKEN 3
68
69 /* global increasing counter, to ensure the key prefix unique */
70 static uint64_t key_prefix_seq = KEY_PREFIX_BASE;
71
72 /* global increasing counter, generating request id for UDP */
73 static ATOMIC uint32_t udp_request_id = 0;
74
75 extern pthread_key_t ms_thread_key;
76
77 /* generate upd request id */
78 static uint32_t ms_get_udp_request_id(void);
79
80 /* connect initialize */
81 static void ms_task_init(ms_conn_t *c);
82 static int ms_conn_udp_init(ms_conn_t *c, const bool is_udp);
83 static int ms_conn_sock_init(ms_conn_t *c);
84 static int ms_conn_event_init(ms_conn_t *c);
85 static int ms_conn_init(ms_conn_t *c, const int init_state, const int read_buffer_size,
86 const bool is_udp);
87 static void ms_warmup_num_init(ms_conn_t *c);
88 static int ms_item_win_init(ms_conn_t *c);
89
90 /* connection close */
91 void ms_conn_free(ms_conn_t *c);
92 static void ms_conn_close(ms_conn_t *c);
93
94 /* create network connection */
95 static int ms_new_socket(struct addrinfo *ai);
96 static void ms_maximize_sndbuf(const int sfd);
97 static int ms_network_connect(ms_conn_t *c, char *srv_host_name, const int srv_port,
98 const bool is_udp, int *ret_sfd);
99 static int ms_reconn(ms_conn_t *c);
100
101 /* read and parse */
102 static int ms_tokenize_command(char *command, token_t *tokens, const int max_tokens);
103 static int ms_ascii_process_line(ms_conn_t *c, char *command);
104 static int ms_try_read_line(ms_conn_t *c);
105 static int ms_sort_udp_packet(ms_conn_t *c, char *buf, int rbytes);
106 static int ms_udp_read(ms_conn_t *c, char *buf, int len);
107 static int ms_try_read_network(ms_conn_t *c);
108 static void ms_verify_value(ms_conn_t *c, ms_mlget_task_item_t *mlget_item, char *value, int vlen);
109 static void ms_ascii_complete_nread(ms_conn_t *c);
110 static void ms_bin_complete_nread(ms_conn_t *c);
111 static void ms_complete_nread(ms_conn_t *c);
112
113 /* send functions */
114 static int ms_add_msghdr(ms_conn_t *c);
115 static int ms_ensure_iov_space(ms_conn_t *c);
116 static int ms_add_iov(ms_conn_t *c, const void *buf, int len);
117 static int ms_build_udp_headers(ms_conn_t *c);
118 static int ms_transmit(ms_conn_t *c);
119
120 /* status adjustment */
121 static void ms_conn_shrink(ms_conn_t *c);
122 static void ms_conn_set_state(ms_conn_t *c, int state);
123 static bool ms_update_event(ms_conn_t *c, const int new_flags);
124 static uint32_t ms_get_rep_sock_index(ms_conn_t *c, int cmd);
125 static uint32_t ms_get_next_sock_index(ms_conn_t *c);
126 static int ms_update_conn_sock_event(ms_conn_t *c);
127 static bool ms_need_yield(ms_conn_t *c);
128 static void ms_update_start_time(ms_conn_t *c);
129
130 /* main loop */
131 static void ms_drive_machine(ms_conn_t *c);
132 void ms_event_handler(const int fd, const short which, void *arg);
133
134 /* ascii protocol */
135 static int ms_build_ascii_write_buf_set(ms_conn_t *c, ms_task_item_t *item);
136 static int ms_build_ascii_write_buf_get(ms_conn_t *c, ms_task_item_t *item);
137 static int ms_build_ascii_write_buf_mlget(ms_conn_t *c);
138
139 /* binary protocol */
140 static int ms_bin_process_response(ms_conn_t *c);
141 static void ms_add_bin_header(ms_conn_t *c, uint8_t opcode, uint8_t hdr_len, uint16_t key_len,
142 uint32_t body_len);
143 static void ms_add_key_to_iov(ms_conn_t *c, ms_task_item_t *item);
144 static int ms_build_bin_write_buf_set(ms_conn_t *c, ms_task_item_t *item);
145 static int ms_build_bin_write_buf_get(ms_conn_t *c, ms_task_item_t *item);
146 static int ms_build_bin_write_buf_mlget(ms_conn_t *c);
147
148 /**
149 * each key has two parts, prefix and suffix. The suffix is a
150 * string random get form the character table. The prefix is a
151 * uint64_t variable. And the prefix must be unique. we use the
152 * prefix to identify a key. And the prefix can't include
153 * character ' ' '\r' '\n' '\0'.
154 *
155 * @return uint64_t
156 */
157 uint64_t ms_get_key_prefix(void) {
158 uint64_t key_prefix;
159
160 pthread_mutex_lock(&ms_global.seq_mutex);
161 key_prefix_seq |= KEY_PREFIX_MASK;
162 key_prefix = key_prefix_seq;
163 key_prefix_seq++;
164 pthread_mutex_unlock(&ms_global.seq_mutex);
165
166 return key_prefix;
167 } /* ms_get_key_prefix */
168
169 /**
170 * get an unique udp request id
171 *
172 * @return an unique UDP request id
173 */
174 static uint32_t ms_get_udp_request_id(void) {
175 return atomic_add_32_nv(&udp_request_id, 1);
176 }
177
178 /**
179 * initialize current task structure
180 *
181 * @param c, pointer of the concurrency
182 */
183 static void ms_task_init(ms_conn_t *c) {
184 c->curr_task.cmd = CMD_NULL;
185 c->curr_task.item = 0;
186 c->curr_task.verify = false;
187 c->curr_task.finish_verify = true;
188 c->curr_task.get_miss = true;
189
190 c->curr_task.get_opt = 0;
191 c->curr_task.set_opt = 0;
192 c->curr_task.cycle_undo_get = 0;
193 c->curr_task.cycle_undo_set = 0;
194 c->curr_task.verified_get = 0;
195 c->curr_task.overwrite_set = 0;
196 } /* ms_task_init */
197
198 /**
199 * initialize udp for the connection structure
200 *
201 * @param c, pointer of the concurrency
202 * @param is_udp, whether it's udp
203 *
204 * @return int, if success, return EXIT_SUCCESS, else return -1
205 */
206 static int ms_conn_udp_init(ms_conn_t *c, const bool is_udp) {
207 c->hdrbuf = 0;
208 c->rudpbuf = 0;
209 c->udppkt = 0;
210
211 c->rudpsize = UDP_DATA_BUFFER_SIZE;
212 c->hdrsize = 0;
213
214 c->rudpbytes = 0;
215 c->packets = 0;
216 c->recvpkt = 0;
217 c->pktcurr = 0;
218 c->ordcurr = 0;
219
220 c->udp = is_udp;
221
222 if (c->udp || (!c->udp && ms_setting.facebook_test)) {
223 c->rudpbuf = (char *) malloc((size_t) c->rudpsize);
224 c->udppkt = (ms_udppkt_t *) malloc(MAX_UDP_PACKET * sizeof(ms_udppkt_t));
225
226 if ((c->rudpbuf == NULL) || (c->udppkt == NULL)) {
227 if (c->rudpbuf)
228 free(c->rudpbuf);
229 if (c->udppkt)
230 free(c->udppkt);
231 fprintf(stderr, "malloc()\n");
232 return -1;
233 }
234 memset(c->udppkt, 0, MAX_UDP_PACKET * sizeof(ms_udppkt_t));
235 }
236
237 return EXIT_SUCCESS;
238 } /* ms_conn_udp_init */
239
240 /**
241 * initialize the connection structure
242 *
243 * @param c, pointer of the concurrency
244 * @param init_state, (conn_read, conn_write, conn_closing)
245 * @param read_buffer_size
246 * @param is_udp, whether it's udp
247 *
248 * @return int, if success, return EXIT_SUCCESS, else return -1
249 */
250 static int ms_conn_init(ms_conn_t *c, const int init_state, const int read_buffer_size,
251 const bool is_udp) {
252 assert(c);
253
254 c->rbuf = c->wbuf = 0;
255 c->iov = 0;
256 c->msglist = 0;
257
258 c->rsize = read_buffer_size;
259 c->wsize = WRITE_BUFFER_SIZE;
260 c->iovsize = IOV_LIST_INITIAL;
261 c->msgsize = MSG_LIST_INITIAL;
262
263 /* for replication, each connection need connect all the server */
264 if (ms_setting.rep_write_srv > 0) {
265 c->total_sfds = ms_setting.srv_cnt * ms_setting.sock_per_conn;
266 } else {
267 c->total_sfds = ms_setting.sock_per_conn;
268 }
269 c->alive_sfds = 0;
270
271 c->rbuf = (char *) malloc((size_t) c->rsize);
272 c->wbuf = (char *) malloc((size_t) c->wsize);
273 c->iov = (struct iovec *) malloc(sizeof(struct iovec) * (size_t) c->iovsize);
274 c->msglist = (struct msghdr *) malloc(sizeof(struct msghdr) * (size_t) c->msgsize);
275 if (ms_setting.mult_key_num > 1) {
276 c->mlget_task.mlget_item = (ms_mlget_task_item_t *) malloc(sizeof(ms_mlget_task_item_t)
277 * (size_t) ms_setting.mult_key_num);
278 }
279 c->tcpsfd = (int *) malloc((size_t) c->total_sfds * sizeof(int));
280
281 if ((c->rbuf == NULL) || (c->wbuf == NULL) || (c->iov == NULL) || (c->msglist == NULL)
282 || (c->tcpsfd == NULL)
283 || ((ms_setting.mult_key_num > 1) && (c->mlget_task.mlget_item == NULL)))
284 {
285 if (c->rbuf)
286 free(c->rbuf);
287 if (c->wbuf)
288 free(c->wbuf);
289 if (c->iov)
290 free(c->iov);
291 if (c->msglist)
292 free(c->msglist);
293 if (c->mlget_task.mlget_item)
294 free(c->mlget_task.mlget_item);
295 if (c->tcpsfd)
296 free(c->tcpsfd);
297 fprintf(stderr, "malloc()\n");
298 return -1;
299 }
300
301 c->state = init_state;
302 c->rvbytes = 0;
303 c->rbytes = 0;
304 c->rcurr = c->rbuf;
305 c->wcurr = c->wbuf;
306 c->iovused = 0;
307 c->msgcurr = 0;
308 c->msgused = 0;
309 c->cur_idx = c->total_sfds; /* default index is a invalid value */
310
311 c->ctnwrite = false;
312 c->readval = false;
313 c->change_sfd = false;
314
315 c->precmd.cmd = c->currcmd.cmd = CMD_NULL;
316 c->precmd.isfinish = true; /* default the previous command finished */
317 c->currcmd.isfinish = false;
318 c->precmd.retstat = c->currcmd.retstat = MCD_FAILURE;
319 c->precmd.key_prefix = c->currcmd.key_prefix = 0;
320
321 c->mlget_task.mlget_num = 0;
322 c->mlget_task.value_index = -1; /* default invalid value */
323
324 if (ms_setting.binary_prot_) {
325 c->protocol = binary_prot;
326 } else {
327 c->protocol = ascii_prot;
328 }
329
330 /* initialize udp */
331 if (ms_conn_udp_init(c, is_udp)) {
332 return -1;
333 }
334
335 /* initialize task */
336 ms_task_init(c);
337
338 if (!(ms_setting.facebook_test && is_udp)) {
339 atomic_add_32(&ms_stats.active_conns, 1);
340 }
341
342 return EXIT_SUCCESS;
343 } /* ms_conn_init */
344
345 /**
346 * when doing 100% get operation, it could preset some objects
347 * to warmup the server. this function is used to initialize the
348 * number of the objects to preset.
349 *
350 * @param c, pointer of the concurrency
351 */
352 static void ms_warmup_num_init(ms_conn_t *c) {
353 /* no set operation, preset all the items in the window */
354 if (ms_setting.cmd_distr[CMD_SET].cmd_prop < PROP_ERROR) {
355 c->warmup_num = c->win_size;
356 c->remain_warmup_num = c->warmup_num;
357 } else {
358 c->warmup_num = 0;
359 c->remain_warmup_num = c->warmup_num;
360 }
361 } /* ms_warmup_num_init */
362
363 /**
364 * each connection has an item window, this function initialize
365 * the window. The window is used to generate task.
366 *
367 * @param c, pointer of the concurrency
368 *
369 * @return int, if success, return EXIT_SUCCESS, else return -1
370 */
371 static int ms_item_win_init(ms_conn_t *c) {
372 ms_thread_t *ms_thread = pthread_getspecific(ms_thread_key);
373 int exp_cnt = 0;
374
375 c->win_size = (int) ms_setting.win_size;
376 c->set_cursor = 0;
377 c->exec_num = ms_thread->thread_ctx->exec_num_perconn;
378 c->remain_exec_num = c->exec_num;
379
380 c->item_win = (ms_task_item_t *) malloc(sizeof(ms_task_item_t) * (size_t) c->win_size);
381 if (c->item_win == NULL) {
382 fprintf(stderr, "Can't allocate task item array for conn.\n");
383 return -1;
384 }
385 memset(c->item_win, 0, sizeof(ms_task_item_t) * (size_t) c->win_size);
386
387 for (int i = 0; i < c->win_size; i++) {
388 c->item_win[i].key_size = (int) ms_setting.distr[i].key_size;
389 c->item_win[i].key_prefix = ms_get_key_prefix();
390 c->item_win[i].key_suffix_offset = ms_setting.distr[i].key_offset;
391 c->item_win[i].value_size = (int) ms_setting.distr[i].value_size;
392 c->item_win[i].value_offset = INVALID_OFFSET; /* default in invalid offset */
393 c->item_win[i].client_time = 0;
394
395 /* set expire time base on the proportion */
396 if (exp_cnt < ms_setting.exp_ver_per * i) {
397 c->item_win[i].exp_time = FIXED_EXPIRE_TIME;
398 exp_cnt++;
399 } else {
400 c->item_win[i].exp_time = 0;
401 }
402 }
403
404 ms_warmup_num_init(c);
405
406 return EXIT_SUCCESS;
407 } /* ms_item_win_init */
408
409 /**
410 * each connection structure can include one or more sock
411 * handlers. this function create these socks and connect the
412 * server(s).
413 *
414 * @param c, pointer of the concurrency
415 *
416 * @return int, if success, return EXIT_SUCCESS, else return -1
417 */
418 static int ms_conn_sock_init(ms_conn_t *c) {
419 ms_thread_t *ms_thread = pthread_getspecific(ms_thread_key);
420 uint32_t i;
421 int ret_sfd;
422 uint32_t srv_idx = 0;
423
424 assert(c);
425 assert(c->tcpsfd);
426
427 for (i = 0; i < c->total_sfds; i++) {
428 ret_sfd = 0;
429 if (ms_setting.rep_write_srv > 0) {
430 /* for replication, each connection need connect all the server */
431 srv_idx = i % ms_setting.srv_cnt;
432 } else {
433 /* all the connections in a thread connects the same server */
434 srv_idx = ms_thread->thread_ctx->srv_idx;
435 }
436
437 if (ms_network_connect(c, ms_setting.servers[srv_idx].srv_host_name,
438 ms_setting.servers[srv_idx].srv_port, ms_setting.udp, &ret_sfd)
439 )
440 {
441 break;
442 }
443
444 if (i == 0) {
445 c->sfd = ret_sfd;
446 }
447
448 if (!ms_setting.udp) {
449 c->tcpsfd[i] = ret_sfd;
450 }
451
452 c->alive_sfds++;
453 }
454
455 /* initialize udp sock handler if necessary */
456 if (ms_setting.facebook_test) {
457 ret_sfd = 0;
458 if (ms_network_connect(c, ms_setting.servers[srv_idx].srv_host_name,
459 ms_setting.servers[srv_idx].srv_port, true, &ret_sfd)
460 )
461 {
462 c->udpsfd = 0;
463 } else {
464 c->udpsfd = ret_sfd;
465 }
466 }
467
468 if ((i != c->total_sfds) || (ms_setting.facebook_test && (c->udpsfd == 0))) {
469 if (ms_setting.udp) {
470 close(c->sfd);
471 } else {
472 for (uint32_t j = 0; j < i; j++) {
473 close(c->tcpsfd[j]);
474 }
475 }
476
477 if (c->udpsfd) {
478 close(c->udpsfd);
479 }
480
481 return -1;
482 }
483
484 return EXIT_SUCCESS;
485 } /* ms_conn_sock_init */
486
487 /**
488 * each connection is managed by libevent, this function
489 * initialize the event of the connection structure.
490 *
491 * @param c, pointer of the concurrency
492 *
493 * @return int, if success, return EXIT_SUCCESS, else return -1
494 */
495 static int ms_conn_event_init(ms_conn_t *c) {
496 ms_thread_t *ms_thread = pthread_getspecific(ms_thread_key);
497 short event_flags = EV_WRITE | EV_PERSIST;
498
499 event_set(&c->event, c->sfd, event_flags, ms_event_handler, (void *) c);
500 event_base_set(ms_thread->base, &c->event);
501 c->ev_flags = event_flags;
502
503 if (event_add(&c->event, NULL) == -1) {
504 return -1;
505 }
506
507 return EXIT_SUCCESS;
508 } /* ms_conn_event_init */
509
510 /**
511 * setup a connection, each connection structure of each
512 * thread must call this function to initialize.
513 *
514 * @param c, pointer of the concurrency
515 *
516 * @return int, if success, return EXIT_SUCCESS, else return -1
517 */
518 int ms_setup_conn(ms_conn_t *c) {
519 if (ms_item_win_init(c)) {
520 return -1;
521 }
522
523 if (ms_conn_init(c, conn_write, DATA_BUFFER_SIZE, ms_setting.udp)) {
524 return -1;
525 }
526
527 if (ms_conn_sock_init(c)) {
528 return -1;
529 }
530
531 if (ms_conn_event_init(c)) {
532 return -1;
533 }
534
535 return EXIT_SUCCESS;
536 } /* ms_setup_conn */
537
538 /**
539 * Frees a connection.
540 *
541 * @param c, pointer of the concurrency
542 */
543 void ms_conn_free(ms_conn_t *c) {
544 ms_thread_t *ms_thread = pthread_getspecific(ms_thread_key);
545 if (c) {
546 if (c->hdrbuf)
547 free(c->hdrbuf);
548 if (c->msglist)
549 free(c->msglist);
550 if (c->rbuf)
551 free(c->rbuf);
552 if (c->wbuf)
553 free(c->wbuf);
554 if (c->iov)
555 free(c->iov);
556 if (c->mlget_task.mlget_item)
557 free(c->mlget_task.mlget_item);
558 if (c->rudpbuf)
559 free(c->rudpbuf);
560 if (c->udppkt)
561 free(c->udppkt);
562 if (c->item_win)
563 free(c->item_win);
564 if (c->tcpsfd)
565 free(c->tcpsfd);
566
567 if (--ms_thread->nactive_conn == 0) {
568 free(ms_thread->conn);
569 }
570 }
571 } /* ms_conn_free */
572
573 /**
574 * close a connection
575 *
576 * @param c, pointer of the concurrency
577 */
578 static void ms_conn_close(ms_conn_t *c) {
579 ms_thread_t *ms_thread = pthread_getspecific(ms_thread_key);
580 assert(c);
581
582 /* delete the event, the socket and the connection */
583 event_del(&c->event);
584
585 for (uint32_t i = 0; i < c->total_sfds; i++) {
586 if (c->tcpsfd[i] > 0) {
587 close(c->tcpsfd[i]);
588 }
589 }
590 c->sfd = 0;
591
592 if (ms_setting.facebook_test) {
593 close(c->udpsfd);
594 }
595
596 atomic_dec_32(&ms_stats.active_conns);
597
598 ms_conn_free(c);
599
600 if (ms_setting.run_time == 0) {
601 pthread_mutex_lock(&ms_global.run_lock.lock);
602 ms_global.run_lock.count++;
603 pthread_cond_signal(&ms_global.run_lock.cond);
604 pthread_mutex_unlock(&ms_global.run_lock.lock);
605 }
606
607 if (ms_thread->nactive_conn == 0) {
608 event_base_loopbreak(ms_thread->base);
609 }
610 } /* ms_conn_close */
611
612 /**
613 * create a new sock
614 *
615 * @param ai, server address information
616 *
617 * @return int, if success, return EXIT_SUCCESS, else return -1
618 */
619 static int ms_new_socket(struct addrinfo *ai) {
620 int sfd;
621
622 if ((sfd = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol)) == -1) {
623 fprintf(stderr, "socket() error: %s.\n", strerror(errno));
624 return -1;
625 }
626
627 return sfd;
628 } /* ms_new_socket */
629
630 /**
631 * Sets a socket's send buffer size to the maximum allowed by the system.
632 *
633 * @param sfd, file descriptor of socket
634 */
635 static void ms_maximize_sndbuf(const int sfd) {
636 socklen_t intsize = sizeof(int);
637 unsigned int last_good = 0;
638 unsigned int min, max, avg;
639 unsigned int old_size;
640
641 /* Start with the default size. */
642 if (getsockopt(sfd, SOL_SOCKET, SO_SNDBUF, &old_size, &intsize)) {
643 fprintf(stderr, "getsockopt(SO_SNDBUF)\n");
644 return;
645 }
646
647 /* Binary-search for the real maximum. */
648 min = old_size;
649 max = MAX_SENDBUF_SIZE;
650
651 while (min <= max) {
652 avg = ((unsigned int) (min + max)) / 2;
653 if (setsockopt(sfd, SOL_SOCKET, SO_SNDBUF, (void *) &avg, intsize) == 0) {
654 last_good = avg;
655 min = avg + 1;
656 } else {
657 max = avg - 1;
658 }
659 }
660 (void) last_good;
661 } /* ms_maximize_sndbuf */
662
663 /**
664 * socket connects the server
665 *
666 * @param c, pointer of the concurrency
667 * @param srv_host_name, the host name of the server
668 * @param srv_port, port of server
669 * @param is_udp, whether it's udp
670 * @param ret_sfd, the connected socket file descriptor
671 *
672 * @return int, if success, return EXIT_SUCCESS, else return -1
673 */
674 static int ms_network_connect(ms_conn_t *c, char *srv_host_name, const int srv_port,
675 const bool is_udp, int *ret_sfd) {
676 int sfd;
677 struct linger ling = {0, 0};
678 struct addrinfo *ai;
679 struct addrinfo *next;
680 struct addrinfo hints;
681 char port_buf[NI_MAXSERV];
682 int error;
683 int success = 0;
684
685 int flags = 1;
686
687 /*
688 * the memset call clears nonstandard fields in some impementations
689 * that otherwise mess things up.
690 */
691 memset(&hints, 0, sizeof(hints));
692 #ifdef AI_ADDRCONFIG
693 hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG;
694 #else
695 hints.ai_flags = AI_PASSIVE;
696 #endif /* AI_ADDRCONFIG */
697 if (is_udp) {
698 hints.ai_protocol = IPPROTO_UDP;
699 hints.ai_socktype = SOCK_DGRAM;
700 hints.ai_family = AF_INET; /* This left here because of issues with OSX 10.5 */
701 } else {
702 hints.ai_family = AF_UNSPEC;
703 hints.ai_protocol = IPPROTO_TCP;
704 hints.ai_socktype = SOCK_STREAM;
705 }
706
707 snprintf(port_buf, NI_MAXSERV, "%d", srv_port);
708 error = getaddrinfo(srv_host_name, port_buf, &hints, &ai);
709 if (error) {
710 if (error != EAI_SYSTEM)
711 fprintf(stderr, "getaddrinfo(): %s.\n", gai_strerror(error));
712 else
713 perror("getaddrinfo()");
714
715 return -1;
716 }
717
718 for (next = ai; next; next = next->ai_next) {
719 if ((sfd = ms_new_socket(next)) == -1) {
720 freeaddrinfo(ai);
721 return -1;
722 }
723
724 setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *) &flags, sizeof(flags));
725 if (is_udp) {
726 ms_maximize_sndbuf(sfd);
727 } else {
728 setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, (void *) &flags, sizeof(flags));
729 setsockopt(sfd, SOL_SOCKET, SO_LINGER, (void *) &ling, sizeof(ling));
730 setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, (void *) &flags, sizeof(flags));
731 }
732
733 if (is_udp) {
734 c->srv_recv_addr_size = sizeof(struct sockaddr);
735 memcpy(&c->srv_recv_addr, next->ai_addr, c->srv_recv_addr_size);
736 } else {
737 if (connect(sfd, next->ai_addr, next->ai_addrlen) == -1) {
738 close(sfd);
739 freeaddrinfo(ai);
740 return -1;
741 }
742 }
743
744 if (((flags = fcntl(sfd, F_GETFL, 0)) < 0) || (fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0)) {
745 fprintf(stderr, "setting O_NONBLOCK\n");
746 close(sfd);
747 freeaddrinfo(ai);
748 return -1;
749 }
750
751 if (ret_sfd) {
752 *ret_sfd = sfd;
753 }
754
755 success++;
756 }
757
758 freeaddrinfo(ai);
759
760 /* Return zero if we detected no errors in starting up connections */
761 return success == 0;
762 } /* ms_network_connect */
763
764 /**
765 * reconnect a disconnected sock
766 *
767 * @param c, pointer of the concurrency
768 *
769 * @return int, if success, return EXIT_SUCCESS, else return -1
770 */
771 static int ms_reconn(ms_conn_t *c) {
772 ms_thread_t *ms_thread = pthread_getspecific(ms_thread_key);
773 uint32_t srv_idx = 0;
774 uint32_t srv_conn_cnt = 0;
775
776 if (ms_setting.rep_write_srv > 0) {
777 srv_idx = c->cur_idx % ms_setting.srv_cnt;
778 srv_conn_cnt = ms_setting.sock_per_conn * ms_setting.nconns;
779 } else {
780 srv_idx = ms_thread->thread_ctx->srv_idx;
781 srv_conn_cnt = ms_setting.nconns / ms_setting.srv_cnt;
782 }
783
784 /* close the old socket handler */
785 close(c->sfd);
786 c->tcpsfd[c->cur_idx] = 0;
787
788 if (atomic_add_32_nv(&ms_setting.servers[srv_idx].disconn_cnt, 1) % srv_conn_cnt == 0) {
789 gettimeofday(&ms_setting.servers[srv_idx].disconn_time, NULL);
790 fprintf(stderr, "Server %s:%d disconnect\n", ms_setting.servers[srv_idx].srv_host_name,
791 ms_setting.servers[srv_idx].srv_port);
792 }
793
794 if (ms_setting.rep_write_srv > 0) {
795 uint32_t i = 0;
796
797 for (i = 0; i < c->total_sfds; i++) {
798 if (c->tcpsfd[i]) {
799 break;
800 }
801 }
802
803 /* all socks disconnect */
804 if (i == c->total_sfds) {
805 return -1;
806 }
807 } else {
808 do {
809 /* reconnect success, break the loop */
810 if (ms_network_connect(c, ms_setting.servers[srv_idx].srv_host_name,
811 ms_setting.servers[srv_idx].srv_port, ms_setting.udp, &c->sfd)
812 == 0)
813 {
814 c->tcpsfd[c->cur_idx] = c->sfd;
815 if (atomic_add_32_nv(&ms_setting.servers[srv_idx].reconn_cnt, 1) % (uint32_t) srv_conn_cnt
816 == 0) {
817 gettimeofday(&ms_setting.servers[srv_idx].reconn_time, NULL);
818 int reconn_time = (int) (ms_setting.servers[srv_idx].reconn_time.tv_sec
819 - ms_setting.servers[srv_idx].disconn_time.tv_sec);
820 fprintf(stderr, "Server %s:%d reconnect after %ds\n",
821 ms_setting.servers[srv_idx].srv_host_name, ms_setting.servers[srv_idx].srv_port,
822 reconn_time);
823 }
824 break;
825 }
826
827 if (ms_setting.rep_write_srv == 0 && c->total_sfds > 0) {
828 /* wait a second and reconnect */
829 sleep(1);
830 }
831 } while (ms_setting.rep_write_srv == 0 && c->total_sfds > 0);
832 }
833
834 if ((c->total_sfds > 1) && (c->tcpsfd[c->cur_idx] == 0)) {
835 c->sfd = 0;
836 c->alive_sfds--;
837 }
838
839 return EXIT_SUCCESS;
840 } /* ms_reconn */
841
842 /**
843 * reconnect several disconnected socks in the connection
844 * structure, the ever-1-second timer of the thread will check
845 * whether some socks in the connections disconnect. if
846 * disconnect, reconnect the sock.
847 *
848 * @param c, pointer of the concurrency
849 *
850 * @return int, if success, return EXIT_SUCCESS, else return -1
851 */
852 int ms_reconn_socks(ms_conn_t *c) {
853 ms_thread_t *ms_thread = pthread_getspecific(ms_thread_key);
854 uint32_t srv_idx = 0;
855 int ret_sfd = 0;
856 uint32_t srv_conn_cnt = 0;
857 struct timeval cur_time;
858
859 assert(c);
860
861 if ((c->total_sfds == 1) || (c->total_sfds == c->alive_sfds)) {
862 return EXIT_SUCCESS;
863 }
864
865 for (uint32_t i = 0; i < c->total_sfds; i++) {
866 if (c->tcpsfd[i] == 0) {
867 gettimeofday(&cur_time, NULL);
868
869 /**
870 * For failover test of replication, reconnect the socks after
871 * it disconnects more than 5 seconds, Otherwise memslap will
872 * block at connect() function and the work threads can't work
873 * in this interval.
874 */
875 if (cur_time.tv_sec - ms_setting.servers[srv_idx].disconn_time.tv_sec < 5) {
876 break;
877 }
878
879 if (ms_setting.rep_write_srv > 0) {
880 srv_idx = i % ms_setting.srv_cnt;
881 srv_conn_cnt = ms_setting.sock_per_conn * ms_setting.nconns;
882 } else {
883 srv_idx = ms_thread->thread_ctx->srv_idx;
884 srv_conn_cnt = ms_setting.nconns / ms_setting.srv_cnt;
885 }
886
887 if (ms_network_connect(c, ms_setting.servers[srv_idx].srv_host_name,
888 ms_setting.servers[srv_idx].srv_port, ms_setting.udp, &ret_sfd)
889 == 0)
890 {
891 c->tcpsfd[i] = ret_sfd;
892 c->alive_sfds++;
893
894 if (atomic_add_32_nv(&ms_setting.servers[srv_idx].reconn_cnt, 1) % (uint32_t) srv_conn_cnt
895 == 0) {
896 gettimeofday(&ms_setting.servers[srv_idx].reconn_time, NULL);
897 int reconn_time = (int) (ms_setting.servers[srv_idx].reconn_time.tv_sec
898 - ms_setting.servers[srv_idx].disconn_time.tv_sec);
899 fprintf(stderr, "Server %s:%d reconnect after %ds\n",
900 ms_setting.servers[srv_idx].srv_host_name, ms_setting.servers[srv_idx].srv_port,
901 reconn_time);
902 }
903 }
904 }
905 }
906
907 return EXIT_SUCCESS;
908 } /* ms_reconn_socks */
909
910 /**
911 * Tokenize the command string by replacing whitespace with '\0' and update
912 * the token array tokens with pointer to start of each token and length.
913 * Returns total number of tokens. The last valid token is the terminal
914 * token (value points to the first unprocessed character of the string and
915 * length zero).
916 *
917 * Usage example:
918 *
919 * while(ms_tokenize_command(command, ncommand, tokens, max_tokens) > 0) {
920 * for(int ix = 0; tokens[ix].length; ix++) {
921 * ...
922 * }
923 * ncommand = tokens[ix].value - command;
924 * command = tokens[ix].value;
925 * }
926 *
927 * @param command, the command string to token
928 * @param tokens, array to store tokens
929 * @param max_tokens, maximum tokens number
930 *
931 * @return int, the number of tokens
932 */
933 static int ms_tokenize_command(char *command, token_t *tokens, const int max_tokens) {
934 char *s, *e;
935 int ntokens = 0;
936
937 assert(command && tokens && max_tokens > 1);
938
939 for (s = e = command; ntokens < max_tokens - 1; ++e) {
940 if (*e == ' ') {
941 if (s != e) {
942 tokens[ntokens].value = s;
943 tokens[ntokens].length = (size_t)(e - s);
944 ntokens++;
945 *e = '\0';
946 }
947 s = e + 1;
948 } else if (*e == '\0') {
949 if (s != e) {
950 tokens[ntokens].value = s;
951 tokens[ntokens].length = (size_t)(e - s);
952 ntokens++;
953 }
954
955 break; /* string end */
956 }
957 }
958
959 return ntokens;
960 } /* ms_tokenize_command */
961
962 /**
963 * parse the response of server.
964 *
965 * @param c, pointer of the concurrency
966 * @param command, the string responded by server
967 *
968 * @return int, if the command completed return EXIT_SUCCESS, else return
969 * -1
970 */
971 static int ms_ascii_process_line(ms_conn_t *c, char *command) {
972 int ret = 0;
973 int64_t value_len;
974 char *buffer = command;
975
976 assert(c);
977
978 /**
979 * for command get, we store the returned value into local buffer
980 * then continue in ms_complete_nread().
981 */
982
983 switch (buffer[0]) {
984 case 'V': /* VALUE || VERSION */
985 if (buffer[1] == 'A') /* VALUE */ {
986 token_t tokens[MAX_TOKENS];
987 ms_tokenize_command(command, tokens, MAX_TOKENS);
988 errno = 0;
989 value_len = strtol(tokens[VALUELEN_TOKEN].value, NULL, 10);
990 if (errno) {
991 printf("<%d ERROR %s\n", c->sfd, strerror(errno));
992 }
993 memcpy(&c->currcmd.key_prefix, tokens[KEY_TOKEN].value, sizeof(c->currcmd.key_prefix));
994
995 /*
996 * We read the \r\n into the string since not doing so is more
997 * cycles then the waster of memory to do so.
998 *
999 * We are null terminating through, which will most likely make
1000 * some people lazy about using the return length.
1001 */
1002 c->rvbytes = (int) (value_len + 2);
1003 c->readval = true;
1004 ret = -1;
1005 }
1006
1007 break;
1008
1009 case 'O': /* OK */
1010 c->currcmd.retstat = MCD_SUCCESS;
1011 break;
1012
1013 case 'S': /* STORED STATS SERVER_ERROR */
1014 if (buffer[2] == 'A') /* STORED STATS */ { /* STATS*/
1015 c->currcmd.retstat = MCD_STAT;
1016 } else if (buffer[1] == 'E') {
1017 /* SERVER_ERROR */
1018 printf("<%d %s\n", c->sfd, buffer);
1019
1020 c->currcmd.retstat = MCD_SERVER_ERROR;
1021 } else if (buffer[1] == 'T') {
1022 /* STORED */
1023 c->currcmd.retstat = MCD_STORED;
1024 } else {
1025 c->currcmd.retstat = MCD_UNKNOWN_READ_FAILURE;
1026 }
1027 break;
1028
1029 case 'D': /* DELETED DATA */
1030 if (buffer[1] == 'E') {
1031 c->currcmd.retstat = MCD_DELETED;
1032 } else {
1033 c->currcmd.retstat = MCD_UNKNOWN_READ_FAILURE;
1034 }
1035
1036 break;
1037
1038 case 'N': /* NOT_FOUND NOT_STORED*/
1039 if (buffer[4] == 'F') {
1040 c->currcmd.retstat = MCD_NOTFOUND;
1041 } else if (buffer[4] == 'S') {
1042 printf("<%d %s\n", c->sfd, buffer);
1043 c->currcmd.retstat = MCD_NOTSTORED;
1044 } else {
1045 c->currcmd.retstat = MCD_UNKNOWN_READ_FAILURE;
1046 }
1047 break;
1048
1049 case 'E': /* PROTOCOL ERROR or END */
1050 if (buffer[1] == 'N') {
1051 /* END */
1052 c->currcmd.retstat = MCD_END;
1053 } else if (buffer[1] == 'R') {
1054 printf("<%d ERROR\n", c->sfd);
1055 c->currcmd.retstat = MCD_PROTOCOL_ERROR;
1056 } else if (buffer[1] == 'X') {
1057 c->currcmd.retstat = MCD_DATA_EXISTS;
1058 printf("<%d %s\n", c->sfd, buffer);
1059 } else {
1060 c->currcmd.retstat = MCD_UNKNOWN_READ_FAILURE;
1061 }
1062 break;
1063
1064 case 'C': /* CLIENT ERROR */
1065 printf("<%d %s\n", c->sfd, buffer);
1066 c->currcmd.retstat = MCD_CLIENT_ERROR;
1067 break;
1068
1069 default:
1070 c->currcmd.retstat = MCD_UNKNOWN_READ_FAILURE;
1071 break;
1072 } /* switch */
1073
1074 return ret;
1075 } /* ms_ascii_process_line */
1076
1077 /**
1078 * after one operation completes, reset the concurrency
1079 *
1080 * @param c, pointer of the concurrency
1081 * @param timeout, whether it's timeout
1082 */
1083 void ms_reset_conn(ms_conn_t *c, bool timeout) {
1084 assert(c);
1085
1086 if (c->udp) {
1087 if ((c->packets > 0) && (c->packets < MAX_UDP_PACKET)) {
1088 memset(c->udppkt, 0, sizeof(ms_udppkt_t) * (size_t) c->packets);
1089 }
1090
1091 c->packets = 0;
1092 c->recvpkt = 0;
1093 c->pktcurr = 0;
1094 c->ordcurr = 0;
1095 c->rudpbytes = 0;
1096 }
1097 c->currcmd.isfinish = true;
1098 c->ctnwrite = false;
1099 c->rbytes = 0;
1100 c->rcurr = c->rbuf;
1101 c->msgcurr = 0;
1102 c->msgused = 0;
1103 c->iovused = 0;
1104 ms_conn_set_state(c, conn_write);
1105 memcpy(&c->precmd, &c->currcmd, sizeof(ms_cmdstat_t)); /* replicate command state */
1106
1107 if (timeout) {
1108 ms_drive_machine(c);
1109 }
1110 } /* ms_reset_conn */
1111
1112 /**
1113 * if we have a complete line in the buffer, process it.
1114 *
1115 * @param c, pointer of the concurrency
1116 *
1117 * @return int, if success, return EXIT_SUCCESS, else return -1
1118 */
1119 static int ms_try_read_line(ms_conn_t *c) {
1120 if (c->protocol == binary_prot) {
1121 /* Do we have the complete packet header? */
1122 if ((uint64_t) c->rbytes < sizeof(c->binary_header)) {
1123 /* need more data! */
1124 return EXIT_SUCCESS;
1125 } else {
1126 #ifdef NEED_ALIGN
1127 if (((long) (c->rcurr)) % 8) {
1128 /* must realign input buffer */
1129 memmove(c->rbuf, c->rcurr, c->rbytes);
1130 c->rcurr = c->rbuf;
1131 if (settings.verbose) {
1132 fprintf(stderr, "%d: Realign input buffer.\n", c->sfd);
1133 }
1134 }
1135 #endif
1136 protocol_binary_response_header *rsp;
1137 rsp = (protocol_binary_response_header *) c->rcurr;
1138
1139 c->binary_header = *rsp;
1140 c->binary_header.response.extlen = rsp->response.extlen;
1141 c->binary_header.response.keylen = ntohs(rsp->response.keylen);
1142 c->binary_header.response.bodylen = ntohl(rsp->response.bodylen);
1143 c->binary_header.response.status = ntohs(rsp->response.status);
1144
1145 if (c->binary_header.response.magic != PROTOCOL_BINARY_RES) {
1146 fprintf(stderr, "Invalid magic: %x\n", c->binary_header.response.magic);
1147 ms_conn_set_state(c, conn_closing);
1148 return EXIT_SUCCESS;
1149 }
1150
1151 /* process this complete response */
1152 if (ms_bin_process_response(c) == 0) {
1153 /* current operation completed */
1154 ms_reset_conn(c, false);
1155 return -1;
1156 } else {
1157 c->rbytes -= (int32_t) sizeof(c->binary_header);
1158 c->rcurr += sizeof(c->binary_header);
1159 }
1160 }
1161 } else {
1162 char *el, *cont;
1163
1164 assert(c);
1165 assert(c->rcurr <= (c->rbuf + c->rsize));
1166
1167 if (c->rbytes == 0)
1168 return EXIT_SUCCESS;
1169
1170 el = memchr(c->rcurr, '\n', (size_t) c->rbytes);
1171 if (!el)
1172 return EXIT_SUCCESS;
1173
1174 cont = el + 1;
1175 if (((el - c->rcurr) > 1) && (*(el - 1) == '\r')) {
1176 el--;
1177 }
1178 *el = '\0';
1179
1180 assert(cont <= (c->rcurr + c->rbytes));
1181
1182 /* process this complete line */
1183 if (ms_ascii_process_line(c, c->rcurr) == 0) {
1184 /* current operation completed */
1185 ms_reset_conn(c, false);
1186 return -1;
1187 } else {
1188 /* current operation didn't complete */
1189 c->rbytes -= (int32_t)(cont - c->rcurr);
1190 c->rcurr = cont;
1191 }
1192
1193 assert(c->rcurr <= (c->rbuf + c->rsize));
1194 }
1195
1196 return -1;
1197 } /* ms_try_read_line */
1198
1199 /**
1200 * because the packet of UDP can't ensure the order, the
1201 * function is used to sort the received udp packet.
1202 *
1203 * @param c, pointer of the concurrency
1204 * @param buf, the buffer to store the ordered packages data
1205 * @param rbytes, the maximum capacity of the buffer
1206 *
1207 * @return int, if success, return the copy bytes, else return
1208 * -1
1209 */
1210 static int ms_sort_udp_packet(ms_conn_t *c, char *buf, int rbytes) {
1211 int len = 0;
1212 int wbytes = 0;
1213 uint16_t req_id = 0;
1214 uint16_t seq_num = 0;
1215 uint16_t packets = 0;
1216 unsigned char *header = NULL;
1217
1218 /* no enough data */
1219 assert(c);
1220 assert(buf);
1221 assert(c->rudpbytes >= UDP_HEADER_SIZE);
1222
1223 /* calculate received packets count */
1224 if (c->rudpbytes % UDP_MAX_PAYLOAD_SIZE >= UDP_HEADER_SIZE) {
1225 /* the last packet has some data */
1226 c->recvpkt = c->rudpbytes / UDP_MAX_PAYLOAD_SIZE + 1;
1227 } else {
1228 c->recvpkt = c->rudpbytes / UDP_MAX_PAYLOAD_SIZE;
1229 }
1230
1231 /* get the total packets count if necessary */
1232 if (c->packets == 0) {
1233 c->packets = HEADER_TO_PACKETS((unsigned char *) c->rudpbuf);
1234 }
1235
1236 /* build the ordered packet array */
1237 for (int i = c->pktcurr; i < c->recvpkt; i++) {
1238 header = (unsigned char *) c->rudpbuf + i * UDP_MAX_PAYLOAD_SIZE;
1239 req_id = (uint16_t) HEADER_TO_REQID(header);
1240 assert(req_id == c->request_id % (1 << 16));
1241
1242 packets = (uint16_t) HEADER_TO_PACKETS(header);
1243 assert(c->packets == HEADER_TO_PACKETS(header));
1244
1245 seq_num = (uint16_t) HEADER_TO_SEQNUM(header);
1246 c->udppkt[seq_num].header = header;
1247 c->udppkt[seq_num].data = (char *) header + UDP_HEADER_SIZE;
1248
1249 if (i == c->recvpkt - 1) {
1250 /* last received packet */
1251 if (c->rudpbytes % UDP_MAX_PAYLOAD_SIZE == 0) {
1252 c->udppkt[seq_num].rbytes = UDP_MAX_PAYLOAD_SIZE - UDP_HEADER_SIZE;
1253 c->pktcurr++;
1254 } else {
1255 c->udppkt[seq_num].rbytes = c->rudpbytes % UDP_MAX_PAYLOAD_SIZE - UDP_HEADER_SIZE;
1256 }
1257 } else {
1258 c->udppkt[seq_num].rbytes = UDP_MAX_PAYLOAD_SIZE - UDP_HEADER_SIZE;
1259 c->pktcurr++;
1260 }
1261 }
1262
1263 for (int i = c->ordcurr; i < c->recvpkt; i++) {
1264 /* there is some data to copy */
1265 if ((c->udppkt[i].data) && (c->udppkt[i].copybytes < c->udppkt[i].rbytes)) {
1266 header = c->udppkt[i].header;
1267 len = c->udppkt[i].rbytes - c->udppkt[i].copybytes;
1268 if (len > rbytes - wbytes) {
1269 len = rbytes - wbytes;
1270 }
1271
1272 assert(len <= rbytes - wbytes);
1273 assert(i == HEADER_TO_SEQNUM(header));
1274
1275 memcpy(buf + wbytes, c->udppkt[i].data + c->udppkt[i].copybytes, (size_t) len);
1276 wbytes += len;
1277 c->udppkt[i].copybytes += len;
1278
1279 if ((c->udppkt[i].copybytes == c->udppkt[i].rbytes)
1280 && (c->udppkt[i].rbytes == UDP_MAX_PAYLOAD_SIZE - UDP_HEADER_SIZE))
1281 {
1282 /* finish copying all the data of this packet, next */
1283 c->ordcurr++;
1284 }
1285
1286 /* last received packet, and finish copying all the data */
1287 if ((c->recvpkt == c->packets) && (i == c->recvpkt - 1)
1288 && (c->udppkt[i].copybytes == c->udppkt[i].rbytes))
1289 {
1290 break;
1291 }
1292
1293 /* no space to copy data */
1294 if (wbytes >= rbytes) {
1295 break;
1296 }
1297
1298 /* it doesn't finish reading all the data of the packet from network */
1299 if ((i != c->recvpkt - 1) && (c->udppkt[i].rbytes < UDP_MAX_PAYLOAD_SIZE - UDP_HEADER_SIZE)) {
1300 break;
1301 }
1302 } else {
1303 /* no data to copy */
1304 break;
1305 }
1306 }
1307 (void) packets;
1308
1309 return wbytes == 0 ? -1 : wbytes;
1310 } /* ms_sort_udp_packet */
1311
1312 /**
1313 * encapsulate upd read like tcp read
1314 *
1315 * @param c, pointer of the concurrency
1316 * @param buf, read buffer
1317 * @param len, length to read
1318 *
1319 * @return int, if success, return the read bytes, else return
1320 * -1
1321 */
1322 static int ms_udp_read(ms_conn_t *c, char *buf, int len) {
1323 int res = 0;
1324 int avail = 0;
1325 int rbytes = 0;
1326 int copybytes = 0;
1327
1328 assert(c->udp);
1329
1330 while (1) {
1331 if (c->rudpbytes + UDP_MAX_PAYLOAD_SIZE > c->rudpsize) {
1332 char *new_rbuf = realloc(c->rudpbuf, (size_t) c->rudpsize * 2);
1333 if (!new_rbuf) {
1334 fprintf(stderr, "Couldn't realloc input buffer.\n");
1335 c->rudpbytes = 0; /* ignore what we read */
1336 return -1;
1337 }
1338 c->rudpbuf = new_rbuf;
1339 c->rudpsize *= 2;
1340 }
1341
1342 avail = c->rudpsize - c->rudpbytes;
1343 /* UDP each time read a packet, 1400 bytes */
1344 res = (int) read(c->sfd, c->rudpbuf + c->rudpbytes, (size_t) avail);
1345
1346 if (res > 0) {
1347 atomic_add_size(&ms_stats.bytes_read, res);
1348 c->rudpbytes += res;
1349 rbytes += res;
1350 if (res == avail) {
1351 continue;
1352 } else {
1353 break;
1354 }
1355 }
1356
1357 if (res == 0) {
1358 /* "connection" closed */
1359 return res;
1360 }
1361
1362 if (res == -1) {
1363 /* no data to read */
1364 return res;
1365 }
1366 }
1367
1368 /* copy data to read buffer */
1369 if (rbytes > 0) {
1370 copybytes = ms_sort_udp_packet(c, buf, len);
1371 }
1372
1373 if (copybytes == -1) {
1374 atomic_add_size(&ms_stats.pkt_disorder, 1);
1375 }
1376
1377 return copybytes;
1378 } /* ms_udp_read */
1379
1380 /*
1381 * read from network as much as we can, handle buffer overflow and connection
1382 * close.
1383 * before reading, move the remaining incomplete fragment of a command
1384 * (if any) to the beginning of the buffer.
1385 * return EXIT_SUCCESS if there's nothing to read on the first read.
1386 */
1387
1388 /**
1389 * read from network as much as we can, handle buffer overflow and connection
1390 * close. before reading, move the remaining incomplete fragment of a command
1391 * (if any) to the beginning of the buffer.
1392 *
1393 * @param c, pointer of the concurrency
1394 *
1395 * @return int,
1396 * return EXIT_SUCCESS if there's nothing to read on the first read.
1397 * return EXIT_FAILURE if get data
1398 * return -1 if error happens
1399 */
1400 static int ms_try_read_network(ms_conn_t *c) {
1401 int gotdata = 0;
1402 int res;
1403 int64_t avail;
1404
1405 assert(c);
1406
1407 if ((c->rcurr != c->rbuf)
1408 && (!c->readval || (c->rvbytes > c->rsize - (c->rcurr - c->rbuf))
1409 || (c->readval && (c->rcurr - c->rbuf > c->rbytes))))
1410 {
1411 if (c->rbytes) /* otherwise there's nothing to copy */
1412 memmove(c->rbuf, c->rcurr, (size_t) c->rbytes);
1413 c->rcurr = c->rbuf;
1414 }
1415
1416 while (1) {
1417 if (c->rbytes >= c->rsize) {
1418 char *new_rbuf = realloc(c->rbuf, (size_t) c->rsize * 2);
1419 if (!new_rbuf) {
1420 fprintf(stderr, "Couldn't realloc input buffer.\n");
1421 c->rbytes = 0; /* ignore what we read */
1422 return -1;
1423 }
1424 c->rcurr = c->rbuf = new_rbuf;
1425 c->rsize *= 2;
1426 }
1427
1428 avail = c->rsize - c->rbytes - (c->rcurr - c->rbuf);
1429 if (avail == 0) {
1430 break;
1431 }
1432
1433 if (c->udp) {
1434 res = (int32_t) ms_udp_read(c, c->rcurr + c->rbytes, (int32_t) avail);
1435 } else {
1436 res = (int) read(c->sfd, c->rcurr + c->rbytes, (size_t) avail);
1437 }
1438
1439 if (res > 0) {
1440 if (!c->udp) {
1441 atomic_add_size(&ms_stats.bytes_read, res);
1442 }
1443 gotdata = 1;
1444 c->rbytes += res;
1445 if (res == avail) {
1446 continue;
1447 } else {
1448 break;
1449 }
1450 }
1451 if (res == 0) {
1452 /* connection closed */
1453 ms_conn_set_state(c, conn_closing);
1454 return -1;
1455 }
1456 if (res == -1) {
1457 if ((errno == EAGAIN) || (errno == EWOULDBLOCK))
1458 break;
1459 /* Should close on unhandled errors. */
1460 ms_conn_set_state(c, conn_closing);
1461 return -1;
1462 }
1463 }
1464
1465 return gotdata;
1466 } /* ms_try_read_network */
1467
1468 /**
1469 * after get the object from server, verify the value if
1470 * necessary.
1471 *
1472 * @param c, pointer of the concurrency
1473 * @param mlget_item, pointer of mulit-get task item structure
1474 * @param value, received value string
1475 * @param vlen, received value string length
1476 */
1477 static void ms_verify_value(ms_conn_t *c, ms_mlget_task_item_t *mlget_item, char *value, int vlen) {
1478 if (c->curr_task.verify) {
1479 assert(c->curr_task.item->value_offset != INVALID_OFFSET);
1480 char *orignval = &ms_setting.char_block[c->curr_task.item->value_offset];
1481 char *orignkey = &ms_setting.char_block[c->curr_task.item->key_suffix_offset];
1482
1483 /* verify expire time if necessary */
1484 if (c->curr_task.item->exp_time > 0) {
1485 struct timeval curr_time;
1486 gettimeofday(&curr_time, NULL);
1487
1488 /* object expired but get it now */
1489 if (curr_time.tv_sec - c->curr_task.item->client_time
1490 > c->curr_task.item->exp_time + EXPIRE_TIME_ERROR)
1491 {
1492 atomic_add_size(&ms_stats.exp_get, 1);
1493
1494 if (ms_setting.verbose) {
1495 char set_time[64];
1496 char cur_time[64];
1497 strftime(set_time, 64, "%Y-%m-%d %H:%M:%S", localtime(&c->curr_task.item->client_time));
1498 strftime(cur_time, 64, "%Y-%m-%d %H:%M:%S", localtime(&curr_time.tv_sec));
1499 fprintf(stderr,
1500 "\n<%d expire time verification failed, "
1501 "object expired but get it now\n"
1502 "\tkey len: %d\n"
1503 "\tkey: %" PRIx64 " %.*s\n"
1504 "\tset time: %s current time: %s "
1505 "diff time: %d expire time: %d\n"
1506 "\texpected data: \n"
1507 "\treceived data len: %d\n"
1508 "\treceived data: %.*s\n",
1509 c->sfd, c->curr_task.item->key_size, c->curr_task.item->key_prefix,
1510 c->curr_task.item->key_size - (int) KEY_PREFIX_SIZE, orignkey, set_time, cur_time,
1511 (int) (curr_time.tv_sec - c->curr_task.item->client_time),
1512 c->curr_task.item->exp_time, vlen, vlen, value);
1513 fflush(stderr);
1514 }
1515 }
1516 } else {
1517 if ((c->curr_task.item->value_size != vlen) || (memcmp(orignval, value, (size_t) vlen)))
1518 {
1519 atomic_add_size(&ms_stats.vef_failed, 1);
1520
1521 if (ms_setting.verbose) {
1522 fprintf(stderr,
1523 "\n<%d data verification failed\n"
1524 "\tkey len: %d\n"
1525 "\tkey: %" PRIx64 " %.*s\n"
1526 "\texpected data len: %d\n"
1527 "\texpected data: %.*s\n"
1528 "\treceived data len: %d\n"
1529 "\treceived data: %.*s\n",
1530 c->sfd, c->curr_task.item->key_size, c->curr_task.item->key_prefix,
1531 c->curr_task.item->key_size - (int) KEY_PREFIX_SIZE, orignkey,
1532 c->curr_task.item->value_size, c->curr_task.item->value_size, orignval, vlen,
1533 vlen, value);
1534 fflush(stderr);
1535 }
1536 }
1537 }
1538
1539 c->curr_task.finish_verify = true;
1540
1541 if (mlget_item) {
1542 mlget_item->finish_verify = true;
1543 }
1544 }
1545 } /* ms_verify_value */
1546
1547 /**
1548 * For ASCII protocol, after store the data into the local
1549 * buffer, run this function to handle the data.
1550 *
1551 * @param c, pointer of the concurrency
1552 */
1553 static void ms_ascii_complete_nread(ms_conn_t *c) {
1554 assert(c);
1555 assert(c->rbytes >= c->rvbytes);
1556 assert(c->protocol == ascii_prot);
1557 if (c->rvbytes > 2) {
1558 assert(c->rcurr[c->rvbytes - 1] == '\n' && c->rcurr[c->rvbytes - 2] == '\r');
1559 }
1560
1561 /* multi-get */
1562 ms_mlget_task_item_t *mlget_item = NULL;
1563 if (((ms_setting.mult_key_num > 1) && (c->mlget_task.mlget_num >= ms_setting.mult_key_num))
1564 || ((c->remain_exec_num == 0) && (c->mlget_task.mlget_num > 0)))
1565 {
1566 c->mlget_task.value_index++;
1567 mlget_item = &c->mlget_task.mlget_item[c->mlget_task.value_index];
1568
1569 if (mlget_item->item->key_prefix == c->currcmd.key_prefix) {
1570 c->curr_task.item = mlget_item->item;
1571 c->curr_task.verify = mlget_item->verify;
1572 c->curr_task.finish_verify = mlget_item->finish_verify;
1573 mlget_item->get_miss = false;
1574 } else {
1575 /* Try to find the task item in multi-get task array */
1576 for (int i = 0; i < c->mlget_task.mlget_num; i++) {
1577 mlget_item = &c->mlget_task.mlget_item[i];
1578 if (mlget_item->item->key_prefix == c->currcmd.key_prefix) {
1579 c->curr_task.item = mlget_item->item;
1580 c->curr_task.verify = mlget_item->verify;
1581 c->curr_task.finish_verify = mlget_item->finish_verify;
1582 mlget_item->get_miss = false;
1583
1584 break;
1585 }
1586 }
1587 }
1588 }
1589
1590 ms_verify_value(c, mlget_item, c->rcurr, c->rvbytes - 2);
1591
1592 c->curr_task.get_miss = false;
1593 c->rbytes -= c->rvbytes;
1594 c->rcurr = c->rcurr + c->rvbytes;
1595 assert(c->rcurr <= (c->rbuf + c->rsize));
1596 c->readval = false;
1597 c->rvbytes = 0;
1598 } /* ms_ascii_complete_nread */
1599
1600 /**
1601 * For binary protocol, after store the data into the local
1602 * buffer, run this function to handle the data.
1603 *
1604 * @param c, pointer of the concurrency
1605 */
1606 static void ms_bin_complete_nread(ms_conn_t *c) {
1607 assert(c);
1608 assert(c->rbytes >= c->rvbytes);
1609 assert(c->protocol == binary_prot);
1610
1611 int extlen = c->binary_header.response.extlen;
1612 int keylen = c->binary_header.response.keylen;
1613 uint8_t opcode = c->binary_header.response.opcode;
1614
1615 /* not get command or not include value, just return */
1616 if (((opcode != PROTOCOL_BINARY_CMD_GET) && (opcode != PROTOCOL_BINARY_CMD_GETQ))
1617 || (c->rvbytes <= extlen + keylen))
1618 {
1619 /* get miss */
1620 if (c->binary_header.response.opcode == PROTOCOL_BINARY_CMD_GET) {
1621 c->currcmd.retstat = MCD_END;
1622 c->curr_task.get_miss = true;
1623 }
1624
1625 c->readval = false;
1626 c->rvbytes = 0;
1627 ms_reset_conn(c, false);
1628 return;
1629 }
1630
1631 /* multi-get */
1632 ms_mlget_task_item_t *mlget_item = NULL;
1633 if (((ms_setting.mult_key_num > 1) && (c->mlget_task.mlget_num >= ms_setting.mult_key_num))
1634 || ((c->remain_exec_num == 0) && (c->mlget_task.mlget_num > 0)))
1635 {
1636 c->mlget_task.value_index++;
1637 mlget_item = &c->mlget_task.mlget_item[c->mlget_task.value_index];
1638
1639 c->curr_task.item = mlget_item->item;
1640 c->curr_task.verify = mlget_item->verify;
1641 c->curr_task.finish_verify = mlget_item->finish_verify;
1642 mlget_item->get_miss = false;
1643 }
1644
1645 ms_verify_value(c, mlget_item, c->rcurr + extlen + keylen, c->rvbytes - extlen - keylen);
1646
1647 c->currcmd.retstat = MCD_END;
1648 c->curr_task.get_miss = false;
1649 c->rbytes -= c->rvbytes;
1650 c->rcurr = c->rcurr + c->rvbytes;
1651 assert(c->rcurr <= (c->rbuf + c->rsize));
1652 c->readval = false;
1653 c->rvbytes = 0;
1654
1655 if (ms_setting.mult_key_num > 1) {
1656 /* multi-get have check all the item */
1657 if (c->mlget_task.value_index == c->mlget_task.mlget_num - 1) {
1658 ms_reset_conn(c, false);
1659 }
1660 } else {
1661 /* single get */
1662 ms_reset_conn(c, false);
1663 }
1664 } /* ms_bin_complete_nread */
1665
1666 /**
1667 * we get here after reading the value of get commands.
1668 *
1669 * @param c, pointer of the concurrency
1670 */
1671 static void ms_complete_nread(ms_conn_t *c) {
1672 assert(c);
1673 assert(c->rbytes >= c->rvbytes);
1674 assert(c->protocol == ascii_prot || c->protocol == binary_prot);
1675
1676 if (c->protocol == binary_prot) {
1677 ms_bin_complete_nread(c);
1678 } else {
1679 ms_ascii_complete_nread(c);
1680 }
1681 } /* ms_complete_nread */
1682
1683 /**
1684 * Adds a message header to a connection.
1685 *
1686 * @param c, pointer of the concurrency
1687 *
1688 * @return int, if success, return EXIT_SUCCESS, else return -1
1689 */
1690 static int ms_add_msghdr(ms_conn_t *c) {
1691 struct msghdr *msg;
1692
1693 assert(c);
1694
1695 if (c->msgsize == c->msgused) {
1696 msg = realloc(c->msglist, (size_t) c->msgsize * 2 * sizeof(struct msghdr));
1697 if (!msg)
1698 return -1;
1699
1700 c->msglist = msg;
1701 c->msgsize *= 2;
1702 }
1703
1704 msg = c->msglist + c->msgused;
1705
1706 /**
1707 * this wipes msg_iovlen, msg_control, msg_controllen, and
1708 * msg_flags, the last 3 of which aren't defined on solaris:
1709 */
1710 memset(msg, 0, sizeof(struct msghdr));
1711
1712 msg->msg_iov = &c->iov[c->iovused];
1713
1714 if (c->udp && (c->srv_recv_addr_size > 0)) {
1715 msg->msg_name = &c->srv_recv_addr;
1716 msg->msg_namelen = c->srv_recv_addr_size;
1717 }
1718
1719 c->msgbytes = 0;
1720 c->msgused++;
1721
1722 if (c->udp) {
1723 /* Leave room for the UDP header, which we'll fill in later. */
1724 return ms_add_iov(c, NULL, UDP_HEADER_SIZE);
1725 }
1726
1727 return EXIT_SUCCESS;
1728 } /* ms_add_msghdr */
1729
1730 /**
1731 * Ensures that there is room for another structure iovec in a connection's
1732 * iov list.
1733 *
1734 * @param c, pointer of the concurrency
1735 *
1736 * @return int, if success, return EXIT_SUCCESS, else return -1
1737 */
1738 static int ms_ensure_iov_space(ms_conn_t *c) {
1739 assert(c);
1740
1741 if (c->iovused >= c->iovsize) {
1742 int i, iovnum;
1743 struct iovec *new_iov =
1744 (struct iovec *) realloc(c->iov, ((size_t) c->iovsize * 2) * sizeof(struct iovec));
1745 if (!new_iov)
1746 return -1;
1747
1748 c->iov = new_iov;
1749 c->iovsize *= 2;
1750
1751 /* Point all the msghdr structures at the new list. */
1752 for (i = 0, iovnum = 0; i < c->msgused; i++) {
1753 c->msglist[i].msg_iov = &c->iov[iovnum];
1754 iovnum += (int) c->msglist[i].msg_iovlen;
1755 }
1756 }
1757
1758 return EXIT_SUCCESS;
1759 } /* ms_ensure_iov_space */
1760
1761 /**
1762 * Adds data to the list of pending data that will be written out to a
1763 * connection.
1764 *
1765 * @param c, pointer of the concurrency
1766 * @param buf, the buffer includes data to send
1767 * @param len, the data length in the buffer
1768 *
1769 * @return int, if success, return EXIT_SUCCESS, else return -1
1770 */
1771 static int ms_add_iov(ms_conn_t *c, const void *buf, int len) {
1772 struct msghdr *m;
1773 int leftover;
1774 bool limit_to_mtu;
1775
1776 assert(c);
1777
1778 do {
1779 m = &c->msglist[c->msgused - 1];
1780
1781 /*
1782 * Limit UDP packets, to UDP_MAX_PAYLOAD_SIZE bytes.
1783 */
1784 limit_to_mtu = c->udp;
1785
1786 #ifdef IOV_MAX
1787 /* We may need to start a new msghdr if this one is full. */
1788 if ((m->msg_iovlen == IOV_MAX) || (limit_to_mtu && (c->msgbytes >= UDP_MAX_SEND_PAYLOAD_SIZE)))
1789 {
1790 ms_add_msghdr(c);
1791 m = &c->msglist[c->msgused - 1];
1792 }
1793 #endif
1794
1795 if (ms_ensure_iov_space(c))
1796 return -1;
1797
1798 /* If the fragment is too big to fit in the datagram, split it up */
1799 if (limit_to_mtu && (len + c->msgbytes > UDP_MAX_SEND_PAYLOAD_SIZE)) {
1800 leftover = len + c->msgbytes - UDP_MAX_SEND_PAYLOAD_SIZE;
1801 len -= leftover;
1802 } else {
1803 leftover = 0;
1804 }
1805
1806 m = &c->msglist[c->msgused - 1];
1807 m->msg_iov[m->msg_iovlen].iov_base = (void *) buf;
1808 m->msg_iov[m->msg_iovlen].iov_len = (size_t) len;
1809
1810 c->msgbytes += len;
1811 c->iovused++;
1812 m->msg_iovlen++;
1813
1814 buf = ((char *) buf) + len;
1815 len = leftover;
1816 } while (leftover > 0);
1817
1818 return EXIT_SUCCESS;
1819 } /* ms_add_iov */
1820
1821 /**
1822 * Constructs a set of UDP headers and attaches them to the outgoing messages.
1823 *
1824 * @param c, pointer of the concurrency
1825 *
1826 * @return int, if success, return EXIT_SUCCESS, else return -1
1827 */
1828 static int ms_build_udp_headers(ms_conn_t *c) {
1829 int i;
1830 unsigned char *hdr;
1831
1832 assert(c);
1833
1834 c->request_id = ms_get_udp_request_id();
1835
1836 if (c->msgused > c->hdrsize) {
1837 void *new_hdrbuf;
1838 if (c->hdrbuf)
1839 new_hdrbuf = realloc(c->hdrbuf, (size_t) c->msgused * 2 * UDP_HEADER_SIZE);
1840 else
1841 new_hdrbuf = malloc((size_t) c->msgused * 2 * UDP_HEADER_SIZE);
1842 if (!new_hdrbuf)
1843 return -1;
1844
1845 c->hdrbuf = (unsigned char *) new_hdrbuf;
1846 c->hdrsize = c->msgused * 2;
1847 }
1848
1849 /* If this is a multi-packet request, drop it. */
1850 if (c->udp && (c->msgused > 1)) {
1851 fprintf(stderr, "multi-packet request for UDP not supported.\n");
1852 return -1;
1853 }
1854
1855 hdr = c->hdrbuf;
1856 for (i = 0; i < c->msgused; i++) {
1857 c->msglist[i].msg_iov[0].iov_base = (void *) hdr;
1858 c->msglist[i].msg_iov[0].iov_len = UDP_HEADER_SIZE;
1859 *hdr++ = (unsigned char) (c->request_id / 256);
1860 *hdr++ = (unsigned char) (c->request_id % 256);
1861 *hdr++ = (unsigned char) (i / 256);
1862 *hdr++ = (unsigned char) (i % 256);
1863 *hdr++ = (unsigned char) (c->msgused / 256);
1864 *hdr++ = (unsigned char) (c->msgused % 256);
1865 *hdr++ = (unsigned char) 1; /* support facebook memcached */
1866 *hdr++ = (unsigned char) 0;
1867 assert(hdr == ((unsigned char *) c->msglist[i].msg_iov[0].iov_base + UDP_HEADER_SIZE));
1868 }
1869
1870 return EXIT_SUCCESS;
1871 } /* ms_build_udp_headers */
1872
1873 /**
1874 * Transmit the next chunk of data from our list of msgbuf structures.
1875 *
1876 * @param c, pointer of the concurrency
1877 *
1878 * @return TRANSMIT_COMPLETE All done writing.
1879 * TRANSMIT_INCOMPLETE More data remaining to write.
1880 * TRANSMIT_SOFT_ERROR Can't write any more right now.
1881 * TRANSMIT_HARD_ERROR Can't write (c->state is set to conn_closing)
1882 */
1883 static int ms_transmit(ms_conn_t *c) {
1884 assert(c);
1885
1886 if ((c->msgcurr < c->msgused) && (c->msglist[c->msgcurr].msg_iovlen == 0)) {
1887 /* Finished writing the current msg; advance to the next. */
1888 c->msgcurr++;
1889 }
1890
1891 if (c->msgcurr < c->msgused) {
1892 ssize_t res;
1893 struct msghdr *m = &c->msglist[c->msgcurr];
1894
1895 res = sendmsg(c->sfd, m, 0);
1896 if (res > 0) {
1897 atomic_add_size(&ms_stats.bytes_written, res);
1898
1899 /* We've written some of the data. Remove the completed
1900 * iovec entries from the list of pending writes. */
1901 while (m->msg_iovlen > 0 && res >= (ssize_t) m->msg_iov->iov_len) {
1902 res -= (ssize_t) m->msg_iov->iov_len;
1903 m->msg_iovlen--;
1904 m->msg_iov++;
1905 }
1906
1907 /* Might have written just part of the last iovec entry;
1908 * adjust it so the next write will do the rest. */
1909 if (res > 0) {
1910 m->msg_iov->iov_base = (void *) ((unsigned char *) m->msg_iov->iov_base + res);
1911 m->msg_iov->iov_len -= (size_t) res;
1912 }
1913 return TRANSMIT_INCOMPLETE;
1914 }
1915 if ((res == -1) && ((errno == EAGAIN) || (errno == EWOULDBLOCK))) {
1916 if (!ms_update_event(c, EV_WRITE | EV_PERSIST)) {
1917 fprintf(stderr, "Couldn't update event.\n");
1918 ms_conn_set_state(c, conn_closing);
1919 return TRANSMIT_HARD_ERROR;
1920 }
1921 return TRANSMIT_SOFT_ERROR;
1922 }
1923
1924 /* if res==0 or res==-1 and error is not EAGAIN or EWOULDBLOCK,
1925 * we have a real error, on which we close the connection */
1926 fprintf(stderr, "Failed to write, and not due to blocking.\n");
1927
1928 ms_conn_set_state(c, conn_closing);
1929 return TRANSMIT_HARD_ERROR;
1930 } else {
1931 return TRANSMIT_COMPLETE;
1932 }
1933 } /* ms_transmit */
1934
1935 /**
1936 * Shrinks a connection's buffers if they're too big. This prevents
1937 * periodic large "mget" response from server chewing lots of client
1938 * memory.
1939 *
1940 * This should only be called in between requests since it can wipe output
1941 * buffers!
1942 *
1943 * @param c, pointer of the concurrency
1944 */
1945 static void ms_conn_shrink(ms_conn_t *c) {
1946 assert(c);
1947
1948 if (c->udp)
1949 return;
1950
1951 if ((c->rsize > READ_BUFFER_HIGHWAT) && (c->rbytes < DATA_BUFFER_SIZE)) {
1952 char *newbuf;
1953
1954 if (c->rcurr != c->rbuf)
1955 memmove(c->rbuf, c->rcurr, (size_t) c->rbytes);
1956
1957 newbuf = (char *) realloc((void *) c->rbuf, DATA_BUFFER_SIZE);
1958
1959 if (newbuf) {
1960 c->rbuf = newbuf;
1961 c->rsize = DATA_BUFFER_SIZE;
1962 }
1963 c->rcurr = c->rbuf;
1964 }
1965
1966 if (c->udp && (c->rudpsize > UDP_DATA_BUFFER_HIGHWAT)
1967 && (c->rudpbytes + UDP_MAX_PAYLOAD_SIZE < UDP_DATA_BUFFER_SIZE))
1968 {
1969 char *new_rbuf = (char *) realloc(c->rudpbuf, (size_t) c->rudpsize * 2);
1970 if (new_rbuf) {
1971 c->rudpbuf = new_rbuf;
1972 c->rudpsize = UDP_DATA_BUFFER_SIZE;
1973 }
1974 /* TODO check error condition? */
1975 }
1976
1977 if (c->msgsize > MSG_LIST_HIGHWAT) {
1978 struct msghdr *newbuf =
1979 (struct msghdr *) realloc((void *) c->msglist, MSG_LIST_INITIAL * sizeof(c->msglist[0]));
1980 if (newbuf) {
1981 c->msglist = newbuf;
1982 c->msgsize = MSG_LIST_INITIAL;
1983 }
1984 /* TODO check error condition? */
1985 }
1986
1987 if (c->iovsize > IOV_LIST_HIGHWAT) {
1988 struct iovec *newbuf =
1989 (struct iovec *) realloc((void *) c->iov, IOV_LIST_INITIAL * sizeof(c->iov[0]));
1990 if (newbuf) {
1991 c->iov = newbuf;
1992 c->iovsize = IOV_LIST_INITIAL;
1993 }
1994 /* TODO check return value */
1995 }
1996 } /* ms_conn_shrink */
1997
1998 /**
1999 * Sets a connection's current state in the state machine. Any special
2000 * processing that needs to happen on certain state transitions can
2001 * happen here.
2002 *
2003 * @param c, pointer of the concurrency
2004 * @param state, connection state
2005 */
2006 static void ms_conn_set_state(ms_conn_t *c, int state) {
2007 assert(c);
2008
2009 if (state != c->state) {
2010 if (state == conn_read) {
2011 ms_conn_shrink(c);
2012 }
2013 c->state = state;
2014 }
2015 } /* ms_conn_set_state */
2016
2017 /**
2018 * update the event if socks change state. for example: when
2019 * change the listen scoket read event to sock write event, or
2020 * change socket handler, we could call this function.
2021 *
2022 * @param c, pointer of the concurrency
2023 * @param new_flags, new event flags
2024 *
2025 * @return bool, if success, return true, else return false
2026 */
2027 static bool ms_update_event(ms_conn_t *c, const int new_flags) {
2028 assert(c);
2029
2030 struct event_base *base = c->event.ev_base;
2031 if ((c->ev_flags == new_flags) && (ms_setting.rep_write_srv == 0)
2032 && (!ms_setting.facebook_test || (c->total_sfds == 1)))
2033 {
2034 return true;
2035 }
2036
2037 if (event_del(&c->event) == -1) {
2038 /* try to delete the event again */
2039 if (event_del(&c->event) == -1) {
2040 return false;
2041 }
2042 }
2043
2044 event_set(&c->event, c->sfd, (short) new_flags, ms_event_handler, (void *) c);
2045 event_base_set(base, &c->event);
2046 c->ev_flags = (short) new_flags;
2047
2048 if (event_add(&c->event, NULL) == -1) {
2049 return false;
2050 }
2051
2052 return true;
2053 } /* ms_update_event */
2054
2055 /**
2056 * If user want to get the expected throughput, we could limit
2057 * the performance of memslap. we could give up some work and
2058 * just wait a short time. The function is used to check this
2059 * case.
2060 *
2061 * @param c, pointer of the concurrency
2062 *
2063 * @return bool, if success, return true, else return false
2064 */
2065 static bool ms_need_yield(ms_conn_t *c) {
2066 ms_thread_t *ms_thread = pthread_getspecific(ms_thread_key);
2067 int64_t tps = 0;
2068 int64_t time_diff = 0;
2069 struct timeval curr_time;
2070 ms_task_t *task = &c->curr_task;
2071
2072 if (ms_setting.expected_tps > 0) {
2073 gettimeofday(&curr_time, NULL);
2074 time_diff = ms_time_diff(&ms_thread->startup_time, &curr_time);
2075 tps = (int64_t)(((task->get_opt + task->set_opt) / (uint64_t) time_diff) * 1000000);
2076
2077 /* current throughput is greater than expected throughput */
2078 if (tps > ms_thread->thread_ctx->tps_perconn) {
2079 return true;
2080 }
2081 }
2082
2083 return false;
2084 } /* ms_need_yield */
2085
2086 /**
2087 * used to update the start time of each operation
2088 *
2089 * @param c, pointer of the concurrency
2090 */
2091 static void ms_update_start_time(ms_conn_t *c) {
2092 ms_task_item_t *item = c->curr_task.item;
2093
2094 if ((ms_setting.stat_freq > 0) || c->udp || ((c->currcmd.cmd == CMD_SET) && (item->exp_time > 0)))
2095 {
2096 gettimeofday(&c->start_time, NULL);
2097 if ((c->currcmd.cmd == CMD_SET) && (item->exp_time > 0)) {
2098 /* record the current time */
2099 item->client_time = c->start_time.tv_sec;
2100 }
2101 }
2102 } /* ms_update_start_time */
2103
2104 /**
2105 * run the state machine
2106 *
2107 * @param c, pointer of the concurrency
2108 */
2109 static void ms_drive_machine(ms_conn_t *c) {
2110 bool stop = false;
2111
2112 assert(c);
2113
2114 while (!stop) {
2115 switch (c->state) {
2116 case conn_read:
2117 if (c->readval) {
2118 if (c->rbytes >= c->rvbytes) {
2119 ms_complete_nread(c);
2120 break;
2121 }
2122 } else {
2123 if (ms_try_read_line(c)) {
2124 break;
2125 }
2126 }
2127
2128 if (ms_try_read_network(c)) {
2129 break;
2130 }
2131
2132 /* doesn't read all the response data, wait event wake up */
2133 if (!c->currcmd.isfinish) {
2134 if (!ms_update_event(c, EV_READ | EV_PERSIST)) {
2135 fprintf(stderr, "Couldn't update event.\n");
2136 ms_conn_set_state(c, conn_closing);
2137 break;
2138 }
2139 stop = true;
2140 break;
2141 }
2142
2143 /* we have no command line and no data to read from network, next write */
2144 ms_conn_set_state(c, conn_write);
2145 memcpy(&c->precmd, &c->currcmd, sizeof(ms_cmdstat_t)); /* replicate command state */
2146
2147 break;
2148
2149 case conn_write:
2150 if (!c->ctnwrite && ms_need_yield(c)) {
2151 usleep(10);
2152
2153 if (!ms_update_event(c, EV_WRITE | EV_PERSIST)) {
2154 fprintf(stderr, "Couldn't update event.\n");
2155 ms_conn_set_state(c, conn_closing);
2156 break;
2157 }
2158 stop = true;
2159 break;
2160 }
2161
2162 if (!c->ctnwrite && (ms_exec_task(c))) {
2163 ms_conn_set_state(c, conn_closing);
2164 break;
2165 }
2166
2167 /* record the start time before starting to send data if necessary */
2168 if (!c->ctnwrite || (c->change_sfd && c->ctnwrite)) {
2169 if (c->change_sfd) {
2170 c->change_sfd = false;
2171 }
2172 ms_update_start_time(c);
2173 }
2174
2175 /* change sfd if necessary */
2176 if (c->change_sfd) {
2177 c->ctnwrite = true;
2178 stop = true;
2179 break;
2180 }
2181
2182 /* execute task until nothing need be written to network */
2183 if (!c->ctnwrite && (c->msgcurr == c->msgused)) {
2184 if (!ms_update_event(c, EV_WRITE | EV_PERSIST)) {
2185 fprintf(stderr, "Couldn't update event.\n");
2186 ms_conn_set_state(c, conn_closing);
2187 break;
2188 }
2189 stop = true;
2190 break;
2191 }
2192
2193 switch (ms_transmit(c)) {
2194 case TRANSMIT_COMPLETE:
2195 /* we have no data to write to network, next wait repose */
2196 if (!ms_update_event(c, EV_READ | EV_PERSIST)) {
2197 fprintf(stderr, "Couldn't update event.\n");
2198 ms_conn_set_state(c, conn_closing);
2199 c->ctnwrite = false;
2200 break;
2201 }
2202 ms_conn_set_state(c, conn_read);
2203 c->ctnwrite = false;
2204 stop = true;
2205 break;
2206
2207 case TRANSMIT_INCOMPLETE:
2208 c->ctnwrite = true;
2209 break; /* Continue in state machine. */
2210
2211 case TRANSMIT_HARD_ERROR:
2212 c->ctnwrite = false;
2213 break;
2214
2215 case TRANSMIT_SOFT_ERROR:
2216 c->ctnwrite = true;
2217 stop = true;
2218 break;
2219
2220 default:
2221 break;
2222 } /* switch */
2223
2224 break;
2225
2226 case conn_closing:
2227 /* recovery mode, need reconnect if connection close */
2228 if (ms_setting.reconnect
2229 && (!ms_global.time_out || ((ms_setting.run_time == 0) && (c->remain_exec_num > 0))))
2230 {
2231 if (ms_reconn(c)) {
2232 ms_conn_close(c);
2233 stop = true;
2234 break;
2235 }
2236
2237 ms_reset_conn(c, false);
2238
2239 if (c->total_sfds == 1) {
2240 if (!ms_update_event(c, EV_WRITE | EV_PERSIST)) {
2241 fprintf(stderr, "Couldn't update event.\n");
2242 ms_conn_set_state(c, conn_closing);
2243 break;
2244 }
2245 }
2246
2247 break;
2248 } else {
2249 ms_conn_close(c);
2250 stop = true;
2251 break;
2252 }
2253
2254 default:
2255 assert(0);
2256 } /* switch */
2257 }
2258 } /* ms_drive_machine */
2259
2260 /**
2261 * the event handler of each thread
2262 *
2263 * @param fd, the file descriptor of socket
2264 * @param which, event flag
2265 * @param arg, argument
2266 */
2267 void ms_event_handler(const int fd, const short which, void *arg) {
2268 ms_conn_t *c = (ms_conn_t *) arg;
2269
2270 assert(c);
2271
2272 c->which = which;
2273
2274 /* sanity */
2275 if (fd != c->sfd) {
2276 fprintf(stderr, "Catastrophic: event fd: %d doesn't match conn fd: %d\n", fd, c->sfd);
2277 ms_conn_close(c);
2278 exit(1);
2279 }
2280 assert(fd == c->sfd);
2281
2282 ms_drive_machine(c);
2283
2284 /* wait for next event */
2285 } /* ms_event_handler */
2286
2287 /**
2288 * get the next socket descriptor index to run for replication
2289 *
2290 * @param c, pointer of the concurrency
2291 * @param cmd, command(get or set )
2292 *
2293 * @return int, if success, return the index, else return EXIT_SUCCESS
2294 */
2295 static uint32_t ms_get_rep_sock_index(ms_conn_t *c, int cmd) {
2296 uint32_t sock_index = 0;
2297 uint32_t i = 0;
2298
2299 if (c->total_sfds == 1) {
2300 return EXIT_SUCCESS;
2301 }
2302
2303 if (ms_setting.rep_write_srv == 0) {
2304 return sock_index;
2305 }
2306
2307 do {
2308 if (cmd == CMD_SET) {
2309 for (i = 0; i < ms_setting.rep_write_srv; i++) {
2310 if (c->tcpsfd[i] > 0) {
2311 break;
2312 }
2313 }
2314
2315 if (i == ms_setting.rep_write_srv) {
2316 /* random get one replication server to read */
2317 sock_index = (uint32_t) random() % c->total_sfds;
2318 } else {
2319 /* random get one replication writing server to write */
2320 sock_index = (uint32_t) random() % ms_setting.rep_write_srv;
2321 }
2322 } else if (cmd == CMD_GET) {
2323 /* random get one replication server to read */
2324 sock_index = (uint32_t) random() % c->total_sfds;
2325 }
2326 } while (c->tcpsfd[sock_index] == 0);
2327
2328 return sock_index;
2329 } /* ms_get_rep_sock_index */
2330
2331 /**
2332 * get the next socket descriptor index to run
2333 *
2334 * @param c, pointer of the concurrency
2335 *
2336 * @return int, return the index
2337 */
2338 static uint32_t ms_get_next_sock_index(ms_conn_t *c) {
2339 uint32_t sock_index = 0;
2340
2341 do {
2342 sock_index = (++c->cur_idx == c->total_sfds) ? 0 : c->cur_idx;
2343 } while (c->tcpsfd[sock_index] == 0);
2344
2345 return sock_index;
2346 } /* ms_get_next_sock_index */
2347
2348 /**
2349 * update socket event of the connections
2350 *
2351 * @param c, pointer of the concurrency
2352 *
2353 * @return int, if success, return EXIT_SUCCESS, else return -1
2354 */
2355 static int ms_update_conn_sock_event(ms_conn_t *c) {
2356 assert(c);
2357
2358 switch (c->currcmd.cmd) {
2359 case CMD_SET:
2360 if (ms_setting.facebook_test && c->udp) {
2361 c->sfd = c->tcpsfd[0];
2362 c->udp = false;
2363 c->change_sfd = true;
2364 }
2365 break;
2366
2367 case CMD_GET:
2368 if (ms_setting.facebook_test && !c->udp) {
2369 c->sfd = c->udpsfd;
2370 c->udp = true;
2371 c->change_sfd = true;
2372 }
2373 break;
2374
2375 default:
2376 break;
2377 } /* switch */
2378
2379 if (!c->udp && (c->total_sfds > 1)) {
2380 if (c->cur_idx != c->total_sfds) {
2381 if (ms_setting.rep_write_srv == 0) {
2382 c->cur_idx = ms_get_next_sock_index(c);
2383 } else {
2384 c->cur_idx = ms_get_rep_sock_index(c, c->currcmd.cmd);
2385 }
2386 } else {
2387 /* must select the first sock of the connection at the beginning */
2388 c->cur_idx = 0;
2389 }
2390
2391 c->sfd = c->tcpsfd[c->cur_idx];
2392 assert(c->sfd);
2393 c->change_sfd = true;
2394 }
2395
2396 if (c->change_sfd) {
2397 if (!ms_update_event(c, EV_WRITE | EV_PERSIST)) {
2398 fprintf(stderr, "Couldn't update event.\n");
2399 ms_conn_set_state(c, conn_closing);
2400 return -1;
2401 }
2402 }
2403
2404 return EXIT_SUCCESS;
2405 } /* ms_update_conn_sock_event */
2406
2407 /**
2408 * for ASCII protocol, this function build the set command
2409 * string and send the command.
2410 *
2411 * @param c, pointer of the concurrency
2412 * @param item, pointer of task item which includes the object
2413 * information
2414 *
2415 * @return int, if success, return EXIT_SUCCESS, else return -1
2416 */
2417 static int ms_build_ascii_write_buf_set(ms_conn_t *c, ms_task_item_t *item) {
2418 int value_offset;
2419 int write_len;
2420 char *buffer = c->wbuf;
2421
2422 write_len = snprintf(buffer, c->wsize, " %u %d %d\r\n", 0, item->exp_time, item->value_size);
2423
2424 if (write_len > c->wsize || write_len < 0) {
2425 /* ought to be always enough. just fail for simplicity */
2426 fprintf(stderr, "output command line too long.\n");
2427 return -1;
2428 }
2429
2430 if (item->value_offset == INVALID_OFFSET) {
2431 value_offset = item->key_suffix_offset;
2432 } else {
2433 value_offset = item->value_offset;
2434 }
2435
2436 if ((ms_add_iov(c, "set ", 4))
2437 || (ms_add_iov(c, (char *) &item->key_prefix, (int) KEY_PREFIX_SIZE))
2438 || (ms_add_iov(c, &ms_setting.char_block[item->key_suffix_offset],
2439 item->key_size - (int) KEY_PREFIX_SIZE)
2440 )
2441 || (ms_add_iov(c, buffer, write_len))
2442 || (ms_add_iov(c, &ms_setting.char_block[value_offset], item->value_size))
2443 || (ms_add_iov(c, "\r\n", 2)) || (c->udp && (ms_build_udp_headers(c))))
2444 {
2445 return -1;
2446 }
2447
2448 return EXIT_SUCCESS;
2449 } /* ms_build_ascii_write_buf_set */
2450
2451 /**
2452 * used to send set command to server
2453 *
2454 * @param c, pointer of the concurrency
2455 * @param item, pointer of task item which includes the object
2456 * information
2457 *
2458 * @return int, if success, return EXIT_SUCCESS, else return -1
2459 */
2460 int ms_mcd_set(ms_conn_t *c, ms_task_item_t *item) {
2461 assert(c);
2462
2463 c->currcmd.cmd = CMD_SET;
2464 c->currcmd.isfinish = false;
2465 c->currcmd.retstat = MCD_FAILURE;
2466
2467 if (ms_update_conn_sock_event(c)) {
2468 return -1;
2469 }
2470
2471 c->msgcurr = 0;
2472 c->msgused = 0;
2473 c->iovused = 0;
2474 if (ms_add_msghdr(c)) {
2475 fprintf(stderr, "Out of memory preparing request.");
2476 return -1;
2477 }
2478
2479 /* binary protocol */
2480 if (c->protocol == binary_prot) {
2481 if (ms_build_bin_write_buf_set(c, item)) {
2482 return -1;
2483 }
2484 } else {
2485 if (ms_build_ascii_write_buf_set(c, item)) {
2486 return -1;
2487 }
2488 }
2489
2490 atomic_add_size(&ms_stats.obj_bytes, item->key_size + item->value_size);
2491 atomic_add_size(&ms_stats.cmd_set, 1);
2492
2493 return EXIT_SUCCESS;
2494 } /* ms_mcd_set */
2495
2496 /**
2497 * for ASCII protocol, this function build the get command
2498 * string and send the command.
2499 *
2500 * @param c, pointer of the concurrency
2501 * @param item, pointer of task item which includes the object
2502 * information
2503 *
2504 * @return int, if success, return EXIT_SUCCESS, else return -1
2505 */
2506 static int ms_build_ascii_write_buf_get(ms_conn_t *c, ms_task_item_t *item) {
2507 if ((ms_add_iov(c, "get ", 4))
2508 || (ms_add_iov(c, (char *) &item->key_prefix, (int) KEY_PREFIX_SIZE))
2509 || (ms_add_iov(c, &ms_setting.char_block[item->key_suffix_offset],
2510 item->key_size - (int) KEY_PREFIX_SIZE)
2511 )
2512 || (ms_add_iov(c, "\r\n", 2)) || (c->udp && (ms_build_udp_headers(c))))
2513 {
2514 return -1;
2515 }
2516
2517 return EXIT_SUCCESS;
2518 } /* ms_build_ascii_write_buf_get */
2519
2520 /**
2521 * used to send the get command to server
2522 *
2523 * @param c, pointer of the concurrency
2524 * @param item, pointer of task item which includes the object
2525 * information
2526 *
2527 * @return int, if success, return EXIT_SUCCESS, else return -1
2528 */
2529 int ms_mcd_get(ms_conn_t *c, ms_task_item_t *item) {
2530 assert(c);
2531
2532 c->currcmd.cmd = CMD_GET;
2533 c->currcmd.isfinish = false;
2534 c->currcmd.retstat = MCD_FAILURE;
2535
2536 if (ms_update_conn_sock_event(c)) {
2537 return -1;
2538 }
2539
2540 c->msgcurr = 0;
2541 c->msgused = 0;
2542 c->iovused = 0;
2543 if (ms_add_msghdr(c)) {
2544 fprintf(stderr, "Out of memory preparing request.");
2545 return -1;
2546 }
2547
2548 /* binary protocol */
2549 if (c->protocol == binary_prot) {
2550 if (ms_build_bin_write_buf_get(c, item)) {
2551 return -1;
2552 }
2553 } else {
2554 if (ms_build_ascii_write_buf_get(c, item)) {
2555 return -1;
2556 }
2557 }
2558
2559 atomic_add_size(&ms_stats.cmd_get, 1);
2560
2561 return EXIT_SUCCESS;
2562 } /* ms_mcd_get */
2563
2564 /**
2565 * for ASCII protocol, this function build the multi-get command
2566 * string and send the command.
2567 *
2568 * @param c, pointer of the concurrency
2569 *
2570 * @return int, if success, return EXIT_SUCCESS, else return -1
2571 */
2572 static int ms_build_ascii_write_buf_mlget(ms_conn_t *c) {
2573 ms_task_item_t *item;
2574
2575 if (ms_add_iov(c, "get", 3)) {
2576 return -1;
2577 }
2578
2579 for (int i = 0; i < c->mlget_task.mlget_num; i++) {
2580 item = c->mlget_task.mlget_item[i].item;
2581 assert(item);
2582 if ((ms_add_iov(c, " ", 1))
2583 || (ms_add_iov(c, (char *) &item->key_prefix, (int) KEY_PREFIX_SIZE))
2584 || (ms_add_iov(c, &ms_setting.char_block[item->key_suffix_offset],
2585 item->key_size - (int) KEY_PREFIX_SIZE)
2586 ))
2587 {
2588 return -1;
2589 }
2590 }
2591
2592 if ((ms_add_iov(c, "\r\n", 2)) || (c->udp && (ms_build_udp_headers(c)))) {
2593 return -1;
2594 }
2595
2596 return EXIT_SUCCESS;
2597 } /* ms_build_ascii_write_buf_mlget */
2598
2599 /**
2600 * used to send the multi-get command to server
2601 *
2602 * @param c, pointer of the concurrency
2603 *
2604 * @return int, if success, return EXIT_SUCCESS, else return -1
2605 */
2606 int ms_mcd_mlget(ms_conn_t *c) {
2607 ms_task_item_t *item;
2608
2609 assert(c);
2610 assert(c->mlget_task.mlget_num >= 1);
2611
2612 c->currcmd.cmd = CMD_GET;
2613 c->currcmd.isfinish = false;
2614 c->currcmd.retstat = MCD_FAILURE;
2615
2616 if (ms_update_conn_sock_event(c)) {
2617 return -1;
2618 }
2619
2620 c->msgcurr = 0;
2621 c->msgused = 0;
2622 c->iovused = 0;
2623 if (ms_add_msghdr(c)) {
2624 fprintf(stderr, "Out of memory preparing request.");
2625 return -1;
2626 }
2627
2628 /* binary protocol */
2629 if (c->protocol == binary_prot) {
2630 if (ms_build_bin_write_buf_mlget(c)) {
2631 return -1;
2632 }
2633 } else {
2634 if (ms_build_ascii_write_buf_mlget(c)) {
2635 return -1;
2636 }
2637 }
2638
2639 /* decrease operation time of each item */
2640 for (int i = 0; i < c->mlget_task.mlget_num; i++) {
2641 item = c->mlget_task.mlget_item[i].item;
2642 atomic_add_size(&ms_stats.cmd_get, 1);
2643 }
2644
2645 (void) item;
2646
2647 return EXIT_SUCCESS;
2648 } /* ms_mcd_mlget */
2649
2650 /**
2651 * binary protocol support
2652 */
2653
2654 /**
2655 * for binary protocol, parse the response of server
2656 *
2657 * @param c, pointer of the concurrency
2658 *
2659 * @return int, if success, return EXIT_SUCCESS, else return -1
2660 */
2661 static int ms_bin_process_response(ms_conn_t *c) {
2662 const char *errstr = NULL;
2663
2664 assert(c);
2665
2666 uint32_t bodylen = c->binary_header.response.bodylen;
2667 uint8_t opcode = c->binary_header.response.opcode;
2668 uint16_t status = c->binary_header.response.status;
2669
2670 if (bodylen > 0) {
2671 c->rvbytes = (int32_t) bodylen;
2672 c->readval = true;
2673 return EXIT_FAILURE;
2674 } else {
2675 switch (status) {
2676 case PROTOCOL_BINARY_RESPONSE_SUCCESS:
2677 if (opcode == PROTOCOL_BINARY_CMD_SET) {
2678 c->currcmd.retstat = MCD_STORED;
2679 } else if (opcode == PROTOCOL_BINARY_CMD_DELETE) {
2680 c->currcmd.retstat = MCD_DELETED;
2681 } else if (opcode == PROTOCOL_BINARY_CMD_GET) {
2682 c->currcmd.retstat = MCD_END;
2683 }
2684 break;
2685
2686 case PROTOCOL_BINARY_RESPONSE_ENOMEM:
2687 errstr = "Out of memory";
2688 c->currcmd.retstat = MCD_SERVER_ERROR;
2689 break;
2690
2691 case PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND:
2692 errstr = "Unknown command";
2693 c->currcmd.retstat = MCD_UNKNOWN_READ_FAILURE;
2694 break;
2695
2696 case PROTOCOL_BINARY_RESPONSE_KEY_ENOENT:
2697 errstr = "Not found";
2698 c->currcmd.retstat = MCD_NOTFOUND;
2699 break;
2700
2701 case PROTOCOL_BINARY_RESPONSE_EINVAL:
2702 errstr = "Invalid arguments";
2703 c->currcmd.retstat = MCD_PROTOCOL_ERROR;
2704 break;
2705
2706 case PROTOCOL_BINARY_RESPONSE_KEY_EEXISTS:
2707 errstr = "Data exists for key.";
2708 break;
2709
2710 case PROTOCOL_BINARY_RESPONSE_E2BIG:
2711 errstr = "Too large.";
2712 c->currcmd.retstat = MCD_SERVER_ERROR;
2713 break;
2714
2715 case PROTOCOL_BINARY_RESPONSE_NOT_STORED:
2716 errstr = "Not stored.";
2717 c->currcmd.retstat = MCD_NOTSTORED;
2718 break;
2719
2720 default:
2721 errstr = "Unknown error";
2722 c->currcmd.retstat = MCD_UNKNOWN_READ_FAILURE;
2723 break;
2724 } /* switch */
2725
2726 if (errstr) {
2727 fprintf(stderr, "%s\n", errstr);
2728 }
2729 }
2730
2731 return EXIT_SUCCESS;
2732 } /* ms_bin_process_response */
2733
2734 /* build binary header and add the header to the buffer to send */
2735
2736 /**
2737 * build binary header and add the header to the buffer to send
2738 *
2739 * @param c, pointer of the concurrency
2740 * @param opcode, operation code
2741 * @param hdr_len, length of header
2742 * @param key_len, length of key
2743 * @param body_len. length of body
2744 */
2745 static void ms_add_bin_header(ms_conn_t *c, uint8_t opcode, uint8_t hdr_len, uint16_t key_len,
2746 uint32_t body_len) {
2747 protocol_binary_request_header *header;
2748
2749 assert(c);
2750
2751 header = (protocol_binary_request_header *) c->wcurr;
2752
2753 header->request.magic = (uint8_t) PROTOCOL_BINARY_REQ;
2754 header->request.opcode = (uint8_t) opcode;
2755 header->request.keylen = htons(key_len);
2756
2757 header->request.extlen = (uint8_t) hdr_len;
2758 header->request.datatype = (uint8_t) PROTOCOL_BINARY_RAW_BYTES;
2759 header->request.vbucket = 0;
2760
2761 header->request.bodylen = htonl(body_len);
2762 header->request.opaque = 0;
2763 header->request.cas = 0;
2764
2765 ms_add_iov(c, c->wcurr, sizeof(header->request));
2766 } /* ms_add_bin_header */
2767
2768 /**
2769 * add the key to the socket write buffer array
2770 *
2771 * @param c, pointer of the concurrency
2772 * @param item, pointer of task item which includes the object
2773 * information
2774 */
2775 static void ms_add_key_to_iov(ms_conn_t *c, ms_task_item_t *item) {
2776 ms_add_iov(c, (char *) &item->key_prefix, (int) KEY_PREFIX_SIZE);
2777 ms_add_iov(c, &ms_setting.char_block[item->key_suffix_offset],
2778 item->key_size - (int) KEY_PREFIX_SIZE);
2779 }
2780
2781 /**
2782 * for binary protocol, this function build the set command
2783 * and add the command to send buffer array.
2784 *
2785 * @param c, pointer of the concurrency
2786 * @param item, pointer of task item which includes the object
2787 * information
2788 *
2789 * @return int, if success, return EXIT_SUCCESS, else return -1
2790 */
2791 static int ms_build_bin_write_buf_set(ms_conn_t *c, ms_task_item_t *item) {
2792 assert(c->wbuf == c->wcurr);
2793
2794 int value_offset;
2795 protocol_binary_request_set *rep = (protocol_binary_request_set *) c->wcurr;
2796 uint16_t keylen = (uint16_t) item->key_size;
2797 uint32_t bodylen =
2798 (uint32_t) sizeof(rep->message.body) + (uint32_t) keylen + (uint32_t) item->value_size;
2799
2800 ms_add_bin_header(c, PROTOCOL_BINARY_CMD_SET, sizeof(rep->message.body), keylen, bodylen);
2801 rep->message.body.flags = 0;
2802 rep->message.body.expiration = htonl((uint32_t) item->exp_time);
2803 ms_add_iov(c, &rep->message.body, sizeof(rep->message.body));
2804 ms_add_key_to_iov(c, item);
2805
2806 if (item->value_offset == INVALID_OFFSET) {
2807 value_offset = item->key_suffix_offset;
2808 } else {
2809 value_offset = item->value_offset;
2810 }
2811 ms_add_iov(c, &ms_setting.char_block[value_offset], item->value_size);
2812
2813 return EXIT_SUCCESS;
2814 } /* ms_build_bin_write_buf_set */
2815
2816 /**
2817 * for binary protocol, this function build the get command and
2818 * add the command to send buffer array.
2819 *
2820 * @param c, pointer of the concurrency
2821 * @param item, pointer of task item which includes the object
2822 * information
2823 *
2824 * @return int, if success, return EXIT_SUCCESS, else return -1
2825 */
2826 static int ms_build_bin_write_buf_get(ms_conn_t *c, ms_task_item_t *item) {
2827 assert(c->wbuf == c->wcurr);
2828
2829 ms_add_bin_header(c, PROTOCOL_BINARY_CMD_GET, 0, (uint16_t) item->key_size,
2830 (uint32_t) item->key_size);
2831 ms_add_key_to_iov(c, item);
2832
2833 return EXIT_SUCCESS;
2834 } /* ms_build_bin_write_buf_get */
2835
2836 /**
2837 * for binary protocol, this function build the multi-get
2838 * command and add the command to send buffer array.
2839 *
2840 * @param c, pointer of the concurrency
2841 * @param item, pointer of task item which includes the object
2842 * information
2843 *
2844 * @return int, if success, return EXIT_SUCCESS, else return -1
2845 */
2846 static int ms_build_bin_write_buf_mlget(ms_conn_t *c) {
2847 ms_task_item_t *item;
2848
2849 assert(c->wbuf == c->wcurr);
2850
2851 for (int i = 0; i < c->mlget_task.mlget_num; i++) {
2852 item = c->mlget_task.mlget_item[i].item;
2853 assert(item);
2854
2855 ms_add_bin_header(c, PROTOCOL_BINARY_CMD_GET, 0, (uint16_t) item->key_size,
2856 (uint32_t) item->key_size);
2857 ms_add_key_to_iov(c, item);
2858 c->wcurr += sizeof(protocol_binary_request_get);
2859 }
2860
2861 c->wcurr = c->wbuf;
2862
2863 return EXIT_SUCCESS;
2864 } /* ms_build_bin_write_buf_mlget */
Resurrection of libmemcached https://launchpad.net/libmemcached