diff options
Diffstat (limited to 'components/drivers/src/pipe.c')
| -rw-r--r-- | components/drivers/src/pipe.c | 512 |
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 |