From 83e95bdff4c48c661b50e1eec56d5fd293b28fbb Mon Sep 17 00:00:00 2001 From: Shell Date: Thu, 28 Mar 2024 23:42:56 +0800 Subject: sync smart & dfs (#8672) Signed-off-by: xqyjlj Signed-off-by: Shell Co-authored-by: xqyjlj --- components/drivers/ipc/completion.c | 32 ++++++- components/drivers/ipc/condvar.c | 173 ++++++++++++++++++++++++++++++++++++ components/drivers/ipc/pipe.c | 45 +++++++--- components/drivers/ipc/waitqueue.c | 89 +++++++++++++++++-- 4 files changed, 320 insertions(+), 19 deletions(-) create mode 100644 components/drivers/ipc/condvar.c (limited to 'components/drivers/ipc') diff --git a/components/drivers/ipc/completion.c b/components/drivers/ipc/completion.c index d74fcdc8d1..824f538129 100644 --- a/components/drivers/ipc/completion.c +++ b/components/drivers/ipc/completion.c @@ -134,7 +134,7 @@ RTM_EXPORT(rt_completion_wait); * * @param completion is a pointer to a completion object. */ -void rt_completion_done(struct rt_completion *completion) +static int _completion_done(struct rt_completion *completion) { rt_base_t level; rt_err_t error; @@ -145,7 +145,7 @@ void rt_completion_done(struct rt_completion *completion) if (RT_COMPLETION_FLAG(completion) == RT_COMPLETED) { rt_spin_unlock_irqrestore(&_completion_lock, level); - return; + return -RT_EBUSY; } suspend_thread = RT_COMPLETION_THREAD(completion); @@ -160,10 +160,38 @@ void rt_completion_done(struct rt_completion *completion) LOG_D("%s: failed to resume thread", __func__); } } + else + { + /* no thread waiting */ + error = -RT_EEMPTY; + } completion->susp_thread_n_flag = RT_COMPLETION_NEW_STAT(RT_NULL, RT_COMPLETED); rt_spin_unlock_irqrestore(&_completion_lock, level); + + return error; +} + +/** + * @brief This function indicates a completion has done. + * + * @param completion is a pointer to a completion object. + */ +void rt_completion_done(struct rt_completion *completion) +{ + _completion_done(completion); } RTM_EXPORT(rt_completion_done); +/** + * @brief This function indicates a completion has done and wakeup the thread + * + * @param completion is a pointer to a completion object. + */ +rt_err_t rt_completion_wakeup(struct rt_completion *completion) +{ + return _completion_done(completion); +} +RTM_EXPORT(rt_completion_wakeup); + diff --git a/components/drivers/ipc/condvar.c b/components/drivers/ipc/condvar.c new file mode 100644 index 0000000000..1eb42dc9ad --- /dev/null +++ b/components/drivers/ipc/condvar.c @@ -0,0 +1,173 @@ +/* + * Copyright (c) 2006-2023, RT-Thread Development Team + * + * SPDX-License-Identifier: Apache-2.0 + * + * Change Logs: + * Date Author Notes + * 2023-11-20 Shell Support of condition variable + */ +#define DBG_TAG "ipc.condvar" +#define DBG_LVL DBG_INFO +#include + +#include +#include +#include + +static struct rt_spinlock _local_cv_queue_lock = RT_SPINLOCK_INIT; + +#define CV_ASSERT_LOCKED(cv) \ + RT_ASSERT(!(cv)->waiting_mtx || \ + rt_mutex_get_owner((rt_mutex_t)(cv)->waiting_mtx) == \ + rt_thread_self()) + +void rt_condvar_init(rt_condvar_t cv, char *name) +{ +#ifdef USING_RT_OBJECT + /* TODO: support rt object */ + rt_object_init(); +#endif + + rt_wqueue_init(&cv->event); + rt_atomic_store(&cv->waiters_cnt, 0); + rt_atomic_store(&cv->waiting_mtx, 0); +} + +static int _waitq_inqueue(rt_wqueue_t *queue, struct rt_wqueue_node *node, + rt_tick_t timeout, int suspend_flag) +{ + rt_thread_t tcb = node->polling_thread; + rt_timer_t timer = &(tcb->thread_timer); + rt_err_t ret; + + if (queue->flag != RT_WQ_FLAG_WAKEUP) + { + ret = rt_thread_suspend_with_flag(tcb, suspend_flag); + if (ret == RT_EOK) + { + rt_wqueue_add(queue, node); + if (timeout != RT_WAITING_FOREVER) + { + rt_timer_control(timer, RT_TIMER_CTRL_SET_TIME, &timeout); + + rt_timer_start(timer); + } + } + } + else + { + ret = RT_EOK; + } + + return ret; +} + +#define INIT_WAITQ_NODE(node) \ + { \ + .polling_thread = rt_thread_self(), .key = 0, \ + .wakeup = __wqueue_default_wake, .wqueue = &cv->event, \ + .list = RT_LIST_OBJECT_INIT(node.list) \ + } + +int rt_condvar_timedwait(rt_condvar_t cv, rt_mutex_t mtx, int suspend_flag, + rt_tick_t timeout) +{ + rt_err_t acq_mtx_succ, rc; + rt_atomic_t waiting_mtx; + struct rt_wqueue_node node = INIT_WAITQ_NODE(node); + + /* not allowed in IRQ & critical section */ + RT_DEBUG_SCHEDULER_AVAILABLE(1); + CV_ASSERT_LOCKED(cv); + + /** + * for the worst case, this is racy with the following works to reset field + * before mutex is taken. The spinlock then comes to rescue. + */ + rt_spin_lock(&_local_cv_queue_lock); + waiting_mtx = rt_atomic_load(&cv->waiting_mtx); + if (!waiting_mtx) + acq_mtx_succ = rt_atomic_compare_exchange_strong( + &cv->waiting_mtx, &waiting_mtx, (size_t)mtx); + else + acq_mtx_succ = 0; + + rt_spin_unlock(&_local_cv_queue_lock); + + if (acq_mtx_succ == 1 || waiting_mtx == (size_t)mtx) + { + rt_atomic_add(&cv->waiters_cnt, 1); + + rt_enter_critical(); + + if (suspend_flag == RT_INTERRUPTIBLE) + rc = _waitq_inqueue(&cv->event, &node, timeout, RT_INTERRUPTIBLE); + else /* UNINTERRUPTIBLE is forbidden, since it's not safe for user space */ + rc = _waitq_inqueue(&cv->event, &node, timeout, RT_KILLABLE); + + acq_mtx_succ = rt_mutex_release(mtx); + RT_ASSERT(acq_mtx_succ == 0); + rt_exit_critical(); + + if (rc == RT_EOK) + { + rt_schedule(); + + rc = rt_get_errno(); + rc = rc > 0 ? -rc : rc; + } + else + { + LOG_D("%s() failed to suspend", __func__); + } + + rt_wqueue_remove(&node); + + rt_spin_lock(&_local_cv_queue_lock); + if (rt_atomic_add(&cv->waiters_cnt, -1) == 1) + { + waiting_mtx = (size_t)mtx; + acq_mtx_succ = rt_atomic_compare_exchange_strong(&cv->waiting_mtx, + &waiting_mtx, 0); + RT_ASSERT(acq_mtx_succ == 1); + } + rt_spin_unlock(&_local_cv_queue_lock); + + acq_mtx_succ = rt_mutex_take(mtx, RT_WAITING_FOREVER); + RT_ASSERT(acq_mtx_succ == 0); + } + else + { + LOG_D("%s: conflict waiting mutex", __func__); + rc = -EBUSY; + } + + return rc; +} + +/** Keep in mind that we always operating when cv.waiting_mtx is taken */ + +int rt_condvar_signal(rt_condvar_t cv) +{ + CV_ASSERT_LOCKED(cv); + + /* to avoid spurious wakeups */ + if (rt_atomic_load(&cv->waiters_cnt) > 0) + rt_wqueue_wakeup(&cv->event, 0); + + cv->event.flag = 0; + return 0; +} + +int rt_condvar_broadcast(rt_condvar_t cv) +{ + CV_ASSERT_LOCKED(cv); + + /* to avoid spurious wakeups */ + if (rt_atomic_load(&cv->waiters_cnt) > 0) + rt_wqueue_wakeup_all(&cv->event, 0); + + cv->event.flag = 0; + return 0; +} diff --git a/components/drivers/ipc/pipe.c b/components/drivers/ipc/pipe.c index acd0c2def6..c39cd6bdc6 100644 --- a/components/drivers/ipc/pipe.c +++ b/components/drivers/ipc/pipe.c @@ -9,11 +9,13 @@ * 2017-11-08 JasonJiaJie fix memory leak issue when close a pipe. * 2023-06-28 shell return POLLHUP when writer closed its channel on poll() * fix flag test on pipe_fops_open() + * 2023-12-02 shell Make read pipe operation interruptable. */ #include #include #include #include +#include #if defined(RT_USING_POSIX_DEVIO) && defined(RT_USING_POSIX_PIPE) #include @@ -69,12 +71,12 @@ static int pipe_fops_open(struct dfs_file *fd) if ((fd->flags & O_ACCMODE) == O_RDONLY) { - pipe->reader = 1; + pipe->reader += 1; } if ((fd->flags & O_ACCMODE) == O_WRONLY) { - pipe->writer = 1; + pipe->writer += 1; } if (fd->vnode->ref_count == 1) { @@ -86,6 +88,21 @@ static int pipe_fops_open(struct dfs_file *fd) } } + if ((fd->flags & O_ACCMODE) == O_RDONLY && !pipe->writer) + { + /* wait for partner */ + rc = rt_condvar_timedwait(&pipe->waitfor_parter, &pipe->lock, + RT_INTERRUPTIBLE, RT_WAITING_FOREVER); + if (rc != 0) + { + pipe->reader--; + } + } + else if ((fd->flags & O_ACCMODE) == O_WRONLY) + { + rt_condvar_broadcast(&pipe->waitfor_parter); + } + __exit: rt_mutex_release(&pipe->lock); @@ -117,12 +134,12 @@ static int pipe_fops_close(struct dfs_file *fd) if ((fd->flags & O_RDONLY) == O_RDONLY) { - pipe->reader = 0; + pipe->reader -= 1; } if ((fd->flags & O_WRONLY) == O_WRONLY) { - pipe->writer = 0; + pipe->writer -= 1; while (!rt_list_isempty(&pipe->reader_queue.waiting_list)) { rt_wqueue_wakeup(&pipe->reader_queue, (void*)POLLIN); @@ -234,7 +251,8 @@ static ssize_t pipe_fops_read(struct dfs_file *fd, void *buf, size_t count) rt_mutex_release(&pipe->lock); rt_wqueue_wakeup(&pipe->writer_queue, (void*)POLLOUT); - rt_wqueue_wait(&pipe->reader_queue, 0, -1); + if (rt_wqueue_wait_interruptible(&pipe->reader_queue, 0, -1) == -RT_EINTR) + return -EINTR; rt_mutex_take(&pipe->lock, RT_WAITING_FOREVER); } } @@ -309,7 +327,8 @@ static ssize_t pipe_fops_write(struct dfs_file *fd, const void *buf, size_t coun rt_mutex_release(&pipe->lock); rt_wqueue_wakeup(&pipe->reader_queue, (void*)POLLIN); /* pipe full, waiting on suspended write list */ - rt_wqueue_wait(&pipe->writer_queue, 0, -1); + if (rt_wqueue_wait_interruptible(&pipe->writer_queue, 0, -1) == -RT_EINTR) + return -EINTR; rt_mutex_take(&pipe->lock, -1); } rt_mutex_release(&pipe->lock); @@ -611,6 +630,8 @@ rt_pipe_t *rt_pipe_create(const char *name, int bufsz) rt_mutex_init(&pipe->lock, name, RT_IPC_FLAG_FIFO); rt_wqueue_init(&pipe->reader_queue); rt_wqueue_init(&pipe->writer_queue); + rt_condvar_init(&pipe->waitfor_parter, "piwfp"); + pipe->writer = 0; pipe->reader = 0; @@ -674,6 +695,7 @@ int rt_pipe_delete(const char *name) pipe = (rt_pipe_t *)device; + rt_condvar_detach(&pipe->waitfor_parter); rt_mutex_detach(&pipe->lock); #if defined(RT_USING_POSIX_DEVIO) && defined(RT_USING_POSIX_PIPE) resource_id_put(&id_mgr, pipe->pipeno); @@ -736,11 +758,6 @@ int pipe(int fildes[2]) pipe->is_named = RT_FALSE; /* unamed pipe */ pipe->pipeno = pipeno; rt_snprintf(dev_name, sizeof(dev_name), "/dev/%s", dname); - fildes[0] = open(dev_name, O_RDONLY, 0); - if (fildes[0] < 0) - { - return -1; - } fildes[1] = open(dev_name, O_WRONLY, 0); if (fildes[1] < 0) @@ -749,6 +766,12 @@ int pipe(int fildes[2]) return -1; } + fildes[0] = open(dev_name, O_RDONLY, 0); + if (fildes[0] < 0) + { + return -1; + } + return 0; } diff --git a/components/drivers/ipc/waitqueue.c b/components/drivers/ipc/waitqueue.c index 03901a9ba3..75e14e5844 100644 --- a/components/drivers/ipc/waitqueue.c +++ b/components/drivers/ipc/waitqueue.c @@ -9,7 +9,11 @@ * to blocked thread. * 2022-01-24 THEWON let rt_wqueue_wait return thread->error when using signal * 2023-09-15 xqyjlj perf rt_hw_interrupt_disable/enable + * 2023-11-21 Shell Support wakeup_all */ +#define DBG_TAG "ipc.waitqueue" +#define DBG_LVL DBG_INFO +#include #include #include @@ -64,7 +68,8 @@ int __wqueue_default_wake(struct rt_wqueue_node *wait, void *key) } /** - * @brief This function will wake up a pending thread on the specified waiting queue that meets the conditions. + * @brief This function will wake up a pending thread on the specified + * waiting queue that meets the conditions. * * @param queue is a pointer to the wait queue. * @@ -94,17 +99,89 @@ void rt_wqueue_wakeup(rt_wqueue_t *queue, void *key) entry = rt_list_entry(node, struct rt_wqueue_node, list); if (entry->wakeup(entry, key) == 0) { - rt_thread_resume(entry->polling_thread); - need_schedule = 1; + /** + * even though another thread may interrupt the thread and + * wakeup it meanwhile, we can asuume that condition is ready + */ + entry->polling_thread->error = RT_EOK; + if (!rt_thread_resume(entry->polling_thread)) + { + need_schedule = 1; + + rt_list_remove(&(entry->list)); + + break; + } + } + } + } + rt_spin_unlock_irqrestore(&(queue->spinlock), level); + if (need_schedule) + rt_schedule(); + + return; +} + +/** + * @brief This function will wake up all pending thread on the specified + * waiting queue that meets the conditions. + * + * @param queue is a pointer to the wait queue. + * + * @param key is the wakeup conditions, but it is not effective now, because + * default wakeup function always return 0. + * If user wants to use it, user should define their own wakeup + * function. + */ +void rt_wqueue_wakeup_all(rt_wqueue_t *queue, void *key) +{ + rt_base_t level; + int need_schedule = 0; - rt_list_remove(&(entry->list)); - break; + rt_list_t *queue_list; + struct rt_list_node *node; + struct rt_wqueue_node *entry; + + queue_list = &(queue->waiting_list); + + level = rt_spin_lock_irqsave(&(queue->spinlock)); + /* set wakeup flag in the queue */ + queue->flag = RT_WQ_FLAG_WAKEUP; + + if (!(rt_list_isempty(queue_list))) + { + for (node = queue_list->next; node != queue_list; ) + { + entry = rt_list_entry(node, struct rt_wqueue_node, list); + if (entry->wakeup(entry, key) == 0) + { + /** + * even though another thread may interrupt the thread and + * wakeup it meanwhile, we can asuume that condition is ready + */ + entry->polling_thread->error = RT_EOK; + if (!rt_thread_resume(entry->polling_thread)) + { + need_schedule = 1; + } + else + { + /* wakeup happened too soon that waker hadn't slept */ + LOG_D("%s: Thread resume failed", __func__); + } + node = node->next; + } + else + { + node = node->next; } } } rt_spin_unlock_irqrestore(&(queue->spinlock), level); if (need_schedule) rt_schedule(); + + return; } /** @@ -184,7 +261,7 @@ __exit_wakeup: rt_wqueue_remove(&__wait); - return tid->error; + return tid->error > 0 ? -tid->error : tid->error; } int rt_wqueue_wait(rt_wqueue_t *queue, int condition, int msec) -- cgit v1.3.1