Line data Source code
1 : #include <stdio.h>
2 : #include <unistd.h>
3 : #include <string.h>
4 : #include <sys/timerfd.h>
5 : #include <sys/eventfd.h>
6 : #include <sys/stat.h>
7 : #include <stdio.h>
8 : #include "log.h"
9 :
10 : #include "netio/netio.h"
11 :
12 : const char* resource_name[] = { "NETIO_TIMER", "NETIO_SIGNAL", "NETIO_CQ", "NETIO_EQ", "NETIO_TCP"};
13 : const char* socket_name[] = {"BSEND", "USEND", "BRECV", "URECV", "BSUB", "USUB", "BPUB", "UPUB", "BLISTEN", "ULISTEN", "NOSOCKET"};
14 :
15 : #if defined DEBUG || defined DEBUG_EV
16 : #define __FILENAME__ (strrchr(__FILE__, '/') ? strrchr(__FILE__, '/') + 1 : __FILE__)
17 : #define log_dbg(...) log_log(LOG_DEBUG, __FILENAME__, __LINE__, __VA_ARGS__)
18 : #define log_trc(...) log_log(LOG_TRACE, __FILENAME__, __LINE__, __VA_ARGS__)
19 : #else
20 : #define log_dbg(...)
21 : #define log_trc(...)
22 : #endif
23 :
24 :
25 : //#define TRACK_ALL_FD //This option is not thread-safe, see FLX-2022.
26 :
27 : /**
28 : * @file
29 : * Functions for the event loop.
30 : */
31 :
32 :
33 : // STATIC FUNCTIONS ////////////////////////////////////////////////////////////
34 :
35 147 : static int check_fd_is_closed(struct closed_fds* closedfds, int fd){
36 416 : for(int i = 0; i < closedfds->count; i++){
37 270 : if(closedfds->fds[i] == fd){
38 : return true;
39 : }
40 : }
41 : return false;
42 : }
43 :
44 :
45 : static void
46 57191901 : process_event(struct netio_event_context* evc, struct closed_fds* closedfds)
47 : {
48 57192048 : if(closedfds->count == 0 || !check_fd_is_closed(closedfds, evc->fd)){
49 57191900 : if(evc->cb != NULL) {
50 57191900 : evc->cb(evc->fd, evc->data);
51 : }
52 : }
53 57191835 : }
54 :
55 2814 : static void add_closed_fd(struct closed_fds* closedfds, int fd){
56 2814 : if (closedfds->count +1 >= 3*NETIO_MAX_POLLED_FIDS){log_warn("Cannot add FD %d to list of closed FDs because Array is full.", fd); return;}
57 2814 : closedfds->fds[closedfds->count++] = fd;
58 : }
59 :
60 54268513 : static void reset_closed_fds(struct closed_fds* closedfds){
61 54268513 : closedfds->count = 0;
62 : }
63 :
64 : static void
65 575370 : set_timerfd(int fd, unsigned s, unsigned ns)
66 : {
67 575370 : struct itimerspec it;
68 575370 : it.it_interval.tv_sec = s;
69 575370 : it.it_interval.tv_nsec = ns;
70 575370 : it.it_value.tv_sec = s;
71 575370 : it.it_value.tv_nsec = ns;
72 575370 : if(timerfd_settime(fd, 0, &it, NULL)) {
73 0 : log_error("Could not set timerfd %d. The timer will not fire.", fd);
74 0 : return;
75 : }
76 : }
77 :
78 : static void
79 8923 : register_fd(int epfd, struct netio_event_context* ctx, int flags)
80 : {
81 8923 : struct epoll_event ev;
82 8923 : ev.events = flags;
83 8923 : ev.data.ptr = ctx;
84 8923 : int rc = fcntl(ctx->fd, F_SETFL, fcntl(ctx->fd, F_GETFL) | O_NONBLOCK );
85 8923 : if (rc < 0) {
86 0 : log_error("Failed to change flags (incl. O_NONBLOCK) of file descriptor %d.", ctx->fd);
87 : }
88 8923 : log_dbg("Adding %d to epoll %d", ctx->fd, epfd);
89 8923 : if(epoll_ctl(epfd, EPOLL_CTL_ADD, ctx->fd, &ev))
90 : {
91 0 : log_error("Could not add file descriptor %d to epoll. Events from this resource will be neglected.", ctx->fd);
92 0 : return;
93 : }
94 : }
95 :
96 : // PRIVATE FUNCTIONS ///////////////////////////////////////////////////////////
97 :
98 : void
99 2516234 : netio_timer_callback(int fd, void* data)
100 : {
101 2516234 : log_trc("timer event on FD %d.", fd);
102 2516234 : struct netio_timer* timer = (struct netio_timer*)data;
103 2516234 : uint64_t buf;
104 2516234 : if(8 != read(fd, &buf, 8)) {
105 2516234 : log_dbg("Did not read 8 bytes.");
106 : }
107 2516234 : if(timer->cb)
108 2516234 : timer->cb(timer->data);
109 2516234 : }
110 :
111 : void
112 7779175 : netio_signal_callback(int fd, void* data)
113 : {
114 7779175 : log_dbg("signal event on FD %d.", fd);
115 :
116 7779175 : struct netio_signal* signal = (struct netio_signal*)data;
117 7779175 : uint64_t buf;
118 7779175 : if(8 != read(fd, &buf, 8)) {
119 0 : log_info("Did not read 8 bytes.");
120 : }
121 7779175 : log_dbg("Count = %lu", buf);
122 7779175 : if(signal->cb)
123 7779175 : signal->cb(signal->data);
124 7779175 : }
125 :
126 : void
127 16 : netio_error_connection_refused_callback(int fd, void* data)
128 : {
129 16 : log_dbg("error event on FD %d.", fd);
130 :
131 16 : struct netio_event_context* ev_ctx = (struct netio_event_context*)data;
132 16 : struct netio_send_socket* socket;
133 16 : if(8 != read(fd, &socket, 8)) {
134 16 : log_dbg("Did not read 8 bytes.");
135 : }
136 16 : if (socket->cb_error_connection_refused) {
137 16 : socket->cb_error_connection_refused(socket);
138 : } else {
139 0 : log_error("Send socket %p has no connection refused callback set. Resources not freed.", socket);
140 : }
141 16 : free(ev_ctx);
142 16 : }
143 :
144 : void
145 19 : netio_error_bind_refused_callback(int fd, void* data)
146 : {
147 19 : log_dbg("error event on FD %d.", fd);
148 :
149 19 : struct netio_event_context* ev_ctx = (struct netio_event_context*)data;
150 19 : struct netio_listen_socket* socket;
151 19 : if(8 != read(fd, &socket, 8)) {
152 19 : log_dbg("Did not read 8 bytes.");
153 : }
154 19 : if (socket->cb_error_bind_refused) {
155 12 : socket->cb_error_bind_refused(socket);
156 : } else {
157 7 : log_error("Listen socket %p has no bin refused callback set. Resources not freed.", socket);
158 : }
159 19 : free(ev_ctx);
160 19 : }
161 :
162 : // INTERNAL RESOURCES ///////////////////////////////////////////////////////////////
163 :
164 : void
165 927 : init_polled_fids(struct polled_fids* pfids, int initial_size){
166 927 : pfids->size = initial_size;
167 927 : pfids->count = 0;
168 927 : pfids->fabric = NULL;
169 927 : pfids->fid_set = malloc(pfids->size*sizeof(struct fid*));
170 927 : pfids->data = malloc(pfids->size*sizeof(struct polled_fids_data));
171 927 : }
172 :
173 : void
174 915 : init_openfds(struct open_fds* fds, int initial_size){
175 915 : fds->size = initial_size;
176 915 : fds->count = 0;
177 915 : fds->data = malloc(fds->size*sizeof(struct open_fd_data));
178 915 : }
179 :
180 :
181 : void
182 0 : print_polled_fids(struct polled_fids* pfids){
183 0 : printf("Number of polled fds %u \n", pfids->count);
184 0 : printf("FD \t FID \n");
185 0 : for(unsigned int i=0; i < pfids->count; ++i){
186 0 : printf("%d \t %p \n", pfids->data[i].fd, pfids->fid_set[i]);
187 : }
188 0 : printf("-------------------\n");
189 0 : }
190 :
191 : void
192 0 : print_openfds(struct open_fds* fds){
193 0 : printf("Number of open fds %u \n", fds->count);
194 0 : printf("===============================================\n");
195 0 : printf("FD \t RESOURCE \t SOCKET \t OBJ ADDR \n");
196 0 : printf("-----------------------------------------------\n");
197 0 : for(unsigned int i=0; i < fds->count; ++i){
198 0 : int r = fds->data[i].rtype;
199 0 : int s = fds->data[i].stype;
200 0 : printf("%d\t%s\t%s\t%p\n", fds->data[i].fd, resource_name[r], socket_name[s], fds->data[i].object);
201 : }
202 0 : printf("===============================================\n");
203 0 : }
204 :
205 :
206 : void
207 3093 : add_polled_fid(struct polled_fids* pfids, struct fid_fabric* fab, struct fid* fid, int fd, void* socket, void (*cb)(int,void*)){
208 3093 : if(pfids->size <= pfids->count){
209 2 : log_dbg("Reallocing polled fids");
210 2 : pfids->fid_set = realloc(pfids->fid_set, 2*(pfids->size)*sizeof(struct fid*));
211 2 : pfids->data = realloc(pfids->data, 2*(pfids->size)*sizeof(struct polled_fids_data));
212 2 : pfids->size *= 2;
213 3093 : };
214 3093 : log_dbg("Polled_fids %p Adding polled fd %d fid %p.", pfids, fd, fid);
215 3093 : pfids->fabric = fab;
216 3093 : pfids->fid_set[pfids->count] = fid;
217 3093 : pfids->data[pfids->count].fd = fd;
218 3093 : pfids->data[pfids->count].socket = socket;
219 3093 : pfids->data[pfids->count].cb = cb;
220 3093 : pfids->count++;
221 : //print_polled_fids(pfids);
222 3093 : };
223 :
224 : void
225 3701 : add_open_fd(struct open_fds* fds, int fd, enum resource_type rtype, enum socket_type stype, void* object){
226 3701 : if(fds->size <= fds->count){
227 0 : log_dbg("Reallocing open fds");
228 0 : fds->data = realloc(fds->data, 2*(fds->size)*sizeof(struct open_fd_data));
229 0 : fds->size *= 2;
230 : };
231 : log_dbg("New open fd %d res type %s socket type %s", fd, resource_name[rtype], socket_name[stype]);
232 11089 : for(unsigned int i=0; i<fds->count; ++i){
233 7388 : if (fd == fds->data[i].fd){
234 0 : log_error("Adding again fd % to open fds!", fd);
235 : }
236 : }
237 3701 : fds->data[fds->count].fd = fd;
238 3701 : fds->data[fds->count].object = object;
239 3701 : fds->data[fds->count].rtype = rtype;
240 3701 : fds->data[fds->count].stype = stype;
241 3701 : fds->count++;
242 3701 : }
243 :
244 :
245 : void
246 2507 : remove_polled_fid(struct polled_fids* pfids, int fd){
247 2507 : log_dbg("Polled_fids %p removing polled fd %d.", pfids, fd);
248 4048 : for(unsigned int i = 0; i < pfids->count; i++){
249 3996 : if(fd == pfids->data[i].fd){
250 : log_dbg("FD %d FID %p removed.", fd, pfids->fid_set[i]);
251 4269 : for(unsigned int j = i; j < pfids->count-1; j++){
252 1814 : pfids->fid_set[j] = pfids->fid_set[j+1];
253 1814 : pfids->data[j] = pfids->data[j+1];
254 : }
255 2455 : pfids->count -= 1;
256 2455 : break;
257 : }
258 : }
259 : //print_polled_fids(pfids);
260 2507 : }
261 :
262 : void
263 2814 : remove_open_fd(struct netio_eventloop* ev, int fd){
264 2814 : struct open_fds* fds = &ev->openfds;
265 5735 : for(unsigned int i = 0; i < fds->count; i++){
266 5735 : if(fd == fds->data[i].fd){
267 : log_dbg("Removing from open fd record fd %d res type %s socket type %s. Current registered fds %u", fd, resource_name[fds->data[i].rtype], socket_name[fds->data[i].stype],fds->count);
268 5279 : for(unsigned int j = i; j < fds->count-1; j++){
269 2465 : fds->data[j] = fds->data[j+1];
270 : }
271 2814 : fds->count -= 1;
272 2814 : break;
273 : }
274 : }
275 2814 : add_closed_fd(&ev->closedfds, fd);
276 2814 : }
277 :
278 :
279 : int
280 0 : check_open_fd_exists(struct open_fds* fds, int fd){
281 0 : for(unsigned int i = 0; i < fds->count; i++){
282 0 : if(fd == fds->data[i].fd){
283 : return true;
284 : }
285 : }
286 : return false;
287 : }
288 :
289 :
290 : // API FUNCTIONS ///////////////////////////////////////////////////////////////
291 :
292 : /*! \brief Initializes a NetIO event loop.
293 : *
294 : * In the background this creates an epoll file descriptor handle.
295 : *
296 : * @param evloop The event loop to initialize
297 : */
298 : void
299 915 : netio_eventloop_init(struct netio_eventloop* evloop)
300 : {
301 915 : evloop->epollfd = epoll_create1(0); // no flag passed, same behaviour as epoll_create
302 915 : evloop->events = malloc(sizeof(struct epoll_event)*NETIO_MAX_EPOLL_EVENTS);
303 915 : init_polled_fids(&evloop->pfids, NETIO_MAX_POLLED_FIDS);
304 915 : init_openfds(&evloop->openfds, NETIO_MAX_POLLED_FIDS);
305 915 : reset_closed_fds(&evloop->closedfds);
306 915 : log_dbg("Creating a new eventloop with fd %d", evloop->epollfd);
307 :
308 915 : if(evloop->epollfd == -1) {
309 0 : log_fatal("Could not create epoll fd. Exit.");
310 0 : exit(2);
311 : }
312 :
313 : //termination signal
314 915 : evloop->stop_signal.data = evloop;
315 915 : evloop->stop_signal.cb = netio_stop;
316 915 : netio_signal_init(evloop, &(evloop->stop_signal));
317 915 : log_dbg("stop signal initialised with fd %d", evloop->stop_signal.ev_ctx.fd );
318 915 : }
319 :
320 : /**
321 : * Initializes a timer and registers it with the event loop.
322 : *
323 : * Internally, timers are implemented using `timerfd`.
324 : *
325 : * @param evloop The event loop in which the timer will be registered
326 : */
327 : void
328 1978 : netio_timer_init(struct netio_eventloop* evloop, struct netio_timer* timer)
329 : {
330 1978 : timer->ev_ctx.fd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK);
331 1976 : timer->ev_ctx.data = timer;
332 1976 : timer->ev_ctx.cb = netio_timer_callback;
333 1976 : if(timer->ev_ctx.fd == -1)
334 : {
335 0 : log_error("Could not create timerfd. The timer will not fire.");
336 0 : return;
337 : }
338 1976 : log_dbg("registering timerfd %d to %d", timer->ev_ctx.fd, evloop->epollfd);
339 1976 : netio_register_read_fd(evloop, &timer->ev_ctx);
340 : #if defined TRACK_ALL_FD
341 : add_open_fd(&evloop->openfds, timer->ev_ctx.fd, NETIO_TIMER, NOSOCKET, timer);
342 : #endif
343 : }
344 :
345 : /**
346 : * Deregisters a timer from the event loop and closes its file descriptor.
347 : *
348 : * @param evloop The event loop in which the timer is registered
349 : * @param timer The timer to unregister and close
350 : */
351 : void
352 1558 : netio_timer_close(struct netio_eventloop* evloop, struct netio_timer* timer)
353 : {
354 1558 : netio_timer_stop(timer);
355 1558 : epoll_ctl(evloop->epollfd, EPOLL_CTL_DEL, timer->ev_ctx.fd, NULL);
356 1558 : log_dbg("netio_timer_close: deregistered timer eventfd %d", timer->ev_ctx.fd);
357 : #if defined TRACK_ALL_FD
358 : int ret = close(timer->ev_ctx.fd);
359 : if(!ret){
360 : remove_open_fd(&evloop, timer->ev_ctx.fd);
361 : }
362 : #else
363 1558 : close(timer->ev_ctx.fd);
364 : #endif
365 1558 : }
366 :
367 : /**
368 : * Starts a timer with the defined period (given in seconds).
369 : *
370 : * The period is given in seconds. The timer callback is executed at the
371 : * defined frequency until it is explicitly stopped.
372 : *
373 : * @param timer The timer to start
374 : * @param seconds The timer period, given in seconds
375 : */
376 : void
377 86 : netio_timer_start_s(struct netio_timer* timer, unsigned long long seconds)
378 : {
379 86 : set_timerfd(timer->ev_ctx.fd, seconds, 0);
380 86 : }
381 :
382 : /**
383 : * Start a timer with the defined period (given in milliseconds).
384 : * @see netio_timer_start_s
385 : */
386 : void
387 287241 : netio_timer_start_ms(struct netio_timer* timer, unsigned long long milliseconds)
388 : {
389 287241 : set_timerfd(timer->ev_ctx.fd, milliseconds/1000, (milliseconds%1000)*1000*1000);
390 287241 : }
391 :
392 : /**
393 : * Start a timer with the defined period (given in microseconds).
394 : * @see netio_timer_start_s
395 : */
396 : void
397 2 : netio_timer_start_us(struct netio_timer* timer, unsigned long long microseconds)
398 : {
399 2 : set_timerfd(timer->ev_ctx.fd, microseconds/(1000*1000), (microseconds%(1000*1000)*1000));
400 2 : }
401 :
402 : /**
403 : * Start a timer with the defined period (given in nanoseconds).
404 : * @see netio_timer_start_s
405 : */
406 : void
407 0 : netio_timer_start_ns(struct netio_timer* timer, unsigned long long nanoseconds)
408 : {
409 0 : set_timerfd(timer->ev_ctx.fd, nanoseconds/(1000*1000*1000), nanoseconds%(1000*1000*1000));
410 0 : }
411 :
412 : /**
413 : * Stops a timer.
414 : *
415 : * The timer will not execute callbacks anymore until it is started again.
416 : *
417 : * @param timer The timer to stop
418 : */
419 : void
420 288041 : netio_timer_stop(struct netio_timer* timer) {
421 288041 : set_timerfd(timer->ev_ctx.fd, 0, 0);
422 288041 : }
423 :
424 :
425 : /**
426 : * Initializes a signal and registers it in the event loop.
427 : *
428 : * Internally, signals are implemented using `eventfd`.
429 : *
430 : * @param evloop The event loop in which the signal will be registered
431 : * @param signal The signal to initialize
432 : */
433 : void
434 3200 : netio_signal_init(struct netio_eventloop* evloop, struct netio_signal* signal)
435 : {
436 3200 : signal->ev_ctx.fd = eventfd(0, EFD_NONBLOCK | EFD_SEMAPHORE);
437 3200 : signal->ev_ctx.data = signal;
438 3200 : if(signal->ev_ctx.fd == -1)
439 : {
440 0 : log_fatal("Could not open eventfd");
441 0 : exit(2);
442 : }
443 :
444 3200 : signal->ev_ctx.cb = netio_signal_callback;
445 3200 : signal->epollfd = evloop->epollfd;
446 3200 : netio_register_read_fd(evloop, &signal->ev_ctx);
447 : #if defined TRACK_ALL_FD
448 : add_open_fd(&evloop->openfds, signal->ev_ctx.fd, NETIO_SIGNAL, NOSOCKET, signal);
449 : #endif
450 3200 : log_dbg("Registering eventfd %d", signal->ev_ctx.fd);
451 3200 : }
452 :
453 :
454 : void
455 44 : netio_signal_no_semaphore_init(struct netio_eventloop* evloop, struct netio_signal* signal)
456 : {
457 44 : signal->ev_ctx.fd = eventfd(0, EFD_NONBLOCK);
458 44 : signal->ev_ctx.data = signal;
459 44 : if(signal->ev_ctx.fd == -1)
460 : {
461 0 : log_fatal("Could not open eventfd");
462 0 : exit(2);
463 : }
464 :
465 44 : signal->ev_ctx.cb = netio_signal_callback;
466 44 : signal->epollfd = evloop->epollfd;
467 44 : netio_register_read_fd(evloop, &signal->ev_ctx);
468 : #if defined TRACK_ALL_FD
469 : add_open_fd(&evloop->openfds, signal->ev_ctx.fd, NETIO_SIGNAL, NOSOCKET, signal);
470 : #endif
471 44 : log_dbg("Registering eventfd %d", signal->ev_ctx.fd);
472 44 : }
473 :
474 :
475 :
476 :
477 : /**
478 : * Deregisters a signal from the event loop and closes its file descriptor.
479 : *
480 : * @param evloop The event loop in which the signal will be registered
481 : * @param signal The signal to initialize
482 : */
483 : void
484 2902 : netio_signal_close(struct netio_eventloop* evloop, struct netio_signal* signal)
485 : {
486 2902 : int rc = epoll_ctl(evloop->epollfd, EPOLL_CTL_DEL, signal->ev_ctx.fd, NULL);
487 2902 : if (rc) {log_warn("Cannot deregister signal fd %d from evloop %d, %s", signal->ev_ctx.fd, evloop->epollfd, strerror(errno));}
488 2902 : log_dbg("netio_signal_close: deregistered signal eventfd %d, ret %d from evloop %d", signal->ev_ctx.fd, rc, signal->epollfd);
489 2902 : int ret = close(signal->ev_ctx.fd);
490 2902 : if(ret) {log_warn("Cannot close %d: %s", signal->ev_ctx.fd, strerror(errno));}
491 : #if defined TRACK_ALL_FD
492 : remove_open_fd(&evloop, signal->ev_ctx.fd);
493 : #endif
494 2902 : }
495 :
496 :
497 : /**
498 : * Fires a signal.
499 : *
500 : * Firing the signal triggers the execution of the signal's callback. Firing
501 : * a signal is thread-safe.
502 : *
503 : * @param signal The signal to fire
504 : */
505 : void
506 13210569 : netio_signal_fire(struct netio_signal* signal)
507 : {
508 13210569 : uint64_t buf = 1;
509 13210569 : int ret = write(signal->ev_ctx.fd, &buf, 8);
510 13210631 : if( ret !=8 ){
511 0 : log_error("Firing signal writing on fd %d, only %d / 8 bytes written. Errno %s", signal->ev_ctx.fd, ret, strerror(errno));
512 : }
513 13210631 : }
514 :
515 : /**
516 : * Fires a callback for error_connection_refused.
517 : *
518 : * Firing triggers the execution of the error_connection_refused callback.
519 : *
520 : * @param socket The socket to use as parameter for the callback
521 : */
522 16 : void netio_error_connection_refused_fire(struct netio_send_socket* socket)
523 : {
524 16 : struct netio_event_context* ev_ctx = malloc(sizeof(struct netio_event_context));
525 16 : ev_ctx->fd = eventfd(0, EFD_NONBLOCK);
526 : // event context is freed by the callback
527 16 : ev_ctx->data = ev_ctx;
528 16 : if (ev_ctx->fd == -1)
529 : {
530 0 : log_fatal("Could not open eventfd for send socket error_connection_refused");
531 0 : exit(2);
532 : }
533 :
534 16 : ev_ctx->cb = netio_error_connection_refused_callback;
535 :
536 16 : netio_register_read_fd(&socket->ctx->evloop, ev_ctx);
537 16 : add_open_fd(&socket->ctx->evloop.openfds, ev_ctx->fd, NETIO_SIGNAL, USEND, socket);
538 16 : log_dbg("listen_socket netio_error_connection_refused_fire: registering eventfd %d", ev_ctx->fd);
539 :
540 : // Socket is transmitted as data to the file descriptor
541 : // struct netio_send_socket* buf = socket;
542 16 : write(ev_ctx->fd, &socket, 8);
543 16 : }
544 :
545 : /**
546 : * Fires a callback for error_bind_refused.
547 : *
548 : * Firing triggers the execution of the error_bind_refused callback.
549 : *
550 : * @param socket The socket to use as parameter for the callback
551 : */
552 19 : void netio_error_bind_refused_fire(struct netio_listen_socket* socket)
553 : {
554 19 : struct netio_event_context* ev_ctx = malloc(sizeof(struct netio_event_context));
555 19 : ev_ctx->fd = eventfd(0, EFD_NONBLOCK);
556 : // event context is freed by the callback
557 19 : ev_ctx->data = ev_ctx;
558 19 : if (ev_ctx->fd == -1)
559 : {
560 0 : log_fatal("Could not open eventfd for listen socket error_bind_refused");
561 0 : exit(2);
562 : }
563 :
564 19 : ev_ctx->cb = netio_error_bind_refused_callback;
565 :
566 19 : netio_register_read_fd(&socket->ctx->evloop, ev_ctx);
567 19 : add_open_fd(&socket->ctx->evloop.openfds, ev_ctx->fd, NETIO_SIGNAL, ULISTEN, socket);
568 19 : log_dbg("listen_socket netio_error_bind_refused_fire: registering eventfd %d", ev_ctx->fd);
569 :
570 : // Socket is transmitted as data to the file descriptor
571 : // struct netio_listen_socket* buf = socket;
572 19 : write(ev_ctx->fd, &socket, 8);
573 19 : }
574 :
575 :
576 : /**
577 : * Executes the event loop.
578 : *
579 : * The event loop is executed in an endless loop until it is explicitly
580 : * terminated by `netio_terminate`. Before any processing any other event,
581 : * `netio_run` will execute this initialization callback, if one was specified.
582 : * The core of the event loop is epoll_wait.
583 : * Note that epoll_wait returns only one event per fd, so MAX_EPOLL_EVENTS actually
584 : * translates into the maximum number of fds that are processed in one iteration
585 : * (the remaining fds are processed in a round-robin fashion in the next iteration).
586 : *
587 : * @param evloop The event loop to execute.
588 : */
589 : void
590 915 : netio_run(struct netio_eventloop* evloop)
591 : {
592 915 : evloop->is_running = 1;
593 915 : int nevents;
594 :
595 915 : if(evloop->cb_init != NULL) {
596 915 : evloop->cb_init(evloop->data);
597 : }
598 : int running=1;
599 54268513 : while(running) {
600 : // don't want to block or wait too long if we're shutting down
601 54267751 : uint64_t timeout = evloop->is_running ? NETIO_EPOLL_TIMEOUT : 10;
602 54267751 : nevents = epoll_wait(evloop->epollfd, evloop->events, NETIO_MAX_EPOLL_EVENTS,
603 : timeout);
604 54267751 : log_trc("%s epoll wait: %d events to process" nevents);
605 :
606 111459583 : for(int i = 0; i < nevents; ++i)
607 : {
608 57191929 : log_trc("event type: %x from fd %d", evloop->events[i].events, ((struct netio_event_context*)evloop->events[i].data.ptr)->fd);
609 57191929 : process_event((struct netio_event_context*)(evloop->events[i].data.ptr), &evloop->closedfds);
610 57191832 : if(evloop->events[i].events & EPOLLRDHUP)
611 : {
612 0 : struct netio_event_context* c = (struct netio_event_context*)(evloop->events[i].data.ptr);
613 0 : log_dbg("EPOLLRDHUP on fd %d, removing it from epoll_wait", c->fd);
614 0 : epoll_ctl(evloop->epollfd, EPOLL_CTL_DEL, c->fd, NULL);
615 : }
616 : }
617 54267598 : if (evloop->is_running==0 && nevents==0) {
618 762 : running=0;
619 : }
620 54267598 : reset_closed_fds(&evloop->closedfds);
621 54267598 : if(unlikely(nevents == -1))
622 : {
623 43 : int errsv = errno;
624 43 : if(errsv==EINTR) {
625 43 : log_dbg("Eventloop: non-blocking epoll_wait returned -1: %s", strerror(errsv));
626 43 : continue;
627 : }
628 : else {
629 0 : log_fatal("Eventloop: non-blocking epoll_wait returned -1: %s", strerror(errsv));
630 0 : free(evloop->events);
631 0 : exit(1);
632 : }
633 : }
634 : }//end of while running
635 :
636 762 : log_dbg("Cleaning up eventloop");
637 762 : close(evloop->epollfd);
638 1274 : for(unsigned int i=0; i < evloop->openfds.count; ++i){
639 512 : struct open_fd_data* data = &evloop->openfds.data[i];
640 512 : log_warn("Evloop terminated, closing fd %d type %s socket type %s", data->fd, resource_name[data->rtype], socket_name[data->stype]);
641 512 : close(data->fd);
642 : }
643 762 : free(evloop->openfds.data);
644 762 : free(evloop->pfids.data);
645 762 : free(evloop->pfids.fid_set);
646 762 : free(evloop->events);
647 762 : }
648 :
649 :
650 : /**
651 : * Terminates the event loop.
652 : *
653 : * `netio_terminate` will stop the execution of the event loop. This will *not*
654 : * interrupt any event processing code or user callback that is currently being
655 : * executed. Event loop execution will halt after the processing of the current
656 : * event has completed.
657 : *
658 : * @param evloop The event loop to terminate
659 : */
660 : void
661 242 : netio_terminate(struct netio_eventloop* evloop)
662 : {
663 242 : netio_stop((void*)evloop);
664 242 : }
665 :
666 :
667 : /**
668 : * Terminates the event loop.
669 : *
670 : * `netio_terminate_signal` will add a signal to the eventloop
671 : * such that queued events, including those related to resources deallocation,
672 : * can be executed.
673 : *
674 : * @param evloop The event loop to terminate
675 : */
676 : void
677 595 : netio_terminate_signal(struct netio_eventloop* evloop)
678 : {
679 595 : log_info("Firing termination signal");
680 595 : netio_signal_fire(&evloop->stop_signal);
681 595 : }
682 :
683 : void
684 801 : netio_stop(void* ptr)
685 : {
686 801 : struct netio_eventloop* evloop = (struct netio_eventloop*)ptr;
687 801 : netio_signal_close(evloop, &evloop->stop_signal);
688 801 : evloop->pfids.count = 0;
689 801 : evloop->is_running = 0;
690 801 : }
691 :
692 : void
693 8923 : netio_register_read_fd(struct netio_eventloop* evloop, struct netio_event_context* ctx)
694 : {
695 8923 : register_fd(evloop->epollfd, ctx, EPOLLIN | EPOLLRDHUP);
696 8923 : }
697 :
698 :
699 : /**
700 : * Closes a socket.
701 : *
702 : * `netio_close_socket` will deallocate the resources associated to a socket
703 : * and close the corresponding file descriptor. The type of socket is specified
704 : * by the third argument. The socket is not freed as the function ignores whether
705 : * it had been allocated statically or dynamically.
706 : *
707 : * @param evloop The event loop to which the socket belongs
708 : * @param socket Pointer to the socket
709 : * @param type Type of socket as defined by enum socket_type.
710 : */
711 : void
712 436 : netio_close_socket(struct netio_eventloop* evloop, void* socket, enum socket_type type)
713 : {
714 : //Sockets that contain more than one other socket
715 436 : switch (type){
716 0 : case BPUB: {
717 0 : struct netio_publish_socket* pub = (struct netio_publish_socket*)socket;
718 0 : struct netio_socket_list* it = pub->subscription_table.socket_list;
719 0 : while(it != NULL){
720 0 : if(it->socket){
721 0 : struct netio_buffered_send_socket* bs_socket = (struct netio_buffered_send_socket*)it->socket;
722 0 : netio_disconnect(&(bs_socket->send_socket)); //Do I need to disconnect for the sake of the other side?
723 0 : free(it->socket);
724 0 : if(it->addrlen > 0){ free(it->addr); }
725 : }
726 0 : struct netio_socket_list* tmp = it;
727 0 : it = it->next;
728 0 : free(tmp);
729 : }
730 0 : netio_close_socket(evloop, (void*)(&pub->lsocket), ULISTEN);
731 0 : return;
732 : }
733 :
734 0 : case UPUB: {
735 0 : struct netio_unbuffered_publish_socket* upub = (struct netio_unbuffered_publish_socket*)socket;
736 0 : struct netio_socket_list* uit = upub->subscription_table.socket_list;
737 0 : while(uit != NULL){
738 0 : if(uit->socket){
739 0 : struct netio_send_socket* s_socket = (struct netio_send_socket*)uit->socket;
740 0 : netio_disconnect(s_socket); //Do I need to disconnec for the sake of the other side?
741 0 : free(uit->socket);
742 0 : if(uit->addrlen > 0){ free (uit->addr); }
743 : }
744 0 : struct netio_socket_list* tmp = uit;
745 0 : uit = uit->next;
746 0 : free(tmp);
747 : }
748 0 : netio_close_socket(evloop, (void*)(&upub->lsocket), ULISTEN);
749 0 : struct netio_completion_stack* cs = &upub->completion_stack;
750 0 : free(cs->stack);
751 0 : free(cs->objects);
752 0 : free(cs->key_array);
753 0 : return;
754 : }
755 :
756 136 : case BSUB: {
757 136 : struct netio_subscribe_socket* sub_socket = (struct netio_subscribe_socket*)socket;
758 136 : netio_disconnect(&sub_socket->socket);
759 136 : netio_close_socket(evloop, &sub_socket->recv_socket, BLISTEN);
760 136 : if (sub_socket->remote_hostname) {
761 136 : free((void*)sub_socket->remote_hostname);
762 136 : sub_socket->remote_hostname=NULL;
763 : }
764 : return;
765 : }
766 :
767 82 : case USUB: {
768 82 : struct netio_unbuffered_subscribe_socket* usub_socket = (struct netio_unbuffered_subscribe_socket*)socket;
769 82 : netio_disconnect(&usub_socket->socket);
770 82 : netio_close_socket(evloop, &usub_socket->recv_socket, ULISTEN);
771 82 : if (usub_socket->remote_hostname) {
772 82 : free((void*)usub_socket->remote_hostname);
773 82 : usub_socket->remote_hostname=NULL;
774 : }
775 : return;
776 : }
777 :
778 : case NOSOCKET:
779 : return;
780 :
781 218 : default:
782 218 : ;
783 : //go on with the function
784 : }
785 :
786 218 : struct netio_signal* signal_close_socket = malloc(sizeof(struct netio_signal));
787 218 : struct signal_data* sd = malloc(sizeof(struct signal_data));
788 218 : sd->signal = signal_close_socket;
789 218 : sd->ptr = socket;
790 218 : sd->evloop = evloop;
791 :
792 218 : switch (type){
793 0 : case USEND:
794 0 : signal_close_socket->cb = close_send_socket;
795 0 : break;
796 0 : case BSEND:
797 0 : signal_close_socket->cb = close_buffered_send_socket;
798 0 : break;
799 0 : case URECV:
800 0 : signal_close_socket->cb = close_recv_socket;
801 0 : break;
802 0 : case BRECV:
803 0 : signal_close_socket->cb = close_buffered_recv_socket;
804 0 : break;
805 82 : case ULISTEN:
806 82 : signal_close_socket->cb = close_listen_socket;
807 82 : break;
808 136 : case BLISTEN: {
809 136 : struct netio_buffered_listen_socket* lsocket = (struct netio_buffered_listen_socket*)socket;
810 136 : sd->ptr = (void*)(&lsocket->listen_socket);
811 136 : signal_close_socket->cb = close_buffered_listen_socket;
812 136 : break;
813 : }
814 : case NOSOCKET:
815 : return;
816 :
817 0 : default:
818 0 : log_error("Could not delete socket: type unknown.");
819 0 : break;
820 : }
821 218 : signal_close_socket->data = sd;
822 218 : netio_signal_init(evloop, signal_close_socket);
823 218 : netio_signal_fire(signal_close_socket);
824 : }
|
