3 * Author: Mingqiang Zhuang
5 * Created on February 10, 2009
7 * (c) Copyright 2009, Schooner Information Technology, Inc.
8 * http://www.schoonerinfotech.com/
11 #include "ms_thread.h"
12 #include "ms_setting.h"
15 __thread ms_thread_t ms_thread
; /* each thread with a private ms_thread structure */
17 /* array of thread context structure, each thread has a thread context structure */
18 static ms_thread_ctx_t
*ms_thread_ctx
;
21 static void ms_set_current_time(void);
22 static void ms_check_sock_timeout(void);
23 static void ms_clock_handler(const int fd
, const short which
, void *arg
);
24 static int ms_set_thread_cpu_affinity(int cpu
);
25 static int ms_setup_thread(ms_thread_ctx_t
*thread_ctx
);
26 static void *ms_worker_libevent(void *arg
);
27 static void ms_create_worker(void *(*func
)(void *), void *arg
);
31 * time-sensitive callers can call it by hand with this,
32 * outside the normal ever-1-second timer
34 static void ms_set_current_time()
38 gettimeofday(&timer
, NULL
);
39 ms_thread
.curr_time
= (rel_time_t
)timer
.tv_sec
;
40 } /* ms_set_current_time */
44 * used to check whether UDP of command are waiting timeout
45 * by the ever-1-second timer
47 static void ms_check_sock_timeout(void)
52 for (int i
= 0; i
< ms_thread
.thread_ctx
->nconns
; i
++)
54 c
= &ms_thread
.conn
[i
];
58 time_diff
= (int)(ms_thread
.curr_time
- c
->start_time
.tv_sec
);
61 if (time_diff
> SOCK_WAIT_TIMEOUT
)
63 /* calculate dropped packets count */
66 __sync_fetch_and_add(&ms_stats
.pkt_drop
, c
->packets
- c
->recvpkt
);
69 __sync_fetch_and_add(&ms_stats
.udp_timeout
, 1);
70 ms_reset_conn(c
, true);
74 } /* ms_check_sock_timeout */
77 /* if disconnect, the ever-1-second timer will call this function to reconnect */
78 static void ms_reconn_thread_socks(void)
80 for (int i
= 0; i
< ms_thread
.thread_ctx
->nconns
; i
++)
82 ms_reconn_socks(&ms_thread
.conn
[i
]);
84 } /* ms_reconn_thread_socks */
88 * the handler of the ever-1-second timer
90 * @param fd, the descriptors of the socket
91 * @param which, event flags
92 * @param arg, argument
94 static void ms_clock_handler(const int fd
, const short which
, void *arg
)
98 .tv_sec
= 1, .tv_usec
= 0
102 UNUSED_ARGUMENT(which
);
103 UNUSED_ARGUMENT(arg
);
105 ms_set_current_time();
107 if (ms_thread
.initialized
)
109 /* only delete the event if it's actually there. */
110 evtimer_del(&ms_thread
.clock_event
);
111 ms_check_sock_timeout();
115 ms_thread
.initialized
= true;
118 ms_reconn_thread_socks();
120 evtimer_set(&ms_thread
.clock_event
, ms_clock_handler
, 0);
121 event_base_set(ms_thread
.base
, &ms_thread
.clock_event
);
122 evtimer_add(&ms_thread
.clock_event
, &t
);
123 } /* ms_clock_handler */
127 * used to bind thread to CPU if the system supports
129 * @param cpu, cpu index
131 * @return if success, return 0, else return -1
133 static int ms_set_thread_cpu_affinity(int cpu
)
137 #ifdef HAVE_CPU_SET_T
140 CPU_SET(cpu
, &cpu_set
);
142 if (sched_setaffinity(0, sizeof(cpu_set_t
), &cpu_set
) == -1)
144 fprintf(stderr
, "WARNING: Could not set CPU Affinity, continuing...\n");
148 UNUSED_ARGUMENT(cpu
);
152 } /* ms_set_thread_cpu_affinity */
156 * Set up a thread's information.
158 * @param thread_ctx, pointer of the thread context structure
160 * @return if success, return 0, else return -1
162 static int ms_setup_thread(ms_thread_ctx_t
*thread_ctx
)
164 ms_thread
.thread_ctx
= thread_ctx
;
165 ms_thread
.nactive_conn
= thread_ctx
->nconns
;
166 ms_thread
.initialized
= false;
169 gettimeofday(&ms_thread
.startup_time
, NULL
);
171 ms_thread
.base
= event_init();
172 if (ms_thread
.base
== NULL
)
174 if (__sync_fetch_and_add(&cnt
, 1) == 0)
176 fprintf(stderr
, "Can't allocate event base.\n");
183 (ms_conn_t
*)malloc((size_t)thread_ctx
->nconns
* sizeof(ms_conn_t
));
184 if (ms_thread
.conn
== NULL
)
186 if (__sync_fetch_and_add(&cnt
, 1) == 0)
190 "Can't allocate concurrency structure for thread descriptors.");
195 memset(ms_thread
.conn
, 0, (size_t)thread_ctx
->nconns
* sizeof(ms_conn_t
));
197 for (int i
= 0; i
< thread_ctx
->nconns
; i
++)
199 ms_thread
.conn
[i
].conn_idx
= i
;
200 if (ms_setup_conn(&ms_thread
.conn
[i
]) != 0)
202 /* only output this error once */
203 if (__sync_fetch_and_add(&cnt
, 1) == 0)
205 fprintf(stderr
, "Initializing connection failed.\n");
213 } /* ms_setup_thread */
217 * Worker thread: main event loop
219 * @param arg, the pointer of argument
223 static void *ms_worker_libevent(void *arg
)
225 ms_thread_ctx_t
*thread_ctx
= (ms_thread_ctx_t
*)arg
;
228 * If system has more than one cpu and supports set cpu
229 * affinity, try to bind each thread to a cpu core;
231 if (ms_setting
.ncpu
> 1)
233 ms_set_thread_cpu_affinity(thread_ctx
->thd_idx
% ms_setting
.ncpu
);
236 if (ms_setup_thread(thread_ctx
) != 0)
241 /* each thread with a timer */
242 ms_clock_handler(0, 0, 0);
244 event_base_loop(ms_thread
.base
, 0);
247 } /* ms_worker_libevent */
251 * Creates a worker thread.
253 * @param func, the callback function
254 * @param arg, the argument to pass to the callback function
256 static void ms_create_worker(void *(*func
)(void *), void *arg
)
262 pthread_attr_init(&attr
);
264 if ((ret
= pthread_create(&thread
, &attr
, func
, arg
)) != 0)
266 fprintf(stderr
, "Can't create thread: %s.\n", strerror(ret
));
269 } /* ms_create_worker */
272 /* initialize threads */
273 void ms_thread_init()
276 (ms_thread_ctx_t
*)malloc(
277 sizeof(ms_thread_ctx_t
) * (size_t)ms_setting
.nthreads
);
278 if (ms_thread_ctx
== NULL
)
280 fprintf(stderr
, "Can't allocate thread descriptors.");
284 for (int i
= 0; i
< ms_setting
.nthreads
; i
++)
286 ms_thread_ctx
[i
].thd_idx
= i
;
287 ms_thread_ctx
[i
].nconns
= ms_setting
.nconns
/ ms_setting
.nthreads
;
290 * If only one server, all the connections in all threads
291 * connects the same server. For support multi-servers, simple
292 * distribute thread to server.
294 ms_thread_ctx
[i
].srv_idx
= i
% ms_setting
.srv_cnt
;
295 ms_thread_ctx
[i
].tps_perconn
= ms_setting
.expected_tps
297 ms_thread_ctx
[i
].exec_num_perconn
= ms_setting
.exec_num
301 /* Create threads after we've done all the epoll setup. */
302 for (int i
= 0; i
< ms_setting
.nthreads
; i
++)
304 ms_create_worker(ms_worker_libevent
, (void *)&ms_thread_ctx
[i
]);
306 } /* ms_thread_init */
309 /* cleanup some resource of threads when all the threads exit */
310 void ms_thread_cleanup()
312 if (ms_thread_ctx
!= NULL
)
316 } /* ms_thread_cleanup */