[DeviceDrivers] change cmd type.

1. Change 'rt_uint8_t' type of cmd to 'int';
2. Add waitqueue;
3. Split device ipc header files;

1 files changed, 323 insertions, 189 deletions

diff --git a/components/drivers/src/pipe.c b/components/drivers/src/pipe.c
index f438a92898..b455870c8a 100644
--- a/components/drivers/src/pipe.c
+++ b/components/drivers/src/pipe.c
@@ -1,7 +1,7 @@
 /*
  * File      : pipe.c
  * This file is part of RT-Thread RTOS
- * COPYRIGHT (C) 2012, RT-Thread Development Team
+ * COPYRIGHT (C) 2012-2017, RT-Thread Development Team
  *
  *  This program is free software; you can redistribute it and/or modify
  *  it under the terms of the GNU General Public License as published by
@@ -21,273 +21,407 @@
  * Date           Author       Notes
  * 2012-09-30     Bernard      first version.
  */
-
 #include <rthw.h>
-#include <rtthread.h>
 #include <rtdevice.h>
+#include <dfs_file.h>
+#include <dfs_posix.h>
 
-static void _rt_pipe_resume_writer(struct rt_pipe_device *pipe)
+#if defined(RT_USING_DFS) && defined(RT_USING_DFS_DEVFS)
+static int pipe_open(struct dfs_fd *fd)
 {
-    if (!rt_list_isempty(&pipe->suspended_write_list))
-    {
-        rt_thread_t thread;
+    rt_device_t device;
+    rt_pipe_t *pipe;
 
-        RT_ASSERT(pipe->flag & RT_PIPE_FLAG_BLOCK_WR);
+    pipe = (rt_pipe_t *)fd->data;
+    if (!pipe) return -1;
 
-        /* get suspended thread */
-        thread = rt_list_entry(pipe->suspended_write_list.next,
-                struct rt_thread,
-                tlist);
+    device = &(pipe->parent);
+    rt_mutex_take(&(pipe->lock), RT_WAITING_FOREVER);
 
-        /* resume the write thread */
-        rt_thread_resume(thread);
+    if (device->ref_count == 0)
+    {
+        pipe->fifo = rt_ringbuffer_create(PIPE_BUFSZ);
+    }
 
-        rt_schedule();
+    switch (fd->flags & O_ACCMODE)
+    {
+    case O_RDONLY:
+        pipe->readers ++;
+        break;
+    case O_WRONLY:
+        pipe->writers ++;
+        break;
+    case O_RDWR:
+        pipe->readers ++;
+        pipe->writers ++;
+        break;
     }
+    device->ref_count ++;
+
+    rt_mutex_release(&(pipe->lock));
+
+    return 0;
 }
 
-static rt_size_t rt_pipe_read(rt_device_t dev,
-                              rt_off_t    pos,
-                              void       *buffer,
-                              rt_size_t   size)
+static int pipe_close(struct dfs_fd *fd)
 {
-    rt_uint32_t level;
-    rt_thread_t thread;
-    struct rt_pipe_device *pipe;
-    rt_size_t read_nbytes;
+    rt_device_t device;
+    rt_pipe_t *pipe;
 
-    pipe = PIPE_DEVICE(dev);
-    RT_ASSERT(pipe != RT_NULL);
+    pipe = (rt_pipe_t *)fd->data;
+    if (!pipe) return -1;
 
-    if (!(pipe->flag & RT_PIPE_FLAG_BLOCK_RD))
+    device = &(pipe->parent);
+    rt_mutex_take(&(pipe->lock), RT_WAITING_FOREVER);
+
+    switch (fd->flags & O_ACCMODE)
     {
-        level = rt_hw_interrupt_disable();
-        read_nbytes = rt_ringbuffer_get(&(pipe->ringbuffer), buffer, size);
+    case O_RDONLY:
+        pipe->readers --;
+        break;
+    case O_WRONLY:
+        pipe->writers --;
+        break;
+    case O_RDWR:
+        pipe->readers --;
+        pipe->writers --;
+        break;
+    }
 
-        /* if the ringbuffer is empty, there won't be any writer waiting */
-        if (read_nbytes)
-            _rt_pipe_resume_writer(pipe);
+    if (pipe->writers == 0)
+    {
+        rt_wqueue_wakeup(&(pipe->reader_queue), (void*)(POLLIN | POLLERR | POLLHUP));
+    }
 
-        rt_hw_interrupt_enable(level);
+    if (pipe->readers == 0)
+    {
+        rt_wqueue_wakeup(&(pipe->writer_queue), (void*)(POLLOUT | POLLERR | POLLHUP));
+    }
 
-        return read_nbytes;
+    if (device->ref_count == 1)
+    {
+        rt_free(pipe->fifo);
+        pipe->fifo = RT_NULL;
     }
+    device->ref_count --;
 
-    thread = rt_thread_self();
+    rt_mutex_release(&(pipe->lock));
 
-    /* current context checking */
-    RT_DEBUG_NOT_IN_INTERRUPT;
+    return 0;
+}
 
-    do {
-        level = rt_hw_interrupt_disable();
-        read_nbytes = rt_ringbuffer_get(&(pipe->ringbuffer), buffer, size);
-        if (read_nbytes == 0)
-        {
-            rt_thread_suspend(thread);
-            /* waiting on suspended read list */
-            rt_list_insert_before(&(pipe->suspended_read_list),
-                                  &(thread->tlist));
-            rt_hw_interrupt_enable(level);
+static int pipe_ioctl(struct dfs_fd *fd, int cmd, void *args)
+{
+    rt_pipe_t *pipe;
+    int ret = 0;
 
-            rt_schedule();
-        }
-        else
-        {
-            _rt_pipe_resume_writer(pipe);
-            rt_hw_interrupt_enable(level);
-            break;
-        }
-    } while (read_nbytes == 0);
+    pipe = (rt_pipe_t *)fd->data;
+
+    switch (cmd)
+    {
+    case FIONREAD:
+        *((int*)args) = rt_ringbuffer_data_len(pipe->fifo);
+        break;
+    case FIONWRITE:
+        *((int*)args) = rt_ringbuffer_space_len(pipe->fifo);
+        break;
+    default:
+        ret = -EINVAL;
+        break;
+    }
 
-    return read_nbytes;
+    return ret;
 }
 
-static void _rt_pipe_resume_reader(struct rt_pipe_device *pipe)
+static int pipe_read(struct dfs_fd *fd, void *buf, size_t count)
 {
-    if (pipe->parent.rx_indicate)
-        pipe->parent.rx_indicate(&pipe->parent,
-                                 rt_ringbuffer_data_len(&pipe->ringbuffer));
+    int len = 0;
+    rt_pipe_t *pipe;
 
-    if (!rt_list_isempty(&pipe->suspended_read_list))
-    {
-        rt_thread_t thread;
+    pipe = (rt_pipe_t *)fd->data;
 
-        RT_ASSERT(pipe->flag & RT_PIPE_FLAG_BLOCK_RD);
+    /* no process has the pipe open for writing, return end-of-file */
+    if (pipe->writers == 0)
+        return 0;
 
-        /* get suspended thread */
-        thread = rt_list_entry(pipe->suspended_read_list.next,
-                struct rt_thread,
-                tlist);
+    rt_mutex_take(&(pipe->lock), RT_WAITING_FOREVER);
 
-        /* resume the read thread */
-        rt_thread_resume(thread);
+    while (1)
+    {
+        if (pipe->writers == 0)
+        {
+            goto out;
+        }
+
+        len = rt_ringbuffer_get(pipe->fifo, buf, count);
+
+        if (len > 0)
+        {
+            break;
+        }
+        else
+        {
+            if (fd->flags & O_NONBLOCK)
+            {
+                len = -EAGAIN;
+                goto out;
+            }
 
-        rt_schedule();
+            rt_mutex_release(&pipe->lock);
+            rt_wqueue_wakeup(&(pipe->writer_queue), (void*)POLLOUT);
+            rt_wqueue_wait(&(pipe->reader_queue), 0, -1);
+            rt_mutex_take(&(pipe->lock), RT_WAITING_FOREVER);
+        }
     }
+
+    /* wakeup writer */
+    rt_wqueue_wakeup(&(pipe->writer_queue), (void*)POLLOUT);
+
+out:
+    rt_mutex_release(&pipe->lock);
+
+    return len;
 }
 
-static rt_size_t rt_pipe_write(rt_device_t dev,
-                               rt_off_t    pos,
-                               const void *buffer,
-                               rt_size_t   size)
+static int pipe_write(struct dfs_fd *fd, const void *buf, size_t count)
 {
-    rt_uint32_t level;
-    rt_thread_t thread;
-    struct rt_pipe_device *pipe;
-    rt_size_t write_nbytes;
+    int len;
+    rt_pipe_t *pipe;
+    int wakeup = 0;
+    int ret = 0;
+    uint8_t *pbuf;
 
-    pipe = PIPE_DEVICE(dev);
-    RT_ASSERT(pipe != RT_NULL);
+    pipe = (rt_pipe_t *)fd->data;
 
-    if ((pipe->flag & RT_PIPE_FLAG_FORCE_WR) ||
-       !(pipe->flag & RT_PIPE_FLAG_BLOCK_WR))
+    if (pipe->readers == 0)
     {
-        level = rt_hw_interrupt_disable();
+        ret = -EPIPE;
+        goto out;
+    }
+
+    if (count == 0)
+        return 0;
 
-        if (pipe->flag & RT_PIPE_FLAG_FORCE_WR)
-            write_nbytes = rt_ringbuffer_put_force(&(pipe->ringbuffer),
-                                                   buffer, size);
+    pbuf = (uint8_t*)buf;
+    rt_mutex_take(&pipe->lock, -1);
+
+    while (1)
+    {
+        if (pipe->readers == 0)
+        {
+            if (ret == 0)
+                ret = -EPIPE;
+            break;
+        }
+
+        len = rt_ringbuffer_put(pipe->fifo, pbuf, count - ret);
+        ret +=  len;
+        pbuf += len;
+        wakeup = 1;
+
+        if (ret == count)
+        {
+            break;
+        }
         else
-            write_nbytes = rt_ringbuffer_put(&(pipe->ringbuffer),
-                                             buffer, size);
+        {
+            if (fd->flags & O_NONBLOCK)
+            {
+                if (ret == 0)
+                {
+                    ret = -EAGAIN;
+                }
 
-        _rt_pipe_resume_reader(pipe);
+                break;
+            }
+        }
 
-        rt_hw_interrupt_enable(level);
+        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);
+        rt_mutex_take(&pipe->lock, -1);
+    }
 
-        return write_nbytes;
+    rt_mutex_release(&pipe->lock);
+    if (wakeup)
+    {
+        rt_wqueue_wakeup(&(pipe->reader_queue), (void*)POLLIN);
     }
 
-    thread = rt_thread_self();
+out:
+
+    return ret;
+}
 
-    /* current context checking */
-    RT_DEBUG_NOT_IN_INTERRUPT;
+static int pipe_poll(struct dfs_fd *fd, rt_pollreq_t *req)
+{
+    int mask = 0;
+    rt_pipe_t *pipe;
+    int mode = 0;
+    pipe = (rt_pipe_t *)fd->data;
+
+    rt_poll_add(&(pipe->reader_queue), req);
+    rt_poll_add(&(pipe->writer_queue), req);
+
+    switch (fd->flags & O_ACCMODE)
+    {
+    case O_RDONLY:
+        mode = 1;
+        break;
+    case O_WRONLY:
+        mode = 2;
+        break;
+    case O_RDWR:
+        mode = 3;
+        break;
+    }
 
-    do {
-        level = rt_hw_interrupt_disable();
-        write_nbytes = rt_ringbuffer_put(&(pipe->ringbuffer), buffer, size);
-        if (write_nbytes == 0)
+    if (mode & 1)
+    {
+        if (rt_ringbuffer_data_len(pipe->fifo) != 0)
+        {
+            mask |= POLLIN;
+        }
+        if (pipe->writers == 0)
         {
-            /* pipe full, waiting on suspended write list */
-            rt_thread_suspend(thread);
-            /* waiting on suspended read list */
-            rt_list_insert_before(&(pipe->suspended_write_list),
-                                  &(thread->tlist));
-            rt_hw_interrupt_enable(level);
+            mask |= POLLHUP;
+        }
+    }
 
-            rt_schedule();
+    if (mode & 2)
+    {
+        if (rt_ringbuffer_space_len(pipe->fifo) != 0)
+        {
+            mask |= POLLOUT;
         }
-        else
+        if (pipe->readers == 0)
         {
-            _rt_pipe_resume_reader(pipe);
-            rt_hw_interrupt_enable(level);
-            break;
+            mask |= POLLERR;
         }
-    } while (write_nbytes == 0);
+    }
 
-    return write_nbytes;
+    return mask;
 }
 
-static rt_err_t rt_pipe_control(rt_device_t dev, rt_uint8_t cmd, void *args)
+static const struct dfs_file_ops pipe_fops =
 {
-    if (cmd == PIPE_CTRL_GET_SPACE && args)
-        *(rt_size_t*)args = rt_ringbuffer_space_len(&PIPE_DEVICE(dev)->ringbuffer);
-    return RT_EOK;
-}
+    pipe_open,
+    pipe_close,
+    pipe_ioctl,
+    pipe_read,
+    pipe_write,
+    RT_NULL,
+    RT_NULL,
+    RT_NULL,
+    pipe_poll,
+};
 
-/**
- * This function will initialize a pipe device and put it under control of
- * resource management.
- *
- * @param pipe the pipe device
- * @param name the name of pipe device
- * @param flag the attribute of the pipe device
- * @param buf  the buffer of pipe device
- * @param size the size of pipe device buffer
- *
- * @return the operation status, RT_EOK on successful
- */
-rt_err_t rt_pipe_init(struct rt_pipe_device *pipe,
-                      const char *name,
-                      enum rt_pipe_flag flag,
-                      rt_uint8_t *buf,
-                      rt_size_t size)
+rt_pipe_t *rt_pipe_create(const char *name)
 {
-    RT_ASSERT(pipe);
-    RT_ASSERT(buf);
+    rt_pipe_t *pipe;
+    rt_device_t dev;
 
-    /* initialize suspended list */
-    rt_list_init(&pipe->suspended_read_list);
-    rt_list_init(&pipe->suspended_write_list);
+    pipe = rt_malloc(sizeof(rt_pipe_t));
+    if (pipe == RT_NULL) return RT_NULL;
 
-    /* initialize ring buffer */
-    rt_ringbuffer_init(&pipe->ringbuffer, buf, size);
+    rt_memset(pipe, 0, sizeof(rt_pipe_t));
+    rt_mutex_init(&(pipe->lock), name, RT_IPC_FLAG_FIFO);
+    rt_list_init(&(pipe->reader_queue));
+    rt_list_init(&(pipe->writer_queue));
 
-    pipe->flag = flag;
+    dev = &(pipe->parent);
+    dev->type = RT_Device_Class_Pipe;
 
-    /* create pipe */
-    pipe->parent.type    = RT_Device_Class_Pipe;
-    pipe->parent.init    = RT_NULL;
-    pipe->parent.open    = RT_NULL;
-    pipe->parent.close   = RT_NULL;
-    pipe->parent.read    = rt_pipe_read;
-    pipe->parent.write   = rt_pipe_write;
-    pipe->parent.control = rt_pipe_control;
+    if (rt_device_register(&(pipe->parent), name, RT_DEVICE_FLAG_RDWR | RT_DEVICE_FLAG_REMOVABLE) != 0)
+    {
+        rt_free(pipe);
+        return RT_NULL;
+    }
+    dev->fops = (void*)&pipe_fops;
 
-    return rt_device_register(&(pipe->parent), name, RT_DEVICE_FLAG_RDWR);
+    return pipe;
 }
-RTM_EXPORT(rt_pipe_init);
 
-/**
- * This function will detach a pipe device from resource management
- *
- * @param pipe the pipe device
- *
- * @return the operation status, RT_EOK on successful
- */
-rt_err_t rt_pipe_detach(struct rt_pipe_device *pipe)
+int rt_pipe_delete(const char *name)
 {
-    return rt_device_unregister(&pipe->parent);
+    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);
+
+            rt_free(pipe);
+        }
+        else
+        {
+            result = -1;
+        }
+    }
+    else
+    {
+        result = -1;
+    }
+
+    return result;
 }
-RTM_EXPORT(rt_pipe_detach);
 
-#ifdef RT_USING_HEAP
-rt_err_t rt_pipe_create(const char *name, enum rt_pipe_flag flag, rt_size_t size)
+int pipe(int fildes[2])
 {
-    rt_uint8_t *rb_memptr = RT_NULL;
-    struct rt_pipe_device *pipe = RT_NULL;
+    rt_pipe_t *pipe;
+    char dname[8];
+    char dev_name[32];
+    static int pipeno = 0;
 
-    /* get aligned size */
-    size = RT_ALIGN(size, RT_ALIGN_SIZE);
-    pipe = (struct rt_pipe_device *)rt_calloc(1, sizeof(struct rt_pipe_device));
+    rt_snprintf(dname, sizeof(dname), "pipe%d", pipeno++);
+
+    pipe = rt_pipe_create(dname);
     if (pipe == RT_NULL)
-        return -RT_ENOMEM;
+    {
+        return -1;
+    }
 
-    /* create ring buffer of pipe */
-    rb_memptr = rt_malloc(size);
-    if (rb_memptr == RT_NULL)
+    rt_snprintf(dev_name, sizeof(dev_name), "/dev/%s", dname);
+    fildes[0] = open(dev_name, O_RDONLY, 0);
+    if (fildes[0] < 0)
     {
-        rt_free(pipe);
-        return -RT_ENOMEM;
+        return -1;
     }
 
-    return rt_pipe_init(pipe, name, flag, rb_memptr, size);
+    fildes[1] = open(dev_name, O_WRONLY, 0);
+    if (fildes[1] < 0)
+    {
+        close(fildes[1]);
+        return -1;
+    }
+
+    return 0;
 }
-RTM_EXPORT(rt_pipe_create);
 
-void rt_pipe_destroy(struct rt_pipe_device *pipe)
+int mkfifo(const char *path, mode_t mode)
 {
+    rt_pipe_t *pipe;
+    
+    pipe = rt_pipe_create(path);
     if (pipe == RT_NULL)
-        return;
-
-    /* un-register pipe device */
-    rt_pipe_detach(pipe);
-
-    /* release memory */
-    rt_free(pipe->ringbuffer.buffer_ptr);
-    rt_free(pipe);
+    {
+        return -1;
+    }
 
-    return;
+    return 0;
 }
-RTM_EXPORT(rt_pipe_destroy);
-#endif /* RT_USING_HEAP */
+#endif