-
Notifications
You must be signed in to change notification settings - Fork 1.2k
/
one-acceptor-multi-workers.c
100 lines (84 loc) · 2.42 KB
/
one-acceptor-multi-workers.c
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
/*
*
* @build make examples
* @server bin/one-acceptor-multi-workers 1234
* @client bin/nc 127.0.0.1 1234
* nc 127.0.0.1 1234
* telnet 127.0.0.1 1234
*/
#include "hloop.h"
#include "hsocket.h"
#include "hthread.h"
static const char* host = "0.0.0.0";
static int port = 1234;
static int thread_num = 4;
static hloop_t* accept_loop = NULL;
static hloop_t** worker_loops = NULL;
static hloop_t* get_next_loop() {
static int s_cur_index = 0;
if (++s_cur_index >= thread_num) {
s_cur_index = 0;
}
return worker_loops[s_cur_index % thread_num];
}
static void on_close(hio_t* io) {
printf("on_close fd=%d error=%d\n", hio_fd(io), hio_error(io));
}
static void on_recv(hio_t* io, void* buf, int readbytes) {
// echo
hio_write(io, buf, readbytes);
}
static void new_conn_event(hevent_t* ev) {
hloop_t* loop = ev->loop;
hio_t* io = (hio_t*)hevent_userdata(ev);
hio_attach(loop, io);
char localaddrstr[SOCKADDR_STRLEN] = {0};
char peeraddrstr[SOCKADDR_STRLEN] = {0};
printf("tid=%ld connfd=%d [%s] <= [%s]\n",
(long)hv_gettid(),
(int)hio_fd(io),
SOCKADDR_STR(hio_localaddr(io), localaddrstr),
SOCKADDR_STR(hio_peeraddr(io), peeraddrstr));
hio_setcb_close(io, on_close);
hio_setcb_read(io, on_recv);
hio_read(io);
}
static void on_accept(hio_t* io) {
hio_detach(io);
hloop_t* worker_loop = get_next_loop();
hevent_t ev;
memset(&ev, 0, sizeof(ev));
ev.loop = worker_loop;
ev.cb = new_conn_event;
ev.userdata = io;
hloop_post_event(worker_loop, &ev);
}
static HTHREAD_ROUTINE(worker_thread) {
hloop_t* loop = (hloop_t*)userdata;
hloop_run(loop);
return 0;
}
static HTHREAD_ROUTINE(accept_thread) {
hloop_t* loop = (hloop_t*)userdata;
hio_t* listenio = hloop_create_tcp_server(loop, host, port, on_accept);
if (listenio == NULL) {
exit(1);
}
hloop_run(loop);
return 0;
}
int main(int argc, char** argv) {
if (argc < 2) {
printf("Usage: cmd port\n");
return -10;
}
port = atoi(argv[1]);
worker_loops = (hloop_t**)malloc(sizeof(hloop_t*) * thread_num);
for (int i = 0; i < thread_num; ++i) {
worker_loops[i] = hloop_new(HLOOP_FLAG_AUTO_FREE);
hthread_create(worker_thread, worker_loops[i]);
}
accept_loop = hloop_new(HLOOP_FLAG_AUTO_FREE);
accept_thread(accept_loop);
return 0;
}