mpipe.h

   1 /* Copyright (C) 2010 Red Hat, Inc.
   2
   3    This program is free software: you can redistribute it and/or modify
   4    it under the terms of the GNU General Public License as published by
   5    the Free Software Foundation, either version 3 of the License, or
   6    (at your option) any later version.
   7
   8    This program is distributed in the hope that it will be useful,
   9    but WITHOUT ANY WARRANTY; without even the implied warranty of
  10    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11    GNU General Public License for more details.
  12
  13    You should have received a copy of the GNU General Public License
  14    along with this program.  If not, see <http://www.gnu.org/licenses/>.  */
  15
  16 #if !defined(_MPIPE_H)
  17 #define _MPIPE_H
  18
  19 #include <fcntl.h>
  20 #include <getopt.h>
  21 #include <poll.h>
  22 #include <pthread.h>
  23 #include <semaphore.h>
  24 #include <stdint.h>
  25 #include <stdio.h>
  26 #include <stdlib.h>
  27 #include <string.h>
  28 #include <strings.h>
  29 #include <unistd.h>
  30 #include <sys/stat.h>
  31
  32 /*
  33  * This is an in-memory "pipe" construct with a twist: it lets you have
  34  * multiple consumers instead of just one.  For example, you might want
  35  * to stream data from a back-end store both to the user and into the
  36  * local cache, or you might want to replicate out to several back ends
  37  * simultaneously.  The basic flow for the producer is as follows:
  38  *
  39  *      while data available
  40  *              read a chunk of data
  41  *              lock shared structure
  42  *              update the shared pointer/data/sequence
  43  *              signal the consumer event
  44  *              wait on the producer event
  45  *              unlock shared structure
  46  *      lock shared structure
  47  *      set prod_done
  48  *      signal the consumer event
  49  *      wait on the producer event
  50  *
  51  * For consumers, it's a mirror image:
  52  *      lock shared structure
  53  *      loop
  54  *              wait on the consumer event
  55  *              continue if shared sequence != own sequence
  56  *              break if len == 0
  57  *              unlock shared structure
  58  *              write the data somewhere
  59  *              increment own sequence
  60  *              lock shared structure
  61  *              signal producer event if ++cons_done == cons_total
  62  *      do cons_count/producer-event handshake one more time
  63  *
  64  * The sequence checking is not strictly necessary, but it makes things a lot
  65  * easier to debug if there is a bug that causes producer and consumers to get
  66  * out of sync.  Instead of corrupting data and continuing, consumers block
  67  * waiting for the "right" sequence number while the producer blocks waiting
  68  * for a signal that will never come.
  69  *
  70  * The cons_error is the "deadweight" that only increments. This way the
  71  * thread ping-pong and zeroing of cons_done are left alone.
  72  */
  73
  74 typedef struct {
  75         void            *owner;
  76         pthread_mutex_t  lock;
  77         pthread_cond_t   prod_cond;
  78         pthread_cond_t   cons_cond;
  79         void            *data_ptr;
  80         size_t           data_len;
  81         unsigned long    sequence;
  82         int              in_init;
  83         unsigned short   cons_total;
  84         unsigned short   cons_done;
  85         unsigned short   cons_error;
  86 } pipe_shared;
  87
  88 typedef struct {
  89         pipe_shared     *shared;
  90         unsigned long    sequence;
  91         size_t           offset;
  92 } pipe_private;
  93
  94
  95 void             pipe_init_shared       (pipe_shared *ps,
  96                                          void *owner, unsigned short ncons);
  97 pipe_private    *pipe_init_private      (pipe_shared *ps);
  98 int              pipe_cons_wait         (pipe_private *pp);
  99 void             pipe_cons_signal       (pipe_private *pp, int error);
 100 void             pipe_cons_siginit      (pipe_shared *ps, int error);
 101 int              pipe_prod_wait_init    (pipe_shared *ps);
 102 void             pipe_prod_signal       (pipe_shared *ps,
 103                                          void *ptr, size_t total);
 104 void             pipe_prod_finish       (pipe_shared *ps);
 105
 106 #endif