2026/7/19大约 5 分钟附录源码附录
os_workqueue.c
路径: kernel\source\os_workqueue.c
/**
***********************************************************************************************************************
* Copyright (c) 2020, China Mobile Communications Group Co.,Ltd.
*
* 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.
*
* @file os_workqueue.c
*
* @brief This function implements workqueue mechanism.
*
* @revision
* Date Author Notes
* 2020-04-06 OneOS Team First version.
***********************************************************************************************************************
*/
#include <oneos_config.h>
#include <os_errno.h>
#include <os_assert.h>
#include <os_stddef.h>
#include <os_workqueue.h>
#include "os_kernel_internal.h"
#ifdef OS_USING_WORKQUEUE
/* Workqueue depends on semaphore. */
#ifndef OS_USING_SEMAPHORE
#error "Make sure semaphore is enable"
#endif
/* Log tag of workqueue */
#define WORKQ_TAG "WORKQ"
static void _k_workqueue_task_entry(void *parameter)
{
struct os_workqueue *queue;
struct os_work *work;
const os_list_node_t *head;
os_ubase_t irq_key;
queue = (os_workqueue_t *)parameter;
head = &queue->work_list_head;
while (1)
{
(void)os_semaphore_wait(&queue->sem_id, OS_WAIT_FOREVER);
while (1)
{
os_spin_lock_irqsave(&queue->lock, &irq_key);
if (!os_list_empty(head))
{
work = os_list_first_entry(head, os_work_t, work_node);
os_list_del(&work->work_node);
queue->work_current = work;
work->flag = OS_WORK_STAGE_IDLE;
os_spin_unlock_irqrestore(&queue->lock, irq_key);
/*done work*/
work->work_func(work->data);
}
else
{
queue->work_current = OS_NULL;
os_spin_unlock_irqrestore(&queue->lock, irq_key);
break;
}
}
}
}
OS_INLINE void _k_work_commit_to_queue_tail(os_workqueue_t *queue, os_work_t *work)
{
work->flag = OS_WORK_STAGE_PENDING;
os_list_add_tail(&queue->work_list_head, &work->work_node);
/*Optimization: reduce bin semaphore release*/
if (OS_NULL == queue->work_current)
{
(void)os_semaphore_post(&queue->sem_id);
}
}
OS_INLINE void _k_work_commit_to_queue_head(os_workqueue_t *queue, os_work_t *work)
{
work->flag = OS_WORK_STAGE_PENDING;
os_list_add(&queue->work_list_head, &work->work_node);
/*Optimization: reduce bin semaphore release*/
if (OS_NULL == queue->work_current)
{
(void)os_semaphore_post(&queue->sem_id);
}
}
static void _k_work_timeout(void *parameter)
{
os_ubase_t irq_key;
struct os_work *work;
os_workqueue_t *queue;
work = (struct os_work *)parameter;
queue = work->workqueue;
os_spin_lock_irqsave(&queue->lock, &irq_key);
/*If the timer can't cancel,can be cancelled by os_cancel_work used work->flag */
if (OS_WORK_STAGE_DELAY == work->flag)
{
_k_work_commit_to_queue_tail(work->workqueue, work);
}
os_spin_unlock_irqrestore(&queue->lock, irq_key);
}
static void _k_work_wait(void *parameter)
{
(void)os_semaphore_post((os_semaphore_id)parameter);
}
static os_err_t _k_work_stop(os_work_id work_id, os_bool_t sync)
{
os_ubase_t key;
os_err_t ret;
os_workqueue_t *queue;
OS_WORK_DEFINE(wait_work);
os_work_id wait_work_id;
OS_SEMAPHORE_DEFINE(wait_sem);
os_semaphore_id wait_sem_id;
int32_t work_runing;
os_work_t *work;
work = (os_work_t *)work_id;
OS_ASSERT(work);
OS_ASSERT(OS_KOBJ_INITED == work->object_inited);
ret = OS_SUCCESS;
work_runing = 0;
/*The work has not been committed since it was initialized.*/
if (OS_NULL == work->workqueue)
{
ret = OS_FAILURE;
}
else
{
queue = work->workqueue;
if (sync)
{
wait_sem_id = os_semaphore_create(&wait_sem, OS_NULL, 0, 1);
wait_work_id = os_work_init(&wait_work, _k_work_wait, wait_sem_id);
}
os_spin_lock_irqsave(&queue->lock, &key);
if (OS_WORK_STAGE_DELAY == work->flag)
{
(void)os_timer_stop(&work->timer);
work->flag = OS_WORK_STAGE_IDLE;
}
else if (OS_WORK_STAGE_PENDING == work->flag)
{
os_list_del(&work->work_node);
work->flag = OS_WORK_STAGE_IDLE;
}
else
{
/*Work is in progress*/
if (queue->work_current == work)
{
work_runing = 1;
if (sync)
{
_k_work_commit_to_queue_head(work->workqueue, (os_work_t *)&wait_work);
}
else
{
OS_KERN_LOG(KERN_WARNING, WORKQ_TAG, "Work(%p) is running", work);
ret = OS_BUSY;
}
}
}
os_spin_unlock_irqrestore(&queue->lock, key);
if (sync)
{
if (work_runing)
{
(void)os_semaphore_wait(wait_sem_id, OS_WAIT_FOREVER);
}
(void)os_semaphore_destroy(wait_sem_id);
os_work_deinit(wait_work_id);
}
}
return ret;
}
static os_err_t _k_workqueue_init(os_workqueue_t *queue,
const char *name,
void *stack_begin,
uint32_t stack_size,
uint8_t priority,
int32_t cpu_index)
{
os_err_t ret;
os_task_id tid;
queue->work_current = OS_NULL;
queue->object_inited = OS_KOBJ_INITED;
os_list_init(&queue->work_list_head);
os_spin_lock_init(&queue->lock);
ret = OS_SUCCESS;
/*binary semaphore*/
if (OS_NULL != os_semaphore_create(&queue->sem_id, OS_NULL, 0, 1))
{
tid = os_task_create(&queue->worker_task,
stack_begin,
stack_size,
name,
_k_workqueue_task_entry,
queue,
priority);
if (tid)
{
#ifdef OS_USING_SMP
ret = os_task_set_cpu_affinity(&queue->worker_task, cpu_index);
if (ret == OS_SUCCESS)
{
ret = os_task_startup(&queue->worker_task);
}
#else
ret = os_task_startup(&queue->worker_task);
#endif
}
else
{
ret = OS_FAILURE;
}
}
else
{
ret = OS_FAILURE;
}
return ret;
}
/**
***********************************************************************************************************************
* @brief Initialize work.
*
* @attention Before using work, must initialize work.
*
* @param[out] work_id The id of work init.
* @param[in] work The static work control block to be create.
* @param[in] func Callback function for work.
* @param[in] data Private data of work.
*
* @return None.
***********************************************************************************************************************
*/
os_work_id os_work_init(os_work_dummy_t *work, void (*func)(void *data), void *data)
{
os_work_t *os_work;
OS_ASSERT(work);
OS_ASSERT(func);
os_work = (os_work_t *)work;
os_work->work_func = func;
os_work->data = data;
os_work->workqueue = OS_NULL;
os_work->flag = OS_WORK_STAGE_IDLE;
os_work->object_inited = OS_KOBJ_INITED;
os_list_init(&os_work->work_node);
os_timer_create(&os_work->timer, OS_NULL, _k_work_timeout, work, 1, OS_TIMER_FLAG_ONE_SHOT);
return work;
}
/**
***********************************************************************************************************************
* @brief This function will deinitialize the work.
*
* @attention Before deinitialize work, must initialize work.
*
* @param[in] work_id The ID of work to deinit.
* @retval None.
***********************************************************************************************************************
*/
void os_work_deinit(os_work_id work_id)
{
os_work_t *work = (os_work_t *)work_id;
OS_ASSERT(work);
OS_ASSERT(OS_KOBJ_INITED == work->object_inited);
work->object_inited = OS_KOBJ_DEINITED;
os_list_del(&work->work_node);
os_timer_destroy(&work->timer);
return;
}
/**
***********************************************************************************************************************
* @brief Submit a work to the specific workqueue.
*
* @details The work does not resubmit if it is in a queue(OS_WORK_STAGE_PENDING) or in a timeout commit
* (OS_WORK_STAGE_DELAY).If you want to recommit work immediately, call first os_cancel_work() cancel
*the last submission.
*
* @param[in] queue_id The ID of the workqueue.
* @param[in] work_id The ID of the work.
* @param[in] delay_time Delay time. Commit immediately if equals 0.If it is greater than 0, the work will
*be delayed commit.
*
* @return Submit work result.
* @retval OS_SUCCESS Submit work success.
* @retval OS_BUSY The work has been submitted.
* @retval OS_FAILURE Not allowing a work to commit to more than one queue.
***********************************************************************************************************************
*/
os_err_t os_work_run(os_workqueue_id queue_id, os_work_id work_id, os_tick_t delay_time)
{
os_ubase_t irq_key;
os_err_t ret;
os_workqueue_t *queue;
os_work_t *work;
queue = (os_workqueue_t *)queue_id;
work = (os_work_t *)work_id;
OS_ASSERT(queue);
OS_ASSERT(work);
OS_ASSERT(OS_KOBJ_INITED == queue->object_inited);
OS_ASSERT(OS_KOBJ_INITED == work->object_inited);
OS_ASSERT(delay_time < (OS_TICK_MAX / 2));
ret = OS_SUCCESS;
os_spin_lock_irqsave(&queue->lock, &irq_key);
/*Not allowing a work to commit to more than one queue.Limiting concurrent running*/
if (work->workqueue && (queue != work->workqueue))
{
os_spin_unlock_irqrestore(&queue->lock, irq_key);
OS_KERN_LOG(KERN_WARNING, WORKQ_TAG, "Submitting work to more than one workqueue is not allowed");
ret = OS_FAILURE;
}
else
{
if (work->flag == OS_WORK_STAGE_IDLE)
{
work->workqueue = queue;
if (delay_time)
{
work->flag = OS_WORK_STAGE_DELAY;
(void)os_timer_set_timeout_ticks(&work->timer, delay_time);
(void)os_timer_start(&work->timer);
}
else
{
_k_work_commit_to_queue_tail(queue, work);
}
ret = OS_SUCCESS;
}
else
{
OS_KERN_LOG(KERN_WARNING, WORKQ_TAG, "Work(%p) had been submitted to workqueue(%p)", work, queue);
ret = OS_BUSY;
}
os_spin_unlock_irqrestore(&queue->lock, irq_key);
}
return ret;
}
/**
***********************************************************************************************************************
* @brief Cancel the work that has been committed(in a queue (OS_WORK_STAGE_DELAY) or in a timeout commit
* (OS_WORK_STAGE_PENDING).
*
* @attention Work cannot be cancelled while it is in progress, and return OS_BUSY.
*
* @param[in] work_id The ID of the work.
* @param[in] sync How to remove the work, 0 is asynchronous, 1 is synchronize.
*
* @return Cancel work result.
* @retval OS_SUCCESS Cancel work success.
* @retval OS_BUSY The work is in progress.
* @retval OS_FAILURE The work has not been committed since it was initialized.
***********************************************************************************************************************
*/
os_err_t os_work_stop(os_work_id work_id, os_bool_t sync)
{
return _k_work_stop(work_id, sync);
}
/**
***********************************************************************************************************************
* @brief This function dynamic creates a workqueue.
*
* @param[in] queue The static workqueue cb to be create, NULL will dynamic create.
* @param[in] name The name of task in workqueue.
* @param[in] stack_begin Pointer to start of stack, NULL will dynamic create
* @param[in] stack_size The stack size of task in workqueue.
* @param[in] priority The priority of task in workqueue.
* @param[in] cpu_index The workqueue task want bind to that CPU(CPU affinity).-1:no CPU affinity.
*
* @return The ID of workqueue.
* @retval The ID of workqueue.
***********************************************************************************************************************
*/
os_workqueue_id os_workqueue_create(os_workqueue_dummy_t *queue_cb,
const char *name,
void *stack_begin,
uint32_t stack_size,
uint8_t priority,
int32_t cpu_index)
{
os_bool_t flag;
os_workqueue_t *queue;
uint8_t alloc_type;
uint32_t ret;
OS_ASSERT(stack_size > 0);
OS_ASSERT(priority < OS_TASK_PRIORITY_MAX);
OS_ASSERT(cpu_index >= -1);
OS_ASSERT(OS_FALSE == os_is_irq_active());
#ifndef OS_USING_HEAP
OS_ASSERT(OS_NULL != queue_cb);
OS_ASSERT(OS_NULL != stack_begin);
#endif
flag = OS_TRUE;
if (OS_NULL != queue_cb)
{
queue = OS_TYPE_CONVERT(os_workqueue_t *, queue_cb);
alloc_type = OS_ALLOC_TYPE_STATIC;
}
#ifdef OS_USING_HEAP
else
{
queue = (os_workqueue_t *)OS_KERNEL_MALLOC(sizeof(os_workqueue_t));
if (OS_NULL == queue)
{
OS_KERN_LOG(KERN_ERROR, WORKQ_TAG, "Malloc workqueue(%s) failed", name);
flag = OS_FALSE;
}
else
{
alloc_type = OS_KOBJ_ALLOC_TYPE_DYNAMIC;
}
}
#endif
#ifdef OS_USING_HEAP
if ((OS_NULL == stack_begin) && (OS_TRUE == flag))
{
stack_size = OS_ALIGN_UP(stack_size, OS_ARCH_STACK_ALIGN_SIZE);
stack_begin = OS_KERNEL_MALLOC_ALIGN(OS_ARCH_STACK_ALIGN_SIZE, stack_size);
if (OS_NULL == stack_begin)
{
OS_KERN_LOG(KERN_ERROR, WORKQ_TAG, "Malloc stack_begin(%s) failed", name);
if (OS_KOBJ_ALLOC_TYPE_DYNAMIC == alloc_type)
{
OS_KERNEL_FREE(queue);
queue = OS_NULL;
}
flag = OS_FALSE;
}
else
{
alloc_type |= OS_KDATA_ALLOC_TYPE_DYNAMIC;
}
}
#endif
if (OS_TRUE == flag)
{
ret = _k_workqueue_init(queue, name, stack_begin, stack_size, priority, cpu_index);
if (OS_SUCCESS != ret)
{
#ifdef OS_USING_HEAP
if (OS_KOBJ_ALLOC_TYPE_DYNAMIC == (OS_KOBJ_ALLOC_TYPE_DYNAMIC & alloc_type))
{
OS_KERNEL_FREE(queue);
}
if (OS_KDATA_ALLOC_TYPE_DYNAMIC == (OS_KDATA_ALLOC_TYPE_DYNAMIC & alloc_type))
{
OS_KERNEL_FREE(stack_begin);
}
#endif
queue = OS_NULL;
}
}
return OS_TYPE_CONVERT(os_workqueue_id, queue);
}
#ifdef OS_USING_SYSTEM_WORKQUEUE
static OS_TASK_STACK_DEFINE(gs_sys_worker_task_stack, OS_SYSTEM_WORKQUEUE_STACK_SIZE);
OS_WORKQUEUE_DEFINE(gs_sys_workq);
static os_workqueue_id gs_sys_workq_id;
/**
***********************************************************************************************************************
* @brief Initialize system workqueue.
*
* @param None.
*
* @return Initialize system workqueue result.
* @retval OS_SUCCESS Initialize system workqueue success.
* @retval OS_FAILURE Initialize system workqueue failed.
***********************************************************************************************************************
*/
os_err_t os_sys_workqueue_init(void)
{
os_err_t ret = OS_SUCCESS;
gs_sys_workq_id = os_workqueue_create(&gs_sys_workq,
"sys_work",
OS_TASK_STACK_BEGIN_ADDR(gs_sys_worker_task_stack),
OS_TASK_STACK_SIZE(gs_sys_worker_task_stack),
OS_SYSTEM_WORKQUEUE_PRIORITY,
-1);
if (!gs_sys_workq_id)
{
OS_KERN_LOG(KERN_ERROR, WORKQ_TAG, "Create system workqueue failed");
ret = OS_FAILURE;
}
return ret;
}
OS_INIT_CALL(os_sys_workqueue_init, OS_INIT_LEVEL_POST_KERNEL, OS_INIT_SUBLEVEL_HIGH);
/**
***********************************************************************************************************************
* @brief Submit a work to the system workqueue.
*
* @details When the delay_time is greater than 0, the work will be delayed.
*
* @param[in] work_id The ID of the work.
* @param[in] delay_time Delay time. If it is greater than 0, the work will be delayed.
*
* @return Submit work result.
* @retval OS_SUCCESS Submit work success.
* @retval OS_BUSY The work has been submitted.
***********************************************************************************************************************
*/
os_err_t os_sys_work_run(os_work_id work_id, os_tick_t delay_time)
{
OS_ASSERT(work_id);
OS_ASSERT(delay_time < (OS_TICK_MAX / 2));
return os_work_run(gs_sys_workq_id, work_id, delay_time);
}
#endif /* OS_USING_SYSTEM_WORKQUEUE */
#endif /* OS_USING_WORKQUEUE */