primes.cc

#include <thread>
#include <mutex>
#include <condition_variable>
#include <deque>
#include <string>
#include <iostream>
#include <utility>

class semaphore {
public:
  void wait() {
    std::unique_lock<std::mutex> lock(mtx);
    while (!up)
      cnd.wait(lock);
    up = false;
  }

  void signal() {
    std::unique_lock<std::mutex> lock(mtx);
    up = true;
    cnd.notify_one();
  }

private:
  std::mutex mtx;
  std::condition_variable cnd;
  bool up = false;
};

template <typename T, bool sync> class channel;

template <typename T> class channel<T, false> {
public:
  template <typename U> void put(U &&t) {
    std::unique_lock<std::mutex> lock(mutex);
    queue.push_back(std::forward<U>(t));
    if (queue.size() == 1)
      cond.notify_one();
  }

  T get() {
    std::unique_lock<std::mutex> lock(mutex);
    while (queue.size() == 0)
      cond.wait(lock);
    T r = queue.front();
    queue.pop_front();
    return r;
  }

private:
  std::mutex mutex;
  std::condition_variable cond;
  std::deque<T> queue;
};

template <typename T> class channel<T, true> {
public:
  void put(const T &t) {
    std::unique_lock<std::mutex> _(put_mutex);
    {
      buf_avail.wait();
      *buf = t;
      data_avail.signal();
    }
  }

  T get() {
    std::unique_lock<std::mutex> _(get_mutex);
    {
      T r;
      buf = &r;
      buf_avail.signal();
      data_avail.wait();
      return r;
    }
  }

private:
  std::mutex put_mutex, get_mutex;
  semaphore data_avail, buf_avail;
  T *buf;
};

template <typename T> using chan = channel<T, false>;

void gen(chan<unsigned long> &ch) {
  unsigned long v = 2;
  while (true)
    ch.put(v++);
}

void filter(unsigned long p, chan<unsigned long> &in,
            chan<unsigned long> &out) {
  while (true) {
    auto v = in.get();
    if (v % p != 0)
      out.put(v);
  }
}

void primes(chan<unsigned long> &out) {
  auto c = new chan<unsigned long>;
  new std::thread(::gen, std::ref(*c));
  while (true) {
    auto p = c->get();
    out.put(p);
    auto newc = new chan<unsigned long>;
    new std::thread(filter, p, std::ref(*c), std::ref(*newc));
    c = newc;
  }
}

int main() {
  chan<unsigned long> ch;
  new std::thread(primes, std::ref(ch));
  for (int i = 0; i < 5000; i++) {
    std::cout << ch.get() << std::endl;
  }
}