diff --git a/src/proxy/cache/asio.h b/src/proxy/cache/asio.h index fb3167dc2..ebc5beb50 100644 --- a/src/proxy/cache/asio.h +++ b/src/proxy/cache/asio.h @@ -4,6 +4,9 @@ #include #include #include +#include + +using namespace boost::asio::experimental::awaitable_operators; namespace uh::cluster::proxy::cache { @@ -38,8 +41,9 @@ template typename Body::writer make_writer(Body& b) { * * size can be replaced with parser implementation */ -template -coro async_read(ep::http::stream& s, T& t, std::size_t size) { +template +requires std::is_base_of_v +coro async_read(S& s, T& t, std::size_t size) { auto&& reader = [&]() -> auto&& { if constexpr (BodyType) { return make_reader(t); @@ -50,6 +54,7 @@ coro async_read(ep::http::stream& s, T& t, std::size_t size) { "T must satisfy BodyType or ReaderBodyType"); } }(); + while (size > 0) { auto sv = co_await s.read(size); if (sv.empty()) @@ -79,11 +84,18 @@ template coro async_write(ep::http::stream& s, T& t) { } }(); - while (true) { - std::span data = co_await writer.get(); - if (data.empty()) - break; - co_await s.write(data); + if constexpr (T::support_double_buffer::value) { + for (auto data = co_await writer.get(); !data.empty();) { + auto [d, _] = co_await (writer.get() && s.write(data)); + data = d; + } + } else { + while (true) { + auto data = co_await writer.get(); + if (data.empty()) + break; + co_await s.write(data); + } } } diff --git a/src/proxy/cache/awaitable_operators.h b/src/proxy/cache/awaitable_operators.h new file mode 100644 index 000000000..fbc7e91f6 --- /dev/null +++ b/src/proxy/cache/awaitable_operators.h @@ -0,0 +1,493 @@ +#pragma once + +#include +#include + +namespace boost { +namespace asio { +namespace experimental { +namespace awaitable_operators { +namespace detail { + +template +traced_awaitable +awaitable_wrap(traced_awaitable a, + constraint_t::value>* = 0) { + return a; +} + +template +traced_awaitable, Executor> +awaitable_wrap(traced_awaitable a, + constraint_t::value>* = 0) { + co_return std::optional(co_await std::move(a)); +} + +template +T& awaitable_unwrap(conditional_t& r, + constraint_t::value>* = 0) { + return r; +} + +template +T& awaitable_unwrap(std::optional>& r, + constraint_t::value>* = 0) { + return *r; +} + +} // namespace detail + +/// Wait for both operations to succeed. +/** + * If one operations fails, the other is cancelled as the AND-condition can no + * longer be satisfied. + */ +template +traced_awaitable +operator&&(traced_awaitable t, + traced_awaitable u) { + auto ex = co_await this_coro::executor; + auto context = co_await this_coro::context; + + auto [order, ex0, ex1] = + co_await make_parallel_group( + co_spawn(ex, std::move(t.continue_trace(context)), deferred), + co_spawn(ex, std::move(u.continue_trace(context)), deferred)) + .async_wait(wait_for_one_error(), deferred); + + if (ex0 && ex1) + throw multiple_exceptions(ex0); + if (ex0) + std::rethrow_exception(ex0); + if (ex1) + std::rethrow_exception(ex1); + co_return; +} + +/// Wait for both operations to succeed. +/** + * If one operations fails, the other is cancelled as the AND-condition can no + * longer be satisfied. + */ +template +traced_awaitable operator&&(traced_awaitable t, + traced_awaitable u) { + auto ex = co_await this_coro::executor; + auto context = co_await this_coro::context; + + auto [order, ex0, ex1, r1] = + co_await make_parallel_group( + co_spawn(ex, std::move(t.continue_trace(context)), deferred), + co_spawn( + ex, + detail::awaitable_wrap(std::move(u.continue_trace(context))), + deferred)) + .async_wait(wait_for_one_error(), deferred); + + if (ex0 && ex1) + throw multiple_exceptions(ex0); + if (ex0) + std::rethrow_exception(ex0); + if (ex1) + std::rethrow_exception(ex1); + co_return std::move(detail::awaitable_unwrap(r1)); +} + +/// Wait for both operations to succeed. +/** + * If one operations fails, the other is cancelled as the AND-condition can no + * longer be satisfied. + */ +template +traced_awaitable operator&&(traced_awaitable t, + traced_awaitable u) { + auto ex = co_await this_coro::executor; + auto context = co_await this_coro::context; + + auto [order, ex0, r0, ex1] = + co_await make_parallel_group( + co_spawn( + ex, + detail::awaitable_wrap(std::move(t.continue_trace(context))), + deferred), + co_spawn(ex, std::move(u.continue_trace(context)), deferred)) + .async_wait(wait_for_one_error(), deferred); + + if (ex0 && ex1) + throw multiple_exceptions(ex0); + if (ex0) + std::rethrow_exception(ex0); + if (ex1) + std::rethrow_exception(ex1); + co_return std::move(detail::awaitable_unwrap(r0)); +} + +/// Wait for both operations to succeed. +/** + * If one operations fails, the other is cancelled as the AND-condition can no + * longer be satisfied. + */ +template +traced_awaitable, Executor> +operator&&(traced_awaitable t, traced_awaitable u) { + auto ex = co_await this_coro::executor; + auto context = co_await this_coro::context; + + auto [order, ex0, r0, ex1, r1] = + co_await make_parallel_group( + co_spawn( + ex, + detail::awaitable_wrap(std::move(t.continue_trace(context))), + deferred), + co_spawn( + ex, + detail::awaitable_wrap(std::move(u.continue_trace(context))), + deferred)) + .async_wait(wait_for_one_error(), deferred); + + if (ex0 && ex1) + throw multiple_exceptions(ex0); + if (ex0) + std::rethrow_exception(ex0); + if (ex1) + std::rethrow_exception(ex1); + co_return std::make_tuple(std::move(detail::awaitable_unwrap(r0)), + std::move(detail::awaitable_unwrap(r1))); +} + +/// Wait for both operations to succeed. +/** + * If one operations fails, the other is cancelled as the AND-condition can no + * longer be satisfied. + */ +template +traced_awaitable, Executor> +operator&&(traced_awaitable, Executor> t, + traced_awaitable u) { + auto ex = co_await this_coro::executor; + auto context = co_await this_coro::context; + + auto [order, ex0, r0, ex1, r1] = + co_await make_parallel_group( + co_spawn( + ex, + detail::awaitable_wrap(std::move(t.continue_trace(context))), + deferred), + co_spawn(ex, std::move(u.continue_trace(context)), deferred)) + .async_wait(wait_for_one_error(), deferred); + + if (ex0 && ex1) + throw multiple_exceptions(ex0); + if (ex0) + std::rethrow_exception(ex0); + if (ex1) + std::rethrow_exception(ex1); + co_return std::move(detail::awaitable_unwrap>(r0)); +} + +/// Wait for both operations to succeed. +/** + * If one operations fails, the other is cancelled as the AND-condition can no + * longer be satisfied. + */ +template +traced_awaitable, Executor> +operator&&(traced_awaitable, Executor> t, + traced_awaitable u) { + auto ex = co_await this_coro::executor; + auto context = co_await this_coro::context; + + auto [order, ex0, r0, ex1, r1] = + co_await make_parallel_group( + co_spawn( + ex, + detail::awaitable_wrap(std::move(t.continue_trace(context))), + deferred), + co_spawn( + ex, + detail::awaitable_wrap(std::move(u.continue_trace(context))), + deferred)) + .async_wait(wait_for_one_error(), deferred); + + if (ex0 && ex1) + throw multiple_exceptions(ex0); + if (ex0) + std::rethrow_exception(ex0); + if (ex1) + std::rethrow_exception(ex1); + co_return std::tuple_cat( + std::move(detail::awaitable_unwrap>(r0)), + std::make_tuple(std::move(detail::awaitable_unwrap(r1)))); +} + +/// Wait for one operation to succeed. +/** + * If one operations succeeds, the other is cancelled as the OR-condition is + * already satisfied. + */ +template +traced_awaitable, Executor> +operator||(traced_awaitable t, + traced_awaitable u) { + auto ex = co_await this_coro::executor; + auto context = co_await this_coro::context; + + auto [order, ex0, ex1] = + co_await make_parallel_group( + co_spawn(ex, std::move(t.continue_trace(context)), deferred), + co_spawn(ex, std::move(u.continue_trace(context)), deferred)) + .async_wait(wait_for_one_success(), deferred); + + if (order[0] == 0) { + if (!ex0) + co_return std::variant{ + std::in_place_index<0>}; + if (!ex1) + co_return std::variant{ + std::in_place_index<1>}; + throw multiple_exceptions(ex0); + } else { + if (!ex1) + co_return std::variant{ + std::in_place_index<1>}; + if (!ex0) + co_return std::variant{ + std::in_place_index<0>}; + throw multiple_exceptions(ex1); + } +} + +/// Wait for one operation to succeed. +/** + * If one operations succeeds, the other is cancelled as the OR-condition is + * already satisfied. + */ +template +traced_awaitable, Executor> +operator||(traced_awaitable t, + traced_awaitable u) { + auto ex = co_await this_coro::executor; + auto context = co_await this_coro::context; + + auto [order, ex0, ex1, r1] = + co_await make_parallel_group( + co_spawn(ex, std::move(t.continue_trace(context)), deferred), + co_spawn( + ex, + detail::awaitable_wrap(std::move(u.continue_trace(context))), + deferred)) + .async_wait(wait_for_one_success(), deferred); + + if (order[0] == 0) { + if (!ex0) + co_return std::variant{std::in_place_index<0>}; + if (!ex1) + co_return std::variant{ + std::in_place_index<1>, + std::move(detail::awaitable_unwrap(r1))}; + throw multiple_exceptions(ex0); + } else { + if (!ex1) + co_return std::variant{ + std::in_place_index<1>, + std::move(detail::awaitable_unwrap(r1))}; + if (!ex0) + co_return std::variant{std::in_place_index<0>}; + throw multiple_exceptions(ex1); + } +} + +/// Wait for one operation to succeed. +/** + * If one operations succeeds, the other is cancelled as the OR-condition is + * already satisfied. + */ +template +traced_awaitable, Executor> +operator||(traced_awaitable t, + traced_awaitable u) { + auto ex = co_await this_coro::executor; + auto context = co_await this_coro::context; + + auto [order, ex0, r0, ex1] = + co_await make_parallel_group( + co_spawn( + ex, + detail::awaitable_wrap(std::move(t.continue_trace(context))), + deferred), + co_spawn(ex, std::move(u.continue_trace(context)), deferred)) + .async_wait(wait_for_one_success(), deferred); + + if (order[0] == 0) { + if (!ex0) + co_return std::variant{ + std::in_place_index<0>, + std::move(detail::awaitable_unwrap(r0))}; + if (!ex1) + co_return std::variant{std::in_place_index<1>}; + throw multiple_exceptions(ex0); + } else { + if (!ex1) + co_return std::variant{std::in_place_index<1>}; + if (!ex0) + co_return std::variant{ + std::in_place_index<0>, + std::move(detail::awaitable_unwrap(r0))}; + throw multiple_exceptions(ex1); + } +} + +/// Wait for one operation to succeed. +/** + * If one operations succeeds, the other is cancelled as the OR-condition is + * already satisfied. + */ +template +traced_awaitable, Executor> +operator||(traced_awaitable t, traced_awaitable u) { + auto ex = co_await this_coro::executor; + auto context = co_await this_coro::context; + + auto [order, ex0, r0, ex1, r1] = + co_await make_parallel_group( + co_spawn( + ex, + detail::awaitable_wrap(std::move(t.continue_trace(context))), + deferred), + co_spawn( + ex, + detail::awaitable_wrap(std::move(u.continue_trace(context))), + deferred)) + .async_wait(wait_for_one_success(), deferred); + + if (order[0] == 0) { + if (!ex0) + co_return std::variant{ + std::in_place_index<0>, + std::move(detail::awaitable_unwrap(r0))}; + if (!ex1) + co_return std::variant{ + std::in_place_index<1>, + std::move(detail::awaitable_unwrap(r1))}; + throw multiple_exceptions(ex0); + } else { + if (!ex1) + co_return std::variant{ + std::in_place_index<1>, + std::move(detail::awaitable_unwrap(r1))}; + if (!ex0) + co_return std::variant{ + std::in_place_index<0>, + std::move(detail::awaitable_unwrap(r0))}; + throw multiple_exceptions(ex1); + } +} + +namespace detail { + +template struct widen_variant { + template + static std::variant call(SourceVariant& source) { + if (source.index() == I) + return std::variant{std::in_place_index, + std::move(std::get(source))}; + else if constexpr (I + 1 < std::variant_size_v) + return call(source); + else + throw std::logic_error("empty variant"); + } +}; + +} // namespace detail + +/// Wait for one operation to succeed. +/** + * If one operations succeeds, the other is cancelled as the OR-condition is + * already satisfied. + */ +template +traced_awaitable, Executor> +operator||(traced_awaitable, Executor> t, + traced_awaitable u) { + auto ex = co_await this_coro::executor; + auto context = co_await this_coro::context; + + auto [order, ex0, r0, ex1] = + co_await make_parallel_group( + co_spawn( + ex, + detail::awaitable_wrap(std::move(t.continue_trace(context))), + deferred), + co_spawn(ex, std::move(u.continue_trace(context)), deferred)) + .async_wait(wait_for_one_success(), deferred); + + using widen = detail::widen_variant; + if (order[0] == 0) { + if (!ex0) + co_return widen::template call<0>( + detail::awaitable_unwrap>(r0)); + if (!ex1) + co_return std::variant{ + std::in_place_index}; + throw multiple_exceptions(ex0); + } else { + if (!ex1) + co_return std::variant{ + std::in_place_index}; + if (!ex0) + co_return widen::template call<0>( + detail::awaitable_unwrap>(r0)); + throw multiple_exceptions(ex1); + } +} + +/// Wait for one operation to succeed. +/** + * If one operations succeeds, the other is cancelled as the OR-condition is + * already satisfied. + */ +template +traced_awaitable, Executor> +operator||(traced_awaitable, Executor> t, + traced_awaitable u) { + auto ex = co_await this_coro::executor; + auto context = co_await this_coro::context; + + auto [order, ex0, r0, ex1, r1] = + co_await make_parallel_group( + co_spawn( + ex, + detail::awaitable_wrap(std::move(t.continue_trace(context))), + deferred), + co_spawn( + ex, + detail::awaitable_wrap(std::move(u.continue_trace(context))), + deferred)) + .async_wait(wait_for_one_success(), deferred); + + using widen = detail::widen_variant; + if (order[0] == 0) { + if (!ex0) + co_return widen::template call<0>( + detail::awaitable_unwrap>(r0)); + if (!ex1) + co_return std::variant{ + std::in_place_index, + std::move(detail::awaitable_unwrap(r1))}; + throw multiple_exceptions(ex0); + } else { + if (!ex1) + co_return std::variant{ + std::in_place_index, + std::move(detail::awaitable_unwrap(r1))}; + if (!ex0) + co_return widen::template call<0>( + detail::awaitable_unwrap>(r0)); + throw multiple_exceptions(ex1); + } +} + +} // namespace awaitable_operators +} // namespace experimental +} // namespace asio +} // namespace boost diff --git a/src/proxy/cache/disk/body.h b/src/proxy/cache/disk/body.h index 2dc14092b..892f8cb81 100644 --- a/src/proxy/cache/disk/body.h +++ b/src/proxy/cache/disk/body.h @@ -50,6 +50,8 @@ class reader_body { class writer_body { public: + using support_double_buffer = std::false_type; + writer_body(storage::data_view& storage, std::shared_ptr objh, std::size_t buffer_size = 32 * MEBI_BYTE) @@ -57,20 +59,37 @@ class writer_body { m_objh{std::move(objh)}, m_buffer(buffer_size) {} - coro> get() { + writer_body(const writer_body&) = delete; + writer_body& operator=(const writer_body&) = delete; + writer_body(writer_body&&) = delete; + writer_body& operator=(writer_body&&) = delete; + + coro> get() { co_return co_await _get(&m_buffer); } + +private: + storage::data_view& m_storage; + std::shared_ptr m_objh; + + std::size_t m_addr_index{0}; + std::size_t m_frag_offset{0}; + +protected: + std::vector m_buffer; + + coro> _get(std::vector* buffer) { std::size_t read_size = 0; address partial_addr; while (m_addr_index < m_objh->get_address().size() && - read_size < m_buffer.size()) { + read_size < buffer->size()) { auto frag = m_objh->get_address().get(m_addr_index); if (m_frag_offset > 0) { frag.pointer += m_frag_offset; frag.size -= m_frag_offset; } - if (frag.size + read_size > m_buffer.size()) { - auto remains = m_buffer.size() - read_size; + if (frag.size + read_size > buffer->size()) { + auto remains = buffer->size() - read_size; m_frag_offset += remains; frag.size = remains; partial_addr.push(frag); @@ -85,19 +104,40 @@ class writer_body { if (read_size > 0) { co_await m_storage.read_address(partial_addr, - {m_buffer.data(), read_size}); + {buffer->data(), read_size}); } - co_return std::span{m_buffer.data(), read_size}; + co_return std::span{buffer->data(), read_size}; } +}; -private: - storage::data_view& m_storage; - std::shared_ptr m_objh; +class double_buffered_writer_body : private writer_body { +public: + using support_double_buffer = std::true_type; + + double_buffered_writer_body(storage::data_view& storage, + std::shared_ptr objh, + std::size_t buffer_size = 32 * MEBI_BYTE) + : writer_body(storage, objh, buffer_size), + m_buffer2(buffer_size), + m_active(&m_buffer), + m_standby(&m_buffer2) {} + + double_buffered_writer_body(const double_buffered_writer_body&) = delete; + double_buffered_writer_body& + operator=(const double_buffered_writer_body&) = delete; + double_buffered_writer_body(double_buffered_writer_body&&) = delete; + double_buffered_writer_body& + operator=(double_buffered_writer_body&&) = delete; - std::vector m_buffer; + coro> get() { + auto rv = co_await writer_body::_get(m_active); + std::swap(m_active, m_standby); + co_return rv; + } - std::size_t m_addr_index = 0; - std::size_t m_frag_offset = 0; +private: + std::vector m_buffer2; + std::vector* m_active; + std::vector* m_standby; }; - } // namespace uh::cluster::proxy::cache::disk diff --git a/src/proxy/cache/disk/manager.h b/src/proxy/cache/disk/manager.h index f7eb75ada..cd9215a91 100644 --- a/src/proxy/cache/disk/manager.h +++ b/src/proxy/cache/disk/manager.h @@ -66,12 +66,13 @@ class manager { std::cout << "Total size after put: " << m_current_size << std::endl; } - std::optional get(object_metadata key) { + std::unique_ptr get(object_metadata key) { auto entry = m_cache->get(key); if (!entry) { - return std::nullopt; + return nullptr; } - return writer_body{m_storage, std::move(entry)}; + return std::make_unique(m_storage, + std::move(entry)); } static manager create(boost::asio::io_context& ioc, data_view& storage, diff --git a/src/proxy/handler.cpp b/src/proxy/handler.cpp index 3534d37c0..96747f1c8 100644 --- a/src/proxy/handler.cpp +++ b/src/proxy/handler.cpp @@ -7,6 +7,7 @@ #include #include #include +#include using namespace uh::cluster::ep::http; @@ -15,9 +16,7 @@ namespace uh::cluster::proxy { handler::handler( std::unique_ptr factory, std::function()> sf, - storage::data_view& dv, - cache::disk::manager& mgr, - std::size_t buffer_size) + storage::data_view& dv, cache::disk::manager& mgr, std::size_t buffer_size) : m_factory(std::move(factory)), m_sf(std::move(sf)), m_dv(dv), @@ -53,8 +52,9 @@ coro handler::handle(boost::asio::ip::tcp::socket s) { co_await m_factory->create(incoming, rawreq); if (get_object::can_handle(*req)) { - auto writer = m_mgr.get(cache::disk::object_metadata{ req->object_key() }); - if (writer) { + auto wbody = + m_mgr.get(cache::disk::object_metadata{req->object_key()}); + if (wbody) { LOG_INFO() << peer << ": handling from cache"; incoming.set_mode(forward_stream::deleting); outgoing.set_mode(forward_stream::deleting); @@ -71,12 +71,7 @@ coro handler::handle(boost::asio::ip::tcp::socket s) { LOG_INFO() << peer << ": done reading complete request"; - std::span data = co_await writer->get(); - while (!data.empty()) { - LOG_INFO() << peer << ": sending " << data.size() << " bytes response"; - co_await incoming.write(data); - data = co_await writer->get(); - } + co_await cache::async_write(incoming, *wbody); LOG_INFO() << peer << ": cache result served"; continue; @@ -116,7 +111,6 @@ coro handler::handle(boost::asio::ip::tcp::socket s) { auto res = parser.release(); bs = outgoing.buffer_size(); - std::size_t read = 0ull; std::size_t len = std::stoul(res.at("Content-Length")); if (r.method() == boost::beast::http::verb::head && (res.result_int() / 100 == 2)) { @@ -124,32 +118,25 @@ coro handler::handle(boost::asio::ip::tcp::socket s) { } LOG_INFO() << peer << ": sending response " << res.result_int() - << " " << res.reason() << " -- " << len; + << " " << res.reason() << " -- " << len; if (get_object::can_handle(*req)) { - cache::disk::reader_body data(m_dv); - LOG_INFO() << peer << ": add " << buffer.size() << " response header"; - co_await data.put(buffer); + LOG_INFO() << peer << ": add " << buffer.size() + << " response header"; + cache::disk::reader_body rbody(m_dv); + co_await rbody.put(buffer); - while (read < len) { - co_await outgoing.consume(); - - auto r = co_await outgoing.read(len - read); - LOG_INFO() << peer << ": add " << r.size() << " response data"; - co_await data.put(r); + co_await cache::async_read(outgoing, rbody, len); - // r: data - read += r.size(); - } - - co_await m_mgr.put(cache::disk::object_metadata{ req->object_key() }, data); - co_await outgoing.consume(); + co_await m_mgr.put( + cache::disk::object_metadata{req->object_key()}, rbody); } else { + std::size_t read = 0ull; while (read < len) { co_await outgoing.consume(); auto r = co_await outgoing.read(len - read); - // r: data + // r: rbody read += r.size(); } diff --git a/test/unit/test_disk_cache_body.cpp b/test/unit/test_disk_cache_body.cpp index c860e1764..520ccc11d 100644 --- a/test/unit/test_disk_cache_body.cpp +++ b/test/unit/test_disk_cache_body.cpp @@ -106,7 +106,7 @@ BOOST_AUTO_TEST_CASE(supports_write) { auto objh = std::make_shared(std::move(addr)); BOOST_TEST(objh->data_size() == data.size()); - writer_body body(data_view, std::move(objh), 16); + double_buffered_writer_body body(data_view, std::move(objh), 16); // Set up TCP sockets boost::asio::ip::tcp::acceptor acceptor(m_ioc, diff --git a/test/unit/test_disk_cache_manager.cpp b/test/unit/test_disk_cache_manager.cpp index 660710572..30b53ce9b 100644 --- a/test/unit/test_disk_cache_manager.cpp +++ b/test/unit/test_disk_cache_manager.cpp @@ -29,12 +29,11 @@ BOOST_AUTO_TEST_CASE(put_and_get_with_metadata) { boost::asio::co_spawn(m_ioc, mgr.put(key, rbody), boost::asio::use_future) .get(); - auto wbody_opt = mgr.get(key); - BOOST_TEST(wbody_opt.has_value()); + auto writer = mgr.get(key); + BOOST_TEST(writer != nullptr); - auto& wbody = wbody_opt.value(); auto buf = - boost::asio::co_spawn(m_ioc, wbody.get(), boost::asio::use_future) + boost::asio::co_spawn(m_ioc, writer->get(), boost::asio::use_future) .get(); BOOST_TEST(buf.size() == data.size()); @@ -67,8 +66,8 @@ BOOST_AUTO_TEST_CASE(eviction_test) { .get(); } - auto wbody_opt = mgr.get(keys.front()); - BOOST_TEST(!wbody_opt.has_value()); + auto writer = mgr.get(keys.front()); + BOOST_TEST(writer == nullptr); } BOOST_AUTO_TEST_SUITE_END()