Async Ranges

[dixlorenz]

Hi! I have started migrating from boost::fiber to tmc; after a few trivial helper classes I hit the first problem. I want to give std::ranges some async capabilities; first order is to consume a range in a tmc::task, pushing the values into a tmc::channel and consuming the channel as a range. Ideally, tmc::channel would implement begin()/end() (fiber::channel does that), I don't know if that's even possible. I tried having AI writing a simple range-wrapper around a chan_tok where the iterator would pull; I didn't even get that to compile because the pull would have to be co_awaited.

My second try was creating a generator (Apple's clang doesn't have it yet, my implementation is based on https://www.scs.stanford.edu/~dm/blog/c++-coroutines.html) that loops over the channel and co_yields the values. That compiled, but asserted.

template <std::ranges::input_range Range>
      requires std::movable<std::ranges::range_value_t<Range>>
   [[nodiscard]] DLCL::Generator<std::ranges::range_value_t<Range>> async(Range source)
   {
      using T      = std::ranges::range_value_t<Range>;
      auto channel = tmc::make_channel<T>();
      tmc::post(ex_cpu(), [](Range source, auto channel) mutable -> CoTask<void> {
         for (auto &&value : source)
            co_await channel.push(std::move(value));
         channel.close();
      }(std::move(source), channel));
      auto data = co_await channel.pull(); // <------ asserts here
      while (data.has_value()) {
         co_yield std::move(data.value());
         data = co_await channel.pull();
      }
   }

the assert is in channel_storage::destroy, called from aw_pull_impl::await_resume. I don't understand coroutines internals enough; AI claims "TMC’s channel.pull() awaitable is only designed to be awaited from a TMC coroutine type (tmc::task / CoTask)".

The documentation says a generator type is missing from tmc, is that what would be needed here?

[tzcnt]

Re: implementing begin() / end() as async. Unforunately, only stackful coroutines (like boost::fiber) have the ability to suspend the caller implicitly. For C++20 stackless coroutines, you will need a co_await expression, which makes it impossible to implement an async API that looks like a standard iterator.

---

Re: your range generator code. It looks good to me. You should be able to await channel.pull() from inside a custom generator.

That assert indicates that the element was never constructed, or was destructed twice. However, since your usage looks correct, I suspect a lifetime issue with the generator itself, since the awaitable exists inside the generator's coroutine frame.

Are you able to share a small, complete example program that demonstrates the issue?

One other thing to double check - the type T should be a value, not a reference.

---

A couple references:
https://github.com/lewissbaker/cppcoro?tab=readme-ov-file#generatort
https://github.com/lewissbaker/cppcoro?tab=readme-ov-file#async_generatort

Notably, a generator<T> (cannot use co_await internally) and async_generator<T> (can use co_await internally) have to be implemented, and invoked, differently.

Also note that the example syntax shown in that README does not work, as for co_await was removed from the standard. You'd have to rewrite the loop like this: lewissbaker/cppcoro@fdb7a74

---

From the caller of the generator, this won't look too different than just using co_await channel.pull() directly. The simplest path would be to have this function return the chan_tok and have the caller make use of that.

template <std::ranges::input_range Range>
      requires std::movable<std::ranges::range_value_t<Range>>
   [[nodiscard]] tmc::chan_tok<std::ranges::range_value_t<Range>> async(Range source)
   {
      using T      = std::ranges::range_value_t<Range>;
      auto channel = tmc::make_channel<T>();
      tmc::post(ex_cpu(), [](Range source, auto channel) mutable -> CoTask<void> {
         for (auto &&value : source)
            co_await channel.push(std::move(value));
         channel.close();
      }(std::move(source), channel));
      return channel;
   }

   tmc::task<void> caller() {
     std::vector<Thing> things;
     auto async_range = async(things);

     auto data = co_await async_range.pull();
     while (data.has_value()) {
       process(data.value());
       data = co_await async_range.pull();
     }
   }

[dixlorenz]

Unforunately, only stackless coroutines (like boost::fiber) have the ability to suspend the caller implicitly
I assume you mean "stackful" here.

As to the range generator: the element wrongly destroyed belongs to the aw_pull_impl; I don't think it can be my generator code somehow not initializing the internal storage. But since you say it should work it tried debugging and anytime I hit breakpoints it mostly did work. in aw_pull_impl::await_ready, if there already is data ("elem->is_data_waiting()") it does work (I can force it by sleeping before the first co_await). When it actually suspends at the co_await, it asserts in the await_resume. As you said, claiming t was never constructed and indeed, breakpoints in channel_storage don't get hit and the value that has been moved from t is (in this case) an empty string. That's as far as I can get, to me it seems it shouldn't have resumed (yet)?

This asserts almost always:

#include <coroutine>
#include <ranges>
#include <print>

#include "tmc/all_headers.hpp"

namespace TMCIssue {
   template <typename T>
   struct [[nodiscard]] Generator {
      using value_type = T;

      struct promise_type;
      using handle_type = std::coroutine_handle<promise_type>;

      struct promise_type {
         T                  value;
         std::exception_ptr exception;

         promise_type() = default;
         Generator get_return_object() noexcept { return Generator(handle_type::from_promise(*this)); }
         std::suspend_always initial_suspend() noexcept { return {}; }
         std::suspend_always final_suspend() noexcept { return {}; }
         void                unhandled_exception() { exception = std::current_exception(); }
         template <std::convertible_to<T> From> // C++20 concept
         std::suspend_always yield_value(From &&from) noexcept
         {
            value = std::forward<From>(from);
            return {};
         }
         void return_void() noexcept {}
      };

      handle_type coro_handle;

      Generator(handle_type h) noexcept : coro_handle(h) {}

      Generator(const Generator &)            = delete;
      Generator &operator=(const Generator &) = delete;

      Generator(Generator &&other) noexcept : coro_handle(std::move(other.coro_handle)), filled(other.filled)
      {
         other.coro_handle = nullptr;
      }
      Generator &operator=(Generator &&other) noexcept
      {
         if (this != &other) {
            if (coro_handle)
               coro_handle.destroy();
            coro_handle       = std::move(other.coro_handle);
            other.coro_handle = nullptr;
            filled            = other.filled;
         }
         return *this;
      }

      ~Generator()
      {
         if (coro_handle)
            coro_handle.destroy();
      }

      explicit operator bool() noexcept
      {
         fill();
         return !coro_handle.done();
      }
      T operator()() noexcept
      {
         fill();
         filled = false;
         return std::move(coro_handle.promise().value);
      }

      class iterator {
      public:
         using iterator_category = std::input_iterator_tag;
         using difference_type   = std::ptrdiff_t;
         using value_type        = T;
         using reference         = T &;
         using pointer           = T *;

         iterator() noexcept        = default;
         iterator(const iterator &) = delete;

         iterator(iterator &&_other) noexcept : coro_handle(std::exchange(_other.coro_handle, {})) {}

         iterator &operator=(iterator &&_other) noexcept
         {
            std::swap(coro_handle, _other.coro_handle);
            return *this;
         }

         ~iterator() noexcept {}

         friend bool operator==(const iterator &it, std::default_sentinel_t) noexcept
         {
            return it.coro_handle.done();
         }

         iterator &operator++() noexcept
         {
            coro_handle.resume();
            return *this;
         }
         void operator++(int) noexcept { (void)operator++(); }

         reference operator*() const noexcept { return static_cast<reference>(coro_handle.promise().value); }

      private:
         friend Generator;

         explicit iterator(handle_type _coro_handle) noexcept : coro_handle(_coro_handle) {}

         handle_type coro_handle;
      };

      iterator begin()
      {
         assert(coro_handle);
         coro_handle.resume();
         return iterator{ coro_handle };
      }

      std::default_sentinel_t end() const noexcept { return {}; }


   private:
      bool filled = false;

      void fill() noexcept
      {
         if (!filled) {
            coro_handle();
            if (coro_handle.promise().exception)
               std::rethrow_exception(coro_handle.promise().exception);
            filled = true;
         }
      }
   };

   template <std::ranges::input_range Range>
      requires std::movable<std::ranges::range_value_t<Range>>
   [[nodiscard]] Generator<std::ranges::range_value_t<Range>> async(Range source)
   {
      using T      = std::ranges::range_value_t<Range>;
      auto channel = tmc::make_channel<T>();
      tmc::post(tmc::cpu_executor(), [](Range source, auto channel) mutable -> tmc::task<void> {
         for (auto &&value : source)
            co_await channel.push(std::move(value));
         channel.close();
      }(std::move(source), channel));
      auto data = co_await channel.pull();
      while (data.has_value()) {
         co_yield std::move(data.value());
         data = co_await channel.pull();
      }
   }
} // namespace TMCIssue

TMCIssue::Generator<int> generate_ints()
{
   for (auto i = 0; i < 10; ++i)
      co_yield i;
}

int main(int /*argc*/, const char * /*argv*/[])
{
   return tmc::async_main([]() -> tmc::task<int> {
      auto vec = std::ranges::to<std::vector>(
         TMCIssue::async(std::views::iota(0) | std::views::take(4)) | std::views::transform([](int x) { return x + 1; }));
      std::println("{}", vec);
      co_return 0;
   }());
}

[tzcnt]

Yes, I meant "stackful", thanks. I've fixed my initial comment, and also updated it with some more information and potential solutions.

---

This code is a problem:

auto vec = std::ranges::to<std::vector>(
  TMCIssue::async(std::views::iota(0)
  | std::views::take(4))
  | std::views::transform([](int x) { return x + 1; })
);

There's no co_await in this expression, so it doesn't have any way for the caller to suspend and wait for the results to be ready. This would only work with a synchronous generator. Note that synchronous generators aren't even really coroutines in the traditional sense - they are pretty just fancy iterators. They are polled by the caller, and the caller can resume them as many times as they want.

What you need is an asynchronous generator. For this to work, you must have a co_await expression in the caller.

---

To give a more detailed explanation of what's happening in your current implementation: You've implemented a synchronous generator type, but then you use co_await internally without suspending the caller. I've annotated the code with suspend point numbers:

  auto data = co_await channel.pull(); // suspend point 1
  while (data.has_value()) {
    co_yield std::move(data.value()); // suspend point 2
    data = co_await channel.pull(); // suspend point 3
  }

The only place where the caller should be allowed to resume the generator is at suspend point 2. When hitting suspend point 1 and 3, the caller needs to also suspend and wait for the other end of the channel (the post() task) to do the resumption.

What you have instead is the caller is attempting to resume the generator repeatedly, at all 3 suspend points. This results in the following sequence of calls (which I verified by adding some print statements):

call	new state
1. main thread: begin()	runs and suspends at point 1
2. executor thread: post()	running...
3. main thread: operator*	retrieves garbage (uninitialized, not ready) value
4. main thread: operator++	resumes after suspend point 1
5. executor thread	produces the result, and also tries to resume the generator (at suspend point 1)

However you usually don't get that far, as the assert occurs inside of step 4, when the channel sees that it was resumed without the data being present. This stops you immediately before the executor thread is allowed to try to resume the generator again. This is good, because you'd end up with multiple simultaneous copies of the generator running.

[tzcnt]

What you're doing here doesn't make a lot of sense. You have a synchronous input (a range, which is a synchronous iterator / equivalent to a synchronous generator). Then you spawn an extra task to push the values into a channel, so you can make it awaitable. But then you want to consume it in a synchronous manner. I suggest for this use case that you just use the original range directly.

If your input is async (the original input is from some async stream like a file) and you want to consume the outputs in a synchronous manner, then you'd need to block on pulling data out of the channel. Currently I don't provide any mechanism for this, so you would need to use channel::try_pull() in a loop. But I discourage you from doing this, as it is very inefficient.

If your input is async, and you want to consume the outputs asynchronously, then just await the data source directly and do the processing inline. Or if you need to fan out data to multiple consumers, then you could return the chan_tok and let the consumers read from that, as I suggested in my original comment. You could implement an async_generator type but that would just be some syntax sugar to let you do some inline processing; it's probably not necessary to achieve your goals.

[dixlorenz]

Thanks for all the explanations; it seems what I am trying to do is just not possible.

my use case: I have a zip file containing a huge (300GB) csv file. I have a pipeline like this:

unzip_4MBData_chunks_generator | convert_into_records | chunk_by_key | compress | store_in_db

This all works with my generator and std::ranges; but it's horribly slow, what I actually need is:

async(unzip_4MBData_chunks) | convert_into_records | chunk_by_key | fork(compress) | store_in_db

convert_into_records and chunk_by_key don't need to know that what they are being fed is created asynchronously. store_in_db doesn't need to know that there are multiple compress-tasks feeding into this, fork returning a channel takes care of that.

I could (and started a few times) to build my own data pipeline engine; it's not trivial and the whole infrastructure is just replicating a lot of what std::ranges already do (without interfering), that's why I am trying this approach. I have more cases where I have huge amounts of data being processed in multiple independent stages, it makes sense to have a framework for it.

Conceptually I can resolve this, replace

data = co_await channel.pull();

with

data =
         tmc::post_waitable(
            tmc::cpu_executor(),
            [channel]() mutable -> tmc::task<std::optional<T>> { co_return co_await channel.pull(); }())
            .get();

Horrible, I'll admit. It also blocks the thread; probably have to run this on its own thread. The overhead might actually be acceptable in my use cases, it just feels there should be a much nicer solution.

[tzcnt]

So, if you want the "ranges pipeline syntax" using |, then I can't offer a solution right now, although this could be implemented along with #60. Additionally, I could implement an async generator type in the future, but for now you'd have to write a processing loop by hand.

If you want your file reads to be truly async, you can do so using tmc-asio and setting ASIO_HAS_IO_URING and ASIO_DISABLE_EPOLL (assuming you are on Linux).

Here are some ways to do this using the current facilities of TooManyCooks:

1. using a single reader, processing inline which I think matches what you were asking for: https://gist.github.com/tzcnt/f3318040e900deb78c90b13fd64a5a3a
2. using a single reader but also using double buffering to ensure that we don't stall on the file access side: https://gist.github.com/tzcnt/bfd1933b034da2761963adbb4025299c
3. fork()s a new task for each file part... this might cause your computer to go OOM: https://gist.github.com/tzcnt/716c691bcb44214cc87628a1e57c6409

These all assume that you only want a single DB writer, but of course it's easy to scale that end up by adding more db_writer tasks. If you find that you are producing data too fast, and your DB writers can't keep up, you can use a tmc::semaphore to limit the number of in-flight chunks.

[dixlorenz]

I'm still adjusting to thinking asynchronously, I am realizing ranges/iterators just isn't suitable for async processing; Channels (or something like it) have to be the base for an async pipeline. I'll start off with the async that returns the channel. "transform" creates a task that iterates over it, forking a task for each item and writing those into a channel, returning that channel; instantly giving me concurrent stages and item-processing. And so on, treating "async" as the normal case instead of mixing async operations into a synchronous workflow.
Thanks for all the help and explanations.

[tzcnt]

I've some ideas on how to provide facilities to make building this kind of pipeline easier, but an implementation will have to wait until after v1.3. The following is just a scratchpad:

---

Method 1: Using channel to send messages and semaphore for backpressure. This makes it easy to have different levels of parallelism at each stage of the pipeline. Implementing with | operator would look cool, but probably make my life a lot harder, so I might go with a single class and a variadic constructor.

FileChunkReader myReader("filename.csv");
tmc::pipeline p{
  tmc::pipeline_start{FileChunkReader::get_next_chunk, myReader},
  tmc::pipeline_func{convert_into_records},
  tmc::pipeline_func{chunk_by_key},
  tmc::pipeline_func{compress}.with_concurrency(tmc::cpu_executor().thread_count()),
  tmc::pipeline_end{db_writer}.with_concurrency(4),
}

There could be a .with_backpressure(size_t) function for each, which would limit the number of allowed waiting inputs to each stage, by suspending the prior stage. This just initializes the value of the semaphore for that stage. I think a reasonable default value for this would be current_stage.concurrency + 1.

Each stage could be implemented as a number of parallel copies of:

tmc::task<void> pipeline_stage(tmc::channel<Input> inChan, tmc::channel<Output> outChan, tmc::semaphore& inSem, tmc::semaphore& outSem, std::function<Output(Input)> process) {
  auto input = co_await inChan.pull();
  while (input.has_value()) {
    inSem.release();
    co_await outSem;
    co_await outChan.push(process(input.value));
    input = co_await inChan.pull();
  }
}

Rather than requiring a semaphore, it would be good to offer a bounded queue option (it would disable post(), or potentially make it blocking?)

Also a specialization pipeline_coro should be provided where the processing function is a coroutine. Some other considerations would be how to setup the input and output:

• input might need to be stateful (a class member, like in the example), or it could be a channel, or an iterator
• output could be void, or a channel, or collected into a vector
• if a stage needs additional static parameters, they could be provided to the constructor (like the myReader "this" pointer in the example)

---

Method 2: Directly invoking each stage - a replacement for #60. That issue originally was designed to let you attach a continuation to task after it was created, but this would need to happen before; letting you create a "pipeline template" which could be called any number of times with new inputs. This would not have any channels, but parallelism would be possible by a call to tmc::spawn_many in the implementation. However, this would require the prior stage to emit a std::vector or std::array.

tmc::pipeline_direct p{
  tmc::pipeline_task{load_file},
  tmc::pipeline_func{file_to_chunks},
  tmc::pipeline_func{decompress} // automatically concurrent if file_to_chunks returns an array
};
std::vector<DecompressedChunks> = co_await p.run("myfile.csv");

This should desugar into a series of regular calls in a single coroutine:

co_await [](std::string input){
  auto a = co_await load_file(input);
  auto b = file_to_chunks(a);
  auto c = co_await tmc::spawn_func_many(
    std::ranges::views::transform(b, [](Chunk& c) -> auto {
      return [&c]() { return decompress(c); };
    })
  );
  co_return c;
}("myfile.csv");

Implementing this desugaring depends on the ability to emit co_awaits while doing the compile-time tuple pack indexing. Also it would be really nice if HALO worked here, to make this equally as performant as the hand-coded version.

---

Method 3: An ideal version would offers the benefits of both of the prior approaches:

• backpressure / concurrency per pipeline stage
• directly invoked stages (no channel overhead)
• HALO so everything fits into a single, static allocation

A single-producer / single-consumer stage should be able to be pulled directly. Double / multiple buffering could be implemented like in https://gist.github.com/tzcnt/bfd1933b034da2761963adbb4025299c, but a new data structure would be needed in order to allow multiplexing M outputs -> N inputs efficiently. This could be a fixed size array with N slots. An atomic bitmap could be used to acquire/release slots. Would also need storage for a suspended consumer (or producer).

To avoid needing to move data multiple times (like channel does), the producer could store the value directly in the array, and consumer could use the value by reference, then free it afterward. If done this way, the array should probably have 2N slots, and we would need multiple states in the bitmap - empty, data ready, data in use. This design is similar to channel, but should be more efficient.

I wonder if any design elements of work contracts https://www.youtube.com/watch?v=oj-_vpZNMVw could be used to make this multiplexer more efficient?

[dixlorenz]

Method 2 is a no go for me. a) it requires the file to fit into Memory. b) it only starts processing once the complete file is loaded.

Method 3 I don't fully understand, it seems to concentrate on reducing the overhead of using channels everywhere?

Method 1 is what I am working on. Syntactic sugar aside (pipe operator, using variadic function/class...), backpressure and channel overhead seem to be the main issues there. You mentioned a bounded queue; indeed I had already thought about opening an issue about that :-)

Maybe it is way to simplistic, but what I am thinking is there are 2 cases:

a) the producer is faster than the consumer(s). What's the point of having multiple items waiting in a channel? They just take up memory, channel size = 1 should be sufficient. One could argue that different items can take different amounts of time to process so that sometimes consumption is faster and then there should be another item waiting already. At most there need to be cpu_count slots, in practice probably less.

b) the consumer(s) is/are faster than the producer: the channel would never fill up to more than 1 item.

A bounded_channel of static size N (implemented using a ring buffer?) that suspends on push would nicely solve the problem of the initial producer being too fast and should be more efficient than the unbounded channel. In this use case a "channel" of size 1 might even be sufficient and could be even more efficient, but that might be too specialized.

A transformer that runs multiple items concurrently would indeed have to use a semaphore; the sub task uses acquire_scope and it cannot return it without being able to push to the bounded_channel.

It almost feels too easy, but I think that solves the backpressure problem und negates the need for the complexity of method 3. Am I missing something?

What I currently have (no sugar, proof of concept):

a) The initial source
auto async(Range source) -> tmc::chan_tok<std::ranges::range_value_t>
(code above, but returning the channel). The intent is for a generator, but it takes any range so it could accept views::iota(0) | take(10), starting a parallel for-loop; take a vector of items to process in parallel...

b) transform

   template <typename T, typename F>
   auto transform(Channel<T> source, F f)
   {
      using Result = std::invoke_result_t<F, T>;
      auto target  = make_channel<CoTask<Result>>();
      tmc::post(ex_cpu(), [](auto source, auto target, auto f) mutable -> tmc::task<void> {
         auto value = co_await source.pull();
         while (value) {
            auto task = [](F f, T value) mutable -> tmc::task<Result> { co_return f(std::move(value)); }(
                                                    f, std::move(*value));
            co_await target.push(std::move(task));
            value = co_await source.pull();
         }

         target.close();
      }(source, target, f));

      return target;
   }

This works, but the sub-tasks aren't started immediately. What I really want is

co_await target.push(tmc::spawn(std::move(task)).fork());

but that won't compile and as far as I can tell from documentation, it is impossible to move the result from fork() into a channel?

I already had a very simple equivalent of a shared_future:

template <typename T>
class SharedTask {
   struct State {
      tmc::manual_reset_event ready_event;
      std::optional<T>        value;

      void set_value(T &&val)
      {
         value = std::move(val);
         ready_event.set();
      }

      T get_value() const
      {
         assert(value.has_value());
         return value.value();
      }
   };

   std::shared_ptr<State> mState;

public:
   SharedTask() = default;
   template <typename Executor>
   SharedTask(Executor &&executor, tmc::task<T> task, size_t priority = 0, size_t threadhint = NO_HINT)
      : mState(std::make_shared<State>())
   {
      tmc::task<void> tp = [](std::shared_ptr<State> state, tmc::task<T> task) -> tmc::task<void> {
         state->set_value(co_await std::move(task));
      }(mState, std::move(task).resume_on(executor));
      tmc::post(executor, std::move(tp), priority, threadhint);
   }

   bool is_ready() const noexcept { return mState && mState->ready_event.is_set(); }

   auto operator co_await() const noexcept
   {
      struct Awaiter {
         using event_awaiter = decltype(std::declval<tmc::manual_reset_event &>().operator co_await());

         std::shared_ptr<State> state;
         event_awaiter          inner;

         bool await_ready() const noexcept
         {
            assert(state);
            return state->ready_event.is_set();
         }

         void await_suspend(std::coroutine_handle<> continuation) noexcept
         {
            inner.await_suspend(std::move(continuation));
         }

         T await_resume()
         {
            inner.await_resume();
            return state->get_value();
         }
      };

      return Awaiter{ mState, mState->ready_event.operator co_await() };
   }
};

template <class Executor, class T>
auto share(Executor &&executor, tmc::task<T> task, size_t priority = 0, size_t threadhint = NO_HINT)
{
   return SharedTask(executor, std::move(task), priority, threadhint);
}

so I can just use

co_await target.push(share(std::move(task)));

It changes the type of the output channel and needs to be adapted as ForkedTask/fork to enable non-copyable items, but that would work. It's not really nice, maybe tmc offers a better way to pass around forked tasks?

As we said, transform needs to use a semaphore internally.

c) await
take a channel<task> and return a channel.

   template <typename T>
   auto await(Channel<tmc::task<T>> source)
   {
      auto target = make_channel<T>();
      tmc::post(ex_cpu(), [](auto source, auto target) mutable -> tmc::task<void> {
         auto value = co_await source.pull();
         while (value) {
            co_await target.push(co_await std::move(*value));
            value = co_await source.pull();
         }
         target.close();
      }(source, target));
      return target;
   }

d) the end of the pipeline.
I don't have anything yet; name could be "join"? It should take a function; if that function returns void it will just consume the channel and return tmc::task. if the return type is not void, a task<vector>. Doesn't seem to be a problem.

There is probably quite a bit of unnecessary overhead; for the heavy processing in my use case it hopefully doesn't matter. So far this approach seems promising to me: IMHO something like tmc::channel (but bounded) is absolutely needed to connect the different stages of a pipeline anyway and instead of having a rigid framework for a pipeline-object which has to be constructed in certain ways with special types (looking at std::ranges) , a "pipeline" is just couple of channels connected by functions, the only requirement for expansions being that the function takes a channel and maybe returns a channel.

Two things I noticed:

auto value = co_await source.pull();
while (value.has_value()) {
   // use value
   value = co_await source.pull();
}

looks so ugly, but the far nicer

while (auto value = co_await source.pull()) {
   // use value
}

crashes for me, no idea why. If that is a technical limitation, I think that should be in the documentation.

I had known/read about problems with using lambda captures and coroutines; I assumed it was only about references, capture-by-value would be ok. Judging by all the crashes I got, using lambdas with coroutines you should never ever capture anything. Although not specific to TMC, a warning in the documentation would be nice.

[tzcnt]

Don't mind my musings, you should definitely just use Method 1 (channel with backpressure) for now.

Re: backpressure. Say your producer can produce an item in 1 second, and a consumer takes 5 seconds to process it.
So you spin up 1 producer and 5 consumers, which should be a balanced configuration.
Due to other tasks running on the executor, the consumers end up being scheduled so that they finish at nearly the same time. You can have a batch of consumers finishing at once:

time	event
0.0s	consumer 0 finishes, takes the 1 item from the channel, and wakes the producer
0.1s	consumer 1 finishes. channel is empty. consumer suspends
0.2s	consumer 2 finishes. channel is empty. consumer suspends
0.3s	consumer 3 finishes. channel is empty. consumer suspends
0.4s	consumer 4 finishes. channel is empty. consumer suspends
1.0s	producer finishes, waking consumer 1
2.0s	producer finishes, waking consumer 2
3.0s	producer finishes, waking consumer 3
4.0s	producer finishes, waking consumer 4

This introduces a lot of extra latency. For this reason, I suggest having at least 1 channel capacity per consumer.
This buffers any latency in between producers and consumers to keep the pipeline flowing.

Re: the performance of channel w/ semaphore vs. a bounded channel. I'll make a bounded channel eventually, but
I don't think it'll make much difference to the performance of your use case.

---

I discourage you from pushing tasks through the channel. Just push data through the channel, and spin up as many workers as you need for each parallel pipeline stage. This also means that you don't need separate transform and await stages. If your individual data elements are large, then just push std::unique_ptrs through the channel.

Here is a complete implementation; please just use this for now... It's not perfect and I had to reach into the detail namespace at one place (this is more a fault of the stdlib - there's no standard way to get the type of an awaitable), so this has been a learning experience in trying to build something complicated on top of the library :)

It has specializations for pipeline stages that are either functions or coroutines themselves.

>> https://gist.github.com/tzcnt/2f50fb953dc9ee737571e422299b1964 <<

---

Thanks for sharing the while (auto value = co_await source.pull()) syntax! This is much better than what I was using. I'm embarassed to say I didn't know about this form. I've updated the main library #152 and examples tzcnt/tmc-examples#81 to use this syntax. All the examples and tests are working, so I don't think there's any inherent issue with this syntax.

I learned one other thing while implementing this - it was possible to convert a tmc::task<> to a boolean implicitly (it exposes operator bool because std::coroutine_handle does) but that's actually very bad in this case - because I could actually pass a coroutine into pipeline_transform_func and it will silently convert to bool before passing the coroutine handle into the channel as if it were a number... I've fixed this in #153 by making the operator explicit

---

Yes, you can't capture anything at all in a lambda coroutine. Although this is a defect in the standard, I should probably add a section "Coroutine Basics" which covers this.

One other thing - if you declare a function as a coroutine (return type is a task type), but the body of the function doesn't contain a co_await or co_return statement, your program will crash unexpectedly. Your compiler should issue a warning in this case, but it really should be an error.

[tzcnt]

One thing that may cause issues is that tmc::semaphore currently wakes waiters in LIFO (rather than FIFO) order.
If tmc::semaphore used FIFO waking, then you'd still get outputs out-of-order due to random jitter in parallel
pipeline stages, but it's more likely that they would be roughly in order. But with the current
LIFO waking, you're going to get outputs that are very out of order.
I think this was an oversight in my design, and I think I'll switch these wait structures to use FIFO order in the future,
perhaps with an optional toggle.

For now, if you really need "roughly FIFO" processing you can do it by not taking data from the input channel until after the output semaphore has been acquired. This cuts the performance on the benchmark significantly, but I was able to regain that performance by doubling the size of the semaphore (which allows for "double buffering").

See the changes to worker_func, worker_coro, and the outSem_ initializer vs. the prior gist here:
>> https://gist.github.com/tzcnt/c5313a8588ab84e0d712520d2ce18410 <<

[dixlorenz]

Glad I could contribute a bit with the syntax; I changed it in my code too, no problems. I guess I was still using lambda captures.

"Coroutine Basics" would be a welcome addition to the docs; problems are hard to debug and coroutines have a lot of pitfalls. AI has absolutely no clue about it either and even less about TMC ;-)

Are you sure the semaphore in your code does what it is supposed to do?. It does constrain the number of workers concurrently pulling the next value to 2 * workerCount, but there are only workerCount workers anyway. And the pulling itself is not the problem; there shouldn't be more than workerCount functions running at the same time (oversubscribing the CPU; using too much memory), that is already ensured by only having workerCount workers. And a channel shouldn't fill up unchecked when consumers are too slow (using too much memory), the semaphore doesn't constrain that. It would have to acquire on push and release on the pull. Except that the 2 operations happen in different stages which can't share a semaphore. That's where a bounded_channel would be awesome; not because of efficiency, but it makes it really easy to keep a fast producer in check ;-)

The biggest problem for me is ordering, though. Some stages (compress, write_to_db) I don't care about the order; but "chunk_by_id" relies on the incoming data being exactly in the right order. Same in some other use cases. The easiest way I saw to ensure that was passing the tasks around, it seems to nicely model the intent. With a bounded_channel as target it would also solve the number of concurrently running tasks and a channel growing too large.

I've come up with this for an ordered transform; borrowing your idea of spawning workerCount tasks; but instead of them pulling in random order (which can never result in the correct order on the other end) I have the stage pulling the items, enumerating them, using another channel to pass that to the workers which use an atomic_condvar to wait for their turn to push the result out:

   template <typename T, typename F>
   [[nodiscard]] auto transform_data(
      Channel<T> source, F f, std::uint16_t concurrency = std::hardware_concurrency())
   {
      using Result = std::invoke_result_t<F, T>;
      auto target  = make_channel<Result>();
      tmc::post(ex_cpu(), [](auto source, auto target, auto f, auto concurrency) mutable -> tmc::task<void> {
         auto          feed = make_channel<std::pair<std::uint64_t, T>>();
         std::uint64_t counter{ 0 };
         auto          expected = std::make_shared<tmc::atomic_condvar<std::uint64_t>>(0);

         auto sg = tmc::spawn_group();
         for (size_t i = 0; i < concurrency; ++i) {
            auto worker = [](auto feed, auto target, auto expected, auto f) -> tmc::task<void> {
               while (auto index_value = co_await feed.pull()) {
                  auto result = f(std::move((*index_value).second));

                  for (;;) {
                     auto observed = expected->ref().load(std::memory_order_acquire);
                     if (observed == (*index_value).first)
                        break;
                     co_await expected->await(observed);
                  }

                  co_await target.push(std::move(result));
                  expected->ref().fetch_add(1, std::memory_order_release);
                  co_await expected->co_notify_all();
               }
            }(feed, target, expected, f);
            sg.add(std::move(worker));
         }

         tmc::spawn([](auto SG, auto target) -> tmc::task<void> {
            co_await std::move(SG);
            co_await target.drain();
         }(std::move(sg), target))
            .detach();

         while (auto value = co_await source.pull())
            co_await feed.push(std::make_pair(counter++, std::move(*value)));

         feed.close();
      }(source, target, f, concurrency));

      return target;
   }

I'm not entirely happy with it; compared to just passing around a forked task in a channel it uses another channel and a notify_all for each item. And there is an obvious problem: a really slow work item would suspend all the other worker threads from pushing their results and processing the next one. If the next stage needs to process in order, that stage would suspend anyway, not a problem. If the order is not important, that is bad.

So transform probably needs 2 different implementations, depending on if the ordering is important or not.

[tzcnt]

The semaphore is a proxy for channel capacity, which allows us to create a bounded channel.
So if we have 10 consumer workers, we initialize a semaphore to 20. Each producer needs to acquire (subtract 1) from the semaphore before it's allowed to produce to the channel. When the semaphore hits 0, 20 items have been produced to the channel. It is full, and producers will suspend. If a consumer then pulls from the channel, it releases (adds 1) the semaphore. An item has been removed from the channel and the capacity is now 1.

This is what you said - "It would have to acquire on push and release on the pull." but your next statement was incorrect "the 2 operations happen in different stages which can't share a semaphore": The output semaphore of 1 stage and the input semaphore of the next stage refer to the same object, so that the producer acquires -> consumer releases pattern works. (the input semaphore is a pointer to the output semaphore of the previous stage)

However I did realize that I had 1 bug in the implementation from last night, with the initial size of the semaphore. I was setting the size of the output semaphore based on the number of workers in the current stage, but I should have been setting the size of the input semaphore. I don't want to do gist-based development any more, so I've turned this into a proper example, and you can see the fix commit here: tzcnt/tmc-examples@899fad7

[tzcnt]

Re: ordering

I finally get what you mean by forked task in a channel, and how it solves the problem of maintaining exact ordering.

As usual there is some weird corner of the language to get this done. https://quuxplusone.github.io/blog/2018/05/17/super-elider-round-2/ lets you construct the non-movable object from the factory function into anything that emplaces. Such as std::optional::emplace, or with std::make_unique. This will let you pass a forked task through a channel.

template <class F> class with_result_of_t {
  F&& fun;

public:
  using T = decltype(std::declval<F&&>()());
  explicit with_result_of_t(F&& f) : fun(std::forward<F>(f)) {}
  operator T() { return fun(); }
};

template <class F> inline with_result_of_t<F> with_result_of(F&& f) {
  return with_result_of_t<F>(std::forward<F>(f));
}



auto chan =
  tmc::make_channel<std::unique_ptr<tmc::aw_spawn_fork<tmc::task<int>>>>();

auto in =
  std::make_unique<tmc::aw_spawn_fork<tmc::task<int>>>(with_result_of([]() {
    return tmc::spawn([]() -> tmc::task<int> { co_return 5; }()).fork();
  }));

chan.post(std::move(in));

std::optional<std::unique_ptr<tmc::aw_spawn_fork<tmc::task<int>>>> out =
  co_await chan.pull();

int a = co_await std::move(**out);

[dixlorenz]

No idea why I didn't see inSem and outSem; it was pretty much was I was looking for. Being able to move a started task is nice; I'll keep it in mind. The syntax isn't too nice and it's yet another type to handle...

I took all the ideas and thoughts here and started implementing what best fits my use case. So far it's going well, thanks for all the help and explanations.

[tzcnt]

Great! I'm going to convert this to a discussion. Feel free to ask any additional questions there.

I have some ideas in mind as to how I can make this easier to implement, which I've created enhancement tickets for. I'll ping you on the discussion when those tickets have been closed.

• Zero-copy channel #154 so you don't need to wrap the forked task in a unique_ptr
• Type traits for users to implement generic code for awaitable / non awaitable #156 so you can write a single handler for both values and awaitables