xiaoxiang781216 commented on code in PR #3743: URL: https://github.com/apache/nuttx-apps/pull/3743#discussion_r3888433497
########## 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: 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: 1. use file_mq API directly in userspace 2. 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 -- 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]
