source: rtems/cpukit/libfs/src/pipe/fifo.c @ 2d1994c9

4.11
Last change on this file since 2d1994c9 was 2d1994c9, checked in by Joel Sherrill <joel.sherrill@…>, on Jun 24, 2010 at 7:57:58 PM

2010-06-24 Bharath Suri <bharath.s.jois@…>

PR 1542/filesystem
PR 1585/filesystem

  • libfs/src/pipe/fifo.c: pipe_control_t was not deallocated if fifo_open() was attempted with (O_WRONLY|O_NONBLOCK). Mutex was locked too many times on this path and we needed an unlock.
  • 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  if (sc != RTEMS_SUCCESSFUL) {
194    /* FIXME */
195    rtems_fatal_error_occurred(0xdeadbeef);
196  }
197}
198
199/*
200 * If called with *pipep = NULL, pipe_new will call pipe_alloc to allocate a
201 * pipe control structure and set *pipep to its address.
202 * pipe is locked, when pipe_new returns with no error.
203 */
204static int pipe_new(
205  pipe_control_t **pipep
206)
207{
208  pipe_control_t *pipe;
209  int err = 0;
210
211  err = pipe_lock();
212  if (err)
213    return err;
214
215  pipe = *pipep;
216  if (pipe == NULL) {
217    err = pipe_alloc(&pipe);
218    if (err)
219      goto out;
220  }
221
222  if (! PIPE_LOCK(pipe))
223    err = -EINTR;
224
225  if (*pipep == NULL) {
226    if (err)
227      pipe_free(pipe);
228    else
229      *pipep = pipe;
230  }
231
232out:
233  pipe_unlock();
234  return err;
235}
236
237/*
238 * Interface to file system close.
239 *
240 * *pipep points to pipe control structure. When the last user releases pipe,
241 * it will be set to NULL.
242 */
243int pipe_release(
244  pipe_control_t **pipep,
245  rtems_libio_t *iop
246)
247{
248  pipe_control_t *pipe = *pipep;
249  uint32_t mode;
250
251  if (pipe_lock())
252    /* WARN pipe not freed and pipep not set to NULL! */
253    /* FIXME */
254    rtems_fatal_error_occurred(0xdeadbeef);
255
256  if (!PIPE_LOCK(pipe))
257    /* WARN pipe not released! */
258    /* FIXME */
259    rtems_fatal_error_occurred(0xdeadbeef);
260
261  mode = LIBIO_ACCMODE(iop);
262  if (mode & LIBIO_FLAGS_READ)
263     pipe->Readers --;
264  if (mode & LIBIO_FLAGS_WRITE)
265     pipe->Writers --;
266
267  PIPE_UNLOCK(pipe);
268
269  if (pipe->Readers == 0 && pipe->Writers == 0) {
270#if 0
271    /* To delete an anonymous pipe file when all users closed it */
272    if (pipe->Anonymous)
273      delfile = TRUE;
274#endif
275    pipe_free(pipe);
276    *pipep = NULL;
277  }
278  else if (pipe->Readers == 0 && mode != LIBIO_FLAGS_WRITE)
279    /* Notify waiting Writers that all their partners left */
280    PIPE_WAKEUPWRITERS(pipe);
281  else if (pipe->Writers == 0 && mode != LIBIO_FLAGS_READ)
282    PIPE_WAKEUPREADERS(pipe);
283
284  pipe_unlock();
285
286#if 0
287  if (! delfile)
288    return 0;
289  if (iop->pathinfo.ops->unlink_h == NULL)
290    return 0;
291
292  /* This is safe for IMFS, but how about other FSes? */
293  iop->flags &= ~LIBIO_FLAGS_OPEN;
294  if(iop->pathinfo.ops->unlink_h(&iop->pathinfo))
295    return -errno;
296#endif
297
298  return 0;
299}
300
301/*
302 * Interface to file system open.
303 *
304 * *pipep points to pipe control structure. If called with *pipep = NULL,
305 * fifo_open will try allocating and initializing a control structure. If the
306 * call succeeds, *pipep will be set to address of new control structure.
307 */
308int fifo_open(
309  pipe_control_t **pipep,
310  rtems_libio_t *iop
311)
312{
313  pipe_control_t *pipe;
314  unsigned int prevCounter;
315  int err;
316
317  err = pipe_new(pipep);
318  if (err)
319    return err;
320  pipe = *pipep;
321
322  switch (LIBIO_ACCMODE(iop)) {
323    case LIBIO_FLAGS_READ:
324      pipe->readerCounter ++;
325      if (pipe->Readers ++ == 0)
326        PIPE_WAKEUPWRITERS(pipe);
327
328      if (pipe->Writers == 0) {
329        /* Not an error */
330        if (LIBIO_NODELAY(iop))
331          break;
332
333        prevCounter = pipe->writerCounter;
334        err = -EINTR;
335        /* Wait until a writer opens the pipe */
336        do {
337          PIPE_UNLOCK(pipe);
338          if (! PIPE_READWAIT(pipe))
339            goto out_error;
340          if (! PIPE_LOCK(pipe))
341            goto out_error;
342        } while (prevCounter == pipe->writerCounter);
343      }
344      break;
345
346    case LIBIO_FLAGS_WRITE:
347      pipe->writerCounter ++;
348
349      if (pipe->Writers ++ == 0)
350        PIPE_WAKEUPREADERS(pipe);
351
352      if (pipe->Readers == 0 && LIBIO_NODELAY(iop)) {
353        PIPE_UNLOCK(pipe);
354        err = -ENXIO;
355        goto out_error;
356      }
357
358      if (pipe->Readers == 0) {
359        prevCounter = pipe->readerCounter;
360        err = -EINTR;
361        do {
362          PIPE_UNLOCK(pipe);
363          if (! PIPE_WRITEWAIT(pipe))
364            goto out_error;
365          if (! PIPE_LOCK(pipe))
366            goto out_error;
367        } while (prevCounter == pipe->readerCounter);
368      }
369      break;
370
371    case LIBIO_FLAGS_READ_WRITE:
372      pipe->readerCounter ++;
373      if (pipe->Readers ++ == 0)
374        PIPE_WAKEUPWRITERS(pipe);
375      pipe->writerCounter ++;
376      if (pipe->Writers ++ == 0)
377        PIPE_WAKEUPREADERS(pipe);
378      break;
379  }
380
381  PIPE_UNLOCK(pipe);
382  return 0;
383
384out_error:
385  pipe_release(pipep, iop);
386  return err;
387}
388
389/*
390 * Interface to file system read.
391 */
392ssize_t pipe_read(
393  pipe_control_t *pipe,
394  void           *buffer,
395  size_t          count,
396  rtems_libio_t  *iop
397)
398{
399  int chunk, chunk1, read = 0, ret = 0;
400
401  if (! PIPE_LOCK(pipe))
402    return -EINTR;
403
404  while (read < count) {
405    while (PIPE_EMPTY(pipe)) {
406      /* Not an error */
407      if (pipe->Writers == 0)
408        goto out_locked;
409
410      if (LIBIO_NODELAY(iop)) {
411        ret = -EAGAIN;
412        goto out_locked;
413      }
414
415      /* Wait until pipe is no more empty or no writer exists */
416      pipe->waitingReaders ++;
417      PIPE_UNLOCK(pipe);
418      if (! PIPE_READWAIT(pipe))
419        ret = -EINTR;
420      if (! PIPE_LOCK(pipe)) {
421        /* WARN waitingReaders not restored! */
422        ret = -EINTR;
423        goto out_nolock;
424      }
425      pipe->waitingReaders --;
426      if (ret != 0)
427        goto out_locked;
428    }
429
430    /* Read chunk bytes */
431    chunk = MIN(count - read,  pipe->Length);
432    chunk1 = pipe->Size - pipe->Start;
433    if (chunk > chunk1) {
434      memcpy(buffer + read, pipe->Buffer + pipe->Start, chunk1);
435      memcpy(buffer + read + chunk1, pipe->Buffer, chunk - chunk1);
436    }
437    else
438      memcpy(buffer + read, pipe->Buffer + pipe->Start, chunk);
439
440    pipe->Start += chunk;
441    pipe->Start %= pipe->Size;
442    pipe->Length -= chunk;
443    /* For buffering optimization */
444    if (PIPE_EMPTY(pipe))
445      pipe->Start = 0;
446
447    if (pipe->waitingWriters > 0)
448      PIPE_WAKEUPWRITERS(pipe);
449    read += chunk;
450  }
451
452out_locked:
453  PIPE_UNLOCK(pipe);
454
455out_nolock:
456  if (read > 0)
457    return read;
458  return ret;
459}
460
461/*
462 * Interface to file system write.
463 */
464ssize_t pipe_write(
465  pipe_control_t *pipe,
466  const void     *buffer,
467  size_t          count,
468  rtems_libio_t  *iop
469)
470{
471  int chunk, chunk1, written = 0, ret = 0;
472
473  /* Write nothing */
474  if (count == 0)
475    return 0;
476
477  if (! PIPE_LOCK(pipe))
478    return -EINTR;
479
480  if (pipe->Readers == 0) {
481    ret = -EPIPE;
482    goto out_locked;
483  }
484
485  /* Write of PIPE_BUF bytes or less shall not be interleaved */
486  chunk = count <= pipe->Size ? count : 1;
487
488  while (written < count) {
489    while (PIPE_SPACE(pipe) < chunk) {
490      if (LIBIO_NODELAY(iop)) {
491        ret = -EAGAIN;
492        goto out_locked;
493      }
494
495      /* Wait until there is chunk bytes space or no reader exists */
496      pipe->waitingWriters ++;
497      PIPE_UNLOCK(pipe);
498      if (! PIPE_WRITEWAIT(pipe))
499        ret = -EINTR;
500      if (! PIPE_LOCK(pipe)) {
501        /* WARN waitingWriters not restored! */
502        ret = -EINTR;
503        goto out_nolock;
504      }
505      pipe->waitingWriters --;
506      if (ret != 0)
507        goto out_locked;
508
509      if (pipe->Readers == 0) {
510        ret = -EPIPE;
511        goto out_locked;
512      }
513    }
514
515    chunk = MIN(count - written, PIPE_SPACE(pipe));
516    chunk1 = pipe->Size - PIPE_WSTART(pipe);
517    if (chunk > chunk1) {
518      memcpy(pipe->Buffer + PIPE_WSTART(pipe), buffer + written, chunk1);
519      memcpy(pipe->Buffer, buffer + written + chunk1, chunk - chunk1);
520    }
521    else
522      memcpy(pipe->Buffer + PIPE_WSTART(pipe), buffer + written, chunk);
523
524    pipe->Length += chunk;
525    if (pipe->waitingReaders > 0)
526      PIPE_WAKEUPREADERS(pipe);
527    written += chunk;
528    /* Write of more than PIPE_BUF bytes can be interleaved */
529    chunk = 1;
530  }
531
532out_locked:
533  PIPE_UNLOCK(pipe);
534
535out_nolock:
536#ifdef RTEMS_POSIX_API
537  /* Signal SIGPIPE */
538  if (ret == -EPIPE)
539    kill(getpid(), SIGPIPE);
540#endif
541
542  if (written > 0)
543    return written;
544  return ret;
545}
546
547/*
548 * Interface to file system ioctl.
549 */
550int pipe_ioctl(
551  pipe_control_t *pipe,
552  uint32_t        cmd,
553  void           *buffer,
554  rtems_libio_t  *iop
555)
556{
557  if (cmd == FIONREAD) {
558    if (buffer == NULL)
559      return -EFAULT;
560
561    if (! PIPE_LOCK(pipe))
562      return -EINTR;
563
564    /* Return length of pipe */
565    *(unsigned int *)buffer = pipe->Length;
566    PIPE_UNLOCK(pipe);
567    return 0;
568  }
569
570  return -EINVAL;
571}
572
573/*
574 * Interface to file system lseek.
575 */
576int pipe_lseek(
577  pipe_control_t *pipe,
578  off_t           offset,
579  int             whence,
580  rtems_libio_t  *iop
581)
582{
583  /* Seek on pipe is not supported */
584  return -ESPIPE;
585}
Note: See TracBrowser for help on using the repository browser.