cf.c

   1 // Demand channels. See squint paper by McIlroy.
   2 //
   3 // TODO: Handle messy thread problems. What happens if a thread quits
   4 // but then another tries to signal and read its channel?
   5 // TODO: What if the continued fraction terminates?
   6 #include <stdio.h>
   7 #include <stdlib.h>
   8 #include <pthread.h>
   9 #include <semaphore.h>
  10 #include <gmp.h>
  11
  12 struct channel_s {
  13   void *data;
  14   struct channel_s *next;
  15 };
  16 typedef struct channel_s channel_t[1];
  17 typedef struct channel_s *channel_ptr;
  18
  19 struct cf_s {
  20   // Each continued fraction is a separate thread.
  21   pthread_t thread;
  22   // Helgrind prints warnings for these condition variables.
  23   // Rewrite with semaphores?
  24   // When queue is empty, and there is demand for the next term.
  25   sem_t demand_sem;
  26   // When the queue was empty, and we just added to it.
  27   pthread_cond_t read_cond;
  28   pthread_mutex_t chan_mu;
  29   channel_ptr chan, next;
  30
  31   // We break the CSP model slightly here: the sign of the continued
  32   // fraction is read directly from a variable, not over channels.
  33   // Only 'thread' may write to 'sign', and it should do so before the first
  34   // cf_put(). Other threads should only read it after they have
  35   // called cf_get() at least once.
  36   int sign;
  37   int quitflag;
  38   void *data;
  39 };
  40
  41 typedef struct cf_s *cf_t;
  42
  43 void *cf_data(cf_t cf) {
  44   return cf->data;
  45 }
  46
  47 void cf_set_sign(cf_t cf, int sign) {
  48   cf->sign = sign;
  49 }
  50
  51 int cf_sign(cf_t cf) {
  52   return cf->sign;
  53 }
  54
  55 int cf_flip_sign(cf_t cf) {
  56   return cf->sign = -cf->sign;
  57 }
  58
  59 // A bit like cooperative multitasking. Continued fractions are expected
  60 // to call this as often as practical, and on a return value of 0,
  61 // to drop everything and stop.
  62 int cf_wait(cf_t cf) {
  63   for (;;) {
  64     sem_wait(&cf->demand_sem);
  65     // The wait is over!
  66     if (cf->quitflag) {
  67       return 0;
  68     }
  69     pthread_mutex_lock(&cf->chan_mu);
  70     // ... but we keep waiting unless the channel is empty.
  71     if (!cf->chan) break;
  72     pthread_mutex_unlock(&cf->chan_mu);
  73     // The channel could be emptied in the meantime, but that
  74     // implies at least one sem_post() call, so we'll notice next iteration.
  75   }
  76   pthread_mutex_unlock(&cf->chan_mu);
  77   return 1;
  78 }
  79
  80 void cf_free(cf_t cf) {
  81   // These two statements force a thread out of its next/current cf_wait.
  82   cf->quitflag = 1;
  83   sem_post(&cf->demand_sem);
  84
  85   pthread_join(cf->thread, NULL);
  86   pthread_mutex_lock(&cf->chan_mu);
  87   channel_ptr c = cf->chan;
  88   while (c) {
  89     channel_ptr cnext = c->next;
  90     mpz_clear(c->data);
  91     free(c->data);
  92     free(c);
  93     c = cnext;
  94   }
  95   pthread_mutex_unlock(&cf->chan_mu);
  96   sem_destroy(&cf->demand_sem);
  97   free(cf);
  98 }
  99
 100 void cf_put(cf_t cf, mpz_t z) {
 101   // TODO: Block or something if there's a large backlog on the queue.
 102   channel_ptr cnew = malloc(sizeof(*cnew));
 103   mpz_ptr znew = malloc(sizeof(*znew));
 104   mpz_init(znew);
 105   mpz_set(znew, z);
 106   cnew->data = znew;
 107   cnew->next = NULL;
 108   pthread_mutex_lock(&cf->chan_mu);
 109   if (cf->chan) {
 110     cf->next->next = cnew;
 111   } else {
 112     // Channel is empty. Now that we're populating it, send signal
 113     // in case someone is waiting for data.
 114     cf->chan = cnew;
 115     pthread_cond_signal(&cf->read_cond);
 116   }
 117   cf->next = cnew;
 118   pthread_mutex_unlock(&cf->chan_mu);
 119 }
 120
 121 void cf_put_int(cf_t cf, int n) {
 122   mpz_t z;
 123   mpz_init(z);
 124   mpz_set_si(z, n);
 125   cf_put(cf, z);
 126   mpz_clear(z);
 127 }
 128
 129 void cf_get(mpz_t z, cf_t cf) {
 130   pthread_mutex_lock(&cf->chan_mu);
 131   if (!cf->chan) {
 132     // If channel is empty, send demand signal and wait for read signal.
 133     sem_post(&cf->demand_sem);
 134     pthread_cond_wait(&cf->read_cond, &cf->chan_mu);
 135   }
 136   channel_ptr c = cf->chan;
 137   cf->chan = c->next;
 138   pthread_mutex_unlock(&cf->chan_mu);
 139   mpz_ptr znew = c->data;
 140   mpz_set(z, znew);
 141   mpz_clear(znew);
 142   free(c->data);
 143   free(c);
 144 }
 145
 146 cf_t cf_new(void *(*func)(cf_t), void *data) {
 147   cf_t cf = malloc(sizeof(*cf));
 148   cf->sign = 1;
 149   cf->chan = NULL;
 150   cf->next = NULL;
 151   cf->quitflag = 0;
 152   cf->data = data;
 153   pthread_attr_t attr;
 154   pthread_attr_init(&attr);
 155   pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
 156   pthread_mutex_init(&cf->chan_mu, NULL);
 157   sem_init(&cf->demand_sem, 0, 0);
 158   pthread_cond_init(&cf->read_cond, NULL);
 159   pthread_create(&cf->thread, &attr, (void*(*)(void *)) func, cf);
 160   pthread_attr_destroy(&attr);
 161   return cf;
 162 }