-
Notifications
You must be signed in to change notification settings - Fork 1.4k
Expand file tree
/
Copy pathhevent.h
More file actions
331 lines (302 loc) · 9.65 KB
/
Copy pathhevent.h
File metadata and controls
331 lines (302 loc) · 9.65 KB
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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
#ifndef HV_EVENT_H_
#define HV_EVENT_H_
#include "hloop.h"
#include "iowatcher.h"
#include "rudp.h"
#include "hbuf.h"
#include "hmutex.h"
#include "array.h"
#include "list.h"
#include "heap.h"
#include "queue.h"
#define HLOOP_READ_BUFSIZE 8192 // 8K
#define READ_BUFSIZE_HIGH_WATER 65536 // 64K
#define WRITE_BUFSIZE_HIGH_WATER (1U << 23) // 8M
#define MAX_READ_BUFSIZE (1U << 24) // 16M
#define MAX_WRITE_BUFSIZE (1U << 24) // 16M
// hio_read_flags
#define HIO_READ_ONCE 0x1
#define HIO_READ_UNTIL_LENGTH 0x2
#define HIO_READ_UNTIL_DELIM 0x4
// Internal NIO transport lifecycle. hio_handle_events is the single readiness
// dispatcher; phase selects its current transport or handshake action.
typedef enum {
HIO_PHASE_NONE = 0,
HIO_PHASE_READY,
HIO_PHASE_ACCEPTING,
HIO_PHASE_CONNECTING,
HIO_PHASE_PROXY_HANDSHAKING,
HIO_PHASE_PROXY_ESTABLISHED,
HIO_PHASE_TLS_SERVER_HANDSHAKING,
HIO_PHASE_TLS_CLIENT_HANDSHAKING,
HIO_PHASE_TLS_ESTABLISHED,
HIO_PHASE_ESTABLISHED,
HIO_PHASE_CLOSED,
} hio_phase_e;
ARRAY_DECL(hio_t*, io_array);
ARRAY_DECL(hsignal_t*, signal_array);
QUEUE_DECL(hevent_t, event_queue);
struct hloop_s {
uint32_t flags;
hloop_status_e status;
uint64_t start_ms; // ms
uint64_t start_hrtime; // us
uint64_t end_hrtime;
uint64_t cur_hrtime;
uint64_t loop_cnt;
long pid;
long tid;
void* userdata;
//private:
// events
uint32_t intern_nevents;
uint32_t nactives;
uint32_t npendings;
// pendings: with priority as array.index
hevent_t* pendings[HEVENT_PRIORITY_SIZE];
// signals
struct signal_array signals;
uint32_t nsignals;
// idles
struct list_head idles;
uint32_t nidles;
// timers
struct heap timers; // monotonic time
struct heap realtimers; // realtime
uint32_t ntimers;
// ios: with fd as array.index
struct io_array ios;
uint32_t nios;
// one loop per thread, so one readbuf per loop is OK.
hbuf_t readbuf;
void* iowatcher;
// custom_events
int eventfds[2];
event_queue custom_events;
hmutex_t custom_events_mutex;
// async dns resolver (event/hdns.c), created lazily, freed in hloop_cleanup
void* dns_resolver;
// per-loop lua_State (lua/), stored as opaque void* so the C core stays
// lua-free. Set via hloop_set_lua_state with a destructor; freed in hloop_cleanup.
void* lua_state;
void (*lua_state_dtor)(void* lua_state);
// per-loop JS runtime (js/), stored as opaque void* so the C core stays
// quickjs-free. Set via hloop_set_js_runtime with a destructor; freed in hloop_cleanup.
void* js_runtime;
void (*js_runtime_dtor)(void* js_runtime);
};
uint64_t hloop_next_event_id();
// async dns resolver (event/hdns.c): free per-loop resolver, called in hloop_cleanup.
void hdns_resolver_free(hloop_t* loop);
struct hidle_s {
HEVENT_FIELDS
uint32_t repeat;
//private:
struct list_node node;
};
#define HTIMER_FIELDS \
HEVENT_FIELDS \
uint32_t repeat; \
uint64_t next_timeout; \
struct heap_node node;
struct htimer_s {
HTIMER_FIELDS
};
struct htimeout_s {
HTIMER_FIELDS
uint32_t timeout; \
};
struct hperiod_s {
HTIMER_FIELDS
int8_t minute;
int8_t hour;
int8_t day;
int8_t week;
int8_t month;
};
QUEUE_DECL(offset_buf_t, write_queue);
// sizeof(struct hio_s)=424 on linux-x64
struct hio_s {
HEVENT_FIELDS
// flags
unsigned ready :1;
unsigned connected :1;
unsigned closed :1;
unsigned accept :1;
unsigned connect :1;
unsigned recv :1;
unsigned send :1;
unsigned recvfrom :1;
unsigned sendto :1;
unsigned close :1;
unsigned alloced_readbuf :1; // for hio_alloc_readbuf
unsigned alloced_ssl_ctx :1; // for hio_new_ssl_ctx
// public:
hio_type_e io_type;
hio_phase_e phase;
uint32_t id; // fd cannot be used as unique identifier, so we provide an id
int fd;
int error;
int events;
int revents;
struct sockaddr* localaddr;
struct sockaddr* peeraddr;
uint64_t last_read_hrtime;
uint64_t last_write_hrtime;
// read
fifo_buf_t readbuf;
unsigned int read_flags;
// for hio_read_until
union {
unsigned int read_until_length;
unsigned char read_until_delim;
};
uint32_t max_read_bufsize;
uint32_t small_readbytes_cnt; // for readbuf autosize
// write
struct write_queue write_queue;
hrecursive_mutex_t write_mutex; // lock write and write_queue
uint32_t write_bufsize;
uint32_t max_write_bufsize;
// callbacks
hread_cb read_cb;
hwrite_cb write_cb;
hclose_cb close_cb;
haccept_cb accept_cb;
hconnect_cb connect_cb;
// timers
int connect_timeout; // ms
int close_timeout; // ms
int read_timeout; // ms
int write_timeout; // ms
int keepalive_timeout; // ms
int heartbeat_interval; // ms
hio_send_heartbeat_fn heartbeat_fn;
htimer_t* connect_timer;
htimer_t* close_timer;
htimer_t* read_timer;
htimer_t* write_timer;
htimer_t* keepalive_timer;
htimer_t* heartbeat_timer;
// upstream
struct hio_s* upstream_io; // for hio_setup_upstream
// unpack
unpack_setting_t* unpack_setting; // for hio_set_unpack
// ssl
void* ssl; // for hio_set_ssl
void* ssl_ctx; // for hio_set_ssl_ctx
char* hostname; // for hssl_set_sni_hostname
// Proxy context. Client connections run a handshake before connect_cb;
// proxy server listeners and accepted connections use the same type.
struct proxy_ctx_s* proxy;
// context
void* ctx; // for hio_context / hio_set_context
// private:
#if defined(EVENT_POLL) || defined(EVENT_KQUEUE)
int event_index[2]; // for poll,kqueue
#endif
#if WITH_RUDP
rudp_t rudp;
#if WITH_KCP
kcp_setting_t* kcp_setting;
#endif
#endif
};
/*
* hio lifeline:
*
* fd =>
* hio_get => HV_ALLOC_SIZEOF(io) => hio_init => hio_ready
*
* hio_read => hio_add(HV_READ) => hio_read_cb
* hio_write => hio_add(HV_WRITE) => hio_write_cb
* hio_close => hio_done => hio_del(HV_RDWR) => hio_close_cb
*
* hloop_stop => hloop_free => hio_free => HV_FREE(io)
*/
void hio_init(hio_t* io);
void hio_ready(hio_t* io);
void hio_done(hio_t* io);
void hio_free(hio_t* io);
uint32_t hio_next_id();
void hio_accept_cb(hio_t* io);
void hio_connect_cb(hio_t* io);
void hio_handle_read(hio_t* io, void* buf, int readbytes);
void hio_read_cb(hio_t* io, void* buf, int len);
void hio_write_cb(hio_t* io, const void* buf, int len);
void hio_close_cb(hio_t* io);
void hio_del_connect_timer(hio_t* io);
void hio_del_close_timer(hio_t* io);
void hio_del_read_timer(hio_t* io);
void hio_del_write_timer(hio_t* io);
void hio_del_keepalive_timer(hio_t* io);
void hio_del_heartbeat_timer(hio_t* io);
static inline void hio_use_loop_readbuf(hio_t* io) {
hloop_t* loop = io->loop;
if (loop->readbuf.len == 0) {
loop->readbuf.len = HLOOP_READ_BUFSIZE;
HV_ALLOC(loop->readbuf.base, loop->readbuf.len);
}
io->readbuf.base = loop->readbuf.base;
io->readbuf.len = loop->readbuf.len;
}
static inline bool hio_is_loop_readbuf(hio_t* io) {
return io->readbuf.base == io->loop->readbuf.base;
}
static inline bool hio_is_alloced_readbuf(hio_t* io) {
return io->alloced_readbuf;
}
void hio_alloc_readbuf(hio_t* io, int len);
void hio_free_readbuf(hio_t* io);
void hio_memmove_readbuf(hio_t* io);
#define EVENT_ENTRY(p) container_of(p, hevent_t, pending_node)
#define IDLE_ENTRY(p) container_of(p, hidle_t, node)
#define TIMER_ENTRY(p) container_of(p, htimer_t, node)
#define EVENT_ACTIVE(ev) \
if (!ev->active) {\
ev->active = 1;\
ev->loop->nactives++;\
}\
#define EVENT_INACTIVE(ev) \
if (ev->active) {\
ev->active = 0;\
ev->loop->nactives--;\
}\
#define EVENT_PENDING(ev) \
do {\
if (!ev->pending) {\
ev->pending = 1;\
ev->loop->npendings++;\
hevent_t** phead = &ev->loop->pendings[HEVENT_PRIORITY_INDEX(ev->priority)];\
ev->pending_next = *phead;\
*phead = (hevent_t*)ev;\
}\
} while(0)
#define EVENT_ADD(loop, ev, cb) \
do {\
ev->loop = loop;\
ev->event_id = hloop_next_event_id();\
ev->cb = (hevent_cb)cb;\
EVENT_ACTIVE(ev);\
} while(0)
#define EVENT_DEL(ev) \
do {\
EVENT_INACTIVE(ev);\
if (!ev->pending) {\
HV_FREE(ev);\
}\
} while(0)
#define EVENT_RESET(ev) \
do {\
ev->destroy = 0;\
EVENT_ACTIVE(ev);\
ev->pending = 0;\
} while(0)
#define EVENT_UNPENDING(ev) \
do {\
if (ev->pending) {\
ev->pending = 0;\
ev->loop->npendings--;\
}\
} while(0)
#endif // HV_EVENT_H_