JorgeGzm commented on code in PR #3743: URL: https://github.com/apache/nuttx-apps/pull/3743#discussion_r3890851288
########## system/zbus/zbus.c: ########## @@ -0,0 +1,973 @@ +/**************************************************************************** + * apps/system/zbus/zbus.c + * + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright (c) 2022 Rodrigo Peixoto <[email protected]> + * Copyright (c) 2026 NuttX port + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. You may obtain + * a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + * + ****************************************************************************/ + +/**************************************************************************** + * Included Files + ****************************************************************************/ + +#include <nuttx/config.h> +#include <nuttx/compiler.h> +#include <nuttx/mqueue.h> + +#include <fcntl.h> +#include <inttypes.h> +#include <mqueue.h> +#include <pthread.h> +#include <sched.h> +#include <stdio.h> +#include <stdlib.h> +#include <syslog.h> + +#include <system/zbus.h> + +#include "zbus_priv.h" + +/**************************************************************************** + * Private Data + ****************************************************************************/ + +static pthread_once_t g_zbus_once = PTHREAD_ONCE_INIT; + +/* Protects observer enabled flags and observation masks */ + +static pthread_mutex_t g_zbus_obs_lock = PTHREAD_MUTEX_INITIALIZER; + +/**************************************************************************** + * Private Functions + ****************************************************************************/ + +/**************************************************************************** + * Name: zb_ts_add_ms / zb_ts_cmp / zb_ts_sub + * + * Description: + * Small timespec helpers. + * + ****************************************************************************/ + +static void zb_ts_add_ms(struct timespec *ts, int32_t ms) +{ + ts->tv_sec += ms / 1000; + ts->tv_nsec += (long)(ms % 1000) * 1000000L; + if (ts->tv_nsec >= 1000000000L) + { + ts->tv_sec += 1; + ts->tv_nsec -= 1000000000L; + } +} + +static int zb_ts_cmp(const struct timespec *a, const struct timespec *b) +{ + if (a->tv_sec != b->tv_sec) + { + return (a->tv_sec < b->tv_sec) ? -1 : 1; + } + + if (a->tv_nsec != b->tv_nsec) + { + return (a->tv_nsec < b->tv_nsec) ? -1 : 1; + } + + return 0; +} + +/**************************************************************************** + * Name: zb_deadline_to_realtime + * + * Description: + * Convert the remaining time of a monotonic deadline into an absolute + * CLOCK_REALTIME timespec as required by mq_timedsend/mq_timedreceive. + * + ****************************************************************************/ + +static void zb_deadline_to_realtime(const struct zb_deadline *d, + struct timespec *rt) +{ + struct timespec now; + + clock_gettime(CLOCK_REALTIME, rt); + + if (d->mode == ZB_DEADLINE_ABS) + { + clock_gettime(CLOCK_MONOTONIC, &now); + if (zb_ts_cmp(&now, &d->abs) < 0) + { + rt->tv_sec += d->abs.tv_sec - now.tv_sec; + rt->tv_nsec += d->abs.tv_nsec - now.tv_nsec; + while (rt->tv_nsec >= 1000000000L) + { + rt->tv_sec += 1; + rt->tv_nsec -= 1000000000L; + } + + while (rt->tv_nsec < 0) + { + rt->tv_sec -= 1; + rt->tv_nsec += 1000000000L; + } + } + } +} + +/**************************************************************************** + * Name: zb_mq_send / zb_mq_recv + * + * Description: + * Message queue send/receive honoring a zb_deadline. Following the + * Zephyr k_msgq semantics, a no-wait failure returns -ENOMSG and a + * timeout returns -EAGAIN. + * + ****************************************************************************/ + +static int zb_mq_send(struct file *mq, const char *buf, size_t len, + const struct zb_deadline *d) +{ + struct timespec rt; + int ret; + + if (mq->f_inode == NULL) + { + return -ENODEV; + } + + if (d->mode == ZB_DEADLINE_FOREVER) + { + do + { + ret = file_mq_send(mq, buf, len, 0); + } + while (ret == -EINTR); + } + else + { + zb_deadline_to_realtime(d, &rt); + do + { + ret = file_mq_timedsend(mq, buf, len, 0, &rt); + } + while (ret == -EINTR); + } + + if (ret == -ETIMEDOUT) + { + return (d->mode == ZB_DEADLINE_NOWAIT) ? -ENOMSG : -EAGAIN; + } + + return ret; +} + +static ssize_t zb_mq_recv(struct file *mq, char *buf, size_t len, + const struct zb_deadline *d) +{ + struct timespec rt; + ssize_t ret; + + if (mq->f_inode == NULL) + { + return -ENODEV; + } + + if (d->mode == ZB_DEADLINE_FOREVER) + { + do + { + ret = file_mq_receive(mq, buf, len, NULL); + } + while (ret == -EINTR); + } + else + { + zb_deadline_to_realtime(d, &rt); + do + { + ret = file_mq_timedreceive(mq, buf, len, NULL, &rt); + } + while (ret == -EINTR); + } + + if (ret == -ETIMEDOUT) + { + return (d->mode == ZB_DEADLINE_NOWAIT) ? -ENOMSG : -EAGAIN; + } + + return ret; +} + +#ifdef CONFIG_ZBUS_ASYNC_LISTENER + +/**************************************************************************** + * Name: zb_async_listener_task + * + * Description: + * Dedicated task of an async listener: block on the listener's queue + * and invoke its callback for every message copy, from an aligned + * buffer. argv[1] carries the observer address. + * + ****************************************************************************/ + +static int zb_async_listener_task(int argc, FAR char *argv[]) +{ + const struct zbus_observer *obs; + char buf[sizeof(struct zbus_channel *) + + CONFIG_ZBUS_MSG_SUBSCRIBER_MAX_MSG_SIZE]; + uint8_t msg[CONFIG_ZBUS_MSG_SUBSCRIBER_MAX_MSG_SIZE] aligned_data(8); + const struct zbus_channel *chan; + struct zb_deadline d; + ssize_t nbytes; + + if (argc < 2) + { + return EXIT_FAILURE; + } + + obs = (const struct zbus_observer *)(uintptr_t)strtoul(argv[1], NULL, 16); + d.mode = ZB_DEADLINE_FOREVER; + + for (; ; ) + { + nbytes = zb_mq_recv(&obs->data->mq, buf, sizeof(buf), &d); + if (nbytes < (ssize_t)sizeof(struct zbus_channel *)) + { + continue; + } + + memcpy(&chan, buf, sizeof(chan)); + memcpy(msg, buf + sizeof(chan), nbytes - sizeof(chan)); + obs->async_callback(chan, msg); + } + + return EXIT_SUCCESS; +} + +/**************************************************************************** + * Name: zb_async_listener_start + * + * Description: + * Spawn the task serving an async listener. A task rather than a + * pthread: the lazy init runs in the context of the first API caller, + * and a pthread would die with that caller's task group. + * + ****************************************************************************/ + +static int zb_async_listener_start(const struct zbus_observer *obs) +{ + char arg[2 + sizeof(uintptr_t) * 2 + 1]; + FAR char *argv[2]; + int pid; + + snprintf(arg, sizeof(arg), "%" PRIxPTR, (uintptr_t)obs); + argv[0] = arg; + argv[1] = NULL; + + pid = task_create("zbus_async", CONFIG_ZBUS_ASYNC_LISTENER_PRIORITY, + CONFIG_ZBUS_ASYNC_LISTENER_STACKSIZE, + zb_async_listener_task, argv); + if (pid < 0) + { + return -errno; + } + + obs->data->pid = pid; + return 0; +} + +#endif /* CONFIG_ZBUS_ASYNC_LISTENER */ + +/**************************************************************************** + * Name: zbus_init_fn + * + * Description: + * One-time initialization: compute the observation index boundaries of + * every channel (relies on the linker sorting the observation section by + * name, which groups entries per channel in priority order) and open the + * notification queues of subscriber-type observers. + * + ****************************************************************************/ + +static void zbus_init_fn(void) Review 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
