in packages/bun-usockets/src/loop.c [277:499]
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) {
us_internal_socket_after_open((struct us_socket_t *) p, error);
} 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_poll_t *accepted_p = us_create_poll(us_socket_context(0, &listen_socket->s)->loop, 0, sizeof(struct us_socket_t) - sizeof(struct us_poll_t) + listen_socket->socket_ext_size);
us_poll_init(accepted_p, client_fd, POLL_TYPE_SOCKET);
us_poll_start(accepted_p, listen_socket->s.context->loop, LIBUS_SOCKET_READABLE);
struct us_socket_t *s = (struct us_socket_t *) accepted_p;
s->context = listen_socket->s.context;
s->connect_state = NULL;
s->timeout = 255;
s->long_timeout = 255;
s->low_prio_state = 0;
/* We always use nodelay */
bsd_socket_nodelay(client_fd, 1);
us_internal_socket_context_link_socket(listen_socket->s.context, s);
listen_socket->s.context->on_open(s, 0, 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;
if (events & LIBUS_SOCKET_WRITABLE && !error) {
/* 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 (!s || 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_socket_context_ref(0, s->context);
us_internal_socket_context_unlink_socket(0, 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;
}
}
size_t repeat_recv_count = 0;
do {
const struct us_loop_t* loop = s->context->loop;
#ifdef _WIN32
const int recv_flags = MSG_PUSH_IMMEDIATE;
#else
const int recv_flags = MSG_DONTWAIT | MSG_NOSIGNAL;
#endif
int length = bsd_recv(us_poll_fd(&s->p), loop->data.recv_buf + LIBUS_RECV_BUFFER_PADDING, LIBUS_RECV_BUFFER_LENGTH, recv_flags);
if (length > 0) {
s = s->context->on_data(s, loop->data.recv_buf + LIBUS_RECV_BUFFER_PADDING, length);
// loop->num_ready_polls isn't accessible on Windows.
#ifndef WIN32
// rare case: we're reading a lot of data, there's more to be read, and either:
// - the socket has hung up, so we will never get more data from it (only applies to macOS, as macOS will send the event the same tick but Linux will not.)
// - the event loop isn't very busy, so we can read multiple times in a row
#define LOOP_ISNT_VERY_BUSY_THRESHOLD 25
if (
s && length >= (LIBUS_RECV_BUFFER_LENGTH - 24 * 1024) && length <= LIBUS_RECV_BUFFER_LENGTH &&
(error || loop->num_ready_polls < LOOP_ISNT_VERY_BUSY_THRESHOLD) &&
!us_socket_is_closed(0, s)
) {
repeat_recv_count += error == 0;
// When not hung up, read a maximum of 10 times to avoid starving other sockets
// We don't bother with ioctl(FIONREAD) because we've set MSG_DONTWAIT
if (!(repeat_recv_count > 10 && loop->num_ready_polls > 2)) {
continue;
}
}
#undef LOOP_ISNT_VERY_BUSY_THRESHOLD
#endif
} 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, LIBUS_SOCKET_CLOSE_CODE_CLEAN_SHUTDOWN, NULL);
return;
} 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, LIBUS_SOCKET_CLOSE_CODE_CLEAN_SHUTDOWN, NULL);
return;
}
break;
} while (s);
}
/* Such as epollerr epollhup */
if (error && s) {
/* Todo: decide what code we give here */
s = us_socket_close(0, s, error, NULL);
return;
}
break;
}
case POLL_TYPE_UDP: {
struct us_udp_socket_t *u = (struct us_udp_socket_t *) p;
if (u->closed) {
break;
}
if (events & LIBUS_SOCKET_READABLE) {
do {
struct udp_recvbuf recvbuf;
bsd_udp_setup_recvbuf(&recvbuf, u->loop->data.recv_buf, LIBUS_RECV_BUFFER_LENGTH);
int npackets = bsd_recvmmsg(us_poll_fd(p), &recvbuf, MSG_DONTWAIT);
if (npackets > 0) {
u->on_data(u, &recvbuf, npackets);
} else {
if (npackets == LIBUS_SOCKET_ERROR) {
// If the error was not EAGAIN, mark the error
if (!bsd_would_block()) {
error = 1;
}
} else {
// 0 messages received, we are done
// this case can happen if either:
// - the total number of messages pending was not divisible by 8
// - recvmsg() was used instead of recvmmsg() and there was no message to read.
}
break;
}
} while (!u->closed);
}
if (events & LIBUS_SOCKET_WRITABLE && !error && !u->closed) {
u->on_drain(u);
if (u->closed) {
break;
}
// We only poll for writable after a read has failed, and only send one drain notification.
// Otherwise we would receive a writable event on every tick of the event loop.
us_poll_change(&u->p, u->loop, us_poll_events(&u->p) & LIBUS_SOCKET_READABLE);
}
if (error && !u->closed) {
us_udp_socket_close(u);
}
break;
}
}
}