diff --git a/src/async/inc/async/deque.hpp b/src/async/inc/async/deque.hpp index 0f3f114..37609dd 100644 --- a/src/async/inc/async/deque.hpp +++ b/src/async/inc/async/deque.hpp @@ -10,7 +10,7 @@ namespace nil::async { /** * Partially specified alias when using container_base with std::deque, - * std::mutex and std::lock_guard. Prefer this over container base for deques + * std::mutex and std::lock_guard. Prefer this over container_base for deques */ template using deque = diff --git a/src/async/inc/async/task.hpp b/src/async/inc/async/task.hpp new file mode 100644 index 0000000..36de267 --- /dev/null +++ b/src/async/inc/async/task.hpp @@ -0,0 +1,81 @@ +#ifndef NIL_SRC_ASYNC_INC_ASYNC_TASK_HPP_ +#define NIL_SRC_ASYNC_INC_ASYNC_TASK_HPP_ + +#include +#include + +#include "async/optional.hpp" + +namespace nil::async { + +template +class send_pipe { + send_pipe(std::shared_ptr> pipe) + : pipe_{std::move(pipe)} {} + + template + void send(P&& p) { + pipe_->push(std::forward

(p)); + } + + private: + std::shared_ptr> pipe_; +}; + +template +class recv_pipe { + recv_pipe(std::shared_ptr> pipe) + : pipe_{std::move(pipe)} {} + + std::optional recv_opt() { pipe_->pop(); } + + Payload blocking_recv() { + while (true) { + auto r = pipe_->pop(); + if (r != boost::none) { + return r.value(); + } + } + } + + template + std::optional try_recv(std::chrono::duration timeout) { + auto start_time = std::chrono::steady_clock::now(); + while ((std::chrono::steady_clock::now() - start_time) < timeout) { + auto r = pipe_->pop(); + if (r != boost::none) { + return r.value(); + } + } + return std::nullopt; + } + + private: + std::shared_ptr> pipe_; +}; + +template +struct task_future { + task_pipe(std::future&& future, task_pipe&& task_pipe) + : future_{std::move(future)}, pipe_{std::move(task_pipe)} {} + + std::future future_; + send_pipe pipe_; +}; + +template +auto task(Func&& func) { + auto payload_ptr = std::make_shared>(std::nullopt); + + auto future = std::async( + [pipe = recv_pipe{payload_ptr}, f = std::forward(func)]() { + std::invoke(std::forward(func), pipe) + }); + + return task_future>>{ + std::move(future), send_pipe{std::move(payload_ptr)}}; +} + +} // namespace nil::async + +#endif // NIL_SRC_ASYNC_INC_ASYNC_TASK_HPP_ \ No newline at end of file diff --git a/src/async/tests/atomic_rw_test.cpp b/src/async/tests/atomic_rw_test.cpp index bdf79f3..26d64d9 100644 --- a/src/async/tests/atomic_rw_test.cpp +++ b/src/async/tests/atomic_rw_test.cpp @@ -23,13 +23,13 @@ BOOST_AUTO_TEST_CASE(ReadTest) { auto read_func_1 = [&]() { for (size_t ii{0}; ii < 100; ++ii) { - auto read_proxy = async_set.read(); - BOOST_CHECK_EQUAL(read_proxy->size(), 4); - BOOST_CHECK_EQUAL(read_proxy.value().count(4), 1u); - BOOST_CHECK_EQUAL((*read_proxy).count(2), 0u); - BOOST_CHECK(!read_proxy->empty()); - BOOST_CHECK_EQUAL(4 + 6 + 7 + 9, std::accumulate(read_proxy->cbegin(), - read_proxy->cend(), 0)); + auto atomic_r_proxy = async_set.read(); + BOOST_CHECK_EQUAL(atomic_r_proxy->size(), 4); + BOOST_CHECK_EQUAL(atomic_r_proxy.value().count(4), 1u); + BOOST_CHECK_EQUAL((*atomic_r_proxy).count(2), 0u); + BOOST_CHECK(!atomic_r_proxy->empty()); + BOOST_CHECK_EQUAL(4 + 6 + 7 + 9, std::accumulate(atomic_r_proxy->cbegin(), + atomic_r_proxy->cend(), 0)); } }; @@ -63,17 +63,17 @@ BOOST_AUTO_TEST_CASE(WriteTest) { auto read_func = [&const_async_set]() { for (size_t ii{0}; ii < 100; ++ii) { { - auto read_proxy = const_async_set.read(); - BOOST_CHECK_EQUAL(read_proxy->size(), 4); - BOOST_CHECK_EQUAL(read_proxy->count(4), 1u); - BOOST_CHECK_EQUAL(read_proxy->count(12), 0u); - BOOST_CHECK_EQUAL(read_proxy->count(13), 0u); - BOOST_CHECK_EQUAL(read_proxy->count(14), 0u); - BOOST_CHECK_EQUAL(read_proxy->count(15), 0u); - BOOST_CHECK(!read_proxy->empty()); + auto atomic_r_proxy = const_async_set.read(); + BOOST_CHECK_EQUAL(atomic_r_proxy->size(), 4); + BOOST_CHECK_EQUAL(atomic_r_proxy->count(4), 1u); + BOOST_CHECK_EQUAL(atomic_r_proxy->count(12), 0u); + BOOST_CHECK_EQUAL(atomic_r_proxy->count(13), 0u); + BOOST_CHECK_EQUAL(atomic_r_proxy->count(14), 0u); + BOOST_CHECK_EQUAL(atomic_r_proxy->count(15), 0u); + BOOST_CHECK(!atomic_r_proxy->empty()); BOOST_CHECK_EQUAL( 1 + 4 + 5 + 6, - std::accumulate(read_proxy->cbegin(), read_proxy->cend(), 0)); + std::accumulate(atomic_r_proxy->cbegin(), atomic_r_proxy->cend(), 0)); } for (size_t ii{0}; ii < 100; ++ii) { BOOST_CHECK_EQUAL(4, 4); @@ -84,18 +84,18 @@ BOOST_AUTO_TEST_CASE(WriteTest) { auto write_func = [&async_set]() { for (size_t ii{0}; ii < 100; ++ii) { { - auto write_proxy = async_set.write(); - write_proxy->insert(12); - write_proxy->insert(13); - write_proxy->insert(14); - write_proxy->insert(15); - write_proxy->erase(12); - write_proxy->erase(13); - write_proxy->erase(14); - write_proxy->erase(15); - auto cp = *write_proxy; - *write_proxy = std::set({1, 2, 3, 4, 5}); - *write_proxy = cp; + auto atomic_rw_proxy = async_set.write(); + atomic_rw_proxy->insert(12); + atomic_rw_proxy->insert(13); + atomic_rw_proxy->insert(14); + atomic_rw_proxy->insert(15); + atomic_rw_proxy->erase(12); + atomic_rw_proxy->erase(13); + atomic_rw_proxy->erase(14); + atomic_rw_proxy->erase(15); + auto cp = *atomic_rw_proxy; + *atomic_rw_proxy = std::set({1, 2, 3, 4, 5}); + *atomic_rw_proxy = cp; } for (size_t ii{0}; ii < 100; ++ii) { BOOST_CHECK_EQUAL(4, async_set->size());