From c7ccb4f5323b577d080021f364fef66e9514b6be Mon Sep 17 00:00:00 2001 From: EvalZero Date: Sun, 17 Mar 2019 16:01:12 +0800 Subject: [PATCH] [components][drivers][workqueue] add system default workqueue and delayed work. --- components/drivers/Kconfig | 14 ++ components/drivers/include/ipc/workqueue.h | 50 +++- components/drivers/src/workqueue.c | 266 +++++++++++++++++---- 3 files changed, 272 insertions(+), 58 deletions(-) diff --git a/components/drivers/Kconfig b/components/drivers/Kconfig index 137402489a..39ed9c7709 100755 --- a/components/drivers/Kconfig +++ b/components/drivers/Kconfig @@ -8,6 +8,20 @@ if RT_USING_DEVICE_IPC config RT_PIPE_BUFSZ int "Set pipe buffer size" default 512 + + config RT_USING_SYSTEM_WORKQUEUE + bool "Using system default workqueue + default n + + if RT_USING_SYSTEM_WORKQUEUE + config RT_SYSTEM_WORKQUEUE_STACKSIZE + int "The stack size for system workqueue thread" + default 512 + + config RT_SYSTEM_WORKQUEUE_PRIORITY + int "The priority level of system workqueue thread" + default "23 + endif endif config RT_USING_SERIAL diff --git a/components/drivers/include/ipc/workqueue.h b/components/drivers/include/ipc/workqueue.h index 9e23e181b1..09dc4354a8 100644 --- a/components/drivers/include/ipc/workqueue.h +++ b/components/drivers/include/ipc/workqueue.h @@ -11,13 +11,26 @@ #include +enum +{ + RT_WORK_STATE_PENDING, /* Work item pending state */ +}; + +/** + * work type defitions + */ +enum +{ + RT_WORK_TYPE_DELAYED, +}; + /* workqueue implementation */ struct rt_workqueue { rt_list_t work_list; struct rt_work *work_current; /* current work */ - struct rt_semaphore sem; + struct rt_semaphore sem; rt_thread_t work_thread; }; @@ -25,27 +38,46 @@ struct rt_work { rt_list_t list; - void (*work_func)(struct rt_work* work, void* work_data); + void (*work_func)(struct rt_work *work, void *work_data); void *work_data; + rt_uint16_t flags; + rt_uint16_t type; +}; + +struct rt_delayed_work +{ + struct rt_work work; + struct rt_timer timer; + struct rt_workqueue *workqueue; }; #ifdef RT_USING_HEAP /** * WorkQueue for DeviceDriver */ -struct rt_workqueue *rt_workqueue_create(const char* name, rt_uint16_t stack_size, rt_uint8_t priority); -rt_err_t rt_workqueue_destroy(struct rt_workqueue* queue); -rt_err_t rt_workqueue_dowork(struct rt_workqueue* queue, struct rt_work* work); -rt_err_t rt_workqueue_cancel_work(struct rt_workqueue* queue, struct rt_work* work); -rt_err_t rt_workqueue_cancel_work_sync(struct rt_workqueue* queue, struct rt_work* work); +struct rt_workqueue *rt_workqueue_create(const char *name, rt_uint16_t stack_size, rt_uint8_t priority); +rt_err_t rt_workqueue_destroy(struct rt_workqueue *queue); +rt_err_t rt_workqueue_dowork(struct rt_workqueue *queue, struct rt_work *work); +rt_err_t rt_workqueue_submit_work(struct rt_workqueue *queue, struct rt_work *work, rt_tick_t time); +rt_err_t rt_workqueue_cancel_work(struct rt_workqueue *queue, struct rt_work *work); +rt_err_t rt_workqueue_cancel_work_sync(struct rt_workqueue *queue, struct rt_work *work); -rt_inline void rt_work_init(struct rt_work* work, void (*work_func)(struct rt_work* work, void* work_data), - void* work_data) +#ifdef RT_USING_SYSTEM_WORKQUEUE +rt_err_t rt_work_submit(struct rt_work *work, rt_tick_t time); +rt_err_t rt_work_cancel(struct rt_work *work); +#endif + +rt_inline void rt_work_init(struct rt_work *work, void (*work_func)(struct rt_work *work, void *work_data), + void *work_data) { rt_list_init(&(work->list)); work->work_func = work_func; work->work_data = work_data; } + +void rt_delayed_work_init(struct rt_delayed_work *work, void (*work_func)(struct rt_work *work, + void *work_data), void *work_data); + #endif #endif diff --git a/components/drivers/src/workqueue.c b/components/drivers/src/workqueue.c index 158ea8b69b..f2b1ae81e4 100644 --- a/components/drivers/src/workqueue.c +++ b/components/drivers/src/workqueue.c @@ -13,11 +13,12 @@ #include #ifdef RT_USING_HEAP + rt_inline rt_err_t _workqueue_work_completion(struct rt_workqueue *queue) { rt_err_t result; - - rt_enter_critical(); + + rt_enter_critical(); while (1) { /* try to take condition semaphore */ @@ -40,17 +41,17 @@ rt_inline rt_err_t _workqueue_work_completion(struct rt_workqueue *queue) } } rt_exit_critical(); - + return result; } -static void _workqueue_thread_entry(void* parameter) +static void _workqueue_thread_entry(void *parameter) { rt_base_t level; - struct rt_work* work; - struct rt_workqueue* queue; + struct rt_work *work; + struct rt_workqueue *queue; - queue = (struct rt_workqueue*) parameter; + queue = (struct rt_workqueue *) parameter; RT_ASSERT(queue != RT_NULL); while (1) @@ -67,6 +68,7 @@ static void _workqueue_thread_entry(void* parameter) work = rt_list_entry(queue->work_list.next, struct rt_work, list); rt_list_remove(&(work->list)); queue->work_current = work; + work->flags &= ~RT_WORK_STATE_PENDING; rt_hw_interrupt_enable(level); /* do work */ @@ -81,11 +83,156 @@ static void _workqueue_thread_entry(void* parameter) } } -struct rt_workqueue *rt_workqueue_create(const char* name, rt_uint16_t stack_size, rt_uint8_t priority) +static rt_err_t _workqueue_submit_work(struct rt_workqueue *queue, struct rt_work *work) +{ + rt_base_t level; + + level = rt_hw_interrupt_disable(); + if (work->flags & RT_WORK_STATE_PENDING) + { + rt_hw_interrupt_enable(level); + return -RT_EBUSY; + } + + if (queue->work_current == work) + { + rt_hw_interrupt_enable(level); + return -RT_EBUSY; + } + + /* NOTE: the work MUST be initialized firstly */ + rt_list_remove(&(work->list)); + + rt_list_insert_after(queue->work_list.prev, &(work->list)); + work->flags |= RT_WORK_STATE_PENDING; + + /* whether the workqueue is doing work */ + if (queue->work_current == RT_NULL) + { + rt_hw_interrupt_enable(level); + /* resume work thread */ + rt_thread_resume(queue->work_thread); + rt_schedule(); + } + else + { + rt_hw_interrupt_enable(level); + } + + return RT_EOK; +} + +static rt_err_t _workqueue_cancel_work(struct rt_workqueue *queue, struct rt_work *work) +{ + rt_base_t level; + + level = rt_hw_interrupt_disable(); + if (queue->work_current == work) + { + rt_hw_interrupt_enable(level); + return -RT_EBUSY; + } + rt_list_remove(&(work->list)); + work->flags &= ~RT_WORK_STATE_PENDING; + rt_hw_interrupt_enable(level); + + return RT_EOK; +} + +static rt_err_t _workqueue_cancel_delayed_work(struct rt_delayed_work *work) +{ + rt_base_t level; + int ret = RT_EOK; + + if (!work->workqueue) + { + ret = -EINVAL; + goto __exit; + } + + if (work->work.flags & RT_WORK_STATE_PENDING) + { + /* Remove from the queue if already submitted */ + ret = rt_workqueue_cancel_work(work->workqueue, &(work->work)); + if (ret) + { + goto __exit; + } + } + else + { + rt_timer_stop(&(work->timer)); + } + + level = rt_hw_interrupt_disable(); + /* Detach from workqueue */ + work->workqueue = RT_NULL; + work->work.flags &= ~(RT_WORK_STATE_PENDING); + rt_hw_interrupt_enable(level); + +__exit: + return ret; +} + +static rt_err_t _workqueue_submit_delayed_work(struct rt_workqueue *queue, + struct rt_delayed_work *work, rt_tick_t ticks) +{ + rt_base_t level; + int ret = RT_EOK; + + + /* Work cannot be active in multiple queues */ + if (work->workqueue && work->workqueue != queue) + { + ret = -RT_EINVAL; + goto __exit; + } + + /* Cancel if work has been submitted */ + if (work->workqueue == queue) + { + ret = _workqueue_cancel_delayed_work(work); + if (ret < 0) + { + goto __exit; + } + } + + level = rt_hw_interrupt_disable(); + /* Attach workqueue so the timeout callback can submit it */ + work->workqueue = queue; + rt_hw_interrupt_enable(level); + + if (!ticks) + { + /* Submit work if no ticks is 0 */ + _workqueue_submit_work(work->workqueue, &(work->work)); + } + else + { + /* Add timeout */ + rt_timer_control(&(work->timer), RT_TIMER_CTRL_SET_TIME, &ticks); + rt_timer_start(&(work->timer)); + } + +__exit: + return ret; +} + +static void _delayed_work_timeout_handler(void *parameter) +{ + struct rt_delayed_work *delayed_work; + + delayed_work = (struct rt_delayed_work *)parameter; + rt_timer_stop(&(delayed_work->timer)); + _workqueue_submit_work(delayed_work->workqueue, &(delayed_work->work)); +} + +struct rt_workqueue *rt_workqueue_create(const char *name, rt_uint16_t stack_size, rt_uint8_t priority) { struct rt_workqueue *queue = RT_NULL; - queue = (struct rt_workqueue*)RT_KERNEL_MALLOC(sizeof(struct rt_workqueue)); + queue = (struct rt_workqueue *)RT_KERNEL_MALLOC(sizeof(struct rt_workqueue)); if (queue != RT_NULL) { /* initialize work list */ @@ -107,7 +254,7 @@ struct rt_workqueue *rt_workqueue_create(const char* name, rt_uint16_t stack_siz return queue; } -rt_err_t rt_workqueue_destroy(struct rt_workqueue* queue) +rt_err_t rt_workqueue_destroy(struct rt_workqueue *queue) { RT_ASSERT(queue != RT_NULL); @@ -117,37 +264,30 @@ rt_err_t rt_workqueue_destroy(struct rt_workqueue* queue) return RT_EOK; } -rt_err_t rt_workqueue_dowork(struct rt_workqueue* queue, struct rt_work* work) +rt_err_t rt_workqueue_dowork(struct rt_workqueue *queue, struct rt_work *work) { - rt_base_t level; RT_ASSERT(queue != RT_NULL); RT_ASSERT(work != RT_NULL); - level = rt_hw_interrupt_disable(); - if (queue->work_current == work) - { - rt_hw_interrupt_enable(level); - return -RT_EBUSY; - } - - /* NOTE: the work MUST be initialized firstly */ - rt_list_remove(&(work->list)); - - rt_list_insert_after(queue->work_list.prev, &(work->list)); - /* whether the workqueue is doing work */ - if (queue->work_current == RT_NULL) - { - rt_hw_interrupt_enable(level); - /* resume work thread */ - rt_thread_resume(queue->work_thread); - rt_schedule(); - } - else rt_hw_interrupt_enable(level); - - return RT_EOK; + return _workqueue_submit_work(queue, work); } -rt_err_t rt_workqueue_critical_work(struct rt_workqueue* queue, struct rt_work* work) +rt_err_t rt_workqueue_submit_work(struct rt_workqueue *queue, struct rt_work *work, rt_tick_t time) +{ + RT_ASSERT(queue != RT_NULL); + RT_ASSERT(work != RT_NULL); + + if (work->type & RT_WORK_TYPE_DELAYED) + { + return _workqueue_submit_delayed_work(queue, (struct rt_delayed_work *)work, time); + } + else + { + return _workqueue_submit_work(queue, work); + } +} + +rt_err_t rt_workqueue_critical_work(struct rt_workqueue *queue, struct rt_work *work) { rt_base_t level; RT_ASSERT(queue != RT_NULL); @@ -176,26 +316,22 @@ rt_err_t rt_workqueue_critical_work(struct rt_workqueue* queue, struct rt_work* return RT_EOK; } -rt_err_t rt_workqueue_cancel_work(struct rt_workqueue* queue, struct rt_work* work) +rt_err_t rt_workqueue_cancel_work(struct rt_workqueue *queue, struct rt_work *work) { - rt_base_t level; - RT_ASSERT(queue != RT_NULL); RT_ASSERT(work != RT_NULL); - level = rt_hw_interrupt_disable(); - if (queue->work_current == work) + if (work->type & RT_WORK_TYPE_DELAYED) { - rt_hw_interrupt_enable(level); - return -RT_EBUSY; + return _workqueue_cancel_delayed_work((struct rt_delayed_work *)work); + } + else + { + return _workqueue_cancel_work(queue, work); } - rt_list_remove(&(work->list)); - rt_hw_interrupt_enable(level); - - return RT_EOK; } -rt_err_t rt_workqueue_cancel_work_sync(struct rt_workqueue* queue, struct rt_work* work) +rt_err_t rt_workqueue_cancel_work_sync(struct rt_workqueue *queue, struct rt_work *work) { rt_base_t level; @@ -212,12 +348,13 @@ rt_err_t rt_workqueue_cancel_work_sync(struct rt_workqueue* queue, struct rt_wor { rt_list_remove(&(work->list)); } + work->flags &= ~RT_WORK_STATE_PENDING; rt_hw_interrupt_enable(level); return RT_EOK; } -rt_err_t rt_workqueue_cancel_all_work(struct rt_workqueue* queue) +rt_err_t rt_workqueue_cancel_all_work(struct rt_workqueue *queue) { struct rt_list_node *node, *next; RT_ASSERT(queue != RT_NULL); @@ -233,5 +370,36 @@ rt_err_t rt_workqueue_cancel_all_work(struct rt_workqueue* queue) return RT_EOK; } -#endif +void rt_delayed_work_init(struct rt_delayed_work *work, void (*work_func)(struct rt_work *work, + void *work_data), void *work_data) +{ + rt_work_init(&(work->work), work_func, work_data); + work->work.type = RT_WORK_TYPE_DELAYED; + rt_timer_init(&(work->timer), "work", _delayed_work_timeout_handler, work, 0, + RT_TIMER_FLAG_ONE_SHOT | RT_TIMER_FLAG_SOFT_TIMER); +} +#ifdef RT_USING_SYSTEM_WORKQUEUE +static struct rt_workqueue *sys_workq; + +rt_err_t rt_work_submit(struct rt_work *work, rt_tick_t time) +{ + return rt_workqueue_submit_work(sys_workq, work, time); +} + +rt_err_t rt_work_cancel(struct rt_work *work) +{ + return rt_workqueue_cancel_work(sys_workq, work); +} + +static int rt_work_sys_workqueue_init(void) +{ + sys_workq = rt_workqueue_create("sys_work", RT_SYSTEM_WORKQUEUE_STACKSIZE * 4, + RT_SYSTEM_WORKQUEUE_PRIORITY); + + return RT_EOK; +} + +INIT_DEVICE_EXPORT(rt_work_sys_workqueue_init); +#endif +#endif