2.1.0版本串口驱动中队列函数rt_err_t rt_data_queue_push的疑问

发布于 2017-08-12 14:34:03 浏览：5585 订阅该版

```rt_err_t rt_data_queue_push(struct rt_data_queue *queue,
                            const void *data_ptr,
                            rt_size_t data_size,
                            rt_int32_t timeout)
{
    rt_uint16_t mask;
    rt_ubase_t  level;
    rt_thread_t thread;
    rt_err_t    result;
    
    RT_ASSERT(queue != RT_NULL);

result = RT_EOK;
    thread = rt_thread_self();
    mask = queue->size - 1;

level = rt_hw_interrupt_disable();
    while (queue->put_index - queue->get_index == queue->size)
    {
        queue->waiting_lwm = RT_TRUE;

/* queue is full */
        if (timeout == 0)
        {
            result = -RT_ETIMEOUT;

goto __exit;
        }

/* current context checking */
        RT_DEBUG_NOT_IN_INTERRUPT;

/* reset thread error number */
        thread->error = RT_EOK;
        
        /* suspend thread on the push list */
        rt_thread_suspend(thread);
        rt_list_insert_before(&(queue->suspended_push_list), &(thread->tlist));
        /* start timer */
        if (timeout > 0)
        {
            /* reset the timeout of thread timer and start it */
            rt_timer_control(&(thread->thread_timer),
                             RT_TIMER_CTRL_SET_TIME,
                             &timeout);
            rt_timer_start(&(thread->thread_timer));
        }

/* enable interrupt */
        rt_hw_interrupt_enable(level);

/* do schedule */
        rt_schedule();

/* thread is waked up */
        result = thread->error;
        level = rt_hw_interrupt_disable();
        if (result != RT_EOK) goto __exit;
    }

queue->queue[queue->put_index & mask].data_ptr  = data_ptr;
    queue->queue[queue->put_index & mask].data_size = data_size;
    queue->put_index += 1;

if (!rt_list_isempty(&(queue->suspended_pop_list)))
    {
        /* there is at least one thread in suspended list */

/* get thread entry */
        thread = rt_list_entry(queue->suspended_pop_list.next,
                               struct rt_thread,
                               tlist);

/* resume it */
        rt_thread_resume(thread);
        rt_hw_interrupt_enable(level);

/* perform a schedule */
        rt_schedule();

return result;
    }

__exit:
    rt_hw_interrupt_enable(level);
    if ((result == RT_EOK) && queue->evt_notify != RT_NULL)
    {
        queue->evt_notify(queue, RT_DATAQUEUE_EVENT_PUSH);
    }

return result;
}```
最近一直在看RTT的串口的驱动，虽说串口简单，但对于新人来说还是想仔细研究一下，大神略过。
这个函数是在串口DMA发送是调用的，根据名字可以看出是push到队列中（队列就是先进先出的fifo），既然有push那么肯定会有pop。pop函数在中断里，也就是dma发送的中断里。开始对这个函数不是很理解，多看几遍终于弄得差不多懂了。
1、push函数就是往队列里写数据，如果队列满则挂起当前任务，并将任务放到push的suspend队列中，函数中queue->waiting_lwm就是标记push队列满有挂起的任务（在pop函数中会用到这个变量，根据条件会resume在 push中的挂起的任务）。
2、对于put_index和get_index为什么没有限制范围呢？这里用到了mask，每次取值或者读取数值都有
queue->put_index & mask
这里将范围限制在离mask，感觉很是巧妙，学习了。

11 个回答

qq_哎 2017-08-12

这家伙很懒，什么也没写！

```
rt_err_t rt_data_queue_pop(struct rt_data_queue *queue,
                           const void** data_ptr,
                           rt_size_t *size, 
                           rt_int32_t timeout)
{
    rt_ubase_t  level;
    rt_thread_t thread;
    rt_err_t    result;
    rt_uint16_t mask;

RT_ASSERT(queue != RT_NULL);
    RT_ASSERT(data_ptr != RT_NULL);
    RT_ASSERT(size != RT_NULL);

result = RT_EOK;
    thread = rt_thread_self();
    mask   = queue->size - 1;

level = rt_hw_interrupt_disable();
    while (queue->get_index == queue->put_index)
    {
        /* queue is empty */
        if (timeout == 0)
        {
            result = -RT_ETIMEOUT;
            goto __exit;
        }

/* current context checking */
        RT_DEBUG_NOT_IN_INTERRUPT;

/* reset thread error number */
        thread->error = RT_EOK;
        
        /* suspend thread on the pop list */
        rt_thread_suspend(thread);
        rt_list_insert_before(&(queue->suspended_pop_list), &(thread->tlist));
        /* start timer */
        if (timeout > 0)
        {
            /* reset the timeout of thread timer and start it */
            rt_timer_control(&(thread->thread_timer),
                             RT_TIMER_CTRL_SET_TIME,
                             &timeout);
            rt_timer_start(&(thread->thread_timer));
        }

/* enable interrupt */
        rt_hw_interrupt_enable(level);

/* do schedule */
        rt_schedule();

/* thread is waked up */
        result = thread->error;
        level  = rt_hw_interrupt_disable();
        if (result != RT_EOK)
            goto __exit;
    }

*data_ptr = queue->queue[queue->get_index & mask].data_ptr;
    *size     = queue->queue[queue->get_index & mask].data_size;

queue->get_index += 1;

if ((queue->waiting_lwm == RT_TRUE) && 
        (queue->put_index - queue->get_index) <= queue->lwm)
    {
        queue->waiting_lwm = RT_FALSE;

/*
         * there is at least one thread in suspended list
         * and less than low water mark
         */
        if (!rt_list_isempty(&(queue->suspended_push_list)))
        {
            /* get thread entry */
            thread = rt_list_entry(queue->suspended_push_list.next,
                                   struct rt_thread,
                                   tlist);

/* resume it */
            rt_thread_resume(thread);
            rt_hw_interrupt_enable(level);

/* perform a schedule */
            rt_schedule();
        }

if (queue->evt_notify != RT_NULL)
            queue->evt_notify(queue, RT_DATAQUEUE_EVENT_LWM);

return result;
    }

__exit:
    rt_hw_interrupt_enable(level);
    if ((result == RT_EOK) && (queue->evt_notify != RT_NULL))
    {
        queue->evt_notify(queue, RT_DATAQUEUE_EVENT_POP);
    }

return result;
}
```

这个是pop函数，这个是判断是否resume 挂起的push任务的条件

```
if ((queue->waiting_lwm == RT_TRUE) && 
        (queue->put_index - queue->get_index) <= queue->lwm)
```

除了waiting_lwm ，另一个是lwm，lwm的值是在初始化设置的，我看初始化size=8，lwm=4，这个值是要设置为size的一半吗？

qq_哎 2017-08-12

这家伙很懒，什么也没写！

```
rt_err_t rt_data_queue_peak(struct rt_data_queue *queue,
                            const void** data_ptr,
                            rt_size_t *size)
{
    rt_ubase_t  level;
    rt_uint16_t mask;

RT_ASSERT(queue != RT_NULL);

mask = queue->size - 1;

level = rt_hw_interrupt_disable();

if (queue->get_index == queue->put_index) 
    {
        rt_hw_interrupt_enable(level);
        
        return -RT_EEMPTY;
    }

*data_ptr = queue->queue[queue->get_index & mask].data_ptr;
    *size     = queue->queue[queue->get_index & mask].data_size;

rt_hw_interrupt_enable(level);

return RT_EOK;
}
```

这个是peak函数，跟pop函数很相似，区别是没有resume任务和读取完了以后get_index没有加1.
下面是调用函数的一部分

```
 rt_data_queue_pop(&(tx_dma->data_queue), &last_data_ptr, &data_size, 0);
            if (rt_data_queue_peak(&(tx_dma->data_queue), &data_ptr, &data_size) == RT_EOK)
            {
                /* transmit next data node */
                tx_dma->activated = RT_TRUE;
                serial->ops->dma_transmit(serial, data_ptr, data_size, RT_SERIAL_DMA_TX);
            }
            else
            {
                tx_dma->activated = RT_FALSE;
            }
            
            /* invoke callback */
            if (serial->parent.tx_complete != RT_NULL)
            {
                serial->parent.tx_complete(&serial->parent, (void*)last_data_ptr);
            }
```

1、这里为什么读了两次queue中的数据？
2、第一次读取用的pop，数据放在last_data_ptr中，这是最后的数据？这个在coplete中发送，为什么这么操作？没看太懂哦
3、第二次读取用的peak函数，数据放在data_ptr然后dma发送。

bernard 2017-08-12

这家伙很懒，什么也没写！

armink 2017-08-12

这家伙很懒，什么也没写！

qq_哎 2017-08-12

这家伙很懒，什么也没写！

armink 2017-08-12

这家伙很懒，什么也没写！

qq_哎 2017-08-12

这家伙很懒，什么也没写！

qq_哎 2017-08-12

这家伙很懒，什么也没写！

armink 2017-08-13

这家伙很懒，什么也没写！

>>疑惑的话，可以自己做一个 finsh 命令不停地 push/pop 试一试就知道了。
>
>---
>
>
>想了一下用finsh做这个的话有点麻烦，，，
>总是感觉每次发数据调用一下两个函数后，会把第一个数据丢了
>rt_data_queue_pop
>rt_data_queue_peak
>因为pop读完queue的数据（读取到last_data中）后会使index加1，而后peak在读取到的是index加1的数据发送。如果通过peak判断不是最后一个数据，那么读取到last_data中的数据就丢了啊？

---

写个命令一点都不麻烦吧，自己手动测试下，才能更直接的验证自己的想法。

帮你找了一个，我之前写的测试命令。

```
static void dq(uint8_t argc, char **argv) {
    static struct rt_data_queue data_queue;
    
    if (argc >= 2) {
        if (!strcmp(argv[1], "init")) {
            rt_data_queue_init(&data_queue, 10, 1, NULL);
        } else if (!strcmp(argv[1], "push")) {
            static int i = 10;
            char *data = rt_malloc(++i);
            rt_data_queue_push(&data_queue, data, i, 20000);
            rt_kprintf("pop data size %d
", i);
        } else if (!strcmp(argv[1], "pop")) {
            char *data = NULL;
            rt_size_t size = 0;
            if(rt_data_queue_pop(&data_queue, (void const **)&data, &size, 20000) == RT_EOK) {
                rt_kprintf("pop data size %d
", size);
                rt_free(data);
            }
        }
    }
}
MSH_CMD_EXPORT(dq, dataqueue test 'dq <init|push|pop>');
```

qq_哎 2017-08-13

这家伙很懒，什么也没写！

>>>疑惑的话，可以自己做一个 finsh 命令不停地 push/pop 试一试就知道了。
>>
>>---
>>
>>
>>想了一下用finsh做这个的话有点麻烦，，，
>>总是感觉每次发数据调用一下两个函数后，会把第一个数据丢了
>>rt_data_queue_pop
>>rt_data_queue_peak
>>因为pop读完queue的数据（读取到last_data中）后会使index加1，而后peak在读取到的是index加1的数据发送。如果通过peak判断不是最后一个数据，那么读取到last_data中的数据就丢了啊？
>
>---
>
>
>
>写个命令一点都不麻烦吧，自己手动测试下，才能更直接的验证自己的想法。
>
>帮你找了一个，我之前写的测试命令。
>
>
>
>```
>static void dq(uint8_t argc, char **argv) {
>    static struct rt_data_queue data_queue;
>    
>    if (argc >= 2) {
>        if (!strcmp(argv[1], "init")) {
>            rt_data_queue_init(&data_queue, 10, 1, NULL);
>        } else if (!strcmp(argv[1], "push")) {
>            static int i = 10;
>            char *data = rt_malloc(++i);
>            rt_data_queue_push(&data_queue, data, i, 20000);
>            rt_kprintf("pop data size %d
>", i);
>        } else if (!strcmp(argv[1], "pop")) {
>            char *data = NULL;
>            rt_size_t size = 0;
>            if(rt_data_queue_pop(&data_queue, (void const **)&data, &size, 20000) == RT_EOK) {
>                rt_kprintf("pop data size %d
>", size);
>                rt_free(data);
>            }
>        }
>    }
>}
>MSH_CMD_EXPORT(dq, dataqueue test 'dq <init|push|pop>');
>```

---

好吧，还是自己太懒了，，，，谢谢啊，多谢！动手试试哈。。。