From bc899642924fa7685956c39bda0b1088627546a5 Mon Sep 17 00:00:00 2001 From: harmon25 Date: Tue, 30 Jun 2026 00:10:56 -0400 Subject: [PATCH 1/7] Fix socket send reporting backpressure as closed Transient send backpressure (lwIP ERR_MEM, BSD EAGAIN/EWOULDBLOCK) fell through to the generic error branch of nif_socket_send_internal and was reported as {error, closed}, so callers could not tell a full send buffer from a peer that went away. Conversely, an actually closed connection was reported as {ok, Data}, as if nothing had been consumed. Report backpressure as {error, eagain} and a closed connection as {error, closed}. Signed-off-by: harmon25 --- src/libAtomVM/otp_socket.c | 46 +++++++++++++++++++++++++------------- 1 file changed, 30 insertions(+), 16 deletions(-) diff --git a/src/libAtomVM/otp_socket.c b/src/libAtomVM/otp_socket.c index a561e0392d..4f7ecc0a59 100644 --- a/src/libAtomVM/otp_socket.c +++ b/src/libAtomVM/otp_socket.c @@ -2491,21 +2491,33 @@ static ssize_t do_socket_send(struct SocketResource *rsrc_obj, const uint8_t *bu pbuf_free(p); } else { // If the socket is connection-mode, dest_addr shall be ignored. - // Because we are copying, we cannot really send tcp_sndbuf(rsrc_obj->tcp_pcb) at once + // Because we are copying, we cannot really send tcp_sndbuf(rsrc_obj->tcp_pcb) at once. + // tcp_write fails with ERR_MEM when the data exceeds the send buffer, so + // cap the write at the available space, consistently with + // lwip_tcp_can_send: a partial write is returned as {ok, Rest}. size_t buflen = MIN(len, TCP_MSS); + if (len > 0) { + size_t sndbuf = tcp_sndbuf(rsrc_obj->tcp_pcb); + if (sndbuf == 0) { + LWIP_END(); + return SocketWouldBlock; + } + buflen = MIN(buflen, sndbuf); + } int flags = TCP_WRITE_FLAG_COPY; if (buflen < len) { flags |= TCP_WRITE_FLAG_MORE; } err = tcp_write(rsrc_obj->tcp_pcb, buf, buflen, flags); if (err == ERR_MEM) { - sent_data = 0; - } else { + // Send buffer or queue full: transient, the caller can retry. + LWIP_END(); + return SocketWouldBlock; + } + if (err == ERR_OK) { + err = tcp_output(rsrc_obj->tcp_pcb); if (err == ERR_OK) { - err = tcp_output(rsrc_obj->tcp_pcb); - if (err == ERR_OK) { - sent_data = buflen; - } + sent_data = buflen; } } } @@ -2513,9 +2525,6 @@ static ssize_t do_socket_send(struct SocketResource *rsrc_obj, const uint8_t *bu if (err == ERR_CLSD) { return SocketClosed; } - if (sent_data == 0) { - return SocketWouldBlock; - } if (err != ERR_OK) { return SocketOtherError; } @@ -2577,6 +2586,13 @@ static term nif_socket_send_internal(Context *ctx, int argc, term argv[], bool i // {ok, RestData} | {error, Reason} + // Transient send backpressure (lwIP ERR_MEM / BSD EAGAIN|EWOULDBLOCK) is + // reported as {error, eagain} so callers can retry rather than mistaking it + // for a closed connection. + if (sent_data == SocketWouldBlock) { + return make_error_tuple(posix_errno_to_term(EAGAIN, global), ctx); + } + size_t rest_len = len - sent_data; if (rest_len == 0) { return OK_ATOM; @@ -2592,12 +2608,10 @@ static term nif_socket_send_internal(Context *ctx, int argc, term argv[], bool i return port_create_tuple2(ctx, OK_ATOM, rest); } else if (sent_data == 0) { - if (UNLIKELY(memory_ensure_free_with_roots(ctx, TUPLE_SIZE(2), 1, &data, MEMORY_CAN_SHRINK) != MEMORY_GC_OK)) { - AVM_LOGW(TAG, "Failed to allocate memory: %s:%i.", __FILE__, __LINE__); - RAISE_ERROR(OUT_OF_MEMORY_ATOM); - } - - return port_create_tuple2(ctx, OK_ATOM, data); + // do_socket_send only returns 0 for a closed connection (SocketClosed: + // lwIP ERR_CLSD, BSD EBADF|ECONNRESET); an empty payload already + // returned ok above via the rest_len == 0 check. Report it as such. + return make_error_tuple(CLOSED_ATOM, ctx); } else { TRACE("Unable to send data: res=%zi.\n", sent_data); return make_error_tuple(CLOSED_ATOM, ctx); From 5ed64920d08085370c8aeddd45af4fec947b40b0 Mon Sep 17 00:00:00 2001 From: harmon25 Date: Sat, 4 Jul 2026 17:04:49 -0400 Subject: [PATCH 2/7] Add enif_select_write and ERL_NIF_SELECT_CANCEL enif_select_write was declared in erl_nif.h but never implemented, and ERL_NIF_SELECT_CANCEL was documented as unimplemented. A select event used to hold a single ref, message and pid shared by the read and the write direction, so selecting one direction silently overwrote the other. Keep them independent, so a resource can have a pending read select and a pending write select at the same time, each with its own owner. Implement ERL_NIF_SELECT_CANCEL (with the *_CANCELLED result flags) to cancel a single direction while keeping the event and the other direction armed. Also fix the ref ticks computation, that tested ref against UNDEFINED_ATOM instead of checking it is a local reference. Signed-off-by: harmon25 --- src/libAtomVM/erl_nif.h | 19 ++- src/libAtomVM/resources.c | 140 +++++++++++---- src/libAtomVM/resources.h | 9 +- tests/test-enif.c | 347 +++++++++++++++++++++++++++++++++++++- 4 files changed, 466 insertions(+), 49 deletions(-) diff --git a/src/libAtomVM/erl_nif.h b/src/libAtomVM/erl_nif.h index 73ab4e9e2b..90c90d8ce6 100644 --- a/src/libAtomVM/erl_nif.h +++ b/src/libAtomVM/erl_nif.h @@ -106,26 +106,26 @@ typedef enum /** * @brief enif_select mode flags - * @details ERL_NIF_SELECT_CANCEL which was introduced with OTP-22, is unimplemented. + * @details `ERL_NIF_SELECT_CANCEL` must be combined with `ERL_NIF_SELECT_READ` + * and/or `ERL_NIF_SELECT_WRITE`, and cannot be combined with `ERL_NIF_SELECT_STOP`. */ enum ErlNifSelectFlags { ERL_NIF_SELECT_READ = 1, ERL_NIF_SELECT_WRITE = 2, ERL_NIF_SELECT_STOP = 4, - // ERL_NIF_SELECT_CANCEL = 8, + ERL_NIF_SELECT_CANCEL = 8, }; /** * @brief enif_select result flags - * @details ERL_NIF_SELECT_CANCEL which was introduced with OTP-22, is unimplemented. */ enum { ERL_NIF_SELECT_STOP_CALLED = 1, ERL_NIF_SELECT_STOP_SCHEDULED = 2, - // ERL_NIF_SELECT_READ_CANCELLED = 4, - // ERL_NIF_SELECT_WRITE_CANCELLED = 8, + ERL_NIF_SELECT_READ_CANCELLED = 4, + ERL_NIF_SELECT_WRITE_CANCELLED = 8, ERL_NIF_SELECT_INVALID_EVENT = -1, ERL_NIF_SELECT_FAILED = -2, @@ -257,12 +257,15 @@ ERL_NIF_TERM enif_make_resource_binary(ErlNifEnv *env, void *obj, const void *da * @param env current environment * @param event event object (typically a file descriptor) * @param mode select mode (`ERL_NIF_SELECT_READ` and/or `ERL_NIF_SELECT_WRITE`) - * optionally with `ERL_NIF_SELECT_CANCEL` to cancel, or `ERL_NIF_SELECT_STOP` - * to stop. + * optionally with `ERL_NIF_SELECT_CANCEL` to cancel a pending select of the + * given direction(s) while keeping the event, or `ERL_NIF_SELECT_STOP` to stop. * @param obj resource object working as a container of the event object. * @param pid process id to send a message to or NULL to use the current process (from `env`) * @param ref reference object used in sent messages or `undefined` atom. - * @return a negative value on failure, 0 or flags on success. + * @return a negative value on failure, 0 or flags on success. With + * `ERL_NIF_SELECT_CANCEL`, `ERL_NIF_SELECT_READ_CANCELLED` and/or + * `ERL_NIF_SELECT_WRITE_CANCELLED` are set for each pending select that was + * cancelled. */ int enif_select(ErlNifEnv *env, ErlNifEvent event, enum ErlNifSelectFlags mode, void *obj, const ErlNifPid *pid, ERL_NIF_TERM ref); diff --git a/src/libAtomVM/resources.c b/src/libAtomVM/resources.c index 87ca0b5eaa..cf2e90d824 100644 --- a/src/libAtomVM/resources.c +++ b/src/libAtomVM/resources.c @@ -169,16 +169,17 @@ static int enif_select_common(ErlNifEnv *env, ErlNifEvent event, enum ErlNifSele resource->resource_type->stop(env, obj, event, true); } refc_binary_decrement_refcount(resource, global); - enif_select_event_message_dispose(select_event->message, global, false); + enif_select_event_message_dispose(select_event->read_message, global, false); + enif_select_event_message_dispose(select_event->write_message, global, false); free((void *) select_event); return ERL_NIF_SELECT_STOP_CALLED; } // We cannot call stop now because scheduler loop unlocks after building // the select set but before calling select (or equivalent) // So instead we flag the event. - select_event->close = 1; - select_event->read = 0; - select_event->write = 0; + select_event->close = true; + select_event->read = false; + select_event->write = false; synclist_unlock(&global->select_events); // Platform loop should check close flag after unregister is called if (was_read) { @@ -189,6 +190,42 @@ static int enif_select_common(ErlNifEnv *env, ErlNifEvent event, enum ErlNifSele } return ERL_NIF_SELECT_STOP_SCHEDULED; } + if (mode & ERL_NIF_SELECT_CANCEL) { + // Cancel only the requested directions that are pending. The event + // (and the resource refcount it holds) is kept, so the other direction + // stays armed and the caller can still select again or stop later. + int result = 0; + bool cancel_read = false; + bool cancel_write = false; + if (select_event != NULL) { + if ((mode & ERL_NIF_SELECT_READ) && select_event->read) { + select_event->read = false; + enif_select_event_message_dispose(select_event->read_message, global, false); + select_event->read_message = NULL; + select_event->read_ref_ticks = 0; + select_event->read_local_pid = INVALID_PROCESS_ID; + cancel_read = true; + result |= ERL_NIF_SELECT_READ_CANCELLED; + } + if ((mode & ERL_NIF_SELECT_WRITE) && select_event->write) { + select_event->write = false; + enif_select_event_message_dispose(select_event->write_message, global, false); + select_event->write_message = NULL; + select_event->write_ref_ticks = 0; + select_event->write_local_pid = INVALID_PROCESS_ID; + cancel_write = true; + result |= ERL_NIF_SELECT_WRITE_CANCELLED; + } + } + synclist_unlock(&global->select_events); + if (cancel_read) { + sys_unregister_select_event(global, event, false); + } + if (cancel_write) { + sys_unregister_select_event(global, event, true); + } + return result; + } // Create new event if it doesn't exist. if (select_event == NULL) { select_event = malloc(sizeof(struct SelectEvent)); @@ -197,34 +234,45 @@ static int enif_select_common(ErlNifEnv *env, ErlNifEvent event, enum ErlNifSele } select_event->event = event; select_event->resource = resource; - select_event->message = NULL; - select_event->ref_ticks = 0; + select_event->read = false; + select_event->write = false; + select_event->close = false; + select_event->read_message = NULL; + select_event->write_message = NULL; + select_event->read_ref_ticks = 0; + select_event->write_ref_ticks = 0; + select_event->read_local_pid = INVALID_PROCESS_ID; + select_event->write_local_pid = INVALID_PROCESS_ID; // Resource is used in select_event, so we increase refcount. refc_binary_increment_refcount(resource); list_init(&select_event->head); list_append(select_events, &select_event->head); } - // Second read or second write overwrite ref/message & pid. - enif_select_event_message_dispose(select_event->message, global, false); - select_event->message = message; - if (message) { - select_event->ref_ticks = 0; - } else { - if (ref == UNDEFINED_ATOM) { - select_event->ref_ticks = 0; - } else { - select_event->ref_ticks = term_to_ref_ticks(ref); - } - } - select_event->local_pid = *pid; - select_event->read = mode & ERL_NIF_SELECT_READ; - select_event->write = mode & ERL_NIF_SELECT_WRITE; - select_event->close = 0; + // Second read or second write overwrites only that direction, preserving + // the other direction when both read and write are selected on the same + // event (e.g. a socket with a pending recv select and a pending send + // select at the same time). + uint64_t ref_ticks = (!message && term_is_local_reference(ref)) ? term_to_ref_ticks(ref) : 0; + if (mode & ERL_NIF_SELECT_READ) { + enif_select_event_message_dispose(select_event->read_message, global, false); + select_event->read_message = message; + select_event->read_ref_ticks = message ? 0 : ref_ticks; + select_event->read_local_pid = *pid; + select_event->read = true; + } + if (mode & ERL_NIF_SELECT_WRITE) { + enif_select_event_message_dispose(select_event->write_message, global, false); + select_event->write_message = message; + select_event->write_ref_ticks = message ? 0 : ref_ticks; + select_event->write_local_pid = *pid; + select_event->write = true; + } + select_event->close = false; synclist_unlock(&global->select_events); - if (select_event->read) { + if (mode & ERL_NIF_SELECT_READ) { sys_register_select_event(global, event, false); } - if (select_event->write) { + if (mode & ERL_NIF_SELECT_WRITE) { sys_register_select_event(global, event, true); } return 0; @@ -235,6 +283,13 @@ int enif_select(ErlNifEnv *env, ErlNifEvent event, enum ErlNifSelectFlags mode, if (!(mode & (ERL_NIF_SELECT_STOP | ERL_NIF_SELECT_READ | ERL_NIF_SELECT_WRITE))) { return ERL_NIF_SELECT_BADARG; } + if (mode & ERL_NIF_SELECT_CANCEL) { + // Cancel applies to read and/or write, and cannot be combined with stop. + if ((mode & ERL_NIF_SELECT_STOP) || !(mode & (ERL_NIF_SELECT_READ | ERL_NIF_SELECT_WRITE))) { + return ERL_NIF_SELECT_BADARG; + } + return enif_select_common(env, event, mode, obj, pid, ref, NULL); + } if (UNLIKELY(mode & (ERL_NIF_SELECT_READ | ERL_NIF_SELECT_WRITE) && !term_is_local_reference(ref) && ref != UNDEFINED_ATOM)) { return ERL_NIF_SELECT_BADARG; } @@ -252,7 +307,17 @@ int enif_select_read(ErlNifEnv *env, ErlNifEvent event, void *obj, const ErlNifP } Message *message = mailbox_message_create_normal_message_from_term(msg); enum ErlNifSelectFlags mode = ERL_NIF_SELECT_READ; - return enif_select_common(env, event, mode, obj, pid, term_nil(), message); + return enif_select_common(env, event, mode, obj, pid, UNDEFINED_ATOM, message); +} + +int enif_select_write(ErlNifEnv *env, ErlNifEvent event, void *obj, const ErlNifPid *pid, ERL_NIF_TERM msg, ErlNifEnv *msg_env) +{ + if (UNLIKELY(msg_env != NULL)) { + return ERL_NIF_SELECT_BADARG; + } + Message *message = mailbox_message_create_normal_message_from_term(msg); + enum ErlNifSelectFlags mode = ERL_NIF_SELECT_WRITE; + return enif_select_common(env, event, mode, obj, pid, UNDEFINED_ATOM, message); } term select_event_make_notification(void *rsrc_obj, uint64_t ref_ticks, bool is_write, Heap *heap) @@ -273,32 +338,36 @@ term select_event_make_notification(void *rsrc_obj, uint64_t ref_ticks, bool is_ static void select_event_send_notification(struct SelectEvent *select_event, bool is_write, GlobalContext *global) { - if (select_event->message) { + Message **message = is_write ? &select_event->write_message : &select_event->read_message; + int32_t local_pid = is_write ? select_event->write_local_pid : select_event->read_local_pid; + uint64_t ref_ticks = is_write ? select_event->write_ref_ticks : select_event->read_ref_ticks; + + if (*message) { enum SendMessageResult result; #ifdef AVM_SELECT_IN_TASK - result = globalcontext_post_message_from_task(global, select_event->local_pid, select_event->message); + result = globalcontext_post_message_from_task(global, local_pid, *message); #else - result = globalcontext_post_message(global, select_event->local_pid, select_event->message); + result = globalcontext_post_message(global, local_pid, *message); #endif if (result == SEND_MESSAGE_OK) { // Ownership was properly transfered. // Otherwise, it will be destroyed when we have a context (when enif_select is called with stop for example) - select_event->message = NULL; + *message = NULL; } } else { BEGIN_WITH_STACK_HEAP(SELECT_EVENT_NOTIFICATION_SIZE, heap) - term notification = select_event_make_notification(select_event->resource->data, select_event->ref_ticks, is_write, &heap); + term notification = select_event_make_notification(select_event->resource->data, ref_ticks, is_write, &heap); #ifdef AVM_SELECT_IN_TASK - globalcontext_send_message_from_task(global, select_event->local_pid, NormalMessage, notification); + globalcontext_send_message_from_task(global, local_pid, NormalMessage, notification); #else - globalcontext_send_message(global, select_event->local_pid, notification); + globalcontext_send_message(global, local_pid, notification); #endif END_WITH_STACK_HEAP(heap, global) } if (is_write) { - select_event->write = 0; + select_event->write = false; } else { - select_event->read = 0; + select_event->read = false; } sys_unregister_select_event(global, select_event->event, is_write); } @@ -342,7 +411,8 @@ static inline void select_event_destroy(struct SelectEvent *select_event, Global #else refc_binary_decrement_refcount(select_event->resource, global); #endif - enif_select_event_message_dispose(select_event->message, global, true); + enif_select_event_message_dispose(select_event->read_message, global, true); + enif_select_event_message_dispose(select_event->write_message, global, true); free((void *) select_event); } diff --git a/src/libAtomVM/resources.h b/src/libAtomVM/resources.h index 58c97cf9b6..654eaa8acc 100644 --- a/src/libAtomVM/resources.h +++ b/src/libAtomVM/resources.h @@ -92,9 +92,12 @@ struct SelectEvent bool read; bool write; bool close; - int32_t local_pid; - uint64_t ref_ticks; - Message *message; + int32_t read_local_pid; + int32_t write_local_pid; + uint64_t read_ref_ticks; + uint64_t write_ref_ticks; + Message *read_message; + Message *write_message; }; /** diff --git a/tests/test-enif.c b/tests/test-enif.c index 200b1d4534..1447e2a8c7 100644 --- a/tests/test-enif.c +++ b/tests/test-enif.c @@ -39,6 +39,15 @@ #include "sys.h" #include "utils.h" +// Evaluate an expression with side effects outside of assert(), so that it is +// still evaluated when assertions are disabled (NDEBUG). +#define VERIFY(expr) \ + do { \ + bool verify_ok = (expr); \ + assert(verify_ok); \ + UNUSED(verify_ok); \ + } while (0) + static uint32_t cb_read_resource = 0; static int32_t down_pid = 0; static ErlNifMonitor down_mon = { NULL, 0 }; @@ -761,9 +770,12 @@ void test_select_event_armed_during_listener_only_rebuild(void) select_event->read = true; select_event->write = false; select_event->close = false; - select_event->local_pid = ctx->process_id; - select_event->ref_ticks = 0; - select_event->message = NULL; + select_event->read_local_pid = ctx->process_id; + select_event->write_local_pid = INVALID_PROCESS_ID; + select_event->read_ref_ticks = 0; + select_event->write_ref_ticks = 0; + select_event->read_message = NULL; + select_event->write_message = NULL; refc_binary_increment_refcount(select_event->resource); list_init(&select_event->head); struct ListHead *select_events = synclist_wrlock(&glb->select_events); @@ -796,6 +808,332 @@ void test_select_event_armed_during_listener_only_rebuild(void) } #endif +// enif_select_read/enif_select_write should track independent messages, refs +// and target pids for the same event, so a resource can have a pending read +// select and a pending write select at the same time (e.g. a socket with a +// permanent recv select and a transient send-backpressure select). +void test_resource_select_read_write_independent_messages(void) +{ + GlobalContext *glb = globalcontext_new(); + Context *read_ctx = context_new(glb); + Context *write_ctx = context_new(glb); + ErlNifEnv *env = erl_nif_env_from_context(read_ctx); + + ErlNifResourceTypeInit init; + init.members = 1; + init.dtor = resource_dtor; + ErlNifResourceFlags flags; + cb_read_resource = 0; + dtor_call_count = 0; + + ErlNifResourceType *resource_type = enif_init_resource_type(env, "test_select_resource", &init, ERL_NIF_RT_CREATE, &flags); + assert(resource_type != NULL); + + void *ptr = enif_alloc_resource(resource_type, sizeof(uint32_t)); + uint32_t *resource = (uint32_t *) ptr; + *resource = 99; + + int pipefds[2]; + VERIFY(pipe(pipefds) == 0); + ErlNifEvent event = (ErlNifEvent) pipefds[0]; + + ErlNifPid read_pid = read_ctx->process_id; + ErlNifPid write_pid = write_ctx->process_id; + + term read_msg = term_from_int(11); + term write_msg = term_from_int(22); + + int r = enif_select_read(env, event, ptr, &read_pid, read_msg, NULL); + assert(r == 0); + r = enif_select_write(env, event, ptr, &write_pid, write_msg, NULL); + assert(r == 0); + + // Firing both read and write readiness must deliver the read message to + // read_ctx and the write message to write_ctx, without either clobbering + // the other (this is the bug being tested: a shared message/ref/pid field + // would either lose one notification or send it to the wrong process). + bool notified = select_event_notify(event, true, true, glb); + assert(notified); + + VERIFY(mailbox_process_outer_list(read_ctx) == NULL); + assert(mailbox_has_next(&read_ctx->mailbox)); + term read_received; + VERIFY(mailbox_peek(read_ctx, &read_received)); + assert(read_received == read_msg); + mailbox_remove_message(&read_ctx->mailbox, &read_ctx->heap); + assert(!mailbox_has_next(&read_ctx->mailbox)); + + VERIFY(mailbox_process_outer_list(write_ctx) == NULL); + assert(mailbox_has_next(&write_ctx->mailbox)); + term write_received; + VERIFY(mailbox_peek(write_ctx, &write_received)); + assert(write_received == write_msg); + mailbox_remove_message(&write_ctx->mailbox, &write_ctx->heap); + assert(!mailbox_has_next(&write_ctx->mailbox)); + + // Once a direction fires, only that direction is consumed; the other + // stays armed until it is separately triggered or explicitly stopped. + // Re-arm both to test the ref-ticks (no custom message) path together. + r = enif_select_read(env, event, ptr, &read_pid, read_msg, NULL); + assert(r == 0); + r = enif_select_write(env, event, ptr, &write_pid, write_msg, NULL); + assert(r == 0); + + // Re-selecting read must not disturb the pending write selection (and + // vice-versa): overwrite only the read side with a new pid/message. + Context *other_read_ctx = context_new(glb); + ErlNifPid other_read_pid = other_read_ctx->process_id; + term other_read_msg = term_from_int(33); + r = enif_select_read(env, event, ptr, &other_read_pid, other_read_msg, NULL); + assert(r == 0); + + notified = select_event_notify(event, true, true, glb); + assert(notified); + + // Original read_ctx should get nothing new (its select was overwritten). + VERIFY(mailbox_process_outer_list(read_ctx) == NULL); + assert(!mailbox_has_next(&read_ctx->mailbox)); + + VERIFY(mailbox_process_outer_list(other_read_ctx) == NULL); + assert(mailbox_has_next(&other_read_ctx->mailbox)); + term other_read_received; + VERIFY(mailbox_peek(other_read_ctx, &other_read_received)); + assert(other_read_received == other_read_msg); + + VERIFY(mailbox_process_outer_list(write_ctx) == NULL); + assert(mailbox_has_next(&write_ctx->mailbox)); + term write_received_again; + VERIFY(mailbox_peek(write_ctx, &write_received_again)); + assert(write_received_again == write_msg); + mailbox_remove_message(&write_ctx->mailbox, &write_ctx->heap); + + // Both directions have now fired and are no longer armed, so stopping the + // select should immediately release the extra refcount it was holding. + int stop_result = enif_select(env, event, ERL_NIF_SELECT_STOP, ptr, NULL, term_nil()); + assert(stop_result == ERL_NIF_SELECT_STOP_CALLED); + + int release_result = enif_release_resource(ptr); + assert(release_result); + + scheduler_terminate(read_ctx); + scheduler_terminate(write_ctx); + scheduler_terminate(other_read_ctx); + + close(pipefds[0]); + close(pipefds[1]); + + globalcontext_destroy(glb); +} + +// The generic enif_select/2 API (no custom message, just a reference) must +// also track read and write refs independently when both directions are +// selected on the same event, e.g. a single ref selecting both directions, +// followed by a read-only re-select that must not clear the write ref. +void test_resource_select_read_write_independent_refs(void) +{ + GlobalContext *glb = globalcontext_new(); + Context *ctx = context_new(glb); + ErlNifEnv *env = erl_nif_env_from_context(ctx); + + ErlNifResourceTypeInit init; + init.members = 1; + init.dtor = resource_dtor; + ErlNifResourceFlags flags; + cb_read_resource = 0; + + ErlNifResourceType *resource_type = enif_init_resource_type(env, "test_select_resource_refs", &init, ERL_NIF_RT_CREATE, &flags); + assert(resource_type != NULL); + + void *ptr = enif_alloc_resource(resource_type, sizeof(uint32_t)); + uint32_t *resource = (uint32_t *) ptr; + *resource = 7; + + int pipefds[2]; + VERIFY(pipe(pipefds) == 0); + ErlNifEvent event = (ErlNifEvent) pipefds[0]; + + ErlNifPid pid = ctx->process_id; + + enum MemoryGCResult ensure_free_result = memory_erl_nif_env_ensure_free(env, TERM_BOXED_REFERENCE_SHORT_SIZE); + assert(ensure_free_result == MEMORY_GC_OK); + UNUSED(ensure_free_result); + term write_ref = term_from_ref_ticks(globalcontext_get_ref_ticks(glb), &env->heap); + + // Select for write with an explicit ref and no custom message. + int r = enif_select(env, event, ERL_NIF_SELECT_WRITE, ptr, &pid, write_ref); + assert(r == 0); + + // Now select read only, with UNDEFINED_ATOM (no ref requested). This must + // not disturb the write ref registered above. + r = enif_select(env, event, ERL_NIF_SELECT_READ, ptr, &pid, UNDEFINED_ATOM); + assert(r == 0); + + bool notified = select_event_notify(event, true, true, glb); + assert(notified); + + // Two notifications should have been queued: {select, Resource, undefined, ready_input} + // and {select, Resource, WriteRef, ready_output}. + VERIFY(mailbox_process_outer_list(ctx) == NULL); + assert(mailbox_has_next(&ctx->mailbox)); + + bool saw_read = false; + bool saw_write = false; + for (int i = 0; i < 2; i++) { + term msg; + VERIFY(mailbox_peek(ctx, &msg)); + assert(term_is_tuple(msg)); + assert(term_get_tuple_arity(msg) == 4); + assert(term_get_tuple_element(msg, 0) == SELECT_ATOM); + term ref_or_undefined = term_get_tuple_element(msg, 2); + term kind = term_get_tuple_element(msg, 3); + if (kind == READY_INPUT_ATOM) { + assert(ref_or_undefined == UNDEFINED_ATOM); + saw_read = true; + } else { + assert(kind == READY_OUTPUT_ATOM); + assert(term_is_reference(ref_or_undefined)); + saw_write = true; + } + if (i == 0) { + assert(mailbox_has_next(&ctx->mailbox)); + mailbox_next(&ctx->mailbox); + } + } + assert(saw_read); + assert(saw_write); + mailbox_remove_message(&ctx->mailbox, &ctx->heap); + mailbox_remove_message(&ctx->mailbox, &ctx->heap); + assert(!mailbox_has_next(&ctx->mailbox)); + + // Both directions have now fired and are no longer armed, so stopping the + // select should immediately release the extra refcount it was holding. + int stop_result = enif_select(env, event, ERL_NIF_SELECT_STOP, ptr, NULL, term_nil()); + assert(stop_result == ERL_NIF_SELECT_STOP_CALLED); + + int release_result = enif_release_resource(ptr); + assert(release_result); + + scheduler_terminate(ctx); + + close(pipefds[0]); + close(pipefds[1]); + +#ifdef AVM_TASK_DRIVER_ENABLED + // Notifications built without a custom message (the ref-only path + // exercised above) release their transient resource reference through + // the task-driver refc queue rather than synchronously; drain it so the + // resource is properly freed instead of merely being reported as a + // (harmless) dangling resource on shutdown. + globalcontext_process_task_driver_queues(glb); +#endif + + globalcontext_destroy(glb); +} + +// ERL_NIF_SELECT_CANCEL must only cancel the requested direction(s), dispose +// their pending messages, and keep the other direction armed. +void test_resource_select_cancel(void) +{ + GlobalContext *glb = globalcontext_new(); + Context *read_ctx = context_new(glb); + Context *write_ctx = context_new(glb); + ErlNifEnv *env = erl_nif_env_from_context(read_ctx); + + ErlNifResourceTypeInit init; + init.members = 1; + init.dtor = resource_dtor; + ErlNifResourceFlags flags; + cb_read_resource = 0; + dtor_call_count = 0; + + ErlNifResourceType *resource_type = enif_init_resource_type(env, "test_select_cancel_resource", &init, ERL_NIF_RT_CREATE, &flags); + assert(resource_type != NULL); + + void *ptr = enif_alloc_resource(resource_type, sizeof(uint32_t)); + + int pipefds[2]; + VERIFY(pipe(pipefds) == 0); + ErlNifEvent event = (ErlNifEvent) pipefds[0]; + + ErlNifPid read_pid = read_ctx->process_id; + ErlNifPid write_pid = write_ctx->process_id; + term read_msg = term_from_int(11); + term write_msg = term_from_int(22); + + // Cancelling an event that was never selected is not an error. + int r = enif_select(env, event, ERL_NIF_SELECT_READ | ERL_NIF_SELECT_CANCEL, ptr, NULL, term_nil()); + assert(r == 0); + + // Invalid combinations. + r = enif_select(env, event, ERL_NIF_SELECT_CANCEL, ptr, NULL, term_nil()); + assert(r == ERL_NIF_SELECT_BADARG); + r = enif_select(env, event, ERL_NIF_SELECT_READ | ERL_NIF_SELECT_STOP | ERL_NIF_SELECT_CANCEL, ptr, NULL, term_nil()); + assert(r == ERL_NIF_SELECT_BADARG); + + VERIFY(enif_select_read(env, event, ptr, &read_pid, read_msg, NULL) == 0); + VERIFY(enif_select_write(env, event, ptr, &write_pid, write_msg, NULL) == 0); + + // Cancel read only: write stays armed. + r = enif_select(env, event, ERL_NIF_SELECT_READ | ERL_NIF_SELECT_CANCEL, ptr, NULL, term_nil()); + assert(r == ERL_NIF_SELECT_READ_CANCELLED); + // Cancelling again has nothing to cancel. + r = enif_select(env, event, ERL_NIF_SELECT_READ | ERL_NIF_SELECT_CANCEL, ptr, NULL, term_nil()); + assert(r == 0); + + VERIFY(select_event_notify(event, true, true, glb)); + VERIFY(mailbox_process_outer_list(read_ctx) == NULL); + assert(!mailbox_has_next(&read_ctx->mailbox)); + VERIFY(mailbox_process_outer_list(write_ctx) == NULL); + term received; + VERIFY(mailbox_peek(write_ctx, &received)); + assert(received == write_msg); + mailbox_remove_message(&write_ctx->mailbox, &write_ctx->heap); + + // Cancel write only: read (armed again) stays armed. + VERIFY(enif_select_read(env, event, ptr, &read_pid, read_msg, NULL) == 0); + VERIFY(enif_select_write(env, event, ptr, &write_pid, write_msg, NULL) == 0); + r = enif_select(env, event, ERL_NIF_SELECT_WRITE | ERL_NIF_SELECT_CANCEL, ptr, NULL, term_nil()); + assert(r == ERL_NIF_SELECT_WRITE_CANCELLED); + + VERIFY(select_event_notify(event, true, true, glb)); + VERIFY(mailbox_process_outer_list(write_ctx) == NULL); + assert(!mailbox_has_next(&write_ctx->mailbox)); + VERIFY(mailbox_process_outer_list(read_ctx) == NULL); + VERIFY(mailbox_peek(read_ctx, &received)); + assert(received == read_msg); + mailbox_remove_message(&read_ctx->mailbox, &read_ctx->heap); + + // Cancel both at once: nothing is delivered afterwards. + VERIFY(enif_select_read(env, event, ptr, &read_pid, read_msg, NULL) == 0); + VERIFY(enif_select_write(env, event, ptr, &write_pid, write_msg, NULL) == 0); + r = enif_select(env, event, ERL_NIF_SELECT_READ | ERL_NIF_SELECT_WRITE | ERL_NIF_SELECT_CANCEL, ptr, NULL, term_nil()); + assert(r == (ERL_NIF_SELECT_READ_CANCELLED | ERL_NIF_SELECT_WRITE_CANCELLED)); + VERIFY(!select_event_notify(event, true, true, glb)); + assert(!mailbox_has_next(&read_ctx->mailbox)); + assert(!mailbox_has_next(&write_ctx->mailbox)); + + // The event is kept after cancelling, so it can be selected again. + VERIFY(enif_select_write(env, event, ptr, &write_pid, write_msg, NULL) == 0); + VERIFY(select_event_notify(event, false, true, glb)); + VERIFY(mailbox_process_outer_list(write_ctx) == NULL); + VERIFY(mailbox_peek(write_ctx, &received)); + assert(received == write_msg); + mailbox_remove_message(&write_ctx->mailbox, &write_ctx->heap); + + // Nothing is armed anymore, so stop is immediate. + VERIFY(enif_select(env, event, ERL_NIF_SELECT_STOP, ptr, NULL, term_nil()) == ERL_NIF_SELECT_STOP_CALLED); + + VERIFY(enif_release_resource(ptr)); + + scheduler_terminate(read_ctx); + scheduler_terminate(write_ctx); + + close(pipefds[0]); + close(pipefds[1]); + + globalcontext_destroy(glb); +} + int main(int argc, char **argv) { UNUSED(argc); @@ -814,6 +1152,9 @@ int main(int argc, char **argv) #ifndef HAVE_KQUEUE test_select_event_armed_during_listener_only_rebuild(); #endif + test_resource_select_read_write_independent_messages(); + test_resource_select_read_write_independent_refs(); + test_resource_select_cancel(); return EXIT_SUCCESS; } From fe3b4fd9060ac2d1c8699a64c1b2fa1d8019b3a9 Mon Sep 17 00:00:00 2001 From: harmon25 Date: Sat, 4 Jul 2026 17:05:17 -0400 Subject: [PATCH 3/7] otp_socket: Add nif_select_write/2 Add a write select to the socket NIFs, so a process can wait until a socket can accept more data instead of polling after {error, eagain}. BSD sockets use enif_select_write. lwIP sockets use the tcp_sent and tcp_poll callbacks, and tcp_sndbuf/tcp_sndqueuelen to tell when the send buffer has room again. A socket has one selecting process and monitor per direction, like resources.c: selecting in one direction replaces the previous owner of that direction only, close and select_stop handle both, and socket_down releases only the direction of the dead process. A reader and a writer can therefore block on the same socket without disturbing each other, and an abort is sent to the owner of each direction. On lwIP the callbacks notify the owner of their direction, and a connect traps its caller on the write slot. The selecting state is initialized by a helper shared by all the ways a socket is created, including accept. Signed-off-by: harmon25 --- libs/estdlib/src/gen_tcp_socket.erl | 14 +- libs/estdlib/src/gen_udp_socket.erl | 4 +- libs/estdlib/src/socket.erl | 15 +- src/libAtomVM/otp_socket.c | 588 ++++++++++++++++++++-------- src/libAtomVM/otp_socket.h | 4 + 5 files changed, 458 insertions(+), 167 deletions(-) diff --git a/libs/estdlib/src/gen_tcp_socket.erl b/libs/estdlib/src/gen_tcp_socket.erl index b17e5059d5..5ef50187a1 100644 --- a/libs/estdlib/src/gen_tcp_socket.erl +++ b/libs/estdlib/src/gen_tcp_socket.erl @@ -379,10 +379,10 @@ handle_info({'$socket', Socket, abort, {Ref, closed}}, State) -> case maps:get(Ref, State#state.pending_selects, undefined) of undefined -> ?LOG_WARNING("Unable to find select ref ~p in pending selects", [Ref]), - socket:nif_select_stop(Socket), + socket:nif_select_read_stop(Socket), {noreply, State}; {accept, From, _AcceptingProc, _Timeout} -> - socket:nif_select_stop(Socket), + socket:nif_select_read_stop(Socket), gen_server:reply(From, {error, closed}), {noreply, State}; active -> @@ -390,7 +390,7 @@ handle_info({'$socket', Socket, abort, {Ref, closed}}, State) -> State#state.controlling_process ! {tcp_closed, WrappedSocket}, {noreply, State}; {passive, From, _Length, _Timeout} -> - socket:nif_select_stop(Socket), + socket:nif_select_read_stop(Socket), gen_server:reply(From, {error, closed}), {noreply, State#state{ pending_selects = maps:remove(Ref, State#state.pending_selects) @@ -465,7 +465,7 @@ handle_accept(State, From, AcceptingProc) -> {error, _Reason} = Error -> ?LOG_ERROR("Failed to accept connection: ~p", [Error]), %% CAUTION: internal API - socket:nif_select_stop(Socket), + socket:nif_select_read_stop(Socket), gen_server:reply(From, Error), State end. @@ -510,7 +510,7 @@ handle_active_recv(State) -> {error, closed} -> ?LOG_INFO("Socket was closed by peer"), %% CAUTION: internal API - socket:nif_select_stop(Socket), + socket:nif_select_read_stop(Socket), %% peer has closed the connection WrappedSocket = {?GEN_TCP_MONIKER, self(), ?MODULE}, State#state.controlling_process ! {tcp_closed, WrappedSocket}, @@ -518,7 +518,7 @@ handle_active_recv(State) -> {error, _} = Error -> ?LOG_ERROR("Error on receive on pending select Error=~p", [Error]), %% CAUTION: internal API - socket:nif_select_stop(Socket), + socket:nif_select_read_stop(Socket), %% TODO is this right? I don't think active receivers would know %% what to do with this message. Maybe log it instead? State#state.controlling_process ! {tcp_error, WrappedSocket, Error}, @@ -547,7 +547,7 @@ handle_passive_recv(State, From, Length, _Timeout) -> State; {error, _} = Error -> ?LOG_ERROR("Error on receive from socket:nif_recv Error=~p", [Error]), - socket:nif_select_stop(Socket), + socket:nif_select_read_stop(Socket), ?LOG_INFO("unable to receive on pending select~n"), gen_server:reply(From, Error), State diff --git a/libs/estdlib/src/gen_udp_socket.erl b/libs/estdlib/src/gen_udp_socket.erl index c3a561e53e..80883f8508 100644 --- a/libs/estdlib/src/gen_udp_socket.erl +++ b/libs/estdlib/src/gen_udp_socket.erl @@ -335,7 +335,7 @@ handle_active_recvfrom(State) -> State; {error, _} = E -> %% CAUTION: internal API - socket:nif_select_stop(Socket), + socket:nif_select_read_stop(Socket), ?LOG_INFO("unable to receive on pending select~n"), State#state.controlling_process ! {udp_error, WrappedSocket, E}, State @@ -353,7 +353,7 @@ handle_passive_recvfrom(State, From, Length, _Timeout) -> gen_server:reply(From, {ok, {Addr, Port, BinaryOrList}}), State; {error, _} = E -> - socket:nif_select_stop(Socket), + socket:nif_select_read_stop(Socket), ?LOG_INFO("unable to receive on pending select~n"), gen_server:reply(From, E), State diff --git a/libs/estdlib/src/socket.erl b/libs/estdlib/src/socket.erl index 810cf563c2..ea212f55d1 100644 --- a/libs/estdlib/src/socket.erl +++ b/libs/estdlib/src/socket.erl @@ -50,6 +50,7 @@ nif_recv/2, nif_recvfrom/2, nif_select_stop/1, + nif_select_read_stop/1, nif_send/2, nif_sendto/3 ]). @@ -269,7 +270,7 @@ accept0(Socket, Timeout) -> {'$socket', Socket, select, Ref} -> case ?MODULE:nif_accept(Socket) of {error, _} = E -> - ?MODULE:nif_select_stop(Socket), + ?MODULE:nif_select_read_stop(Socket), E; {ok, _Socket} = Reply -> Reply @@ -382,7 +383,7 @@ recv0(Socket, Length, Timeout) -> {'$socket', Socket, select, Ref} -> case ?MODULE:nif_recv(Socket, Length) of {error, _} = E -> - ?MODULE:nif_select_stop(Socket), + ?MODULE:nif_select_read_stop(Socket), E; {ok, Data} -> {ok, Data} @@ -433,7 +434,7 @@ recv0_r(Socket, Length, Timeout, EndQuery, Acc) -> {'$socket', Socket, select, Ref} -> case ?MODULE:nif_recv(Socket, Length) of {error, _} = E -> - ?MODULE:nif_select_stop(Socket), + ?MODULE:nif_select_read_stop(Socket), E; {ok, Data} -> NewAcc = [Data | Acc], @@ -537,7 +538,7 @@ recvfrom0(Socket, Length, Timeout) -> {'$socket', Socket, select, Ref} -> case ?MODULE:nif_recvfrom(Socket, Length) of {error, _} = E -> - ?MODULE:nif_select_stop(Socket), + ?MODULE:nif_select_read_stop(Socket), E; {ok, {_Address, _Data}} = Reply -> Reply @@ -724,6 +725,12 @@ nif_recvfrom(_Socket, _Length) -> nif_select_stop(_Socket) -> erlang:nif_error(undefined). +%% @private +%% Like nif_select_stop/1, but if another process is selecting for write on the +%% socket, only the read direction is stopped. +nif_select_read_stop(_Socket) -> + erlang:nif_error(undefined). + %% @private nif_send(_Socket, _Data) -> erlang:nif_error(undefined). diff --git a/src/libAtomVM/otp_socket.c b/src/libAtomVM/otp_socket.c index 4f7ecc0a59..4b2a3b1871 100644 --- a/src/libAtomVM/otp_socket.c +++ b/src/libAtomVM/otp_socket.c @@ -113,6 +113,7 @@ enum SocketState SocketStateSelectingRead = 1 << 3, SocketStateConnected = 1 << 4, SocketStateListening = 1 << 5, + SocketStateSelectingWrite = 1 << 6, // Actual states SocketStateUDPIdle = SocketStateUDP, @@ -120,6 +121,8 @@ enum SocketState SocketStateTCPNew = SocketStateTCP, SocketStateTCPConnected = SocketStateTCP | SocketStateConnected, SocketStateTCPSelectingRead = SocketStateTCPConnected | SocketStateSelectingRead, + SocketStateTCPSelectingWrite = SocketStateTCPConnected | SocketStateSelectingWrite, + SocketStateTCPSelectingReadWrite = SocketStateTCPConnected | SocketStateSelectingRead | SocketStateSelectingWrite, SocketStateTCPListening = SocketStateTCP | SocketStateListening, SocketStateTCPSelectingAccept = SocketStateTCPListening | SocketStateSelectingRead, }; @@ -146,19 +149,30 @@ struct UDPReceivedItem }; static err_t tcp_recv_cb(void *arg, struct tcp_pcb *tpcb, struct pbuf *p, err_t err); +static err_t tcp_sent_cb(void *arg, struct tcp_pcb *tpcb, u16_t len); +static err_t tcp_poll_cb(void *arg, struct tcp_pcb *tpcb); static void udp_recv_cb(void *arg, struct udp_pcb *pcb, struct pbuf *p, const ip_addr_t *addr, u16_t port); static void tcp_err_cb(void *arg, err_t err); #endif +// A process selecting a socket in one direction. Like in resources.c, a socket +// has at most one owner per direction: selecting again in the same direction +// replaces the previous owner, while the other direction is left untouched. +struct SocketSelect +{ + int32_t process_id; + ErlNifMonitor monitor; + uint64_t ref_ticks; +}; + #if OTP_SOCKET_BSD struct SocketResource { int fd; uint64_t socket_ref_ticks; - uint64_t select_ref_ticks; - int32_t selecting_process_id; - ErlNifMonitor selecting_process_monitor; + struct SocketSelect read_select; + struct SocketSelect write_select; size_t buf_size; #ifndef AVM_NO_SMP RWLock *socket_lock; @@ -174,9 +188,9 @@ struct SocketResource struct udp_pcb *udp_pcb; }; uint64_t socket_ref_ticks; - uint64_t select_ref_ticks; - int32_t selecting_process_id; // trapped or selecting - ErlNifMonitor selecting_process_monitor; + struct SocketSelect read_select; + // Also used to trap the process waiting for a connect to complete. + struct SocketSelect write_select; bool linger_on; int linger_sec; size_t pos; @@ -240,6 +254,12 @@ static const AtomStringIntPair otp_socket_setopt_level_table[] = { #define DEFAULT_BUFFER_SIZE 512 +#if OTP_SOCKET_LWIP +// Interval of the lwIP tcp_poll callback, in units of 500 ms (TCP_SLOW_INTERVAL). +// The callback is the fallback wakeup of processes selecting for write. +#define SOCKET_TCP_POLL_INTERVAL 1 +#endif + #ifndef MIN #define MIN(A, B) (((A) < (B)) ? (A) : (B)) #endif @@ -247,12 +267,35 @@ static const AtomStringIntPair otp_socket_setopt_level_table[] = { static ErlNifResourceType *socket_resource_type; #define SOCKET_MAKE_SELECT_NOTIFICATION_SIZE (TUPLE_SIZE(4) + TERM_BOXED_REFERENCE_SHORT_SIZE + TUPLE_SIZE(2) + TERM_BOXED_REFERENCE_SHORT_SIZE + TERM_BOXED_REFERENCE_RESOURCE_SIZE) -static term socket_make_select_notification(struct SocketResource *rsrc_obj, Heap *heap); +static term socket_make_select_notification(struct SocketResource *rsrc_obj, Heap *heap, enum ErlNifSelectFlags direction); // // resource operations // +#if OTP_SOCKET_LWIP +// Detach every callback (and the callback argument) from a TCP pcb that is +// about to be closed or aborted. lwIP keeps a closed pcb alive until the +// connection is fully shut down (e.g. in FIN_WAIT_1) and keeps calling its +// callbacks (poll, sent, err, ...) with the old argument, which would point to +// a socket resource that may already be gone. +// Must be called with the lwIP lock held. +static void socket_tcp_detach_callbacks(struct tcp_pcb *pcb, enum SocketState socket_state) +{ + tcp_arg(pcb, NULL); + if ((socket_state & SocketStateTCPListening) == SocketStateTCPListening) { + // lwIP only accepts the accept callback on listening pcbs and asserts + // when recv, sent, poll or err are set on them. + tcp_accept(pcb, NULL); + } else { + tcp_recv(pcb, NULL); + tcp_sent(pcb, NULL); + tcp_poll(pcb, NULL, 0); + tcp_err(pcb, NULL); + } +} +#endif + static void socket_dtor(ErlNifEnv *caller_env, void *obj) { UNUSED(caller_env); @@ -273,6 +316,7 @@ static void socket_dtor(ErlNifEnv *caller_env, void *obj) } else if (rsrc_obj->socket_state & SocketStateTCP) { // Try to nicely close the connection here // NOTE: we can't handle linger if the socket is gone. + socket_tcp_detach_callbacks(rsrc_obj->tcp_pcb, rsrc_obj->socket_state); if (UNLIKELY(tcp_close(rsrc_obj->tcp_pcb) != ERR_OK)) { // The resource will be gone, there is not much we can do here. if (!(rsrc_obj->socket_state & SocketStateTCPListening)) { @@ -290,22 +334,56 @@ static void socket_dtor(ErlNifEnv *caller_env, void *obj) #endif } +static void socket_select_init(struct SocketSelect *select) +{ + select->process_id = INVALID_PROCESS_ID; + select->monitor.ref_ticks = 0; + select->monitor.resource_type = NULL; + select->ref_ticks = 0; +} + +// Must be used for every socket resource, whatever way it was created, so that +// a new field cannot be left uninitialized on one of the creation paths. +static void socket_init_select_state(struct SocketResource *rsrc_obj) +{ + socket_select_init(&rsrc_obj->read_select); + socket_select_init(&rsrc_obj->write_select); +} + +static struct SocketSelect *socket_get_select(struct SocketResource *rsrc_obj, enum ErlNifSelectFlags direction) +{ + return (direction == ERL_NIF_SELECT_WRITE) ? &rsrc_obj->write_select : &rsrc_obj->read_select; +} + +// Forget the owner of a direction, demonitoring it. +// If the monitor already fired, socket_down will balance the refcount. +static void socket_select_disown(ErlNifEnv *env, struct SocketResource *rsrc_obj, struct SocketSelect *select) +{ + if (select->process_id == INVALID_PROCESS_ID) { + return; + } + // An owner can have no monitor (a process trapped waiting for a connect). + if (select->monitor.resource_type != NULL) { + // demonitor can fail if process is gone. + if (LIKELY(enif_demonitor_process(env, rsrc_obj, &select->monitor) == 0)) { + // decrement ref count as we are demonitoring + refc_binary_decrement_refcount(refc_binary_from_data(rsrc_obj), env->global); + } + select->monitor.resource_type = NULL; + } + select->process_id = INVALID_PROCESS_ID; +} + #if OTP_SOCKET_BSD static void socket_stop(ErlNifEnv *caller_env, void *obj, ErlNifEvent event, int is_direct_call) { - UNUSED(caller_env); UNUSED(event); UNUSED(is_direct_call); struct SocketResource *rsrc_obj = (struct SocketResource *) obj; - if (rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { - if (LIKELY(enif_demonitor_process(caller_env, rsrc_obj, &rsrc_obj->selecting_process_monitor) == 0)) { - struct RefcBinary *rsrc_refc = refc_binary_from_data(rsrc_obj); - refc_binary_decrement_refcount(rsrc_refc, caller_env->global); - } - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; - } + socket_select_disown(caller_env, rsrc_obj, &rsrc_obj->read_select); + socket_select_disown(caller_env, rsrc_obj, &rsrc_obj->write_select); TRACE("socket_stop called on fd=%i\n", rsrc_obj->fd); } @@ -313,9 +391,7 @@ static void socket_stop(ErlNifEnv *caller_env, void *obj, ErlNifEvent event, int static void socket_down(ErlNifEnv *caller_env, void *obj, ErlNifPid *pid, ErlNifMonitor *mon) { - UNUSED(caller_env); UNUSED(pid); - UNUSED(mon); struct SocketResource *rsrc_obj = (struct SocketResource *) obj; @@ -328,34 +404,39 @@ static void socket_down(ErlNifEnv *caller_env, void *obj, ErlNifPid *pid, ErlNif struct RefcBinary *rsrc_refc = refc_binary_from_data(rsrc_obj); SMP_RWLOCK_WRLOCK(rsrc_obj->socket_lock); - if (rsrc_obj->selecting_process_id == INVALID_PROCESS_ID) { + // Only the direction owned by the dead process is cleared. The monitor + // identifies it: if none matches, the owner was already replaced or + // removed and only the reference held for the monitor is left to drop. + enum ErlNifSelectFlags direction; + if (rsrc_obj->read_select.process_id != INVALID_PROCESS_ID && rsrc_obj->read_select.monitor.ref_ticks == mon->ref_ticks) { + direction = ERL_NIF_SELECT_READ; + } else if (rsrc_obj->write_select.process_id != INVALID_PROCESS_ID && rsrc_obj->write_select.monitor.ref_ticks == mon->ref_ticks) { + direction = ERL_NIF_SELECT_WRITE; + } else { SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); + refc_binary_decrement_refcount(rsrc_refc, caller_env->global); return; } + // Monitor fired, so make sure we don't try to demonitor it later + // as it could crash trying to reacquire lock on process table. + struct SocketSelect *select = socket_get_select(rsrc_obj, direction); + select->process_id = INVALID_PROCESS_ID; + select->monitor.resource_type = NULL; + select->ref_ticks = 0; + #if OTP_SOCKET_BSD - // Monitor fired, so make sure we don't try to demonitor in select_stop - // as it could crash trying to reacquire lock on process table // enif_select can decrement ref count but it's at least 2 in this case (1 for monitor and 1 for select) - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; - enif_select(caller_env, rsrc_obj->fd, ERL_NIF_SELECT_STOP, rsrc_obj, NULL, term_nil()); + enif_select(caller_env, rsrc_obj->fd, direction | ERL_NIF_SELECT_CANCEL, rsrc_obj, NULL, term_nil()); + if (rsrc_obj->read_select.process_id == INVALID_PROCESS_ID && rsrc_obj->write_select.process_id == INVALID_PROCESS_ID) { + // No owner left on the other direction, release the select event. + enif_select(caller_env, rsrc_obj->fd, ERL_NIF_SELECT_STOP, rsrc_obj, NULL, term_nil()); + } #elif OTP_SOCKET_LWIP - // Monitor can be called when we're selecting, accepting or connecting. + // The connection is not aborted: like on BSD sockets, it is up to the + // other owner or to the resource destructor to close it. LWIP_BEGIN(); - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; - if (rsrc_obj->socket_state & SocketStateTCP) { - if (rsrc_obj->socket_state & SocketStateTCPListening) { - (void) tcp_close(rsrc_obj->tcp_pcb); - } else { - tcp_abort(rsrc_obj->tcp_pcb); - } - rsrc_obj->tcp_pcb = NULL; - rsrc_obj->socket_state = SocketStateClosed; - } else if (rsrc_obj->socket_state & SocketStateUDP) { - udp_remove(rsrc_obj->udp_pcb); - rsrc_obj->udp_pcb = NULL; - rsrc_obj->socket_state = SocketStateClosed; - } + rsrc_obj->socket_state &= (direction == ERL_NIF_SELECT_WRITE) ? ~SocketStateSelectingWrite : ~SocketStateSelectingRead; LWIP_END(); #endif @@ -377,7 +458,7 @@ static const ErlNifResourceTypeInit SocketResourceTypeInit = { }; // Make a notification message, using SOCKET_MAKE_SELECT_NOTIFICATION_SIZE on heap -static term socket_make_select_notification(struct SocketResource *rsrc_obj, Heap *heap) +static term socket_make_select_notification(struct SocketResource *rsrc_obj, Heap *heap, enum ErlNifSelectFlags direction) { term notification = term_alloc_tuple(4, heap); term_put_tuple_element(notification, 0, DOLLAR_SOCKET_ATOM); @@ -392,11 +473,12 @@ static term socket_make_select_notification(struct SocketResource *rsrc_obj, Hea term_put_tuple_element(socket_tuple, 1, socket_ref); term_put_tuple_element(notification, 1, socket_tuple); term_put_tuple_element(notification, 2, SELECT_ATOM); + uint64_t select_ref_ticks = socket_get_select(rsrc_obj, direction)->ref_ticks; term select_ref; - if (rsrc_obj->select_ref_ticks == 0) { + if (select_ref_ticks == 0) { select_ref = UNDEFINED_ATOM; } else { - select_ref = term_from_ref_ticks(rsrc_obj->select_ref_ticks, heap); + select_ref = term_from_ref_ticks(select_ref_ticks, heap); } term_put_tuple_element(notification, 3, select_ref); return notification; @@ -404,20 +486,20 @@ static term socket_make_select_notification(struct SocketResource *rsrc_obj, Hea // select emulation for lwIP that doesn't have select. #if OTP_SOCKET_LWIP -static void select_event_send_notification_from_nif(struct SocketResource *rsrc_obj, Context *locked_ctx) +static void select_event_send_notification_from_nif(struct SocketResource *rsrc_obj, Context *locked_ctx, enum ErlNifSelectFlags direction) { BEGIN_WITH_STACK_HEAP(SOCKET_MAKE_SELECT_NOTIFICATION_SIZE, heap) - term notification = socket_make_select_notification(rsrc_obj, &heap); + term notification = socket_make_select_notification(rsrc_obj, &heap, direction); mailbox_send(locked_ctx, notification); END_WITH_STACK_HEAP(heap, locked_ctx->global) } -static void select_event_send_notification_from_handler(struct SocketResource *rsrc_obj, int32_t process_id) +static void select_event_send_notification_from_handler(struct SocketResource *rsrc_obj, int32_t process_id, enum ErlNifSelectFlags direction) { struct RefcBinary *rsrc_refc = refc_binary_from_data(rsrc_obj); GlobalContext *global = rsrc_refc->resource_type->global; BEGIN_WITH_STACK_HEAP(SOCKET_MAKE_SELECT_NOTIFICATION_SIZE, heap) - term notification = socket_make_select_notification(rsrc_obj, &heap); + term notification = socket_make_select_notification(rsrc_obj, &heap, direction); globalcontext_send_message(global, process_id, notification); END_WITH_STACK_HEAP(heap, global) } @@ -587,7 +669,7 @@ static term nif_socket_open(Context *ctx, int argc, term argv[]) return make_errno_tuple(ctx); } else { TRACE("nif_socket_open: Created socket fd=%i\n", rsrc_obj->fd); - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + socket_init_select_state(rsrc_obj); if (type != SOCK_STREAM) { // TCP sockets are made non-blocking after connect, for now. @@ -618,7 +700,7 @@ static term nif_socket_open(Context *ctx, int argc, term argv[]) enif_release_resource(rsrc_obj); RAISE_ERROR(OUT_OF_MEMORY_ATOM); } else { - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + socket_init_select_state(rsrc_obj); rsrc_obj->linger_on = false; rsrc_obj->linger_sec = 0; rsrc_obj->pos = 0; @@ -628,6 +710,8 @@ static term nif_socket_open(Context *ctx, int argc, term argv[]) tcp_arg(rsrc_obj->tcp_pcb, rsrc_obj); tcp_err(rsrc_obj->tcp_pcb, tcp_err_cb); tcp_recv(rsrc_obj->tcp_pcb, tcp_recv_cb); + tcp_sent(rsrc_obj->tcp_pcb, tcp_sent_cb); + tcp_poll(rsrc_obj->tcp_pcb, tcp_poll_cb, SOCKET_TCP_POLL_INTERVAL); LWIP_END(); } else { LWIP_BEGIN(); @@ -692,7 +776,7 @@ bool term_is_otp_socket(term socket_term) // close // -static int send_closed_notification(Context *ctx, term socket_term, int32_t selecting_process_id, struct SocketResource *rsrc_obj) +static int send_closed_notification(Context *ctx, term socket_term, int32_t selecting_process_id, struct SocketResource *rsrc_obj, enum ErlNifSelectFlags direction) { // send a {'$socket', Socket, abort, {Ref | undefined, closed}} message to the pid if (UNLIKELY(memory_ensure_free_with_roots(ctx, TUPLE_SIZE(4) + TUPLE_SIZE(2) + TERM_BOXED_REFERENCE_SHORT_SIZE, 1, &socket_term, MEMORY_CAN_SHRINK) != MEMORY_GC_OK)) { @@ -706,7 +790,8 @@ static int send_closed_notification(Context *ctx, term socket_term, int32_t sele term_put_tuple_element(socket_tuple, 2, ABORT_ATOM); term error_tuple = term_alloc_tuple(2, &ctx->heap); - term ref = (rsrc_obj->select_ref_ticks == 0) ? UNDEFINED_ATOM : term_from_ref_ticks(rsrc_obj->select_ref_ticks, &ctx->heap); + uint64_t select_ref_ticks = socket_get_select(rsrc_obj, direction)->ref_ticks; + term ref = (select_ref_ticks == 0) ? UNDEFINED_ATOM : term_from_ref_ticks(select_ref_ticks, &ctx->heap); term_put_tuple_element(error_tuple, 0, ref); term_put_tuple_element(error_tuple, 1, CLOSED_ATOM); term_put_tuple_element(socket_tuple, 3, error_tuple); @@ -748,29 +833,51 @@ static term nif_socket_close(Context *ctx, int argc, term argv[]) // process, as documented in specification of the abort message. // So we handle closing a socket while another process is selecting - if (rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { - // Save process id as socket_stop may be called by enif_select. - int32_t selecting_process_id = rsrc_obj->selecting_process_id; + if (rsrc_obj->read_select.process_id != INVALID_PROCESS_ID || rsrc_obj->write_select.process_id != INVALID_PROCESS_ID) { + // Save the owners as socket_stop may be called by enif_select. + int32_t read_process_id = rsrc_obj->read_select.process_id; + int32_t write_process_id = rsrc_obj->write_select.process_id; // Another process is selecting, therefore ref_count >= 2 // 1. this caller's context heap (parameter to close) // 2. select + // Cancel each direction first: only a select that is still pending + // needs to be told that the socket is gone. A select that already + // fired has delivered its notification. + ErlNifEnv *env = erl_nif_env_from_context(ctx); + bool read_pending = false; + bool write_pending = false; + if (read_process_id != INVALID_PROCESS_ID) { + int cancel_res = enif_select(env, rsrc_obj->fd, ERL_NIF_SELECT_READ | ERL_NIF_SELECT_CANCEL, rsrc_obj, NULL, term_nil()); + read_pending = cancel_res > 0 && (cancel_res & ERL_NIF_SELECT_READ_CANCELLED); + } + if (write_process_id != INVALID_PROCESS_ID) { + int cancel_res = enif_select(env, rsrc_obj->fd, ERL_NIF_SELECT_WRITE | ERL_NIF_SELECT_CANCEL, rsrc_obj, NULL, term_nil()); + write_pending = cancel_res > 0 && (cancel_res & ERL_NIF_SELECT_WRITE_CANCELLED); + } + // Stop selecting. - int stop_res = enif_select(erl_nif_env_from_context(ctx), rsrc_obj->fd, ERL_NIF_SELECT_STOP, rsrc_obj, NULL, term_nil()); + int stop_res = enif_select(env, rsrc_obj->fd, ERL_NIF_SELECT_STOP, rsrc_obj, NULL, term_nil()); if (UNLIKELY(stop_res < 0)) { SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(BADARG_ATOM); } - // If there is a selecting process who may be waiting on select, - // send a closed notification to it, so that it can break - // out of its receive statement. + // If there are selecting processes who may be waiting on select, + // send a closed notification to each of them, so that they can + // break out of their receive statements. // // When using asynchronous API, the selecting process can be the // calling process. In this case we don't send any notification. // - if (selecting_process_id != ctx->process_id) { - // send a {'$socket', Socket, abort, {Ref | undefined, closed}} message to the pid - if (UNLIKELY(send_closed_notification(ctx, argv[0], selecting_process_id, rsrc_obj) < 0)) { + // send a {'$socket', Socket, abort, {Ref | undefined, closed}} message to the pid + if (read_pending && read_process_id != ctx->process_id) { + if (UNLIKELY(send_closed_notification(ctx, argv[0], read_process_id, rsrc_obj, ERL_NIF_SELECT_READ) < 0)) { + SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); + RAISE_ERROR(OUT_OF_MEMORY_ATOM); + } + } + if (write_pending && write_process_id != ctx->process_id) { + if (UNLIKELY(send_closed_notification(ctx, argv[0], write_process_id, rsrc_obj, ERL_NIF_SELECT_WRITE) < 0)) { SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(OUT_OF_MEMORY_ATOM); } @@ -789,12 +896,21 @@ static term nif_socket_close(Context *ctx, int argc, term argv[]) TRACE("Double close on socket fd %i\n", rsrc_obj->fd); } #elif OTP_SOCKET_LWIP - // If the socket is being selected by another process, send a closed notification. - if (rsrc_obj->socket_state & SocketStateSelectingRead - && rsrc_obj->selecting_process_id != INVALID_PROCESS_ID - && rsrc_obj->selecting_process_id != ctx->process_id) { - // send a {'$socket', Socket, abort, {Ref | undefined, closed}} message to the pid - if (UNLIKELY(send_closed_notification(ctx, argv[0], rsrc_obj->selecting_process_id, rsrc_obj) < 0)) { + // If the socket is being selected by other processes, send a closed + // notification to the owner of each direction being selected. + // send a {'$socket', Socket, abort, {Ref | undefined, closed}} message to the pid + if ((rsrc_obj->socket_state & SocketStateSelectingRead) + && rsrc_obj->read_select.process_id != INVALID_PROCESS_ID + && rsrc_obj->read_select.process_id != ctx->process_id) { + if (UNLIKELY(send_closed_notification(ctx, argv[0], rsrc_obj->read_select.process_id, rsrc_obj, ERL_NIF_SELECT_READ) < 0)) { + SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); + RAISE_ERROR(OUT_OF_MEMORY_ATOM); + } + } + if ((rsrc_obj->socket_state & SocketStateSelectingWrite) + && rsrc_obj->write_select.process_id != INVALID_PROCESS_ID + && rsrc_obj->write_select.process_id != ctx->process_id) { + if (UNLIKELY(send_closed_notification(ctx, argv[0], rsrc_obj->write_select.process_id, rsrc_obj, ERL_NIF_SELECT_WRITE) < 0)) { SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(OUT_OF_MEMORY_ATOM); } @@ -804,7 +920,7 @@ static term nif_socket_close(Context *ctx, int argc, term argv[]) } else { if (rsrc_obj->socket_state & SocketStateTCP) { LWIP_BEGIN(); - tcp_arg(rsrc_obj->tcp_pcb, NULL); + socket_tcp_detach_callbacks(rsrc_obj->tcp_pcb, rsrc_obj->socket_state); err_t err = tcp_close(rsrc_obj->tcp_pcb); LWIP_END(); if (UNLIKELY(err != ERR_OK)) { @@ -848,7 +964,7 @@ static struct SocketResource *make_accepted_socket_resource(struct tcp_pcb *newp } conn_rsrc_obj->socket_state = SocketStateTCPConnected; conn_rsrc_obj->tcp_pcb = newpcb; - conn_rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + socket_init_select_state(conn_rsrc_obj); conn_rsrc_obj->pos = 0; conn_rsrc_obj->linger_on = false; conn_rsrc_obj->linger_sec = 0; @@ -864,6 +980,8 @@ static struct SocketResource *make_accepted_socket_resource(struct tcp_pcb *newp tcp_arg(newpcb, conn_rsrc_obj); tcp_recv(newpcb, tcp_recv_cb); + tcp_sent(newpcb, tcp_sent_cb); + tcp_poll(newpcb, tcp_poll_cb, SOCKET_TCP_POLL_INTERVAL); return conn_rsrc_obj; } @@ -878,8 +996,8 @@ static void tcp_accept_handler(struct LWIPEvent *event) if (rsrc_obj->socket_state & SocketStateSelectingRead) { // Clear flag to avoid sending a message again. rsrc_obj->socket_state &= ~SocketStateSelectingRead; - if (rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { - select_event_send_notification_from_handler(rsrc_obj, rsrc_obj->selecting_process_id); + if (rsrc_obj->read_select.process_id != INVALID_PROCESS_ID) { + select_event_send_notification_from_handler(rsrc_obj, rsrc_obj->read_select.process_id, ERL_NIF_SELECT_READ); } // otherwise, selecting process died but we can just wait for monitor to handle it } } @@ -889,7 +1007,7 @@ static err_t tcp_accept_cb(void *arg, struct tcp_pcb *newpcb, err_t err) UNUSED(err); struct SocketResource *rsrc_obj = (struct SocketResource *) arg; - if (rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { + if (rsrc_obj->read_select.process_id != INVALID_PROCESS_ID) { if (newpcb != NULL) { // Because data can come in, we need to immediately set the receive callback // However, we cannot allocate the resource because we can't call malloc from ISR. @@ -941,8 +1059,8 @@ static void tcp_recv_handler(struct LWIPEvent *event) if (rsrc_obj->socket_state & SocketStateSelectingRead) { // Clear flag to avoid sending a message again. rsrc_obj->socket_state &= ~SocketStateSelectingRead; - if (rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { - select_event_send_notification_from_handler(rsrc_obj, rsrc_obj->selecting_process_id); + if (rsrc_obj->read_select.process_id != INVALID_PROCESS_ID) { + select_event_send_notification_from_handler(rsrc_obj, rsrc_obj->read_select.process_id, ERL_NIF_SELECT_READ); } // otherwise, selecting process died but we can just wait for monitor to handle it } } @@ -963,6 +1081,73 @@ static err_t tcp_recv_cb(void *arg, struct tcp_pcb *tpcb, struct pbuf *p, err_t return ERR_OK; } +// Returns true if the lwIP TCP send buffer has room for more data, i.e. a +// subsequent tcp_write/tcp_output is likely to succeed rather than return +// ERR_MEM. +// Must be called with the lwIP lock held. +static bool lwip_tcp_can_send(const struct SocketResource *rsrc_obj) +{ + return rsrc_obj->tcp_pcb != NULL && tcp_sndbuf(rsrc_obj->tcp_pcb) > 0 && tcp_sndqueuelen(rsrc_obj->tcp_pcb) < TCP_SND_QUEUELEN; +} + +static void tcp_sent_handler(struct LWIPEvent *event) +{ + struct SocketResource *rsrc_obj = event->tcp_sent.rsrc_obj; + + // Send notification if we are selecting for write and the send buffer + // has room again. + if (!(rsrc_obj->socket_state & SocketStateSelectingWrite)) { + return; + } + LWIP_BEGIN(); + bool can_send = lwip_tcp_can_send(rsrc_obj); + LWIP_END(); + if (can_send) { + // Clear flag to avoid sending a message again. + rsrc_obj->socket_state &= ~SocketStateSelectingWrite; + if (rsrc_obj->write_select.process_id != INVALID_PROCESS_ID) { + select_event_send_notification_from_handler(rsrc_obj, rsrc_obj->write_select.process_id, ERL_NIF_SELECT_WRITE); + } // otherwise, selecting process died but we can just wait for monitor to handle it + } +} + +// Called by lwIP when previously written data has been acknowledged by the +// peer, freeing up room in the send buffer. +static err_t tcp_sent_cb(void *arg, struct tcp_pcb *tpcb, u16_t len) +{ + UNUSED(tpcb); + UNUSED(len); + + // Only post an event when a process is waiting for write-readiness: the + // event queue is small and shared with the other socket events, and every + // post can wake the scheduler. + struct SocketResource *rsrc_obj = (struct SocketResource *) arg; + if (LIKELY(rsrc_obj) && (rsrc_obj->socket_state & SocketStateSelectingWrite)) { + struct LWIPEvent event; + event.handler = tcp_sent_handler; + event.tcp_sent.rsrc_obj = rsrc_obj; + otp_socket_lwip_enqueue(&event); + } + return ERR_OK; +} + +// Periodic poll callback, used as a fallback wakeup for write-readiness: some +// lwIP send failure paths (e.g. hitting TCP_SND_QUEUELEN without any bytes +// being freshly acknowledged) would otherwise never trigger tcp_sent_cb. +static err_t tcp_poll_cb(void *arg, struct tcp_pcb *tpcb) +{ + UNUSED(tpcb); + + struct SocketResource *rsrc_obj = (struct SocketResource *) arg; + if (LIKELY(rsrc_obj) && (rsrc_obj->socket_state & SocketStateSelectingWrite)) { + struct LWIPEvent event; + event.handler = tcp_sent_handler; + event.tcp_sent.rsrc_obj = rsrc_obj; + otp_socket_lwip_enqueue(&event); + } + return ERR_OK; +} + static void udp_recv_handler(struct LWIPEvent *event) { struct SocketResource *rsrc_obj = event->udp_recv.rsrc_obj; @@ -976,8 +1161,8 @@ static void udp_recv_handler(struct LWIPEvent *event) if (rsrc_obj->socket_state & SocketStateSelectingRead) { // Clear flag to avoid sending a message again. rsrc_obj->socket_state &= ~SocketStateSelectingRead; - if (rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { - select_event_send_notification_from_handler(rsrc_obj, rsrc_obj->selecting_process_id); + if (rsrc_obj->read_select.process_id != INVALID_PROCESS_ID) { + select_event_send_notification_from_handler(rsrc_obj, rsrc_obj->read_select.process_id, ERL_NIF_SELECT_READ); } // otherwise, selecting process died } } @@ -1021,9 +1206,9 @@ int tcpip_try_callback(tcpip_callback_fn function, void *ctx) #endif -static term nif_socket_select_read(Context *ctx, int argc, term argv[]) +static term nif_socket_select(Context *ctx, int argc, term argv[], enum ErlNifSelectFlags direction) { - TRACE("nif_socket_select_read\n"); + TRACE("nif_socket_select direction=%i\n", (int) direction); UNUSED(argc); @@ -1046,46 +1231,42 @@ static term nif_socket_select_read(Context *ctx, int argc, term argv[]) SMP_RWLOCK_WRLOCK(rsrc_obj->socket_lock); ErlNifEnv *env = erl_nif_env_from_context(ctx); - if (rsrc_obj->selecting_process_id != ctx->process_id && rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { - // demonitor can fail if process is gone. - if (LIKELY(enif_demonitor_process(env, rsrc_obj, &rsrc_obj->selecting_process_monitor) == 0)) { - // decrement ref count as we are demonitoring - refc_binary_decrement_refcount(rsrc_refc, ctx->global); - } - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; - } - // Monitor first as select is less likely to fail and it's less expensive to demonitor - // if select fails than to stop select if monitor fails - if (rsrc_obj->selecting_process_id != ctx->process_id) { - if (UNLIKELY(enif_monitor_process(env, rsrc_obj, &ctx->process_id, &rsrc_obj->selecting_process_monitor) != 0)) { + // Only the requested direction is touched, the other one keeps its owner. + // Another process selecting the same direction is replaced, not queued. + struct SocketSelect *select = socket_get_select(rsrc_obj, direction); + if (select->process_id != ctx->process_id) { + socket_select_disown(env, rsrc_obj, select); + // Monitor first as select is less likely to fail and it's less expensive to demonitor + // if select fails than to stop select if monitor fails + if (UNLIKELY(enif_monitor_process(env, rsrc_obj, &ctx->process_id, &select->monitor) != 0)) { SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(NOPROC_ATOM); } // increment ref count so the resource doesn't go away until monitor is fired refc_binary_increment_refcount(rsrc_refc); - rsrc_obj->selecting_process_id = ctx->process_id; + select->process_id = ctx->process_id; } - rsrc_obj->select_ref_ticks = (select_ref_term == UNDEFINED_ATOM) ? 0 : term_to_ref_ticks(select_ref_term); + select->ref_ticks = (select_ref_term == UNDEFINED_ATOM) ? 0 : term_to_ref_ticks(select_ref_term); #if OTP_SOCKET_BSD TRACE("rsrc_obj->fd=%i\n", (int) rsrc_obj->fd); // The socket may be closed here. if (rsrc_obj->fd == CLOSED_FD) { - send_closed_notification(ctx, argv[0], ctx->process_id, rsrc_obj); + send_closed_notification(ctx, argv[0], ctx->process_id, rsrc_obj, direction); } else { if (UNLIKELY(memory_ensure_free_with_roots(ctx, SOCKET_MAKE_SELECT_NOTIFICATION_SIZE, 2, argv, MEMORY_CAN_SHRINK) != MEMORY_GC_OK)) { AVM_LOGW(TAG, "Failed to allocate memory: %s:%i.", __FILE__, __LINE__); SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(OUT_OF_MEMORY_ATOM); } - term notification = socket_make_select_notification(rsrc_obj, &ctx->heap); - if (UNLIKELY(enif_select_read(erl_nif_env_from_context(ctx), rsrc_obj->fd, rsrc_obj, &ctx->process_id, notification, NULL) < 0)) { - if (LIKELY(enif_demonitor_process(env, rsrc_obj, &rsrc_obj->selecting_process_monitor) == 0)) { - refc_binary_decrement_refcount(rsrc_refc, ctx->global); - } - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + term notification = socket_make_select_notification(rsrc_obj, &ctx->heap, direction); + int select_result = (direction == ERL_NIF_SELECT_WRITE) + ? enif_select_write(env, rsrc_obj->fd, rsrc_obj, &ctx->process_id, notification, NULL) + : enif_select_read(env, rsrc_obj->fd, rsrc_obj, &ctx->process_id, notification, NULL); + if (UNLIKELY(select_result < 0)) { + socket_select_disown(env, rsrc_obj, select); SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(BADARG_ATOM); } @@ -1093,50 +1274,75 @@ static term nif_socket_select_read(Context *ctx, int argc, term argv[]) #elif OTP_SOCKET_LWIP LWIP_BEGIN(); - switch (rsrc_obj->socket_state) { - case SocketStateTCPListening: { - if (!list_is_empty(&rsrc_obj->received_list)) { + if (direction == ERL_NIF_SELECT_WRITE) { + if ((rsrc_obj->socket_state & SocketStateTCPConnected) == SocketStateTCPConnected) { + if (lwip_tcp_can_send(rsrc_obj)) { + // Clear the flag, in case a previous write select was still + // pending (replaced by this one), to avoid sending a second + // notification from tcp_sent_handler. + rsrc_obj->socket_state &= ~SocketStateSelectingWrite; // Send (or resend) notification - select_event_send_notification_from_nif(rsrc_obj, ctx); + select_event_send_notification_from_nif(rsrc_obj, ctx, ERL_NIF_SELECT_WRITE); } else { - // Set flag to send it when a packet will arrive. - rsrc_obj->socket_state = SocketStateTCPSelectingAccept; - } - } break; - case SocketStateTCPSelectingAccept: - // noop - break; - case SocketStateTCPConnected: { - if (!list_is_empty(&rsrc_obj->received_list)) { - // Send (or resend) notification - select_event_send_notification_from_nif(rsrc_obj, ctx); - } else { - // Set flag to send it when a packet will arrive. - rsrc_obj->socket_state = SocketStateTCPSelectingRead; - } - } break; - case SocketStateTCPSelectingRead: - // noop - break; - case SocketStateUDPIdle: { - if (!list_is_empty(&rsrc_obj->received_list)) { - // Send (or resend) notification - select_event_send_notification_from_nif(rsrc_obj, ctx); - } else { - rsrc_obj->socket_state = SocketStateUDPSelectingRead; - } - } break; - case SocketStateUDPSelectingRead: - // noop - break; - default: - if (LIKELY(enif_demonitor_process(env, rsrc_obj, &rsrc_obj->selecting_process_monitor) == 0)) { - refc_binary_decrement_refcount(rsrc_refc, ctx->global); + // Set flag to send it when the send buffer has room again. + rsrc_obj->socket_state |= SocketStateSelectingWrite; } - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + } else if (rsrc_obj->socket_state & SocketStateUDP) { + // UDP datagrams are never buffered by lwIP in a way that blocks + // the caller, so writability is immediate. + select_event_send_notification_from_nif(rsrc_obj, ctx, ERL_NIF_SELECT_WRITE); + } else { + socket_select_disown(env, rsrc_obj, select); LWIP_END(); SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(BADARG_ATOM); + } + } else { + switch (rsrc_obj->socket_state) { + case SocketStateTCPListening: { + if (!list_is_empty(&rsrc_obj->received_list)) { + // Send (or resend) notification + select_event_send_notification_from_nif(rsrc_obj, ctx, ERL_NIF_SELECT_READ); + } else { + // Set flag to send it when a packet will arrive. + rsrc_obj->socket_state = SocketStateTCPSelectingAccept; + } + } break; + case SocketStateTCPSelectingAccept: + // noop + break; + case SocketStateTCPConnected: + case SocketStateTCPSelectingWrite: { + if (!list_is_empty(&rsrc_obj->received_list)) { + // Send (or resend) notification + select_event_send_notification_from_nif(rsrc_obj, ctx, ERL_NIF_SELECT_READ); + } else { + // Set flag to send it when a packet will arrive, preserving + // a pending write select on the same socket. + rsrc_obj->socket_state |= SocketStateSelectingRead; + } + } break; + case SocketStateTCPSelectingRead: + case SocketStateTCPSelectingReadWrite: + // noop + break; + case SocketStateUDPIdle: { + if (!list_is_empty(&rsrc_obj->received_list)) { + // Send (or resend) notification + select_event_send_notification_from_nif(rsrc_obj, ctx, ERL_NIF_SELECT_READ); + } else { + rsrc_obj->socket_state = SocketStateUDPSelectingRead; + } + } break; + case SocketStateUDPSelectingRead: + // noop + break; + default: + socket_select_disown(env, rsrc_obj, select); + LWIP_END(); + SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); + RAISE_ERROR(BADARG_ATOM); + } } LWIP_END(); #endif @@ -1145,36 +1351,61 @@ static term nif_socket_select_read(Context *ctx, int argc, term argv[]) return OK_ATOM; } -static term nif_socket_select_stop(Context *ctx, int argc, term argv[]) +static term nif_socket_select_read(Context *ctx, int argc, term argv[]) { - TRACE("nif_socket_stop\n"); - UNUSED(argc); + return nif_socket_select(ctx, argc, argv, ERL_NIF_SELECT_READ); +} - VALIDATE_VALUE(argv[0], term_is_otp_socket); +static term nif_socket_select_write(Context *ctx, int argc, term argv[]) +{ + return nif_socket_select(ctx, argc, argv, ERL_NIF_SELECT_WRITE); +} + +// Stop selecting on a socket. When read_only is true and another process is +// selecting for write on the same socket, only the read direction is stopped, +// so that cleaning up after a failed read does not disown a blocked writer. +static term nif_socket_select_stop_internal(Context *ctx, term socket, bool read_only) +{ + VALIDATE_VALUE(socket, term_is_otp_socket); struct SocketResource *rsrc_obj; - if (UNLIKELY(!term_to_otp_socket(argv[0], &rsrc_obj, ctx))) { + if (UNLIKELY(!term_to_otp_socket(socket, &rsrc_obj, ctx))) { RAISE_ERROR(BADARG_ATOM); } + ErlNifEnv *env = erl_nif_env_from_context(ctx); // Avoid the race condition with select object here. SMP_RWLOCK_WRLOCK(rsrc_obj->socket_lock); - if (rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { - if (LIKELY(enif_demonitor_process(erl_nif_env_from_context(ctx), rsrc_obj, &rsrc_obj->selecting_process_monitor) == 0)) { - struct RefcBinary *rsrc_refc = refc_binary_from_data(rsrc_obj); - refc_binary_decrement_refcount(rsrc_refc, ctx->global); + if (read_only && rsrc_obj->write_select.process_id != INVALID_PROCESS_ID) { + // The write direction has an owner: stop the read direction only. + socket_select_disown(env, rsrc_obj, &rsrc_obj->read_select); + rsrc_obj->read_select.ref_ticks = 0; +#if OTP_SOCKET_BSD + if (UNLIKELY(enif_select(env, rsrc_obj->fd, ERL_NIF_SELECT_READ | ERL_NIF_SELECT_CANCEL, rsrc_obj, NULL, term_nil()) < 0)) { + SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); + RAISE_ERROR(BADARG_ATOM); } - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; +#elif OTP_SOCKET_LWIP + LWIP_BEGIN(); + rsrc_obj->socket_state &= ~SocketStateSelectingRead; + LWIP_END(); +#endif + SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); + return OK_ATOM; } + + // Stop both directions. + socket_select_disown(env, rsrc_obj, &rsrc_obj->read_select); + socket_select_disown(env, rsrc_obj, &rsrc_obj->write_select); + rsrc_obj->read_select.ref_ticks = 0; + rsrc_obj->write_select.ref_ticks = 0; #if OTP_SOCKET_BSD - if (UNLIKELY(enif_select(erl_nif_env_from_context(ctx), rsrc_obj->fd, ERL_NIF_SELECT_STOP, rsrc_obj, NULL, term_nil()) < 0)) { + if (UNLIKELY(enif_select(env, rsrc_obj->fd, ERL_NIF_SELECT_STOP, rsrc_obj, NULL, term_nil()) < 0)) { SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(BADARG_ATOM); } #elif OTP_SOCKET_LWIP LWIP_BEGIN(); - if (rsrc_obj->socket_state & SocketStateSelectingRead) { - rsrc_obj->socket_state &= ~SocketStateSelectingRead; - } + rsrc_obj->socket_state &= ~(SocketStateSelectingRead | SocketStateSelectingWrite); LWIP_END(); #endif SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); @@ -1182,6 +1413,22 @@ static term nif_socket_select_stop(Context *ctx, int argc, term argv[]) return OK_ATOM; } +static term nif_socket_select_stop(Context *ctx, int argc, term argv[]) +{ + TRACE("nif_socket_select_stop\n"); + UNUSED(argc); + + return nif_socket_select_stop_internal(ctx, argv[0], false); +} + +static term nif_socket_select_read_stop(Context *ctx, int argc, term argv[]) +{ + TRACE("nif_socket_select_read_stop\n"); + UNUSED(argc); + + return nif_socket_select_stop_internal(ctx, argv[0], true); +} + // // getopt // @@ -1895,7 +2142,7 @@ static term nif_socket_accept(Context *ctx, int argc, term argv[]) } struct SocketResource *conn_rsrc_obj = enif_alloc_resource(socket_resource_type, sizeof(struct SocketResource)); conn_rsrc_obj->fd = fd; - conn_rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + socket_init_select_state(conn_rsrc_obj); conn_rsrc_obj->buf_size = DEFAULT_BUFFER_SIZE; #ifndef AVM_NO_SMP conn_rsrc_obj->socket_lock = smp_rwlock_create(); @@ -2668,8 +2915,9 @@ static err_t tcp_connected_cb(void *arg, struct tcp_pcb *tpcb, err_t err) struct SocketResource *rsrc_obj = (struct SocketResource *) arg; struct RefcBinary *rsrc_refc = refc_binary_from_data(rsrc_obj); GlobalContext *global = rsrc_refc->resource_type->global; - int32_t target_pid = rsrc_obj->selecting_process_id; - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + // The process waiting for the connect to complete is trapped on the write slot. + int32_t target_pid = rsrc_obj->write_select.process_id; + rsrc_obj->write_select.process_id = INVALID_PROCESS_ID; rsrc_obj->socket_state = SocketStateTCPConnected; if (target_pid != INVALID_PROCESS_ID) { if (err == ERR_OK) { @@ -2694,10 +2942,19 @@ static void tcp_err_cb(void *arg, err_t err) UNUSED(err); struct SocketResource *rsrc_obj = (struct SocketResource *) arg; + if (UNLIKELY(rsrc_obj == NULL)) { + // Socket was closed. + return; + } struct RefcBinary *rsrc_refc = refc_binary_from_data(rsrc_obj); GlobalContext *global = rsrc_refc->resource_type->global; - int32_t target_pid = rsrc_obj->selecting_process_id; - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + // Only a process trapped waiting for a connect is answered: once connected + // the write slot belongs to a process selecting for write, which is not trapped. + int32_t target_pid = INVALID_PROCESS_ID; + if (rsrc_obj->socket_state == SocketStateTCPNew) { + target_pid = rsrc_obj->write_select.process_id; + rsrc_obj->write_select.process_id = INVALID_PROCESS_ID; + } rsrc_obj->socket_state = SocketStateTCPConnected; if (target_pid != INVALID_PROCESS_ID) { struct LWIPEvent event; @@ -2818,7 +3075,7 @@ static term nif_socket_connect(Context *ctx, int argc, term argv[]) SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); return OK_ATOM; } else { - rsrc_obj->selecting_process_id = ctx->process_id; + rsrc_obj->write_select.process_id = ctx->process_id; // Trap caller waiting for completion context_update_flags(ctx, ~NoFlags, Trap); SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); @@ -2891,6 +3148,13 @@ static term nif_socket_shutdown(Context *ctx, int argc, term argv[]) if (rsrc_obj->socket_state & SocketStateTCP) { err_t res = tcp_shutdown(rsrc_obj->tcp_pcb, how != SHUT_WR, how != SHUT_RD); if (how == SHUT_RDWR && res == ERR_OK) { + // tcp_shutdown(1, 1) closes the pcb, but lwIP keeps it (e.g. in + // FIN_WAIT) and its callbacks running with our argument. We will + // lose the pointer to the pcb, so detach them now. This is done + // after tcp_shutdown succeeded so that a failed shutdown leaves the + // socket intact, and under the lwIP lock so no callback can run in + // between. + socket_tcp_detach_callbacks(rsrc_obj->tcp_pcb, rsrc_obj->socket_state); rsrc_obj->tcp_pcb = NULL; } if (res != ERR_OK) { @@ -2920,6 +3184,10 @@ static const struct Nif socket_select_stop_nif = { .base.type = NIFFunctionType, .nif_ptr = nif_socket_select_stop }; +static const struct Nif socket_select_read_stop_nif = { + .base.type = NIFFunctionType, + .nif_ptr = nif_socket_select_read_stop +}; static const struct Nif socket_getopt_nif = { .base.type = NIFFunctionType, .nif_ptr = nif_socket_getopt @@ -2948,6 +3216,10 @@ static const struct Nif socket_select_read_nif = { .base.type = NIFFunctionType, .nif_ptr = nif_socket_select_read }; +static const struct Nif socket_select_write_nif = { + .base.type = NIFFunctionType, + .nif_ptr = nif_socket_select_write +}; static const struct Nif socket_accept_nif = { .base.type = NIFFunctionType, .nif_ptr = nif_socket_accept @@ -2994,6 +3266,10 @@ const struct Nif *otp_socket_nif_get_nif(const char *nifname) TRACE("Resolved platform nif %s ...\n", nifname); return &socket_select_stop_nif; } + if (strcmp("nif_select_read_stop/1", rest) == 0) { + TRACE("Resolved platform nif %s ...\n", nifname); + return &socket_select_read_stop_nif; + } if (strcmp("getopt/2", rest) == 0) { TRACE("Resolved platform nif %s ...\n", nifname); return &socket_getopt_nif; @@ -3022,6 +3298,10 @@ const struct Nif *otp_socket_nif_get_nif(const char *nifname) TRACE("Resolved platform nif %s ...\n", nifname); return &socket_select_read_nif; } + if (strcmp("nif_select_write/2", rest) == 0) { + TRACE("Resolved platform nif %s ...\n", nifname); + return &socket_select_write_nif; + } if (strcmp("nif_accept/1", rest) == 0) { TRACE("Resolved platform nif %s ...\n", nifname); return &socket_accept_nif; diff --git a/src/libAtomVM/otp_socket.h b/src/libAtomVM/otp_socket.h index cfd9626531..ca8cd43f02 100644 --- a/src/libAtomVM/otp_socket.h +++ b/src/libAtomVM/otp_socket.h @@ -111,6 +111,10 @@ struct LWIPEvent err_t err; } tcp_recv; struct + { + struct SocketResource *rsrc_obj; + } tcp_sent; + struct { struct SocketResource *rsrc_obj; struct pbuf *buf; From daaad3b5938de082cb15686d4ee24152dd0efe92 Mon Sep 17 00:00:00 2001 From: harmon25 Date: Sat, 4 Jul 2026 17:05:44 -0400 Subject: [PATCH 4/7] socket: Retry sends on backpressure Make socket:send/2 and socket:sendto/3 retry partial sends and {error, eagain} internally, waiting on nif_select_write/2 for the socket to become writable again. Callers get ok | {error, Reason} and do not need their own retry and backoff loop. Signed-off-by: harmon25 --- libs/estdlib/src/socket.erl | 100 +++++++++++++++++++++++++++++++++--- 1 file changed, 92 insertions(+), 8 deletions(-) diff --git a/libs/estdlib/src/socket.erl b/libs/estdlib/src/socket.erl index ea212f55d1..7709af80b7 100644 --- a/libs/estdlib/src/socket.erl +++ b/libs/estdlib/src/socket.erl @@ -46,6 +46,7 @@ %% internal nifs -export([ nif_select_read/2, + nif_select_write/2, nif_accept/1, nif_recv/2, nif_recvfrom/2, @@ -572,7 +573,10 @@ recvfrom0_nowait(Socket, Length, Ref) -> %%----------------------------------------------------------------------------- %% @param Socket the socket %% @param Data the data to send -%% @returns `{ok, Rest}' if successful; `{error, Reason}', otherwise. +%% @returns `ok' if successful; `{error, Reason}', otherwise, or +%% `{error, {Reason, RestData}}' if an error occurred after some of +%% the data was already sent, `RestData' being the data that was not +%% sent. %% @doc Send data on the specified socket. %% %% Note that this function will block until data is sent @@ -580,23 +584,29 @@ recvfrom0_nowait(Socket, Length, Ref) -> %% intended recipient, and the data may not even have been sent %% over the network. %% +%% Partial sends and transient send backpressure (e.g. a full TCP +%% send buffer) are retried internally, waiting for the socket to +%% become writable again, until all data has been accepted by the +%% stack or an error occurs. Callers therefore do not need to +%% implement their own retry logic for `eagain'-like conditions. +%% %% Example: %% %% `ok = socket:send(ConnectedSocket, Data)' %% @end %%----------------------------------------------------------------------------- -spec send(Socket :: socket(), Data :: iodata()) -> - ok | {ok, Rest :: binary()} | {error, Reason :: term()}. + ok | {error, Reason :: term()} | {error, {Reason :: term(), RestData :: binary()}}. send(Socket, Data) when is_binary(Data) -> - ?MODULE:nif_send(Socket, Data); + send_all(Socket, Data); send(Socket, Data) -> - ?MODULE:nif_send(Socket, erlang:iolist_to_binary(Data)). + send_all(Socket, erlang:iolist_to_binary(Data)). %%----------------------------------------------------------------------------- %% @param Socket the socket %% @param Data the data to send %% @param Dest the destination to which to send the data -%% @returns `{ok, Rest}' if successful; `{error, Reason}', otherwise. +%% @returns `ok' if successful; `{error, Reason}', otherwise. %% @doc Send data on the specified socket to the specified destination. %% %% Note that this function will block until data is sent @@ -604,17 +614,20 @@ send(Socket, Data) -> %% intended recipient, and the data may not even have been sent %% over the network. %% +%% Transient send backpressure is retried internally, waiting for +%% the socket to become writable again, as with `send/2'. +%% %% Example: %% %% `ok = socket:sendto(ConnectedSocket, Data, Dest)' %% @end %%----------------------------------------------------------------------------- -spec sendto(Socket :: socket(), Data :: iodata(), Dest :: sockaddr()) -> - ok | {ok, Rest :: binary()} | {error, Reason :: term()}. + ok | {error, Reason :: term()}. sendto(Socket, Data, Dest) when is_binary(Data) -> - ?MODULE:nif_sendto(Socket, Data, Dest); + sendto_all(Socket, Data, Dest); sendto(Socket, Data, Dest) -> - ?MODULE:nif_sendto(Socket, erlang:iolist_to_binary(Data), Dest). + sendto_all(Socket, erlang:iolist_to_binary(Data), Dest). %%----------------------------------------------------------------------------- %% @param Socket the socket @@ -709,6 +722,10 @@ shutdown(_Socket, _How) -> nif_select_read(_Socket, _Ref) -> erlang:nif_error(undefined). +%% @private +nif_select_write(_Socket, _Ref) -> + erlang:nif_error(undefined). + %% @private nif_accept(_Socket) -> erlang:nif_error(undefined). @@ -738,3 +755,70 @@ nif_send(_Socket, _Data) -> %% @private nif_sendto(_Socket, _Data, _Dest) -> erlang:nif_error(undefined). + +%% @private +%% Send all of Data on Socket, retrying partial sends and transient send +%% backpressure ({error, eagain}) by waiting for the socket to become +%% writable again, until all data has been accepted or a real error occurs. +%% As in OTP, an error that follows a partial send is reported as +%% {error, {Reason, RestData}}, and as {error, Reason} if nothing was accepted. +send_all(Socket, Data) -> + send_all(Socket, Data, false). + +%% @private +send_all(Socket, Data, Partial) -> + case ?MODULE:nif_send(Socket, Data) of + ok -> + ok; + {ok, Rest} -> + send_all(Socket, Rest, true); + {error, eagain} -> + case wait_write(Socket) of + ok -> + send_all(Socket, Data, Partial); + {error, Reason} -> + send_error(Reason, Data, Partial) + end; + {error, Reason} -> + send_error(Reason, Data, Partial) + end. + +%% @private +send_error(Reason, RestData, true) -> + {error, {Reason, RestData}}; +send_error(Reason, _RestData, false) -> + {error, Reason}. + +%% @private +sendto_all(Socket, Data, Dest) -> + case ?MODULE:nif_sendto(Socket, Data, Dest) of + ok -> + ok; + {ok, Rest} -> + sendto_all(Socket, Rest, Dest); + {error, eagain} -> + case wait_write(Socket) of + ok -> + sendto_all(Socket, Data, Dest); + {error, _Reason} = Error -> + Error + end; + {error, _Reason} = Error -> + Error + end. + +%% @private +%% Block the calling process until Socket is writable (or closed). +wait_write(Socket) -> + Ref = erlang:make_ref(), + case ?MODULE:nif_select_write(Socket, Ref) of + ok -> + receive + {'$socket', Socket, select, Ref} -> + ok; + {'$socket', Socket, abort, {Ref, Reason}} -> + {error, Reason} + end; + {error, _Reason} = Error -> + Error + end. From cd0dd3368b46ad5a1b7daf8e8beb91a145341b0c Mon Sep 17 00:00:00 2001 From: harmon25 Date: Sat, 4 Jul 2026 17:06:20 -0400 Subject: [PATCH 5/7] Change socket:send callers to the new semantics socket:send/2 no longer returns {ok, Rest} or {error, eagain}. - ssl: wait for write readiness in the handshake, close notify, send and recv loops instead of busy looping on want_write. - socket_dist_controller: terminate the connection when socket:send/2 fails on a tick or on outgoing data, instead of ignoring the error. - Remove the {ok, Rest} branches that can no longer happen from epmd and the examples. Signed-off-by: harmon25 --- examples/erlang/rp2/picow_tcp_server.erl | 2 - examples/erlang/tcp_socket_server.erl | 2 - examples/erlang/udp_socket_client.erl | 4 -- libs/avm_network/src/epmd.erl | 2 +- libs/estdlib/src/gen_tcp_socket.erl | 8 +++- libs/estdlib/src/socket_dist_controller.erl | 24 +++++++--- libs/estdlib/src/ssl.erl | 49 +++++++++++++++++---- 7 files changed, 66 insertions(+), 25 deletions(-) diff --git a/examples/erlang/rp2/picow_tcp_server.erl b/examples/erlang/rp2/picow_tcp_server.erl index 053bf170c2..79ff6923a8 100644 --- a/examples/erlang/rp2/picow_tcp_server.erl +++ b/examples/erlang/rp2/picow_tcp_server.erl @@ -78,8 +78,6 @@ echo(ConnectedSocket) -> case socket:send(ConnectedSocket, Data) of ok -> io:format("All data was sent~n"); - {ok, Rest} -> - io:format("Some data was sent. Remaining: ~p~n", [Rest]); {error, Reason} -> io:format("An error occurred sending data: ~p~n", [Reason]) end, diff --git a/examples/erlang/tcp_socket_server.erl b/examples/erlang/tcp_socket_server.erl index 96fdb53a1e..029e9b6f9f 100644 --- a/examples/erlang/tcp_socket_server.erl +++ b/examples/erlang/tcp_socket_server.erl @@ -61,8 +61,6 @@ echo(ConnectedSocket) -> case socket:send(ConnectedSocket, Data) of ok -> io:format("All data was sent~n"); - {ok, Rest} -> - io:format("Some data was sent. Remaining: ~p~n", [Rest]); {error, Reason} -> io:format("An error occurred sending data: ~p~n", [Reason]) end, diff --git a/examples/erlang/udp_socket_client.erl b/examples/erlang/udp_socket_client.erl index 3389d31b11..a4b6df97ad 100644 --- a/examples/erlang/udp_socket_client.erl +++ b/examples/erlang/udp_socket_client.erl @@ -37,10 +37,6 @@ loop(Socket) -> io:format("Send packet ~p to ~p.~n", [ Data, Dest ]); - {ok, Rest} -> - io:format("Send packet ~p to ~p. Remaining: ~p~n", [ - Data, Dest, Rest - ]); {error, Reason} -> io:format("An error occurred sending a packet: ~p~n", [Reason]) end, diff --git a/libs/avm_network/src/epmd.erl b/libs/avm_network/src/epmd.erl index b012fe1eb7..fc9cc839cb 100644 --- a/libs/avm_network/src/epmd.erl +++ b/libs/avm_network/src/epmd.erl @@ -157,7 +157,7 @@ process_req( ) -> case socket:recv(Socket, 1, nowait) of {select, {select_info, recv, Ref}} -> - socket:send(Socket, <>), + ok = socket:send(Socket, <>), State#state{ recv_selects = [{Ref, client} | RecvSelects], clients = [ diff --git a/libs/estdlib/src/gen_tcp_socket.erl b/libs/estdlib/src/gen_tcp_socket.erl index 5ef50187a1..4ee3eb0e39 100644 --- a/libs/estdlib/src/gen_tcp_socket.erl +++ b/libs/estdlib/src/gen_tcp_socket.erl @@ -263,7 +263,13 @@ handle_call({send, Packet}, From, State) -> {send, Packet}, From, State ]), ?LOG_INFO("Sending packet"), - {reply, socket:send(State#state.socket, Packet), State}; + Reply = + case socket:send(State#state.socket, Packet) of + % gen_tcp:send/2 does not report the data that was not sent + {error, {Reason, _RestData}} -> {error, Reason}; + Result -> Result + end, + {reply, Reply, State}; handle_call({recv, Length, Timeout}, From, State) -> ?LOG_DEBUG("handle_call [~p], ~p, ~p]", [ {recv, Length, Timeout}, From, State diff --git a/libs/estdlib/src/socket_dist_controller.erl b/libs/estdlib/src/socket_dist_controller.erl index d97f7eb59b..3e66782025 100644 --- a/libs/estdlib/src/socket_dist_controller.erl +++ b/libs/estdlib/src/socket_dist_controller.erl @@ -142,8 +142,12 @@ handle_call(getstat, _From, #state{received = Received, sent = Sent} = State) -> {reply, {ok, Received, Sent, 0}, State}. handle_cast(tick, #state{socket = Socket, sent = Sent} = State) -> - socket:send(Socket, <<0:32>>), - {noreply, State#state{sent = Sent + 1}}; + case socket:send(Socket, <<0:32>>) of + ok -> + {noreply, State#state{sent = Sent + 1}}; + {error, Reason} -> + {stop, {send_error, Reason}, State} + end; handle_cast({handshake_complete, DHandle}, State0) -> ok = erlang:dist_ctrl_get_data_notification(DHandle), % We now need to get messages when data is coming. @@ -154,8 +158,10 @@ handle_cast({handshake_complete, DHandle}, State0) -> end. handle_info(dist_data, State0) -> - State1 = send_data_loop(State0), - {noreply, State1}; + case send_data_loop(State0) of + {ok, State1} -> {noreply, State1}; + {stop, Reason, State1} -> {stop, Reason, State1} + end; handle_info( {'$socket', Socket, select, SelectHandle}, #state{socket = Socket, select_handle = SelectHandle} = State0 @@ -209,10 +215,14 @@ send_data_loop(#state{dhandle = DHandle, socket = Socket, sent = Sent} = State) case erlang:dist_ctrl_get_data(DHandle) of none -> ok = erlang:dist_ctrl_get_data_notification(DHandle), - State; + {ok, State}; Data -> DataBin = ?PRE_PROCESS(Data), DataSize = byte_size(DataBin), - socket:send(Socket, <>), - send_data_loop(State#state{sent = Sent + 1}) + case socket:send(Socket, <>) of + ok -> + send_data_loop(State#state{sent = Sent + 1}); + {error, Reason} -> + {stop, {send_error, Reason}, State} + end end. diff --git a/libs/estdlib/src/ssl.erl b/libs/estdlib/src/ssl.erl index 46385b0662..90ffc39506 100644 --- a/libs/estdlib/src/ssl.erl +++ b/libs/estdlib/src/ssl.erl @@ -200,8 +200,13 @@ handshake_loop(SSLContext, Socket) -> Error end; want_write -> - % We're currrently missing non-blocking writes - handshake_loop(SSLContext, Socket); + case wait_write(Socket) of + ok -> + handshake_loop(SSLContext, Socket); + {error, _Reason} = Error -> + socket:close(Socket), + Error + end; {error, _Reason} = Error -> socket:close(Socket), Error @@ -253,8 +258,13 @@ close_notify_loop(SSLContext, Socket) -> Error end; want_write -> - % We're currrently missing non-blocking writes - close_notify_loop(SSLContext, Socket); + case wait_write(Socket) of + ok -> + close_notify_loop(SSLContext, Socket); + {error, _Reason} = Error -> + socket:close(Socket), + Error + end; {error, _Reason} = Error -> socket:close(Socket), Error @@ -283,8 +293,12 @@ send({SSLContext, Socket} = SSLSocket, Binary) -> Error end; want_write -> - % We're currrently missing non-blocking writes - send(SSLSocket, Binary); + case wait_write(Socket) of + ok -> + send(SSLSocket, Binary); + {error, _Reason} = Error -> + Error + end; {error, _Reason} = Error -> Error end. @@ -318,8 +332,27 @@ recv0({SSLContext, Socket} = SSLSocket, Length, Remaining, Acc) -> Error end; want_write -> - % We're currrently missing non-blocking writes - recv0(SSLSocket, Length, Remaining, Acc); + case wait_write(Socket) of + ok -> + recv0(SSLSocket, Length, Remaining, Acc); + {error, _Reason} = Error -> + Error + end; + {error, _Reason} = Error -> + Error + end. + +%% @private +wait_write(Socket) -> + Ref = erlang:make_ref(), + case socket:nif_select_write(Socket, Ref) of + ok -> + receive + {'$socket', Socket, select, Ref} -> + ok; + {'$socket', Socket, abort, {Ref, Reason}} -> + {error, Reason} + end; {error, _Reason} = Error -> Error end. From 5daaee11330706b75996425cf6eaaed958c56818 Mon Sep 17 00:00:00 2001 From: harmon25 Date: Sat, 4 Jul 2026 17:06:40 -0400 Subject: [PATCH 6/7] Add tests for socket write select and send - A write select on a socket with a full send buffer wakes up when the peer drains it, and socket:send/2 then completes transparently. - Closing an accepted socket selected by another process. It would have caught accepted sockets not initializing their selecting state, under valgrind. - A process blocked in recv and another blocked in send on the same socket are both aborted when a third process closes it. - When the process blocked in send dies, the one blocked in recv still receives data. Signed-off-by: harmon25 --- tests/libs/estdlib/test_tcp_socket.erl | 420 ++++++++++++++++++++++++- tests/libs/estdlib/test_udp_socket.erl | 13 + 2 files changed, 432 insertions(+), 1 deletion(-) diff --git a/tests/libs/estdlib/test_tcp_socket.erl b/tests/libs/estdlib/test_tcp_socket.erl index 4ceed655b9..2897214f17 100644 --- a/tests/libs/estdlib/test_tcp_socket.erl +++ b/tests/libs/estdlib/test_tcp_socket.erl @@ -31,15 +31,23 @@ test() -> ok = test_recv_nowait(), ok = test_accept_nowait(), ok = test_setopt_getopt(), + ok = test_close_accepted_socket_while_selecting(), + ok = test_send_empty(), case erlang:system_info(machine) of "ATOM" -> - ok = test_abandon_select(); + ok = test_abandon_select(), + ok = test_send_backpressure(), + ok = test_close_with_reader_and_writer(), + ok = test_writer_killed_reader_survives(), + ok = test_reader_cleanup_keeps_writer(), + ok = test_send_error_after_partial_send(); "BEAM" -> ok end, ok. -define(PACKET_SIZE, 7). +-define(LARGE_PAYLOAD_SIZE, (64 * 1024 * 1024)). test_echo_server() -> etest:flush_msg_queue(), @@ -570,5 +578,415 @@ test_abandon_select() -> erlang:garbage_collect(), ok. +%% +%% test_send_backpressure +%% +%% Exercises the write-select mechanism used internally by socket:send/2 to +%% wait for transient send backpressure (a full TCP send buffer) to clear, +%% instead of leaking {error, eagain} or {ok, Rest} to the caller. +%% + +test_send_backpressure() -> + etest:flush_msg_queue(), + + {ok, ListenSocket} = socket:open(inet, stream, tcp), + ok = socket:setopt(ListenSocket, {socket, reuseaddr}, true), + ok = socket:setopt(ListenSocket, {socket, linger}, #{onoff => true, linger => 0}), + ok = socket:bind(ListenSocket, #{family => inet, addr => loopback, port => 0}), + ok = socket:listen(ListenSocket), + {ok, #{port := Port}} = socket:sockname(ListenSocket), + + Self = self(), + Acceptor = spawn_link(fun() -> backpressure_acceptor(Self, ListenSocket) end), + + {ok, ClientSocket} = socket:open(inet, stream, tcp), + ok = try_connect(ClientSocket, Port, 10), + + ok = + receive + {server_socket, _ServerSocket} -> ok + after 5000 -> + error({timeout, waiting_for_server_socket}) + end, + + %% Fill the client's send buffer (the server never reads) using the raw + %% nif_send/2 directly, bypassing socket:send/2's automatic retry, until + %% we either observe a real {error, eagain} or give up after a generous + %% number of attempts. Different platforms/kernels size their socket + %% buffers differently, so we tolerate never observing backpressure + %% rather than failing the test outright. + Chunk = binary:copy(<<0>>, 65536), + {TotalSent, EAgainObserved} = fill_send_buffer(ClientSocket, Chunk, 0, 256), + + ok = + case EAgainObserved andalso TotalSent > 0 of + true -> + %% socket:nif_select_write/2 should let us wait until the + %% socket becomes writable again. Depending on how the + %% platform/kernel sizes and accounts for socket buffers, a + %% small partial drain on the peer may not be enough to + %% cross the low-water mark for writability, so we have the + %% acceptor drain everything (as a real reader normally + %% would) to reliably free up space. + Ref = erlang:make_ref(), + ok = socket:nif_select_write(ClientSocket, Ref), + + Acceptor ! {drain_all, self()}, + ok = + receive + {'$socket', ClientSocket, select, Ref} -> + ok + after 30000 -> + error({timeout, waiting_for_select_write}) + end, + + %% socket:send/2 should now transparently retry (internally + %% waiting for write-readiness as needed) and complete + %% successfully instead of returning {error, eagain} or + %% {ok, Rest}. + ok = socket:send(ClientSocket, Chunk), + ok = socket:close(ClientSocket), + ok = + receive + {drained_all, N} when is_integer(N) -> ok + after 30000 -> error({timeout, waiting_for_drain_all}) + end, + ok; + false -> + %% We never managed to fill the send buffer; nothing more to + %% verify on this platform. + Acceptor ! {drain_all, self()}, + ok = socket:close(ClientSocket), + ok = + receive + {drained_all, N} when is_integer(N) -> ok + after 30000 -> error({timeout, waiting_for_drain_all}) + end, + ok + end, + + ok. + +%% @private +%% Accepts a single connection and gives control of when to start reading +%% on it to the test process, so the test can deliberately stall the +%% receiver in order to build up send backpressure on the client side. +backpressure_acceptor(Owner, ListenSocket) -> + {ok, ServerSocket} = socket:accept(ListenSocket), + Owner ! {server_socket, ServerSocket}, + backpressure_acceptor_loop(Owner, ServerSocket), + ok = socket:close(ListenSocket). + +backpressure_acceptor_loop(Owner, ServerSocket) -> + receive + {drain_all, Owner} -> + Total = recv_until_closed(ServerSocket, 0), + Owner ! {drained_all, Total}, + ok = socket:close(ServerSocket) + after 60000 -> + ok = socket:close(ServerSocket) + end. + +recv_until_closed(Socket, Acc) -> + case socket:recv(Socket, 0, 30000) of + {ok, Data} -> + recv_until_closed(Socket, Acc + byte_size(Data)); + {error, closed} -> + Acc; + {error, timeout} -> + Acc + end. + +%% @private +%% Repeatedly calls the raw nif_send/2 (bypassing socket:send/2's automatic +%% retry) with the same chunk of data, until either an {error, eagain} is +%% observed (returns {TotalSent, true}) or MaxAttempts is reached without +%% ever observing backpressure (returns {TotalSent, false}). +fill_send_buffer(_Socket, _Chunk, TotalSent, 0) -> + {TotalSent, false}; +fill_send_buffer(Socket, Chunk, TotalSent, AttemptsLeft) -> + case socket:nif_send(Socket, Chunk) of + ok -> + fill_send_buffer(Socket, Chunk, TotalSent + byte_size(Chunk), AttemptsLeft - 1); + {ok, Rest} -> + Sent = byte_size(Chunk) - byte_size(Rest), + fill_send_buffer(Socket, Chunk, TotalSent + Sent, AttemptsLeft - 1); + {error, eagain} -> + {TotalSent, true}; + {error, Reason} -> + error({unexpected_send_error, Reason}) + end. + +%% +%% test_close_accepted_socket_while_selecting +%% +%% A process selects on an accepted socket and another process closes it. The +%% selecting state of a socket must be initialized whatever way the socket was +%% created, including by accept. +%% + +test_close_accepted_socket_while_selecting() -> + etest:flush_msg_queue(), + + {ok, ListenSocket} = socket:open(inet, stream, tcp), + ok = socket:setopt(ListenSocket, {socket, reuseaddr}, true), + ok = socket:bind(ListenSocket, #{family => inet, addr => loopback, port => 0}), + ok = socket:listen(ListenSocket), + {ok, #{port := Port}} = socket:sockname(ListenSocket), + + {ok, ClientSocket} = socket:open(inet, stream, tcp), + ok = try_connect(ClientSocket, Port, 10), + {ok, ServerSocket} = socket:accept(ListenSocket), + + Self = self(), + Reader = spawn_link(fun() -> + Self ! {recv_result, self(), socket:recv(ServerSocket, 0, 30000)} + end), + timer:sleep(200), + ok = socket:close(ServerSocket), + ok = expect_result(recv_result, Reader, {error, closed}), + + ok = socket:close(ClientSocket), + ok = socket:close(ListenSocket), + ok. + +%% +%% test_close_with_reader_and_writer +%% +%% A socket has one selecting process per direction: a process blocked in +%% recv and another one blocked in send, against a peer that does not read. +%% Closing the socket from a third process must abort both of them. +%% + +test_close_with_reader_and_writer() -> + etest:flush_msg_queue(), + case setup_full_send_buffer() of + {ok, Setup = #{client := Client}} -> + {Reader, Writer} = start_reader_and_writer(Client), + ok = socket:close(Client), + ok = expect_result(recv_result, Reader, {error, closed}), + ok = expect_send_closed(Writer), + teardown_full_send_buffer(Setup); + no_backpressure -> + ok + end. + +%% +%% test_writer_killed_reader_survives +%% +%% When the process blocked in send dies, only its direction is released: the +%% process blocked in recv must still receive data sent afterwards. +%% + +test_writer_killed_reader_survives() -> + etest:flush_msg_queue(), + case setup_full_send_buffer() of + {ok, Setup = #{client := Client, server := Server}} -> + {Reader, Writer} = start_reader_and_writer(Client), + MonitorRef = erlang:monitor(process, Writer), + exit(Writer, kill), + ok = + receive + {'DOWN', MonitorRef, process, Writer, killed} -> ok + after 5000 -> error({timeout, waiting_for_writer_down}) + end, + %% Give the dead writer's monitor time to fire. + timer:sleep(200), + ok = socket:send(Server, <<"hello">>), + ok = expect_result(recv_result, Reader, {ok, <<"hello">>}), + ok = socket:close(Client), + teardown_full_send_buffer(Setup); + no_backpressure -> + ok + end. + +%% +%% test_send_empty +%% +%% Sending an empty payload goes through the NIF as any other payload: it +%% succeeds on a connected socket and fails on a closed one. +%% + +test_send_empty() -> + etest:flush_msg_queue(), + + {ListenSocket, Port} = start_echo_server(0), + {ok, Socket} = socket:open(inet, stream, tcp), + ok = try_connect(Socket, Port, 10), + ok = socket:send(Socket, <<>>), + ok = socket:send(Socket, [<<>>]), + ok = socket:send(Socket, <<"echo:01">>), + {ok, <<"echo:01">>} = socket:recv(Socket, ?PACKET_SIZE, 5000), + ok = close_client_socket(Socket), + {error, _} = socket:send(Socket, <<>>), + {error, _} = socket:send(Socket, <<1>>), + + ok = close_listen_socket(ListenSocket), + ok. + +%% +%% test_reader_cleanup_keeps_writer +%% +%% A process blocked in recv and another one blocked in send, against a peer +%% that does not read. When the peer half-closes (shutdown write), the reader +%% gets {error, closed} and cleans up its select. This cleanup must not disown +%% the writer: when the peer then reads everything, the send completes. +%% + +test_reader_cleanup_keeps_writer() -> + etest:flush_msg_queue(), + case setup_full_send_buffer() of + {ok, Setup = #{client := Client, server := Server, sent := Sent}} -> + {Reader, Writer} = start_reader_and_writer(Client), + ok = socket:shutdown(Server, write), + ok = expect_result(recv_result, Reader, {error, closed}), + %% The writer is still blocked: read everything on the peer. + ok = socket:setopt(Server, {otp, rcvbuf}, 65536), + Total = Sent + ?LARGE_PAYLOAD_SIZE, + ok = recv_at_least(Server, Total), + ok = expect_result(send_result, Writer, ok), + ok = socket:close(Client), + teardown_full_send_buffer(Setup); + no_backpressure -> + ok + end. + +%% +%% test_send_error_after_partial_send +%% +%% As in OTP, an error following a partial send is reported with the data that +%% was not sent: {error, {Reason, RestData}}. Without any data accepted, it is +%% a plain {error, Reason}. +%% + +test_send_error_after_partial_send() -> + etest:flush_msg_queue(), + case setup_full_send_buffer() of + {ok, Setup = #{client := Client, server := Server}} -> + Self = self(), + Data = binary:copy(<<0>>, ?LARGE_PAYLOAD_SIZE), + Writer = spawn(fun() -> + Self ! {send_result, self(), socket:send(Client, Data)} + end), + timer:sleep(100), + %% Read some data on the peer so that the writer makes progress, + %% then let it block again. + ok = socket:setopt(Server, {otp, rcvbuf}, 65536), + ok = recv_at_least(Server, 1024 * 1024), + timer:sleep(200), + ok = socket:close(Client), + receive + {send_result, Writer, {error, {_Reason, Rest}}} when + is_binary(Rest), byte_size(Rest) > 0, byte_size(Rest) < ?LARGE_PAYLOAD_SIZE + -> + ok; + {send_result, Writer, Other} -> + error({unexpected_send_result, Other}) + after 5000 -> + error({timeout, send_result}) + end, + teardown_full_send_buffer(Setup); + no_backpressure -> + ok + end. + +%% @private +%% Receives at least Remaining bytes (a recv can return more than needed). +recv_at_least(_Socket, Remaining) when Remaining =< 0 -> + ok; +recv_at_least(Socket, Remaining) -> + case socket:recv(Socket, 0, 10000) of + {ok, Data} -> + recv_at_least(Socket, Remaining - byte_size(Data)); + {error, Reason} -> + error({recv_failed, Reason, Remaining}) + end. + +%% @private +%% Starts a process blocked in recv on Client, and then a process blocked in +%% send on it (the peer of Client does not read). +start_reader_and_writer(Client) -> + Self = self(), + Reader = spawn(fun() -> + Self ! {recv_result, self(), socket:recv(Client, 0, 30000)} + end), + timer:sleep(100), + %% Much more than any socket buffer can hold (buffers can grow after the + %% first eagain), so that the send cannot complete while the peer is not + %% reading. + Data = binary:copy(<<0>>, ?LARGE_PAYLOAD_SIZE), + Writer = spawn(fun() -> + Self ! {send_result, self(), socket:send(Client, Data)} + end), + timer:sleep(200), + {Reader, Writer}. + +%% @private +%% The writer may or may not have sent part of its data before being aborted, +%% so the error is either {error, closed} or {error, {closed, RestData}}. +expect_send_closed(Writer) -> + receive + {send_result, Writer, {error, closed}} -> ok; + {send_result, Writer, {error, {closed, Rest}}} when is_binary(Rest) -> ok; + {send_result, Writer, Other} -> error({unexpected_result, send_result, Other}) + after 5000 -> + error({timeout, send_result}) + end. + +%% @private +expect_result(Tag, Pid, Expected) -> + receive + {Tag, Pid, Expected} -> ok; + {Tag, Pid, Other} -> error({unexpected_result, Tag, Other}) + after 5000 -> + error({timeout, Tag}) + end. + +%% @private +%% Connects a client to a server that never reads, and fills the client send +%% buffer. Returns no_backpressure if the platform never reported eagain. +setup_full_send_buffer() -> + {ok, ListenSocket} = socket:open(inet, stream, tcp), + ok = socket:setopt(ListenSocket, {socket, reuseaddr}, true), + ok = socket:setopt(ListenSocket, {socket, linger}, #{onoff => true, linger => 0}), + ok = socket:bind(ListenSocket, #{family => inet, addr => loopback, port => 0}), + ok = socket:listen(ListenSocket), + {ok, #{port := Port}} = socket:sockname(ListenSocket), + + Self = self(), + Acceptor = spawn_link(fun() -> backpressure_acceptor(Self, ListenSocket) end), + + {ok, Client} = socket:open(inet, stream, tcp), + ok = try_connect(Client, Port, 10), + Server = + receive + {server_socket, ServerSocket} -> ServerSocket + after 5000 -> + error({timeout, waiting_for_server_socket}) + end, + + Chunk = binary:copy(<<0>>, 65536), + Setup = #{client => Client, server => Server, acceptor => Acceptor}, + case fill_send_buffer(Client, Chunk, 0, 256) of + {TotalSent, true} -> + {ok, Setup#{sent => TotalSent}}; + {_TotalSent, false} -> + teardown_full_send_buffer(Setup), + no_backpressure + end. + +%% @private +%% Lets the acceptor drain and close its side. The client socket must be +%% closed by the caller, or is closed here if it was not already. +teardown_full_send_buffer(#{client := Client, acceptor := Acceptor}) -> + _ = socket:close(Client), + Acceptor ! {drain_all, self()}, + ok = + receive + {drained_all, N} when is_integer(N) -> ok + after 30000 -> error({timeout, waiting_for_drain_all}) + end, + ok. + id(X) -> X. diff --git a/tests/libs/estdlib/test_udp_socket.erl b/tests/libs/estdlib/test_udp_socket.erl index ff57169669..fea90f1c93 100644 --- a/tests/libs/estdlib/test_udp_socket.erl +++ b/tests/libs/estdlib/test_udp_socket.erl @@ -24,6 +24,7 @@ test() -> ok = test_echo(), + ok = test_send_empty(), ok = test_buf_size(), ok = test_timeout(), ok = test_nowait(), @@ -50,6 +51,18 @@ test() -> -define(PACKET_SIZE, 7). +%% An empty datagram is a valid datagram: sendto/3 must hand it to the socket. +test_send_empty() -> + {ok, Receiver} = socket:open(inet, dgram, udp), + ok = socket:bind(Receiver, #{family => inet, addr => loopback, port => 0}), + {ok, #{port := Port}} = socket:sockname(Receiver), + {ok, Sender} = socket:open(inet, dgram, udp), + ok = socket:sendto(Sender, <<>>, #{family => inet, addr => loopback, port => Port}), + ok = socket:close(Sender), + {error, _} = socket:sendto(Sender, <<>>, #{family => inet, addr => loopback, port => Port}), + ok = socket:close(Receiver), + ok. + start_echo_server(Port) -> {ok, Socket} = socket:open(inet, dgram, udp), From c34bb11ac7a99416e1937895c00594fbc72577cd Mon Sep 17 00:00:00 2001 From: harmon25 Date: Sat, 4 Jul 2026 17:06:56 -0400 Subject: [PATCH 7/7] Document socket write select in the changelog Signed-off-by: harmon25 --- CHANGELOG.md | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index c5d8d94de8..0a314c9569 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [0.7.0-beta.1] - Unreleased +### Added +- Added `enif_select_write` and `ERL_NIF_SELECT_CANCEL` (both previously declared but not + implemented), and reworked `enif_select`/`enif_select_read`/`enif_select_write` so a resource + can hold independent pending read and write selects (each with its own ref/message/pid) on the + same event at the same time +- Added `socket:nif_select_write/2`; sockets now keep one selecting process per direction, so a + process blocked in `socket:send/2` no longer disturbs another process blocked in `socket:recv/2,3` + ### Fixed - Fixed maps decoded by `binary_to_term/1,2` and received over distribution keeping the encoder's key order, which broke `==` and made `=>` updates on an existing key produce @@ -23,6 +31,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed the ESP32 UART driver blocking the scheduler, until the task watchdog reset the chip, when a read had already taken the bytes of a pending data event, or when `uart:read/2` counted bytes that the other scheduler then took +- `socket:send/2,3` now correctly distinguishes transient send backpressure (lwIP `ERR_MEM` / BSD + `EAGAIN`|`EWOULDBLOCK`) from a closed connection at the NIF level, returning `{error, closed}` + (rather than `{ok, Data}`) when the peer has closed the connection, and waits for the socket to + become writable again and retries internally, so callers get `ok | {error, Reason}` (or + `{error, {Reason, RestData}}` when an error follows a partial send, as in OTP) and no longer + need to implement their own retry/backoff for partial sends or backpressure; `ssl:send/2`, + `ssl:recv/2` and the TLS handshake/close-notify loops likewise wait for write-readiness instead + of busy-looping on `want_write` +- Fixed Erlang distribution over sockets (`socket_dist_controller`) silently ignoring + `socket:send/2` errors on `tick` and outgoing distribution data; the connection is now + terminated instead of continuing to run against a broken socket ## [0.7.0-beta.0] - 2026-09-28