/* Generic one-way pipes Copyright (C) 1995 Free Software Foundation, Inc. Written by Miles Bader 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 /* For bzero() */ #include #include #include #include #include "pipe.h" static inline void timestamp (time_value_t *stamp) { host_get_time (mach_host_self (), stamp); } /* Hold this lock before attempting to lock multiple pipes. */ struct mutex pipe_multiple_lock = MUTEX_INITIALIZER; /* ---------------------------------------------------------------- */ #define pipe_is_connless(p) ((p)->class->flags & PIPE_CLASS_CONNECTIONLESS) /* 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->readers = 0; new->writers = 0; new->flags = 0; new->class = class; new->write_limit = 16*1024; new->write_atomic = 16*1024; 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_read_selects); condition_init (&new->pending_writes); condition_init (&new->pending_write_selects); mutex_init (&new->lock); pq_create (&new->queue); if (! pipe_is_connless (new)) new->flags |= PIPE_BROKEN; *pipe = new; return 0; } /* Free PIPE and any resources it holds. */ void pipe_free (struct pipe *pipe) { pq_free (pipe->queue); free (pipe); } /* Take any actions necessary when PIPE acquires its first writer. */ void _pipe_first_writer (struct pipe *pipe) { if (pipe->readers > 0) pipe->flags &= ~PIPE_BROKEN; } /* Take any actions necessary when PIPE acquires its first reader. */ void _pipe_first_reader (struct pipe *pipe) { if (pipe->writers > 0) pipe->flags &= ~PIPE_BROKEN; } /* Take any actions necessary when PIPE's last reader has gone away. PIPE should be locked. */ void _pipe_no_readers (struct pipe *pipe) { if (pipe->writers == 0) pipe_free (pipe); else { if (! pipe_is_connless (pipe)) { pipe->flags |= PIPE_BROKEN; if (pipe->readers) /* Wake up writers for the bad news... */ condition_broadcast (&pipe->pending_writes); } mutex_unlock (&pipe->lock); } } /* Take any actions necessary when PIPE's last writer has gone away. PIPE should be locked. */ void _pipe_no_writers (struct pipe *pipe) { if (pipe->readers == 0) pipe_free (pipe); else { if (! pipe_is_connless (pipe)) { pipe->flags |= PIPE_BROKEN; if (pipe->readers) /* Wake up readers for the bad news... */ { condition_broadcast (&pipe->pending_reads); condition_broadcast (&pipe->pending_read_selects); } } mutex_unlock (&pipe->lock); } } /* Return when either RPIPE is available for reading (if SELECT_READ is set in *SELECT_TYPE), or WPIPE is available for writing (if select_write is set in *SELECT_TYPE). *SELECT_TYPE is modified to reflect which (or both) is now available. DATA_ONLY should be true if only data packets should be waited for on RPIPE. Neither RPIPE or WPIPE should be locked when calling this function (unlike most pipe functions). */ error_t pipe_pair_select (struct pipe *rpipe, struct pipe *wpipe, int *select_type, int data_only) { error_t err = 0; *select_type &= SELECT_READ | SELECT_WRITE; if (*select_type == SELECT_READ) { mutex_lock (&rpipe->lock); err = pipe_select_readable (rpipe, data_only); mutex_unlock (&rpipe->lock); } else if (*select_type == SELECT_WRITE) { mutex_lock (&wpipe->lock); err = pipe_select_writable (wpipe); mutex_unlock (&wpipe->lock); } else /* ugh */ { int rpipe_blocked, wpipe_blocked; struct condition pending_read_write_select; size_t wlimit = wpipe->write_limit; struct mutex *lock = (wpipe == rpipe ? &rpipe->lock : &pipe_multiple_lock); condition_init (&pending_read_write_select); condition_implies (&rpipe->pending_read_selects, &pending_read_write_select); condition_implies (&wpipe->pending_write_selects, &pending_read_write_select); mutex_lock (lock); if (rpipe != wpipe) { mutex_lock (&rpipe->lock); mutex_lock (&wpipe->lock); } rpipe_blocked = ! ((rpipe->flags & PIPE_BROKEN) || pipe_is_readable (rpipe, data_only)); wpipe_blocked = ! ((wpipe->flags & PIPE_BROKEN) || pipe_readable (wpipe, 1) < wlimit); while (!err && rpipe_blocked && wpipe_blocked) { if (rpipe != wpipe) { mutex_unlock (&rpipe->lock); mutex_unlock (&wpipe->lock); } if (hurd_condition_wait (&pending_read_write_select, lock)) err = EINTR; if (rpipe != wpipe) { mutex_lock (&rpipe->lock); mutex_lock (&wpipe->lock); } rpipe_blocked = ! ((rpipe->flags & PIPE_BROKEN) || pipe_is_readable (rpipe, data_only)); wpipe_blocked = ! ((wpipe->flags & PIPE_BROKEN) || pipe_readable (wpipe, 1) < wlimit); } if (!err) { if (rpipe_blocked) *select_type &= ~SELECT_READ; if (wpipe_blocked) *select_type &= ~SELECT_WRITE; } if (rpipe != wpipe) { mutex_unlock (&rpipe->lock); mutex_unlock (&wpipe->lock); } mutex_unlock (lock); condition_unimplies (&rpipe->pending_read_selects, &pending_read_write_select); condition_unimplies (&wpipe->pending_write_selects, &pending_read_write_select); } return err; } /* 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, int noblock, 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; err = pipe_wait_writable (pipe, noblock); if (err) return err; if (noblock) { size_t left = pipe->write_limit - pipe_readable (pipe, 1); if (left > data_len) if (data_len <= pipe->write_atomic) return EWOULDBLOCK; else data_len = left; } if (control_len > 0 || num_ports > 0) /* Write a control packet. */ { /* Note that we don't record the source address in control packets, as it's recorded in the following data packet anyway, and this prevents it from being dealloc'd twice; this depends on the fact that we always write a data packet. */ struct packet *control_packet = pq_queue (pipe->queue, PACKET_TYPE_CONTROL, NULL); 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_broadcast (&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_broadcast (&pipe->pending_read_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_readable (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. */ { 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 = pq_next (pq, PACKET_TYPE_DATA, NULL); assert (packet); /* pipe_write always writes a data packet. */ } 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); /* And wakeup anyone that might be interested in it. */ condition_broadcast (&pipe->pending_writes); mutex_unlock (&pipe->lock); mutex_lock (&pipe->lock); /* Get back the lock on PIPE. */ /* Only wakeup selects if there's still writing space available. */ if (pipe_readable (pipe, 1) < pipe->write_limit) { condition_broadcast (&pipe->pending_write_selects); /* We leave PIPE locked here, assuming the caller will soon unlock it and allow others access. */ } } return err; }