sync smart & dfs (#8672)

Signed-off-by: xqyjlj <[email protected]>
Signed-off-by: Shell <[email protected]>
Co-authored-by: xqyjlj <[email protected]>

4 files changed, 320 insertions, 19 deletions

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 <rtdbg.h>
+
+#include <rtdevice.h>
+#include <rtatomic.h>
+#include <rtthread.h>
+
+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 <rthw.h>
 #include <rtdevice.h>
 #include <stdint.h>
 #include <sys/errno.h>
+#include <ipc/condvar.h>
 
 #if defined(RT_USING_POSIX_DEVIO) && defined(RT_USING_POSIX_PIPE)
 #include <unistd.h>
@@ -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 <rtdbg.h>
 
 #include <stdint.h>
 #include <rthw.h>
@@ -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_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_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)