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 <poll.h>
  21 #include <pthread.h>
  22 #include <semaphore.h>
  23 #include <stdint.h>
  24 #include <stdio.h>
  25 #include <stdlib.h>
  26 #include <string.h>
  27 #include <strings.h>
  28 #include <unistd.h>
  29 #include <sys/stat.h>
  30
  31 /*
  32  * This is an in-memory "pipe" construct with a twist: it lets you have
  33  * multiple consumers instead of just one.  For example, you might want
  34  * to stream data from a back-end store both to the user and into the
  35  * local cache, or you might want to replicate out to several back ends
  36  * simultaneously.  The basic flow for the producer is as follows:
  37  *
  38  *      while data available
  39  *              read a chunk of data
  40  *              lock shared structure
  41  *              update the shared pointer/data/sequence
  42  *              signal the consumer event
  43  *              wait on the producer event
  44  *              unlock shared structure
  45  *      lock shared structure
  46  *      set prod_done
  47  *      signal the consumer event
  48  *      wait on the producer event
  49  *
  50  * For consumers, it's a mirror image:
  51  *      lock shared structure
  52  *      loop
  53  *              wait on the consumer event
  54  *              continue if shared sequence != own sequence
  55  *              break if len == 0
  56  *              unlock shared structure
  57  *              write the data somewhere
  58  *              increment own sequence
  59  *              lock shared structure
  60  *              signal producer event if ++cons_done == cons_total
  61  *      do cons_count/producer-event handshake one more time
  62  *
  63  * The sequence checking is not strictly necessary, but it makes things a lot
  64  * easier to debug if there is a bug that causes producer and consumers to get
  65  * out of sync.  Instead of corrupting data and continuing, consumers block
  66  * waiting for the "right" sequence number while the producer blocks waiting
  67  * for a signal that will never come.
  68  *
  69  * The cons_error is the "deadweight" that only increments. This way the
  70  * thread ping-pong and zeroing of cons_done are left alone.
  71  */
  72
  73 typedef struct {
  74         void            *owner;
  75         pthread_mutex_t  lock;
  76         pthread_cond_t   prod_cond;
  77         pthread_cond_t   cons_cond;
  78         void            *data_ptr;
  79         size_t           data_len;
  80         unsigned long    sequence;
  81         unsigned short   cons_total;
  82         unsigned short   cons_init_done;
  83         unsigned short   cons_init_error;
  84         unsigned short   cons_done;
  85         unsigned short   cons_error;
  86         enum { PROD_INIT, PROD_RUNNING, PROD_ERROR } prod_state;
  87 } pipe_shared;
  88
  89 typedef struct {
  90         pipe_shared     *shared;
  91         unsigned long    sequence;
  92         size_t           offset;
  93         void            *prov;
  94 } pipe_private;
  95
  96
  97 void             pipe_init_shared       (pipe_shared *ps,
  98                                          void *owner, unsigned short ncons);
  99 pipe_private    *pipe_init_private      (pipe_shared *ps);
 100 int              pipe_cons_wait         (pipe_private *pp);
 101 void             pipe_cons_signal       (pipe_private *pp, int error);
 102 void             pipe_cons_siginit      (pipe_shared *ps, int error);
 103 int              pipe_prod_wait_init    (pipe_shared *ps);
 104 void             pipe_prod_siginit      (pipe_shared *ps, int error);
 105 int              pipe_cons_wait_init    (pipe_shared *ps);
 106 void             pipe_prod_signal       (pipe_shared *ps,
 107                                          void *ptr, size_t total);
 108 void             pipe_prod_finish       (pipe_shared *ps);
 109 void             pipe_reset             (pipe_shared *ps, unsigned short ncons);
 110
 111 #endif