source: rtems/cpukit/libfs/src/pipe/fifo.c @ 8851c0a

4.11
Last change on this file since 8851c0a was 8851c0a, checked in by Joel Sherrill <joel.sherrill@…>, on Aug 2, 2010 at 6:27:23 PM

2010-08-02 Joel Sherrill <joel.sherrill@…>

  • libfs/src/imfs/imfs_creat.c, libfs/src/imfs/imfs_eval.c, libfs/src/imfs/imfs_mknod.c, libfs/src/imfs/imfs_readlink.c, libfs/src/pipe/fifo.c: Clean up for coverage improvements and formatting.
  • Property mode set to 100644
File size: 12.6 KB
Line 
1/*
2 * fifo.c: POSIX FIFO/pipe for RTEMS
3 *
4 * Author: Wei Shen <cquark@gmail.com>
5 *
6 * The license and distribution terms for this file may be
7 * found in the file LICENSE in this distribution or at
8 * http://www.rtems.com/license/LICENSE.
9 *
10 * $Id$
11 */
12
13
14#if HAVE_CONFIG_H
15#include "config.h"
16#endif
17
18#ifdef RTEMS_POSIX_API
19#define __RTEMS_VIOLATE_KERNEL_VISIBILITY__
20#endif
21
22#include <errno.h>
23#include <stdlib.h>
24
25#include <rtems.h>
26#include <rtems/libio_.h>
27
28#include "pipe.h"
29
30
31#define MIN(a, b) ((a) < (b)? (a): (b))
32
33#define LIBIO_ACCMODE(_iop) ((_iop)->flags & LIBIO_FLAGS_READ_WRITE)
34#define LIBIO_NODELAY(_iop) ((_iop)->flags & LIBIO_FLAGS_NO_DELAY)
35
36static rtems_id pipe_semaphore = RTEMS_ID_NONE;
37
38
39#define PIPE_EMPTY(_pipe) (_pipe->Length == 0)
40#define PIPE_FULL(_pipe)  (_pipe->Length == _pipe->Size)
41#define PIPE_SPACE(_pipe) (_pipe->Size - _pipe->Length)
42#define PIPE_WSTART(_pipe) ((_pipe->Start + _pipe->Length) % _pipe->Size)
43
44#define PIPE_LOCK(_pipe)  \
45  ( rtems_semaphore_obtain(_pipe->Semaphore, RTEMS_WAIT, RTEMS_NO_TIMEOUT)  \
46   == RTEMS_SUCCESSFUL )
47
48#define PIPE_UNLOCK(_pipe)  rtems_semaphore_release(_pipe->Semaphore)
49
50#define PIPE_READWAIT(_pipe)  \
51  ( rtems_barrier_wait(_pipe->readBarrier, RTEMS_NO_TIMEOUT)  \
52   == RTEMS_SUCCESSFUL)
53
54#define PIPE_WRITEWAIT(_pipe)  \
55  ( rtems_barrier_wait(_pipe->writeBarrier, RTEMS_NO_TIMEOUT)  \
56   == RTEMS_SUCCESSFUL)
57
58#define PIPE_WAKEUPREADERS(_pipe) \
59  do {uint32_t n; rtems_barrier_release(_pipe->readBarrier, &n); } while(0)
60
61#define PIPE_WAKEUPWRITERS(_pipe) \
62  do {uint32_t n; rtems_barrier_release(_pipe->writeBarrier, &n); } while(0)
63
64
65#ifdef RTEMS_POSIX_API
66#define __RTEMS_VIOLATE_KERNEL_VISIBILITY__
67
68#include <rtems/rtems/barrier.h>
69#include <rtems/score/thread.h>
70
71/* Set barriers to be interruptible by signals. */
72static void pipe_interruptible(pipe_control_t *pipe)
73{
74  Objects_Locations location;
75
76  _Barrier_Get(pipe->readBarrier, &location)->Barrier.Wait_queue.state
77    |= STATES_INTERRUPTIBLE_BY_SIGNAL;
78  _Thread_Enable_dispatch();
79  _Barrier_Get(pipe->writeBarrier, &location)->Barrier.Wait_queue.state
80    |= STATES_INTERRUPTIBLE_BY_SIGNAL;
81  _Thread_Enable_dispatch();
82}
83#endif
84
85/*
86 * Alloc pipe control structure, buffer, and resources.
87 * Called with pipe_semaphore held.
88 */
89static int pipe_alloc(
90  pipe_control_t **pipep
91)
92{
93  static char c = 'a';
94  pipe_control_t *pipe;
95  int err = -ENOMEM;
96
97  pipe = malloc(sizeof(pipe_control_t));
98  if (pipe == NULL)
99    return err;
100  memset(pipe, 0, sizeof(pipe_control_t));
101
102  pipe->Size = PIPE_BUF;
103  pipe->Buffer = malloc(pipe->Size);
104  if (! pipe->Buffer)
105    goto err_buf;
106
107  err = -ENOMEM;
108
109  if (rtems_barrier_create(
110        rtems_build_name ('P', 'I', 'r', c),
111        RTEMS_BARRIER_MANUAL_RELEASE, 0,
112        &pipe->readBarrier) != RTEMS_SUCCESSFUL)
113    goto err_rbar;
114  if (rtems_barrier_create(
115        rtems_build_name ('P', 'I', 'w', c),
116        RTEMS_BARRIER_MANUAL_RELEASE, 0,
117        &pipe->writeBarrier) != RTEMS_SUCCESSFUL)
118    goto err_wbar;
119  if (rtems_semaphore_create(
120        rtems_build_name ('P', 'I', 's', c), 1,
121        RTEMS_BINARY_SEMAPHORE | RTEMS_FIFO,
122        RTEMS_NO_PRIORITY, &pipe->Semaphore) != RTEMS_SUCCESSFUL)
123    goto err_sem;
124
125#ifdef RTEMS_POSIX_API
126  pipe_interruptible(pipe);
127#endif
128
129  *pipep = pipe;
130  if (c ++ == 'z')
131    c = 'a';
132  return 0;
133
134err_sem:
135  rtems_barrier_delete(pipe->writeBarrier);
136err_wbar:
137  rtems_barrier_delete(pipe->readBarrier);
138err_rbar:
139  free(pipe->Buffer);
140err_buf:
141  free(pipe);
142  return err;
143}
144
145/* Called with pipe_semaphore held. */
146static inline void pipe_free(
147  pipe_control_t *pipe
148)
149{
150  rtems_barrier_delete(pipe->readBarrier);
151  rtems_barrier_delete(pipe->writeBarrier);
152  rtems_semaphore_delete(pipe->Semaphore);
153  free(pipe->Buffer);
154  free(pipe);
155}
156
157static rtems_status_code pipe_lock(void)
158{
159  rtems_status_code sc = RTEMS_SUCCESSFUL;
160
161  if (pipe_semaphore == RTEMS_ID_NONE) {
162    rtems_libio_lock();
163
164    if (pipe_semaphore == RTEMS_ID_NONE) {
165      sc = rtems_semaphore_create(
166        rtems_build_name('P', 'I', 'P', 'E'),
167        1,
168        RTEMS_BINARY_SEMAPHORE | RTEMS_INHERIT_PRIORITY | RTEMS_PRIORITY,
169        RTEMS_NO_PRIORITY,
170        &pipe_semaphore
171      );
172    }
173
174    rtems_libio_unlock();
175  }
176
177  if (sc == RTEMS_SUCCESSFUL) {
178    sc = rtems_semaphore_obtain(pipe_semaphore, RTEMS_WAIT, RTEMS_NO_TIMEOUT);
179  }
180
181  if (sc == RTEMS_SUCCESSFUL) {
182    return 0;
183  } else {
184    return -ENOMEM;
185  }
186}
187
188static void pipe_unlock(void)
189{
190  rtems_status_code sc = RTEMS_SUCCESSFUL;
191
192  sc = rtems_semaphore_release(pipe_semaphore);
193  #ifdef RTEMS_DEBUG
194    if (sc != RTEMS_SUCCESSFUL) {
195      rtems_fatal_error_occurred(0xdeadbeef);
196    }
197  #endif
198}
199
200/*
201 * If called with *pipep = NULL, pipe_new will call pipe_alloc to allocate a
202 * pipe control structure and set *pipep to its address.
203 * pipe is locked, when pipe_new returns with no error.
204 */
205static int pipe_new(
206  pipe_control_t **pipep
207)
208{
209  pipe_control_t *pipe;
210  int err = 0;
211
212  err = pipe_lock();
213  if (err)
214    return err;
215
216  pipe = *pipep;
217  if (pipe == NULL) {
218    err = pipe_alloc(&pipe);
219    if (err)
220      goto out;
221  }
222
223  if (! PIPE_LOCK(pipe))
224    err = -EINTR;
225
226  if (*pipep == NULL) {
227    if (err)
228      pipe_free(pipe);
229    else
230      *pipep = pipe;
231  }
232
233out:
234  pipe_unlock();
235  return err;
236}
237
238/*
239 * Interface to file system close.
240 *
241 * *pipep points to pipe control structure. When the last user releases pipe,
242 * it will be set to NULL.
243 */
244int pipe_release(
245  pipe_control_t **pipep,
246  rtems_libio_t *iop
247)
248{
249  pipe_control_t *pipe = *pipep;
250  uint32_t mode;
251
252  #if defined(RTEMS_DEBUG)
253    /* WARN pipe not freed and pipep not set to NULL! */
254    if (pipe_lock())
255      rtems_fatal_error_occurred(0xdeadbeef);
256
257    /* WARN pipe not released! */
258    if (!PIPE_LOCK(pipe))
259      rtems_fatal_error_occurred(0xdeadbeef);
260  #endif
261
262  mode = LIBIO_ACCMODE(iop);
263  if (mode & LIBIO_FLAGS_READ)
264     pipe->Readers --;
265  if (mode & LIBIO_FLAGS_WRITE)
266     pipe->Writers --;
267
268  PIPE_UNLOCK(pipe);
269
270  if (pipe->Readers == 0 && pipe->Writers == 0) {
271#if 0
272    /* To delete an anonymous pipe file when all users closed it */
273    if (pipe->Anonymous)
274      delfile = TRUE;
275#endif
276    pipe_free(pipe);
277    *pipep = NULL;
278  }
279  else if (pipe->Readers == 0 && mode != LIBIO_FLAGS_WRITE)
280    /* Notify waiting Writers that all their partners left */
281    PIPE_WAKEUPWRITERS(pipe);
282  else if (pipe->Writers == 0 && mode != LIBIO_FLAGS_READ)
283    PIPE_WAKEUPREADERS(pipe);
284
285  pipe_unlock();
286
287#if 0
288  if (! delfile)
289    return 0;
290  if (iop->pathinfo.ops->unlink_h == NULL)
291    return 0;
292
293  /* This is safe for IMFS, but how about other FSes? */
294  iop->flags &= ~LIBIO_FLAGS_OPEN;
295  if(iop->pathinfo.ops->unlink_h(&iop->pathinfo))
296    return -errno;
297#endif
298
299  return 0;
300}
301
302/*
303 * Interface to file system open.
304 *
305 * *pipep points to pipe control structure. If called with *pipep = NULL,
306 * fifo_open will try allocating and initializing a control structure. If the
307 * call succeeds, *pipep will be set to address of new control structure.
308 */
309int fifo_open(
310  pipe_control_t **pipep,
311  rtems_libio_t *iop
312)
313{
314  pipe_control_t *pipe;
315  unsigned int prevCounter;
316  int err;
317
318  err = pipe_new(pipep);
319  if (err)
320    return err;
321  pipe = *pipep;
322
323  switch (LIBIO_ACCMODE(iop)) {
324    case LIBIO_FLAGS_READ:
325      pipe->readerCounter ++;
326      if (pipe->Readers ++ == 0)
327        PIPE_WAKEUPWRITERS(pipe);
328
329      if (pipe->Writers == 0) {
330        /* Not an error */
331        if (LIBIO_NODELAY(iop))
332          break;
333
334        prevCounter = pipe->writerCounter;
335        err = -EINTR;
336        /* Wait until a writer opens the pipe */
337        do {
338          PIPE_UNLOCK(pipe);
339          if (! PIPE_READWAIT(pipe))
340            goto out_error;
341          if (! PIPE_LOCK(pipe))
342            goto out_error;
343        } while (prevCounter == pipe->writerCounter);
344      }
345      break;
346
347    case LIBIO_FLAGS_WRITE:
348      pipe->writerCounter ++;
349
350      if (pipe->Writers ++ == 0)
351        PIPE_WAKEUPREADERS(pipe);
352
353      if (pipe->Readers == 0 && LIBIO_NODELAY(iop)) {
354        PIPE_UNLOCK(pipe);
355        err = -ENXIO;
356        goto out_error;
357      }
358
359      if (pipe->Readers == 0) {
360        prevCounter = pipe->readerCounter;
361        err = -EINTR;
362        do {
363          PIPE_UNLOCK(pipe);
364          if (! PIPE_WRITEWAIT(pipe))
365            goto out_error;
366          if (! PIPE_LOCK(pipe))
367            goto out_error;
368        } while (prevCounter == pipe->readerCounter);
369      }
370      break;
371
372    case LIBIO_FLAGS_READ_WRITE:
373      pipe->readerCounter ++;
374      if (pipe->Readers ++ == 0)
375        PIPE_WAKEUPWRITERS(pipe);
376      pipe->writerCounter ++;
377      if (pipe->Writers ++ == 0)
378        PIPE_WAKEUPREADERS(pipe);
379      break;
380  }
381
382  PIPE_UNLOCK(pipe);
383  return 0;
384
385out_error:
386  pipe_release(pipep, iop);
387  return err;
388}
389
390/*
391 * Interface to file system read.
392 */
393ssize_t pipe_read(
394  pipe_control_t *pipe,
395  void           *buffer,
396  size_t          count,
397  rtems_libio_t  *iop
398)
399{
400  int chunk, chunk1, read = 0, ret = 0;
401
402  if (! PIPE_LOCK(pipe))
403    return -EINTR;
404
405  while (read < count) {
406    while (PIPE_EMPTY(pipe)) {
407      /* Not an error */
408      if (pipe->Writers == 0)
409        goto out_locked;
410
411      if (LIBIO_NODELAY(iop)) {
412        ret = -EAGAIN;
413        goto out_locked;
414      }
415
416      /* Wait until pipe is no more empty or no writer exists */
417      pipe->waitingReaders ++;
418      PIPE_UNLOCK(pipe);
419      if (! PIPE_READWAIT(pipe))
420        ret = -EINTR;
421      if (! PIPE_LOCK(pipe)) {
422        /* WARN waitingReaders not restored! */
423        ret = -EINTR;
424        goto out_nolock;
425      }
426      pipe->waitingReaders --;
427      if (ret != 0)
428        goto out_locked;
429    }
430
431    /* Read chunk bytes */
432    chunk = MIN(count - read,  pipe->Length);
433    chunk1 = pipe->Size - pipe->Start;
434    if (chunk > chunk1) {
435      memcpy(buffer + read, pipe->Buffer + pipe->Start, chunk1);
436      memcpy(buffer + read + chunk1, pipe->Buffer, chunk - chunk1);
437    }
438    else
439      memcpy(buffer + read, pipe->Buffer + pipe->Start, chunk);
440
441    pipe->Start += chunk;
442    pipe->Start %= pipe->Size;
443    pipe->Length -= chunk;
444    /* For buffering optimization */
445    if (PIPE_EMPTY(pipe))
446      pipe->Start = 0;
447
448    if (pipe->waitingWriters > 0)
449      PIPE_WAKEUPWRITERS(pipe);
450    read += chunk;
451  }
452
453out_locked:
454  PIPE_UNLOCK(pipe);
455
456out_nolock:
457  if (read > 0)
458    return read;
459  return ret;
460}
461
462/*
463 * Interface to file system write.
464 */
465ssize_t pipe_write(
466  pipe_control_t *pipe,
467  const void     *buffer,
468  size_t          count,
469  rtems_libio_t  *iop
470)
471{
472  int chunk, chunk1, written = 0, ret = 0;
473
474  /* Write nothing */
475  if (count == 0)
476    return 0;
477
478  if (! PIPE_LOCK(pipe))
479    return -EINTR;
480
481  if (pipe->Readers == 0) {
482    ret = -EPIPE;
483    goto out_locked;
484  }
485
486  /* Write of PIPE_BUF bytes or less shall not be interleaved */
487  chunk = count <= pipe->Size ? count : 1;
488
489  while (written < count) {
490    while (PIPE_SPACE(pipe) < chunk) {
491      if (LIBIO_NODELAY(iop)) {
492        ret = -EAGAIN;
493        goto out_locked;
494      }
495
496      /* Wait until there is chunk bytes space or no reader exists */
497      pipe->waitingWriters ++;
498      PIPE_UNLOCK(pipe);
499      if (! PIPE_WRITEWAIT(pipe))
500        ret = -EINTR;
501      if (! PIPE_LOCK(pipe)) {
502        /* WARN waitingWriters not restored! */
503        ret = -EINTR;
504        goto out_nolock;
505      }
506      pipe->waitingWriters --;
507      if (ret != 0)
508        goto out_locked;
509
510      if (pipe->Readers == 0) {
511        ret = -EPIPE;
512        goto out_locked;
513      }
514    }
515
516    chunk = MIN(count - written, PIPE_SPACE(pipe));
517    chunk1 = pipe->Size - PIPE_WSTART(pipe);
518    if (chunk > chunk1) {
519      memcpy(pipe->Buffer + PIPE_WSTART(pipe), buffer + written, chunk1);
520      memcpy(pipe->Buffer, buffer + written + chunk1, chunk - chunk1);
521    }
522    else
523      memcpy(pipe->Buffer + PIPE_WSTART(pipe), buffer + written, chunk);
524
525    pipe->Length += chunk;
526    if (pipe->waitingReaders > 0)
527      PIPE_WAKEUPREADERS(pipe);
528    written += chunk;
529    /* Write of more than PIPE_BUF bytes can be interleaved */
530    chunk = 1;
531  }
532
533out_locked:
534  PIPE_UNLOCK(pipe);
535
536out_nolock:
537#ifdef RTEMS_POSIX_API
538  /* Signal SIGPIPE */
539  if (ret == -EPIPE)
540    kill(getpid(), SIGPIPE);
541#endif
542
543  if (written > 0)
544    return written;
545  return ret;
546}
547
548/*
549 * Interface to file system ioctl.
550 */
551int pipe_ioctl(
552  pipe_control_t *pipe,
553  uint32_t        cmd,
554  void           *buffer,
555  rtems_libio_t  *iop
556)
557{
558  if (cmd == FIONREAD) {
559    if (buffer == NULL)
560      return -EFAULT;
561
562    if (! PIPE_LOCK(pipe))
563      return -EINTR;
564
565    /* Return length of pipe */
566    *(unsigned int *)buffer = pipe->Length;
567    PIPE_UNLOCK(pipe);
568    return 0;
569  }
570
571  return -EINVAL;
572}
573
574/*
575 * Interface to file system lseek.
576 */
577int pipe_lseek(
578  pipe_control_t *pipe,
579  off_t           offset,
580  int             whence,
581  rtems_libio_t  *iop
582)
583{
584  /* Seek on pipe is not supported */
585  return -ESPIPE;
586}
Note: See TracBrowser for help on using the repository browser.