P2300 proposes two algorithms similar to unifex::then, upon_done and upon_error. This PR intends to add those algorithms to libunifex with names: unifex::upon_done and unifex::upon_error. I am trying to make coding conventions similar to already existing modules in libunifex and would be happy to change if some coding conventions doesn't match coding conventions of project.

Current Progress:

• upon_done implementation and test added and tests are passing. Tests are kept similar to unifex::then tests.
• upon_error implementation and tests added and tests are passing.

Changes:

• Port third party dependency management over to cmake hunter, and fetch gtest from there from now on.
• Use cmake toolchain files for alternative compilers and STL implementations.
• Add in LLFIO and liburing as an external project.
• Make libunifex installable, and a good player for third party cmake inclusion.
• Add Ninja build support for Windows (untested, but should work)
• Coroutines detection no longer forces STL to libc++ on Linux, and now supports future GCC coroutines flags.

I am confused about when and where should I define a receiver. From the example in libunifex, I only find the definition for sender instead of receiver. I want to know in what situation should we define a receiver and how should we implement it.

I'm very interested in using libunifex and one of the first things I need support for is supporting ranges beyond the simple range_stream in examples. I noticed this issue was open and decided this was worth tackling.

So far I modified range_stream.hpp to take in an rvalue and lvalue range, and then we can grab the iterators from that range. It got tricky because I couldn't figure out how to support both taking in a single range as an argument as well as begin and end iterators using the same stream struct, but maybe this isn't necessary.

I'd like to improve this pull request as well because I have definitely overlooked something and would appreciate a review plus feedback on how I can improve this PR.

Changes made:

• Updated examples to use range_stream{std::views::iota(0, 10)} instead of range_stream{0, 10}
• Added a test using a few simple examples beyond integers
• Updated the range_stream.hpp header file to support streams and constrained the type using std::ranges::range

Adds a move-only unifex::any_object type-erasing wrapper that supports the small-object optimisation.

Also refactors some of the type-erasing helper classes from any_unique to make them more reusable.

Some questions on semantics:

• If I pass an allocator with std::allocator_arg to the constructor but the type to be constructed fits in the small-object buffer then should it use the allocator or should it still use the small-object buffer, ignoring the allocator (currently does the latter)
• Should any_object allow non-movable objects that would be heap-allocated because it's a large object (move ctor is not called on concrete type in this case)?
• If so, should any_object also store small objects that are not movable in a heap-allocation?
• Or should we just always require that wrapped types are always move-constructible? (this is the current behaviour)
• Are we ok with requiring that destructors are noexcept?
• any_object currently utilises ADL-isolation techniques but this means we can't deduce template args for this type. Is this going to break any use-cases for writing customisations of algorithms for any_object?
• Are there better ways of structuring the template args to make them more ergonomic? Currently we have any_object<InlineSize, InlineAlignment, RequireNoExceptMove, DefaultAllocator, CPOs...>
• Should the move constructor leave the source object in a valid but moved from state when moving a small object? Or should it destroy the source object after a successful move?

TODO

• [ ] Add more tests for any untested use-cases
• [ ] Write documentation for type-erasing wrappers (any_unique as well as any_object)

This PR disables gtest if one wants to build the examples but not the tests. In my case, I have issues compiling gtest with GCC 11.0.1 on Fedora 34 but this enables me to "test" the examples.

Hi,

I am trying to schedule a lot of tasks on the threadpool. An example of what I am trying to achieve is probably easier to understand:

#include <iostream>

#include <unifex/static_thread_pool.hpp>
#include <unifex/transform.hpp>
#include <unifex/await_transform.hpp>
#include <unifex/sync_wait.hpp>
#include <unifex/when_all.hpp>
#include <unifex/task.hpp>

//template <typename Scheduler, typename F>
//auto run_on(Scheduler&& s, F&& func) {
//	return unifex::transform(unifex::schedule((Scheduler&&)s), (F&&)func);
//}

int main(int argc, char** argv)
{
	unifex::static_thread_pool tp;
	auto sched = tp.get_scheduler();
	std::atomic<uint32_t> x = 0;

	auto makeTask = [&](unifex::static_thread_pool::scheduler scheduler, uint32_t a) -> unifex::task<void> {
		co_await unifex::schedule(scheduler);
		std::printf("task %i\n", a);
		std::this_thread::sleep_for(std::chrono::milliseconds{10});
		x++;

		co_return;
	};

	// Generate 1000 tasks
	std::vector<unifex::task<void>> tasks;
	tasks.reserve(1000);
	for(uint32_t i = 0; i < 1000; i++)
	{
		tasks.push_back(makeTask(sched, i));
	}


	// Schedule on the pool
	unifex::sync_wait(unifex::when_all(makeTask(sched, 1), makeTask(sched, 2)));
	//unifex::sync_wait(unifex::when_all(std::move(tasks)));

	auto start = std::chrono::system_clock::now();
	//for(auto&& t : tasks) {
	//	unifex::sync_wait(std::move(t));
	//}

	std::cout << "time elapsed: " << std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now() - start).count() << "ms" << std::endl;
	std::cout << "result: " << x << std::endl;
	return 0;
}

When using the for loop with 'sync_wait' it is still sequential (obvious). How can I achieve this in such a way that the coroutines are scheduled and executed on the threadpool parallel?

Regards, Matthijs

I use repeat_effect_until and execute algorithm to execute an infinitely repeating Sender task, but the program crashes shortly after startup, I checked the core file with GDB and found that there are millions of frames about repeat_effect_until, however when I use execute to add an infinite number of frames There is nothing wrong with the loop, I don't have any ideas to solve this problem, hope to provide some suggestions

void Controller::Start() {
  auto schedule_sender =  unifex::just_from([this]{ Schedule();});
  auto repeat_sender = unifex::repeat_effect_until(schedule_sender, [this] { return stopped_; });
  unifex::execute(main_loop_.get_scheduler(), [repeat_sender] {
    unifex::sync_wait(repeat_sender);  // use repeat_sender way
    // while (!stopped_) Schedule();  // use infinitely loop way
  });
}

When compiling the coroutine_stream_consumer example with -Werror, I observed errors such as:

    ... has accessible non-virtual destructor [-Werror=non-virtual-dtor]
    220 |         struct concrete_receiver final : next_receiver_base {

Fix these by adding default virtual destructors to all base structures which define virtual functions.

Signed-off-by: Patrick Williams [email protected]

These are the scheduler affinity changes from the unstable branch, without changes. I will work on refining these changes over the coming weeks as time allows to make them ready for trunk. I will use the review feedback from #290 as a starting point. Please add any future comments here.

bulk_via returns a ManySender that produces the results from predecessor on the execution context of specified scheduler, every result scheduled individually.

The example use case I am trying to fulfill is: we are parsing some data, which should be done sequentially, but once we have a parsed an independent chunk of data - we want to send it for processing immediately (processing can be done in parallel):

            unifex::bulk_transform(
                unifex::bulk_via(
                    thread_pool_scheduler,
                    unifex::bulk_transform(
                        unifex::bulk_schedule(thread_pool_scheduler, count),
                        [](std::size_t index) noexcept { /* Parse data sequentially*/ return handleToParsedObject; },
                        unifex::seq
                    )
                ),
                [&](std::size_t handleToParsedObject) noexcept { /* Do post-processing in parallel*/ },
                unifex::par_unseq
            )

what does the following error mean?

/usr/bin/ld: /root/.conan/data/libunifex/cci.20220430/_/_/package/5e4e6991faac82ea7d12cd1059daf88b6cacdb96/lib/libunifex.a(inplace_stop_token.cpp.o): relocation R_X86_64_32 against `.rodata' can not be used when making a PIE object; recompile with -fPIE

I have tried to do what the compiler says. But did not help

add_compile_options(-O0 -Wall -Wextra -pedantic -fconcepts -fPIE) 

Other details used Conan for dependency management

[requires]
boost/1.69.0
libunifex/cci.20220430
nlohmann_json/3.11.2

[options]
boost:shared=True

[generators]
cmake

Distributor ID: Ubuntu
Description:    Ubuntu 22.04 LTS
Release:        22.04
Codename:       jammy

• class low_latency_iocp_context
  • CPO open_file_read_only -> class low_latency_iocp_context::async_read_only_file
  • CPO open_file_write_only -> class low_latency_iocp_context::async_write_only_file
  • CPO open_file_read_write -> class low_latency_iocp_context::async_read_write_file
• class low_latency_iocp_context::async_read_only_file
  • CPO async_read_some_at
• class low_latency_iocp_context::async_write_only_file
  • CPO async_write_some_at
• class low_latency_iocp_context::async_read_write_file
  • CPO async_read_some_at
  • CPO async_write_some_at

GCC 10.2 coroutine support is not great. In particular, this version of GCC does not handle promise type constructors that accept the coroutine function parameters correctly. GCC passes a reference to the pre-copy object to the promise type constructor rather than a reference to the coroutine function argument after it's been copied to the coroutine frame. connect_awaitable relies on this working correctly, so we will require GCC 10.3 or newer to use coroutines with unifex.

Replaces #443

values construction may throw. Use UNIFEX_TRY {} UNIFEX_CATCH {} to forward current_exception to set_error.

https://github.com/facebookexperimental/libunifex/blob/99b39667e04f6f12ef1bed9a5c69312cc1a9abd3/include/unifex/let_error.hpp#L92

I think it's really bad that mutex serialize all waiters to single thread.

I think unlock should be something like this, if it async: Executor.Schedule(after unlock code); ExecuteHere(next critical section)

But we don't have async dtor :(, so for unique lock something like: Executor.Schedule(next critical section) ExecuteHere(after unlock code)

But now test like this doesn't work: https://github.com/YACLib/YACLib/blob/main/test/unit/coro/async_mutex.cpp#L273

Check Unlock in our library: https://github.com/YACLib/YACLib/blob/main/include/yaclib/coro/async_mutex.hpp