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   int quitflag;
  32   void *data;
  33 };
  34
  35 typedef struct cf_s *cf_t;
  36
  37 void *cf_data(cf_t cf) {
  38   return cf->data;
  39 }
  40
  41 // A bit like cooperative multitasking. Continued fractions are expected
  42 // to call this as often as practical, and on a return value of 0,
  43 // to drop everything and stop.
  44 int cf_wait(cf_t cf) {
  45   for (;;) {
  46     sem_wait(&cf->demand_sem);
  47     // The wait is over!
  48     if (cf->quitflag) {
  49       return 0;
  50     }
  51     pthread_mutex_lock(&cf->chan_mu);
  52     // ... but we keep waiting unless the channel is empty.
  53     if (!cf->chan) break;
  54     pthread_mutex_unlock(&cf->chan_mu);
  55     // The channel could be emptied in the meantime, but that
  56     // implies at least one sem_post() call, so we'll notice next iteration.
  57   }
  58   pthread_mutex_unlock(&cf->chan_mu);
  59   return 1;
  60 }
  61
  62 void cf_free(cf_t cf) {
  63   // These two statements force a thread out of its next/current cf_wait.
  64   cf->quitflag = 1;
  65   sem_post(&cf->demand_sem);
  66
  67   pthread_join(cf->thread, NULL);
  68   pthread_mutex_lock(&cf->chan_mu);
  69   channel_ptr c = cf->chan;
  70   while (c) {
  71     channel_ptr cnext = c->next;
  72     free(c->data);
  73     free(c);
  74     c = cnext;
  75   }
  76   pthread_mutex_unlock(&cf->chan_mu);
  77   sem_destroy(&cf->demand_sem);
  78   free(cf);
  79 }
  80
  81 void cf_put(cf_t cf, mpz_t z) {
  82   // TODO: Block or something if there's a large backlog on the queue.
  83   channel_ptr cnew = malloc(sizeof(*cnew));
  84   mpz_ptr znew = malloc(sizeof(*znew));
  85   mpz_init(znew);
  86   mpz_set(znew, z);
  87   cnew->data = znew;
  88   cnew->next = NULL;
  89   pthread_mutex_lock(&cf->chan_mu);
  90   if (cf->chan) {
  91     cf->next->next = cnew;
  92   } else {
  93     // Channel is empty. Now that we're populating it, send signal
  94     // in case someone is waiting for data.
  95     cf->chan = cnew;
  96     pthread_cond_signal(&cf->read_cond);
  97   }
  98   cf->next = cnew;
  99   pthread_mutex_unlock(&cf->chan_mu);
 100 }
 101
 102 void cf_put_int(cf_t cf, int n) {
 103   mpz_t z;
 104   mpz_init(z);
 105   mpz_set_si(z, n);
 106   cf_put(cf, z);
 107   mpz_clear(z);
 108 }
 109
 110 void cf_get(mpz_t z, cf_t cf) {
 111   pthread_mutex_lock(&cf->chan_mu);
 112   if (!cf->chan) {
 113     // If channel is empty, send demand signal and wait for read signal.
 114     sem_post(&cf->demand_sem);
 115     pthread_cond_wait(&cf->read_cond, &cf->chan_mu);
 116   }
 117   channel_ptr c = cf->chan;
 118   cf->chan = c->next;
 119   pthread_mutex_unlock(&cf->chan_mu);
 120   mpz_ptr znew = c->data;
 121   mpz_set(z, znew);
 122   mpz_clear(znew);
 123   free(c->data);
 124   free(c);
 125 }
 126
 127 cf_t cf_new(void *(*func)(cf_t), void *data) {
 128   cf_t cf = malloc(sizeof(*cf));
 129   cf->chan = NULL;
 130   cf->next = NULL;
 131   cf->quitflag = 0;
 132   cf->data = data;
 133   pthread_attr_t attr;
 134   pthread_attr_init(&attr);
 135   pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
 136   pthread_mutex_init(&cf->chan_mu, NULL);
 137   sem_init(&cf->demand_sem, 0, 0);
 138   pthread_cond_init(&cf->read_cond, NULL);
 139   pthread_create(&cf->thread, &attr, (void*(*)(void *)) func, cf);
 140   pthread_attr_destroy(&attr);
 141   return cf;
 142 }