3 * Author: Mingqiang Zhuang
5 * Created on February 10, 2009
7 * (c) Copyright 2009, Schooner Information Technology, Inc.
8 * http://www.schoonerinfotech.com/
12 #include "mem_config.h"
20 #include <netinet/tcp.h>
21 #include <netinet/in.h>
22 #include <arpa/inet.h>
24 #if defined(HAVE_SYS_TIME_H)
25 # include <sys/time.h>
28 #if defined(HAVE_TIME_H)
32 #include "ms_setting.h"
33 #include "ms_thread.h"
34 #include "ms_atomic.h"
37 /* /usr/include/netinet/in.h defines macros from ntohs() to _bswap_nn to
38 * optimize the conversion functions, but the prototypes generate warnings
39 * from gcc. The conversion methods isn't the bottleneck for my app, so
40 * just remove the warnings by undef'ing the optimization ..
48 /* for network write */
49 #define TRANSMIT_COMPLETE 0
50 #define TRANSMIT_INCOMPLETE 1
51 #define TRANSMIT_SOFT_ERROR 2
52 #define TRANSMIT_HARD_ERROR 3
54 /* for generating key */
55 #define KEY_PREFIX_BASE 0x1010101010101010 /* not include ' ' '\r' '\n' '\0' */
56 #define KEY_PREFIX_MASK 0x1010101010101010
58 /* For parse the value length return by server */
60 #define VALUELEN_TOKEN 3
62 /* global increasing counter, to ensure the key prefix unique */
63 static uint64_t key_prefix_seq
= KEY_PREFIX_BASE
;
65 /* global increasing counter, generating request id for UDP */
66 static volatile uint32_t udp_request_id
= 0;
68 extern pthread_key_t ms_thread_key
;
70 /* generate upd request id */
71 static uint32_t ms_get_udp_request_id(void);
74 /* connect initialize */
75 static void ms_task_init(ms_conn_t
*c
);
76 static int ms_conn_udp_init(ms_conn_t
*c
, const bool is_udp
);
77 static int ms_conn_sock_init(ms_conn_t
*c
);
78 static int ms_conn_event_init(ms_conn_t
*c
);
79 static int ms_conn_init(ms_conn_t
*c
,
81 const int read_buffer_size
,
83 static void ms_warmup_num_init(ms_conn_t
*c
);
84 static int ms_item_win_init(ms_conn_t
*c
);
87 /* connection close */
88 void ms_conn_free(ms_conn_t
*c
);
89 static void ms_conn_close(ms_conn_t
*c
);
92 /* create network connection */
93 static int ms_new_socket(struct addrinfo
*ai
);
94 static void ms_maximize_sndbuf(const int sfd
);
95 static int ms_network_connect(ms_conn_t
*c
,
100 static int ms_reconn(ms_conn_t
*c
);
104 static int ms_tokenize_command(char *command
,
106 const int max_tokens
);
107 static int ms_ascii_process_line(ms_conn_t
*c
, char *command
);
108 static int ms_try_read_line(ms_conn_t
*c
);
109 static int ms_sort_udp_packet(ms_conn_t
*c
, char *buf
, int rbytes
);
110 static int ms_udp_read(ms_conn_t
*c
, char *buf
, int len
);
111 static int ms_try_read_network(ms_conn_t
*c
);
112 static void ms_verify_value(ms_conn_t
*c
,
113 ms_mlget_task_item_t
*mlget_item
,
116 static void ms_ascii_complete_nread(ms_conn_t
*c
);
117 static void ms_bin_complete_nread(ms_conn_t
*c
);
118 static void ms_complete_nread(ms_conn_t
*c
);
122 static int ms_add_msghdr(ms_conn_t
*c
);
123 static int ms_ensure_iov_space(ms_conn_t
*c
);
124 static int ms_add_iov(ms_conn_t
*c
, const void *buf
, int len
);
125 static int ms_build_udp_headers(ms_conn_t
*c
);
126 static int ms_transmit(ms_conn_t
*c
);
129 /* status adjustment */
130 static void ms_conn_shrink(ms_conn_t
*c
);
131 static void ms_conn_set_state(ms_conn_t
*c
, int state
);
132 static bool ms_update_event(ms_conn_t
*c
, const int new_flags
);
133 static uint32_t ms_get_rep_sock_index(ms_conn_t
*c
, int cmd
);
134 static uint32_t ms_get_next_sock_index(ms_conn_t
*c
);
135 static int ms_update_conn_sock_event(ms_conn_t
*c
);
136 static bool ms_need_yield(ms_conn_t
*c
);
137 static void ms_update_start_time(ms_conn_t
*c
);
141 static void ms_drive_machine(ms_conn_t
*c
);
142 void ms_event_handler(const int fd
, const short which
, void *arg
);
146 static int ms_build_ascii_write_buf_set(ms_conn_t
*c
, ms_task_item_t
*item
);
147 static int ms_build_ascii_write_buf_get(ms_conn_t
*c
, ms_task_item_t
*item
);
148 static int ms_build_ascii_write_buf_mlget(ms_conn_t
*c
);
151 /* binary protocol */
152 static int ms_bin_process_response(ms_conn_t
*c
);
153 static void ms_add_bin_header(ms_conn_t
*c
,
158 static void ms_add_key_to_iov(ms_conn_t
*c
, ms_task_item_t
*item
);
159 static int ms_build_bin_write_buf_set(ms_conn_t
*c
, ms_task_item_t
*item
);
160 static int ms_build_bin_write_buf_get(ms_conn_t
*c
, ms_task_item_t
*item
);
161 static int ms_build_bin_write_buf_mlget(ms_conn_t
*c
);
165 * each key has two parts, prefix and suffix. The suffix is a
166 * string random get form the character table. The prefix is a
167 * uint64_t variable. And the prefix must be unique. we use the
168 * prefix to identify a key. And the prefix can't include
169 * character ' ' '\r' '\n' '\0'.
173 uint64_t ms_get_key_prefix(void)
177 pthread_mutex_lock(&ms_global
.seq_mutex
);
178 key_prefix_seq
|= KEY_PREFIX_MASK
;
179 key_prefix
= key_prefix_seq
;
181 pthread_mutex_unlock(&ms_global
.seq_mutex
);
184 } /* ms_get_key_prefix */
188 * get an unique udp request id
190 * @return an unique UDP request id
192 static uint32_t ms_get_udp_request_id(void)
194 return atomic_add_32_nv(&udp_request_id
, 1);
199 * initialize current task structure
201 * @param c, pointer of the concurrency
203 static void ms_task_init(ms_conn_t
*c
)
205 c
->curr_task
.cmd
= CMD_NULL
;
206 c
->curr_task
.item
= 0;
207 c
->curr_task
.verify
= false;
208 c
->curr_task
.finish_verify
= true;
209 c
->curr_task
.get_miss
= true;
211 c
->curr_task
.get_opt
= 0;
212 c
->curr_task
.set_opt
= 0;
213 c
->curr_task
.cycle_undo_get
= 0;
214 c
->curr_task
.cycle_undo_set
= 0;
215 c
->curr_task
.verified_get
= 0;
216 c
->curr_task
.overwrite_set
= 0;
221 * initialize udp for the connection structure
223 * @param c, pointer of the concurrency
224 * @param is_udp, whether it's udp
226 * @return int, if success, return EXIT_SUCCESS, else return -1
228 static int ms_conn_udp_init(ms_conn_t
*c
, const bool is_udp
)
234 c
->rudpsize
= UDP_DATA_BUFFER_SIZE
;
245 if (c
->udp
|| (! c
->udp
&& ms_setting
.facebook_test
))
247 c
->rudpbuf
= (char *)malloc((size_t)c
->rudpsize
);
248 c
->udppkt
= (ms_udppkt_t
*)malloc(MAX_UDP_PACKET
* sizeof(ms_udppkt_t
));
250 if ((c
->rudpbuf
== NULL
) || (c
->udppkt
== NULL
))
252 if (c
->rudpbuf
!= NULL
)
254 if (c
->udppkt
!= NULL
)
256 fprintf(stderr
, "malloc()\n");
259 memset(c
->udppkt
, 0, MAX_UDP_PACKET
* sizeof(ms_udppkt_t
));
263 } /* ms_conn_udp_init */
267 * initialize the connection structure
269 * @param c, pointer of the concurrency
270 * @param init_state, (conn_read, conn_write, conn_closing)
271 * @param read_buffer_size
272 * @param is_udp, whether it's udp
274 * @return int, if success, return EXIT_SUCCESS, else return -1
276 static int ms_conn_init(ms_conn_t
*c
,
277 const int init_state
,
278 const int read_buffer_size
,
287 c
->rsize
= read_buffer_size
;
288 c
->wsize
= WRITE_BUFFER_SIZE
;
289 c
->iovsize
= IOV_LIST_INITIAL
;
290 c
->msgsize
= MSG_LIST_INITIAL
;
292 /* for replication, each connection need connect all the server */
293 if (ms_setting
.rep_write_srv
> 0)
295 c
->total_sfds
= ms_setting
.srv_cnt
* ms_setting
.sock_per_conn
;
299 c
->total_sfds
= ms_setting
.sock_per_conn
;
303 c
->rbuf
= (char *)malloc((size_t)c
->rsize
);
304 c
->wbuf
= (char *)malloc((size_t)c
->wsize
);
305 c
->iov
= (struct iovec
*)malloc(sizeof(struct iovec
) * (size_t)c
->iovsize
);
306 c
->msglist
= (struct msghdr
*)malloc(
307 sizeof(struct msghdr
) * (size_t)c
->msgsize
);
308 if (ms_setting
.mult_key_num
> 1)
310 c
->mlget_task
.mlget_item
= (ms_mlget_task_item_t
*)
312 sizeof(ms_mlget_task_item_t
) * (size_t)ms_setting
.mult_key_num
);
314 c
->tcpsfd
= (int *)malloc((size_t)c
->total_sfds
* sizeof(int));
316 if ((c
->rbuf
== NULL
) || (c
->wbuf
== NULL
) || (c
->iov
== NULL
)
317 || (c
->msglist
== NULL
) || (c
->tcpsfd
== NULL
)
318 || ((ms_setting
.mult_key_num
> 1)
319 && (c
->mlget_task
.mlget_item
== NULL
)))
327 if (c
->msglist
!= NULL
)
329 if (c
->mlget_task
.mlget_item
!= NULL
)
330 free(c
->mlget_task
.mlget_item
);
331 if (c
->tcpsfd
!= NULL
)
333 fprintf(stderr
, "malloc()\n");
337 c
->state
= init_state
;
345 c
->cur_idx
= c
->total_sfds
; /* default index is a invalid value */
349 c
->change_sfd
= false;
351 c
->precmd
.cmd
= c
->currcmd
.cmd
= CMD_NULL
;
352 c
->precmd
.isfinish
= true; /* default the previous command finished */
353 c
->currcmd
.isfinish
= false;
354 c
->precmd
.retstat
= c
->currcmd
.retstat
= MCD_FAILURE
;
355 c
->precmd
.key_prefix
= c
->currcmd
.key_prefix
= 0;
357 c
->mlget_task
.mlget_num
= 0;
358 c
->mlget_task
.value_index
= -1; /* default invalid value */
360 if (ms_setting
.binary_prot_
)
362 c
->protocol
= binary_prot
;
366 c
->protocol
= ascii_prot
;
370 if (ms_conn_udp_init(c
, is_udp
) != 0)
375 /* initialize task */
378 if (! (ms_setting
.facebook_test
&& is_udp
))
380 atomic_add_32(&ms_stats
.active_conns
, 1);
388 * when doing 100% get operation, it could preset some objects
389 * to warmup the server. this function is used to initialize the
390 * number of the objects to preset.
392 * @param c, pointer of the concurrency
394 static void ms_warmup_num_init(ms_conn_t
*c
)
396 /* no set operation, preset all the items in the window */
397 if (ms_setting
.cmd_distr
[CMD_SET
].cmd_prop
< PROP_ERROR
)
399 c
->warmup_num
= c
->win_size
;
400 c
->remain_warmup_num
= c
->warmup_num
;
405 c
->remain_warmup_num
= c
->warmup_num
;
407 } /* ms_warmup_num_init */
411 * each connection has an item window, this function initialize
412 * the window. The window is used to generate task.
414 * @param c, pointer of the concurrency
416 * @return int, if success, return EXIT_SUCCESS, else return -1
418 static int ms_item_win_init(ms_conn_t
*c
)
420 ms_thread_t
*ms_thread
= pthread_getspecific(ms_thread_key
);
423 c
->win_size
= (int)ms_setting
.win_size
;
425 c
->exec_num
= ms_thread
->thread_ctx
->exec_num_perconn
;
426 c
->remain_exec_num
= c
->exec_num
;
428 c
->item_win
= (ms_task_item_t
*)malloc(
429 sizeof(ms_task_item_t
) * (size_t)c
->win_size
);
430 if (c
->item_win
== NULL
)
432 fprintf(stderr
, "Can't allocate task item array for conn.\n");
435 memset(c
->item_win
, 0, sizeof(ms_task_item_t
) * (size_t)c
->win_size
);
437 for (int i
= 0; i
< c
->win_size
; i
++)
439 c
->item_win
[i
].key_size
= (int)ms_setting
.distr
[i
].key_size
;
440 c
->item_win
[i
].key_prefix
= ms_get_key_prefix();
441 c
->item_win
[i
].key_suffix_offset
= ms_setting
.distr
[i
].key_offset
;
442 c
->item_win
[i
].value_size
= (int)ms_setting
.distr
[i
].value_size
;
443 c
->item_win
[i
].value_offset
= INVALID_OFFSET
; /* default in invalid offset */
444 c
->item_win
[i
].client_time
= 0;
446 /* set expire time base on the proportion */
447 if (exp_cnt
< ms_setting
.exp_ver_per
* i
)
449 c
->item_win
[i
].exp_time
= FIXED_EXPIRE_TIME
;
454 c
->item_win
[i
].exp_time
= 0;
458 ms_warmup_num_init(c
);
461 } /* ms_item_win_init */
465 * each connection structure can include one or more sock
466 * handlers. this function create these socks and connect the
469 * @param c, pointer of the concurrency
471 * @return int, if success, return EXIT_SUCCESS, else return -1
473 static int ms_conn_sock_init(ms_conn_t
*c
)
475 ms_thread_t
*ms_thread
= pthread_getspecific(ms_thread_key
);
481 assert(c
->tcpsfd
!= NULL
);
483 for (i
= 0; i
< c
->total_sfds
; i
++)
486 if (ms_setting
.rep_write_srv
> 0)
488 /* for replication, each connection need connect all the server */
489 srv_idx
= i
% ms_setting
.srv_cnt
;
493 /* all the connections in a thread connects the same server */
494 srv_idx
= ms_thread
->thread_ctx
->srv_idx
;
497 if (ms_network_connect(c
, ms_setting
.servers
[srv_idx
].srv_host_name
,
498 ms_setting
.servers
[srv_idx
].srv_port
,
499 ms_setting
.udp
, &ret_sfd
) != 0)
509 if (! ms_setting
.udp
)
511 c
->tcpsfd
[i
]= ret_sfd
;
517 /* initialize udp sock handler if necessary */
518 if (ms_setting
.facebook_test
)
521 if (ms_network_connect(c
, ms_setting
.servers
[srv_idx
].srv_host_name
,
522 ms_setting
.servers
[srv_idx
].srv_port
,
523 true, &ret_sfd
) != 0)
533 if ((i
!= c
->total_sfds
) || (ms_setting
.facebook_test
&& (c
->udpsfd
== 0)))
541 for (uint32_t j
= 0; j
< i
; j
++)
556 } /* ms_conn_sock_init */
560 * each connection is managed by libevent, this function
561 * initialize the event of the connection structure.
563 * @param c, pointer of the concurrency
565 * @return int, if success, return EXIT_SUCCESS, else return -1
567 static int ms_conn_event_init(ms_conn_t
*c
)
569 ms_thread_t
*ms_thread
= pthread_getspecific(ms_thread_key
);
570 short event_flags
= EV_WRITE
| EV_PERSIST
;
572 event_set(&c
->event
, c
->sfd
, event_flags
, ms_event_handler
, (void *)c
);
573 event_base_set(ms_thread
->base
, &c
->event
);
574 c
->ev_flags
= event_flags
;
576 if (event_add(&c
->event
, NULL
) == -1)
582 } /* ms_conn_event_init */
586 * setup a connection, each connection structure of each
587 * thread must call this function to initialize.
589 * @param c, pointer of the concurrency
591 * @return int, if success, return EXIT_SUCCESS, else return -1
593 int ms_setup_conn(ms_conn_t
*c
)
595 if (ms_item_win_init(c
) != 0)
600 if (ms_conn_init(c
, conn_write
, DATA_BUFFER_SIZE
, ms_setting
.udp
) != 0)
605 if (ms_conn_sock_init(c
) != 0)
610 if (ms_conn_event_init(c
) != 0)
616 } /* ms_setup_conn */
620 * Frees a connection.
622 * @param c, pointer of the concurrency
624 void ms_conn_free(ms_conn_t
*c
)
626 ms_thread_t
*ms_thread
= pthread_getspecific(ms_thread_key
);
629 if (c
->hdrbuf
!= NULL
)
631 if (c
->msglist
!= NULL
)
639 if (c
->mlget_task
.mlget_item
!= NULL
)
640 free(c
->mlget_task
.mlget_item
);
641 if (c
->rudpbuf
!= NULL
)
643 if (c
->udppkt
!= NULL
)
645 if (c
->item_win
!= NULL
)
647 if (c
->tcpsfd
!= NULL
)
650 if (--ms_thread
->nactive_conn
== 0)
652 free(ms_thread
->conn
);
661 * @param c, pointer of the concurrency
663 static void ms_conn_close(ms_conn_t
*c
)
665 ms_thread_t
*ms_thread
= pthread_getspecific(ms_thread_key
);
668 /* delete the event, the socket and the connection */
669 event_del(&c
->event
);
671 for (uint32_t i
= 0; i
< c
->total_sfds
; i
++)
673 if (c
->tcpsfd
[i
] > 0)
680 if (ms_setting
.facebook_test
)
685 atomic_dec_32(&ms_stats
.active_conns
);
689 if (ms_setting
.run_time
== 0)
691 pthread_mutex_lock(&ms_global
.run_lock
.lock
);
692 ms_global
.run_lock
.count
++;
693 pthread_cond_signal(&ms_global
.run_lock
.cond
);
694 pthread_mutex_unlock(&ms_global
.run_lock
.lock
);
697 if (ms_thread
->nactive_conn
== 0)
701 } /* ms_conn_close */
707 * @param ai, server address information
709 * @return int, if success, return EXIT_SUCCESS, else return -1
711 static int ms_new_socket(struct addrinfo
*ai
)
715 if ((sfd
= socket(ai
->ai_family
, ai
->ai_socktype
, ai
->ai_protocol
)) == -1)
717 fprintf(stderr
, "socket() error: %s.\n", strerror(errno
));
722 } /* ms_new_socket */
726 * Sets a socket's send buffer size to the maximum allowed by the system.
728 * @param sfd, file descriptor of socket
730 static void ms_maximize_sndbuf(const int sfd
)
732 socklen_t intsize
= sizeof(int);
733 unsigned int last_good
= 0;
734 unsigned int min
, max
, avg
;
735 unsigned int old_size
;
737 /* Start with the default size. */
738 if (getsockopt(sfd
, SOL_SOCKET
, SO_SNDBUF
, &old_size
, &intsize
) != 0)
740 fprintf(stderr
, "getsockopt(SO_SNDBUF)\n");
744 /* Binary-search for the real maximum. */
746 max
= MAX_SENDBUF_SIZE
;
750 avg
= ((unsigned int)(min
+ max
)) / 2;
751 if (setsockopt(sfd
, SOL_SOCKET
, SO_SNDBUF
, (void *)&avg
, intsize
) == 0)
762 } /* ms_maximize_sndbuf */
766 * socket connects the server
768 * @param c, pointer of the concurrency
769 * @param srv_host_name, the host name of the server
770 * @param srv_port, port of server
771 * @param is_udp, whether it's udp
772 * @param ret_sfd, the connected socket file descriptor
774 * @return int, if success, return EXIT_SUCCESS, else return -1
776 static int ms_network_connect(ms_conn_t
*c
,
788 struct addrinfo
*next
;
789 struct addrinfo hints
;
790 char port_buf
[NI_MAXSERV
];
797 * the memset call clears nonstandard fields in some impementations
798 * that otherwise mess things up.
800 memset(&hints
, 0, sizeof(hints
));
802 hints
.ai_flags
= AI_PASSIVE
| AI_ADDRCONFIG
;
804 hints
.ai_flags
= AI_PASSIVE
;
805 #endif /* AI_ADDRCONFIG */
808 hints
.ai_protocol
= IPPROTO_UDP
;
809 hints
.ai_socktype
= SOCK_DGRAM
;
810 hints
.ai_family
= AF_INET
; /* This left here because of issues with OSX 10.5 */
814 hints
.ai_family
= AF_UNSPEC
;
815 hints
.ai_protocol
= IPPROTO_TCP
;
816 hints
.ai_socktype
= SOCK_STREAM
;
819 snprintf(port_buf
, NI_MAXSERV
, "%d", srv_port
);
820 error
= getaddrinfo(srv_host_name
, port_buf
, &hints
, &ai
);
823 if (error
!= EAI_SYSTEM
)
824 fprintf(stderr
, "getaddrinfo(): %s.\n", gai_strerror(error
));
826 perror("getaddrinfo()\n");
831 for (next
= ai
; next
; next
= next
->ai_next
)
833 if ((sfd
= ms_new_socket(next
)) == -1)
839 setsockopt(sfd
, SOL_SOCKET
, SO_REUSEADDR
, (void *)&flags
, sizeof(flags
));
842 ms_maximize_sndbuf(sfd
);
846 setsockopt(sfd
, SOL_SOCKET
, SO_KEEPALIVE
, (void *)&flags
,
848 setsockopt(sfd
, SOL_SOCKET
, SO_LINGER
, (void *)&ling
, sizeof(ling
));
849 setsockopt(sfd
, IPPROTO_TCP
, TCP_NODELAY
, (void *)&flags
,
855 c
->srv_recv_addr_size
= sizeof(struct sockaddr
);
856 memcpy(&c
->srv_recv_addr
, next
->ai_addr
, c
->srv_recv_addr_size
);
860 if (connect(sfd
, next
->ai_addr
, next
->ai_addrlen
) == -1)
868 if (((flags
= fcntl(sfd
, F_GETFL
, 0)) < 0)
869 || (fcntl(sfd
, F_SETFL
, flags
| O_NONBLOCK
) < 0))
871 fprintf(stderr
, "setting O_NONBLOCK\n");
887 /* Return zero if we detected no errors in starting up connections */
889 } /* ms_network_connect */
893 * reconnect a disconnected sock
895 * @param c, pointer of the concurrency
897 * @return int, if success, return EXIT_SUCCESS, else return -1
899 static int ms_reconn(ms_conn_t
*c
)
901 ms_thread_t
*ms_thread
= pthread_getspecific(ms_thread_key
);
903 uint32_t srv_conn_cnt
= 0;
905 if (ms_setting
.rep_write_srv
> 0)
907 srv_idx
= c
->cur_idx
% ms_setting
.srv_cnt
;
908 srv_conn_cnt
= ms_setting
.sock_per_conn
* ms_setting
.nconns
;
912 srv_idx
= ms_thread
->thread_ctx
->srv_idx
;
913 srv_conn_cnt
= ms_setting
.nconns
/ ms_setting
.srv_cnt
;
916 /* close the old socket handler */
918 c
->tcpsfd
[c
->cur_idx
]= 0;
920 if (atomic_add_32_nv(&ms_setting
.servers
[srv_idx
].disconn_cnt
, 1)
923 gettimeofday(&ms_setting
.servers
[srv_idx
].disconn_time
, NULL
);
924 fprintf(stderr
, "Server %s:%d disconnect\n",
925 ms_setting
.servers
[srv_idx
].srv_host_name
,
926 ms_setting
.servers
[srv_idx
].srv_port
);
929 if (ms_setting
.rep_write_srv
> 0)
933 for (i
= 0; i
< c
->total_sfds
; i
++)
935 if (c
->tcpsfd
[i
] != 0)
941 /* all socks disconnect */
942 if (i
== c
->total_sfds
)
951 /* reconnect success, break the loop */
952 if (ms_network_connect(c
, ms_setting
.servers
[srv_idx
].srv_host_name
,
953 ms_setting
.servers
[srv_idx
].srv_port
,
954 ms_setting
.udp
, &c
->sfd
) == 0)
956 c
->tcpsfd
[c
->cur_idx
]= c
->sfd
;
957 if (atomic_add_32_nv(&ms_setting
.servers
[srv_idx
].reconn_cnt
, 1)
958 % (uint32_t)srv_conn_cnt
== 0)
960 gettimeofday(&ms_setting
.servers
[srv_idx
].reconn_time
, NULL
);
962 (int)(ms_setting
.servers
[srv_idx
].reconn_time
.tv_sec
963 - ms_setting
.servers
[srv_idx
].disconn_time
965 fprintf(stderr
, "Server %s:%d reconnect after %ds\n",
966 ms_setting
.servers
[srv_idx
].srv_host_name
,
967 ms_setting
.servers
[srv_idx
].srv_port
, reconn_time
);
972 if (ms_setting
.rep_write_srv
== 0 && c
->total_sfds
> 0)
974 /* wait a second and reconnect */
978 while (ms_setting
.rep_write_srv
== 0 && c
->total_sfds
> 0);
981 if ((c
->total_sfds
> 1) && (c
->tcpsfd
[c
->cur_idx
] == 0))
992 * reconnect several disconnected socks in the connection
993 * structure, the ever-1-second timer of the thread will check
994 * whether some socks in the connections disconnect. if
995 * disconnect, reconnect the sock.
997 * @param c, pointer of the concurrency
999 * @return int, if success, return EXIT_SUCCESS, else return -1
1001 int ms_reconn_socks(ms_conn_t
*c
)
1003 ms_thread_t
*ms_thread
= pthread_getspecific(ms_thread_key
);
1004 uint32_t srv_idx
= 0;
1006 uint32_t srv_conn_cnt
= 0;
1007 struct timeval cur_time
;
1011 if ((c
->total_sfds
== 1) || (c
->total_sfds
== c
->alive_sfds
))
1013 return EXIT_SUCCESS
;
1016 for (uint32_t i
= 0; i
< c
->total_sfds
; i
++)
1018 if (c
->tcpsfd
[i
] == 0)
1020 gettimeofday(&cur_time
, NULL
);
1023 * For failover test of replication, reconnect the socks after
1024 * it disconnects more than 5 seconds, Otherwise memslap will
1025 * block at connect() function and the work threads can't work
1029 - ms_setting
.servers
[srv_idx
].disconn_time
.tv_sec
< 5)
1034 if (ms_setting
.rep_write_srv
> 0)
1036 srv_idx
= i
% ms_setting
.srv_cnt
;
1037 srv_conn_cnt
= ms_setting
.sock_per_conn
* ms_setting
.nconns
;
1041 srv_idx
= ms_thread
->thread_ctx
->srv_idx
;
1042 srv_conn_cnt
= ms_setting
.nconns
/ ms_setting
.srv_cnt
;
1045 if (ms_network_connect(c
, ms_setting
.servers
[srv_idx
].srv_host_name
,
1046 ms_setting
.servers
[srv_idx
].srv_port
,
1047 ms_setting
.udp
, &ret_sfd
) == 0)
1049 c
->tcpsfd
[i
]= ret_sfd
;
1052 if (atomic_add_32_nv(&ms_setting
.servers
[srv_idx
].reconn_cnt
, 1)
1053 % (uint32_t)srv_conn_cnt
== 0)
1055 gettimeofday(&ms_setting
.servers
[srv_idx
].reconn_time
, NULL
);
1057 (int)(ms_setting
.servers
[srv_idx
].reconn_time
.tv_sec
1058 - ms_setting
.servers
[srv_idx
].disconn_time
1060 fprintf(stderr
, "Server %s:%d reconnect after %ds\n",
1061 ms_setting
.servers
[srv_idx
].srv_host_name
,
1062 ms_setting
.servers
[srv_idx
].srv_port
, reconn_time
);
1068 return EXIT_SUCCESS
;
1069 } /* ms_reconn_socks */
1073 * Tokenize the command string by replacing whitespace with '\0' and update
1074 * the token array tokens with pointer to start of each token and length.
1075 * Returns total number of tokens. The last valid token is the terminal
1076 * token (value points to the first unprocessed character of the string and
1081 * while(ms_tokenize_command(command, ncommand, tokens, max_tokens) > 0) {
1082 * for(int ix = 0; tokens[ix].length != 0; ix++) {
1085 * ncommand = tokens[ix].value - command;
1086 * command = tokens[ix].value;
1089 * @param command, the command string to token
1090 * @param tokens, array to store tokens
1091 * @param max_tokens, maximum tokens number
1093 * @return int, the number of tokens
1095 static int ms_tokenize_command(char *command
,
1097 const int max_tokens
)
1102 assert(command
!= NULL
&& tokens
!= NULL
&& max_tokens
> 1);
1104 for (s
= e
= command
; ntokens
< max_tokens
- 1; ++e
)
1110 tokens
[ntokens
].value
= s
;
1111 tokens
[ntokens
].length
= (size_t)(e
- s
);
1117 else if (*e
== '\0')
1121 tokens
[ntokens
].value
= s
;
1122 tokens
[ntokens
].length
= (size_t)(e
- s
);
1126 break; /* string end */
1131 } /* ms_tokenize_command */
1135 * parse the response of server.
1137 * @param c, pointer of the concurrency
1138 * @param command, the string responded by server
1140 * @return int, if the command completed return EXIT_SUCCESS, else return
1143 static int ms_ascii_process_line(ms_conn_t
*c
, char *command
)
1147 char *buffer
= command
;
1152 * for command get, we store the returned value into local buffer
1153 * then continue in ms_complete_nread().
1158 case 'V': /* VALUE || VERSION */
1159 if (buffer
[1] == 'A') /* VALUE */
1161 token_t tokens
[MAX_TOKENS
];
1162 ms_tokenize_command(command
, tokens
, MAX_TOKENS
);
1164 value_len
= strtol(tokens
[VALUELEN_TOKEN
].value
, NULL
, 10);
1167 printf("<%d ERROR %s\n", c
->sfd
, strerror(errno
));
1169 c
->currcmd
.key_prefix
= *(uint64_t *)tokens
[KEY_TOKEN
].value
;
1172 * We read the \r\n into the string since not doing so is more
1173 * cycles then the waster of memory to do so.
1175 * We are null terminating through, which will most likely make
1176 * some people lazy about using the return length.
1178 c
->rvbytes
= (int)(value_len
+ 2);
1186 c
->currcmd
.retstat
= MCD_SUCCESS
;
1189 case 'S': /* STORED STATS SERVER_ERROR */
1190 if (buffer
[2] == 'A') /* STORED STATS */
1192 c
->currcmd
.retstat
= MCD_STAT
;
1194 else if (buffer
[1] == 'E')
1197 printf("<%d %s\n", c
->sfd
, buffer
);
1199 c
->currcmd
.retstat
= MCD_SERVER_ERROR
;
1201 else if (buffer
[1] == 'T')
1204 c
->currcmd
.retstat
= MCD_STORED
;
1208 c
->currcmd
.retstat
= MCD_UNKNOWN_READ_FAILURE
;
1212 case 'D': /* DELETED DATA */
1213 if (buffer
[1] == 'E')
1215 c
->currcmd
.retstat
= MCD_DELETED
;
1219 c
->currcmd
.retstat
= MCD_UNKNOWN_READ_FAILURE
;
1224 case 'N': /* NOT_FOUND NOT_STORED*/
1225 if (buffer
[4] == 'F')
1227 c
->currcmd
.retstat
= MCD_NOTFOUND
;
1229 else if (buffer
[4] == 'S')
1231 printf("<%d %s\n", c
->sfd
, buffer
);
1232 c
->currcmd
.retstat
= MCD_NOTSTORED
;
1236 c
->currcmd
.retstat
= MCD_UNKNOWN_READ_FAILURE
;
1240 case 'E': /* PROTOCOL ERROR or END */
1241 if (buffer
[1] == 'N')
1244 c
->currcmd
.retstat
= MCD_END
;
1246 else if (buffer
[1] == 'R')
1248 printf("<%d ERROR\n", c
->sfd
);
1249 c
->currcmd
.retstat
= MCD_PROTOCOL_ERROR
;
1251 else if (buffer
[1] == 'X')
1253 c
->currcmd
.retstat
= MCD_DATA_EXISTS
;
1254 printf("<%d %s\n", c
->sfd
, buffer
);
1258 c
->currcmd
.retstat
= MCD_UNKNOWN_READ_FAILURE
;
1262 case 'C': /* CLIENT ERROR */
1263 printf("<%d %s\n", c
->sfd
, buffer
);
1264 c
->currcmd
.retstat
= MCD_CLIENT_ERROR
;
1268 c
->currcmd
.retstat
= MCD_UNKNOWN_READ_FAILURE
;
1273 } /* ms_ascii_process_line */
1277 * after one operation completes, reset the concurrency
1279 * @param c, pointer of the concurrency
1280 * @param timeout, whether it's timeout
1282 void ms_reset_conn(ms_conn_t
*c
, bool timeout
)
1288 if ((c
->packets
> 0) && (c
->packets
< MAX_UDP_PACKET
))
1290 memset(c
->udppkt
, 0, sizeof(ms_udppkt_t
) * (size_t)c
->packets
);
1299 c
->currcmd
.isfinish
= true;
1306 ms_conn_set_state(c
, conn_write
);
1307 memcpy(&c
->precmd
, &c
->currcmd
, sizeof(ms_cmdstat_t
)); /* replicate command state */
1311 ms_drive_machine(c
);
1313 } /* ms_reset_conn */
1317 * if we have a complete line in the buffer, process it.
1319 * @param c, pointer of the concurrency
1321 * @return int, if success, return EXIT_SUCCESS, else return -1
1323 static int ms_try_read_line(ms_conn_t
*c
)
1325 if (c
->protocol
== binary_prot
)
1327 /* Do we have the complete packet header? */
1328 if ((uint64_t)c
->rbytes
< sizeof(c
->binary_header
))
1330 /* need more data! */
1331 return EXIT_SUCCESS
;
1336 if (((long)(c
->rcurr
)) % 8 != 0)
1338 /* must realign input buffer */
1339 memmove(c
->rbuf
, c
->rcurr
, c
->rbytes
);
1341 if (settings
.verbose
)
1343 fprintf(stderr
, "%d: Realign input buffer.\n", c
->sfd
);
1347 protocol_binary_response_header
*rsp
;
1348 rsp
= (protocol_binary_response_header
*)c
->rcurr
;
1350 c
->binary_header
= *rsp
;
1351 c
->binary_header
.response
.extlen
= rsp
->response
.extlen
;
1352 c
->binary_header
.response
.keylen
= ntohs(rsp
->response
.keylen
);
1353 c
->binary_header
.response
.bodylen
= ntohl(rsp
->response
.bodylen
);
1354 c
->binary_header
.response
.status
= ntohs(rsp
->response
.status
);
1356 if (c
->binary_header
.response
.magic
!= PROTOCOL_BINARY_RES
)
1358 fprintf(stderr
, "Invalid magic: %x\n",
1359 c
->binary_header
.response
.magic
);
1360 ms_conn_set_state(c
, conn_closing
);
1361 return EXIT_SUCCESS
;
1364 /* process this complete response */
1365 if (ms_bin_process_response(c
) == 0)
1367 /* current operation completed */
1368 ms_reset_conn(c
, false);
1373 c
->rbytes
-= (int32_t)sizeof(c
->binary_header
);
1374 c
->rcurr
+= sizeof(c
->binary_header
);
1383 assert(c
->rcurr
<= (c
->rbuf
+ c
->rsize
));
1386 return EXIT_SUCCESS
;
1388 el
= memchr(c
->rcurr
, '\n', (size_t)c
->rbytes
);
1390 return EXIT_SUCCESS
;
1393 if (((el
- c
->rcurr
) > 1) && (*(el
- 1) == '\r'))
1399 assert(cont
<= (c
->rcurr
+ c
->rbytes
));
1401 /* process this complete line */
1402 if (ms_ascii_process_line(c
, c
->rcurr
) == 0)
1404 /* current operation completed */
1405 ms_reset_conn(c
, false);
1410 /* current operation didn't complete */
1411 c
->rbytes
-= (int32_t)(cont
- c
->rcurr
);
1415 assert(c
->rcurr
<= (c
->rbuf
+ c
->rsize
));
1419 } /* ms_try_read_line */
1423 * because the packet of UDP can't ensure the order, the
1424 * function is used to sort the received udp packet.
1426 * @param c, pointer of the concurrency
1427 * @param buf, the buffer to store the ordered packages data
1428 * @param rbytes, the maximum capacity of the buffer
1430 * @return int, if success, return the copy bytes, else return
1433 static int ms_sort_udp_packet(ms_conn_t
*c
, char *buf
, int rbytes
)
1438 uint16_t seq_num
= 0;
1439 uint16_t packets
= 0;
1440 unsigned char *header
= NULL
;
1442 /* no enough data */
1444 assert(buf
!= NULL
);
1445 assert(c
->rudpbytes
>= UDP_HEADER_SIZE
);
1447 /* calculate received packets count */
1448 if (c
->rudpbytes
% UDP_MAX_PAYLOAD_SIZE
>= UDP_HEADER_SIZE
)
1450 /* the last packet has some data */
1451 c
->recvpkt
= c
->rudpbytes
/ UDP_MAX_PAYLOAD_SIZE
+ 1;
1455 c
->recvpkt
= c
->rudpbytes
/ UDP_MAX_PAYLOAD_SIZE
;
1458 /* get the total packets count if necessary */
1459 if (c
->packets
== 0)
1461 c
->packets
= HEADER_TO_PACKETS((unsigned char *)c
->rudpbuf
);
1464 /* build the ordered packet array */
1465 for (int i
= c
->pktcurr
; i
< c
->recvpkt
; i
++)
1467 header
= (unsigned char *)c
->rudpbuf
+ i
* UDP_MAX_PAYLOAD_SIZE
;
1468 req_id
= (uint16_t)HEADER_TO_REQID(header
);
1469 assert(req_id
== c
->request_id
% (1 << 16));
1471 packets
= (uint16_t)HEADER_TO_PACKETS(header
);
1472 assert(c
->packets
== HEADER_TO_PACKETS(header
));
1474 seq_num
= (uint16_t)HEADER_TO_SEQNUM(header
);
1475 c
->udppkt
[seq_num
].header
= header
;
1476 c
->udppkt
[seq_num
].data
= (char *)header
+ UDP_HEADER_SIZE
;
1478 if (i
== c
->recvpkt
- 1)
1480 /* last received packet */
1481 if (c
->rudpbytes
% UDP_MAX_PAYLOAD_SIZE
== 0)
1483 c
->udppkt
[seq_num
].rbytes
= UDP_MAX_PAYLOAD_SIZE
- UDP_HEADER_SIZE
;
1488 c
->udppkt
[seq_num
].rbytes
= c
->rudpbytes
% UDP_MAX_PAYLOAD_SIZE
1494 c
->udppkt
[seq_num
].rbytes
= UDP_MAX_PAYLOAD_SIZE
- UDP_HEADER_SIZE
;
1499 for (int i
= c
->ordcurr
; i
< c
->recvpkt
; i
++)
1501 /* there is some data to copy */
1502 if ((c
->udppkt
[i
].data
!= NULL
)
1503 && (c
->udppkt
[i
].copybytes
< c
->udppkt
[i
].rbytes
))
1505 header
= c
->udppkt
[i
].header
;
1506 len
= c
->udppkt
[i
].rbytes
- c
->udppkt
[i
].copybytes
;
1507 if (len
> rbytes
- wbytes
)
1509 len
= rbytes
- wbytes
;
1512 assert(len
<= rbytes
- wbytes
);
1513 assert(i
== HEADER_TO_SEQNUM(header
));
1515 memcpy(buf
+ wbytes
, c
->udppkt
[i
].data
+ c
->udppkt
[i
].copybytes
,
1518 c
->udppkt
[i
].copybytes
+= len
;
1520 if ((c
->udppkt
[i
].copybytes
== c
->udppkt
[i
].rbytes
)
1521 && (c
->udppkt
[i
].rbytes
== UDP_MAX_PAYLOAD_SIZE
- UDP_HEADER_SIZE
))
1523 /* finish copying all the data of this packet, next */
1527 /* last received packet, and finish copying all the data */
1528 if ((c
->recvpkt
== c
->packets
) && (i
== c
->recvpkt
- 1)
1529 && (c
->udppkt
[i
].copybytes
== c
->udppkt
[i
].rbytes
))
1534 /* no space to copy data */
1535 if (wbytes
>= rbytes
)
1540 /* it doesn't finish reading all the data of the packet from network */
1541 if ((i
!= c
->recvpkt
- 1)
1542 && (c
->udppkt
[i
].rbytes
< UDP_MAX_PAYLOAD_SIZE
- UDP_HEADER_SIZE
))
1549 /* no data to copy */
1555 return wbytes
== 0 ? -1 : wbytes
;
1556 } /* ms_sort_udp_packet */
1560 * encapsulate upd read like tcp read
1562 * @param c, pointer of the concurrency
1563 * @param buf, read buffer
1564 * @param len, length to read
1566 * @return int, if success, return the read bytes, else return
1569 static int ms_udp_read(ms_conn_t
*c
, char *buf
, int len
)
1580 if (c
->rudpbytes
+ UDP_MAX_PAYLOAD_SIZE
> c
->rudpsize
)
1582 char *new_rbuf
= realloc(c
->rudpbuf
, (size_t)c
->rudpsize
* 2);
1585 fprintf(stderr
, "Couldn't realloc input buffer.\n");
1586 c
->rudpbytes
= 0; /* ignore what we read */
1589 c
->rudpbuf
= new_rbuf
;
1593 avail
= c
->rudpsize
- c
->rudpbytes
;
1594 /* UDP each time read a packet, 1400 bytes */
1595 res
= (int)read(c
->sfd
, c
->rudpbuf
+ c
->rudpbytes
, (size_t)avail
);
1599 atomic_add_size(&ms_stats
.bytes_read
, res
);
1614 /* "connection" closed */
1620 /* no data to read */
1625 /* copy data to read buffer */
1628 copybytes
= ms_sort_udp_packet(c
, buf
, len
);
1631 if (copybytes
== -1)
1633 atomic_add_size(&ms_stats
.pkt_disorder
, 1);
1641 * read from network as much as we can, handle buffer overflow and connection
1643 * before reading, move the remaining incomplete fragment of a command
1644 * (if any) to the beginning of the buffer.
1645 * return EXIT_SUCCESS if there's nothing to read on the first read.
1649 * read from network as much as we can, handle buffer overflow and connection
1650 * close. before reading, move the remaining incomplete fragment of a command
1651 * (if any) to the beginning of the buffer.
1653 * @param c, pointer of the concurrency
1656 * return EXIT_SUCCESS if there's nothing to read on the first read.
1657 * return EXIT_FAILURE if get data
1658 * return -1 if error happens
1660 static int ms_try_read_network(ms_conn_t
*c
)
1668 if ((c
->rcurr
!= c
->rbuf
)
1669 && (! c
->readval
|| (c
->rvbytes
> c
->rsize
- (c
->rcurr
- c
->rbuf
))
1670 || (c
->readval
&& (c
->rcurr
- c
->rbuf
> c
->rbytes
))))
1672 if (c
->rbytes
!= 0) /* otherwise there's nothing to copy */
1673 memmove(c
->rbuf
, c
->rcurr
, (size_t)c
->rbytes
);
1679 if (c
->rbytes
>= c
->rsize
)
1681 char *new_rbuf
= realloc(c
->rbuf
, (size_t)c
->rsize
* 2);
1684 fprintf(stderr
, "Couldn't realloc input buffer.\n");
1685 c
->rbytes
= 0; /* ignore what we read */
1688 c
->rcurr
= c
->rbuf
= new_rbuf
;
1692 avail
= c
->rsize
- c
->rbytes
- (c
->rcurr
- c
->rbuf
);
1700 res
= (int32_t)ms_udp_read(c
, c
->rcurr
+ c
->rbytes
, (int32_t)avail
);
1704 res
= (int)read(c
->sfd
, c
->rcurr
+ c
->rbytes
, (size_t)avail
);
1711 atomic_add_size(&ms_stats
.bytes_read
, res
);
1726 /* connection closed */
1727 ms_conn_set_state(c
, conn_closing
);
1732 if ((errno
== EAGAIN
) || (errno
== EWOULDBLOCK
))
1734 /* Should close on unhandled errors. */
1735 ms_conn_set_state(c
, conn_closing
);
1741 } /* ms_try_read_network */
1745 * after get the object from server, verify the value if
1748 * @param c, pointer of the concurrency
1749 * @param mlget_item, pointer of mulit-get task item structure
1750 * @param value, received value string
1751 * @param vlen, received value string length
1753 static void ms_verify_value(ms_conn_t
*c
,
1754 ms_mlget_task_item_t
*mlget_item
,
1758 if (c
->curr_task
.verify
)
1760 assert(c
->curr_task
.item
->value_offset
!= INVALID_OFFSET
);
1761 char *orignval
= &ms_setting
.char_block
[c
->curr_task
.item
->value_offset
];
1763 &ms_setting
.char_block
[c
->curr_task
.item
->key_suffix_offset
];
1765 /* verify expire time if necessary */
1766 if (c
->curr_task
.item
->exp_time
> 0)
1768 struct timeval curr_time
;
1769 gettimeofday(&curr_time
, NULL
);
1771 /* object expired but get it now */
1772 if (curr_time
.tv_sec
- c
->curr_task
.item
->client_time
1773 > c
->curr_task
.item
->exp_time
+ EXPIRE_TIME_ERROR
)
1775 atomic_add_size(&ms_stats
.exp_get
, 1);
1777 if (ms_setting
.verbose
)
1781 strftime(set_time
, 64, "%Y-%m-%d %H:%M:%S",
1782 localtime(&c
->curr_task
.item
->client_time
));
1783 strftime(cur_time
, 64, "%Y-%m-%d %H:%M:%S",
1784 localtime(&curr_time
.tv_sec
));
1786 "\n<%d expire time verification failed, "
1787 "object expired but get it now\n"
1789 "\tkey: %" PRIx64
" %.*s\n"
1790 "\tset time: %s current time: %s "
1791 "diff time: %d expire time: %d\n"
1792 "\texpected data: \n"
1793 "\treceived data len: %d\n"
1794 "\treceived data: %.*s\n",
1796 c
->curr_task
.item
->key_size
,
1797 c
->curr_task
.item
->key_prefix
,
1798 c
->curr_task
.item
->key_size
- (int)KEY_PREFIX_SIZE
,
1802 (int)(curr_time
.tv_sec
- c
->curr_task
.item
->client_time
),
1803 c
->curr_task
.item
->exp_time
,
1813 if ((c
->curr_task
.item
->value_size
!= vlen
)
1814 || (memcmp(orignval
, value
, (size_t)vlen
) != 0))
1816 atomic_add_size(&ms_stats
.vef_failed
, 1);
1818 if (ms_setting
.verbose
)
1821 "\n<%d data verification failed\n"
1823 "\tkey: %" PRIx64
" %.*s\n"
1824 "\texpected data len: %d\n"
1825 "\texpected data: %.*s\n"
1826 "\treceived data len: %d\n"
1827 "\treceived data: %.*s\n",
1829 c
->curr_task
.item
->key_size
,
1830 c
->curr_task
.item
->key_prefix
,
1831 c
->curr_task
.item
->key_size
- (int)KEY_PREFIX_SIZE
,
1833 c
->curr_task
.item
->value_size
,
1834 c
->curr_task
.item
->value_size
,
1844 c
->curr_task
.finish_verify
= true;
1846 if (mlget_item
!= NULL
)
1848 mlget_item
->finish_verify
= true;
1851 } /* ms_verify_value */
1855 * For ASCII protocol, after store the data into the local
1856 * buffer, run this function to handle the data.
1858 * @param c, pointer of the concurrency
1860 static void ms_ascii_complete_nread(ms_conn_t
*c
)
1863 assert(c
->rbytes
>= c
->rvbytes
);
1864 assert(c
->protocol
== ascii_prot
);
1868 c
->rcurr
[c
->rvbytes
- 1] == '\n' && c
->rcurr
[c
->rvbytes
- 2] == '\r');
1872 ms_mlget_task_item_t
*mlget_item
= NULL
;
1873 if (((ms_setting
.mult_key_num
> 1)
1874 && (c
->mlget_task
.mlget_num
>= ms_setting
.mult_key_num
))
1875 || ((c
->remain_exec_num
== 0) && (c
->mlget_task
.mlget_num
> 0)))
1877 c
->mlget_task
.value_index
++;
1878 mlget_item
= &c
->mlget_task
.mlget_item
[c
->mlget_task
.value_index
];
1880 if (mlget_item
->item
->key_prefix
== c
->currcmd
.key_prefix
)
1882 c
->curr_task
.item
= mlget_item
->item
;
1883 c
->curr_task
.verify
= mlget_item
->verify
;
1884 c
->curr_task
.finish_verify
= mlget_item
->finish_verify
;
1885 mlget_item
->get_miss
= false;
1889 /* Try to find the task item in multi-get task array */
1890 for (int i
= 0; i
< c
->mlget_task
.mlget_num
; i
++)
1892 mlget_item
= &c
->mlget_task
.mlget_item
[i
];
1893 if (mlget_item
->item
->key_prefix
== c
->currcmd
.key_prefix
)
1895 c
->curr_task
.item
= mlget_item
->item
;
1896 c
->curr_task
.verify
= mlget_item
->verify
;
1897 c
->curr_task
.finish_verify
= mlget_item
->finish_verify
;
1898 mlget_item
->get_miss
= false;
1906 ms_verify_value(c
, mlget_item
, c
->rcurr
, c
->rvbytes
- 2);
1908 c
->curr_task
.get_miss
= false;
1909 c
->rbytes
-= c
->rvbytes
;
1910 c
->rcurr
= c
->rcurr
+ c
->rvbytes
;
1911 assert(c
->rcurr
<= (c
->rbuf
+ c
->rsize
));
1914 } /* ms_ascii_complete_nread */
1918 * For binary protocol, after store the data into the local
1919 * buffer, run this function to handle the data.
1921 * @param c, pointer of the concurrency
1923 static void ms_bin_complete_nread(ms_conn_t
*c
)
1926 assert(c
->rbytes
>= c
->rvbytes
);
1927 assert(c
->protocol
== binary_prot
);
1929 int extlen
= c
->binary_header
.response
.extlen
;
1930 int keylen
= c
->binary_header
.response
.keylen
;
1931 uint8_t opcode
= c
->binary_header
.response
.opcode
;
1933 /* not get command or not include value, just return */
1934 if (((opcode
!= PROTOCOL_BINARY_CMD_GET
)
1935 && (opcode
!= PROTOCOL_BINARY_CMD_GETQ
))
1936 || (c
->rvbytes
<= extlen
+ keylen
))
1939 if (c
->binary_header
.response
.opcode
== PROTOCOL_BINARY_CMD_GET
)
1941 c
->currcmd
.retstat
= MCD_END
;
1942 c
->curr_task
.get_miss
= true;
1947 ms_reset_conn(c
, false);
1952 ms_mlget_task_item_t
*mlget_item
= NULL
;
1953 if (((ms_setting
.mult_key_num
> 1)
1954 && (c
->mlget_task
.mlget_num
>= ms_setting
.mult_key_num
))
1955 || ((c
->remain_exec_num
== 0) && (c
->mlget_task
.mlget_num
> 0)))
1957 c
->mlget_task
.value_index
++;
1958 mlget_item
= &c
->mlget_task
.mlget_item
[c
->mlget_task
.value_index
];
1960 c
->curr_task
.item
= mlget_item
->item
;
1961 c
->curr_task
.verify
= mlget_item
->verify
;
1962 c
->curr_task
.finish_verify
= mlget_item
->finish_verify
;
1963 mlget_item
->get_miss
= false;
1968 c
->rcurr
+ extlen
+ keylen
,
1969 c
->rvbytes
- extlen
- keylen
);
1971 c
->currcmd
.retstat
= MCD_END
;
1972 c
->curr_task
.get_miss
= false;
1973 c
->rbytes
-= c
->rvbytes
;
1974 c
->rcurr
= c
->rcurr
+ c
->rvbytes
;
1975 assert(c
->rcurr
<= (c
->rbuf
+ c
->rsize
));
1979 if (ms_setting
.mult_key_num
> 1)
1981 /* multi-get have check all the item */
1982 if (c
->mlget_task
.value_index
== c
->mlget_task
.mlget_num
- 1)
1984 ms_reset_conn(c
, false);
1990 ms_reset_conn(c
, false);
1992 } /* ms_bin_complete_nread */
1996 * we get here after reading the value of get commands.
1998 * @param c, pointer of the concurrency
2000 static void ms_complete_nread(ms_conn_t
*c
)
2003 assert(c
->rbytes
>= c
->rvbytes
);
2004 assert(c
->protocol
== ascii_prot
2005 || c
->protocol
== binary_prot
);
2007 if (c
->protocol
== binary_prot
)
2009 ms_bin_complete_nread(c
);
2013 ms_ascii_complete_nread(c
);
2015 } /* ms_complete_nread */
2019 * Adds a message header to a connection.
2021 * @param c, pointer of the concurrency
2023 * @return int, if success, return EXIT_SUCCESS, else return -1
2025 static int ms_add_msghdr(ms_conn_t
*c
)
2031 if (c
->msgsize
== c
->msgused
)
2034 realloc(c
->msglist
, (size_t)c
->msgsize
* 2 * sizeof(struct msghdr
));
2042 msg
= c
->msglist
+ c
->msgused
;
2045 * this wipes msg_iovlen, msg_control, msg_controllen, and
2046 * msg_flags, the last 3 of which aren't defined on solaris:
2048 memset(msg
, 0, sizeof(struct msghdr
));
2050 msg
->msg_iov
= &c
->iov
[c
->iovused
];
2052 if (c
->udp
&& (c
->srv_recv_addr_size
> 0))
2054 msg
->msg_name
= &c
->srv_recv_addr
;
2055 msg
->msg_namelen
= c
->srv_recv_addr_size
;
2063 /* Leave room for the UDP header, which we'll fill in later. */
2064 return ms_add_iov(c
, NULL
, UDP_HEADER_SIZE
);
2067 return EXIT_SUCCESS
;
2068 } /* ms_add_msghdr */
2072 * Ensures that there is room for another structure iovec in a connection's
2075 * @param c, pointer of the concurrency
2077 * @return int, if success, return EXIT_SUCCESS, else return -1
2079 static int ms_ensure_iov_space(ms_conn_t
*c
)
2083 if (c
->iovused
>= c
->iovsize
)
2086 struct iovec
*new_iov
= (struct iovec
*)realloc(c
->iov
,
2089 * sizeof(struct iovec
));
2096 /* Point all the msghdr structures at the new list. */
2097 for (i
= 0, iovnum
= 0; i
< c
->msgused
; i
++)
2099 c
->msglist
[i
].msg_iov
= &c
->iov
[iovnum
];
2100 iovnum
+= (int)c
->msglist
[i
].msg_iovlen
;
2104 return EXIT_SUCCESS
;
2105 } /* ms_ensure_iov_space */
2109 * Adds data to the list of pending data that will be written out to a
2112 * @param c, pointer of the concurrency
2113 * @param buf, the buffer includes data to send
2114 * @param len, the data length in the buffer
2116 * @return int, if success, return EXIT_SUCCESS, else return -1
2118 static int ms_add_iov(ms_conn_t
*c
, const void *buf
, int len
)
2128 m
= &c
->msglist
[c
->msgused
- 1];
2131 * Limit UDP packets, to UDP_MAX_PAYLOAD_SIZE bytes.
2133 limit_to_mtu
= c
->udp
;
2136 /* We may need to start a new msghdr if this one is full. */
2137 if ((m
->msg_iovlen
== IOV_MAX
)
2138 || (limit_to_mtu
&& (c
->msgbytes
>= UDP_MAX_SEND_PAYLOAD_SIZE
)))
2141 m
= &c
->msglist
[c
->msgused
- 1];
2145 if (ms_ensure_iov_space(c
) != 0)
2148 /* If the fragment is too big to fit in the datagram, split it up */
2149 if (limit_to_mtu
&& (len
+ c
->msgbytes
> UDP_MAX_SEND_PAYLOAD_SIZE
))
2151 leftover
= len
+ c
->msgbytes
- UDP_MAX_SEND_PAYLOAD_SIZE
;
2159 m
= &c
->msglist
[c
->msgused
- 1];
2160 m
->msg_iov
[m
->msg_iovlen
].iov_base
= (void *)buf
;
2161 m
->msg_iov
[m
->msg_iovlen
].iov_len
= (size_t)len
;
2167 buf
= ((char *)buf
) + len
;
2170 while (leftover
> 0);
2172 return EXIT_SUCCESS
;
2177 * Constructs a set of UDP headers and attaches them to the outgoing messages.
2179 * @param c, pointer of the concurrency
2181 * @return int, if success, return EXIT_SUCCESS, else return -1
2183 static int ms_build_udp_headers(ms_conn_t
*c
)
2190 c
->request_id
= ms_get_udp_request_id();
2192 if (c
->msgused
> c
->hdrsize
)
2196 new_hdrbuf
= realloc(c
->hdrbuf
,
2197 (size_t)c
->msgused
* 2 * UDP_HEADER_SIZE
);
2199 new_hdrbuf
= malloc((size_t)c
->msgused
* 2 * UDP_HEADER_SIZE
);
2203 c
->hdrbuf
= (unsigned char *)new_hdrbuf
;
2204 c
->hdrsize
= c
->msgused
* 2;
2207 /* If this is a multi-packet request, drop it. */
2208 if (c
->udp
&& (c
->msgused
> 1))
2210 fprintf(stderr
, "multi-packet request for UDP not supported.\n");
2215 for (i
= 0; i
< c
->msgused
; i
++)
2217 c
->msglist
[i
].msg_iov
[0].iov_base
= (void *)hdr
;
2218 c
->msglist
[i
].msg_iov
[0].iov_len
= UDP_HEADER_SIZE
;
2219 *hdr
++= (unsigned char)(c
->request_id
/ 256);
2220 *hdr
++= (unsigned char)(c
->request_id
% 256);
2221 *hdr
++= (unsigned char)(i
/ 256);
2222 *hdr
++= (unsigned char)(i
% 256);
2223 *hdr
++= (unsigned char)(c
->msgused
/ 256);
2224 *hdr
++= (unsigned char)(c
->msgused
% 256);
2225 *hdr
++= (unsigned char)1; /* support facebook memcached */
2226 *hdr
++= (unsigned char)0;
2228 ((unsigned char *)c
->msglist
[i
].msg_iov
[0].iov_base
2229 + UDP_HEADER_SIZE
));
2232 return EXIT_SUCCESS
;
2233 } /* ms_build_udp_headers */
2237 * Transmit the next chunk of data from our list of msgbuf structures.
2239 * @param c, pointer of the concurrency
2241 * @return TRANSMIT_COMPLETE All done writing.
2242 * TRANSMIT_INCOMPLETE More data remaining to write.
2243 * TRANSMIT_SOFT_ERROR Can't write any more right now.
2244 * TRANSMIT_HARD_ERROR Can't write (c->state is set to conn_closing)
2246 static int ms_transmit(ms_conn_t
*c
)
2250 if ((c
->msgcurr
< c
->msgused
)
2251 && (c
->msglist
[c
->msgcurr
].msg_iovlen
== 0))
2253 /* Finished writing the current msg; advance to the next. */
2257 if (c
->msgcurr
< c
->msgused
)
2260 struct msghdr
*m
= &c
->msglist
[c
->msgcurr
];
2262 res
= sendmsg(c
->sfd
, m
, 0);
2265 atomic_add_size(&ms_stats
.bytes_written
, res
);
2267 /* We've written some of the data. Remove the completed
2268 * iovec entries from the list of pending writes. */
2269 while (m
->msg_iovlen
> 0 && res
>= (ssize_t
)m
->msg_iov
->iov_len
)
2271 res
-= (ssize_t
)m
->msg_iov
->iov_len
;
2276 /* Might have written just part of the last iovec entry;
2277 * adjust it so the next write will do the rest. */
2280 m
->msg_iov
->iov_base
= (void *)((unsigned char *)m
->msg_iov
->iov_base
+ res
);
2281 m
->msg_iov
->iov_len
-= (size_t)res
;
2283 return TRANSMIT_INCOMPLETE
;
2285 if ((res
== -1) && ((errno
== EAGAIN
) || (errno
== EWOULDBLOCK
)))
2287 if (! ms_update_event(c
, EV_WRITE
| EV_PERSIST
))
2289 fprintf(stderr
, "Couldn't update event.\n");
2290 ms_conn_set_state(c
, conn_closing
);
2291 return TRANSMIT_HARD_ERROR
;
2293 return TRANSMIT_SOFT_ERROR
;
2296 /* if res==0 or res==-1 and error is not EAGAIN or EWOULDBLOCK,
2297 * we have a real error, on which we close the connection */
2298 fprintf(stderr
, "Failed to write, and not due to blocking.\n");
2300 ms_conn_set_state(c
, conn_closing
);
2301 return TRANSMIT_HARD_ERROR
;
2305 return TRANSMIT_COMPLETE
;
2311 * Shrinks a connection's buffers if they're too big. This prevents
2312 * periodic large "mget" response from server chewing lots of client
2315 * This should only be called in between requests since it can wipe output
2318 * @param c, pointer of the concurrency
2320 static void ms_conn_shrink(ms_conn_t
*c
)
2327 if ((c
->rsize
> READ_BUFFER_HIGHWAT
) && (c
->rbytes
< DATA_BUFFER_SIZE
))
2331 if (c
->rcurr
!= c
->rbuf
)
2332 memmove(c
->rbuf
, c
->rcurr
, (size_t)c
->rbytes
);
2334 newbuf
= (char *)realloc((void *)c
->rbuf
, DATA_BUFFER_SIZE
);
2339 c
->rsize
= DATA_BUFFER_SIZE
;
2344 if (c
->udp
&& (c
->rudpsize
> UDP_DATA_BUFFER_HIGHWAT
)
2345 && (c
->rudpbytes
+ UDP_MAX_PAYLOAD_SIZE
< UDP_DATA_BUFFER_SIZE
))
2347 char *new_rbuf
= (char *)realloc(c
->rudpbuf
, (size_t)c
->rudpsize
* 2);
2350 c
->rudpbuf
= new_rbuf
;
2351 c
->rudpsize
= UDP_DATA_BUFFER_SIZE
;
2353 /* TODO check error condition? */
2356 if (c
->msgsize
> MSG_LIST_HIGHWAT
)
2358 struct msghdr
*newbuf
= (struct msghdr
*)realloc(
2361 * sizeof(c
->msglist
[0]));
2365 c
->msgsize
= MSG_LIST_INITIAL
;
2367 /* TODO check error condition? */
2370 if (c
->iovsize
> IOV_LIST_HIGHWAT
)
2372 struct iovec
*newbuf
= (struct iovec
*)realloc((void *)c
->iov
,
2374 * sizeof(c
->iov
[0]));
2378 c
->iovsize
= IOV_LIST_INITIAL
;
2380 /* TODO check return value */
2382 } /* ms_conn_shrink */
2386 * Sets a connection's current state in the state machine. Any special
2387 * processing that needs to happen on certain state transitions can
2390 * @param c, pointer of the concurrency
2391 * @param state, connection state
2393 static void ms_conn_set_state(ms_conn_t
*c
, int state
)
2397 if (state
!= c
->state
)
2399 if (state
== conn_read
)
2405 } /* ms_conn_set_state */
2409 * update the event if socks change state. for example: when
2410 * change the listen scoket read event to sock write event, or
2411 * change socket handler, we could call this function.
2413 * @param c, pointer of the concurrency
2414 * @param new_flags, new event flags
2416 * @return bool, if success, return true, else return false
2418 static bool ms_update_event(ms_conn_t
*c
, const int new_flags
)
2422 struct event_base
*base
= c
->event
.ev_base
;
2423 if ((c
->ev_flags
== new_flags
) && (ms_setting
.rep_write_srv
== 0)
2424 && (! ms_setting
.facebook_test
|| (c
->total_sfds
== 1)))
2429 if (event_del(&c
->event
) == -1)
2431 /* try to delete the event again */
2432 if (event_del(&c
->event
) == -1)
2438 event_set(&c
->event
,
2443 event_base_set(base
, &c
->event
);
2444 c
->ev_flags
= (short)new_flags
;
2446 if (event_add(&c
->event
, NULL
) == -1)
2452 } /* ms_update_event */
2456 * If user want to get the expected throughput, we could limit
2457 * the performance of memslap. we could give up some work and
2458 * just wait a short time. The function is used to check this
2461 * @param c, pointer of the concurrency
2463 * @return bool, if success, return true, else return false
2465 static bool ms_need_yield(ms_conn_t
*c
)
2467 ms_thread_t
*ms_thread
= pthread_getspecific(ms_thread_key
);
2469 int64_t time_diff
= 0;
2470 struct timeval curr_time
;
2471 ms_task_t
*task
= &c
->curr_task
;
2473 if (ms_setting
.expected_tps
> 0)
2475 gettimeofday(&curr_time
, NULL
);
2476 time_diff
= ms_time_diff(&ms_thread
->startup_time
, &curr_time
);
2477 tps
= (int64_t)(((task
->get_opt
+ task
->set_opt
) / (uint64_t)time_diff
) * 1000000);
2479 /* current throughput is greater than expected throughput */
2480 if (tps
> ms_thread
->thread_ctx
->tps_perconn
)
2487 } /* ms_need_yield */
2491 * used to update the start time of each operation
2493 * @param c, pointer of the concurrency
2495 static void ms_update_start_time(ms_conn_t
*c
)
2497 ms_task_item_t
*item
= c
->curr_task
.item
;
2499 if ((ms_setting
.stat_freq
> 0) || c
->udp
2500 || ((c
->currcmd
.cmd
== CMD_SET
) && (item
->exp_time
> 0)))
2502 gettimeofday(&c
->start_time
, NULL
);
2503 if ((c
->currcmd
.cmd
== CMD_SET
) && (item
->exp_time
> 0))
2505 /* record the current time */
2506 item
->client_time
= c
->start_time
.tv_sec
;
2509 } /* ms_update_start_time */
2513 * run the state machine
2515 * @param c, pointer of the concurrency
2517 static void ms_drive_machine(ms_conn_t
*c
)
2530 if (c
->rbytes
>= c
->rvbytes
)
2532 ms_complete_nread(c
);
2538 if (ms_try_read_line(c
) != 0)
2544 if (ms_try_read_network(c
) != 0)
2549 /* doesn't read all the response data, wait event wake up */
2550 if (! c
->currcmd
.isfinish
)
2552 if (! ms_update_event(c
, EV_READ
| EV_PERSIST
))
2554 fprintf(stderr
, "Couldn't update event.\n");
2555 ms_conn_set_state(c
, conn_closing
);
2562 /* we have no command line and no data to read from network, next write */
2563 ms_conn_set_state(c
, conn_write
);
2564 memcpy(&c
->precmd
, &c
->currcmd
, sizeof(ms_cmdstat_t
)); /* replicate command state */
2569 if (! c
->ctnwrite
&& ms_need_yield(c
))
2573 if (! ms_update_event(c
, EV_WRITE
| EV_PERSIST
))
2575 fprintf(stderr
, "Couldn't update event.\n");
2576 ms_conn_set_state(c
, conn_closing
);
2583 if (! c
->ctnwrite
&& (ms_exec_task(c
) != 0))
2585 ms_conn_set_state(c
, conn_closing
);
2589 /* record the start time before starting to send data if necessary */
2590 if (! c
->ctnwrite
|| (c
->change_sfd
&& c
->ctnwrite
))
2594 c
->change_sfd
= false;
2596 ms_update_start_time(c
);
2599 /* change sfd if necessary */
2607 /* execute task until nothing need be written to network */
2608 if (! c
->ctnwrite
&& (c
->msgcurr
== c
->msgused
))
2610 if (! ms_update_event(c
, EV_WRITE
| EV_PERSIST
))
2612 fprintf(stderr
, "Couldn't update event.\n");
2613 ms_conn_set_state(c
, conn_closing
);
2620 switch (ms_transmit(c
))
2622 case TRANSMIT_COMPLETE
:
2623 /* we have no data to write to network, next wait repose */
2624 if (! ms_update_event(c
, EV_READ
| EV_PERSIST
))
2626 fprintf(stderr
, "Couldn't update event.\n");
2627 ms_conn_set_state(c
, conn_closing
);
2631 ms_conn_set_state(c
, conn_read
);
2636 case TRANSMIT_INCOMPLETE
:
2638 break; /* Continue in state machine. */
2640 case TRANSMIT_HARD_ERROR
:
2644 case TRANSMIT_SOFT_ERROR
:
2656 /* recovery mode, need reconnect if connection close */
2657 if (ms_setting
.reconnect
&& (! ms_global
.time_out
2658 || ((ms_setting
.run_time
== 0)
2659 && (c
->remain_exec_num
> 0))))
2661 if (ms_reconn(c
) != 0)
2668 ms_reset_conn(c
, false);
2670 if (c
->total_sfds
== 1)
2672 if (! ms_update_event(c
, EV_WRITE
| EV_PERSIST
))
2674 fprintf(stderr
, "Couldn't update event.\n");
2675 ms_conn_set_state(c
, conn_closing
);
2693 } /* ms_drive_machine */
2697 * the event handler of each thread
2699 * @param fd, the file descriptor of socket
2700 * @param which, event flag
2701 * @param arg, argument
2703 void ms_event_handler(const int fd
, const short which
, void *arg
)
2705 ms_conn_t
*c
= (ms_conn_t
*)arg
;
2715 "Catastrophic: event fd: %d doesn't match conn fd: %d\n",
2721 assert(fd
== c
->sfd
);
2723 ms_drive_machine(c
);
2725 /* wait for next event */
2726 } /* ms_event_handler */
2730 * get the next socket descriptor index to run for replication
2732 * @param c, pointer of the concurrency
2733 * @param cmd, command(get or set )
2735 * @return int, if success, return the index, else return EXIT_SUCCESS
2737 static uint32_t ms_get_rep_sock_index(ms_conn_t
*c
, int cmd
)
2739 uint32_t sock_index
= 0;
2742 if (c
->total_sfds
== 1)
2744 return EXIT_SUCCESS
;
2747 if (ms_setting
.rep_write_srv
== 0)
2756 for (i
= 0; i
< ms_setting
.rep_write_srv
; i
++)
2758 if (c
->tcpsfd
[i
] > 0)
2764 if (i
== ms_setting
.rep_write_srv
)
2766 /* random get one replication server to read */
2767 sock_index
= (uint32_t)random() % c
->total_sfds
;
2771 /* random get one replication writing server to write */
2772 sock_index
= (uint32_t)random() % ms_setting
.rep_write_srv
;
2775 else if (cmd
== CMD_GET
)
2777 /* random get one replication server to read */
2778 sock_index
= (uint32_t)random() % c
->total_sfds
;
2781 while (c
->tcpsfd
[sock_index
] == 0);
2784 } /* ms_get_rep_sock_index */
2788 * get the next socket descriptor index to run
2790 * @param c, pointer of the concurrency
2792 * @return int, return the index
2794 static uint32_t ms_get_next_sock_index(ms_conn_t
*c
)
2796 uint32_t sock_index
= 0;
2800 sock_index
= (++c
->cur_idx
== c
->total_sfds
) ? 0 : c
->cur_idx
;
2802 while (c
->tcpsfd
[sock_index
] == 0);
2805 } /* ms_get_next_sock_index */
2809 * update socket event of the connections
2811 * @param c, pointer of the concurrency
2813 * @return int, if success, return EXIT_SUCCESS, else return -1
2815 static int ms_update_conn_sock_event(ms_conn_t
*c
)
2819 switch (c
->currcmd
.cmd
)
2822 if (ms_setting
.facebook_test
&& c
->udp
)
2824 c
->sfd
= c
->tcpsfd
[0];
2826 c
->change_sfd
= true;
2831 if (ms_setting
.facebook_test
&& ! c
->udp
)
2835 c
->change_sfd
= true;
2843 if (! c
->udp
&& (c
->total_sfds
> 1))
2845 if (c
->cur_idx
!= c
->total_sfds
)
2847 if (ms_setting
.rep_write_srv
== 0)
2849 c
->cur_idx
= ms_get_next_sock_index(c
);
2853 c
->cur_idx
= ms_get_rep_sock_index(c
, c
->currcmd
.cmd
);
2858 /* must select the first sock of the connection at the beginning */
2862 c
->sfd
= c
->tcpsfd
[c
->cur_idx
];
2863 assert(c
->sfd
!= 0);
2864 c
->change_sfd
= true;
2869 if (! ms_update_event(c
, EV_WRITE
| EV_PERSIST
))
2871 fprintf(stderr
, "Couldn't update event.\n");
2872 ms_conn_set_state(c
, conn_closing
);
2877 return EXIT_SUCCESS
;
2878 } /* ms_update_conn_sock_event */
2882 * for ASCII protocol, this function build the set command
2883 * string and send the command.
2885 * @param c, pointer of the concurrency
2886 * @param item, pointer of task item which includes the object
2889 * @return int, if success, return EXIT_SUCCESS, else return -1
2891 static int ms_build_ascii_write_buf_set(ms_conn_t
*c
, ms_task_item_t
*item
)
2895 char *buffer
= c
->wbuf
;
2897 write_len
= snprintf(buffer
,
2904 if (write_len
> c
->wsize
|| write_len
< 0)
2906 /* ought to be always enough. just fail for simplicity */
2907 fprintf(stderr
, "output command line too long.\n");
2911 if (item
->value_offset
== INVALID_OFFSET
)
2913 value_offset
= item
->key_suffix_offset
;
2917 value_offset
= item
->value_offset
;
2920 if ((ms_add_iov(c
, "set ", 4) != 0)
2921 || (ms_add_iov(c
, (char *)&item
->key_prefix
,
2922 (int)KEY_PREFIX_SIZE
) != 0)
2923 || (ms_add_iov(c
, &ms_setting
.char_block
[item
->key_suffix_offset
],
2924 item
->key_size
- (int)KEY_PREFIX_SIZE
) != 0)
2925 || (ms_add_iov(c
, buffer
, write_len
) != 0)
2926 || (ms_add_iov(c
, &ms_setting
.char_block
[value_offset
],
2927 item
->value_size
) != 0)
2928 || (ms_add_iov(c
, "\r\n", 2) != 0)
2929 || (c
->udp
&& (ms_build_udp_headers(c
) != 0)))
2934 return EXIT_SUCCESS
;
2935 } /* ms_build_ascii_write_buf_set */
2939 * used to send set command to server
2941 * @param c, pointer of the concurrency
2942 * @param item, pointer of task item which includes the object
2945 * @return int, if success, return EXIT_SUCCESS, else return -1
2947 int ms_mcd_set(ms_conn_t
*c
, ms_task_item_t
*item
)
2951 c
->currcmd
.cmd
= CMD_SET
;
2952 c
->currcmd
.isfinish
= false;
2953 c
->currcmd
.retstat
= MCD_FAILURE
;
2955 if (ms_update_conn_sock_event(c
) != 0)
2963 if (ms_add_msghdr(c
) != 0)
2965 fprintf(stderr
, "Out of memory preparing request.");
2969 /* binary protocol */
2970 if (c
->protocol
== binary_prot
)
2972 if (ms_build_bin_write_buf_set(c
, item
) != 0)
2979 if (ms_build_ascii_write_buf_set(c
, item
) != 0)
2985 atomic_add_size(&ms_stats
.obj_bytes
,
2986 item
->key_size
+ item
->value_size
);
2987 atomic_add_size(&ms_stats
.cmd_set
, 1);
2989 return EXIT_SUCCESS
;
2994 * for ASCII protocol, this function build the get command
2995 * string and send the command.
2997 * @param c, pointer of the concurrency
2998 * @param item, pointer of task item which includes the object
3001 * @return int, if success, return EXIT_SUCCESS, else return -1
3003 static int ms_build_ascii_write_buf_get(ms_conn_t
*c
, ms_task_item_t
*item
)
3005 if ((ms_add_iov(c
, "get ", 4) != 0)
3006 || (ms_add_iov(c
, (char *)&item
->key_prefix
,
3007 (int)KEY_PREFIX_SIZE
) != 0)
3008 || (ms_add_iov(c
, &ms_setting
.char_block
[item
->key_suffix_offset
],
3009 item
->key_size
- (int)KEY_PREFIX_SIZE
) != 0)
3010 || (ms_add_iov(c
, "\r\n", 2) != 0)
3011 || (c
->udp
&& (ms_build_udp_headers(c
) != 0)))
3016 return EXIT_SUCCESS
;
3017 } /* ms_build_ascii_write_buf_get */
3021 * used to send the get command to server
3023 * @param c, pointer of the concurrency
3024 * @param item, pointer of task item which includes the object
3027 * @return int, if success, return EXIT_SUCCESS, else return -1
3029 int ms_mcd_get(ms_conn_t
*c
, ms_task_item_t
*item
)
3033 c
->currcmd
.cmd
= CMD_GET
;
3034 c
->currcmd
.isfinish
= false;
3035 c
->currcmd
.retstat
= MCD_FAILURE
;
3037 if (ms_update_conn_sock_event(c
) != 0)
3045 if (ms_add_msghdr(c
) != 0)
3047 fprintf(stderr
, "Out of memory preparing request.");
3051 /* binary protocol */
3052 if (c
->protocol
== binary_prot
)
3054 if (ms_build_bin_write_buf_get(c
, item
) != 0)
3061 if (ms_build_ascii_write_buf_get(c
, item
) != 0)
3067 atomic_add_size(&ms_stats
.cmd_get
, 1);
3069 return EXIT_SUCCESS
;
3074 * for ASCII protocol, this function build the multi-get command
3075 * string and send the command.
3077 * @param c, pointer of the concurrency
3079 * @return int, if success, return EXIT_SUCCESS, else return -1
3081 static int ms_build_ascii_write_buf_mlget(ms_conn_t
*c
)
3083 ms_task_item_t
*item
;
3085 if (ms_add_iov(c
, "get", 3) != 0)
3090 for (int i
= 0; i
< c
->mlget_task
.mlget_num
; i
++)
3092 item
= c
->mlget_task
.mlget_item
[i
].item
;
3093 assert(item
!= NULL
);
3094 if ((ms_add_iov(c
, " ", 1) != 0)
3095 || (ms_add_iov(c
, (char *)&item
->key_prefix
,
3096 (int)KEY_PREFIX_SIZE
) != 0)
3097 || (ms_add_iov(c
, &ms_setting
.char_block
[item
->key_suffix_offset
],
3098 item
->key_size
- (int)KEY_PREFIX_SIZE
) != 0))
3104 if ((ms_add_iov(c
, "\r\n", 2) != 0)
3105 || (c
->udp
&& (ms_build_udp_headers(c
) != 0)))
3110 return EXIT_SUCCESS
;
3111 } /* ms_build_ascii_write_buf_mlget */
3115 * used to send the multi-get command to server
3117 * @param c, pointer of the concurrency
3119 * @return int, if success, return EXIT_SUCCESS, else return -1
3121 int ms_mcd_mlget(ms_conn_t
*c
)
3123 ms_task_item_t
*item
;
3126 assert(c
->mlget_task
.mlget_num
>= 1);
3128 c
->currcmd
.cmd
= CMD_GET
;
3129 c
->currcmd
.isfinish
= false;
3130 c
->currcmd
.retstat
= MCD_FAILURE
;
3132 if (ms_update_conn_sock_event(c
) != 0)
3140 if (ms_add_msghdr(c
) != 0)
3142 fprintf(stderr
, "Out of memory preparing request.");
3146 /* binary protocol */
3147 if (c
->protocol
== binary_prot
)
3149 if (ms_build_bin_write_buf_mlget(c
) != 0)
3156 if (ms_build_ascii_write_buf_mlget(c
) != 0)
3162 /* decrease operation time of each item */
3163 for (int i
= 0; i
< c
->mlget_task
.mlget_num
; i
++)
3165 item
= c
->mlget_task
.mlget_item
[i
].item
;
3166 atomic_add_size(&ms_stats
.cmd_get
, 1);
3171 return EXIT_SUCCESS
;
3172 } /* ms_mcd_mlget */
3176 * binary protocol support
3180 * for binary protocol, parse the response of server
3182 * @param c, pointer of the concurrency
3184 * @return int, if success, return EXIT_SUCCESS, else return -1
3186 static int ms_bin_process_response(ms_conn_t
*c
)
3188 const char *errstr
= NULL
;
3192 uint32_t bodylen
= c
->binary_header
.response
.bodylen
;
3193 uint8_t opcode
= c
->binary_header
.response
.opcode
;
3194 uint16_t status
= c
->binary_header
.response
.status
;
3198 c
->rvbytes
= (int32_t)bodylen
;
3200 return EXIT_FAILURE
;
3206 case PROTOCOL_BINARY_RESPONSE_SUCCESS
:
3207 if (opcode
== PROTOCOL_BINARY_CMD_SET
)
3209 c
->currcmd
.retstat
= MCD_STORED
;
3211 else if (opcode
== PROTOCOL_BINARY_CMD_DELETE
)
3213 c
->currcmd
.retstat
= MCD_DELETED
;
3215 else if (opcode
== PROTOCOL_BINARY_CMD_GET
)
3217 c
->currcmd
.retstat
= MCD_END
;
3221 case PROTOCOL_BINARY_RESPONSE_ENOMEM
:
3222 errstr
= "Out of memory";
3223 c
->currcmd
.retstat
= MCD_SERVER_ERROR
;
3226 case PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND
:
3227 errstr
= "Unknown command";
3228 c
->currcmd
.retstat
= MCD_UNKNOWN_READ_FAILURE
;
3231 case PROTOCOL_BINARY_RESPONSE_KEY_ENOENT
:
3232 errstr
= "Not found";
3233 c
->currcmd
.retstat
= MCD_NOTFOUND
;
3236 case PROTOCOL_BINARY_RESPONSE_EINVAL
:
3237 errstr
= "Invalid arguments";
3238 c
->currcmd
.retstat
= MCD_PROTOCOL_ERROR
;
3241 case PROTOCOL_BINARY_RESPONSE_KEY_EEXISTS
:
3242 errstr
= "Data exists for key.";
3245 case PROTOCOL_BINARY_RESPONSE_E2BIG
:
3246 errstr
= "Too large.";
3247 c
->currcmd
.retstat
= MCD_SERVER_ERROR
;
3250 case PROTOCOL_BINARY_RESPONSE_NOT_STORED
:
3251 errstr
= "Not stored.";
3252 c
->currcmd
.retstat
= MCD_NOTSTORED
;
3256 errstr
= "Unknown error";
3257 c
->currcmd
.retstat
= MCD_UNKNOWN_READ_FAILURE
;
3263 fprintf(stderr
, "%s\n", errstr
);
3267 return EXIT_SUCCESS
;
3268 } /* ms_bin_process_response */
3271 /* build binary header and add the header to the buffer to send */
3274 * build binary header and add the header to the buffer to send
3276 * @param c, pointer of the concurrency
3277 * @param opcode, operation code
3278 * @param hdr_len, length of header
3279 * @param key_len, length of key
3280 * @param body_len. length of body
3282 static void ms_add_bin_header(ms_conn_t
*c
,
3288 protocol_binary_request_header
*header
;
3292 header
= (protocol_binary_request_header
*)c
->wcurr
;
3294 header
->request
.magic
= (uint8_t)PROTOCOL_BINARY_REQ
;
3295 header
->request
.opcode
= (uint8_t)opcode
;
3296 header
->request
.keylen
= htons(key_len
);
3298 header
->request
.extlen
= (uint8_t)hdr_len
;
3299 header
->request
.datatype
= (uint8_t)PROTOCOL_BINARY_RAW_BYTES
;
3300 header
->request
.vbucket
= 0;
3302 header
->request
.bodylen
= htonl(body_len
);
3303 header
->request
.opaque
= 0;
3304 header
->request
.cas
= 0;
3306 ms_add_iov(c
, c
->wcurr
, sizeof(header
->request
));
3307 } /* ms_add_bin_header */
3311 * add the key to the socket write buffer array
3313 * @param c, pointer of the concurrency
3314 * @param item, pointer of task item which includes the object
3317 static void ms_add_key_to_iov(ms_conn_t
*c
, ms_task_item_t
*item
)
3319 ms_add_iov(c
, (char *)&item
->key_prefix
, (int)KEY_PREFIX_SIZE
);
3320 ms_add_iov(c
, &ms_setting
.char_block
[item
->key_suffix_offset
],
3321 item
->key_size
- (int)KEY_PREFIX_SIZE
);
3326 * for binary protocol, this function build the set command
3327 * and add the command to send buffer array.
3329 * @param c, pointer of the concurrency
3330 * @param item, pointer of task item which includes the object
3333 * @return int, if success, return EXIT_SUCCESS, else return -1
3335 static int ms_build_bin_write_buf_set(ms_conn_t
*c
, ms_task_item_t
*item
)
3337 assert(c
->wbuf
== c
->wcurr
);
3340 protocol_binary_request_set
*rep
= (protocol_binary_request_set
*)c
->wcurr
;
3341 uint16_t keylen
= (uint16_t)item
->key_size
;
3342 uint32_t bodylen
= (uint32_t)sizeof(rep
->message
.body
)
3343 + (uint32_t)keylen
+ (uint32_t)item
->value_size
;
3345 ms_add_bin_header(c
,
3346 PROTOCOL_BINARY_CMD_SET
,
3347 sizeof(rep
->message
.body
),
3350 rep
->message
.body
.flags
= 0;
3351 rep
->message
.body
.expiration
= htonl((uint32_t)item
->exp_time
);
3352 ms_add_iov(c
, &rep
->message
.body
, sizeof(rep
->message
.body
));
3353 ms_add_key_to_iov(c
, item
);
3355 if (item
->value_offset
== INVALID_OFFSET
)
3357 value_offset
= item
->key_suffix_offset
;
3361 value_offset
= item
->value_offset
;
3363 ms_add_iov(c
, &ms_setting
.char_block
[value_offset
], item
->value_size
);
3365 return EXIT_SUCCESS
;
3366 } /* ms_build_bin_write_buf_set */
3370 * for binary protocol, this function build the get command and
3371 * add the command to send buffer array.
3373 * @param c, pointer of the concurrency
3374 * @param item, pointer of task item which includes the object
3377 * @return int, if success, return EXIT_SUCCESS, else return -1
3379 static int ms_build_bin_write_buf_get(ms_conn_t
*c
, ms_task_item_t
*item
)
3381 assert(c
->wbuf
== c
->wcurr
);
3383 ms_add_bin_header(c
, PROTOCOL_BINARY_CMD_GET
, 0, (uint16_t)item
->key_size
,
3384 (uint32_t)item
->key_size
);
3385 ms_add_key_to_iov(c
, item
);
3387 return EXIT_SUCCESS
;
3388 } /* ms_build_bin_write_buf_get */
3392 * for binary protocol, this function build the multi-get
3393 * command and add the command to send buffer array.
3395 * @param c, pointer of the concurrency
3396 * @param item, pointer of task item which includes the object
3399 * @return int, if success, return EXIT_SUCCESS, else return -1
3401 static int ms_build_bin_write_buf_mlget(ms_conn_t
*c
)
3403 ms_task_item_t
*item
;
3405 assert(c
->wbuf
== c
->wcurr
);
3407 for (int i
= 0; i
< c
->mlget_task
.mlget_num
; i
++)
3409 item
= c
->mlget_task
.mlget_item
[i
].item
;
3410 assert(item
!= NULL
);
3412 ms_add_bin_header(c
,
3413 PROTOCOL_BINARY_CMD_GET
,
3415 (uint16_t)item
->key_size
,
3416 (uint32_t)item
->key_size
);
3417 ms_add_key_to_iov(c
, item
);
3418 c
->wcurr
+= sizeof(protocol_binary_request_get
);
3423 return EXIT_SUCCESS
;
3424 } /* ms_build_bin_write_buf_mlget */