Fixes and improvements

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);
 

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

Modules/app-module/source/fd_table.cpp

@@ -2,6 +2,7 @@
 #include <app/private/fd_table.h>
 #include <app/private/null_device.h>
 
+#include <tactility/check.h>
 #include <tactility/concurrent/mutex.h>
 #include <tactility/error.h>
 
@@ -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;

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);

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<AppStream*>(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<AppStream*>(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<ssize_t>(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<ssize_t>(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<AppStream*>(object));
+    auto* stream = static_cast<AppStream*>(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<AppStream*>(object), wait, timeout);
+    auto* stream = static_cast<AppStream*>(object);
+    StreamOperationGuard guard(stream);
+    return app_stream_await(stream, wait, timeout);
 }
 
 uint32_t stream_file_poll(void* object) {
     auto* stream = static_cast<AppStream*>(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, &current) && 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;