mpipe.h

   1 #if !defined(_MPIPE_H)
   2 #define _MPIPE_H
   3
   4 #include <fcntl.h>
   5 #include <getopt.h>
   6 #include <poll.h>
   7 #include <pthread.h>
   8 #include <semaphore.h>
   9 #include <stdint.h>
  10 #include <stdio.h>
  11 #include <stdlib.h>
  12 #include <string.h>
  13 #include <strings.h>
  14 #include <unistd.h>
  15 #include <sys/stat.h>
  16
  17 /*
  18  * This is an in-memory "pipe" construct with a twist: it lets you have
  19  * multiple consumers instead of just one.  For example, you might want
  20  * to stream data from a back-end store both to the user and into the
  21  * local cache, or you might want to replicate out to several back ends
  22  * simultaneously.  The basic flow for the producer is as follows:
  23  *
  24  *      while data available
  25  *              read a chunk of data
  26  *              lock shared structure
  27  *              update the shared pointer/data/sequence
  28  *              signal the consumer event
  29  *              wait on the producer event
  30  *              unlock shared structure
  31  *      lock shared structure
  32  *      set prod_done
  33  *      signal the consumer event
  34  *      wait on the producer event
  35  *
  36  * For consumers, it's a mirror image:
  37  *      lock shared structure
  38  *      loop
  39  *              wait on the consumer event
  40  *              continue if shared sequence != own sequence
  41  *              break if len == 0
  42  *              unlock shared structure
  43  *              write the data somewhere
  44  *              increment own sequence
  45  *              lock shared structure
  46  *              signal producer event if ++cons_done == cons_total
  47  *      do cons_count/producer-event handshake one more time
  48  *
  49  * The sequence checking is not strictly necessary, but it makes things a lot
  50  * easier to debug if there is a bug that causes producer and consumers to get
  51  * out of sync.  Instead of corrupting data and continuing, consumers block
  52  * waiting for the "right" sequence number while the producer blocks waiting
  53  * for a signal that will never come.
  54  */
  55
  56 typedef struct {
  57         void            *owner;
  58         pthread_mutex_t  lock;
  59         pthread_cond_t   prod_cond;
  60         pthread_cond_t   cons_cond;
  61         void            *data_ptr;
  62         size_t           data_len;
  63         unsigned long    sequence;
  64         unsigned short   cons_total;
  65         unsigned short   cons_done;
  66 } pipe_shared;
  67
  68 typedef struct {
  69         pipe_shared     *shared;
  70         unsigned long    sequence;
  71         size_t           offset;
  72 } pipe_private;
  73
  74
  75 void             pipe_init_shared       (pipe_shared *ps,
  76                                          void *owner, unsigned short ncons);
  77 pipe_private    *pipe_init_private      (pipe_shared *ps);
  78 int              pipe_cons_wait         (pipe_private *pp);
  79 void             pipe_cons_signal       (pipe_private *pp);
  80 void             pipe_prod_signal       (pipe_shared *ps,
  81                                          void *ptr, size_t total);
  82 void             pipe_prod_finish       (pipe_shared *ps);
  83
  84 #endif