diff --git a/Modules/app-module/include/app/stream.h b/Modules/app-module/include/app/stream.h index d9e4d9ec5..cb870523d 100644 --- a/Modules/app-module/include/app/stream.h +++ b/Modules/app-module/include/app/stream.h @@ -47,6 +47,10 @@ struct AppStream { uint32_t readable_bit; uint32_t writable_bit; bool closed; + /** Count of AppFileOps calls currently executing against this stream; app_stream_unsubscribe() + * waits for this to reach 0 before destructing `mutex`, since a call already blocked in + * app_stream_await() when unsubscribe starts only wakes (it doesn't vanish) when closed. */ + int active_operations; } internal; }; @@ -56,8 +60,10 @@ struct AppStream { * There is at most one subscriber for a given app fd. Subscribing over an existing one * atomically closes it first (any task blocked on it wakes; already-buffered bytes remain * readable until drained) and installs @a stream in its place. - * @param[in,out] stream caller-owned. Storage must remain valid until - * app_stream_unsubscribe()/app_stream_close() or app-exit teardown. + * @param[in,out] stream caller-owned. Storage must remain valid until app_stream_unsubscribe() + * returns. That is the only operation guaranteeing both that the fd-table binding is gone and + * that no AppFileOps call is still executing against @a stream; app_stream_close() alone does + * neither. * @param[in] buffer ring buffer storage; caller-owned, same validity requirement as @a stream. * @param[in] buffer_capacity size of @a buffer in bytes. * @param[in] event_group caller-owned; must outlive @a stream (i.e. be destructed only after @@ -70,8 +76,13 @@ struct AppStream { */ error_t app_stream_subscribe(struct AppStream* stream, void* buffer, size_t buffer_capacity, struct TaskEventGroup* event_group, AppInstanceId producer_id, int producer_fd); -/** Removes @a stream from whatever fd table it was installed in, restoring the null device there, - * and releases its two bits back to the event_group given to app_stream_subscribe(). */ +/** + * Removes @a stream's binding from whichever fd table it was installed in (further use of that + * fd then fails), wakes and waits for every AppFileOps call currently in flight against + * @a stream to finish, then releases its two bits back to the event_group given to + * app_stream_subscribe() and destructs its internal mutex. Only after this returns is + * @a stream's storage safe to free or reuse. + */ error_t app_stream_unsubscribe(struct AppStream* stream); /** @@ -88,12 +99,15 @@ error_t app_stream_await(struct AppStream* stream, AppFileWait wait, TickType_t /** Non-blocking: copies up to @a buffer_size currently-available bytes out of @a stream. */ size_t app_stream_read(struct AppStream* stream, void* buffer, size_t buffer_size); -/** Non-blocking: copies up to @a buffer_size bytes into @a stream, as much as currently fits. */ +/** Non-blocking: copies up to @a buffer_size bytes into @a stream, as much as currently fits. + * Copies nothing and returns 0 once @a stream is closed. */ size_t app_stream_write(struct AppStream* stream, const void* buffer, size_t buffer_size); /** * Marks @a stream closed: wakes any task blocked in app_stream_await(), and causes the other side - * to observe EOF (reader, once drained) or a write error (writer). + * to observe EOF (reader, once drained) or a write error (writer). Safe to call while another + * task may be blocked on @a stream, unlike app_stream_unsubscribe() (see its own doc); does not + * by itself make @a stream's storage safe to free. */ error_t app_stream_close(struct AppStream* stream); diff --git a/Modules/app-module/private/app/private/fd_table.h b/Modules/app-module/private/app/private/fd_table.h index 3ebe92891..68d010215 100644 --- a/Modules/app-module/private/app/private/fd_table.h +++ b/Modules/app-module/private/app/private/fd_table.h @@ -16,11 +16,18 @@ * copies the AppFile out under the lock first), since those calls belong to independently * synchronized objects (AppStream's own mutex/event group) and may run concurrently with a * close()/bind() replacing an unrelated slot. + * + * `shutting_down` is set under `mutex` by app_fd_table_teardown() before it destructs `mutex`, + * and checked (also under `mutex`) by every other entry point. Callers are expected to + * serialize against teardown externally (see app_stream_subscribe()/app_stream_unsubscribe()), + * so a check() failure here means that external synchronization broke, not a condition to + * handle gracefully. */ struct AppFdTable { struct AppFile slots[APP_MAX_FDS]; struct AppFile* fds[APP_MAX_FDS]; struct Mutex mutex; + bool shutting_down; }; #ifdef __cplusplus diff --git a/Modules/app-module/source/fd_table.cpp b/Modules/app-module/source/fd_table.cpp index 3376b3202..56b7df1a1 100644 --- a/Modules/app-module/source/fd_table.cpp +++ b/Modules/app-module/source/fd_table.cpp @@ -2,6 +2,7 @@ #include #include +#include #include #include @@ -24,16 +25,28 @@ void app_fd_table_construct(AppFdTable* table) { table->slots[fd] = *null_file; table->fds[fd] = &table->slots[fd]; } + table->shutting_down = false; mutex_construct(&table->mutex); } void app_fd_table_teardown(AppFdTable* table) { + AppFile closing[APP_MAX_FDS]; + int closing_count = 0; + + mutex_lock(&table->mutex); + table->shutting_down = true; for (int fd = 0; fd < APP_MAX_FDS; fd++) { if (table->fds[fd] != nullptr) { - table->slots[fd].ops->close(table->slots[fd].object); + closing[closing_count++] = table->slots[fd]; table->fds[fd] = nullptr; } } + mutex_unlock(&table->mutex); + + for (int i = 0; i < closing_count; i++) { + closing[i].ops->close(closing[i].object); + } + mutex_destruct(&table->mutex); } @@ -43,6 +56,7 @@ error_t app_fd_table_bind(AppFdTable* table, int fd, const AppFileOps* ops, void } mutex_lock(&table->mutex); + check(!table->shutting_down); AppFile old = table->slots[fd]; bool had_old = table->fds[fd] != nullptr; table->slots[fd] = { .ops = ops, .object = object }; @@ -57,6 +71,7 @@ error_t app_fd_table_bind(AppFdTable* table, int fd, const AppFileOps* ops, void error_t app_fd_table_allocate(AppFdTable* table, const AppFileOps* ops, void* object, int* out_fd) { mutex_lock(&table->mutex); + check(!table->shutting_down); for (int fd = 3; fd < APP_MAX_FDS; fd++) { if (table->fds[fd] == nullptr) { table->slots[fd] = { .ops = ops, .object = object }; @@ -76,6 +91,7 @@ bool app_fd_table_get(AppFdTable* table, int fd, AppFile* out_file) { } mutex_lock(&table->mutex); + check(!table->shutting_down); bool in_use = table->fds[fd] != nullptr; if (in_use) { *out_file = table->slots[fd]; @@ -90,6 +106,7 @@ error_t app_fd_table_close(AppFdTable* table, int fd) { } mutex_lock(&table->mutex); + check(!table->shutting_down); bool in_use = table->fds[fd] != nullptr; AppFile old = table->slots[fd]; table->fds[fd] = nullptr; diff --git a/Modules/app-module/source/manager.cpp b/Modules/app-module/source/manager.cpp index 4bd470819..dfa1a4256 100644 --- a/Modules/app-module/source/manager.cpp +++ b/Modules/app-module/source/manager.cpp @@ -94,6 +94,11 @@ char** copy_arguments(int argc, const char* const argv[]) { // new instance's fd table before app_scheduler_start() is called, so they're in place before its // task begins executing (see app_manager_start_with_streams()). error_t start_internal(const char* id, AppInstanceId parent_instance_id, int argc, char* argv[], const AppStreamBinding* bindings, size_t binding_count, AppInstanceId* out_app_instance_id) { + if (binding_count != 0 && bindings == nullptr) { + app_ledger_free_arguments(argc, argv); + return ERROR_INVALID_ARGUMENT; + } + auto& ledger = app_ledger(); mutex_lock(&ledger.mutex); diff --git a/Modules/app-module/source/stream.cpp b/Modules/app-module/source/stream.cpp index 6abbb40ba..c483bf72a 100644 --- a/Modules/app-module/source/stream.cpp +++ b/Modules/app-module/source/stream.cpp @@ -20,8 +20,33 @@ bool is_writable_locked(AppStream* stream) { return stream->internal.buffer.count < stream->internal.buffer.capacity || stream->internal.closed; } +// Brackets one AppFileOps call against `stream` for app_stream_unsubscribe()'s drain wait: a +// call already blocked in app_stream_await() when unsubscribe starts only wakes, it doesn't +// vanish, so unsubscribe must know it's still running before destructing stream->internal.mutex. +class StreamOperationGuard { +public: + explicit StreamOperationGuard(AppStream* stream) : stream_(stream) { + mutex_lock(&stream_->internal.mutex); + stream_->internal.active_operations++; + mutex_unlock(&stream_->internal.mutex); + } + + ~StreamOperationGuard() { + mutex_lock(&stream_->internal.mutex); + stream_->internal.active_operations--; + mutex_unlock(&stream_->internal.mutex); + } + + StreamOperationGuard(const StreamOperationGuard&) = delete; + StreamOperationGuard& operator=(const StreamOperationGuard&) = delete; + +private: + AppStream* stream_; +}; + ssize_t stream_file_read(void* object, void* buffer, size_t size) { auto* stream = static_cast(object); + StreamOperationGuard guard(stream); while (true) { if (app_stream_await(stream, APP_FILE_WAIT_READABLE, portMAX_DELAY) != ERROR_NONE) { return -1; @@ -43,28 +68,39 @@ ssize_t stream_file_read(void* object, void* buffer, size_t size) { ssize_t stream_file_write(void* object, const void* buffer, size_t size) { auto* stream = static_cast(object); + StreamOperationGuard guard(stream); if (app_stream_await(stream, APP_FILE_WAIT_WRITABLE, portMAX_DELAY) != ERROR_NONE) { return -1; } + // app_stream_write() itself refuses to copy anything once closed (checked under the same + // lock as the copy, so a close() racing right after this await() can't slip bytes in), and + // reports that as 0, indistinguishable here from "woke writable, nothing to copy" without + // checking closed separately. + size_t written = app_stream_write(stream, buffer, size); + if (written > 0) { + return static_cast(written); + } mutex_lock(&stream->internal.mutex); bool is_closed = stream->internal.closed; mutex_unlock(&stream->internal.mutex); - if (is_closed) { - return -1; // broken pipe: woke writable only because the consumer closed - } - return static_cast(app_stream_write(stream, buffer, size)); + return is_closed ? -1 : 0; // broken pipe, or a spurious wake with nothing to copy } error_t stream_file_close(void* object) { - return app_stream_close(static_cast(object)); + auto* stream = static_cast(object); + StreamOperationGuard guard(stream); + return app_stream_close(stream); } error_t stream_file_await(void* object, AppFileWait wait, TickType_t timeout) { - return app_stream_await(static_cast(object), wait, timeout); + auto* stream = static_cast(object); + StreamOperationGuard guard(stream); + return app_stream_await(stream, wait, timeout); } uint32_t stream_file_poll(void* object) { auto* stream = static_cast(object); + StreamOperationGuard guard(stream); mutex_lock(&stream->internal.mutex); uint32_t bits = 0; if (is_readable_locked(stream)) { @@ -98,17 +134,6 @@ error_t app_stream_subscribe(AppStream* stream, void* buffer, size_t buffer_capa return ERROR_OUT_OF_RANGE; } - auto& ledger = app_ledger(); - mutex_lock(&ledger.mutex); - auto iterator = ledger.instances.find(producer_id); - if (iterator == ledger.instances.end()) { - mutex_unlock(&ledger.mutex); - return ERROR_NOT_FOUND; - } - TaskHandle_t producer_task = iterator->second.task; - AppFdTable* table = &iterator->second.fd_table; - mutex_unlock(&ledger.mutex); - uint32_t readable_bit; error_t claim_result = task_event_group_claim_bit(event_group, &readable_bit); if (claim_result != ERROR_NONE) { @@ -123,7 +148,6 @@ error_t app_stream_subscribe(AppStream* stream, void* buffer, size_t buffer_capa stream->internal.producer_id = producer_id; stream->internal.producer_fd = producer_fd; - stream->internal.producer_task = producer_task; stream->internal.event_group = event_group; stream->internal.readable_bit = readable_bit; stream->internal.writable_bit = writable_bit; @@ -133,9 +157,27 @@ error_t app_stream_subscribe(AppStream* stream, void* buffer, size_t buffer_capa stream->internal.buffer.write_pos = 0; stream->internal.buffer.count = 0; stream->internal.closed = false; + stream->internal.active_operations = 0; mutex_construct(&stream->internal.mutex); - error_t bind_result = app_fd_table_bind(table, producer_fd, &STREAM_OPS, stream); + // Held across the fd-table lookup and bind so this can't race app_fd_table_teardown(): + // both teardown call sites (scheduler.cpp, manager.cpp) hold this same ledger mutex for + // their whole teardown call, and app_fd_table_bind()/close()/get() check for it (see + // fd_table.h's AppFdTable::shutting_down). + auto& ledger = app_ledger(); + mutex_lock(&ledger.mutex); + auto iterator = ledger.instances.find(producer_id); + if (iterator == ledger.instances.end()) { + mutex_unlock(&ledger.mutex); + mutex_destruct(&stream->internal.mutex); + task_event_group_release_bit(event_group, readable_bit); + task_event_group_release_bit(event_group, writable_bit); + return ERROR_NOT_FOUND; + } + stream->internal.producer_task = iterator->second.task; + error_t bind_result = app_fd_table_bind(&iterator->second.fd_table, producer_fd, &STREAM_OPS, stream); + mutex_unlock(&ledger.mutex); + if (bind_result != ERROR_NONE) { mutex_destruct(&stream->internal.mutex); task_event_group_release_bit(event_group, readable_bit); @@ -145,20 +187,35 @@ error_t app_stream_subscribe(AppStream* stream, void* buffer, size_t buffer_capa } error_t app_stream_unsubscribe(AppStream* stream) { + // Held across the fd-table lookup, get, and close. See app_stream_subscribe()'s own + // comment on why this must stay exclusive with app_fd_table_teardown(). auto& ledger = app_ledger(); mutex_lock(&ledger.mutex); auto iterator = ledger.instances.find(stream->internal.producer_id); - AppFdTable* table = (iterator != ledger.instances.end()) ? &iterator->second.fd_table : nullptr; - mutex_unlock(&ledger.mutex); - - if (table != nullptr) { + if (iterator != ledger.instances.end()) { + AppFdTable* table = &iterator->second.fd_table; AppFile current {}; if (app_fd_table_get(table, stream->internal.producer_fd, ¤t) && current.object == stream) { app_fd_table_close(table, stream->internal.producer_fd); // -> stream_file_close() -> app_stream_close() } } + mutex_unlock(&ledger.mutex); + + app_stream_close(stream); // idempotent; wakes anyone already blocked in app_stream_await() + + // Waking a blocked call doesn't mean it has finished. It still has to get scheduled, notice + // it's closed, and return. Wait for that before destructing mutex/bits below, mirroring + // scheduler.cpp's reap_self()/reaper_task_main() waiting out a task before freeing its stack. + while (true) { + mutex_lock(&stream->internal.mutex); + int active_operations = stream->internal.active_operations; + mutex_unlock(&stream->internal.mutex); + if (active_operations == 0) { + break; + } + taskYIELD(); + } - app_stream_close(stream); // idempotent; covers the case where the producer already exited task_event_group_release_bit(stream->internal.event_group, stream->internal.readable_bit); task_event_group_release_bit(stream->internal.event_group, stream->internal.writable_bit); mutex_destruct(&stream->internal.mutex); @@ -199,6 +256,10 @@ size_t app_stream_read(AppStream* stream, void* buffer, size_t buffer_size) { size_t app_stream_write(AppStream* stream, const void* buffer, size_t buffer_size) { mutex_lock(&stream->internal.mutex); + if (stream->internal.closed) { + mutex_unlock(&stream->internal.mutex); + return 0; + } AppStreamBuffer& ring = stream->internal.buffer; size_t available = ring.capacity - ring.count; size_t to_copy = buffer_size < available ? buffer_size : available;