Merge pull request #143 from grissiom/ringbuffer-pipe

Ringbuffer and Pipe

2 files changed, 187 insertions, 48 deletions

diff --git a/components/drivers/src/pipe.c b/components/drivers/src/pipe.c
index 469c2d06e6..3f14e6a837 100644
--- a/components/drivers/src/pipe.c
+++ b/components/drivers/src/pipe.c
@@ -26,6 +26,26 @@
 #include <rtthread.h>
 #include <rtdevice.h>
 
+static void _rt_pipe_resume_writer(struct rt_pipe_device *pipe)
+{
+    if (!rt_list_isempty(&pipe->suspended_write_list))
+    {
+        rt_thread_t thread;
+
+        RT_ASSERT(pipe->flag & RT_PIPE_FLAG_BLOCK_WR);
+
+        /* get suspended thread */
+        thread = rt_list_entry(pipe->suspended_write_list.next,
+                struct rt_thread,
+                tlist);
+
+        /* resume the write thread */
+        rt_thread_resume(thread);
+
+        rt_schedule();
+    }
+}
+
 static rt_size_t rt_pipe_read(rt_device_t dev,
                               rt_off_t    pos,
                               void       *buffer,
@@ -39,13 +59,26 @@ static rt_size_t rt_pipe_read(rt_device_t dev,
     pipe = PIPE_DEVICE(dev);
     RT_ASSERT(pipe != RT_NULL);
 
+    if (!(pipe->flag & RT_PIPE_FLAG_BLOCK_RD))
+    {
+        level = rt_hw_interrupt_disable();
+        read_nbytes = rt_ringbuffer_get(&(pipe->ringbuffer), buffer, size);
+
+        /* if the ringbuffer is empty, there won't be any writer waiting */
+        if (read_nbytes)
+            _rt_pipe_resume_writer(pipe);
+
+        rt_hw_interrupt_enable(level);
+
+        return read_nbytes;
+    }
+
     thread = rt_thread_self();
 
     /* current context checking */
     RT_DEBUG_NOT_IN_INTERRUPT;
 
-    do 
-    {
+    do {
         level = rt_hw_interrupt_disable();
         read_nbytes = rt_ringbuffer_get(&(pipe->ringbuffer), buffer, size);
         if (read_nbytes == 0)
@@ -60,23 +93,8 @@ static rt_size_t rt_pipe_read(rt_device_t dev,
         }
         else
         {
-            if (!rt_list_isempty(&pipe->suspended_write_list))
-            {
-                /* get suspended thread */
-                thread = rt_list_entry(pipe->suspended_write_list.next,
-                                       struct rt_thread,
-                                       tlist);
-
-                /* resume the write thread */
-                rt_thread_resume(thread);
-                rt_hw_interrupt_enable(level);
-
-                rt_schedule();
-            }
-            else
-            {
-                rt_hw_interrupt_enable(level);
-            }
+            _rt_pipe_resume_writer(pipe);
+            rt_hw_interrupt_enable(level);
             break;
         }
     } while (read_nbytes == 0);
@@ -84,7 +102,30 @@ static rt_size_t rt_pipe_read(rt_device_t dev,
     return read_nbytes;
 }
 
-struct rt_pipe_device *_pipe = RT_NULL;
+static void _rt_pipe_resume_reader(struct rt_pipe_device *pipe)
+{
+    if (pipe->parent.rx_indicate)
+        pipe->parent.rx_indicate(&pipe->parent,
+                                 rt_ringbuffer_data_len(&pipe->ringbuffer));
+
+    if (!rt_list_isempty(&pipe->suspended_read_list))
+    {
+        rt_thread_t thread;
+
+        RT_ASSERT(pipe->flag & RT_PIPE_FLAG_BLOCK_RD);
+
+        /* get suspended thread */
+        thread = rt_list_entry(pipe->suspended_read_list.next,
+                struct rt_thread,
+                tlist);
+
+        /* resume the read thread */
+        rt_thread_resume(thread);
+
+        rt_schedule();
+    }
+}
+
 static rt_size_t rt_pipe_write(rt_device_t dev,
                                rt_off_t    pos,
                                const void *buffer,
@@ -97,16 +138,32 @@ static rt_size_t rt_pipe_write(rt_device_t dev,
 
     pipe = PIPE_DEVICE(dev);
     RT_ASSERT(pipe != RT_NULL);
-    if (_pipe == RT_NULL)
-        _pipe = pipe;
+
+    if ((pipe->flag & RT_PIPE_FLAG_FORCE_WR) ||
+       !(pipe->flag & RT_PIPE_FLAG_BLOCK_WR))
+    {
+        level = rt_hw_interrupt_disable();
+
+        if (pipe->flag & RT_PIPE_FLAG_FORCE_WR)
+            write_nbytes = rt_ringbuffer_put_force(&(pipe->ringbuffer),
+                                                   buffer, size);
+        else
+            write_nbytes = rt_ringbuffer_put(&(pipe->ringbuffer),
+                                             buffer, size);
+
+        _rt_pipe_resume_reader(pipe);
+
+        rt_hw_interrupt_enable(level);
+
+        return write_nbytes;
+    }
 
     thread = rt_thread_self();
 
     /* current context checking */
     RT_DEBUG_NOT_IN_INTERRUPT;
 
-    do
-    {
+    do {
         level = rt_hw_interrupt_disable();
         write_nbytes = rt_ringbuffer_put(&(pipe->ringbuffer), buffer, size);
         if (write_nbytes == 0)
@@ -122,26 +179,11 @@ static rt_size_t rt_pipe_write(rt_device_t dev,
         }
         else
         {
-            if (!rt_list_isempty(&pipe->suspended_read_list))
-            {
-                /* get suspended thread */
-                thread = rt_list_entry(pipe->suspended_read_list.next,
-                                       struct rt_thread,
-                                       tlist);
-
-                /* resume the read thread */
-                rt_thread_resume(thread);
-                rt_hw_interrupt_enable(level);
-
-                rt_schedule();
-            }
-            else
-            {
-                rt_hw_interrupt_enable(level);
-            }
+            _rt_pipe_resume_reader(pipe);
+            rt_hw_interrupt_enable(level);
             break;
         }
-    }while (write_nbytes == 0);
+    } while (write_nbytes == 0);
 
     return write_nbytes;
 }
@@ -157,6 +199,7 @@ static rt_err_t rt_pipe_control(rt_device_t dev, rt_uint8_t cmd, void *args)
  *
  * @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
  *
@@ -164,6 +207,7 @@ static rt_err_t rt_pipe_control(rt_device_t dev, rt_uint8_t cmd, void *args)
  */
 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)
 {
@@ -177,6 +221,8 @@ rt_err_t rt_pipe_init(struct rt_pipe_device *pipe,
     /* initialize ring buffer */
     rt_ringbuffer_init(&pipe->ringbuffer, buf, size);
 
+    pipe->flag = flag;
+
     /* create pipe */
     pipe->parent.type    = RT_Device_Class_Char;
     pipe->parent.init    = RT_NULL;
@@ -204,7 +250,7 @@ rt_err_t rt_pipe_detach(struct rt_pipe_device *pipe)
 RTM_EXPORT(rt_pipe_detach);
 
 #ifdef RT_USING_HEAP
-rt_err_t rt_pipe_create(const char *name, rt_size_t size)
+rt_err_t rt_pipe_create(const char *name, enum rt_pipe_flag flag, rt_size_t size)
 {
     rt_uint8_t *rb_memptr = RT_NULL;
     struct rt_pipe_device *pipe = RT_NULL;
@@ -223,7 +269,7 @@ rt_err_t rt_pipe_create(const char *name, rt_size_t size)
         return -RT_ENOMEM;
     }
 
-    return rt_pipe_init(pipe, name, rb_memptr, size);
+    return rt_pipe_init(pipe, name, flag, rb_memptr, size);
 }
 RTM_EXPORT(rt_pipe_create);
 
diff --git a/components/drivers/src/ringbuffer.c b/components/drivers/src/ringbuffer.c
index 8e7e8c7895..6212b10258 100644
--- a/components/drivers/src/ringbuffer.c
+++ b/components/drivers/src/ringbuffer.c
@@ -44,6 +44,9 @@ void rt_ringbuffer_init(struct rt_ringbuffer *rb,
 }
 RTM_EXPORT(rt_ringbuffer_init);
 
+/**
+ * put a block of data into ring buffer
+ */
 rt_size_t rt_ringbuffer_put(struct rt_ringbuffer *rb,
                             const rt_uint8_t     *ptr,
                             rt_uint16_t           length)
@@ -53,7 +56,7 @@ rt_size_t rt_ringbuffer_put(struct rt_ringbuffer *rb,
     RT_ASSERT(rb != RT_NULL);
 
     /* whether has enough space */
-    size = RT_RINGBUFFER_EMPTY(rb);
+    size = rt_ringbuffer_space_len(rb);
 
     /* no space */
     if (size == 0)
@@ -89,6 +92,59 @@ rt_size_t rt_ringbuffer_put(struct rt_ringbuffer *rb,
 RTM_EXPORT(rt_ringbuffer_put);
 
 /**
+ * put a block of data into ring buffer
+ *
+ * When the buffer is full, it will discard the old data.
+ */
+rt_size_t rt_ringbuffer_put_force(struct rt_ringbuffer *rb,
+                            const rt_uint8_t     *ptr,
+                            rt_uint16_t           length)
+{
+    enum rt_ringbuffer_state old_state;
+
+    RT_ASSERT(rb != RT_NULL);
+
+    old_state = rt_ringbuffer_status(rb);
+
+    if (length > rb->buffer_size)
+        length = rb->buffer_size;
+
+    if (rb->buffer_size - rb->write_index > length)
+    {
+        /* read_index - write_index = empty space */
+        memcpy(&rb->buffer_ptr[rb->write_index], ptr, length);
+        /* this should not cause overflow because there is enough space for
+         * length of data in current mirror */
+        rb->write_index += length;
+
+        if (old_state == RT_RINGBUFFER_FULL)
+            rb->read_index = rb->write_index;
+
+        return length;
+    }
+
+    memcpy(&rb->buffer_ptr[rb->write_index],
+           &ptr[0],
+           rb->buffer_size - rb->write_index);
+    memcpy(&rb->buffer_ptr[0],
+           &ptr[rb->buffer_size - rb->write_index],
+           length - (rb->buffer_size - rb->write_index));
+
+    /* we are going into the other side of the mirror */
+    rb->write_mirror = ~rb->write_mirror;
+    rb->write_index = length - (rb->buffer_size - rb->write_index);
+
+    if (old_state == RT_RINGBUFFER_FULL)
+    {
+        rb->read_mirror = ~rb->read_mirror;
+        rb->read_index = rb->write_index;
+    }
+
+    return length;
+}
+RTM_EXPORT(rt_ringbuffer_put_force);
+
+/**
  *  get data from ring buffer
  */
 rt_size_t rt_ringbuffer_get(struct rt_ringbuffer *rb,
@@ -100,7 +156,7 @@ rt_size_t rt_ringbuffer_get(struct rt_ringbuffer *rb,
     RT_ASSERT(rb != RT_NULL);
 
     /* whether has enough data  */
-    size = RT_RINGBUFFER_SIZE(rb);
+    size = rt_ringbuffer_data_len(rb);
 
     /* no data */
     if (size == 0)
@@ -143,7 +199,7 @@ rt_size_t rt_ringbuffer_putchar(struct rt_ringbuffer *rb, const rt_uint8_t ch)
     RT_ASSERT(rb != RT_NULL);
 
     /* whether has enough space */
-    if (!RT_RINGBUFFER_EMPTY(rb))
+    if (!rt_ringbuffer_space_len(rb))
         return 0;
 
     rb->buffer_ptr[rb->write_index] = ch;
@@ -164,6 +220,43 @@ rt_size_t rt_ringbuffer_putchar(struct rt_ringbuffer *rb, const rt_uint8_t ch)
 RTM_EXPORT(rt_ringbuffer_putchar);
 
 /**
+ * put a character into ring buffer
+ *
+ * When the buffer is full, it will discard one old data.
+ */
+rt_size_t rt_ringbuffer_putchar_force(struct rt_ringbuffer *rb, const rt_uint8_t ch)
+{
+    enum rt_ringbuffer_state old_state;
+
+    RT_ASSERT(rb != RT_NULL);
+
+    old_state = rt_ringbuffer_status(rb);
+
+    rb->buffer_ptr[rb->write_index] = ch;
+
+    /* flip mirror */
+    if (rb->write_index == rb->buffer_size-1)
+    {
+        rb->write_mirror = ~rb->write_mirror;
+        rb->write_index = 0;
+        if (old_state == RT_RINGBUFFER_FULL)
+        {
+            rb->read_mirror = ~rb->read_mirror;
+            rb->read_index = rb->write_index;
+        }
+    }
+    else
+    {
+        rb->write_index++;
+        if (old_state == RT_RINGBUFFER_FULL)
+            rb->read_index = rb->write_index;
+    }
+
+    return 1;
+}
+RTM_EXPORT(rt_ringbuffer_putchar_force);
+
+/**
  * get a character from a ringbuffer
  */
 rt_size_t rt_ringbuffer_getchar(struct rt_ringbuffer *rb, rt_uint8_t *ch)
@@ -171,7 +264,7 @@ rt_size_t rt_ringbuffer_getchar(struct rt_ringbuffer *rb, rt_uint8_t *ch)
     RT_ASSERT(rb != RT_NULL);
 
     /* ringbuffer is empty */
-    if (!RT_RINGBUFFER_SIZE(rb))
+    if (!rt_ringbuffer_data_len(rb))
         return 0;
 
     /* put character */