pipeline.cpp

#include <experimental/executor>
#include <condition_variable>
#include <memory>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>

using std::experimental::execution_context;
using std::experimental::executor_binder;
using std::experimental::get_associated_executor;
using std::experimental::post;
using std::experimental::system_executor;
using std::experimental::use_future;
using std::experimental::use_service;

// An executor that launches a new thread for each function submitted to it.
// This class satisfies the Executor requirements.
class thread_executor
{
private:
  // Service to track all threads started through a thread_executor.
  class thread_bag : public execution_context::service
  {
  public:
    typedef thread_bag key_type;

    explicit thread_bag(execution_context& ctx)
      : execution_context::service(ctx)
    {
    }

    void add_thread(std::thread&& t)
    {
      std::unique_lock<std::mutex> lock(mutex_);
      threads_.push_back(std::move(t));
    }

  private:
    void shutdown() noexcept override
    {
      for (auto& t : threads_)
        t.join();
    }

    std::mutex mutex_;
    std::vector<std::thread> threads_;
  };

public:
  execution_context& context() noexcept
  {
    return system_executor().context();
  }

  void on_work_started() noexcept
  {
    // This executor doesn't count work.
  }

  void on_work_finished() noexcept
  {
    // This executor doesn't count work.
  }

  template <class Func, class Alloc>
  void dispatch(Func&& f, const Alloc& a)
  {
    post(std::forward<Func>(f), a);
  }

  template <class Func, class Alloc>
  void post(Func f, const Alloc&)
  {
    thread_bag& bag = use_service<thread_bag>(context());
    bag.add_thread(std::thread(std::move(f)));
  }

  template <class Func, class Alloc>
  void defer(Func&& f, const Alloc& a)
  {
    post(std::forward<Func>(f), a);
  }

  friend bool operator==(const thread_executor&, const thread_executor&) noexcept
  {
    return true;
  }

  friend bool operator!=(const thread_executor&, const thread_executor&) noexcept
  {
    return false;
  }
};

// Base class for all thread-safe queue implementations.
class queue_impl_base
{
  template <class> friend class queue_front;
  template <class> friend class queue_back;
  std::mutex mutex_;
  std::condition_variable condition_;
  bool stop_ = false;
};

// Underlying implementation of a thread-safe queue, shared between the
// queue_front and queue_back classes.
template <class T>
class queue_impl : public queue_impl_base
{
  template <class> friend class queue_front;
  template <class> friend class queue_back;
  std::queue<T> queue_;
};

// The front end of a queue between consecutive pipeline stages.
template <class T>
class queue_front
{
public:
  typedef T value_type;

  explicit queue_front(std::shared_ptr<queue_impl<T>> impl)
    : impl_(impl)
  {
  }

  void push(T t)
  {
    std::unique_lock<std::mutex> lock(impl_->mutex_);
    impl_->queue_.push(std::move(t));
    impl_->condition_.notify_one();
  }

  void stop()
  {
    std::unique_lock<std::mutex> lock(impl_->mutex_);
    impl_->stop_ = true;
    impl_->condition_.notify_one();
  }

private:
  std::shared_ptr<queue_impl<T>> impl_;
};

// The back end of a queue between consecutive pipeline stages.
template <class T>
class queue_back
{
public:
  typedef T value_type;

  explicit queue_back(std::shared_ptr<queue_impl<T>> impl)
    : impl_(impl)
  {
  }

  bool pop(T& t)
  {
    std::unique_lock<std::mutex> lock(impl_->mutex_);
    while (impl_->queue_.empty() && !impl_->stop_)
      impl_->condition_.wait(lock);
    if (!impl_->queue_.empty())
    {
      t = impl_->queue_.front();
      impl_->queue_.pop();
      return true;
    }
    return false;
  }

private:
  std::shared_ptr<queue_impl<T>> impl_;
};

// Helper to determine the argument type of a function.
template <class F>
struct argument_type
{
  typedef typename std::reference_wrapper<F>::argument_type type;
};

template <class F, class E>
struct argument_type<executor_binder<F, E>>
{
  typedef typename std::reference_wrapper<F>::argument_type type;
};

// Helper to determine the second argument type of a function.
template <class F>
struct second_argument_type
{
  typedef typename std::reference_wrapper<F>::second_argument_type type;
};

template <class F, class E>
struct second_argument_type<executor_binder<F, E>>
{
  typedef typename std::reference_wrapper<F>::second_argument_type type;
};

// Launch the last stage in a pipeline.
template <class T, class F>
std::future<void> pipeline(queue_back<T> in, F f)
{
  // Get the function's associated executor, defaulting to thread_executor.
  auto ex = get_associated_executor(f, thread_executor());

  // Run the function, and as we're the last stage return a future so that the
  // caller can wait for the pipeline to finish.
  return post(ex, use_future([in, f = std::move(f)]() mutable { f(in); }));
}

// Launch an intermediate stage in a pipeline.
template <class T, class F, class... Tail>
std::future<void> pipeline(queue_back<T> in, F f, Tail... t)
{
  // Determine the output queue type.
  typedef typename second_argument_type<F>::type::value_type output_value_type;

  // Create the output queue and its implementation.
  auto out_impl = std::make_shared<queue_impl<output_value_type>>();
  queue_front<output_value_type> out(out_impl);
  queue_back<output_value_type> next_in(out_impl);

  // Get the function's associated executor, defaulting to thread_executor.
  auto ex = get_associated_executor(f, thread_executor());

  // Run the function.
  post(ex, [in, out, f = std::move(f)]() mutable
      {
        f(in, out);
        out.stop();
      });

  // Launch the rest of the pipeline.
  return pipeline(next_in, std::move(t)...);
}

// Launch the first stage in a pipeline.
template <class F, class... Tail>
std::future<void> pipeline(F f, Tail... t)
{
  // Determine the output queue type.
  typedef typename argument_type<F>::type::value_type output_value_type;

  // Create the output queue and its implementation.
  auto out_impl = std::make_shared<queue_impl<output_value_type>>();
  queue_front<output_value_type> out(out_impl);
  queue_back<output_value_type> next_in(out_impl);

  // Get the function's associated executor, defaulting to thread_executor.
  auto ex = get_associated_executor(f, thread_executor());

  // Run the function.
  post(ex, [out, f = std::move(f)]() mutable
      {
        f(out);
        out.stop();
      });

  // Launch the rest of the pipeline.
  return pipeline(next_in, std::move(t)...);
}

//------------------------------------------------------------------------------

#include <experimental/thread_pool>
#include <iostream>
#include <string>

using std::experimental::bind_executor;
using std::experimental::thread_pool;

void reader(queue_front<std::string> out)
{
  std::string line;
  while (std::getline(std::cin, line))
    out.push(line);
}

void filter(queue_back<std::string> in, queue_front<std::string> out)
{
  std::string line;
  while (in.pop(line))
    if (line.length() > 5)
      out.push(line);
}

void upper(queue_back<std::string> in, queue_front<std::string> out)
{
  std::string line;
  while (in.pop(line))
  {
    std::string new_line;
    for (char c : line)
      new_line.push_back(std::toupper(c));
    out.push(new_line);
  }
}

void writer(queue_back<std::string> in)
{
  std::size_t count = 0;
  std::string line;
  while (in.pop(line))
    std::cout << count++ << ": " << line << std::endl;
}

int main()
{
  thread_pool pool;

  auto f = pipeline(reader, filter, bind_executor(pool, upper), writer);
  f.wait();
}