RT-Thread-RT-Thread 隐藏的宝藏之管道RT-Thread问答社区

RT-Thread 隐藏的宝藏之管道

发布于 2021-03-27 11:40:10 浏览：1668 订阅该版

[tocm]

### 一. 什么是管道

`pipe`： 匿名管道。

对于熟悉 `linux` 开发的人来说，`pipe` 就很熟悉了。`pipe` 是一种 IPC 机制，他的作用是用作有血缘进程间完成数据传递，只能从一端写入，从另外一端读出。为了解决 `pipe` 的弊端，`linux` 的大神门又引入了 `mkfifo(实名管道)`。这些的讲解网络上有更清晰的讲解，就不再赘述。

`RT-Thread` 也实现了一套 `pipe`，不仅有自己的接口 `rt_pipe_xxx` , 也对接了 `posix` 接口。

### 二. 怎么 使用管道

在使用之前先看一下 `pipe` 的结构体

```
struct rt_pipe_device
{
    struct rt_device parent;
    rt_bool_t is_named;

/* ring buffer in pipe device */
    struct rt_ringbuffer *fifo;
    rt_uint16_t bufsz;

rt_uint8_t readers;
    rt_uint8_t writers;

rt_wqueue_t reader_queue;
    rt_wqueue_t writer_queue;

struct rt_mutex lock;
};
typedef struct rt_pipe_device rt_pipe_t;
```

通过上面的结构体可以先尝试分析以下管道的实现原理：

is_named ：确定是匿名管道，还是实名管道

fifo ：通过 ringbuff 来缓存数据

readers/writes : 确定读取/写入用户的数量

reader_queue/writer_queue ：通过工作队列来实现异步操作

lock ：使用互斥锁来实现半双工

用 `RT-Thread` 的管道的时候，有两个使用方法：

1. 使用 `RT-Thread` 的 `API`  , 即 `rt_pipe_xxx` 的API

2. 使用 `posix` 的管道 `API`

##### 使用 `RT-Thread` 管道 `API`

1. 创建管道

`rt_pipe_t *rt_pipe_create(const char *name, int bufsz)`

name : 创建管道的名字，设备会注册到设备管理器

bufsz：`ringbuff` 缓存去的大小

返回管道的对象

2. 删除管道

`int rt_pipe_delete(const char *name)`

name : 管道的名字，删除函数会自动在设备管理器查找到该设备

删除成功返回 0

3. 打开管道

`rt_err_t  rt_pipe_open (rt_device_t device, rt_uint16_t oflag)`

device : 设备对象

oflag : 没有用到

4. 关闭管道

`rt_err_t  rt_pipe_close  (rt_device_t device)`

device : 设备对象

5. 读取管道数据

`rt_size_t rt_pipe_read   (rt_device_t device, rt_off_t pos, void *buffer, rt_size_t count)`

device : 设备对象

pos : 未使用

buffer : 读管道数据的存储区的指针

count : 读取数据的长度

6. 读取管道数据

`rt_size_t rt_pipe_write  (rt_device_t device, rt_off_t pos, const void *buffer, rt_size_t count)`

device : 设备对象

pos : 未使用

buffer : 写管道数据的存储区的指针

count : 写取数据的长度

##### 使用 `POSIX` 管道 `API`

1. 创建匿名管道

`int pipe(int fildes[2])`

fildes[0] : 读文件描述符

fildes[1] : 写文件描述符

成功返回 0

2. 创建实名管道

`int mkfifo(const char *path, mode_t mode)`

path : 文件名

mode : 未使用

成功返回 0

3. 读写数据

使用 `read/write` 接口

### 三. 原理分析管道

##### 1. 创建管道

```
rt_pipe_t *rt_pipe_create(const char *name, int bufsz)
{
    rt_pipe_t *pipe;
    rt_device_t dev;

pipe = (rt_pipe_t *)rt_malloc(sizeof(rt_pipe_t));// 申请内存
    if (pipe == RT_NULL) return RT_NULL;

rt_memset(pipe, 0, sizeof(rt_pipe_t));
    pipe->is_named = RT_TRUE; /* initialize as a named pipe */
    rt_mutex_init(&(pipe->lock), name, RT_IPC_FLAG_FIFO);//初始化互斥锁
    rt_wqueue_init(&(pipe->reader_queue));//初始化读工作队列
    rt_wqueue_init(&(pipe->writer_queue));//初始化写工作队列

RT_ASSERT(bufsz < 0xFFFF);
    pipe->bufsz = bufsz;//ringbuff 缓存区大小

dev = &(pipe->parent);
    dev->type = RT_Device_Class_Pipe;// 设置设备类型

dev->init        = RT_NULL;//对接 ops 接口
    dev->open        = rt_pipe_open;
    dev->read        = rt_pipe_read;
    dev->write       = rt_pipe_write;
    dev->close       = rt_pipe_close;
    dev->control     = rt_pipe_control;

dev->rx_indicate = RT_NULL;// 不设置回调函数
    dev->tx_complete = RT_NULL;

if (rt_device_register(&(pipe->parent), name, RT_DEVICE_FLAG_RDWR | RT_DEVICE_FLAG_REMOVABLE) != 0) // 注册设备
    {
        rt_free(pipe);
        return RT_NULL;
    }
#ifdef RT_USING_POSIX
    dev->fops = (void*)&pipe_fops;
#endif

return pipe;// 返回 pipe 
}
```

##### 2. 删除管道

```
int rt_pipe_delete(const char *name)
{
    int result = 0;
    rt_device_t device;

device = rt_device_find(name);//查找设备
    if (device)
    {
        if (device->type == RT_Device_Class_Pipe)//检查设备类型
        {
            rt_pipe_t *pipe;

if (device->ref_count != 0)// 如果设备中还有任务没处理完，则不能删除
            {
                return -RT_EBUSY;
            }

pipe = (rt_pipe_t *)device;

rt_mutex_detach(&(pipe->lock));// 脱离互斥锁
            rt_device_unregister(device);// 取消设备注册

/* close fifo ringbuffer */
            if (pipe->fifo) 
            {
                rt_ringbuffer_destroy(pipe->fifo);// 摧毁 ringbuff
                pipe->fifo = RT_NULL;
            }
            rt_free(pipe);// 释放内存
        }
        else
        {
            result = -ENODEV;
        }
    }
    else
    {
        result = -ENODEV;
    }

return result;
}
```

##### 3. 打开管道

```
rt_err_t  rt_pipe_open (rt_device_t device, rt_uint16_t oflag)
{
    rt_pipe_t *pipe = (rt_pipe_t *)device;
    rt_err_t ret = RT_EOK;

if (device == RT_NULL)
    {
        ret = -RT_EINVAL;
        goto __exit;
    }
    
    rt_mutex_take(&(pipe->lock), RT_WAITING_FOREVER);// 加锁

if (pipe->fifo == RT_NULL)
    {
        pipe->fifo = rt_ringbuffer_create(pipe->bufsz);//创建 ringbuff
        if (pipe->fifo == RT_NULL)
        {
            ret = -RT_ENOMEM;
        }
    }

rt_mutex_release(&(pipe->lock));// 释放锁

__exit:
    return ret;
}
```

##### 4. 关闭管道
```
rt_err_t  rt_pipe_close  (rt_device_t device)
{
    rt_pipe_t *pipe = (rt_pipe_t *)device;

if (device == RT_NULL) return -RT_EINVAL;
    rt_mutex_take(&(pipe->lock), RT_WAITING_FOREVER);// 加锁

if (device->ref_count == 1)
    {
        rt_ringbuffer_destroy(pipe->fifo);//摧毁 ringbuff
        pipe->fifo = RT_NULL;
    }

rt_mutex_release(&(pipe->lock));//释放锁

return RT_EOK;
}
```

##### 5. 管道读数据

```
rt_size_t rt_pipe_read(rt_device_t device, rt_off_t pos, void *buffer, rt_size_t count)
{
    uint8_t *pbuf;
    rt_size_t read_bytes = 0;
    rt_pipe_t *pipe = (rt_pipe_t *)device;

if (device == RT_NULL)
    {
        rt_set_errno(-EINVAL);
        return 0;
    }
    if (count == 0) return 0;

pbuf = (uint8_t*)buffer;
    rt_mutex_take(&(pipe->lock), RT_WAITING_FOREVER);// 加锁

while (read_bytes < count)
    {// 循环从 ringbuff 中读取数据
        int len = rt_ringbuffer_get(pipe->fifo, &pbuf[read_bytes], count - read_bytes);
        if (len <= 0) break;

read_bytes += len;
    }
    rt_mutex_release(&pipe->lock);//释放锁

return read_bytes;//返回读取到的数据的长度
}
```

##### 6. 管道写数据

```
rt_size_t rt_pipe_write(rt_device_t device, rt_off_t pos, const void *buffer, rt_size_t count)
{
    uint8_t *pbuf;
    rt_size_t write_bytes = 0;
    rt_pipe_t *pipe = (rt_pipe_t *)device;

if (device == RT_NULL)
    {
        rt_set_errno(-EINVAL);
        return 0;
    }
    if (count == 0) return 0;

pbuf = (uint8_t*)buffer;
    rt_mutex_take(&pipe->lock, -1);//加锁

while (write_bytes < count)
    {// 往 ringbuff 写数据
        int len = rt_ringbuffer_put(pipe->fifo, &pbuf[write_bytes], count - write_bytes);
        if (len <= 0) break;

write_bytes += len;
    }
    rt_mutex_release(&pipe->lock);//释放锁

return write_bytes;//返回写入数据的长度
}
```

##### 7. 匿名管道的实现

```
int pipe(int fildes[2])
{
    rt_pipe_t *pipe;
    char dname[8];// 这里应该写 RT_NAME_MAX
    char dev_name[32];
    static int pipeno = 0;
    //拼接字符串，作为管道的名字
    rt_snprintf(dname, sizeof(dname), "pipe%d", pipeno++);

pipe = rt_pipe_create(dname, PIPE_BUFSZ);// 创建管道
    if (pipe == RT_NULL)
    {
        return -1;
    }
    // 设置为匿名管道
    pipe->is_named = RT_FALSE; /* unamed pipe */
    //拼接字符串，作为管道的名字
    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)
    {
        close(fildes[0]);
        return -1;
    }

return 0;
}
```

##### 8. 实名管道的实现

```
int mkfifo(const char *path, mode_t mode)
{
    rt_pipe_t *pipe;
    
    pipe = rt_pipe_create(path, PIPE_BUFSZ);// 创建管道
    if (pipe == RT_NULL)
    {
        return -1;
    }

return 0;
}
```

注意 `mode` 未使用。

##### 9. POSIX 接口对接

`posix` 接口的实现是通过 `fd` 文件描述符来找到 `pipe` 的对象，这样就和前面一样的操作了。唯一的区别就是在 `posix` 接口实现里面对接了 `readers/writers` ，用来记录管道的使用者。

### 四. 总结

`RT-Thread` 的管道有什么特点呢？

1. `RTOS` 中没有进程的概念，所以在 `RTOS` 中的管道用于线程间通讯
   2. `RT-Thread` 的管道也是半双工，半双工通过互斥锁实现
   3. 支持匿名管道和实名管道
   4. 支持 `posix` 标准，通过源码分析，`posix` 标准实现的功能更加完善