What's the recommended way to wait a list of senders?

[lixin-wei]

e.g. I have a vector of senders.

int n = <some runtime-known number>;

std::vector<MySender> senders;
for (int i = 0; i < n; ++i) {
  senders.push_back(...);
}

How to execute them in parallel and get all results? when_all doesn't work because all input senders of when_all should be known at compile-time. I can't pass a vector to when_all.

[bizehao]

use the any_sender_of

[lixin-wei]

@bizehao Hi thanks for reply, IIUC any_sender_of only allow me to define the sender vector in advance.

But how to execute them in parallel and attach a then after all of them are done? just like when_all. Could you pls show me an example?

[bizehao]

OK, tomorrow, I don't have a computer right now.

[bizehao]

template <class... Ts>
using any_sender_of =
typename exec::any_receiver_ref<stdexec::completion_signatures<Ts...>>::template any_sender<>;

using my_sender_t = any_sender_of<stdexec::set_value_t(int), stdexec::set_error_t(std::exception_ptr)>;

template<std::size_t N>
auto for_handle(std::vector<my_sender_t> ss)
{
	assert(ss.size() == N);
	return[ss = std::move(ss)] <std::size_t... I>(std::index_sequence<I...>) mutable
	{
		auto rs = stdexec::when_all(std::move(ss[I])...)
		| stdexec::let_value(
			[](auto... args)
			{
				std::vector<int> rst;
				((rst.push_back(args)), ...);
				return stdexec::just(std::move(rst));
			}
		);
		return rs;
	}(std::make_index_sequence<N>{});
}

auto main() -> int
{
	constexpr int Task_Num = 15;
	std::vector<my_sender_t> senders;
	for (int i = 0; i < Task_Num; ++i)
	{
		senders.push_back(stdexec::just(i));
	}
	auto task = for_handle<Task_Num>(std::move(senders))
		| stdexec::then([](std::vector<int> rst)
			{
				for (auto&& it : rst)
				{
					std::cout << it << ' ';
				}
				std::cout << std::endl;
			});
	stdexec::sync_wait(std::move(task));
}

the Task_Num It cannot be too large, because when_all is compiled at compile time, which will result in fatal error C1060: insufficient compiler heap space. Because it is compiled at compile time, for this task, you can use thread pools, futures, and then use them with stdexec.

[lixin-wei]

@bizehao thanks for the example, but it doesn't solve my problem. In my case, the number of tasks is only known at runtime.

Do you know how to to do this when tasks is only known at runtime?

[bizehao]

Then don't use when_all. If you can accept it, I can write an example

[lixin-wei]

Sure, would you pls show me an example?

[bizehao]

template <class... Ts>
using any_sender_of =
typename exec::any_receiver_ref<stdexec::completion_signatures<Ts...>>::template any_sender<>;

using my_sender_t = any_sender_of<stdexec::set_value_t(int), stdexec::set_error_t(std::exception_ptr), stdexec::set_stopped_t()>;

auto for_handle(std::vector<my_sender_t> ss)
{
	return stdexec::just(std::move(ss)) | stdexec::then([](std::vector<my_sender_t> ss)
		{
			std::atomic_int32_t c{ 0 };
			int count = ss.size();
			std::mutex mtx;
			std::condition_variable cv;
			std::vector<int> rst(ss.size());
			for (int i = 0; i < count; i++)
			{
				auto t = std::move(ss[i]) | stdexec::then([&rst, i, &c, &count, &cv](int v)
					{
						rst[i] = v;
						c++;
						if (c == count)
						{
							cv.notify_one();
						}
					});
				stdexec::start_detached(std::move(t));
			}
			std::unique_lock lock{ mtx };
			cv.wait(lock, [&c, &count]() { return c == count; });

			return rst;
		});
}

auto main() -> int
{
	exec::static_thread_pool pool{ 4 };
	int task_num = 100;
	std::vector<my_sender_t> senders;
	for (int i = 0; i < task_num; ++i)
	{
		auto t = stdexec::just(i) | stdexec::continues_on(pool.get_scheduler()) |
			stdexec::then([](int v) { return v * v; });
		senders.push_back(std::move(t));
	}
	auto task = for_handle(std::move(senders))
		| stdexec::then([](std::vector<int> rst)
			{
				for (auto&& it : rst)
				{
					std::cout << it << ' ';
				}
				std::cout << std::endl;
			});
	stdexec::sync_wait(std::move(task));
}

Do you think this is suitable? stdexec is very flexible and can be customized.

[lixin-wei]

Thank you so much for your time writing this. But it's not what I want. Your cv.wait will block the thread in the scheduler. It's not an real async way.

I've found someone wrting a when_all(range), looks promising: #1461

[bizehao]

It's easier to use wait, I understand what you mean, you can pass the callback in each sub-task, count it, and then set_value. I deal with business ideas. This when_all_range is indeed the correct solution