system/zbus: Port the Zephyr zbus message bus to NuttX - #3743
Conversation
dfd91d6 to
45592db
Compare
linguini1
left a comment
There was a problem hiding this comment.
Please add the Assisted-by field to your commit message to indicate AI use.
0806149 to
1c29f07
Compare
| * per-task mqd_t descriptors. mq.f_inode == NULL means "not opened". | ||
| */ | ||
|
|
||
| struct file mq; |
There was a problem hiding this comment.
file can only be used by kernel
| * | ||
| ****************************************************************************/ | ||
|
|
||
| static void zbus_init_fn(void) |
There was a problem hiding this comment.
NuttX design is totally different from zephyr, but more like traditional POSIX OS, so it isn't suitable to port RTOS code blindly to userspace without the careful consideration. for example:
- use file_mq API directly in userspace
- access obs field from sender, or access chan field from receiver
if you want to port zbus without the significant architecture change, the only choice is put it into the kernel space.
or you can study uorb design to learn how to make a message subsystem work in the POSIX context:
https://nuttx.apache.org/docs/latest/applications/system/uorb/index.html
https://nuttx.apache.org/docs/latest/components/drivers/special/sensors/sensors_uorb.htmle
especially the usensor design: userspace sensor register implementation
There was a problem hiding this comment.
The objection is correct: the code assumes every task shares one address space, which only holds in BUILD_FLAT. Passing a channel pointer between tasks, reading chan->message on the receiver, touching observer state from the publisher and calling file_mq_* from userspace all follow from that, and it has to be fixed.
My proposal is to fix it by declaring the scope rather than by moving the bus into the kernel: zbus as a process-local event bus, the threads of one application. That makes it POSIX only (mq_open() per task, sem_t, pthread_once()), takes struct file out of the public header, removes the global state opened by whichever task runs first, and no pointer ever crosses a process boundary.
Measured on an STM32H7 at 480 MHz with up_perf_gettime():
- publish to a synchronous listener: 4.4 us
- the same notification to a callback in another task: 16.2 us end to end
That is 3.7x, plus a context switch per message. [...] the cost is losing the synchronous callback, which is what zbus exists for, and a kernel-side bus cannot offer that callback to an application.
This is also what zbus is upstream. It has no CONFIG_USERSPACE support in Zephyr: supervisor mode, one address space, and the subscriber path passes the channel pointer through a msgq exactly as it does here. When a channel has to cross a domain, upstream adds an explicit forwarder (subsys/zbus/proxy_agent) instead of making the bus global.
| * mq_open() whose descriptor belongs to the calling task only. | ||
| */ | ||
|
|
||
| ret = file_mq_open(&obs->data->mq, name, O_RDWR | O_CREAT, 0644, |
There was a problem hiding this comment.
it's wrong to open/modify ALL obs field from the first zbus task.
1819694 to
bd5c346
Compare
Port of the Zephyr RTOS zbus (many-to-many message bus with typed channels and decoupled observers), built entirely on native NuttX primitives and preserving the original declarative API (ZBUS_CHAN_DEFINE, ZBUS_LISTENER_DEFINE, ZBUS_SUBSCRIBER_DEFINE, ...). Features: listeners (synchronous callbacks), subscribers (queue of channel references), message subscribers (ordered message copies), async listeners (callback on a dedicated task), runtime observers, per-observation notification masks, observer enable/disable, message validators, channel user data, publish statistics, lookup by name/numeric id and channel/observer iteration. Mapping to NuttX primitives: - Channel/observer registration: link-time iterable sections (include/nuttx/iterable_sections.h); the observers of a channel are named after their position in the definition, so the linker sorts the notification order, and the declarative macros are built on nuttx/macro.h (CONCATENATE, FOREACH_ARG and the FOREACH_IDX_ARG added in a companion nuttx commit) rather than on a private macro engine. Notification masks live in .bss with their initial value preserved in ROM and applied on lazy init. - Channel lock: sem_t (enable CONFIG_PRIORITY_INHERITANCE instead of the Zephyr priority-boost/HLP); timeouts are computed with the clock_timespec_* helpers from nuttx/clock.h. - Subscriber queues: kernel message queues (file_mq_*) opened lazily via pthread_once, usable from any task; mq payload copying replaces the Zephyr net_buf machinery entirely. - Async listeners: one task per listener (task_create, priority and stack size configurable) blocking on the listener queue; a task rather than a pthread so it outlives the first API caller. - Timeouts: milliseconds with CLOCK_MONOTONIC deadlines (ZBUS_NO_WAIT/ZBUS_FOREVER). Includes a runnable example (examples/zbus, CONFIG_EXAMPLES_ZBUS) and a cmocka test suite (testing/zbus, CONFIG_TESTING_ZBUS) covering the full API: 17/17 tests passing on linum-stm32h753bi hardware, including multi-channel index grouping, mask semantics, runtime observer error paths, notification order (the observers of a channel run in the order they are listed, and an observation bound with ZBUS_CHAN_ADD_OBS() runs after all of them), queue overflow/timeout semantics, async listener bursts, bit-exact float/double payload delivery across every observer type (sensor-style messages with a float-math validator) and an interrupt-driven publisher (kernel timer interrupt -> signal -> sampling thread -> zbus_chan_pub, the recommended pattern for interrupt sources). Requirements: FLAT build; CONFIG_MQ_MAXMSGSIZE >= pointer size + CONFIG_ZBUS_MSG_SUBSCRIBER_MAX_MSG_SIZE for message subscribers; board linker script including <nuttx/linker/common-rom.ld> or the generic CONFIG_ITERABLE_SECTIONS_LINKER_INSERT mode. Not ported: multi-domain proxy agent (experimental upstream); publishing from interrupt handlers (userspace library: hand the data to a thread). Documentation lives in the nuttx repository (Documentation/applications/system/zbus). Assisted-by: Claude Code Signed-off-by: Jorge Guzman <jorge.gzm@gmail.com>
Summary
This PR ports the Zephyr RTOS zbus (a many-to-many message bus with
typed channels and decoupled observers) to NuttX, preserving the
original declarative API (
ZBUS_CHAN_DEFINE,ZBUS_LISTENER_DEFINE,ZBUS_SUBSCRIBER_DEFINE,zbus_chan_pub/read/notify, ...) so thatexisting Zephyr application code and documentation translate directly.
Original zbus by Rodrigo Peixoto (Apache-2.0); copyright preserved in
the derived files.
Ported features: listeners (synchronous callbacks), subscribers
(queue of channel references), message subscribers (ordered message
copies), async listeners (callback on the LP work queue), a deferred
ISR-safe publisher (
ZBUS_ISR_PUBLISHER_DEFINE/zbus_isr_pub),runtime observers, per-observation notification masks, observer
enable/disable, message validators, channel user data, publish
statistics, lookup by name/id and channel/observer iteration.
Everything is built on native NuttX primitives, with no compatibility
shim layer:
k_semchannel lock + priority boost (HLP)sem_t+ nativeCONFIG_PRIORITY_INHERITANCEk_msgq/k_fifo+net_bufpoolsfile_mq_*), lazy-opened;mqpayload copy replacesnet_bufentirelyk_workwork_queue()(LP queue)SYS_INITpthread_once()include/nuttx/iterable_sections.h(companion PR)k_timeout_tCLOCK_MONOTONICdeadlines (ZBUS_NO_WAIT/ZBUS_FOREVER)Board integration (why one linker-script line matters here)
In Zephyr, zbus works on every board out of the box because boards have
no linker scripts: a single common per-arch linker template already
includes the shared
common-rom.ld/common-ram.ldfragments where thezbus iterable sections are collected. In NuttX each board owns its
.ld,so an adopting board needs one of:
#include <nuttx/linker/common-{rom,ram}.ld>lines in its boardscript (done for
linum-stm32h753bi, the first adopter, in thecompanion PR), or
CONFIG_ZBUS_LINKER_INSERT=y(zero-touchINSERT AFTERmode, withthe MEMORY-region constraint documented in the companion PR).
This is documented in the Kconfig help, in
README.rstand in theSphinx page added by the companion PR.
Also included:
examples/zbus(CONFIG_EXAMPLES_ZBUS): a runnable demo with one channel,one listener, one subscriber, runtime masking.
testing/zbus(CONFIG_TESTING_ZBUS): a cmocka suite, 16 testscovering the full API surface: pub/read/listener/subscriber,
multi-channel index isolation, validators, message subscribers
(ordered copies), bit-exact float/double payload delivery
(sensor-style message with a float-math validator rejecting
non-finite samples), claim/finish/notify, masks, enable/disable,
runtime observers (error paths included), async listeners (bursts),
ISR publisher, from_name/from_id, iteration, accessors, and
timeout/overflow semantics.
Not ported (documented in the "Not ported" section of the docs):
multi-domain proxy agent (experimental upstream), direct publishing from
ISRs (the deferred
zbus_isr_pubhelper covers the use case), thepriority-boost/HLP scheme (superseded by native priority inheritance)
and the
net_bufpool machinery (unnecessary with mq payload copies).Impact
CONFIG_ZBUS, defaultn); no impactwhen disabled.
file_mq_*so queuessurvive the creating task, since
mqd_tis a per-task fd in NuttX);CONFIG_MQ_MAXMSGSIZE >= sizeof(pointer) + CONFIG_ZBUS_MSG_SUBSCRIBER_MAX_MSG_SIZEwhen message subscribers areenabled (checked and documented in Kconfig); board linker integration
as described above.
the upstream zbus diagrams, Apache-2.0) lands with the companion
nuttx PR.
retained.
Testing
Host: Ubuntu 24.04.4 x86_64, Arm GNU Toolchain 13.2.rel1
(arm-none-eabi-gcc 13.2.1).
Target: linum-stm32h753bi:zbus (STM32H753BI; board configuration
added by the companion nuttx PR):
CONFIG_ZBUSwith all optionsenabled (message/async/ISR observers, runtime observers),
CONFIG_EXAMPLES_ZBUS,CONFIG_TESTING_ZBUS. Flashed via ST-LINK V3;console on the ST-LINK VCP.
Example (
zbus): 5 messages published; listener correctly skips themasked message #4; listener notified before the subscriber (observer
priority order):
cmocka suite: 16/16 passing, run twice in the same boot (guards
against the lazy-init/fd-lifetime regression class):
(The
could not notify observer: -42line duringtest_timeoutsis theexpected
-ENOMSGqueue-overflow path being exercised.)Builds: Make (
make -j) and CMake (cmake -GNinja+ninja) bothOK for the target config;
_zbus_*iterable-section symbols verified inthe map on both. Stock build with
CONFIG_ZBUSdisabled: no zbussections/symbols (no-op).
tools/checkpatch.sh -gon the commit: allchecks pass.
The companion nuttx PR adds a
linum-stm32h753bi:zbusboardconfiguration reproducing this exact run
(
./tools/configure.sh linum-stm32h753bi:zbus); since that defconfigenables Kconfig symbols introduced here, the two PRs should land
together.