Mirrors_Framework/rt-thread
1
0
Fork 0
mirror of https://github.com/RT-Thread/rt-thread.git synced 2026-09-21 18:57:16 +08:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

sync smart & dfs (#8672)

Browse Source 
Signed-off-by: xqyjlj <xqyjlj@126.com>
Signed-off-by: Shell <smokewood@qq.com>
Co-authored-by: xqyjlj <xqyjlj@126.com>
This commit is contained in:
Shell
2024-03-28 23:42:56 +08:00
committed by GitHub
co-authored by xqyjlj
parent 40e26f4909
commit 83e95bdff4
131 changed files with 14954 additions and 6478 deletions
Download Patch File Download Diff File Expand all files Collapse all files
components/drivers/ipc/completion.c
+30 -2
View File
@@ -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);
components/drivers/ipc/condvar.c
+173
View File
@@ -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;
}
components/drivers/ipc/pipe.c
+34 -11
View File
@@ -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;
}
components/drivers/ipc/waitqueue.c
+83 -6
View File
@@ -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)