From 6e5e35610b959c08f4a13c2550dafe2d19b4ecaa Mon Sep 17 00:00:00 2001 From: Ken Van Hoeylandt Date: Sun, 30 Aug 2026 12:07:40 +0200 Subject: [PATCH] Implement app streams: stdio for apps (#640) Implement file IO for applications to facilitate text input/output capturing. Apps can now launch apps and: - Read their stdio output - Write to their stdio input --- CMakeLists.txt | 4 + Modules/app-module/include/app/file.h | 52 +++ Modules/app-module/include/app/io.h | 35 ++ Modules/app-module/include/app/manager.h | 25 ++ Modules/app-module/include/app/stream.h | 115 +++++++ .../app-module/private/app/private/fd_table.h | 98 ++++++ .../app-module/private/app/private/ledger.h | 5 + .../private/app/private/null_device.h | 18 + .../private/app/private/stream_internal.h | 15 + Modules/app-module/source/fd_table.cpp | 151 +++++++++ Modules/app-module/source/io.cpp | 106 ++++++ Modules/app-module/source/manager.cpp | 51 ++- Modules/app-module/source/module.cpp | 14 + Modules/app-module/source/null_device.cpp | 45 +++ Modules/app-module/source/scheduler.cpp | 17 + Modules/app-module/source/stream.cpp | 313 ++++++++++++++++++ Modules/app-module/tests/source/io_test.cpp | 294 ++++++++++++++++ .../app-module/tests/source/manager_test.cpp | 10 +- .../app-module/tests/source/stream_test.cpp | 158 +++++++++ Tactility/Source/AppStdioWrap.cpp | 24 ++ 20 files changed, 1540 insertions(+), 10 deletions(-) create mode 100644 Modules/app-module/include/app/file.h create mode 100644 Modules/app-module/include/app/io.h create mode 100644 Modules/app-module/include/app/stream.h create mode 100644 Modules/app-module/private/app/private/fd_table.h create mode 100644 Modules/app-module/private/app/private/null_device.h create mode 100644 Modules/app-module/private/app/private/stream_internal.h create mode 100644 Modules/app-module/source/fd_table.cpp create mode 100644 Modules/app-module/source/io.cpp create mode 100644 Modules/app-module/source/null_device.cpp create mode 100644 Modules/app-module/source/stream.cpp create mode 100644 Modules/app-module/tests/source/io_test.cpp create mode 100644 Modules/app-module/tests/source/stream_test.cpp create mode 100644 Tactility/Source/AppStdioWrap.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 84818cca..6b184e33 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -64,6 +64,10 @@ if (DEFINED ENV{ESP_IDF_VERSION}) idf_build_set_property(LINK_OPTIONS "-Wl,--wrap=esp_panic_handler" APPEND) endif () + idf_build_set_property(LINK_OPTIONS "-Wl,--wrap=read" APPEND) + idf_build_set_property(LINK_OPTIONS "-Wl,--wrap=write" APPEND) + idf_build_set_property(LINK_OPTIONS "-Wl,--wrap=close" APPEND) + idf_build_set_property(LINK_OPTIONS "-Wl,--wrap=lv_button_create" APPEND) idf_build_set_property(LINK_OPTIONS "-Wl,--wrap=lv_dropdown_create" APPEND) idf_build_set_property(LINK_OPTIONS "-Wl,--wrap=lv_list_create" APPEND) diff --git a/Modules/app-module/include/app/file.h b/Modules/app-module/include/app/file.h new file mode 100644 index 00000000..91afbdcd --- /dev/null +++ b/Modules/app-module/include/app/file.h @@ -0,0 +1,52 @@ +// SPDX-License-Identifier: Apache-2.0 +#pragma once + +#include +#include +#include + +#include +#include + +#ifdef __cplusplus +extern "C" { +#endif + +/** Which readiness condition to block for in AppFileOps::await(). */ +typedef enum { + APP_FILE_WAIT_READABLE, + APP_FILE_WAIT_WRITABLE, +} AppFileWait; + +/** Bits returned by AppFileOps::poll(). */ +#define APP_FILE_READABLE (1u << 0) +#define APP_FILE_WRITABLE (1u << 1) + +/** + * Operations table for one kind of file-like object (stream, file, device, ...). Every function + * takes the type-erased `object` a concrete AppFile instance was constructed with. + */ +struct AppFileOps { + ssize_t (*read)(void* object, void* buffer, size_t size); + ssize_t (*write)(void* object, const void* buffer, size_t size); + error_t (*close)(void* object); + 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. */ +struct AppFile { + const struct AppFileOps* ops; + void* object; +}; + +#ifdef __cplusplus +} +#endif diff --git a/Modules/app-module/include/app/io.h b/Modules/app-module/include/app/io.h new file mode 100644 index 00000000..76206ea7 --- /dev/null +++ b/Modules/app-module/include/app/io.h @@ -0,0 +1,35 @@ +// SPDX-License-Identifier: Apache-2.0 +#pragma once + +#include +#include + +#ifdef __cplusplus +extern "C" { +#endif + +#define STDIN_FILENO 0 +#define STDOUT_FILENO 1 +#define STDERR_FILENO 2 + +/** Fixed size of an app instance's fd table. FD allocation uses the lowest unused index >= 3. */ +#define APP_MAX_FDS 16 + +/** + * FD-table dispatch for read()/write()/close(). When the calling task belongs to a running app + * instance and @a fd is one that instance bound/allocated itself, it's looked up in that + * instance's own fd table. Otherwise (no app instance, e.g. a kernel service task; or an app + * instance's own fd that was never bound through app-module, e.g. a real file from fopen()/ + * open(), which this layer never intercepts) @a fd is a real underlying fd and the call falls + * through to the real syscall unchanged. + * @return number of bytes transferred, 0 on EOF (read) or a closed peer (write), or -1 with + * errno set on failure. This is the fd-layer translation of AppFileOps::read()/write()'s ssize_t + * and AppFileOps::close()'s error_t. + */ +ssize_t app_io_read(int fd, void* buffer, size_t size); +ssize_t app_io_write(int fd, const void* buffer, size_t size); +int app_io_close(int fd); + +#ifdef __cplusplus +} +#endif diff --git a/Modules/app-module/include/app/manager.h b/Modules/app-module/include/app/manager.h index b5427631..6b4de946 100644 --- a/Modules/app-module/include/app/manager.h +++ b/Modules/app-module/include/app/manager.h @@ -3,6 +3,7 @@ #include #include +#include #include @@ -86,6 +87,30 @@ error_t app_manager_start_with_parameters(const char* id, int argc, const char* */ error_t app_manager_start_for_result(const char* id, AppInstanceId parent_instance_id, int argc, const char* const argv[], AppInstanceId* out_app_instance_id); +/** One fd-to-stream binding for app_manager_start_with_streams(). Every field is passed through + * to app_stream_subscribe() as-is; see its own doc for the ownership contracts. */ +struct AppStreamBinding { + int producer_fd; + struct AppStream* stream; + void* buffer; + size_t buffer_capacity; + struct TaskEventGroup* event_group; +}; + +/** + * Same as app_manager_start(), but installs @a bindings into the new instance's fd table before + * its task begins executing (e.g. a child's stdio, piped through parent-owned AppStreams; see + * app/stream.h). Writes the new instance's id into each bound stream's producer_id itself, since + * the caller cannot know it in advance. + * @param[in] bindings @a binding_count entries; each stream and buffer must stay alive (see + * app_stream_subscribe()) until unsubscribed or the child exits. + * @retval ERROR_NOT_FOUND no manifest with this id is registered, or no AppLoaderApi is registered + * @retval ERROR_OUT_OF_RANGE a binding's producer_fd is out of range + * @retval ERROR_RESOURCE a binding's event_group has no free bits left to claim + * @retval ERROR_NONE on success + */ +error_t app_manager_start_with_streams(const char* id, const struct AppStreamBinding* bindings, size_t binding_count, AppInstanceId* out_app_instance_id); + /** * Stop an app instance permanently. Emits APP_EVENT_CLOSE and bound-waits for its task to exit * if it was running. diff --git a/Modules/app-module/include/app/stream.h b/Modules/app-module/include/app/stream.h new file mode 100644 index 00000000..bf5127f0 --- /dev/null +++ b/Modules/app-module/include/app/stream.h @@ -0,0 +1,115 @@ +// SPDX-License-Identifier: Apache-2.0 +#pragma once + +#include +#include + +#include +#include + +#include +#include +#include +#include +#include + +#ifdef __cplusplus +extern "C" { +#endif + +/** @warning This is internal data. Do not read/write to it directly. */ +struct AppStreamBuffer { + uint8_t* data; + size_t capacity; + size_t read_pos; + size_t write_pos; + size_t count; +}; + +/** + * Buffered byte-oriented communication between one producer task and one consumer task. The + * consumer owns the AppStream object and its lifetime. See app_stream_subscribe(). + * + * @warning This is internal data. Do not read/write to it directly. + */ +struct AppStream { + struct AppStreamBuffer buffer; + AppInstanceId producer_id; + TaskHandle_t producer_task; + /** fd this stream is installed at in producer_id's fd table; set by + * app_stream_subscribe()/app_manager_start_with_streams(), used by + * app_stream_unsubscribe() to find it again. */ + int producer_fd; + + struct Mutex mutex; + /** Caller-owned group readiness is signalled on; see app_stream_subscribe(). */ + struct TaskEventGroup* event_group; + 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; +}; + +/** + * Registers @a stream as the file-like object bound to @a producer_id's fd table at + * @a producer_fd, claiming two bits from @a event_group for its readable/writable readiness. + * 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() + * 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 + * app_stream_unsubscribe()). Lets a task wait on this stream's readiness together with other + * event sources sharing the same group via task_event_group_wait()/task_event_group_wait_any(). + * @retval ERROR_NOT_FOUND no instance with this id is running + * @retval ERROR_OUT_OF_RANGE @a producer_fd is out of range + * @retval ERROR_RESOURCE @a event_group has no free bits left to claim + * @retval ERROR_NONE on success + */ +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'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); + +/** + * Blocks until @a stream becomes readable/writable (per @a wait) or @a timeout elapses. Checks + * the current state directly before blocking, so a permanently-true condition (e.g. closed) + * keeps returning immediately on every call, even though the underlying event bit itself is + * one-shot (task_event_group_wait() clears a matched bit on exit). + * @retval ERROR_NONE the condition is true + * @retval ERROR_TIMEOUT @a timeout elapsed + * @retval ERROR_ISR_STATUS called from an ISR + */ +error_t app_stream_await(struct AppStream* stream, AppFileWait wait, TickType_t timeout); + +/** 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. + * 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). 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); + +#ifdef __cplusplus +} +#endif diff --git a/Modules/app-module/private/app/private/fd_table.h b/Modules/app-module/private/app/private/fd_table.h new file mode 100644 index 00000000..d1ee41d9 --- /dev/null +++ b/Modules/app-module/private/app/private/fd_table.h @@ -0,0 +1,98 @@ +// SPDX-License-Identifier: Apache-2.0 +#pragma once + +#include +#include + +#include + +#include +#include + +/** + * One fd's worth of state in an AppFdTable. `file` is valid only while `in_use` is true. + * `ever_used` is set once `in_use` is ever set true (including the default stdio binding at + * construct time) and never cleared, even once `in_use` goes back to false on close. This is + * what lets app_fd_table_is_app_owned() distinguish "this fd number belongs to us, it's just + * currently closed" (an app-level EBADF) from "app-module has never touched this fd number" (a + * real underlying fd, e.g. from fopen()/open(), that a caller should fall through on). + * + * @warning This is internal data. Do not read/write to it directly. + */ +struct AppFdSlot { + struct AppFile file; + bool in_use; + bool ever_used; +}; + +/** + * Every app instance's fd table (see AppInstanceRecord). `mutex` guards `slots` only; it is + * never held while calling into an AppFile's ops (dispatch 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. + * + * @warning This is internal data. Do not read/write to it directly. + */ +struct AppFdTable { + struct AppFdSlot slots[APP_MAX_FDS]; + struct Mutex mutex; + bool shutting_down; +}; + +#ifdef __cplusplus +extern "C" { +#endif + +/** Initializes every fd to the null device (see app/private/null_device.h) and constructs `mutex`. */ +void app_fd_table_construct(struct AppFdTable* table); + +/** Closes every still-open fd (via AppFileOps::close()) and destructs `mutex`. */ +void app_fd_table_teardown(struct AppFdTable* table); + +/** + * Installs {ops, object} at @a fd, closing whatever was previously there first (the null device + * counts as "previously there" for fds 0-2, so this doubles as their initial stdio binding). + * @retval ERROR_OUT_OF_RANGE @a fd is outside [0, APP_MAX_FDS) + */ +error_t app_fd_table_bind(struct AppFdTable* table, int fd, const struct AppFileOps* ops, void* object); + +/** + * Installs {ops, object} at the lowest unused fd >= 3. + * @retval ERROR_RESOURCE the table is full + */ +error_t app_fd_table_allocate(struct AppFdTable* table, const struct AppFileOps* ops, void* object, int* out_fd); + +/** @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 + */ +error_t app_fd_table_close(struct AppFdTable* table, int fd); + +/** + * @return true if @a fd is currently in use, or was in the past (even if since closed). This is + * whether the table claims @a fd as one of its own, regardless of its current state. False means + * @a fd is a real underlying fd this table has never bound or allocated. + */ +bool app_fd_table_is_app_owned(struct AppFdTable* table, int fd); + +#ifdef __cplusplus +} +#endif diff --git a/Modules/app-module/private/app/private/ledger.h b/Modules/app-module/private/app/private/ledger.h index 00392d60..52efa5ae 100644 --- a/Modules/app-module/private/app/private/ledger.h +++ b/Modules/app-module/private/app/private/ledger.h @@ -3,6 +3,7 @@ #include #include +#include #include #include @@ -49,6 +50,10 @@ struct AppInstanceRecord { /** This instance's completion signal - see AppCompletionSignal. Set once by * app_scheduler_start(), never reassigned. */ AppCompletionSignal* completion = nullptr; + + /** This instance's fd table. Constructed by start_internal() before insertion into + * AppLedger::instances, torn down (every open fd closed) when the instance's task exits. */ + AppFdTable fd_table {}; }; struct AppLedger { diff --git a/Modules/app-module/private/app/private/null_device.h b/Modules/app-module/private/app/private/null_device.h new file mode 100644 index 00000000..21ea76ff --- /dev/null +++ b/Modules/app-module/private/app/private/null_device.h @@ -0,0 +1,18 @@ +// SPDX-License-Identifier: Apache-2.0 +#pragma once + +#include + +#ifdef __cplusplus +extern "C" { +#endif + +/** + * @return the shared null-device AppFile: write() discards data and reports success, read() + * reports EOF, close() is a no-op, and it is always reported readable/writable. + */ +const struct AppFile* app_null_file(void); + +#ifdef __cplusplus +} +#endif diff --git a/Modules/app-module/private/app/private/stream_internal.h b/Modules/app-module/private/app/private/stream_internal.h new file mode 100644 index 00000000..e0f7d7bc --- /dev/null +++ b/Modules/app-module/private/app/private/stream_internal.h @@ -0,0 +1,15 @@ +// SPDX-License-Identifier: Apache-2.0 +#pragma once + +#include + +#ifdef __cplusplus +extern "C" { +#endif + +/** @return the AppFileOps used for stream-backed fd table entries (see app_stream_subscribe()). */ +const struct AppFileOps* app_stream_ops(void); + +#ifdef __cplusplus +} +#endif diff --git a/Modules/app-module/source/fd_table.cpp b/Modules/app-module/source/fd_table.cpp new file mode 100644 index 00000000..0a016f2c --- /dev/null +++ b/Modules/app-module/source/fd_table.cpp @@ -0,0 +1,151 @@ +// SPDX-License-Identifier: Apache-2.0 +#include +#include + +#include +#include +#include + +namespace { + +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); + AppFdSlot& slot = table->slots[fd]; + bool in_use = slot.in_use; + if (in_use) { + *out_file = slot.file; + if (retain && out_file->ops->retain != nullptr) { + out_file->ops->retain(out_file->object); + } + } + mutex_unlock(&table->mutex); + return in_use; +} + +} // namespace + +extern "C" { + +void app_fd_table_construct(AppFdTable* table) { + const AppFile* null_file = app_null_file(); + for (int fd = 0; fd < APP_MAX_FDS; fd++) { + table->slots[fd].in_use = false; + table->slots[fd].ever_used = false; + } + for (int fd = STDIN_FILENO; fd <= STDERR_FILENO; fd++) { + table->slots[fd].file = *null_file; + table->slots[fd].in_use = true; + table->slots[fd].ever_used = true; + } + 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->slots[fd].in_use) { + closing[closing_count++] = table->slots[fd].file; + table->slots[fd].in_use = false; + } + } + mutex_unlock(&table->mutex); + + for (int i = 0; i < closing_count; i++) { + closing[i].ops->close(closing[i].object); + } + + mutex_destruct(&table->mutex); +} + +error_t app_fd_table_bind(AppFdTable* table, int fd, const AppFileOps* ops, void* object) { + if (!fd_in_range(fd)) { + return ERROR_OUT_OF_RANGE; + } + + mutex_lock(&table->mutex); + check(!table->shutting_down); + AppFdSlot& slot = table->slots[fd]; + AppFile old = slot.file; + bool had_old = slot.in_use; + slot.file = { .ops = ops, .object = object }; + slot.in_use = true; + slot.ever_used = true; + mutex_unlock(&table->mutex); + + if (had_old) { + old.ops->close(old.object); + } + return ERROR_NONE; +} + +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++) { + AppFdSlot& slot = table->slots[fd]; + if (!slot.in_use) { + slot.file = { .ops = ops, .object = object }; + slot.in_use = true; + slot.ever_used = true; + mutex_unlock(&table->mutex); + *out_fd = fd; + return ERROR_NONE; + } + } + mutex_unlock(&table->mutex); + return ERROR_RESOURCE; +} + +bool app_fd_table_get(AppFdTable* table, int fd, AppFile* out_file) { + return get_internal(table, fd, out_file, /*retain=*/false); +} + +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) { + if (!fd_in_range(fd)) { + return ERROR_NOT_FOUND; + } + + mutex_lock(&table->mutex); + check(!table->shutting_down); + AppFdSlot& slot = table->slots[fd]; + bool in_use = slot.in_use; + AppFile old = slot.file; + slot.in_use = false; + mutex_unlock(&table->mutex); + + if (!in_use) { + return ERROR_NOT_FOUND; + } + return old.ops->close(old.object); +} + +bool app_fd_table_is_app_owned(AppFdTable* table, int fd) { + if (!fd_in_range(fd)) { + return false; + } + + mutex_lock(&table->mutex); + bool owned = table->slots[fd].in_use || table->slots[fd].ever_used; + mutex_unlock(&table->mutex); + return owned; +} + +} // extern "C" diff --git a/Modules/app-module/source/io.cpp b/Modules/app-module/source/io.cpp new file mode 100644 index 00000000..ff2aafdb --- /dev/null +++ b/Modules/app-module/source/io.cpp @@ -0,0 +1,106 @@ +// SPDX-License-Identifier: Apache-2.0 +#include + +#include +#include +#include + +#include + +#ifdef ESP_PLATFORM +extern "C" { +ssize_t __real_read(int fd, void* buffer, size_t size); +ssize_t __real_write(int fd, const void* buffer, size_t size); +int __real_close(int fd); +} +#else +#include +#endif + +namespace { + +// NULL for a caller not running as an app instance (e.g. a kernel service task). @a fd is then +// a real underlying fd, handled by the caller falling through to the real syscall. +AppFdTable* current_app_fd_table() { + AppInstanceId app_id = app_scheduler_current_app_id(); + if (app_id == 0) { + return nullptr; + } + auto& ledger = app_ledger(); + mutex_lock(&ledger.mutex); + auto iterator = ledger.instances.find(app_id); + AppFdTable* table = (iterator != ledger.instances.end()) ? &iterator->second.fd_table : nullptr; + mutex_unlock(&ledger.mutex); + return table; +} + +} // namespace + +extern "C" { + +ssize_t app_io_read(int fd, void* buffer, size_t size) { + AppFdTable* table = current_app_fd_table(); + AppFile file {}; + if (table != nullptr && app_fd_table_get_and_retain(table, fd, &file)) { + ssize_t result = file.ops->read(file.object, buffer, size); + if (file.ops->release != nullptr) { + file.ops->release(file.object); + } + return result; + } + // @a fd isn't currently bound. If this table has never touched it either, it's a real + // underlying fd (e.g. from fopen()/open(), which app-module never intercepts; see app/io.h), + // so fall through. Otherwise it's one of ours that's already been closed: a real EBADF, not + // a real fd to hand to the platform (which could belong to something else entirely by now). + if (table != nullptr && app_fd_table_is_app_owned(table, fd)) { + errno = EBADF; + return -1; + } +#ifdef ESP_PLATFORM + return __real_read(fd, buffer, size); +#else + return ::read(fd, buffer, size); +#endif +} + +ssize_t app_io_write(int fd, const void* buffer, size_t size) { + AppFdTable* table = current_app_fd_table(); + AppFile file {}; + if (table != nullptr && app_fd_table_get_and_retain(table, fd, &file)) { + ssize_t result = file.ops->write(file.object, buffer, size); + if (file.ops->release != nullptr) { + file.ops->release(file.object); + } + return result; + } + if (table != nullptr && app_fd_table_is_app_owned(table, fd)) { + errno = EBADF; + return -1; + } +#ifdef ESP_PLATFORM + return __real_write(fd, buffer, size); +#else + return ::write(fd, buffer, size); +#endif +} + +int app_io_close(int fd) { + AppFdTable* table = current_app_fd_table(); + if (table != nullptr) { + error_t result = app_fd_table_close(table, fd); + if (result == ERROR_NONE) { + return 0; + } + if (app_fd_table_is_app_owned(table, fd)) { + errno = EBADF; + return -1; + } + } +#ifdef ESP_PLATFORM + return __real_close(fd); +#else + return ::close(fd); +#endif +} + +} // extern "C" diff --git a/Modules/app-module/source/manager.cpp b/Modules/app-module/source/manager.cpp index ae200ac1..5facfea8 100644 --- a/Modules/app-module/source/manager.cpp +++ b/Modules/app-module/source/manager.cpp @@ -1,6 +1,7 @@ // SPDX-License-Identifier: Apache-2.0 #include #include +#include #include #include #include @@ -89,8 +90,15 @@ char** copy_arguments(int argc, const char* const argv[]) { // Takes ownership of argv (already a deep copy, or NULL/argc==0) regardless of outcome - // app_scheduler_start() frees it on any failure path, and the spawned task frees it once its -// run() returns. -error_t start_internal(const char* id, AppInstanceId parent_instance_id, int argc, char* argv[], AppInstanceId* out_app_instance_id) { +// run() returns. @a bindings (@a binding_count entries, may be NULL/0) are subscribed into the +// 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); @@ -106,13 +114,42 @@ error_t start_internal(const char* id, AppInstanceId parent_instance_id, int arg AppInstanceRecord record { .id = target_id, .manifest = manifest, .state = APP_INSTANCE_STATE_STARTING, .task = nullptr }; record.parent_id = parent_instance_id; ledger.instances[target_id] = record; + // Constructed on the map-resident copy, not the local `record` about to go out of scope. + // AppFdTable::fds[] entries point into AppFdTable::slots[] by address (see fd_table.h), so + // constructing before the copy above would leave them pointing at stack storage. + app_fd_table_construct(&ledger.instances[target_id].fd_table); mutex_unlock(&ledger.mutex); LOG_I(TAG, "[instance %d] starting %s with parent %d", target_id, manifest->id, parent_instance_id); + for (size_t i = 0; i < binding_count; i++) { + 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); + mutex_unlock(&ledger.mutex); + app_ledger_free_arguments(argc, argv); + return bind_result; + } + } + 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); mutex_unlock(&ledger.mutex); LOG_I(TAG, "[instance %d] Failed to start: %s", target_id, error_to_string(error)); @@ -126,15 +163,19 @@ error_t start_internal(const char* id, AppInstanceId parent_instance_id, int arg } // namespace error_t app_manager_start(const char* id, AppInstanceId* out_app_instance_id) { - return start_internal(id, 0, 0, nullptr, out_app_instance_id); + return start_internal(id, 0, 0, nullptr, nullptr, 0, out_app_instance_id); } error_t app_manager_start_with_parameters(const char* id, int argc, const char* const argv[], AppInstanceId* out_app_instance_id) { - return start_internal(id, 0, argc, copy_arguments(argc, argv), out_app_instance_id); + return start_internal(id, 0, argc, copy_arguments(argc, argv), nullptr, 0, out_app_instance_id); } error_t app_manager_start_for_result(const char* id, AppInstanceId parent_instance_id, int argc, const char* const argv[], AppInstanceId* out_app_instance_id) { - return start_internal(id, parent_instance_id, argc, copy_arguments(argc, argv), out_app_instance_id); + return start_internal(id, parent_instance_id, argc, copy_arguments(argc, argv), nullptr, 0, out_app_instance_id); +} + +error_t app_manager_start_with_streams(const char* id, const AppStreamBinding* bindings, size_t binding_count, AppInstanceId* out_app_instance_id) { + return start_internal(id, 0, 0, nullptr, bindings, binding_count, out_app_instance_id); } error_t app_manager_stop(AppInstanceId app_instance_id) { diff --git a/Modules/app-module/source/module.cpp b/Modules/app-module/source/module.cpp index 80ef97fa..7000288e 100644 --- a/Modules/app-module/source/module.cpp +++ b/Modules/app-module/source/module.cpp @@ -1,11 +1,13 @@ // SPDX-License-Identifier: Apache-2.0 #include #include +#include #include #include #include #include #include +#include #include @@ -27,10 +29,15 @@ static const ModuleSymbol SYMBOLS[] = { DEFINE_MODULE_SYMBOL(app_get_install_path), DEFINE_MODULE_SYMBOL(app_install), DEFINE_MODULE_SYMBOL(app_uninstall), + // app/io + DEFINE_MODULE_SYMBOL(app_io_read), + DEFINE_MODULE_SYMBOL(app_io_write), + DEFINE_MODULE_SYMBOL(app_io_close), // app/manager DEFINE_MODULE_SYMBOL(app_manager_start), DEFINE_MODULE_SYMBOL(app_manager_start_with_parameters), DEFINE_MODULE_SYMBOL(app_manager_start_for_result), + DEFINE_MODULE_SYMBOL(app_manager_start_with_streams), DEFINE_MODULE_SYMBOL(app_manager_stop), DEFINE_MODULE_SYMBOL(app_manager_get_state), DEFINE_MODULE_SYMBOL(app_manager_find_manifest), @@ -53,6 +60,13 @@ static const ModuleSymbol SYMBOLS[] = { DEFINE_MODULE_SYMBOL(app_paths_get_assets_path), // app/scheduler DEFINE_MODULE_SYMBOL(app_scheduler_current_app_id), + // app/stream + DEFINE_MODULE_SYMBOL(app_stream_subscribe), + DEFINE_MODULE_SYMBOL(app_stream_unsubscribe), + DEFINE_MODULE_SYMBOL(app_stream_await), + DEFINE_MODULE_SYMBOL(app_stream_read), + DEFINE_MODULE_SYMBOL(app_stream_write), + DEFINE_MODULE_SYMBOL(app_stream_close), // terminator MODULE_SYMBOL_TERMINATOR, }; diff --git a/Modules/app-module/source/null_device.cpp b/Modules/app-module/source/null_device.cpp new file mode 100644 index 00000000..de53d05a --- /dev/null +++ b/Modules/app-module/source/null_device.cpp @@ -0,0 +1,45 @@ +// SPDX-License-Identifier: Apache-2.0 +#include + +namespace { + +ssize_t null_read(void*, void*, size_t) { + return 0; // EOF +} + +ssize_t null_write(void*, const void* buffer, size_t size) { + (void)buffer; + return static_cast(size); // discarded, reported as fully written +} + +error_t null_close(void*) { + return ERROR_NONE; +} + +error_t null_await(void*, AppFileWait, TickType_t) { + return ERROR_NONE; // always ready +} + +uint32_t null_poll(void*) { + return APP_FILE_READABLE | APP_FILE_WRITABLE; +} + +const AppFileOps NULL_OPS = { + .read = null_read, + .write = null_write, + .close = null_close, + .await = null_await, + .poll = null_poll, +}; + +constexpr AppFile NULL_FILE = { .ops = &NULL_OPS, .object = nullptr }; + +} // namespace + +extern "C" { + +const AppFile* app_null_file(void) { + return &NULL_FILE; +} + +} // extern "C" diff --git a/Modules/app-module/source/scheduler.cpp b/Modules/app-module/source/scheduler.cpp index 52e96c25..8a838df9 100644 --- a/Modules/app-module/source/scheduler.cpp +++ b/Modules/app-module/source/scheduler.cpp @@ -2,9 +2,12 @@ #include #include #include +#include #include #include +#include #include +#include #include #include @@ -98,6 +101,14 @@ void set_task(AppInstanceId app_instance_id, TaskHandle_t task) { auto iterator = ledger.instances.find(app_instance_id); if (iterator != ledger.instances.end()) { iterator->second.task = task; + // Streams bound before this instance's task existed (app_manager_start_with_streams()) + // only got producer_task filled in as NULL at subscribe time. Backfill it now. + AppFdTable& fd_table = iterator->second.fd_table; + for (auto& slot : fd_table.slots) { + if (slot.in_use && slot.file.ops == app_stream_ops()) { + static_cast(slot.file.object)->producer_task = task; + } + } } mutex_unlock(&ledger.mutex); } @@ -218,8 +229,14 @@ void app_task_main(void* context) { // Erase the ledger entry before self-deleting - see "Reap self-terminated app tasks": // nothing else is guaranteed to ever call app_scheduler_stop() for this instance (the // common case is the app just closing itself), so this can't wait for that to happen. + // Every non-null fd is closed here too. Stream-backed entries wake/mark closed whoever is + // on the other end (e.g. a parent reading this instance's stdout). auto& ledger = app_ledger(); mutex_lock(&ledger.mutex); + auto fd_table_iterator = ledger.instances.find(app_instance_id); + if (fd_table_iterator != ledger.instances.end()) { + app_fd_table_teardown(&fd_table_iterator->second.fd_table); + } ledger.instances.erase(app_instance_id); mutex_unlock(&ledger.mutex); diff --git a/Modules/app-module/source/stream.cpp b/Modules/app-module/source/stream.cpp new file mode 100644 index 00000000..e0727156 --- /dev/null +++ b/Modules/app-module/source/stream.cpp @@ -0,0 +1,313 @@ +// SPDX-License-Identifier: Apache-2.0 +#include + +#include +#include +#include + +#include +#include + +namespace { + +// Caller must hold stream->mutex. +bool is_readable_locked(AppStream* stream) { + return stream->buffer.count > 0 || stream->closed; +} + +// Caller must hold stream->mutex. +bool is_writable_locked(AppStream* stream) { + return stream->buffer.count < stream->buffer.capacity || stream->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->mutex. +class StreamOperationGuard { +public: + explicit StreamOperationGuard(AppStream* stream) : stream_(stream) { + mutex_lock(&stream_->mutex); + stream_->active_operations++; + mutex_unlock(&stream_->mutex); + } + + ~StreamOperationGuard() { + mutex_lock(&stream_->mutex); + stream_->active_operations--; + mutex_unlock(&stream_->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; + } + size_t read = app_stream_read(stream, buffer, size); + if (read > 0) { + return static_cast(read); + } + mutex_lock(&stream->mutex); + bool is_closed = stream->closed; + mutex_unlock(&stream->mutex); + if (is_closed) { + return 0; // EOF: readable-because-closed, and nothing left buffered + } + // Woke readable but another party drained it first. Re-await rather than assume EOF; + // this is a defensive fallback since AppStream expects a single reader. + } +} + +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->mutex); + bool is_closed = stream->closed; + mutex_unlock(&stream->mutex); + return is_closed ? -1 : 0; // broken pipe, or a spurious wake with nothing to copy +} + +error_t stream_file_close(void* 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) { + 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->mutex); + uint32_t bits = 0; + if (is_readable_locked(stream)) { + bits |= APP_FILE_READABLE; + } + if (is_writable_locked(stream)) { + bits |= APP_FILE_WRITABLE; + } + mutex_unlock(&stream->mutex); + 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->mutex); + stream->active_operations++; + mutex_unlock(&stream->mutex); +} + +void stream_file_release(void* object) { + auto* stream = static_cast(object); + mutex_lock(&stream->mutex); + stream->active_operations--; + mutex_unlock(&stream->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 + +extern "C" { + +const AppFileOps* app_stream_ops(void) { + return &STREAM_OPS; +} + +error_t app_stream_subscribe(AppStream* stream, void* buffer, size_t buffer_capacity, TaskEventGroup* event_group, AppInstanceId producer_id, int producer_fd) { + if (producer_fd < 0 || producer_fd >= APP_MAX_FDS) { + return ERROR_OUT_OF_RANGE; + } + + uint32_t readable_bit; + error_t claim_result = task_event_group_claim_bit(event_group, &readable_bit); + if (claim_result != ERROR_NONE) { + return claim_result; + } + uint32_t writable_bit; + claim_result = task_event_group_claim_bit(event_group, &writable_bit); + if (claim_result != ERROR_NONE) { + task_event_group_release_bit(event_group, readable_bit); + return claim_result; + } + + stream->producer_id = producer_id; + stream->producer_fd = producer_fd; + stream->event_group = event_group; + stream->readable_bit = readable_bit; + stream->writable_bit = writable_bit; + stream->buffer.data = static_cast(buffer); + stream->buffer.capacity = buffer_capacity; + stream->buffer.read_pos = 0; + stream->buffer.write_pos = 0; + stream->buffer.count = 0; + stream->closed = false; + stream->active_operations = 0; + mutex_construct(&stream->mutex); + + // 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->mutex); + task_event_group_release_bit(event_group, readable_bit); + task_event_group_release_bit(event_group, writable_bit); + return ERROR_NOT_FOUND; + } + stream->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->mutex); + task_event_group_release_bit(event_group, readable_bit); + task_event_group_release_bit(event_group, writable_bit); + } + return bind_result; +} + +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->producer_id); + if (iterator != ledger.instances.end()) { + AppFdTable* table = &iterator->second.fd_table; + AppFile current {}; + if (app_fd_table_get(table, stream->producer_fd, ¤t) && current.object == stream) { + app_fd_table_close(table, stream->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->mutex); + int active_operations = stream->active_operations; + mutex_unlock(&stream->mutex); + if (active_operations == 0) { + break; + } + taskYIELD(); + } + + task_event_group_release_bit(stream->event_group, stream->readable_bit); + task_event_group_release_bit(stream->event_group, stream->writable_bit); + mutex_destruct(&stream->mutex); + return ERROR_NONE; +} + +error_t app_stream_await(AppStream* stream, AppFileWait wait, TickType_t timeout) { + mutex_lock(&stream->mutex); + bool ready = (wait == APP_FILE_WAIT_READABLE) ? is_readable_locked(stream) : is_writable_locked(stream); + mutex_unlock(&stream->mutex); + if (ready) { + return ERROR_NONE; + } + + uint32_t bit = (wait == APP_FILE_WAIT_READABLE) ? stream->readable_bit : stream->writable_bit; + return task_event_group_wait(stream->event_group, bit, /*await_all=*/false, nullptr, timeout); +} + +size_t app_stream_read(AppStream* stream, void* buffer, size_t buffer_size) { + mutex_lock(&stream->mutex); + AppStreamBuffer& ring = stream->buffer; + size_t to_copy = buffer_size < ring.count ? buffer_size : ring.count; + for (size_t i = 0; i < to_copy; i++) { + static_cast(buffer)[i] = ring.data[(ring.read_pos + i) % ring.capacity]; + } + if (ring.capacity > 0) { + ring.read_pos = (ring.read_pos + to_copy) % ring.capacity; + } + ring.count -= to_copy; + TaskEventGroup* event_group = stream->event_group; + uint32_t writable_bit = stream->writable_bit; + mutex_unlock(&stream->mutex); + if (to_copy > 0) { + task_event_group_signal(event_group, writable_bit); // space freed up + } + return to_copy; +} + +size_t app_stream_write(AppStream* stream, const void* buffer, size_t buffer_size) { + mutex_lock(&stream->mutex); + if (stream->closed) { + mutex_unlock(&stream->mutex); + return 0; + } + AppStreamBuffer& ring = stream->buffer; + size_t available = ring.capacity - ring.count; + size_t to_copy = buffer_size < available ? buffer_size : available; + for (size_t i = 0; i < to_copy; i++) { + ring.data[(ring.write_pos + i) % ring.capacity] = static_cast(buffer)[i]; + } + if (ring.capacity > 0) { + ring.write_pos = (ring.write_pos + to_copy) % ring.capacity; + } + ring.count += to_copy; + TaskEventGroup* event_group = stream->event_group; + uint32_t readable_bit = stream->readable_bit; + mutex_unlock(&stream->mutex); + if (to_copy > 0) { + task_event_group_signal(event_group, readable_bit); // data became available + } + return to_copy; +} + +error_t app_stream_close(AppStream* stream) { + mutex_lock(&stream->mutex); + stream->closed = true; + TaskEventGroup* event_group = stream->event_group; + uint32_t readable_bit = stream->readable_bit; + uint32_t writable_bit = stream->writable_bit; + mutex_unlock(&stream->mutex); + task_event_group_signal(event_group, readable_bit); + task_event_group_signal(event_group, writable_bit); + return ERROR_NONE; +} + +} // extern "C" diff --git a/Modules/app-module/tests/source/io_test.cpp b/Modules/app-module/tests/source/io_test.cpp new file mode 100644 index 00000000..aa7b37de --- /dev/null +++ b/Modules/app-module/tests/source/io_test.cpp @@ -0,0 +1,294 @@ +// SPDX-License-Identifier: Apache-2.0 +#include "doctest.h" + +#include +#include +#include +#include +#include + +#include + +#include + +#include +#include + +#include +#include +#include + +extern ServiceManifest app_internal_loader_service_manifest; + +namespace { + +// See manager_test.cpp's own copy of this helper for why this checks the registry directly +// rather than a per-translation-unit static bool. +void ensure_memory_loader_registered() { + if (service_manager_find_instance(APP_LOADER_MEMORY_SERVICE_ID) == nullptr) { + service_manager_add(&app_internal_loader_service_manifest, /*auto_start=*/true); + } +} + +bool wait_for_state(AppInstanceId id, AppInstanceState target, uint32_t timeout_ms) { + uint32_t waited = 0; + while (waited < timeout_ms) { + if (app_manager_get_state(id) == target) { + return true; + } + delay_millis(10); + waited += 10; + } + return app_manager_get_state(id) == target; +} + +std::atomic g_stdio_write_result { -2 }; +std::atomic g_stdio_read_result { -2 }; + +int32_t unbound_stdio_app_main(int, char*[]) { + g_stdio_write_result.store(app_io_write(STDOUT_FILENO, "x", 1), std::memory_order_release); + uint8_t buffer[1]; + g_stdio_read_result.store(app_io_read(STDIN_FILENO, buffer, sizeof(buffer)), std::memory_order_release); + return 0; +} + +int32_t stdout_writer_app_main(int, char*[]) { + const char message[] = "hello"; + size_t sent = 0; + while (sent < sizeof(message) - 1) { + ssize_t written = app_io_write(STDOUT_FILENO, message + sent, sizeof(message) - 1 - sent); + if (written < 0) { + break; + } + sent += static_cast(written); + } + return 0; +} + +std::atomic g_blocked_writer_saw_error { false }; +std::atomic g_blocked_writer_done { false }; + +int32_t blocked_writer_app_main(int, char*[]) { + const char message[] = "0123456789"; // larger than the test's 4-byte stream capacity + size_t sent = 0; + while (sent < sizeof(message) - 1) { + ssize_t written = app_io_write(STDOUT_FILENO, message + sent, sizeof(message) - 1 - sent); + if (written < 0) { + g_blocked_writer_saw_error.store(true, std::memory_order_release); + break; + } + sent += static_cast(written); + } + g_blocked_writer_done.store(true, std::memory_order_release); + return 0; +} + +std::atomic g_real_file_write_result { -2 }; +std::atomic g_real_file_read_result { -2 }; +std::atomic g_real_file_read_matches { false }; +std::atomic g_real_file_close_result { -2 }; + +// A real file fd from a bare open() call: app-module never intercepts open(), so this fd is +// never bound/allocated in the app's own fd table. app_io_read/write/close() must still pass it +// straight through to the real syscall instead of treating it as an unknown app-level fd. +int32_t real_file_io_app_main(int, char*[]) { + const char* path = "/tmp/tactility_app_io_passthrough_test.txt"; + int real_fd = ::open(path, O_CREAT | O_TRUNC | O_RDWR, 0600); + if (real_fd < 0) { + return 0; + } + + g_real_file_write_result.store(app_io_write(real_fd, "hi", 2), std::memory_order_release); + ::lseek(real_fd, 0, SEEK_SET); + char buffer[2] = {}; + ssize_t read_result = app_io_read(real_fd, buffer, sizeof(buffer)); + g_real_file_read_result.store(read_result, std::memory_order_release); + g_real_file_read_matches.store(read_result == 2 && buffer[0] == 'h' && buffer[1] == 'i', std::memory_order_release); + g_real_file_close_result.store(app_io_close(real_fd), std::memory_order_release); + + ::unlink(path); + return 0; +} + +std::atomic g_double_close_first_result { -2 }; +std::atomic g_double_close_second_result { -2 }; +std::atomic g_write_after_close_result { -2 }; + +// A second close() of an already-closed app fd, and a write() after that, must both report +// EBADF, never fall through to the platform syscall, which by then could be operating on a real +// fd that fd number was recycled for (e.g. the process's real stdout). +int32_t double_close_app_main(int, char*[]) { + g_double_close_first_result.store(app_io_close(STDOUT_FILENO), std::memory_order_release); + g_double_close_second_result.store(app_io_close(STDOUT_FILENO), std::memory_order_release); + g_write_after_close_result.store(app_io_write(STDOUT_FILENO, "x", 1), std::memory_order_release); + return 0; +} + +} // namespace + +TEST_CASE("an app's stdio fds default to the null device: write succeeds and discards, read reports EOF") { + ensure_memory_loader_registered(); + g_stdio_write_result.store(-2, std::memory_order_relaxed); + g_stdio_read_result.store(-2, std::memory_order_relaxed); + + AppManifest manifest { "test.io.unbound", "Unbound", APP_CATEGORY_USER, { APP_LOCATION_MEMORY, reinterpret_cast(unbound_stdio_app_main) } }; + REQUIRE_EQ(app_manager_add(&manifest), ERROR_NONE); + + AppInstanceId instance_id = 0; + REQUIRE_EQ(app_manager_start("test.io.unbound", &instance_id), ERROR_NONE); + REQUIRE(wait_for_state(instance_id, APP_INSTANCE_STATE_STOPPED, 1000)); + + CHECK_EQ(g_stdio_write_result.load(std::memory_order_acquire), 1); + CHECK_EQ(g_stdio_read_result.load(std::memory_order_acquire), 0); + + app_manager_remove("test.io.unbound"); +} + +TEST_CASE("app_manager_start_with_streams pipes a child's app_io_write() calls into a parent-owned AppStream, EOF at exit") { + ensure_memory_loader_registered(); + + AppManifest manifest { "test.io.writer", "Writer", APP_CATEGORY_USER, { APP_LOCATION_MEMORY, reinterpret_cast(stdout_writer_app_main) } }; + REQUIRE_EQ(app_manager_add(&manifest), ERROR_NONE); + + TaskEventGroup event_group {}; + task_event_group_construct(&event_group); + + uint8_t storage[64]; + 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.writer", &binding, 1, &child_id), ERROR_NONE); + + std::vector received; + while (app_stream_await(&child_stdout, APP_FILE_WAIT_READABLE, pdMS_TO_TICKS(1000)) == ERROR_NONE) { + uint8_t chunk[16]; + size_t n = app_stream_read(&child_stdout, chunk, sizeof(chunk)); + if (n == 0) { + break; // EOF + } + received.insert(received.end(), chunk, chunk + n); + } + + REQUIRE_EQ(received.size(), 5u); + CHECK_EQ(std::memcmp(received.data(), "hello", 5), 0); + + REQUIRE(wait_for_state(child_id, APP_INSTANCE_STATE_STOPPED, 1000)); + app_stream_unsubscribe(&child_stdout); + task_event_group_destruct(&event_group); + app_manager_remove("test.io.writer"); +} + +TEST_CASE("a write blocked on a full stream wakes with an error once the consumer closes it") { + 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.blocked", "Blocked", 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.blocked", &binding, 1, &child_id), ERROR_NONE); + + // Never drained: the child fills the 4-byte buffer and blocks awaiting space for the rest. + delay_millis(200); + CHECK_FALSE(g_blocked_writer_done.load(std::memory_order_acquire)); + + // Unblocks the writer without touching child_stdout's sync primitives, safe even while it + // may still be blocked in app_stream_await() (unlike app_stream_unsubscribe()). + app_stream_close(&child_stdout); + + REQUIRE(wait_for_state(child_id, APP_INSTANCE_STATE_STOPPED, 1000)); + CHECK(g_blocked_writer_saw_error.load(std::memory_order_acquire)); + + // Only safe now that the child's task (the only other party that could be blocked on this + // stream) has fully exited. + app_stream_unsubscribe(&child_stdout); + 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->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"); +} + +TEST_CASE("app_io_read/write/close pass through a real file fd app-module never bound") { + ensure_memory_loader_registered(); + g_real_file_write_result.store(-2, std::memory_order_relaxed); + g_real_file_read_result.store(-2, std::memory_order_relaxed); + g_real_file_read_matches.store(false, std::memory_order_relaxed); + g_real_file_close_result.store(-2, std::memory_order_relaxed); + + AppManifest manifest { "test.io.real_file", "RealFile", APP_CATEGORY_USER, { APP_LOCATION_MEMORY, reinterpret_cast(real_file_io_app_main) } }; + REQUIRE_EQ(app_manager_add(&manifest), ERROR_NONE); + + AppInstanceId instance_id = 0; + REQUIRE_EQ(app_manager_start("test.io.real_file", &instance_id), ERROR_NONE); + REQUIRE(wait_for_state(instance_id, APP_INSTANCE_STATE_STOPPED, 1000)); + + CHECK_EQ(g_real_file_write_result.load(std::memory_order_acquire), 2); + CHECK_EQ(g_real_file_read_result.load(std::memory_order_acquire), 2); + CHECK(g_real_file_read_matches.load(std::memory_order_acquire)); + CHECK_EQ(g_real_file_close_result.load(std::memory_order_acquire), 0); + + app_manager_remove("test.io.real_file"); +} + +TEST_CASE("closing an already-closed app fd reports EBADF instead of falling through to the platform") { + ensure_memory_loader_registered(); + g_double_close_first_result.store(-2, std::memory_order_relaxed); + g_double_close_second_result.store(-2, std::memory_order_relaxed); + g_write_after_close_result.store(-2, std::memory_order_relaxed); + + AppManifest manifest { "test.io.double_close", "DoubleClose", APP_CATEGORY_USER, { APP_LOCATION_MEMORY, reinterpret_cast(double_close_app_main) } }; + REQUIRE_EQ(app_manager_add(&manifest), ERROR_NONE); + + AppInstanceId instance_id = 0; + REQUIRE_EQ(app_manager_start("test.io.double_close", &instance_id), ERROR_NONE); + REQUIRE(wait_for_state(instance_id, APP_INSTANCE_STATE_STOPPED, 1000)); + + CHECK_EQ(g_double_close_first_result.load(std::memory_order_acquire), 0); + CHECK_EQ(g_double_close_second_result.load(std::memory_order_acquire), -1); + CHECK_EQ(g_write_after_close_result.load(std::memory_order_acquire), -1); + + app_manager_remove("test.io.double_close"); +} diff --git a/Modules/app-module/tests/source/manager_test.cpp b/Modules/app-module/tests/source/manager_test.cpp index 1c2e7770..9dc8fca1 100644 --- a/Modules/app-module/tests/source/manager_test.cpp +++ b/Modules/app-module/tests/source/manager_test.cpp @@ -128,12 +128,12 @@ void ensure_fake_loader_registered() { // app-module's real APP_LOCATION_MEMORY loader (source/app_internal_loader.cpp) - not a fake, // since it has no platform dependency and is exactly what a statically-linked app would go -// through. +// through. Checks the registry directly rather than a per-translation-unit static bool: other +// test files (stream_test.cpp, io_test.cpp) register the same manifest the same way, and +// doctest doesn't guarantee which file's tests run first. void ensure_memory_loader_registered() { - static bool registered = false; - if (!registered) { - CHECK_EQ(service_manager_add(&app_internal_loader_service_manifest, /*auto_start=*/true), ERROR_NONE); - registered = true; + if (service_manager_find_instance(APP_LOADER_MEMORY_SERVICE_ID) == nullptr) { + service_manager_add(&app_internal_loader_service_manifest, /*auto_start=*/true); } } diff --git a/Modules/app-module/tests/source/stream_test.cpp b/Modules/app-module/tests/source/stream_test.cpp new file mode 100644 index 00000000..f11dc1ed --- /dev/null +++ b/Modules/app-module/tests/source/stream_test.cpp @@ -0,0 +1,158 @@ +// SPDX-License-Identifier: Apache-2.0 +#include "doctest.h" + +#include +#include +#include +#include +#include + +#include + +#include + +#include + +extern ServiceManifest app_internal_loader_service_manifest; + +namespace { + +// See manager_test.cpp's own copy of this helper for why this checks the registry directly +// rather than a per-translation-unit static bool. +void ensure_memory_loader_registered() { + if (service_manager_find_instance(APP_LOADER_MEMORY_SERVICE_ID) == nullptr) { + service_manager_add(&app_internal_loader_service_manifest, /*auto_start=*/true); + } +} + +// Stays Active, doing nothing, until asked to close. Just an anchor instance for +// app_stream_subscribe() to target; the tests below drive the resulting AppStream directly from +// the test thread, the same way a parent consumes a child's stream without going through its +// own fd table. +int32_t idle_app_main(int, char*[]) { + TaskEventGroup event_group {}; + task_event_group_construct(&event_group); + + AppEventSubscription sub {}; + app_event_subscribe(&sub, &event_group); + + while (true) { + if (task_event_group_wait_any(&event_group, nullptr, pdMS_TO_TICKS(5000)) != ERROR_NONE) { + break; // safety net + } + bool done = false; + AppEvent event {}; + while (app_event_poll(&sub, &event) == ERROR_NONE) { + if (event.type == APP_EVENT_CLOSE) { + done = true; + break; + } + } + if (done) { + break; + } + } + + app_event_unsubscribe(&sub); + task_event_group_destruct(&event_group); + return 0; +} + +bool wait_for_state(AppInstanceId id, AppInstanceState target, uint32_t timeout_ms) { + uint32_t waited = 0; + while (waited < timeout_ms) { + if (app_manager_get_state(id) == target) { + return true; + } + delay_millis(10); + waited += 10; + } + return app_manager_get_state(id) == target; +} + +AppInstanceId start_idle_app(const char* id) { + ensure_memory_loader_registered(); + AppManifest manifest { id, id, APP_CATEGORY_USER, { APP_LOCATION_MEMORY, reinterpret_cast(idle_app_main) } }; + REQUIRE_EQ(app_manager_add(&manifest), ERROR_NONE); + AppInstanceId instance_id = 0; + REQUIRE_EQ(app_manager_start(id, &instance_id), ERROR_NONE); + REQUIRE(wait_for_state(instance_id, APP_INSTANCE_STATE_ACTIVE, 1000)); + return instance_id; +} + +} // namespace + +TEST_CASE("app_stream_read/write move bytes through a ring buffer, respecting capacity") { + AppInstanceId producer_id = start_idle_app("test.stream.ringbuffer"); + + TaskEventGroup event_group {}; + task_event_group_construct(&event_group); + + uint8_t storage[4]; + AppStream stream {}; + REQUIRE_EQ(app_stream_subscribe(&stream, storage, sizeof(storage), &event_group, producer_id, 5), ERROR_NONE); + + CHECK_EQ(app_stream_write(&stream, "ab", 2), 2u); + CHECK_EQ(app_stream_write(&stream, "cdef", 4), 2u); // only 2 bytes of space left + + uint8_t out[8] = {}; + CHECK_EQ(app_stream_read(&stream, out, sizeof(out)), 4u); + CHECK_EQ(std::memcmp(out, "abcd", 4), 0); + + app_stream_unsubscribe(&stream); + task_event_group_destruct(&event_group); + app_manager_stop(producer_id); + app_manager_remove("test.stream.ringbuffer"); +} + +TEST_CASE("app_stream_await blocks until data/space is available, and closing forces both bits") { + AppInstanceId producer_id = start_idle_app("test.stream.await"); + + TaskEventGroup event_group {}; + task_event_group_construct(&event_group); + + uint8_t storage[2]; + AppStream stream {}; + REQUIRE_EQ(app_stream_subscribe(&stream, storage, sizeof(storage), &event_group, producer_id, 5), ERROR_NONE); + + // Nothing written yet, so the readable wait times out. + CHECK_EQ(app_stream_await(&stream, APP_FILE_WAIT_READABLE, pdMS_TO_TICKS(50)), ERROR_TIMEOUT); + + app_stream_write(&stream, "x", 1); + CHECK_EQ(app_stream_await(&stream, APP_FILE_WAIT_READABLE, pdMS_TO_TICKS(50)), ERROR_NONE); + + app_stream_close(&stream); + CHECK_EQ(app_stream_await(&stream, APP_FILE_WAIT_READABLE, pdMS_TO_TICKS(50)), ERROR_NONE); + CHECK_EQ(app_stream_await(&stream, APP_FILE_WAIT_WRITABLE, pdMS_TO_TICKS(50)), ERROR_NONE); + + app_stream_unsubscribe(&stream); + task_event_group_destruct(&event_group); + app_manager_stop(producer_id); + app_manager_remove("test.stream.await"); +} + +TEST_CASE("app_stream_subscribe over an existing subscription atomically replaces it") { + AppInstanceId producer_id = start_idle_app("test.stream.replace"); + + TaskEventGroup event_group {}; + task_event_group_construct(&event_group); + + uint8_t storage_a[4]; + uint8_t storage_b[4]; + AppStream stream_a {}; + REQUIRE_EQ(app_stream_subscribe(&stream_a, storage_a, sizeof(storage_a), &event_group, producer_id, 5), ERROR_NONE); + + AppStream stream_b {}; + REQUIRE_EQ(app_stream_subscribe(&stream_b, storage_b, sizeof(storage_b), &event_group, producer_id, 5), ERROR_NONE); + + // stream_a was replaced: it's now closed (readable reports ready, and reads 0 bytes / EOF), + // even though nobody explicitly unsubscribed it. + CHECK_EQ(app_stream_await(&stream_a, APP_FILE_WAIT_READABLE, pdMS_TO_TICKS(50)), ERROR_NONE); + CHECK_EQ(app_stream_read(&stream_a, storage_a, sizeof(storage_a)), 0u); + + app_stream_unsubscribe(&stream_a); + app_stream_unsubscribe(&stream_b); + task_event_group_destruct(&event_group); + app_manager_stop(producer_id); + app_manager_remove("test.stream.replace"); +} diff --git a/Tactility/Source/AppStdioWrap.cpp b/Tactility/Source/AppStdioWrap.cpp new file mode 100644 index 00000000..464c6a74 --- /dev/null +++ b/Tactility/Source/AppStdioWrap.cpp @@ -0,0 +1,24 @@ +// SPDX-License-Identifier: Apache-2.0 +#ifdef ESP_PLATFORM + +#include + +#include + +extern "C" { + +ssize_t __wrap_read(int fd, void* buffer, size_t size) { + return app_io_read(fd, buffer, size); +} + +ssize_t __wrap_write(int fd, const void* buffer, size_t size) { + return app_io_write(fd, buffer, size); +} + +int __wrap_close(int fd) { + return app_io_close(fd); +} + +} + +#endif // ESP_PLATFORM