Initial revision

9 files changed, 1357 insertions, 0 deletions

diff --git a/libpipe/Makefile b/libpipe/Makefile
new file mode 100644
index 00000000..2b995a18
--- /dev/null
+++ b/libpipe/Makefile
@@ -0,0 +1,30 @@
+# Makefile for libpipe
+# 
+#   Copyright (C) 1995 Free Software Foundation, Inc.
+#
+#   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 the Free Software Foundation; either version 2, or (at
+#   your option) any later version.
+#
+#   This program is distributed in the hope that it will be useful, but
+#   WITHOUT ANY WARRANTY; without even the implied warranty of
+#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+#   General Public License for more details.
+#
+#   You should have received a copy of the GNU General Public License
+#   along with this program; if not, write to the Free Software
+#   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+
+dir := libpipe
+makemode := library
+
+libname = libpipe
+installhdrs = pipe.h pq.h
+
+SRCS = pq.c dgram.c pipe.c stream.c seqpack.c addr.c
+LCLHDRS = pipe.h pq.h
+
+OBJS = $(SRCS:.c=.o)
+
+include ../Makeconf
diff --git a/libpipe/addr.c b/libpipe/addr.c
new file mode 100644
index 00000000..3af3ef8f
--- /dev/null
+++ b/libpipe/addr.c
@@ -0,0 +1,29 @@
+/* Default address deallocate function
+
+   Copyright (C) 1995 Free Software Foundation, Inc.
+
+   Written by Miles Bader <miles@gnu.ai.mit.edu>
+
+   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 the Free Software Foundation; either version 2, or (at
+   your option) any later version.
+
+   This program is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. */
+
+#include <hurd/ports.h>
+
+/* This routine may be provided by the user, in which case, it should be a
+   function taking a non-NULL source address and deallocating it.  It
+   defaults to calling ports_port_deref.  */
+void pipe_dealloc_addr (void *addr)
+{
+  ports_port_deref (addr);
+}
diff --git a/libpipe/dgram.c b/libpipe/dgram.c
new file mode 100644
index 00000000..3f3b2ab6
--- /dev/null
+++ b/libpipe/dgram.c
@@ -0,0 +1,51 @@
+/* The SOCK_DGRAM pipe class
+
+   Copyright (C) 1995 Free Software Foundation, Inc.
+
+   Written by Miles Bader <miles@gnu.ai.mit.edu>
+
+   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 the Free Software Foundation; either version 2, or (at
+   your option) any later version.
+
+   This program is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. */
+
+#include <sys/socket.h>		/* For SOCK_DGRAM */
+
+#include "pipe.h"
+#include "pq.h"
+
+/* See the definition of struct pipe_class in "pipe.h" for documentation.  */
+
+static error_t 
+dgram_write (struct pq *pq, void *source,
+	     char *data, size_t data_len, size_t *amount)
+{
+  struct packet *packet = pq_queue (pq, PACKET_TYPE_DATA, source);
+  if (!packet)
+    return ENOBUFS;
+  else
+    return packet_write (packet, data, data_len, amount);
+}
+
+static error_t 
+dgram_read (struct packet *packet, int *dequeue, unsigned *flags,
+	    char **data, size_t *data_len, size_t amount)
+{
+  *dequeue = 1;
+  return packet_read (packet, data, data_len, amount);
+}
+
+struct pipe_class _dgram_pipe_class =
+{
+  SOCK_DGRAM, PIPE_CLASS_CONNECTIONLESS, dgram_read, dgram_write
+};
+struct pipe_class *dgram_pipe_class = &_dgram_pipe_class;
diff --git a/libpipe/pipe.c b/libpipe/pipe.c
new file mode 100644
index 00000000..73463acd
--- /dev/null
+++ b/libpipe/pipe.c
@@ -0,0 +1,246 @@
+/* Generic one-way pipes
+
+   Copyright (C) 1995 Free Software Foundation, Inc.
+
+   Written by Miles Bader <miles@gnu.ai.mit.edu>
+
+   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 the Free Software Foundation; either version 2, or (at
+   your option) any later version.
+
+   This program is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. */
+
+#include <string.h>		/* For bzero() */
+
+#include <mach/time_value.h>
+#include <mach/mach_host.h>
+
+#include "pipe.h"
+
+static inline void
+timestamp (time_value_t *stamp)
+{
+  host_get_time (mach_host_self (), stamp);
+}
+
+/* ---------------------------------------------------------------- */
+
+/* Creates a new pipe of class CLASS and returns it in RESULT.  */
+error_t
+pipe_create (struct pipe_class *class, struct pipe **pipe)
+{
+  struct pipe *new = malloc (sizeof (struct pipe));
+
+  if (new == NULL)
+    return ENOMEM;
+
+  new->refs = 0;
+  new->flags = 0;
+  new->class = class;
+
+  bzero (&new->read_time, sizeof (new->read_time));
+  bzero (&new->write_time, sizeof (new->write_time));
+
+  condition_init (&new->pending_reads);
+  condition_init (&new->pending_selects);
+  mutex_init (&new->lock);
+
+  new->interrupt_seq_num = 0;
+  pq_create (&new->queue);  
+
+  *pipe = new;
+  return 0;
+}
+
+/* Free PIPE and any resources it holds.  */
+void 
+pipe_free (struct pipe *pipe)
+{
+  pq_free (pipe->queue);
+  free (pipe);
+}
+
+/* Wake up all threads waiting on PIPE, which should be locked.  */
+inline void
+pipe_kick (struct pipe *pipe)
+{
+  /* Now wake them all up for the bad news... */
+  condition_broadcast (&pipe->pending_reads);
+  mutex_unlock (&pipe->lock);
+  condition_broadcast (&pipe->pending_selects);
+  mutex_lock (&pipe->lock);	/* Get back the lock on PIPE.  */
+}
+
+/* Discard a reference to PIPE, which should be unlocked, being sure to make
+   users aware of this.  */
+void
+pipe_break (struct pipe *pipe)
+{
+  mutex_lock (&pipe->lock);
+
+  /* As there may be multiple writers on a connectionless socket, we
+     never allow EOF to be signaled on the reader.  */
+  if (! (pipe->class->flags & PIPE_CLASS_CONNECTIONLESS))
+    pipe->flags |= PIPE_BROKEN;
+
+  if (pipe->refs > 1)
+    /* Other references to PIPE besides ours?  Wake 'em up.  */
+    pipe_kick (pipe);
+
+  pipe_release (pipe);
+}
+
+/* Writes up to LEN bytes of DATA, to PIPE, which should be locked, and
+   returns the amount written in AMOUNT.  If present, the information in
+   CONTROL & PORTS is written in a preceding control packet.  If an error is
+   returned, nothing is done.  */
+error_t
+pipe_send (struct pipe *pipe, void *source,
+	   char *data, size_t data_len,
+	   char *control, size_t control_len,
+	   mach_port_t *ports, size_t num_ports,
+	   size_t *amount)
+{
+  error_t err = 0;
+
+  if (pipe->flags & PIPE_BROKEN)
+    return EPIPE;
+
+  if (control || ports)
+    /* Write a control packet.  */
+    {
+      struct packet *control_packet =
+	pq_queue (pipe->queue, PACKET_TYPE_CONTROL, source);
+      if (control_packet == NULL)
+	err = ENOBUFS;
+      else
+	{
+	  err = packet_write (control_packet, control, control_len, NULL);
+	  if (!err)
+	    err = packet_set_ports (control_packet, ports, num_ports);
+	  if (err)
+	    /* Trash CONTROL_PACKET somehow XXX */;
+	}
+    }
+
+  if (!err)
+    err = (*pipe->class->write)(pipe->queue, source, data, data_len, amount);
+
+  if (!err)
+    {
+      timestamp (&pipe->write_time);
+      
+      /* And wakeup anyone that might be interested in it.  */
+      condition_signal (&pipe->pending_reads);
+      mutex_unlock (&pipe->lock);
+
+      mutex_lock (&pipe->lock);	/* Get back the lock on PIPE.  */
+      /* Only wakeup selects if there's still data available.  */
+      if (pipe_is_readable (pipe, 0))
+	{
+	  condition_signal (&pipe->pending_selects);
+	  /* We leave PIPE locked here, assuming the caller will soon unlock
+	     it and allow others access.  */
+	}
+    }
+
+  return err;
+}
+
+/* Reads up to AMOUNT bytes from PIPE, which should be locked, into DATA, and
+   returns the amount read in DATA_LEN.  If NOBLOCK is true, EWOULDBLOCK is
+   returned instead of block when no data is immediately available.  If an
+   error is returned, nothing is done.  If source isn't NULL, the address of
+   the socket from which the data was sent is returned in it; this may be
+   NULL if it wasn't specified by the sender (which is usually the case with
+   connection-oriented protcols).
+
+   If there is control data waiting (before any data), then the behavior
+   depends on whether this is an `ordinary read' (when CONTROL and PORTS are
+   both NULL), in which case any control data is skipped, or a `msg read', in
+   which case the contents of the first control packet is returned (in
+   CONTROL and PORTS), as well as the first data packet following that (if
+   the control packet is followed by another control packet or no packet in
+   this case, a zero length data buffer is returned; the user should be
+   careful to distinguish this case from EOF (when no control or ports data
+   is returned either).  */
+error_t
+pipe_recv (struct pipe *pipe, int noblock, unsigned *flags, void **source,
+	   char **data, size_t *data_len, size_t amount,
+	   char **control, size_t *control_len,
+	   mach_port_t **ports, size_t *num_ports)
+{
+  error_t err;
+  struct packet *packet;
+  struct pq *pq = pipe->queue;
+  /* True if the user isn't asking for any `control' data.  */
+  int data_only = (control == NULL && ports == NULL);
+
+  err = pipe_wait (pipe, noblock, data_only);
+  if (err)
+    return err;
+
+  packet = pq_head (pq, PACKET_TYPE_ANY, 0);
+
+  if (data_only)
+    /* The user doesn't want to know about control info, so skip any...  */
+    while (packet && packet->type == PACKET_TYPE_CONTROL)
+      packet = pq_next (pq, PACKET_TYPE_ANY, 0);
+  else if (packet && packet->type == PACKET_TYPE_CONTROL)
+    /* Read this control packet first, before looking for a data packet. */
+    {
+      void *control_source;
+
+      if (control != NULL)
+	packet_read (packet, control, control_len, packet_readable (packet));
+      if (ports != NULL)
+	/* Copy out the port rights being sent.  */
+	packet_read_ports (packet, ports, num_ports);
+
+      packet_read_source (packet, &control_source);
+      packet = pq_next (pq, PACKET_TYPE_DATA, control_source);
+      if (!packet && source)
+	/* Since there is no data, say where the control data came from.  */
+	*source = control_source;
+      else if (control_source)
+	/* Otherwise be sure to get rid of our reference to the address. */
+	pipe_dealloc_addr (control_source);
+    }
+  else
+    /* No control data... */
+    {
+      if (control_len)
+	*control_len = 0;
+      if (num_ports)
+	*num_ports = 0;
+    }
+
+  if (!err)
+    if (packet)
+      /* Read some data (PACKET must be a data packet at this point).  */
+      {
+	int dq = 1;	/* True if we should dequeue this packet.  */
+
+ 	if (source)
+	  packet_read_source (packet, source);
+	err = (*pipe->class->read)(packet, &dq, flags, data, data_len, amount);
+	if (dq)
+	  pq_dequeue (pq);
+      }
+    else
+      /* Return EOF.  */
+      *data_len = 0;
+
+  if (!err && packet)
+    timestamp (&pipe->read_time);
+
+  return err;
+}
diff --git a/libpipe/pipe.h b/libpipe/pipe.h
new file mode 100644
index 00000000..022db1f3
--- /dev/null
+++ b/libpipe/pipe.h
@@ -0,0 +1,254 @@
+/* Generic one-way pipes
+
+   Copyright (C) 1995 Free Software Foundation, Inc.
+
+   Written by Miles Bader <miles@gnu.ai.mit.edu>
+
+   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 the Free Software Foundation; either version 2, or (at
+   your option) any later version.
+
+   This program is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. */
+
+#ifndef __PIPE_H__
+#define __PIPE_H__
+
+#include <cthreads.h>		/* For conditions & mutexes */
+
+#include "pq.h"
+
+/* A description of a class of pipes and how to operate on them.  */
+struct pipe_class
+{
+  /* What sort of socket this corresponds too.  */
+  int sock_type;
+
+  /* Flags, from PIPE_CLASS_*, below.  */
+  unsigned flags;
+
+  /* Operations: */
+  /* Read from PACKET into DATA &c, and set *DEQUEUE to true if PACKET should
+     be subsequently discarded.  */
+  error_t (*read)(struct packet *packet, int *dequeue, unsigned *flags,
+		  char **data, size_t *data_len, size_t amount);
+  /* Write DATA &c into the packet queue PQ.  */
+  error_t (*write)(struct pq *pq, void *source,
+		   char *data, size_t data_len, size_t *amount);
+};
+
+/* pipe_class flags  */
+#define PIPE_CLASS_CONNECTIONLESS	0x1 /* A non-stream protocol.  */
+
+/* Some pre-defined pipe_classes.  */
+struct pipe_class *stream_pipe_class;
+struct pipe_class *dgram_pipe_class;
+struct pipe_class *seqpacket_pipe_class;
+
+/* A unidirectional data pipe; it transfers data from READER to WRITER.  */
+struct pipe
+{
+  /* What kind of pipe we are.  */
+  struct pipe_class *class;
+
+  /* We use this to keep track of active threads using this pipe, so that
+     while a thread is waiting to read from a pipe and that pipe gets
+     deallocated (say by socket_shutdown), it doesn't actually go away until
+     the reader realizes what happened.  It is normally frobbed using
+     pipe_aquire & pipe_release, which do locking as well..  */
+  unsigned refs;
+
+  /* Various flags, from PIPE_* below.  */
+  unsigned flags;
+
+  /* Various timestamps for this pipe.  */
+  time_value_t read_time;
+  time_value_t write_time;
+
+  struct condition pending_reads;
+  struct condition pending_selects;
+
+  struct mutex lock;
+
+  /* When a pipe receives an interrupt, we want to wake up all pending read
+     threads, and have them realize they've been interrupted; reads that
+     happen after the interrupt shouldn't return EINTR.  When a thread waits
+     on this pipe's PENDING_READS condition, it remembers this sequence
+     number; any interrupt bumps this number and broadcasts on the condition.
+     A reader thread will try to read from the pipe only if the sequence
+     number is the same as when it went to sleep. */
+  unsigned long interrupt_seq_num;
+
+  /* A queue of incoming packets, of type either PACKET_TYPE_DATA or
+     PACKET_TYPE_CONTROL.  Each data packet represents one datagram for
+     protocols that maintain record boundaries.  Control packets always
+     represent the control information to be returned from one read
+     operation, and will be returned in conjuction with the following data
+     packet (if any).  Reads interested only in data just skip control
+     packets until they find a data packet.  */
+  struct pq *queue;
+};
+
+/* Pipe flags.  */
+#define PIPE_BROKEN	0x1	/* This pipe isn't connected.  */
+
+/* Returns the number of characters quickly readable from PIPE.  If DATA_ONLY
+   is true, then `control' packets are ignored.  */
+extern inline size_t
+pipe_readable (struct pipe *pipe, int data_only)
+{
+  size_t readable = 0;
+  struct pq *pq = pipe->queue;
+  struct packet *packet = pq_head (pq, PACKET_TYPE_ANY, NULL);
+  while (packet)
+    {
+      if (packet->type == PACKET_TYPE_DATA)
+	readable += packet_readable (packet);
+      packet = packet->next;
+    }
+  return readable;
+}
+
+/* Returns true if there's any data available in PIPE.  If DATA_ONLY is true,
+   then `control' packets are ignored.  Note that this is different than
+   (pipe_readable (PIPE) > 0) in the case where a control packet containing
+   only ports is present.  */
+extern inline int
+pipe_is_readable (struct pipe *pipe, int data_only)
+{
+  struct pq *pq = pipe->queue;
+  struct packet *packet = pq_head (pq, PACKET_TYPE_ANY, NULL);
+  if (data_only)
+    while (packet && packet->type == PACKET_TYPE_CONTROL)
+      packet = packet->next;
+  return (packet != NULL);
+}
+
+/* Waits for PIPE to be readable, or an error to occurr.  If NOBLOCK is true,
+   this operation will return EWOULDBLOCK instead of blocking when no data is
+   immediately available.  If DATA_ONLY is true, then `control' packets are
+   ignored.  */
+extern inline error_t
+pipe_wait (struct pipe *pipe, int noblock, int data_only)
+{
+  while (! pipe_is_readable (pipe, data_only) && ! (pipe->flags & PIPE_BROKEN))
+    {
+      unsigned seq_num = pipe->interrupt_seq_num;
+      if (noblock)
+	return EWOULDBLOCK;
+      condition_wait (&pipe->pending_reads, &pipe->lock);
+      if (seq_num != pipe->interrupt_seq_num)
+	return EINTR;
+    }
+  return 0;
+}
+ 
+/* Wake up all threads waiting on PIPE, which should be locked.  */
+void pipe_kick (struct pipe *pipe);
+
+/* Creates a new pipe of class CLASS and returns it in RESULT.  */
+error_t pipe_create (struct pipe_class *class, struct pipe **pipe);
+
+/* Free PIPE and any resources it holds.  */
+void pipe_free (struct pipe *pipe);
+
+/* Discard a reference to PIPE, which should be unlocked, being sure to make
+   users aware of this.  */
+void pipe_break (struct pipe *pipe);
+
+/* Lock PIPE and increment its ref count.  */
+extern inline void
+pipe_aquire (struct pipe *pipe)
+{
+  mutex_lock (&pipe->lock);
+  pipe->refs++;
+}
+
+/* Decrement PIPE's (which should be locked) ref count and unlock it.  If the
+   ref count goes to zero, PIPE will be destroyed.  */
+extern inline void
+pipe_release (struct pipe *pipe)
+{
+  if (--pipe->refs == 0)
+    pipe_free (pipe);
+  else
+    mutex_unlock (&pipe->lock);
+}
+
+/* Empty out PIPE of any data.  PIPE should be locked.  */
+extern inline void
+pipe_drain (struct pipe *pipe)
+{
+  pq_drain (pipe->queue);
+}
+
+/* Writes up to LEN bytes of DATA, to PIPE, which should be locked, and
+   returns the amount written in AMOUNT.  If present, the information in
+   CONTROL & PORTS is written in a preceding control packet.  If an error is
+   returned, nothing is done.  If non-NULL, SOURCE is recorded as the source
+   of the data, to be provided to any readers of it; if no reader ever reads
+   it, it's deallocated by calling pipe_dealloc_addr.  */
+error_t pipe_send (struct pipe *pipe, void *source,
+		   char *data, size_t data_len,
+		   char *control, size_t control_len,
+		   mach_port_t *ports, size_t num_ports,
+		   size_t *amount);
+
+/* Writes up to LEN bytes of DATA, to PIPE, which should be locked, and
+   returns the amount written in AMOUNT.  If an error is returned, nothing is
+   done.  If non-NULL, SOURCE is recorded as the source of the data, to be
+   provided to any readers of it; if no reader ever reads it, it's
+   deallocated by calling pipe_dealloc_addr.  */
+#define pipe_write(pipe, source, data, data_len, amount) \
+  pipe_send (pipe, source, data, data_len, 0, 0, 0, 0, amount)
+
+/* Reads up to AMOUNT bytes from PIPE, which should be locked, into DATA, and
+   returns the amount read in DATA_LEN.  If NOBLOCK is true, EWOULDBLOCK is
+   returned instead of block when no data is immediately available.  If an
+   error is returned, nothing is done.  If source isn't NULL, the
+   corresponding source provided by the sender is returned in it; this may be
+   NULL if it wasn't specified by the sender (which is usually the case with
+   connection-oriented protcols).
+
+   If there is control data waiting (before any data), then the behavior
+   depends on whether this is an `ordinary read' (when CONTROL and PORTS are
+   both NULL), in which case any control data is skipped, or a `msg read', in
+   which case the contents of the first control packet is returned (in
+   CONTROL and PORTS), as well as the first data packet following that (if
+   the control packet is followed by another control packet or no packet in
+   this case, a zero length data buffer is returned; the user should be
+   careful to distinguish this case from EOF (when no control or ports data
+   is returned either).  */
+error_t pipe_recv (struct pipe *pipe, int noblock, unsigned *flags,
+		   void **source,
+		   char **data, size_t *data_len, size_t amount,
+		   char **control, size_t *control_len,
+		   mach_port_t **ports, size_t *num_ports);
+
+
+/* Reads up to AMOUNT bytes from PIPE, which should be locked, into DATA, and
+   returns the amount read in DATA_LEN.  If NOBLOCK is true, EWOULDBLOCK is
+   returned instead of block when no data is immediately available.  If an
+   error is returned, nothing is done.  If source isn't NULL, the
+   corresponding source provided by the sender is returned in it; this may be
+   NULL if it wasn't specified by the sender (which is usually the case with
+   connection-oriented protcols).  */
+#define pipe_read(pipe, noblock, source, data, data_len, amount) \
+  pipe_recv (pipe, noblock, 0, source, data, data_len, amount, 0,0,0,0)
+
+/* ---------------------------------------------------------------- */
+/* User-provided functions.  */
+
+/* This routine may be provided by the user, in which case, it should be a
+   function taking a non-NULL source address and deallocating it.  It
+   defaults to calling ports_port_deref.  */
+void pipe_dealloc_addr (void *addr);
+
+#endif /* __PIPE_H__ */
diff --git a/libpipe/pq.c b/libpipe/pq.c
new file mode 100644
index 00000000..bd611155
--- /dev/null
+++ b/libpipe/pq.c
@@ -0,0 +1,394 @@
+/* Packet queues
+
+   Copyright (C) 1995 Free Software Foundation, Inc.
+
+   Written by Miles Bader <miles@gnu.ai.mit.edu>
+
+   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 the Free Software Foundation; either version 2, or (at
+   your option) any later version.
+
+   This program is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. */
+
+#include <malloc.h>
+#include <string.h>		/* for bcopy */
+#include <stddef.h>
+#include <assert.h>
+
+#include "pq.h"
+
+/* ---------------------------------------------------------------- */
+
+/* Create a new packet queue, returning it in PQ.  The only possible error is
+   ENOMEM.  */
+error_t
+pq_create (struct pq **pq)
+{
+  *pq = malloc (sizeof (struct pq));
+
+  if (! *pq)
+    return ENOMEM;
+
+  (*pq)->head = (*pq)->tail = 0;
+  (*pq)->free = 0;
+
+  return 0;
+}
+
+/* Free every packet (and its contents) in the linked list rooted at HEAD.  */
+static void
+free_packets (struct packet *head)
+{
+  if (head)
+    {
+      struct packet *next = head->next;
+      if (head->ports)
+	free (head->ports);
+      if (head->buf_len > 0)
+	if (head->buf_vm_alloced)
+	  vm_deallocate (mach_task_self (),
+			 (vm_address_t)head->buf, head->buf_len);
+	else
+	  free (head->buf);
+      free (head);
+      free_packets (next);
+    }
+}
+
+/* Frees PQ and any resources it holds, including deallocating any ports in
+   packets left in the queue.  */
+void
+pq_free (struct pq *pq)
+{
+  pq_drain (pq);
+  free_packets (pq->free);
+  free (pq);
+}
+
+/* ---------------------------------------------------------------- */
+
+/* Remove the first packet (if any) in PQ, deallocating any resources it
+   holds.  True is returned if a packet was found, false otherwise.  */
+int
+pq_dequeue (struct pq *pq)
+{
+  extern void pipe_dealloc_addr (void *addr);
+  struct packet *packet = pq->head;
+
+  if (! packet)
+    return 0;
+
+  /* Deallocate any resource in PACKET.  */
+  if (packet->num_ports)
+    packet_dealloc_ports (packet);
+  if (packet->source)
+    pipe_dealloc_addr (packet->source);
+
+  pq->head = packet->next;
+  packet->next = pq->free;
+  pq->free = packet;
+  if (pq->head)
+    pq->head->prev = 0;
+  else
+    pq->tail = 0;
+
+  return 1;
+}
+
+/* Empties out PQ.  This *will* deallocate any ports in any of the packets.  */
+void
+pq_drain (struct pq *pq)
+{
+  while (pq_dequeue (pq))
+    ;
+}
+
+/* Pushes a new packet of type TYPE and source SOURCE onto the tail of the
+   queue, and returns it, or 0 if there was an allocation error. */
+struct packet *
+pq_queue (struct pq *pq, unsigned type, void *source)
+{
+  struct packet *packet = pq->free;
+
+  if (!packet)
+    {
+      packet = malloc (sizeof (struct packet));
+      if (!packet)
+	return 0;
+      packet->buf = 0;
+      packet->buf_len = 0;
+      packet->ports = 0;
+      packet->num_ports = 0;
+      packet->buf_start = packet->buf_end = packet->buf;
+    }
+  else
+    pq->free = packet->next;
+
+  packet->type = type;
+  packet->source = source;
+  packet->next = 0;
+  packet->prev = pq->tail;
+  if (pq->tail)
+    pq->tail->next = packet;
+  pq->tail = packet;
+  if (!pq->head)
+    pq->head = packet;
+
+  return packet;
+}
+
+/* ---------------------------------------------------------------- */
+
+/* Returns a size to which a packet can be set, which will be at least GOAL,
+   but perhaps more.  */
+size_t
+packet_size_adjust (size_t goal)
+{
+  if (goal > PACKET_SIZE_LARGE)
+    /* Round GOAL up to a page boundary (OLD_LEN should already be).  */
+    return round_page (goal);
+  else
+    /* Otherwise, just round up to a multiple of 512 bytes.  */
+    return (goal + 511) & ~511;
+}
+
+/* Try to extend PACKET to be NEW_LEN bytes long, which should be greater
+   than the current packet size.  This should be a valid length -- i.e., if
+   it's greater than PACKET_SIZE_LARGE, it should be a mulitple of
+   VM_PAGE_SIZE.  If PACKET cannot be extended for some reason, false is
+   returned, otherwise true.  */
+int
+packet_extend (struct packet *packet, size_t new_len)
+{
+  size_t old_len = packet->buf_len;
+
+  if (old_len == 0)
+    /* No existing buffer to extend.  */
+    return 0;
+
+  if (packet->buf_vm_alloced)
+    /* A vm_alloc'd packet.  */
+    {
+      char *extension = packet->buf + old_len;
+
+      /* Try to allocate memory at the end of our current buffer.  */
+      if (vm_allocate (mach_task_self (),
+		       (vm_address_t *)&extension, new_len - old_len, 0) != 0)
+	return 0;
+    }
+  else
+    /* A malloc'd packet.  */
+    {
+      char *new_buf;
+      char *old_buf = packet->buf;
+
+      if (new_len >= PACKET_SIZE_LARGE)
+	/* The old packet length is malloc'd, but we want to vm_allocate the
+	   new length, so we'd have to copy the old contents.  */
+	return 0;
+
+      new_buf = realloc (old_buf, new_len);
+      if (! new_buf)
+	return 0;
+
+      packet->buf = new_buf;
+      packet->buf_start = new_buf + (packet->buf_start  - old_buf);
+      packet->buf_end = new_buf + (packet->buf_end  - old_buf);
+    }
+  
+  packet->buf_len = new_len;
+
+  return 1;
+}
+
+/* Reallocate PACKET to have NEW_LEN bytes of buffer space, which should be
+   greater than the current packet size.  This should be a valid length --
+   i.e., if it's greater than PACKET_SIZE_LARGE, it should be a multiple of
+   VM_PAGE_SIZE.  If an error occurs, PACKET is not modified and the error is
+   returned.  */
+error_t
+packet_realloc (struct packet *packet, size_t new_len)
+{
+  error_t err;
+  char *new_buf;
+  char *old_buf = packet->buf;
+  int vm_alloc = (new_len >= PACKET_SIZE_LARGE);
+
+  /* Make a new buffer.  */
+  if (vm_alloc)
+    err =
+      vm_allocate (mach_task_self (), (vm_address_t *)&new_buf, new_len, 1);
+  else
+    {
+      new_buf = malloc (new_len);
+      err = (new_buf ? 0 : ENOMEM);
+    }
+
+  if (! err)
+    {
+      size_t old_len = packet->buf_len;
+      char *start = packet->buf_start, *end = packet->buf_end;
+
+      /* Copy what we must.  */
+      if (end != start)
+	/* If there was an operation like vm_move, we could use that in the
+	   case where both the old and the new buffers were vm_alloced (on
+	   the assumption that creating COW pages is somewhat more costly).
+	   But there's not, and bcopy will do vm_copy where it can.  Will we
+	   still takes faults on the new copy, even though we've deallocated
+	   the old one???  XXX */
+	bcopy (start, new_buf, end - start);
+
+      /* And get rid of the old buffer.  */
+      if (old_len > 0)
+	if (packet->buf_vm_alloced)
+	  vm_deallocate (mach_task_self (), (vm_address_t)old_buf, old_len);
+	else
+	  free (old_buf);
+
+      packet->buf = new_buf;
+      packet->buf_len = new_len;
+      packet->buf_vm_alloced = vm_alloc;
+      packet->buf_start = new_buf + (start - old_buf);
+      packet->buf_end = new_buf + (end - old_buf);
+    }
+
+  return err;
+}
+
+/* ---------------------------------------------------------------- */
+
+/* If PACKET has any ports, deallocates them.  */
+void 
+packet_dealloc_ports (struct packet *packet)
+{
+  unsigned i;
+  for (i = 0; i < packet->num_ports; i++)
+    {
+      mach_port_t port = packet->ports[i];
+      if (port != MACH_PORT_NULL)
+	mach_port_deallocate (mach_task_self (), port);
+    }
+}
+
+/* Sets PACKET's ports to be PORTS, of length NUM_PORTS.  ENOMEM is returned
+   if a memory allocation error occurred, otherwise, 0.  */
+error_t
+packet_set_ports (struct packet *packet,
+		  mach_port_t *ports, size_t num_ports)
+{
+  if (packet->num_ports > 0)
+    packet_dealloc_ports (packet);
+  if (num_ports > packet->ports_alloced)
+    {
+      mach_port_t *new_ports = malloc (sizeof (mach_port_t *) * num_ports);
+      if (! new_ports)
+	return ENOMEM;
+      free (packet->ports);
+      packet->ports_alloced = num_ports;
+    }
+  bcopy (ports, packet->ports, sizeof (mach_port_t *) * num_ports);
+  packet->num_ports = num_ports;
+  return 0;
+}
+
+/* Returns any ports in PACKET in PORTS and NUM_PORTS, and removes them from
+   PACKET.  */
+error_t
+packet_read_ports (struct packet *packet,
+		   mach_port_t **ports, size_t *num_ports)
+{
+  int length = packet->num_ports * sizeof (mach_port_t *);
+  if (*num_ports < packet->num_ports)
+    {
+      error_t err =
+	vm_allocate (mach_task_self (), (vm_address_t *)ports, length, 1);
+      if (err)
+	return err;
+    }
+  *num_ports = packet->num_ports;
+  bcopy (packet->ports, *ports, length);
+  packet->num_ports = 0;
+  return 0;
+}
+
+/* Append the bytes in DATA, of length DATA_LEN, to what's already in PACKET,
+   and return the amount appended in AMOUNT.  */
+error_t 
+packet_write (struct packet *packet,
+	      char *data, size_t data_len, size_t *amount)
+{
+  error_t err = packet_ensure (packet, data_len);
+
+  if (err)
+    return err;
+
+  /* Add the new data.  */
+  bcopy (data, packet->buf_end, data_len);
+  packet->buf_end += data_len;
+  *amount = data_len;
+
+  return 0;
+}
+
+/* Removes up to AMOUNT bytes from the beginning of the data in PACKET, and
+   puts it into *DATA, and the amount read into DATA_LEN.  If more than the
+   original *DATA_LEN bytes are available, new memory is vm_allocated, and
+   the address and length of this array put into DATA and DATA_LEN.  */
+error_t
+packet_read (struct packet *packet,
+	     char **data, size_t *data_len, size_t amount)
+{
+  char *start = packet->buf_start;
+
+  if (amount > packet->buf_end - start)
+    amount = packet->buf_end - start;
+
+  if (amount > 0)
+    {
+      if (packet->buf_vm_alloced && amount > vm_page_size)
+	/* We can return memory from BUF directly without copying.  */
+	{
+	  *data = start;
+	  if (start + amount < packet->buf_end)
+	    /* Since returning a partial page actually means returning the
+	       whole page, we have to be careful not to grab past the page
+	       boundary before the end of the data we want, unless the rest
+	       of the page is unimportant.  */
+	    {
+	      amount = (char *)trunc_page (start + amount) - start;
+	      packet->buf_start = (char *)round_page (start + amount);
+	    }
+	  else
+	    /* As we're at the end of the buffer, we don't care about
+	       BUF_START remaining page-aligned (the buffer size will be
+	       zero anyway), and this way we ensure that BUF_START doesn't
+	       go past BUF_END (which causes all sorts of fun).  */
+	    packet->buf_start = start + amount;
+
+	  /* We've actually consumed the memory at the start of BUF, so
+	     adjust it and BUF_LEN to reflect this.  */
+	  packet->buf_len -= (packet->buf_start - packet->buf);
+	  packet->buf = packet->buf_start;
+	}
+      else
+	/* Just copy the data the old fashioned way....  */
+	{
+	  if (*data_len < amount)
+	    vm_allocate (mach_task_self (), (vm_address_t *)data, amount, 1);
+	  bcopy (start, *data, amount);
+	  packet->buf_start = start + amount;
+	}
+    }
+  *data_len = amount;
+
+  return 0;
+}
diff --git a/libpipe/pq.h b/libpipe/pq.h
new file mode 100644
index 00000000..f7a99c64
--- /dev/null
+++ b/libpipe/pq.h
@@ -0,0 +1,230 @@
+/* Packet queues
+
+   Copyright (C) 1995 Free Software Foundation, Inc.
+
+   Written by Miles Bader <miles@gnu.ai.mit.edu>
+
+   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 the Free Software Foundation; either version 2, or (at
+   your option) any later version.
+
+   This program is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. */
+
+#ifndef __PQ_H__
+#define __PQ_H__
+
+#include <errno.h>
+#include <stddef.h>		/* for size_t */
+#include <mach/mach.h>
+
+struct packet
+{
+  /* The packet type, from PACKET_* below.  */
+  unsigned short type;
+
+  /* Where this packet was sent from.  */
+  void *source;
+
+  /* Buffer space. */
+  char *buf;
+  size_t buf_len;
+  /* Pointers to the data within BUF.  */
+  char *buf_start, *buf_end;
+  /* True if BUF was allocated using vm_allocate rather than malloc; only
+     valid if BUF_LEN > 0.  */
+  int buf_vm_alloced;
+
+  /* Port data */
+  mach_port_t *ports;
+  size_t num_ports, ports_alloced;
+
+  /* Next and previous packets within the packet queue we're part of.  If
+     PREV is null, we're at the head of the queue, and if NEXT is null, we're
+     at the tail.  */
+  struct packet *next, *prev;
+};
+
+#define PACKET_TYPE_ANY		0 /* matches any type of packet */
+#define PACKET_TYPE_DATA	1
+#define PACKET_TYPE_CONTROL	2
+
+/* Sets PACKET's ports to be PORTS, of length NUM_PORTS.  ENOMEM is returned
+   if a memory allocation error occurred, otherwise, 0.  */
+error_t packet_set_ports (struct packet *packet,
+			  mach_port_t *ports, size_t num_ports);
+
+/* If PACKET has any ports, deallocates them.  */
+void packet_dealloc_ports (struct packet *packet);
+
+/* Returns the number of bytes of data in PACKET.  */
+extern inline size_t
+packet_readable (struct packet *packet)
+{
+  return packet->buf_end - packet->buf_start;
+}
+
+/* Append the bytes in DATA, of length DATA_LEN, to what's already in PACKET,
+   and return the amount appended in AMOUNT.  */
+error_t packet_write (struct packet *packet,
+		      char *data, size_t data_len, size_t *amount);
+
+/* Removes up to AMOUNT bytes from the beginning of the data in PACKET, and
+   puts it into *DATA, and the amount read into DATA_LEN.  If more than the
+   original *DATA_LEN bytes are available, new memory is vm_allocated, and
+   the address and length of this array put into DATA and DATA_LEN.  */
+error_t packet_read (struct packet *packet,
+		     char **data, size_t *data_len, size_t amount);
+
+/* Returns any ports in PACKET in PORTS and NUM_PORTS, and removes them from
+   PACKET.  */
+error_t packet_read_ports (struct packet *packet,
+			   mach_port_t **ports, size_t *num_ports);
+
+/* Return the source addressd in PACKET in SOURCE, deallocating it from
+   PACKET.  */
+extern inline void
+packet_read_source (struct packet *packet, void **source)
+{
+  *source = packet->source;
+  packet->source = 0;
+}
+
+/* The packet size above which we start to do things differently to avoid
+   copying around data.  */
+#define PACKET_SIZE_LARGE	8192
+
+/* Returns a size to which a packet can be set, which will be at least GOAL,
+   but perhaps more.  */
+size_t packet_size_adjust (size_t goal);
+
+/* Try to extend PACKET to be NEW_LEN bytes long, which should be greater
+   than the current packet size.  This should be a valid length -- i.e., if
+   it's greater than PAGE_PACKET_SIZE, it should be a mulitple of
+   VM_PAGE_SIZE.  If PACKET cannot be extended for some reason, false is
+   returned, otherwise true.  */
+int packet_extend (struct packet *packet, size_t new_len);
+
+/* Reallocate PACKET to have NEW_LEN bytes of buffer space, which should be
+   greater than the current packet size.  This should be a valid length --
+   i.e., if it's greater than PAGE_PACKET_SIZE, it should be a multiple of
+   VM_PAGE_SIZE.  If an error occurs, PACKET is not modified and the error is
+   returned.  */
+error_t packet_realloc (struct packet *packet, size_t new_len);
+
+/* Make sure that PACKET has room for at least AMOUNT more bytes, or return
+   the reason why not.  */
+extern inline error_t
+packet_ensure (struct packet *packet, size_t amount)
+{
+  size_t new_len;
+  size_t old_len = packet->buf_len;
+  size_t left = packet->buf + old_len - packet->buf_end;
+
+  if (amount < left)
+    return 0;
+
+  new_len = packet_size_adjust (old_len + amount - left);
+  if (packet_extend (packet, new_len))
+    return 0;
+  else
+    return packet_realloc (packet, new_len);
+}
+
+/* Make sure that PACKET has room for at least AMOUNT more bytes, *only* if
+   it can be done efficiently, e.g., the packet can be grown in place, rather
+   than moving the contents (or there is little enough data so that copying
+   it is OK).  True is returned if room was made, false otherwise.  */
+extern inline int
+packet_ensure_efficiently (struct packet *packet, size_t amount)
+{
+  size_t new_len;
+  size_t old_len = packet->buf_len;
+  size_t left = packet->buf + old_len - packet->buf_end;
+
+  if (amount < left)
+    return 1;
+
+  new_len = packet_size_adjust (old_len + amount - left);
+  if (packet_extend (packet, new_len))
+    return 1;
+  if ((packet->buf_end - packet->buf_start) < PACKET_SIZE_LARGE)
+    return packet_realloc (packet, new_len) == 0;
+  return 0;
+}
+
+struct pq
+{
+  struct packet *head, *tail;	/* Packet queue */
+  struct packet *free;		/* Free packets */
+};
+
+/* Pushes a new packet of type TYPE and source SOURCE, and returns it, or
+   NULL if there was an allocation error.  SOURCE is returned to readers of
+   the packet, or deallocated by calling pipe_dealloc_addr.  */
+struct packet *pq_queue (struct pq *pq, unsigned type, void *source);
+
+/* Returns the tail of the packet queue PQ, which may mean pushing a new
+   packet if TYPE and SOURCE do not match the current tail, or this is the
+   first packet.  */
+extern inline struct packet *
+pq_tail (struct pq *pq, unsigned type, void *source)
+{
+  struct packet *tail = pq->tail;
+  if (!tail
+      || (type && tail->type != type) || (source && tail->source != source))
+    tail = pq_queue (pq, type, source);
+  return tail;
+}
+
+/* Remove the first packet (if any) in PQ, deallocating any resources it
+   holds.  True is returned if a packet was found, false otherwise.  */
+int pq_dequeue (struct pq *pq);
+
+/* Returns the next available packet in PQ, without removing it from the
+   queue, or NULL if there is none, or the next packet isn't appropiate.  
+   A packet is inappropiate if SOURCE is non-NULL its source field doesn't
+   match it, or TYPE is non-NULL and the packet's type field doesn't match
+   it.  */
+extern inline struct packet *
+pq_head (struct pq *pq, unsigned type, void *source)
+{
+  struct packet *head = pq->head;
+  if (!head)
+    return 0;
+  if (type && head->type != type)
+    return 0;
+  if (source && head->source != source)
+    return 0;
+  return head;
+}
+
+/* The same as pq_head, but first discards the head of the queue.  */
+extern inline struct packet *
+pq_next (struct pq *pq, unsigned type, void *source)
+{
+  if (!pq->head)
+    return 0;
+  pq_dequeue (pq);
+  return pq_head (pq, type, source);
+}
+
+/* Dequeues all packets in PQ.  */
+void pq_drain (struct pq *pq);
+
+/* Create a new packet queue, returning it in PQ.  The only possible error is
+   ENOMEM.  */
+error_t pq_create (struct pq **pq);
+
+/* Frees PQ and any resources it holds, including deallocating any ports in
+   packets left in the queue.  */
+void pq_free (struct pq *pq);
+
+#endif /* __PQ_H__ */
diff --git a/libpipe/seqpack.c b/libpipe/seqpack.c
new file mode 100644
index 00000000..44a15a03
--- /dev/null
+++ b/libpipe/seqpack.c
@@ -0,0 +1,55 @@
+/* The SOCK_SEQPACKET pipe class
+
+   Copyright (C) 1995 Free Software Foundation, Inc.
+
+   Written by Miles Bader <miles@gnu.ai.mit.edu>
+
+   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 the Free Software Foundation; either version 2, or (at
+   your option) any later version.
+
+   This program is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. */
+
+#include <sys/socket.h>		/* For SOCK_SEQPACKET */
+
+#include "pipe.h"
+#include "pq.h"
+
+/* See the definition of struct pipe_class in "pipe.h" for documentation.  */
+
+/* This type of pipe is the same as a SOCK_STREAM, but maintains record
+   boundaries.  */
+
+static error_t 
+seqpack_write (struct pq *pq, void *source,
+	       char *data, size_t data_len, size_t *amount)
+{
+  struct packet *packet = pq_queue (pq, PACKET_TYPE_DATA, source);
+  if (!packet)
+    return ENOBUFS;
+  else
+    return packet_write (packet, data, data_len, amount);
+}
+
+static error_t 
+seqpack_read (struct packet *packet, int *dequeue, unsigned *flags,
+	      char **data, size_t *data_len, size_t amount)
+{
+  error_t err = packet_read (packet, data, data_len, amount);
+  *dequeue = (packet_readable (packet) == 0);
+  return err;
+}
+
+struct pipe_class _seqpack_pipe_class =
+{
+  SOCK_SEQPACKET, 0, seqpack_read, seqpack_write
+};
+struct pipe_class *seqpack_pipe_class = &_seqpack_pipe_class;
diff --git a/libpipe/stream.c b/libpipe/stream.c
new file mode 100644
index 00000000..8eb90435
--- /dev/null
+++ b/libpipe/stream.c
@@ -0,0 +1,68 @@
+/* The SOCK_STREAM pipe class
+
+   Copyright (C) 1995 Free Software Foundation, Inc.
+
+   Written by Miles Bader <miles@gnu.ai.mit.edu>
+
+   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 the Free Software Foundation; either version 2, or (at
+   your option) any later version.
+
+   This program is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. */
+
+#include <sys/socket.h>		/* For SOCK_STREAM */
+
+#include "pipe.h"
+#include "pq.h"
+
+/* See the definition of struct pipe_class in "pipe.h" for an explanation.  */
+
+/* This should be in some system header...  XXX  */
+static inline int page_aligned (vm_offset_t num)
+{
+  return trunc_page (num) == num;
+}
+
+static error_t 
+stream_write (struct pq *pq, void *source,
+	      char *data, size_t data_len, size_t *amount)
+{
+  struct packet *packet = pq_tail (pq, PACKET_TYPE_DATA, source);
+
+  if (packet_readable (packet) > 0
+      && data_len > PACKET_SIZE_LARGE
+      && (! page_aligned (data - packet->buf_end)
+	  || ! packet_ensure_efficiently (packet, data_len)))
+    /* Put a large page-aligned transfer in its own packet, if it's
+       page-aligned `differently' than the end of the current packet, or if
+       the current packet can't be extended in place.  */
+    packet = pq_queue (pq, PACKET_TYPE_DATA, source);
+
+  if (!packet)
+    return ENOBUFS;
+  else
+    return packet_write (packet, data, data_len, amount);
+}
+
+static error_t 
+stream_read (struct packet *packet, int *dequeue, unsigned *flags,
+	     char **data, size_t *data_len, size_t amount)
+{
+  error_t err = packet_read (packet, data, data_len, amount);
+  *dequeue = (packet_readable (packet) == 0);
+  return err;
+}
+
+struct pipe_class _stream_pipe_class =
+{
+  SOCK_STREAM, 0, stream_read, stream_write
+};
+struct pipe_class *stream_pipe_class = &_stream_pipe_class;