/src/uWebSockets/uSockets/src/loop.c

Source
/*
 * Authored by Alex Hultman, 2018-2021.
 * Intellectual property of third-party.

 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at

 *     http://www.apache.org/licenses/LICENSE-2.0

 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#ifndef LIBUS_USE_IO_URING

#include "libusockets.h"
#include "internal/internal.h"
#include <stdlib.h>

/* The loop has 2 fallthrough polls */
void us_internal_loop_data_init(struct us_loop_t *loop, void (*wakeup_cb)(struct us_loop_t *loop),
    void (*pre_cb)(struct us_loop_t *loop), void (*post_cb)(struct us_loop_t *loop)) {
    loop->data.sweep_timer = us_create_timer(loop, 1, 0);
    loop->data.recv_buf = malloc(LIBUS_RECV_BUFFER_LENGTH + LIBUS_RECV_BUFFER_PADDING * 2);
    loop->data.ssl_data = 0;
    loop->data.head = 0;
    loop->data.iterator = 0;
    loop->data.closed_head = 0;
    loop->data.low_prio_head = 0;
    loop->data.low_prio_budget = 0;

    loop->data.pre_cb = pre_cb;
    loop->data.post_cb = post_cb;
    loop->data.iteration_nr = 0;

    loop->data.wakeup_async = us_internal_create_async(loop, 1, 0);
    us_internal_async_set(loop->data.wakeup_async, (void (*)(struct us_internal_async *)) wakeup_cb);
}

void us_internal_loop_data_free(struct us_loop_t *loop) {
#ifndef LIBUS_NO_SSL
    us_internal_free_loop_ssl_data(loop);
#endif

    free(loop->data.recv_buf);

    us_timer_close(loop->data.sweep_timer);
    us_internal_async_close(loop->data.wakeup_async);
}

void us_wakeup_loop(struct us_loop_t *loop) {
    us_internal_async_wakeup(loop->data.wakeup_async);
}

void us_internal_loop_link(struct us_loop_t *loop, struct us_socket_context_t *context) {
    /* Insert this context as the head of loop */
    context->next = loop->data.head;
    context->prev = 0;
    if (loop->data.head) {
        loop->data.head->prev = context;
    }
    loop->data.head = context;
}

/* Unlink is called before free */
void us_internal_loop_unlink(struct us_loop_t *loop, struct us_socket_context_t *context) {
    if (loop->data.head == context) {
        loop->data.head = context->next;
        if (loop->data.head) {
            loop->data.head->prev = 0;
        }
    } else {
        context->prev->next = context->next;
        if (context->next) {
            context->next->prev = context->prev;
        }
    }
}

/* This functions should never run recursively */
void us_internal_timer_sweep(struct us_loop_t *loop) {
    struct us_internal_loop_data_t *loop_data = &loop->data;
    /* For all socket contexts in this loop */
    for (loop_data->iterator = loop_data->head; loop_data->iterator; loop_data->iterator = loop_data->iterator->next) {

        struct us_socket_context_t *context = loop_data->iterator;

        /* Update this context's timestamps (this could be moved to loop and done once) */
        context->global_tick++;
        unsigned char short_ticks = context->timestamp = context->global_tick % 240;
        unsigned char long_ticks = context->long_timestamp = (context->global_tick / 15) % 240;

        /* Begin at head */
        struct us_socket_t *s = context->head_sockets;
        while (s) {
            /* Seek until end or timeout found (tightest loop) */
            while (1) {
                /* We only read from 1 random cache line here */
                if (short_ticks == s->timeout || long_ticks == s->long_timeout) {
                    break;
                }

                /* Did we reach the end without a find? */
                if ((s = s->next) == 0) {
                    goto next_context;
                }
            }

            /* Here we have a timeout to emit (slow path) */
            context->iterator = s;

            if (short_ticks == s->timeout) {
                s->timeout = 255;
                context->on_socket_timeout(s);
            }

            if (context->iterator == s && long_ticks == s->long_timeout) {
                s->long_timeout = 255;
                context->on_socket_long_timeout(s);
            }   

            /* Check for unlink / link (if the event handler did not modify the chain, we step 1) */
            if (s == context->iterator) {
                s = s->next;
            } else {
                /* The iterator was changed by event handler */
                s = context->iterator;
            }
        }
        /* We always store a 0 to context->iterator here since we are no longer iterating this context */
        next_context:
        context->iterator = 0;
    }
}

/* We do not want to block the loop with tons and tons of CPU-intensive work for SSL handshakes.
 * Spread it out during many loop iterations, prioritizing already open connections, they are far
 * easier on CPU */
static const int MAX_LOW_PRIO_SOCKETS_PER_LOOP_ITERATION = 5;

void us_internal_handle_low_priority_sockets(struct us_loop_t *loop) {
    struct us_internal_loop_data_t *loop_data = &loop->data;
    struct us_socket_t *s;

    loop_data->low_prio_budget = MAX_LOW_PRIO_SOCKETS_PER_LOOP_ITERATION;

    for (s = loop_data->low_prio_head; s && loop_data->low_prio_budget > 0; s = loop_data->low_prio_head, loop_data->low_prio_budget--) {
        /* Unlink this socket from the low-priority queue */
        loop_data->low_prio_head = s->next;
        if (s->next) s->next->prev = 0;
        s->next = 0;

        us_internal_socket_context_link_socket(s->context, s);
        us_poll_change(&s->p, us_socket_context(0, s)->loop, us_poll_events(&s->p) | LIBUS_SOCKET_READABLE);

        s->low_prio_state = 2;
    }
}

/* Note: Properly takes the linked list and timeout sweep into account */
void us_internal_free_closed_sockets(struct us_loop_t *loop) {
    /* Free all closed sockets (maybe it is better to reverse order?) */
    if (loop->data.closed_head) {
        for (struct us_socket_t *s = loop->data.closed_head; s; ) {
            struct us_socket_t *next = s->next;
            us_poll_free((struct us_poll_t *) s, loop);
            s = next;
        }
        loop->data.closed_head = 0;
    }
}

void sweep_timer_cb(struct us_internal_callback_t *cb) {
    us_internal_timer_sweep(cb->loop);
}

long long us_loop_iteration_number(struct us_loop_t *loop) {
    return loop->data.iteration_nr;
}

/* These may have somewhat different meaning depending on the underlying event library */
void us_internal_loop_pre(struct us_loop_t *loop) {
    loop->data.iteration_nr++;
    us_internal_handle_low_priority_sockets(loop);
    loop->data.pre_cb(loop);
}

void us_internal_loop_post(struct us_loop_t *loop) {
    us_internal_free_closed_sockets(loop);
    loop->data.post_cb(loop);
}

struct us_socket_t *us_adopt_accepted_socket(int ssl, struct us_socket_context_t *context, LIBUS_SOCKET_DESCRIPTOR accepted_fd,
    unsigned int socket_ext_size, char *addr_ip, int addr_ip_length) {
#ifndef LIBUS_NO_SSL
    if (ssl) {
        return (struct us_socket_t *)us_internal_ssl_adopt_accepted_socket((struct us_internal_ssl_socket_context_t *)context, accepted_fd,
            socket_ext_size, addr_ip, addr_ip_length);
    }
#endif
    struct us_poll_t *accepted_p = us_create_poll(context->loop, 0, sizeof(struct us_socket_t) - sizeof(struct us_poll_t) + socket_ext_size);
    us_poll_init(accepted_p, accepted_fd, POLL_TYPE_SOCKET);
    us_poll_start(accepted_p, context->loop, LIBUS_SOCKET_READABLE);

    struct us_socket_t *s = (struct us_socket_t *) accepted_p;

    s->context = context;
    s->timeout = 255;
    s->long_timeout = 255;
    s->low_prio_state = 0;

    /* We always use nodelay */
    bsd_socket_nodelay(accepted_fd, 1);

    us_internal_socket_context_link_socket(context, s);

    context->on_open(s, 0, addr_ip, addr_ip_length);
    return s;
}

void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int events) {
    switch (us_internal_poll_type(p)) {
    case POLL_TYPE_CALLBACK: {
            struct us_internal_callback_t *cb = (struct us_internal_callback_t *) p;
            /* Timers, asyncs should accept (read), while UDP sockets should obviously not */
            if (!cb->leave_poll_ready) {
                /* Let's just have this macro to silence the CodeQL alert regarding empty function when using libuv */
    #ifndef LIBUS_USE_LIBUV
                us_internal_accept_poll_event(p);
    #endif
            }
            cb->cb(cb->cb_expects_the_loop ? (struct us_internal_callback_t *) cb->loop : (struct us_internal_callback_t *) &cb->p);
        }
        break;
    case POLL_TYPE_SEMI_SOCKET: {
            /* Both connect and listen sockets are semi-sockets
             * but they poll for different events */
            if (us_poll_events(p) == LIBUS_SOCKET_WRITABLE) {
                struct us_socket_t *s = (struct us_socket_t *) p;

                /* It is perfectly possible to come here with an error */
                if (error) {
                    /* Emit error, close without emitting on_close */
                    s->context->on_connect_error(s, 0);
                    us_socket_close_connecting(0, s);
                } else {
                    /* All sockets poll for readable */
                    us_poll_change(p, s->context->loop, LIBUS_SOCKET_READABLE);

                    /* We always use nodelay */
                    bsd_socket_nodelay(us_poll_fd(p), 1);

                    /* We are now a proper socket */
                    us_internal_poll_set_type(p, POLL_TYPE_SOCKET);

                    /* If we used a connection timeout we have to reset it here */
                    us_socket_timeout(0, s, 0);

                    s->context->on_open(s, 1, 0, 0);
                }
            } else {
                struct us_listen_socket_t *listen_socket = (struct us_listen_socket_t *) p;
                struct bsd_addr_t addr;

                LIBUS_SOCKET_DESCRIPTOR client_fd = bsd_accept_socket(us_poll_fd(p), &addr);
                if (client_fd == LIBUS_SOCKET_ERROR) {
                    /* Todo: start timer here */

                } else {

                    /* Todo: stop timer if any */

                    do {
                        struct us_socket_context_t *context = us_socket_context(0, &listen_socket->s);
                        /* See if we want to export the FD or keep it here (this event can be unset) */
                        if (context->on_pre_open == 0 || context->on_pre_open(context, client_fd) == client_fd) {

                            /* Adopt the newly accepted socket */
                            us_adopt_accepted_socket(0, context,
                                client_fd, listen_socket->socket_ext_size, bsd_addr_get_ip(&addr), bsd_addr_get_ip_length(&addr));

                            /* Exit accept loop if listen socket was closed in on_open handler */
                            if (us_socket_is_closed(0, &listen_socket->s)) {
                                break;
                            }

                        }

                    } while ((client_fd = bsd_accept_socket(us_poll_fd(p), &addr)) != LIBUS_SOCKET_ERROR);
                }
            }
        }
        break;
    case POLL_TYPE_SOCKET_SHUT_DOWN:
    case POLL_TYPE_SOCKET: {
            /* We should only use s, no p after this point */
            struct us_socket_t *s = (struct us_socket_t *) p;

            /* Such as epollerr epollhup */
            if (error) {
                /* Todo: decide what code we give here */
                s = us_socket_close(0, s, 0, NULL);
                return;
            }

            if (events & LIBUS_SOCKET_WRITABLE) {
                /* Note: if we failed a write as a socket of one loop then adopted
                 * to another loop, this will be wrong. Absurd case though */
                s->context->loop->data.last_write_failed = 0;

                s = s->context->on_writable(s);

                if (us_socket_is_closed(0, s)) {
                    return;
                }

                /* If we have no failed write or if we shut down, then stop polling for more writable */
                if (!s->context->loop->data.last_write_failed || us_socket_is_shut_down(0, s)) {
                    us_poll_change(&s->p, us_socket_context(0, s)->loop, us_poll_events(&s->p) & LIBUS_SOCKET_READABLE);
                }
            }

            if (events & LIBUS_SOCKET_READABLE) {
                /* Contexts may prioritize down sockets that are currently readable, e.g. when SSL handshake has to be done.
                 * SSL handshakes are CPU intensive, so we limit the number of handshakes per loop iteration, and move the rest
                 * to the low-priority queue */
                if (s->context->is_low_prio(s)) {
                    if (s->low_prio_state == 2) {
                        s->low_prio_state = 0; /* Socket has been delayed and now it's time to process incoming data for one iteration */
                    } else if (s->context->loop->data.low_prio_budget > 0) {
                        s->context->loop->data.low_prio_budget--; /* Still having budget for this iteration - do normal processing */
                    } else {
                        us_poll_change(&s->p, us_socket_context(0, s)->loop, us_poll_events(&s->p) & LIBUS_SOCKET_WRITABLE);
                        us_internal_socket_context_unlink_socket(s->context, s);

                        /* Link this socket to the low-priority queue - we use a LIFO queue, to prioritize newer clients that are
                         * maybe not already timeouted - sounds unfair, but works better in real-life with smaller client-timeouts
                         * under high load */
                        s->prev = 0;
                        s->next = s->context->loop->data.low_prio_head;
                        if (s->next) s->next->prev = s;
                        s->context->loop->data.low_prio_head = s;

                        s->low_prio_state = 1;

                        break;
                    }
                }

                int length;
                read_more:
                length = bsd_recv(us_poll_fd(&s->p), s->context->loop->data.recv_buf + LIBUS_RECV_BUFFER_PADDING, LIBUS_RECV_BUFFER_LENGTH, 0);
                if (length > 0) {
                    s = s->context->on_data(s, s->context->loop->data.recv_buf + LIBUS_RECV_BUFFER_PADDING, length);

                    /* If we filled the entire recv buffer, we need to immediately read again since otherwise a
                     * pending hangup event in the same even loop iteration can close the socket before we get
                     * the chance to read again next iteration */
                    if (length == LIBUS_RECV_BUFFER_LENGTH && s && !us_socket_is_closed(0, s)) {
                        goto read_more;
                    }

                } else if (!length) {
                    if (us_socket_is_shut_down(0, s)) {
                        /* We got FIN back after sending it */
                        /* Todo: We should give "CLEAN SHUTDOWN" as reason here */
                        s = us_socket_close(0, s, 0, NULL);
                    } else {
                        /* We got FIN, so stop polling for readable */
                        us_poll_change(&s->p, us_socket_context(0, s)->loop, us_poll_events(&s->p) & LIBUS_SOCKET_WRITABLE);
                        s = s->context->on_end(s);
                    }
                } else if (length == LIBUS_SOCKET_ERROR && !bsd_would_block()) {
                    /* Todo: decide also here what kind of reason we should give */
                    s = us_socket_close(0, s, 0, NULL);
                }
            }
        }
        break;
    }
}

/* Integration only requires the timer to be set up */
void us_loop_integrate(struct us_loop_t *loop) {
    us_timer_set(loop->data.sweep_timer, (void (*)(struct us_timer_t *)) sweep_timer_cb, LIBUS_TIMEOUT_GRANULARITY * 1000, LIBUS_TIMEOUT_GRANULARITY * 1000);
}

void *us_loop_ext(struct us_loop_t *loop) {
    return loop + 1;
}

#endif

Coverage Report

Created: 2025-12-18 06:45

Line	Count	Source
1		/*
2		* Authored by Alex Hultman, 2018-2021.
3		* Intellectual property of third-party.
4
5		* Licensed under the Apache License, Version 2.0 (the "License");
6		* you may not use this file except in compliance with the License.
7		* You may obtain a copy of the License at
8
9		* http://www.apache.org/licenses/LICENSE-2.0
10
11		* Unless required by applicable law or agreed to in writing, software
12		* distributed under the License is distributed on an "AS IS" BASIS,
13		* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14		* See the License for the specific language governing permissions and
15		* limitations under the License.
16		*/
17
18		#ifndef LIBUS_USE_IO_URING
19
20		#include "libusockets.h"
21		#include "internal/internal.h"
22		#include <stdlib.h>
23
24		/* The loop has 2 fallthrough polls */
25		void us_internal_loop_data_init(struct us_loop_t loop, void (wakeup_cb)(struct us_loop_t *loop),
26	22.7k	void (pre_cb)(struct us_loop_t loop), void (post_cb)(struct us_loop_t loop)) {
27	22.7k	loop->data.sweep_timer = us_create_timer(loop, 1, 0);
28	22.7k	loop->data.recv_buf = malloc(LIBUS_RECV_BUFFER_LENGTH + LIBUS_RECV_BUFFER_PADDING * 2);
29	22.7k	loop->data.ssl_data = 0;
30	22.7k	loop->data.head = 0;
31	22.7k	loop->data.iterator = 0;
32	22.7k	loop->data.closed_head = 0;
33	22.7k	loop->data.low_prio_head = 0;
34	22.7k	loop->data.low_prio_budget = 0;
35
36	22.7k	loop->data.pre_cb = pre_cb;
37	22.7k	loop->data.post_cb = post_cb;
38	22.7k	loop->data.iteration_nr = 0;
39
40	22.7k	loop->data.wakeup_async = us_internal_create_async(loop, 1, 0);
41	22.7k	us_internal_async_set(loop->data.wakeup_async, (void ()(struct us_internal_async )) wakeup_cb);
42	22.7k	}
43
44	22.7k	void us_internal_loop_data_free(struct us_loop_t *loop) {
45		#ifndef LIBUS_NO_SSL
46		us_internal_free_loop_ssl_data(loop);
47		#endif
48
49	22.7k	free(loop->data.recv_buf);
50
51	22.7k	us_timer_close(loop->data.sweep_timer);
52	22.7k	us_internal_async_close(loop->data.wakeup_async);
53	22.7k	}
54
55	37.7k	void us_wakeup_loop(struct us_loop_t *loop) {
56	37.7k	us_internal_async_wakeup(loop->data.wakeup_async);
57	37.7k	}
58
59	62.8k	void us_internal_loop_link(struct us_loop_t loop, struct us_socket_context_t context) {
60		/* Insert this context as the head of loop */
61	62.8k	context->next = loop->data.head;
62	62.8k	context->prev = 0;
63	62.8k	if (loop->data.head) {
64	40.0k	loop->data.head->prev = context;
65	40.0k	}
66	62.8k	loop->data.head = context;
67	62.8k	}
68
69		/* Unlink is called before free */
70	62.8k	void us_internal_loop_unlink(struct us_loop_t loop, struct us_socket_context_t context) {
71	62.8k	if (loop->data.head == context) {
72	29.5k	loop->data.head = context->next;
73	29.5k	if (loop->data.head) {
74	6.80k	loop->data.head->prev = 0;
75	6.80k	}
76	33.2k	} else {
77	33.2k	context->prev->next = context->next;
78	33.2k	if (context->next) {
79	0	context->next->prev = context->prev;
80	0	}
81	33.2k	}
82	62.8k	}
83
84		/* This functions should never run recursively */
85	8.98M	void us_internal_timer_sweep(struct us_loop_t *loop) {
86	8.98M	struct us_internal_loop_data_t *loop_data = &loop->data;
87		/* For all socket contexts in this loop */
88	32.3M	for (loop_data->iterator = loop_data->head; loop_data->iterator; loop_data->iterator = loop_data->iterator->next) {
89
90	23.4M	struct us_socket_context_t *context = loop_data->iterator;
91
92		/* Update this context's timestamps (this could be moved to loop and done once) */
93	23.4M	context->global_tick++;
94	23.4M	unsigned char short_ticks = context->timestamp = context->global_tick % 240;
95	23.4M	unsigned char long_ticks = context->long_timestamp = (context->global_tick / 15) % 240;
96
97		/* Begin at head */
98	23.4M	struct us_socket_t *s = context->head_sockets;
99	23.4M	while (s) {
100		/* Seek until end or timeout found (tightest loop) */
101	2.30M	while (1) {
102		/* We only read from 1 random cache line here */
103	2.30M	if (short_ticks == s->timeout \|\| long_ticks == s->long_timeout) {
104	66.1k	break;
105	66.1k	}
106
107		/* Did we reach the end without a find? */
108	2.24M	if ((s = s->next) == 0) {
109	546k	goto next_context;
110	546k	}
111	2.24M	}
112
113		/* Here we have a timeout to emit (slow path) */
114	66.1k	context->iterator = s;
115
116	66.1k	if (short_ticks == s->timeout) {
117	66.1k	s->timeout = 255;
118	66.1k	context->on_socket_timeout(s);
119	66.1k	}
120
121	66.1k	if (context->iterator == s && long_ticks == s->long_timeout) {
122	0	s->long_timeout = 255;
123	0	context->on_socket_long_timeout(s);
124	0	}
125
126		/* Check for unlink / link (if the event handler did not modify the chain, we step 1) */
127	66.1k	if (s == context->iterator) {
128	16.8k	s = s->next;
129	49.3k	} else {
130		/* The iterator was changed by event handler */
131	49.3k	s = context->iterator;
132	49.3k	}
133	66.1k	}
134		/* We always store a 0 to context->iterator here since we are no longer iterating this context */
135	23.4M	next_context:
136	23.4M	context->iterator = 0;
137	23.4M	}
138	8.98M	}
139
140		/* We do not want to block the loop with tons and tons of CPU-intensive work for SSL handshakes.
141		* Spread it out during many loop iterations, prioritizing already open connections, they are far
142		* easier on CPU */
143		static const int MAX_LOW_PRIO_SOCKETS_PER_LOOP_ITERATION = 5;
144
145	10.6M	void us_internal_handle_low_priority_sockets(struct us_loop_t *loop) {
146	10.6M	struct us_internal_loop_data_t *loop_data = &loop->data;
147	10.6M	struct us_socket_t *s;
148
149	10.6M	loop_data->low_prio_budget = MAX_LOW_PRIO_SOCKETS_PER_LOOP_ITERATION;
150
151	10.6M	for (s = loop_data->low_prio_head; s && loop_data->low_prio_budget > 0; s = loop_data->low_prio_head, loop_data->low_prio_budget--) {
152		/* Unlink this socket from the low-priority queue */
153	0	loop_data->low_prio_head = s->next;
154	0	if (s->next) s->next->prev = 0;
155	0	s->next = 0;
156
157	0	us_internal_socket_context_link_socket(s->context, s);
158	0	us_poll_change(&s->p, us_socket_context(0, s)->loop, us_poll_events(&s->p) \| LIBUS_SOCKET_READABLE);
159
160	0	s->low_prio_state = 2;
161	0	}
162	10.6M	}
163
164		/* Note: Properly takes the linked list and timeout sweep into account */
165	10.6M	void us_internal_free_closed_sockets(struct us_loop_t *loop) {
166		/* Free all closed sockets (maybe it is better to reverse order?) */
167	10.6M	if (loop->data.closed_head) {
168	6.28M	for (struct us_socket_t *s = loop->data.closed_head; s; ) {
169	5.19M	struct us_socket_t *next = s->next;
170	5.19M	us_poll_free((struct us_poll_t *) s, loop);
171	5.19M	s = next;
172	5.19M	}
173	1.09M	loop->data.closed_head = 0;
174	1.09M	}
175	10.6M	}
176
177	8.98M	void sweep_timer_cb(struct us_internal_callback_t *cb) {
178	8.98M	us_internal_timer_sweep(cb->loop);
179	8.98M	}
180
181	6.80k	long long us_loop_iteration_number(struct us_loop_t *loop) {
182	6.80k	return loop->data.iteration_nr;
183	6.80k	}
184
185		/* These may have somewhat different meaning depending on the underlying event library */
186	10.6M	void us_internal_loop_pre(struct us_loop_t *loop) {
187	10.6M	loop->data.iteration_nr++;
188	10.6M	us_internal_handle_low_priority_sockets(loop);
189	10.6M	loop->data.pre_cb(loop);
190	10.6M	}
191
192	10.6M	void us_internal_loop_post(struct us_loop_t *loop) {
193	10.6M	us_internal_free_closed_sockets(loop);
194	10.6M	loop->data.post_cb(loop);
195	10.6M	}
196
197		struct us_socket_t us_adopt_accepted_socket(int ssl, struct us_socket_context_t context, LIBUS_SOCKET_DESCRIPTOR accepted_fd,
198	5.17M	unsigned int socket_ext_size, char *addr_ip, int addr_ip_length) {
199		#ifndef LIBUS_NO_SSL
200		if (ssl) {
201		return (struct us_socket_t )us_internal_ssl_adopt_accepted_socket((struct us_internal_ssl_socket_context_t )context, accepted_fd,
202		socket_ext_size, addr_ip, addr_ip_length);
203		}
204		#endif
205	5.17M	struct us_poll_t *accepted_p = us_create_poll(context->loop, 0, sizeof(struct us_socket_t) - sizeof(struct us_poll_t) + socket_ext_size);
206	5.17M	us_poll_init(accepted_p, accepted_fd, POLL_TYPE_SOCKET);
207	5.17M	us_poll_start(accepted_p, context->loop, LIBUS_SOCKET_READABLE);
208
209	5.17M	struct us_socket_t s = (struct us_socket_t ) accepted_p;
210
211	5.17M	s->context = context;
212	5.17M	s->timeout = 255;
213	5.17M	s->long_timeout = 255;
214	5.17M	s->low_prio_state = 0;
215
216		/* We always use nodelay */
217	5.17M	bsd_socket_nodelay(accepted_fd, 1);
218
219	5.17M	us_internal_socket_context_link_socket(context, s);
220
221	5.17M	context->on_open(s, 0, addr_ip, addr_ip_length);
222	5.17M	return s;
223	5.17M	}
224
225	42.0M	void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int events) {
226	42.0M	switch (us_internal_poll_type(p)) {
227	26.3M	case POLL_TYPE_CALLBACK: {
228	26.3M	struct us_internal_callback_t cb = (struct us_internal_callback_t ) p;
229		/* Timers, asyncs should accept (read), while UDP sockets should obviously not */
230	26.3M	if (!cb->leave_poll_ready) {
231		/* Let's just have this macro to silence the CodeQL alert regarding empty function when using libuv */
232	26.3M	#ifndef LIBUS_USE_LIBUV
233	26.3M	us_internal_accept_poll_event(p);
234	26.3M	#endif
235	26.3M	}
236	26.3M	cb->cb(cb->cb_expects_the_loop ? (struct us_internal_callback_t ) cb->loop : (struct us_internal_callback_t ) &cb->p);
237	26.3M	}
238	26.3M	break;
239	8.54M	case POLL_TYPE_SEMI_SOCKET: {
240		/* Both connect and listen sockets are semi-sockets
241		* but they poll for different events */
242	8.54M	if (us_poll_events(p) == LIBUS_SOCKET_WRITABLE) {
243	1.84k	struct us_socket_t s = (struct us_socket_t ) p;
244
245		/* It is perfectly possible to come here with an error */
246	1.84k	if (error) {
247		/* Emit error, close without emitting on_close */
248	1.60k	s->context->on_connect_error(s, 0);
249	1.60k	us_socket_close_connecting(0, s);
250	1.60k	} else {
251		/* All sockets poll for readable */
252	234	us_poll_change(p, s->context->loop, LIBUS_SOCKET_READABLE);
253
254		/* We always use nodelay */
255	234	bsd_socket_nodelay(us_poll_fd(p), 1);
256
257		/* We are now a proper socket */
258	234	us_internal_poll_set_type(p, POLL_TYPE_SOCKET);
259
260		/* If we used a connection timeout we have to reset it here */
261	234	us_socket_timeout(0, s, 0);
262
263	234	s->context->on_open(s, 1, 0, 0);
264	234	}
265	8.54M	} else {
266	8.54M	struct us_listen_socket_t listen_socket = (struct us_listen_socket_t ) p;
267	8.54M	struct bsd_addr_t addr;
268
269	8.54M	LIBUS_SOCKET_DESCRIPTOR client_fd = bsd_accept_socket(us_poll_fd(p), &addr);
270	8.54M	if (client_fd == LIBUS_SOCKET_ERROR) {
271		/* Todo: start timer here */
272
273	7.71M	} else {
274
275		/* Todo: stop timer if any */
276
277	5.17M	do {
278	5.17M	struct us_socket_context_t *context = us_socket_context(0, &listen_socket->s);
279		/* See if we want to export the FD or keep it here (this event can be unset) */
280	5.17M	if (context->on_pre_open == 0 \|\| context->on_pre_open(context, client_fd) == client_fd) {
281
282		/* Adopt the newly accepted socket */
283	5.17M	us_adopt_accepted_socket(0, context,
284	5.17M	client_fd, listen_socket->socket_ext_size, bsd_addr_get_ip(&addr), bsd_addr_get_ip_length(&addr));
285
286		/* Exit accept loop if listen socket was closed in on_open handler */
287	5.17M	if (us_socket_is_closed(0, &listen_socket->s)) {
288	0	break;
289	0	}
290
291	5.17M	}
292
293	5.17M	} while ((client_fd = bsd_accept_socket(us_poll_fd(p), &addr)) != LIBUS_SOCKET_ERROR);
294	827k	}
295	8.54M	}
296	8.54M	}
297	8.54M	break;
298	4.61k	case POLL_TYPE_SOCKET_SHUT_DOWN:
299	7.15M	case POLL_TYPE_SOCKET: {
300		/* We should only use s, no p after this point */
301	7.15M	struct us_socket_t s = (struct us_socket_t ) p;
302
303		/* Such as epollerr epollhup */
304	7.15M	if (error) {
305		/* Todo: decide what code we give here */
306	3.35M	s = us_socket_close(0, s, 0, NULL);
307	3.35M	return;
308	3.35M	}
309
310	3.80M	if (events & LIBUS_SOCKET_WRITABLE) {
311		/* Note: if we failed a write as a socket of one loop then adopted
312		* to another loop, this will be wrong. Absurd case though */
313	116k	s->context->loop->data.last_write_failed = 0;
314
315	116k	s = s->context->on_writable(s);
316
317	116k	if (us_socket_is_closed(0, s)) {
318	1.53k	return;
319	1.53k	}
320
321		/* If we have no failed write or if we shut down, then stop polling for more writable */
322	115k	if (!s->context->loop->data.last_write_failed \|\| us_socket_is_shut_down(0, s)) {
323	9.91k	us_poll_change(&s->p, us_socket_context(0, s)->loop, us_poll_events(&s->p) & LIBUS_SOCKET_READABLE);
324	9.91k	}
325	115k	}
326
327	3.80M	if (events & LIBUS_SOCKET_READABLE) {
328		/* Contexts may prioritize down sockets that are currently readable, e.g. when SSL handshake has to be done.
329		* SSL handshakes are CPU intensive, so we limit the number of handshakes per loop iteration, and move the rest
330		* to the low-priority queue */
331	3.73M	if (s->context->is_low_prio(s)) {
332	0	if (s->low_prio_state == 2) {
333	0	s->low_prio_state = 0; /* Socket has been delayed and now it's time to process incoming data for one iteration */
334	0	} else if (s->context->loop->data.low_prio_budget > 0) {
335	0	s->context->loop->data.low_prio_budget--; /* Still having budget for this iteration - do normal processing */
336	0	} else {
337	0	us_poll_change(&s->p, us_socket_context(0, s)->loop, us_poll_events(&s->p) & LIBUS_SOCKET_WRITABLE);
338	0	us_internal_socket_context_unlink_socket(s->context, s);
339
340		/* Link this socket to the low-priority queue - we use a LIFO queue, to prioritize newer clients that are
341		* maybe not already timeouted - sounds unfair, but works better in real-life with smaller client-timeouts
342		* under high load */
343	0	s->prev = 0;
344	0	s->next = s->context->loop->data.low_prio_head;
345	0	if (s->next) s->next->prev = s;
346	0	s->context->loop->data.low_prio_head = s;
347
348	0	s->low_prio_state = 1;
349
350	0	break;
351	0	}
352	0	}
353
354	3.73M	int length;
355	3.73M	read_more:
356	3.73M	length = bsd_recv(us_poll_fd(&s->p), s->context->loop->data.recv_buf + LIBUS_RECV_BUFFER_PADDING, LIBUS_RECV_BUFFER_LENGTH, 0);
357	3.73M	if (length > 0) {
358	3.54M	s = s->context->on_data(s, s->context->loop->data.recv_buf + LIBUS_RECV_BUFFER_PADDING, length);
359
360		/* If we filled the entire recv buffer, we need to immediately read again since otherwise a
361		* pending hangup event in the same even loop iteration can close the socket before we get
362		* the chance to read again next iteration */
363	3.54M	if (length == LIBUS_RECV_BUFFER_LENGTH && s && !us_socket_is_closed(0, s)) {
364	0	goto read_more;
365	0	}
366
367	3.54M	} else if (!length) {
368	167k	if (us_socket_is_shut_down(0, s)) {
369		/* We got FIN back after sending it */
370		/* Todo: We should give "CLEAN SHUTDOWN" as reason here */
371	977	s = us_socket_close(0, s, 0, NULL);
372	166k	} else {
373		/* We got FIN, so stop polling for readable */
374	166k	us_poll_change(&s->p, us_socket_context(0, s)->loop, us_poll_events(&s->p) & LIBUS_SOCKET_WRITABLE);
375	166k	s = s->context->on_end(s);
376	166k	}
377	167k	} else if (length == LIBUS_SOCKET_ERROR && !bsd_would_block()) {
378		/* Todo: decide also here what kind of reason we should give */
379	0	s = us_socket_close(0, s, 0, NULL);
380	0	}
381	3.73M	}
382	3.80M	}
383	3.80M	break;
384	42.0M	}
385	42.0M	}
386
387		/* Integration only requires the timer to be set up */
388	22.7k	void us_loop_integrate(struct us_loop_t *loop) {
389	22.7k	us_timer_set(loop->data.sweep_timer, (void ()(struct us_timer_t )) sweep_timer_cb, LIBUS_TIMEOUT_GRANULARITY * 1000, LIBUS_TIMEOUT_GRANULARITY * 1000);
390	22.7k	}
391
392	55.3M	void us_loop_ext(struct us_loop_t loop) {
393	55.3M	return loop + 1;
394	55.3M	}
395
396		#endif