Add pipe()

git-svn-id: svn://svn.code.sf.net/p/nuttx/code/trunk@779 42af7a65-404d-4744-a932-0658087f49c3

ChangeLog

@@ -375,3 +375,5 @@
 	* Add test for recursive mutexes
 	* Correct bug in recursive mutex logic
 	* Add mkfifo()
+	* Add pipe() and test for both pipes and fifos
+

Documentation/NuttX.html

@@ -8,7 +8,7 @@
   <tr align="center" bgcolor="#e4e4e4">
     <td>
       <h1><big><font color="#3c34ec"><i>NuttX RTOS</i></font></big></h1>
-      <p>Last Updated: July 20, 2008</p>
+      <p>Last Updated: July 26, 2008</p>
     </td>
   </tr>
 </table>
@@ -1024,6 +1024,7 @@ nuttx-0.3.12 2008-xx-xx Gregory Nutt &lt;spudmonkey@racsa.co.cr&gt;
 	* Add test for recursive mutexes
 	* Correct bug in recursive mutex logic
 	* Add mkfifo()
+	* Add pipe() and test for both pipes and fifos
 
 pascal-0.1.3 2008-xx-xx Gregory Nutt &lt;spudmonkey@racsa.co.cr&gt;
 

drivers/Makefile

@@ -44,7 +44,7 @@ AOBJS		= $(ASRCS:.S=$(OBJEXT))
 
 CSRCS		=
 ifneq ($(CONFIG_NFILE_DESCRIPTORS),0)
-CSRCS		+= dev_null.c dev_zero.c fifo.c serial.c lowconsole.c can.c
+CSRCS		+= dev_null.c dev_zero.c pipe.c fifo.c pipe-common.c serial.c lowconsole.c can.c
 ifneq ($(CONFIG_DISABLE_MOUNTPOINT),y)
 CSRCS           += ramdisk.c
 endif

drivers/fifo.c

@@ -44,79 +44,35 @@
 #include <nuttx/config.h>
 
 #include <sys/types.h>
-#include <sys/stat.h>
-#include <stdlib.h>
-#include <string.h>
-#include <sched.h>
-#include <semaphore.h>
-#include <fcntl.h>
-#include <errno.h>
-#include <assert.h>
 #include <nuttx/fs.h>
+#include <errno.h>
+
+#include "pipe-common.h"
 
-#ifndef CONFIG_DEV_FIFO_SIZE
-#  define CONFIG_DEV_FIFO_SIZE 1024
-#endif
 #if CONFIG_DEV_FIFO_SIZE > 0
 
 /****************************************************************************
  * Definitions
  ****************************************************************************/
 
-/* Maximum number of open's supported on FIFO */
-
-#define CONFIG_DEV_FIFO_MAXUSER 255
-
 /****************************************************************************
  * Private Types
  ****************************************************************************/
 
-/* Make the FIFO index as small as possible for the configured FIFO size */
- 
-#if CONFIG_DEV_FIFO_SIZE > 65535
-typedef uint32 fifo_ndx_t;  /* 32-bit index */
-#elif CONFIG_DEV_FIFO_SIZE > 255
-typedef uint16 fifo_ndx_t;  /* 16-bit index */
-#else
-typedef ubyte fifo_ndx_t;   /*  8-bit index */
-#endif
-
-struct fifo_dev_s
-{
-  sem_t      d_bfsem;       /* Used to serialize access to d_buffer and indices */
-  sem_t      d_rdsem;       /* Empty buffer - Reader waits for data write */
-  sem_t      d_wrsem;       /* Full buffer - Writer waits for data read */
-  fifo_ndx_t d_wrndx;       /* Index in d_buffer to save next byte written */
-  fifo_ndx_t d_rdndx;      /* Index in d_buffer to return the next byte read */
-  ubyte      d_refs;        /* References counts on FIFO (limited to 255) */
-  ubyte      d_nreaders;    /* Number of readers waiting for write data to empty buffer */
-  ubyte      d_nwriters;    /* Number of writers wiating for data read out of full buffer */
-  ubyte      d_buffer[CONFIG_DEV_FIFO_SIZE];
-};
-
 /****************************************************************************
  * Private Function Prototypes
  ****************************************************************************/
 
-static inline FAR struct fifo_dev_s *fifo_allocdev(void);
-static inline void fifo_freedev(FAR struct fifo_dev_s *dev);
-static void    fifo_semtake(sem_t *sem);
-
-static int     fifo_open(FAR struct file *filep);
-static int     fifo_close(FAR struct file *filep);
-static ssize_t fifo_read(FAR struct file *, FAR char *, size_t);
-static ssize_t fifo_write(FAR struct file *, FAR const char *, size_t);
-
 /****************************************************************************
  * Private Data
  ****************************************************************************/
 
 static struct file_operations fifo_fops =
 {
-  fifo_open,  /* open */
+  pipecommon_open,  /* open */
-  fifo_close, /* close */
+  pipecommon_close, /* close */
-  fifo_read,  /* read */
+  pipecommon_read,  /* read */
-  fifo_write, /* write */
+  pipecommon_write, /* write */
   0,                /* seek */
   0                 /* ioctl */
 };
@@ -125,325 +81,6 @@ static struct file_operations fifo_fops =
  * Private Functions
  ****************************************************************************/
 
-/****************************************************************************
- * Name: fifo_allocdev
- ****************************************************************************/
-static inline FAR struct fifo_dev_s *fifo_allocdev(void)
-{
- struct fifo_dev_s *dev;
-
-  /* Allocate a private structure to manage the FIFO */
-
-  dev = (struct fifo_dev_s *)malloc(sizeof(struct fifo_dev_s));
-  if (dev)
-    {
-      /* Initialize the private structure */
-
-      sem_init(&dev->d_bfsem, 0, 1);
-      sem_init(&dev->d_rdsem, 0, 0);
-      sem_init(&dev->d_wrsem, 0, 0);
-      dev->d_wrndx    = 0;
-      dev->d_rdndx   = 0;
-      dev->d_refs     = 0;
-      dev->d_nreaders = 0;
-      dev->d_nwriters = 0;
-    }
-  return dev;
-}
-
-/****************************************************************************
- * Name: fifo_freedev
- ****************************************************************************/
-static inline void fifo_freedev(FAR struct fifo_dev_s *dev)
-{
-   sem_destroy(&dev->d_bfsem);
-   sem_destroy(&dev->d_rdsem);
-   sem_destroy(&dev->d_wrsem);
-   free(dev);
-}
-
-/****************************************************************************
- * Name: fifo_semtake
- ****************************************************************************/
-static void fifo_semtake(sem_t *sem)
-{
-  while (sem_wait(sem) != 0)
-    {
-      /* The only case that an error should occur here is if the wait was
-       * awakened by a signal.
-       */
-
-      ASSERT(errno == EINTR);
-    }
-}
-
-/****************************************************************************
- * Name: fifo_open
- ****************************************************************************/
-static int fifo_open(FAR struct file *filep)
-{
-  struct inode      *inode = filep->f_inode;
-  struct fifo_dev_s *dev   = inode->i_private;
- 
-  /* The device structure will be allocated the first time that the FIFO is
-   * opened
-   */
-
-  if (!dev)
-    {
-      /* Allocate and initialize a new device structure instance */
-
-      dev = fifo_allocdev();
-      if (!dev)
-        {
-          return -ENOMEM;
-        }
-
-      /* Set the private data on the inode to this instance */
-
-      inode->i_private = dev;
-    }
-
-  /* Make sure that we have exclusive access to the device structure */
-
-  if (sem_wait(&dev->d_bfsem) == 0)
-    {
-      /* Increment the reference count on the fifo instance */
-
-      dev->d_refs++;
-      (void)sem_post(&dev->d_bfsem);
-      return OK;
-  }
-  return ERROR;
-}
-
-/****************************************************************************
- * Name: fifo_close
- ****************************************************************************/
-static int fifo_close(FAR struct file *filep)
-{
-  struct inode      *inode = filep->f_inode;
-  struct fifo_dev_s *dev   = inode->i_private;
-
-  /* Some sanity checking */
-#if CONFIG_DEBUG
-  if (!dev)
-    {
-       return -EBADF;
-    }
-#endif
-
-  /* Make sure that we have exclusive access to the device structure.
-   * NOTE: close() is supposed to return EINTR if interrupted, however
-   * I've never seen anyone check that.
-   */
-
-  fifo_semtake(&dev->d_bfsem);
-
-  /* Check if the decremented reference count would be zero */
-
-  if (dev->d_refs > 1)
-    {
-       /* No.. then just decrement the reference count */
-
-       dev->d_refs--;
-       sem_post(&dev->d_bfsem);
-    }
-  else
-    {
-       /* Then nothing else can be holding the semaphore, so it is save to */
-
-       inode->i_private = NULL;
-       sem_post(&dev->d_bfsem);
-
-       /* Then free the fifo structure instance */
-
-       fifo_freedev(dev);
-    }
-  return OK;
-}
-
-/****************************************************************************
- * Name: fifo_read
- ****************************************************************************/
-static ssize_t fifo_read(FAR struct file *filep, FAR char *buffer, size_t len)
-{
-  struct inode      *inode  = filep->f_inode;
-  struct fifo_dev_s *dev    = inode->i_private;
-  ssize_t            nread  = 0;
-  int                ret;
-
-  /* Some sanity checking */
-#if CONFIG_DEBUG
-  if (!dev)
-    {
-      return -ENODEV;
-    }
-#endif
-
-  /* Make sure that we have exclusive access to the device structure */
-
-  if (sem_wait(&dev->d_bfsem) < 0)
-    {
-      return ERROR;
-    }
-
-  /* If the fifo is empty, then wait for something to be written to it */
-
-  while (dev->d_wrndx == dev->d_rdndx)
-    {
-      /* If O_NONBLOCK was set, then return EGAIN */
-
-      if (filep->f_oflags & O_NONBLOCK)
-        {
-          sem_post(&dev->d_bfsem);
-          return -EAGAIN;
-        }
-
-      /* Otherwise, wait for something to be written to the FIFO */
-
-      dev->d_nreaders++;
-      sched_lock();
-      sem_post(&dev->d_bfsem);
-      ret = sem_wait(&dev->d_rdsem);
-      sched_unlock();
-      if (ret < 0  || sem_wait(&dev->d_bfsem) < 0) 
-        {
-          return ERROR;
-        }
-    }
-
-  /* Then return whatever is available in the FIFO (which is at least one byte) */
-
-  nread = 0;
-  while (nread < len && dev->d_wrndx != dev->d_rdndx)
-    {
-      *buffer++ = dev->d_buffer[dev->d_rdndx];
-      if (++dev->d_rdndx >= CONFIG_DEV_FIFO_SIZE)
-        {
-          dev->d_rdndx = 0; 
-        }
-      nread++;
-    }
-
-  /* Notify any waiting writers that bytes have been removed from the buffer */
-
-  if (dev->d_nwriters > 0)
-    {
-      dev->d_nwriters--;
-      sem_post(&dev->d_wrsem);
-    }
-
-  sem_post(&dev->d_bfsem);
-  return nread;	    
-}
-
-/****************************************************************************
- * Name: fifo_write
- ****************************************************************************/
-static ssize_t fifo_write(FAR struct file *filep, FAR const char *buffer, size_t len)
-{
-  struct inode      *inode    = filep->f_inode;
-  struct fifo_dev_s *dev      = inode->i_private;
-  ssize_t            nwritten = 0;
-  ssize_t            last;
-  int                nxtwrndx;
-
-  /* Some sanity checking */
-#if CONFIG_DEBUG
-  if (!dev)
-    {
-      return -ENODEV;
-    }
-#endif
-
-  /* Make sure that we have exclusive access to the device structure */
-
-  if (sem_wait(&dev->d_bfsem) < 0)
-    {
-      return ERROR;
-    }
-
-  /* Loop until all of the bytes have been written */
-
-  last = 0;
-  for (;;)
-    {
-      /* Calculate the write index AFTER the next byte is written */
-
-      nxtwrndx = dev->d_wrndx + 1;
-      if (nxtwrndx >= CONFIG_DEV_FIFO_SIZE)
-        {
-          nxtwrndx = 0;
-        }
-
-      /* Would the next write overflow the circular buffer? */
-
-      if (nxtwrndx != dev->d_rdndx)
-        {
-          /* No... copy the byte */
-
-          dev->d_buffer[dev->d_wrndx] = *buffer++;
-          dev->d_wrndx = nxtwrndx;
-
-          /* Is the write complete? */
-
-          if (++nwritten >= len)
-            {
-              /* Yes.. Notify the waiting readers that more data is available */
-
-              if (dev->d_nreaders > 0)
-                {
-                  dev->d_nreaders--;
-                  sem_post(&dev->d_rdsem);
-                }
-
-              /* Return the number of bytes written */
-
-              sem_post(&dev->d_bfsem);
-              return len;
-            }
-        }
-      else
-        {
-          /* There is not enough room for the next byte.  Was anything written in this pass? */
-
-          if (last < nwritten)
-            {
-              /* Yes.. Notify the waiting readers that more data is available */
-
-              if (dev->d_nreaders > 0)
-                {
-                  dev->d_nreaders--;
-                  sem_post(&dev->d_rdsem);
-                }
-            }
-          last = nwritten;
-
-          /* If O_NONBLOCK was set, then return partial bytes written or EGAIN */
-
-          if (filep->f_oflags & O_NONBLOCK)
-            {
-              if (nwritten == 0)
-                {
-                  nwritten = -EAGAIN;
-                }
-              sem_post(&dev->d_bfsem);
-              return nwritten;
-            }
-
-          /* There is more to be written.. wait for data to be removed from the FIFO */
-
-          dev->d_nwriters++;
-          sched_lock();
-          sem_post(&dev->d_bfsem);
-          fifo_semtake(&dev->d_wrsem);
-          sched_unlock();
-          fifo_semtake(&dev->d_bfsem);
-        }
-    }
-}
-
 /****************************************************************************
  * Public Functions
  ****************************************************************************/
@@ -457,8 +94,10 @@ static ssize_t fifo_write(FAR struct file *filep, FAR const char *buffer, size_t
  *   instance.  'mode' specifies the FIFO's permissions. 
  *
  *   Once the FIFO has been created by mkfifo(), any thread can open it for
- *   reading or writing, in the same way as an ordinary file.  NuttX FIFOs need
+ *   reading or writing, in the same way as an ordinary file. However, it must
- *   not be open at both ends before input or output operations on it.
+ *   have been opened from both reading and writing before input or output
+ *   can be performed.  Unlike other mkfifo() implementations, this one will
+ *   NOT block when the FIFO is opened on only one end.
  *
  * Inputs:
  *   pathname - The full path to the FIFO instance to attach to or to create
@@ -472,6 +111,22 @@ static ssize_t fifo_write(FAR struct file *filep, FAR const char *buffer, size_t
  ****************************************************************************/
 int mkfifo(FAR const char *pathname, mode_t mode)
 {
-  return register_driver(pathname, &fifo_fops, mode, NULL);
+  struct pipe_dev_s *dev;
+  int ret;
+ 
+  /* Allocate and initialize a new device structure instance */
+
+  dev = pipecommon_allocdev();
+  if (!dev)
+    {
+      return -ENOMEM;
+    }
+
+  ret = register_driver(pathname, &fifo_fops, mode, (void*)dev);
+  if (ret != 0)
+    {
+      pipecommon_freedev(dev);
+    }
+  return ret;
 }
-#endif
+#endif /* CONFIG_DEV_FIFO_SIZE > 0 */

drivers/pipe-common.c

@@ -0,0 +1,421 @@
+/****************************************************************************
+ * drivers/pipe-common.c
+ *
+ *   Copyright (C) 2008 Gregory Nutt. All rights reserved.
+ *   Author: Gregory Nutt <spudmonkey@racsa.co.cr>
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in
+ *    the documentation and/or other materials provided with the
+ *    distribution.
+ * 3. Neither the name NuttX nor the names of its contributors may be
+ *    used to endorse or promote products derived from this software
+ *    without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
+ * FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
+ * COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+ * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
+ * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS
+ * OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
+ * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+ * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+ * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ *
+ ****************************************************************************/
+
+/****************************************************************************
+ * Compilation Switches
+ ****************************************************************************/
+
+/****************************************************************************
+ * Included Files
+ ****************************************************************************/
+
+#include <nuttx/config.h>
+
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sched.h>
+#include <semaphore.h>
+#include <fcntl.h>
+#include <errno.h>
+#include <assert.h>
+#include <nuttx/fs.h>
+
+#include "pipe-common.h"
+
+#if CONFIG_DEV_FIFO_SIZE > 0
+
+/****************************************************************************
+ * Definitions
+ ****************************************************************************/
+
+/****************************************************************************
+ * Private Types
+ ****************************************************************************/
+
+/****************************************************************************
+ * Private Function Prototypes
+ ****************************************************************************/
+
+static void    pipecommon_semtake(sem_t *sem);
+
+/****************************************************************************
+ * Private Data
+ ****************************************************************************/
+
+/****************************************************************************
+ * Private Functions
+ ****************************************************************************/
+
+/****************************************************************************
+ * Name: pipecommon_semtake
+ ****************************************************************************/
+static void pipecommon_semtake(sem_t *sem)
+{
+  while (sem_wait(sem) != 0)
+    {
+      /* The only case that an error should occur here is if the wait was
+       * awakened by a signal.
+       */
+
+      ASSERT(errno == EINTR);
+    }
+}
+
+/****************************************************************************
+ * Public Functions
+ ****************************************************************************/
+
+/****************************************************************************
+ * Name: pipecommon_allocdev
+ ****************************************************************************/
+FAR struct pipe_dev_s *pipecommon_allocdev(void)
+{
+ struct pipe_dev_s *dev;
+
+  /* Allocate a private structure to manage the FIFO */
+
+  dev = (struct pipe_dev_s *)malloc(sizeof(struct pipe_dev_s));
+  if (dev)
+    {
+      /* Initialize the private structure */
+
+      memset(&dev->s, 0, sizeof(struct pipe_state_s));
+      sem_init(&dev->s.d_bfsem, 0, 1);
+      sem_init(&dev->s.d_rdsem, 0, 0);
+      sem_init(&dev->s.d_wrsem, 0, 0);
+    }
+  return dev;
+}
+
+/****************************************************************************
+ * Name: pipecommon_freedev
+ ****************************************************************************/
+void pipecommon_freedev(FAR struct pipe_dev_s *dev)
+{
+   sem_destroy(&dev->s.d_bfsem);
+   sem_destroy(&dev->s.d_rdsem);
+   sem_destroy(&dev->s.d_wrsem);
+   free(dev);
+}
+
+/****************************************************************************
+ * Name: pipecommon_open
+ ****************************************************************************/
+int pipecommon_open(FAR struct file *filep)
+{
+  struct inode      *inode = filep->f_inode;
+  struct pipe_dev_s *dev   = inode->i_private;
+ 
+  /* Some sanity checking */
+#if CONFIG_DEBUG
+  if (!dev)
+    {
+       return -EBADF;
+    }
+#endif
+  /* Make sure that we have exclusive access to the device structure */
+
+  if (sem_wait(&dev->s.d_bfsem) == 0)
+    {
+      /* Increment the reference count on the pipe instance */
+
+      dev->s.d_refs++;
+
+      /* If opened for writing, increment the count of writers on on the pipe instance */
+
+      if ((filep->f_oflags & O_WROK) != 0)
+        {
+          dev->s.d_nwriters++;
+        }
+
+      (void)sem_post(&dev->s.d_bfsem);
+      return OK;
+  }
+  return ERROR;
+}
+
+/****************************************************************************
+ * Name: pipecommon_close
+ ****************************************************************************/
+int pipecommon_close(FAR struct file *filep)
+{
+  struct inode      *inode = filep->f_inode;
+  struct pipe_dev_s *dev   = inode->i_private;
+
+  /* Some sanity checking */
+#if CONFIG_DEBUG
+  if (!dev)
+    {
+       return -EBADF;
+    }
+#endif
+
+  /* Make sure that we have exclusive access to the device structure.
+   * NOTE: close() is supposed to return EINTR if interrupted, however
+   * I've never seen anyone check that.
+   */
+
+  pipecommon_semtake(&dev->s.d_bfsem);
+
+  /* Check if the decremented reference count would be zero */
+
+  if (dev->s.d_refs > 1)
+    {
+       /* No.. then just decrement the reference count */
+
+       dev->s.d_refs--;
+
+      /* If opened for writing, decrement the count of writers on on the pipe instance */
+
+      if ((filep->f_oflags & O_WROK) != 0)
+        {
+          /* If there are no longer any writers on the pipe, then notify any
+           * waiting readers that must return end-of-file.
+           */
+
+          if (--dev->s.d_nwriters <= 0 && dev->s.d_rdwaiters > 0)
+            {
+              dev->s.d_rdwaiters--;
+              sem_post(&dev->s.d_rdsem);
+            }
+        }
+       sem_post(&dev->s.d_bfsem);
+    }
+  else
+    {
+       /* Then nothing else can be holding the semaphore, so it is save to */
+
+       inode->i_private = NULL;
+       sem_post(&dev->s.d_bfsem);
+
+       /* Then free the fifo structure instance */
+
+       pipecommon_freedev(dev);
+    }
+  return OK;
+}
+
+/****************************************************************************
+ * Name: pipecommon_read
+ ****************************************************************************/
+ssize_t pipecommon_read(FAR struct file *filep, FAR char *buffer, size_t len)
+{
+  struct inode      *inode  = filep->f_inode;
+  struct pipe_dev_s *dev    = inode->i_private;
+  ssize_t            nread  = 0;
+  int                ret;
+
+  /* Some sanity checking */
+#if CONFIG_DEBUG
+  if (!dev)
+    {
+      return -ENODEV;
+    }
+#endif
+
+  /* Make sure that we have exclusive access to the device structure */
+
+  if (sem_wait(&dev->s.d_bfsem) < 0)
+    {
+      return ERROR;
+    }
+
+  /* If the fifo is empty, then wait for something to be written to it */
+
+  while (dev->s.d_wrndx == dev->s.d_rdndx)
+    {
+      /* If O_NONBLOCK was set, then return EGAIN */
+
+      if (filep->f_oflags & O_NONBLOCK)
+        {
+          sem_post(&dev->s.d_bfsem);
+          return -EAGAIN;
+        }
+
+      /* If there are no writers on the pipe, then return end of file */
+
+      if (dev->s.d_nwriters <= 0)
+        {
+          sem_post(&dev->s.d_bfsem);
+          return 0;
+        }
+
+      /* Otherwise, wait for something to be written to the FIFO */
+
+      dev->s.d_rdwaiters++;
+      sched_lock();
+      sem_post(&dev->s.d_bfsem);
+      ret = sem_wait(&dev->s.d_rdsem);
+      sched_unlock();
+      if (ret < 0  || sem_wait(&dev->s.d_bfsem) < 0) 
+        {
+          return ERROR;
+        }
+    }
+
+  /* Then return whatever is available in the FIFO (which is at least one byte) */
+
+  nread = 0;
+  while (nread < len && dev->s.d_wrndx != dev->s.d_rdndx)
+    {
+      *buffer++ = dev->d_buffer[dev->s.d_rdndx];
+      if (++dev->s.d_rdndx >= CONFIG_DEV_FIFO_SIZE)
+        {
+          dev->s.d_rdndx = 0; 
+        }
+      nread++;
+    }
+
+  /* Notify any waiting writers that bytes have been removed from the buffer */
+
+  if (dev->s.d_nwrwaiters > 0)
+    {
+      dev->s.d_nwrwaiters--;
+      sem_post(&dev->s.d_wrsem);
+    }
+
+  sem_post(&dev->s.d_bfsem);
+  return nread;	    
+}
+
+/****************************************************************************
+ * Name: pipecommon_write
+ ****************************************************************************/
+ssize_t pipecommon_write(FAR struct file *filep, FAR const char *buffer, size_t len)
+{
+  struct inode      *inode    = filep->f_inode;
+  struct pipe_dev_s *dev      = inode->i_private;
+  ssize_t            nwritten = 0;
+  ssize_t            last;
+  int                nxtwrndx;
+
+  /* Some sanity checking */
+#if CONFIG_DEBUG
+  if (!dev)
+    {
+      return -ENODEV;
+    }
+#endif
+
+  /* Make sure that we have exclusive access to the device structure */
+
+  if (sem_wait(&dev->s.d_bfsem) < 0)
+    {
+      return ERROR;
+    }
+
+  /* Loop until all of the bytes have been written */
+
+  last = 0;
+  for (;;)
+    {
+      /* Calculate the write index AFTER the next byte is written */
+
+      nxtwrndx = dev->s.d_wrndx + 1;
+      if (nxtwrndx >= CONFIG_DEV_FIFO_SIZE)
+        {
+          nxtwrndx = 0;
+        }
+
+      /* Would the next write overflow the circular buffer? */
+
+      if (nxtwrndx != dev->s.d_rdndx)
+        {
+          /* No... copy the byte */
+
+          dev->d_buffer[dev->s.d_wrndx] = *buffer++;
+          dev->s.d_wrndx = nxtwrndx;
+
+          /* Is the write complete? */
+
+          if (++nwritten >= len)
+            {
+              /* Yes.. Notify the waiting readers that more data is available */
+
+              if (dev->s.d_rdwaiters > 0)
+                {
+                  dev->s.d_rdwaiters--;
+                  sem_post(&dev->s.d_rdsem);
+                }
+
+              /* Return the number of bytes written */
+
+              sem_post(&dev->s.d_bfsem);
+              return len;
+            }
+        }
+      else
+        {
+          /* There is not enough room for the next byte.  Was anything written in this pass? */
+
+          if (last < nwritten)
+            {
+              /* Yes.. Notify the waiting readers that more data is available */
+
+              if (dev->s.d_rdwaiters > 0)
+                {
+                  dev->s.d_rdwaiters--;
+                  sem_post(&dev->s.d_rdsem);
+                }
+            }
+          last = nwritten;
+
+          /* If O_NONBLOCK was set, then return partial bytes written or EGAIN */
+
+          if (filep->f_oflags & O_NONBLOCK)
+            {
+              if (nwritten == 0)
+                {
+                  nwritten = -EAGAIN;
+                }
+              sem_post(&dev->s.d_bfsem);
+              return nwritten;
+            }
+
+          /* There is more to be written.. wait for data to be removed from the FIFO */
+
+          dev->s.d_nwrwaiters++;
+          sched_lock();
+          sem_post(&dev->s.d_bfsem);
+          pipecommon_semtake(&dev->s.d_wrsem);
+          sched_unlock();
+          pipecommon_semtake(&dev->s.d_bfsem);
+        }
+    }
+}
+
+#endif /* CONFIG_DEV_FIFO_SIZE > 0 */

drivers/pipe-common.h

@@ -0,0 +1,117 @@
+/****************************************************************************
+ * drivers/pipe-common.h
+ *
+ *   Copyright (C) 2008 Gregory Nutt. All rights reserved.
+ *   Author: Gregory Nutt <spudmonkey@racsa.co.cr>
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in
+ *    the documentation and/or other materials provided with the
+ *    distribution.
+ * 3. Neither the name NuttX nor the names of its contributors may be
+ *    used to endorse or promote products derived from this software
+ *    without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
+ * FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
+ * COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+ * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
+ * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS
+ * OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
+ * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+ * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+ * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ *
+ ****************************************************************************/
+
+#ifndef __DRIVERS_PIPE_COMMON_H
+#define __DRIVERS_PIPE_COMMON_H
+
+/****************************************************************************
+ * Included Files
+ ****************************************************************************/
+
+#include <nuttx/config.h>
+#include <sys/types.h>
+
+#ifndef CONFIG_DEV_FIFO_SIZE
+#  define CONFIG_DEV_FIFO_SIZE 1024
+#endif
+#if CONFIG_DEV_FIFO_SIZE > 0
+
+/****************************************************************************
+ * Definitions
+ ****************************************************************************/
+
+/* Maximum number of open's supported on FIFO */
+
+#define CONFIG_DEV_FIFO_MAXUSER 255
+
+/****************************************************************************
+ * Public Types
+ ****************************************************************************/
+
+/* Make the FIFO index as small as possible for the configured FIFO size */
+ 
+#if CONFIG_DEV_FIFO_SIZE > 65535
+typedef uint32 pipe_ndx_t;  /* 32-bit index */
+#elif CONFIG_DEV_FIFO_SIZE > 255
+typedef uint16 pipe_ndx_t;  /* 16-bit index */
+#else
+typedef ubyte pipe_ndx_t;   /*  8-bit index */
+#endif
+
+struct pipe_state_s
+{
+  sem_t      d_bfsem;       /* Used to serialize access to d_buffer and indices */
+  sem_t      d_rdsem;       /* Empty buffer - Reader waits for data write */
+  sem_t      d_wrsem;       /* Full buffer - Writer waits for data read */
+  pipe_ndx_t d_wrndx;       /* Index in d_buffer to save next byte written */
+  pipe_ndx_t d_rdndx;       /* Index in d_buffer to return the next byte read */
+  ubyte      d_refs;        /* References counts on FIFO (limited to 255) */
+  ubyte      d_nwriters;    /* Number of open counts for write access */
+  ubyte      d_rdwaiters;   /* Number of readers waiting for write data to empty buffer */
+  ubyte      d_nwrwaiters;  /* Number of writers wiating for data read out of full buffer */
+  ubyte      d_pipeno;      /* Pipe minor number */
+};
+
+struct pipe_dev_s
+{
+  struct pipe_state_s s;
+  ubyte d_buffer[CONFIG_DEV_FIFO_SIZE];
+};
+
+/****************************************************************************
+ * Public Function Prototypes
+ ****************************************************************************/
+
+#ifdef __cplusplus
+#  define EXTERN extern "C"
+extern "C" {
+#else
+#  define EXTERN extern
+#endif
+
+EXTERN FAR struct pipe_dev_s *pipecommon_allocdev(void);
+EXTERN void    pipecommon_freedev(FAR struct pipe_dev_s *dev);
+EXTERN int     pipecommon_open(FAR struct file *filep);
+EXTERN int     pipecommon_close(FAR struct file *filep);
+EXTERN ssize_t pipecommon_read(FAR struct file *, FAR char *, size_t);
+EXTERN ssize_t pipecommon_write(FAR struct file *, FAR const char *, size_t);
+
+#undef EXTERN
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* CONFIG_DEV_FIFO_SIZE > 0 */
+#endif /* __DRIVERS_PIPE_COMMON_H */

drivers/pipe.c

@@ -0,0 +1,257 @@
+/****************************************************************************
+ * drivers/pipe.c
+ *
+ *   Copyright (C) 2008 Gregory Nutt. All rights reserved.
+ *   Author: Gregory Nutt <spudmonkey@racsa.co.cr>
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in
+ *    the documentation and/or other materials provided with the
+ *    distribution.
+ * 3. Neither the name NuttX nor the names of its contributors may be
+ *    used to endorse or promote products derived from this software
+ *    without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
+ * FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
+ * COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+ * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
+ * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS
+ * OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
+ * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+ * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+ * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ *
+ ****************************************************************************/
+
+/****************************************************************************
+ * Compilation Switches
+ ****************************************************************************/
+
+/****************************************************************************
+ * Included Files
+ ****************************************************************************/
+
+#include <nuttx/config.h>
+
+#include <sys/types.h>
+#include <nuttx/fs.h>
+#include <stdio.h>
+#include <unistd.h>
+#include <semaphore.h>
+#include <fcntl.h>
+#include <errno.h>
+
+#include "pipe-common.h"
+
+#if CONFIG_DEV_FIFO_SIZE > 0
+
+/****************************************************************************
+ * Definitions
+ ****************************************************************************/
+
+#define MAX_PIPES 32
+
+/****************************************************************************
+ * Private Types
+ ****************************************************************************/
+
+/****************************************************************************
+ * Private Function Prototypes
+ ****************************************************************************/
+
+static int pipe_close(FAR struct file *filep);
+
+/****************************************************************************
+ * Private Data
+ ****************************************************************************/
+
+static struct file_operations pipe_fops =
+{
+  pipecommon_open,   /* open */
+  pipe_close,        /* close */
+  pipecommon_read,   /* read */
+  pipecommon_write,  /* write */
+  0,                 /* seek */
+  0                  /* ioctl */
+};
+
+static sem_t  g_pipesem = { 1 };
+static uint32 g_pipeset = 0;
+
+/****************************************************************************
+ * Private Functions
+ ****************************************************************************/
+
+/****************************************************************************
+ * Name: pipe_allocate
+ ****************************************************************************/
+static inline int pipe_allocate(void)
+{
+  int pipeno;
+  int ret = sem_wait(&g_pipesem);
+  if (ret >= 0)
+    {
+      ret = -ENFILE;
+      for (pipeno = 0; pipeno < MAX_PIPES; pipeno++)
+        {
+          if ((g_pipeset & (1 << pipeno)) == 0)
+            {
+              g_pipeset |= (1 << pipeno);
+              ret = pipeno;
+              break;
+            }
+        }
+      (void)sem_post(&g_pipesem);
+    }
+  return ret;
+}
+
+/****************************************************************************
+ * Name: pipe_free
+ ****************************************************************************/
+static inline void pipe_free(int pipeno)
+{
+  int ret = sem_wait(&g_pipesem);
+  if (ret == 0)
+    {
+      g_pipeset &= ~(1 << pipeno);
+      (void)sem_post(&g_pipesem);
+    }
+}
+
+/****************************************************************************
+ * Name: pipe_close
+ ****************************************************************************/
+static int pipe_close(FAR struct file *filep)
+{
+  struct inode      *inode = filep->f_inode;
+  struct pipe_dev_s *dev   = inode->i_private;
+  ubyte              pipeno;
+  int                ret;
+
+  /* Some sanity checking */
+#if CONFIG_DEBUG
+  if (!dev)
+    {
+       return -EBADF;
+    }
+#endif
+  pipeno = dev->s.d_pipeno;
+
+  /* Perform common close operations */
+
+  ret =  pipecommon_close(filep);
+  if (ret == 0 && !inode->i_private)
+    {
+      char devname[16];
+      sprintf(devname, "/dev/pipe%d", pipeno);
+      unlink(devname);
+      pipe_free(pipeno);
+    }
+  return ret;
+}
+
+/****************************************************************************
+ * Public Functions
+ ****************************************************************************/
+
+/****************************************************************************
+ * Name: pipe
+ *
+ * Description:
+ *   pipe() creates a pair of file descriptors, pointing to a pipe inode, and
+ *   places them in the array pointed to by 'filedes'. filedes[0] is for reading,
+ *   filedes[1] is for writing. 
+ *
+ * Inputs:
+ *   filedes[2] - The user provided array in which to catch the pipe file
+ *   descriptors
+ *
+ * Return:
+ *   0 is returned on success; otherwise, -1 is returned with errno set
+ *   appropriately.
+ *
+ ****************************************************************************/
+int pipe(int filedes[2])
+{
+  struct pipe_dev_s *dev;
+  char devname[16];
+  int pipeno;
+  int err;
+  int ret;
+
+  /* Allocate a minor number for the pipe device */
+
+  pipeno = pipe_allocate();
+  if (pipeno < 0)
+    {
+      err = -pipeno;
+      goto errout;
+    }
+
+  /* Allocate and initialize a new device structure instance */
+
+  dev = pipecommon_allocdev();
+  if (!dev)
+    {
+      pipe_free(pipeno);
+      err = ENOMEM;
+      goto errout;
+    }
+  dev->s.d_pipeno = pipeno;
+
+  /* Create a pathname to the pipe device */
+
+  sprintf(devname, "/dev/pipe%d", pipeno);
+
+  /* Register the pipe device */
+
+  ret = register_driver(devname, &pipe_fops, 0666, (void*)dev);
+  if (ret != 0)
+    {
+      err = -ret;
+      goto errout_with_dev;
+    }
+
+  /* Get a write file descriptor */
+
+  filedes[1] = open(devname, O_WRONLY);
+  if (filedes[1] < 0)
+    {
+      err = -filedes[1];
+      goto errout_with_driver;
+    }
+
+  /* Get a read file descriptor */
+
+  filedes[0] = open(devname, O_RDONLY);
+  if (filedes[0] < 0)
+    {
+      err = -filedes[0];
+      goto errout_with_wrfd;
+    }
+
+  return OK;
+
+errout_with_wrfd:
+  close(filedes[1]);
+errout_with_driver:
+  unregister_driver(devname);
+errout_with_dev:
+  pipecommon_freedev(dev);
+errout:
+  errno = err;
+  return ERROR;
+}
+
+#endif /* CONFIG_DEV_FIFO_SIZE > 0 */

examples/pipe/pipe_main.c

@@ -102,10 +102,13 @@ static void *reader(pthread_addr_t pvarg)
         }
       else if (ret == 0)
         {
-#warning BUG: pipe should return zero when writer closes
+          if (nbytes < NREAD_BYTES)
-           fprintf(stderr, "reader: Received zero bytes\n");
+            {
+              fprintf(stderr, "reader: Too few bytes read -- aborting: %d\n", nbytes);
               return (void*)2;
             }
+          break;
+        }
       for (ndx = 0; ndx < ret; ndx++)
         {
           if (value >= WRITE_SIZE)
@@ -121,6 +124,11 @@ static void *reader(pthread_addr_t pvarg)
           value++;
         }
       nbytes += ret;
+      if (nbytes > NREAD_BYTES)
+        {
+          fprintf(stderr, "reader: Too many bytes read -- aborting: %d\n", nbytes);
+          return (void*)3;
+        }
     }
   printf("reader: %d bytes read\n", nbytes);
   return (void*)0;
@@ -252,8 +260,7 @@ void user_initialize(void)
 
 int user_start(int argc, char *argv[])
 {
-  int fdin;
+  int filedes[2];
-  int fdout;
   int ret;
 
   /* Test FIFO logic */
@@ -266,26 +273,33 @@ int user_start(int argc, char *argv[])
       return 1;
     }
 
-  fdin = open(CONFIG_EXAMPLES_FIFO_PATH, O_RDONLY);
+  /* Open open end of the FIFO for reading and the other end for writing.  NOTE:
-  if (fdin < 0)
+   * the following would not work on most FIFO implementations because the attempt
+   * to open just one end of the FIFO would block.  The NuttX FIFOs do not block.
+   */
+
+  filedes[1] = open(CONFIG_EXAMPLES_FIFO_PATH, O_WRONLY);
+  if (filedes[1] < 0)
+    {
+      fprintf(stderr, "user_start: Failed to open FIFO %s for writing, errno=%d\n",
+              CONFIG_EXAMPLES_FIFO_PATH, errno);
+      close(filedes[0]);
+      return 3;
+    }
+
+  filedes[0] = open(CONFIG_EXAMPLES_FIFO_PATH, O_RDONLY);
+  if (filedes[0] < 0)
     {
       fprintf(stderr, "user_start: Failed to open FIFO %s for reading, errno=%d\n",
               CONFIG_EXAMPLES_FIFO_PATH, errno);
       return 2;
     }
 
-  fdout = open(CONFIG_EXAMPLES_FIFO_PATH, O_WRONLY);
+  /* Then perform the test using those file descriptors */
-  if (fdout < 0)
-    {
-      fprintf(stderr, "user_start: Failed to open FIFO %s for writing, errno=%d\n",
-              CONFIG_EXAMPLES_FIFO_PATH, errno);
-      close(fdin);
-      return 3;
-    }
 
-  ret = perform_test(fdin, fdout);
+  ret = perform_test(filedes[0], filedes[1]);
-  close(fdin);
+  close(filedes[0]);
-  close(fdout);
+  close(filedes[1]);
   unlink(CONFIG_EXAMPLES_FIFO_PATH);
   if (ret != 0)
     {
@@ -297,7 +311,25 @@ int user_start(int argc, char *argv[])
   /* Test PIPE logic */
 
   printf("user_start: Performing pipe test\n");
-  /* Not yet implemented */
+  ret = pipe(filedes);
+  if (ret < 0)
+    {
+      fprintf(stderr, "user_start: pipe failed with errno=%d\n", errno);
+      return 1;
+    }
+
+  /* Then perform the test using those file descriptors */
+
+  ret = perform_test(filedes[0], filedes[1]);
+  close(filedes[0]);
+  close(filedes[1]);
+  unlink(CONFIG_EXAMPLES_FIFO_PATH);
+  if (ret != 0)
+    {
+      fprintf(stderr, "user_start: PIPE test FAILED\n");
+      return 4;
+    }
+  printf("user_start: PIPE test PASSED\n");
 
   fflush(stdout);
   return 0;

include/stddef.h

@@ -1,7 +1,7 @@
-/************************************************************
+/****************************************************************************
- * stddef.h
+ * include/stddef.h
  *
- *   Copyright (C) 2007 Gregory Nutt. All rights reserved.
+ *   Copyright (C) 2007, 2008 Gregory Nutt. All rights reserved.
  *   Author: Gregory Nutt <spudmonkey@racsa.co.cr>
  *
  * Redistribution and use in source and binary forms, with or without
@@ -14,7 +14,7 @@
  *    notice, this list of conditions and the following disclaimer in
  *    the documentation and/or other materials provided with the
  *    distribution.
- * 3. Neither the name Gregory Nutt nor the names of its contributors may be
+ * 3. Neither the name NuttX nor the names of its contributors may be
  *    used to endorse or promote products derived from this software
  *    without specific prior written permission.
  *
@@ -31,19 +31,19 @@
  * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  * POSSIBILITY OF SUCH DAMAGE.
  *
- ************************************************************/
+ ****************************************************************************/
 
 #ifndef __STDDEF_H
 #define __STDDEF_H
 
-/************************************************************
+/****************************************************************************
  * Included Files
- ************************************************************/
+ ****************************************************************************/
 
 #include <sys/types.h>
 
-/************************************************************
+/****************************************************************************
  * Type Definitions
- ************************************************************/
+ ****************************************************************************/
 
 #endif /* __STDDEF_H */

include/unistd.h

@@ -135,6 +135,9 @@ EXTERN off_t lseek(int fd, off_t offset, int whence);
 EXTERN int   read(int fd, FAR void *buf, unsigned int nbytes);
 EXTERN int   write(int fd, FAR const void *buf, unsigned int nbytes);
 
+/* Special devices */
+EXTERN int   pipe(int filedes[2]);
+
 /* File path operations */
 
 EXTERN int   unlink(FAR const char *pathname);
@@ -143,7 +146,6 @@ EXTERN int rmdir(FAR const char *pathname);
 /* Other */
 
 EXTERN int   getopt(int argc, FAR char *const argv[], FAR const char *optstring);
-EXTERN int pipe(int filedes[2]);
 
 #undef EXTERN
 #if defined(__cplusplus)