3 * Author: Mingqiang Zhuang
5 * Created on February 10, 2009
7 * (c) Copyright 2009, Schooner Information Technology, Inc.
8 * http://www.schoonerinfotech.com/
14 #include "ms_thread.h"
15 #include "ms_setting.h"
16 #include "ms_atomic.h"
19 __thread ms_thread_t ms_thread
; /* each thread with a private ms_thread structure */
21 /* array of thread context structure, each thread has a thread context structure */
22 static ms_thread_ctx_t
*ms_thread_ctx
;
25 static void ms_set_current_time(void);
26 static void ms_check_sock_timeout(void);
27 static void ms_clock_handler(const int fd
, const short which
, void *arg
);
28 static int ms_set_thread_cpu_affinity(int cpu
);
29 static int ms_setup_thread(ms_thread_ctx_t
*thread_ctx
);
30 static void *ms_worker_libevent(void *arg
);
31 static void ms_create_worker(void *(*func
)(void *), void *arg
);
35 * time-sensitive callers can call it by hand with this,
36 * outside the normal ever-1-second timer
38 static void ms_set_current_time()
42 gettimeofday(&timer
, NULL
);
43 ms_thread
.curr_time
= (rel_time_t
)timer
.tv_sec
;
44 } /* ms_set_current_time */
48 * used to check whether UDP of command are waiting timeout
49 * by the ever-1-second timer
51 static void ms_check_sock_timeout(void)
56 for (int i
= 0; i
< ms_thread
.thread_ctx
->nconns
; i
++)
58 c
= &ms_thread
.conn
[i
];
62 time_diff
= (int)(ms_thread
.curr_time
- c
->start_time
.tv_sec
);
65 if (time_diff
> SOCK_WAIT_TIMEOUT
)
67 /* calculate dropped packets count */
70 __sync_fetch_and_add(&ms_stats
.pkt_drop
, c
->packets
- c
->recvpkt
);
73 __sync_fetch_and_add(&ms_stats
.udp_timeout
, 1);
74 ms_reset_conn(c
, true);
78 } /* ms_check_sock_timeout */
81 /* if disconnect, the ever-1-second timer will call this function to reconnect */
82 static void ms_reconn_thread_socks(void)
84 for (int i
= 0; i
< ms_thread
.thread_ctx
->nconns
; i
++)
86 ms_reconn_socks(&ms_thread
.conn
[i
]);
88 } /* ms_reconn_thread_socks */
92 * the handler of the ever-1-second timer
94 * @param fd, the descriptors of the socket
95 * @param which, event flags
96 * @param arg, argument
98 static void ms_clock_handler(const int fd
, const short which
, void *arg
)
102 .tv_sec
= 1, .tv_usec
= 0
106 UNUSED_ARGUMENT(which
);
107 UNUSED_ARGUMENT(arg
);
109 ms_set_current_time();
111 if (ms_thread
.initialized
)
113 /* only delete the event if it's actually there. */
114 evtimer_del(&ms_thread
.clock_event
);
115 ms_check_sock_timeout();
119 ms_thread
.initialized
= true;
122 ms_reconn_thread_socks();
124 evtimer_set(&ms_thread
.clock_event
, ms_clock_handler
, 0);
125 event_base_set(ms_thread
.base
, &ms_thread
.clock_event
);
126 evtimer_add(&ms_thread
.clock_event
, &t
);
127 } /* ms_clock_handler */
131 * used to bind thread to CPU if the system supports
133 * @param cpu, cpu index
135 * @return if success, return 0, else return -1
137 static int ms_set_thread_cpu_affinity(int cpu
)
141 #ifdef HAVE_CPU_SET_T
144 CPU_SET(cpu
, &cpu_set
);
146 if (sched_setaffinity(0, sizeof(cpu_set_t
), &cpu_set
) == -1)
148 fprintf(stderr
, "WARNING: Could not set CPU Affinity, continuing...\n");
152 UNUSED_ARGUMENT(cpu
);
156 } /* ms_set_thread_cpu_affinity */
160 * Set up a thread's information.
162 * @param thread_ctx, pointer of the thread context structure
164 * @return if success, return 0, else return -1
166 static int ms_setup_thread(ms_thread_ctx_t
*thread_ctx
)
168 ms_thread
.thread_ctx
= thread_ctx
;
169 ms_thread
.nactive_conn
= thread_ctx
->nconns
;
170 ms_thread
.initialized
= false;
173 gettimeofday(&ms_thread
.startup_time
, NULL
);
175 ms_thread
.base
= event_init();
176 if (ms_thread
.base
== NULL
)
178 if (__sync_fetch_and_add(&cnt
, 1) == 0)
180 fprintf(stderr
, "Can't allocate event base.\n");
187 (ms_conn_t
*)malloc((size_t)thread_ctx
->nconns
* sizeof(ms_conn_t
));
188 if (ms_thread
.conn
== NULL
)
190 if (__sync_fetch_and_add(&cnt
, 1) == 0)
194 "Can't allocate concurrency structure for thread descriptors.");
199 memset(ms_thread
.conn
, 0, (size_t)thread_ctx
->nconns
* sizeof(ms_conn_t
));
201 for (int i
= 0; i
< thread_ctx
->nconns
; i
++)
203 ms_thread
.conn
[i
].conn_idx
= i
;
204 if (ms_setup_conn(&ms_thread
.conn
[i
]) != 0)
206 /* only output this error once */
207 if (__sync_fetch_and_add(&cnt
, 1) == 0)
209 fprintf(stderr
, "Initializing connection failed.\n");
217 } /* ms_setup_thread */
221 * Worker thread: main event loop
223 * @param arg, the pointer of argument
227 static void *ms_worker_libevent(void *arg
)
229 ms_thread_ctx_t
*thread_ctx
= (ms_thread_ctx_t
*)arg
;
232 * If system has more than one cpu and supports set cpu
233 * affinity, try to bind each thread to a cpu core;
235 if (ms_setting
.ncpu
> 1)
237 ms_set_thread_cpu_affinity(thread_ctx
->thd_idx
% ms_setting
.ncpu
);
240 if (ms_setup_thread(thread_ctx
) != 0)
245 /* each thread with a timer */
246 ms_clock_handler(0, 0, 0);
248 event_base_loop(ms_thread
.base
, 0);
251 } /* ms_worker_libevent */
255 * Creates a worker thread.
257 * @param func, the callback function
258 * @param arg, the argument to pass to the callback function
260 static void ms_create_worker(void *(*func
)(void *), void *arg
)
266 pthread_attr_init(&attr
);
268 if ((ret
= pthread_create(&thread
, &attr
, func
, arg
)) != 0)
270 fprintf(stderr
, "Can't create thread: %s.\n", strerror(ret
));
273 } /* ms_create_worker */
276 /* initialize threads */
277 void ms_thread_init()
280 (ms_thread_ctx_t
*)malloc(
281 sizeof(ms_thread_ctx_t
) * (size_t)ms_setting
.nthreads
);
282 if (ms_thread_ctx
== NULL
)
284 fprintf(stderr
, "Can't allocate thread descriptors.");
288 for (int i
= 0; i
< ms_setting
.nthreads
; i
++)
290 ms_thread_ctx
[i
].thd_idx
= i
;
291 ms_thread_ctx
[i
].nconns
= ms_setting
.nconns
/ ms_setting
.nthreads
;
294 * If only one server, all the connections in all threads
295 * connects the same server. For support multi-servers, simple
296 * distribute thread to server.
298 ms_thread_ctx
[i
].srv_idx
= i
% ms_setting
.srv_cnt
;
299 ms_thread_ctx
[i
].tps_perconn
= ms_setting
.expected_tps
301 ms_thread_ctx
[i
].exec_num_perconn
= ms_setting
.exec_num
305 /* Create threads after we've done all the epoll setup. */
306 for (int i
= 0; i
< ms_setting
.nthreads
; i
++)
308 ms_create_worker(ms_worker_libevent
, (void *)&ms_thread_ctx
[i
]);
310 } /* ms_thread_init */
313 /* cleanup some resource of threads when all the threads exit */
314 void ms_thread_cleanup()
316 if (ms_thread_ctx
!= NULL
)
320 } /* ms_thread_cleanup */