tee.c

   1 // Tee: read input channel and write to two output channels
   2
   3 #include <stdio.h>
   4 #include <stdlib.h>
   5 #include <string.h>
   6 #include <gmp.h>
   7 #include "cf.h"
   8
   9 struct tee_s {
  10   int n;
  11   cf_t parent;
  12 };
  13 typedef struct tee_s tee_t[1];
  14 typedef struct tee_s *tee_ptr;
  15
  16 static void *branch_loop(cf_t cf) {
  17   tee_ptr t = cf_data(cf);
  18   mpz_t z;
  19   mpz_init(z);
  20   while(cf_wait(cf)) {
  21     cf_put_int(t->parent, t->n);  // Causes parent to put to cf.
  22     cf_signal(t->parent);
  23   }
  24   cf_put_int(t->parent, -1 - t->n);  // Notify parent of our destruction.
  25   mpz_clear(z);
  26   free(t);
  27   return NULL;
  28 }
  29
  30 struct parent_data_s {
  31   cf_t kid[2];
  32   cf_t in;
  33 };
  34 typedef struct parent_data_s parent_data_t[1];
  35 typedef struct parent_data_s *parent_data_ptr;
  36
  37 // TODO: Join after destruction.
  38 static void *parent_loop(cf_t cf) {
  39   struct backlog_s {
  40     mpz_t z;
  41     struct backlog_s *next;
  42   };
  43   typedef struct backlog_s *backlog_ptr;
  44
  45   backlog_ptr head[2];
  46   backlog_ptr last[2];
  47   head[0] = NULL;
  48   head[1] = NULL;
  49   last[0] = NULL;
  50   last[1] = NULL;
  51   mpz_t z;
  52   mpz_init(z);
  53   parent_data_ptr pd = cf_data(cf);
  54   for (;;) {
  55     cf_wait_special(cf);
  56     cf_get(z, cf);
  57     int k = mpz_get_ui(z);
  58     if (k < 0) {
  59       k = -k + 1;
  60       pd->kid[k] = NULL;
  61       backlog_ptr pnext = head[k], p;
  62       do {
  63         p = pnext;
  64         pnext = p->next;
  65         mpz_clear(p->z);
  66         free(p);
  67       } while(pnext);
  68       if (!pd->kid[1-k]) break;
  69     } else {
  70       if (head[k]) {
  71         backlog_ptr p = head[k];
  72         mpz_set(z, p->z);
  73         mpz_clear(p->z);
  74         head[k] = p->next;
  75         free(p);
  76         if (!head[k]) last[k] = NULL;
  77       } else {
  78         cf_get(z, pd->in);
  79         if (pd->kid[1-k]) {
  80           backlog_ptr p = malloc(sizeof(*p));
  81           mpz_init(p->z);
  82           mpz_set(p->z, z);
  83           p->next = NULL;
  84           if (last[1-k]) last[1-k]->next = p;
  85           last[1-k] = p;
  86           if (!head[1-k]) head[1-k] = p;
  87         }
  88       }
  89     }
  90     cf_put(pd->kid[k], z);
  91   }
  92   mpz_clear(z);
  93   return NULL;
  94 }
  95
  96 cf_t cf_new_parent(cf_t in) {
  97   parent_data_ptr p = malloc(sizeof(*p));
  98   p->in = in;
  99   p->kid[0] = NULL;
 100   p->kid[1] = NULL;
 101   return cf_new(parent_loop, p);
 102 }
 103
 104 cf_t cf_new_branch(cf_t parent, int n) {
 105   tee_ptr t = malloc(sizeof(*t));
 106   t->n = n;
 107   t->parent = parent;
 108   parent_data_ptr p = cf_data(parent);
 109   cf_t res = cf_new(branch_loop, t);
 110   p->kid[n] = res;
 111   return res;
 112 }
 113
 114 void cf_tee(cf_t *out_array, cf_t in) {
 115   cf_t parent = cf_new_parent(in);
 116   out_array[0] = cf_new_branch(parent, 0);
 117   out_array[1] = cf_new_branch(parent, 1);
 118 }