From 7780860446f45ee14c7b214ce493b0197e2ca719 Mon Sep 17 00:00:00 2001 From: Ken Van Hoeylandt Date: Sat, 29 Aug 2026 23:52:38 +0200 Subject: [PATCH] Fixes --- Modules/app-module/include/app/file.h | 6 ++++ .../app-module/private/app/private/fd_table.h | 9 +++++ Modules/app-module/source/fd_table.cpp | 33 +++++++++++------ Modules/app-module/source/io.cpp | 16 ++++++--- Modules/app-module/source/manager.cpp | 11 ++++++ Modules/app-module/source/stream.cpp | 19 ++++++++++ Modules/app-module/tests/source/io_test.cpp | 36 +++++++++++++++++++ 7 files changed, 115 insertions(+), 15 deletions(-) diff --git a/Modules/app-module/include/app/file.h b/Modules/app-module/include/app/file.h index f8dcf6bfe..91afbdcdc 100644 --- a/Modules/app-module/include/app/file.h +++ b/Modules/app-module/include/app/file.h @@ -33,6 +33,12 @@ struct AppFileOps { error_t (*await)(void* object, AppFileWait wait, TickType_t timeout); /** @return bitmask of APP_FILE_READABLE / APP_FILE_WRITABLE. */ uint32_t (*poll)(void* object); + /** Optional (may be NULL). Called by app_fd_table_get_and_retain() atomically with the fd + * lookup, before handing `object` to a caller about to dispatch into it, so a concurrent + * teardown of `object` can wait for that dispatch to finish instead of racing it. The caller + * calls release() once done, regardless of what the dispatched call returned. */ + void (*retain)(void* object); + void (*release)(void* object); }; /** A file-descriptor-table entry: an operations table paired with the object it operates on. */ diff --git a/Modules/app-module/private/app/private/fd_table.h b/Modules/app-module/private/app/private/fd_table.h index 68d010215..669f4372f 100644 --- a/Modules/app-module/private/app/private/fd_table.h +++ b/Modules/app-module/private/app/private/fd_table.h @@ -56,6 +56,15 @@ error_t app_fd_table_allocate(struct AppFdTable* table, const struct AppFileOps* /** @return true and fills @a out_file if @a fd is currently in use, false otherwise. */ bool app_fd_table_get(struct AppFdTable* table, int fd, struct AppFile* out_file); +/** + * Like app_fd_table_get(), but also calls the returned AppFile's AppFileOps::retain() (if any) + * while still holding `mutex`, atomically with the lookup. A caller that gets true back can + * therefore never have @a fd closed/torn down out from under it between this call and actually + * dispatching into the returned AppFile. Caller must call AppFileOps::release() once done + * dispatching. + */ +bool app_fd_table_get_and_retain(struct AppFdTable* table, int fd, struct AppFile* out_file); + /** * Closes @a fd (via AppFileOps::close()) and frees the slot for reuse. * @retval ERROR_NOT_FOUND @a fd is out of range or not currently in use diff --git a/Modules/app-module/source/fd_table.cpp b/Modules/app-module/source/fd_table.cpp index 56b7df1a1..acbfab6fe 100644 --- a/Modules/app-module/source/fd_table.cpp +++ b/Modules/app-module/source/fd_table.cpp @@ -12,6 +12,24 @@ bool fd_in_range(int fd) { return fd >= 0 && fd < APP_MAX_FDS; } +bool get_internal(AppFdTable* table, int fd, AppFile* out_file, bool retain) { + if (!fd_in_range(fd)) { + return false; + } + + mutex_lock(&table->mutex); + check(!table->shutting_down); + bool in_use = table->fds[fd] != nullptr; + if (in_use) { + *out_file = table->slots[fd]; + if (retain && out_file->ops->retain != nullptr) { + out_file->ops->retain(out_file->object); + } + } + mutex_unlock(&table->mutex); + return in_use; +} + } // namespace extern "C" { @@ -86,18 +104,11 @@ error_t app_fd_table_allocate(AppFdTable* table, const AppFileOps* ops, void* ob } bool app_fd_table_get(AppFdTable* table, int fd, AppFile* out_file) { - if (!fd_in_range(fd)) { - return false; - } + return get_internal(table, fd, out_file, /*retain=*/false); +} - mutex_lock(&table->mutex); - check(!table->shutting_down); - bool in_use = table->fds[fd] != nullptr; - if (in_use) { - *out_file = table->slots[fd]; - } - mutex_unlock(&table->mutex); - return in_use; +bool app_fd_table_get_and_retain(AppFdTable* table, int fd, AppFile* out_file) { + return get_internal(table, fd, out_file, /*retain=*/true); } error_t app_fd_table_close(AppFdTable* table, int fd) { diff --git a/Modules/app-module/source/io.cpp b/Modules/app-module/source/io.cpp index bd1f584f6..5885e60f8 100644 --- a/Modules/app-module/source/io.cpp +++ b/Modules/app-module/source/io.cpp @@ -49,11 +49,15 @@ ssize_t app_io_read(int fd, void* buffer, size_t size) { } AppFile file {}; - if (!app_fd_table_get(table, fd, &file)) { + if (!app_fd_table_get_and_retain(table, fd, &file)) { errno = EBADF; return -1; } - return file.ops->read(file.object, buffer, size); + ssize_t result = file.ops->read(file.object, buffer, size); + if (file.ops->release != nullptr) { + file.ops->release(file.object); + } + return result; } ssize_t app_io_write(int fd, const void* buffer, size_t size) { @@ -67,11 +71,15 @@ ssize_t app_io_write(int fd, const void* buffer, size_t size) { } AppFile file {}; - if (!app_fd_table_get(table, fd, &file)) { + if (!app_fd_table_get_and_retain(table, fd, &file)) { errno = EBADF; return -1; } - return file.ops->write(file.object, buffer, size); + ssize_t result = file.ops->write(file.object, buffer, size); + if (file.ops->release != nullptr) { + file.ops->release(file.object); + } + return result; } int app_io_close(int fd) { diff --git a/Modules/app-module/source/manager.cpp b/Modules/app-module/source/manager.cpp index dfa1a4256..5facfea8e 100644 --- a/Modules/app-module/source/manager.cpp +++ b/Modules/app-module/source/manager.cpp @@ -126,6 +126,12 @@ error_t start_internal(const char* id, AppInstanceId parent_instance_id, int arg error_t bind_result = app_stream_subscribe(bindings[i].stream, bindings[i].buffer, bindings[i].buffer_capacity, bindings[i].event_group, target_id, bindings[i].producer_fd); if (bind_result != ERROR_NONE) { LOG_E(TAG, "[instance %d] Failed to bind stream at fd %d: %s", target_id, bindings[i].producer_fd, error_to_string(bind_result)); + // Undo bindings[0..i): app_fd_table_teardown() below only closes each stream. It + // doesn't release the event bits app_stream_subscribe() claimed or destruct + // stream->internal.mutex; only app_stream_unsubscribe() does that. + for (size_t j = 0; j < i; j++) { + app_stream_unsubscribe(bindings[j].stream); + } mutex_lock(&ledger.mutex); app_fd_table_teardown(&ledger.instances[target_id].fd_table); ledger.instances.erase(target_id); @@ -137,6 +143,11 @@ error_t start_internal(const char* id, AppInstanceId parent_instance_id, int arg error_t error = app_scheduler_start(target_id, manifest->location, manifest->stack, argc, argv); if (error != ERROR_NONE) { + // Every binding succeeded before app_scheduler_start() failed. Unsubscribe all of them, + // same reasoning as the bind-failure path above. + for (size_t j = 0; j < binding_count; j++) { + app_stream_unsubscribe(bindings[j].stream); + } mutex_lock(&ledger.mutex); app_fd_table_teardown(&ledger.instances[target_id].fd_table); ledger.instances.erase(target_id); diff --git a/Modules/app-module/source/stream.cpp b/Modules/app-module/source/stream.cpp index c483bf72a..0ce5d5002 100644 --- a/Modules/app-module/source/stream.cpp +++ b/Modules/app-module/source/stream.cpp @@ -113,12 +113,31 @@ uint32_t stream_file_poll(void* object) { return bits; } +// Fused with app_fd_table_get_and_retain()'s own lock, so a caller that gets a live AppFile back +// is already counted here before it can be paused and raced by app_stream_unsubscribe()'s drain +// wait (see StreamOperationGuard above for the equivalent per-call bracketing). +void stream_file_retain(void* object) { + auto* stream = static_cast(object); + mutex_lock(&stream->internal.mutex); + stream->internal.active_operations++; + mutex_unlock(&stream->internal.mutex); +} + +void stream_file_release(void* object) { + auto* stream = static_cast(object); + mutex_lock(&stream->internal.mutex); + stream->internal.active_operations--; + mutex_unlock(&stream->internal.mutex); +} + constexpr AppFileOps STREAM_OPS = { .read = stream_file_read, .write = stream_file_write, .close = stream_file_close, .await = stream_file_await, .poll = stream_file_poll, + .retain = stream_file_retain, + .release = stream_file_release, }; } // namespace diff --git a/Modules/app-module/tests/source/io_test.cpp b/Modules/app-module/tests/source/io_test.cpp index c9b2d6d8d..9487852f0 100644 --- a/Modules/app-module/tests/source/io_test.cpp +++ b/Modules/app-module/tests/source/io_test.cpp @@ -170,3 +170,39 @@ TEST_CASE("a write blocked on a full stream wakes with an error once the consume task_event_group_destruct(&event_group); app_manager_remove("test.io.blocked"); } + +TEST_CASE("app_stream_unsubscribe is safe to call while a write is actively blocked") { + ensure_memory_loader_registered(); + g_blocked_writer_saw_error.store(false, std::memory_order_relaxed); + g_blocked_writer_done.store(false, std::memory_order_relaxed); + + AppManifest manifest { "test.io.unsub_race", "UnsubRace", APP_CATEGORY_USER, { APP_LOCATION_MEMORY, reinterpret_cast(blocked_writer_app_main) } }; + REQUIRE_EQ(app_manager_add(&manifest), ERROR_NONE); + + TaskEventGroup event_group {}; + task_event_group_construct(&event_group); + + uint8_t storage[4]; // smaller than the 10 bytes blocked_writer_app_main sends + AppStream child_stdout {}; + + AppStreamBinding binding { STDOUT_FILENO, &child_stdout, storage, sizeof(storage), &event_group }; + AppInstanceId child_id = 0; + REQUIRE_EQ(app_manager_start_with_streams("test.io.unsub_race", &binding, 1, &child_id), ERROR_NONE); + + // Give the child time to fill the 4-byte buffer and block inside app_io_write(), already + // dispatched through app_fd_table_get_and_retain() and currently waiting in + // app_stream_await(). This is the exact state app_stream_unsubscribe() must be safe to run + // against, with no prior app_stream_close() or wait for the child to stop first. + delay_millis(200); + REQUIRE_FALSE(g_blocked_writer_done.load(std::memory_order_acquire)); + + // Regression: unsubscribing directly here used to be able to destruct + // stream->internal.mutex while the blocked write above was still executing against it. + REQUIRE_EQ(app_stream_unsubscribe(&child_stdout), ERROR_NONE); + + REQUIRE(wait_for_state(child_id, APP_INSTANCE_STATE_STOPPED, 1000)); + CHECK(g_blocked_writer_saw_error.load(std::memory_order_acquire)); + + task_event_group_destruct(&event_group); + app_manager_remove("test.io.unsub_race"); +}