libdpf/include/dpf/asio.hpp
Ryan Henry e4e666f459 Initial import of libdpf.
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-24 14:08:32 -06:00

1167 lines
44 KiB
C++

/// @file dpf/asio.hpp
/// @brief
/// @details
/// @author Ryan Henry <ryan.henry@ucalgary.ca>
/// @copyright Copyright (c) 2019-2023 Ryan Henry and [others](@ref authors)
/// @license Released under a GNU General Public v2.0 (GPLv2) license;
/// see [LICENSE.md](@ref license) for details.
#ifndef LIBDPF_INCLUDE_DPF_ASIO_HPP__
#define LIBDPF_INCLUDE_DPF_ASIO_HPP__
#include "hedley/hedley.h"
#include <asio.hpp>
#include "dpf/prg.hpp"
#include "dpf/dpf_key.hpp"
extern bool do_quickack;
using quickack = asio::detail::socket_option::boolean<IPPROTO_TCP, TCP_QUICKACK>;
quickack quickack_toggle{false};
namespace dpf
{
namespace asio
{
namespace detail
{
template <class ...> using void_t = void;
template <typename T, typename = void> struct has_lowest_layer : std::false_type {};
template <typename T> struct has_lowest_layer<T, void_t<decltype(std::declval<T>().get_lowest_layer())>> : std::true_type {};
template <typename T> static constexpr bool has_lowest_layer_v = has_lowest_layer<T>::value;
}
template <typename ExecutorT,
typename Function,
typename CompletionToken>
auto async_post(ExecutorT executor, Function && func, CompletionToken && token)
{
#include <asio/yield.hpp>
return ::asio::async_compose<CompletionToken, void()>(
[
executor,
func = std::move(func)
]
(auto & self)
mutable
{
::asio::post(executor,
[
func = std::move(func),
self = std::move(self)
]
()
mutable
{
func();
self.complete();
});
}, token, executor);
#include <asio/unyield.hpp>
}
//
// make_dpf
//
template <typename InteriorPRG = dpf::prg::aes128,
typename ExteriorPRG = InteriorPRG,
typename PeerT,
typename InputT,
typename OutputT,
typename ...OutputTs>
auto make_dpf(PeerT & peer0, PeerT & peer1, std::size_t count, dpfargs<InputT, OutputT, OutputTs...> & args, ::asio::error_code & error, root_sampler_t<InteriorPRG> && root_sampler = dpf::uniform_sample<typename InteriorPRG::block_type>)
{
using dpf_type = utils::dpf_type_t<InteriorPRG, ExteriorPRG, InputT, OutputT, OutputTs...>;
using correction_words_array = typename dpf_type::correction_words_array;
using correction_advice_array = typename dpf_type::correction_advice_array;
using interior_node = typename dpf_type::interior_node;
using leaf_tuple = typename dpf_type::leaf_tuple;
using beaver_tuple = typename dpf_type::beaver_tuple;
using input_type = typename dpf_type::input_type;
std::size_t bytes_written0 = 0, bytes_written1 = 0;
for (std::size_t num_written = 0; num_written < count; ++num_written)
{
auto [correction_words, correction_advice, priv0, priv1]
= dpf::detail::make_dpf_impl<InteriorPRG, ExteriorPRG>(args, std::forward<root_sampler_t<InteriorPRG>>(root_sampler));
auto & [root0, leaves0, beavers0, offset_share0] = priv0;
auto & [root1, leaves1, beavers1, offset_share1] = priv1;
bytes_written0 += ::asio::write(peer0,
std::array<::asio::const_buffer, 6>{
::asio::buffer(&correction_words, sizeof(correction_words_array)),
::asio::buffer(&correction_advice, sizeof(correction_advice_array)),
::asio::buffer(&root0, sizeof(interior_node)),
::asio::buffer(&leaves0, sizeof(leaf_tuple)),
::asio::buffer(&beavers0, sizeof(beaver_tuple)),
::asio::buffer(&offset_share0, sizeof(input_type))
}, error);
if (error)
{
return std::make_tuple(bytes_written0, bytes_written1, num_written);
}
bytes_written1 += ::asio::write(peer1,
std::array<::asio::const_buffer, 6>{
::asio::buffer(&correction_words, sizeof(correction_words_array)),
::asio::buffer(&correction_advice, sizeof(correction_advice_array)),
::asio::buffer(&root1, sizeof(interior_node)),
::asio::buffer(&leaves1, sizeof(leaf_tuple)),
::asio::buffer(&beavers1, sizeof(beaver_tuple)),
::asio::buffer(&offset_share1, sizeof(input_type))
}, error);
if (error)
{
return std::make_tuple(bytes_written0, bytes_written1, num_written);
}
}
return std::make_tuple(bytes_written0, bytes_written1, count);
}
template <typename InteriorPRG = dpf::prg::aes128,
typename ExteriorPRG = InteriorPRG,
typename PeerT,
typename InputT,
typename OutputT,
typename ...OutputTs>
HEDLEY_ALWAYS_INLINE
auto make_dpf(PeerT & peer0, PeerT & peer1, std::size_t count, dpfargs<InputT, OutputT, OutputTs...> & args, root_sampler_t<InteriorPRG> && root_sampler = dpf::uniform_sample<typename InteriorPRG::block_type>)
{
::asio::error_code error{};
auto ret = dpf::asio::make_dpf(peer0, peer1, count, args, error, std::forward<root_sampler_t<InteriorPRG>>(root_sampler));
if (error) throw error;
return ret;
}
template <typename InteriorPRG = dpf::prg::aes128,
typename ExteriorPRG = InteriorPRG,
typename PeerT,
typename InputT,
typename OutputT,
typename ...OutputTs>
HEDLEY_ALWAYS_INLINE
auto make_dpf(PeerT & peer0, PeerT & peer1, dpfargs<InputT, OutputT, OutputTs...> args, ::asio::error_code & error, root_sampler_t<InteriorPRG> && root_sampler = dpf::uniform_sample<typename InteriorPRG::block_type>)
{
auto [bytes_written0, bytes_written1, num_written]
= dpf::asio::make_dpf<InteriorPRG, ExteriorPRG>(peer0, peer1, static_cast<std::size_t>(1), args, error, std::forward<root_sampler_t<InteriorPRG>>(root_sampler));
return std::make_tuple(bytes_written0, bytes_written1);
}
template <typename InteriorPRG = dpf::prg::aes128,
typename ExteriorPRG = InteriorPRG,
typename PeerT,
typename InputT,
typename OutputT,
typename ...OutputTs>
HEDLEY_ALWAYS_INLINE
auto make_dpf(PeerT & peer0, PeerT & peer1, dpfargs<InputT, OutputT, OutputTs...> args, root_sampler_t<InteriorPRG> && root_sampler = dpf::uniform_sample<typename InteriorPRG::block_type>)
{
auto [bytes_written0, bytes_written1, num_written]
= dpf::asio::make_dpf<InteriorPRG, ExteriorPRG>(peer0, peer1, static_cast<std::size_t>(1), args, std::forward<root_sampler_t<InteriorPRG>>(root_sampler));
return std::make_tuple(bytes_written0, bytes_written1);
}
//
// async_make_dpf
//
template <typename InteriorPRG = dpf::prg::aes128,
typename ExteriorPRG = InteriorPRG,
typename PeerT,
typename ExecutorT,
typename InputT,
typename OutputT,
typename ...OutputTs,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
auto async_make_dpf(PeerT & peer0, PeerT & peer1, ExecutorT work_executor,
std::size_t count, dpfargs<InputT, OutputT, OutputTs...> args,
CompletionToken && token, root_sampler_t<InteriorPRG> && root_sampler = dpf::uniform_sample<typename InteriorPRG::block_type>)
{
using dpf_type = utils::dpf_type_t<InteriorPRG, ExteriorPRG, InputT, OutputT, OutputTs...>;
using correction_words_array = typename dpf_type::correction_words_array;
using correction_advice_array = typename dpf_type::correction_advice_array;
using interior_node = typename dpf_type::interior_node;
using leaf_tuple = typename dpf_type::leaf_tuple;
using beaver_tuple = typename dpf_type::beaver_tuple;
using input_type = typename dpf_type::input_type;
using dpf_priv_values = std::tuple<interior_node, leaf_tuple, beaver_tuple, input_type>;
using dpf_values = std::tuple<correction_words_array, correction_advice_array, dpf_priv_values, dpf_priv_values>;
#include <asio/yield.hpp>
return ::asio::async_compose<
CompletionToken, void(::asio::error_code, // error status
std::size_t, // bytes_written0
std::size_t, // bytes_written1
std::size_t)>( // num_written
[
&peer0,
&peer1,
work_executor,
args,
count = std::size_t(count),
dpf_data = std::make_shared<dpf_values>(),
num_written = std::size_t(0),
bytes_written0 = std::size_t(0),
bytes_written1 = std::size_t(0),
coro = ::asio::coroutine()
]
(
auto & self,
const ::asio::error_code & error = {},
std::size_t bytes_just_written = 0
)
mutable
{
reenter (coro)
{
while (num_written++ < count)
{
yield async_post(work_executor, [dpf_data, args]()
{
*dpf_data = dpf::detail::make_dpf_impl<InteriorPRG, ExteriorPRG>(args);
}, std::move(self));
yield ::asio::async_write(peer0, std::array<::asio::const_buffer, 6>{
::asio::buffer(&utils::get<0>(*dpf_data), sizeof(correction_words_array)),
::asio::buffer(&utils::get<1>(*dpf_data), sizeof(correction_advice_array)),
::asio::buffer(&utils::get<0>(utils::get<2>(*dpf_data)), sizeof(interior_node)),
::asio::buffer(&utils::get<1>(utils::get<2>(*dpf_data)), sizeof(leaf_tuple)),
::asio::buffer(&utils::get<2>(utils::get<2>(*dpf_data)), sizeof(beaver_tuple)),
::asio::buffer(&utils::get<3>(utils::get<2>(*dpf_data)), sizeof(input_type))},
std::move(self));
bytes_written0 += bytes_just_written;
if (error)
{
self.complete(error, bytes_written0, bytes_written1, num_written);
break;
}
yield ::asio::async_write(peer1, std::array<::asio::const_buffer, 6>{
::asio::buffer(&utils::get<0>(*dpf_data), sizeof(correction_words_array)),
::asio::buffer(&utils::get<1>(*dpf_data), sizeof(correction_advice_array)),
::asio::buffer(&utils::get<0>(utils::get<3>(*dpf_data)), sizeof(interior_node)),
::asio::buffer(&utils::get<1>(utils::get<3>(*dpf_data)), sizeof(leaf_tuple)),
::asio::buffer(&utils::get<2>(utils::get<3>(*dpf_data)), sizeof(beaver_tuple)),
::asio::buffer(&utils::get<3>(utils::get<3>(*dpf_data)), sizeof(input_type))},
std::move(self));
bytes_written1 += bytes_just_written;
if (error)
{
self.complete(error, bytes_written0, bytes_written1, num_written);
break;
}
}
self.complete(error, bytes_written0, bytes_written1, count);
}
},
token, peer0, peer1, work_executor);
#include <asio/unyield.hpp>
}
template <typename InteriorPRG = dpf::prg::aes128,
typename ExteriorPRG = InteriorPRG,
typename PeerT,
typename ExecutorT,
typename InputT,
typename OutputT,
typename ...OutputTs,
typename CompletionToken,
std::enable_if_t<!std::is_integral_v<ExecutorT>, bool> = false>
HEDLEY_WARN_UNUSED_RESULT
HEDLEY_ALWAYS_INLINE
auto async_make_dpf(PeerT & peer0, PeerT & peer1, ExecutorT work_executor,
dpfargs<InputT, OutputT, OutputTs...> args, CompletionToken && token,
root_sampler_t<InteriorPRG> && root_sampler = dpf::uniform_sample<typename InteriorPRG::block_type>)
{
std::size_t count = 1;
return async_make_dpf<InteriorPRG, ExteriorPRG>(peer0, peer1, work_executor, count, args, std::forward<CompletionToken>(token), std::forward<root_sampler_t<InteriorPRG>>(root_sampler));
}
template <typename InteriorPRG = dpf::prg::aes128,
typename ExteriorPRG = InteriorPRG,
typename PeerT,
typename InputT,
typename OutputT,
typename ...OutputTs,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
HEDLEY_ALWAYS_INLINE
auto async_make_dpf(PeerT & peer0, PeerT & peer1, std::size_t count, dpfargs<InputT, OutputT, OutputTs...> args, CompletionToken && token,
root_sampler_t<InteriorPRG> && root_sampler = dpf::uniform_sample<typename InteriorPRG::block_type>)
{
auto work_executor = ::asio::system_executor();
return async_make_dpf<InteriorPRG, ExteriorPRG>(peer0, peer1, work_executor, count, args, std::forward<CompletionToken>(token), std::forward<root_sampler_t<InteriorPRG>>(root_sampler));
}
template <typename InteriorPRG = dpf::prg::aes128,
typename ExteriorPRG = InteriorPRG,
typename PeerT,
typename InputT,
typename OutputT,
typename ...OutputTs,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
HEDLEY_ALWAYS_INLINE
auto async_make_dpf(PeerT & peer0, PeerT & peer1, dpfargs<InputT, OutputT, OutputTs...> args, CompletionToken && token,
root_sampler_t<InteriorPRG> && root_sampler = dpf::uniform_sample<typename InteriorPRG::block_type>)
{
auto work_executor = ::asio::system_executor();
std::size_t count = 1;
return dpf::asio::async_make_dpf<InteriorPRG, ExteriorPRG>(peer0, peer1, work_executor, count, args, std::forward<CompletionToken>(token), std::forward<root_sampler_t<InteriorPRG>>(root_sampler));
}
//
// read_dpf
//
template <typename DpfKey,
typename DealerT,
typename BackEmplaceable>
auto read_dpf(DealerT & dealer, BackEmplaceable & output, std::size_t count, ::asio::error_code & error)
{
using dpf_type = DpfKey;
using interior_node = typename dpf_type::interior_node;
using leaf_tuple = typename dpf_type::leaf_tuple;
using beaver_tuple = typename dpf_type::beaver_tuple;
using input_type = typename dpf_type::input_type;
using correction_words_array = typename dpf_type::correction_words_array;
using correction_advice_array = typename dpf_type::correction_advice_array;
interior_node root;
correction_words_array correction_words;
correction_advice_array correction_advice;
leaf_tuple leaves;
beaver_tuple beavers;
input_type offset_share;
std::size_t bytes_read = 0;
for (std::size_t num_read = 0; num_read < count; ++num_read)
{
bytes_read += ::asio::read(dealer,
std::array<::asio::mutable_buffer, 6>{
::asio::buffer(&correction_words, sizeof(correction_words_array)),
::asio::buffer(&correction_advice, sizeof(correction_advice_array)),
::asio::buffer(&root, sizeof(interior_node)),
::asio::buffer(&leaves, sizeof(leaf_tuple)),
::asio::buffer(&beavers, sizeof(beaver_tuple)),
::asio::buffer(&offset_share, sizeof(input_type))
}, error);
if constexpr(detail::has_lowest_layer_v<DealerT>)
{
if (do_quickack) dealer.get_lowest_layer().set_option(quickack_toggle);
}
if (error)
{
return std::make_pair(bytes_read, num_read);
}
dpf_type::emplace_back(output, root, correction_words, correction_advice, leaves, beavers, offset_share);
}
return std::make_pair(bytes_read, count);
}
template <typename DpfKey,
typename DealerT,
typename BackEmplaceable>
HEDLEY_ALWAYS_INLINE
auto read_dpf(DealerT & dealer, BackEmplaceable & output, std::size_t count)
{
::asio::error_code error{};
auto ret = dpf::asio::read_dpf<DpfKey>(dealer, output, count, error);
if constexpr(detail::has_lowest_layer_v<DealerT>)
{
if (do_quickack) dealer.get_lowest_layer().set_option(quickack_toggle);
}
if (error) throw error;
return ret;
}
template <typename DpfKey,
typename DealerT,
typename Emplaceable>
auto read_dpf(DealerT & dealer, Emplaceable & output, ::asio::error_code & error)
{
using dpf_type = DpfKey;
using interior_node = typename dpf_type::interior_node;
using leaf_tuple = typename dpf_type::leaf_tuple;
using beaver_tuple = typename dpf_type::beaver_tuple;
using input_type = typename dpf_type::input_type;
using correction_words_array = typename dpf_type::correction_words_array;
using correction_advice_array = typename dpf_type::correction_advice_array;
interior_node root;
correction_words_array correction_words;
correction_advice_array correction_advice;
leaf_tuple leaves;
beaver_tuple beavers;
input_type offset_share;
std::size_t bytes_read = ::asio::read(dealer,
std::array<::asio::mutable_buffer, 6>{
::asio::buffer(&correction_words, sizeof(correction_words_array)),
::asio::buffer(&correction_advice, sizeof(correction_advice_array)),
::asio::buffer(&root, sizeof(interior_node)),
::asio::buffer(&leaves, sizeof(leaf_tuple)),
::asio::buffer(&beavers, sizeof(beaver_tuple)),
::asio::buffer(&offset_share, sizeof(input_type))
}, error);
if constexpr(detail::has_lowest_layer_v<DealerT>)
{
if (do_quickack) dealer.get_lowest_layer().set_option(quickack_toggle);
}
if (error)
{
return bytes_read;
}
dpf_type::emplace(output, root, correction_words, correction_advice, leaves, beavers, offset_share);
return bytes_read;
}
template <typename DpfKey,
typename DealerT,
typename Emplaceable>
HEDLEY_ALWAYS_INLINE
auto read_dpf(DealerT & dealer, Emplaceable & output)
{
::asio::error_code error{};
auto ret = dpf::asio::read_dpf<DpfKey>(dealer, output, error);
if constexpr(detail::has_lowest_layer_v<DealerT>)
{
if (do_quickack) dealer.get_lowest_layer().set_option(quickack_toggle);
}
if (error) throw error;
return ret;
}
//
// async_read_dpf
//
template <typename DpfKey,
typename DealerT,
typename ExecutorT,
typename BackEmplaceable,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
auto async_read_dpf(DealerT & dealer, ExecutorT work_executor, BackEmplaceable & output, std::size_t count, CompletionToken && token)
{
using dpf_type = DpfKey;
using correction_words_array = typename dpf_type::correction_words_array;
using correction_advice_array = typename dpf_type::correction_advice_array;
using interior_node = typename dpf_type::interior_node;
using leaf_tuple = typename dpf_type::leaf_tuple;
using beaver_tuple = typename dpf_type::beaver_tuple;
using input_type = typename dpf_type::input_type;
using dpf_priv_values = std::tuple<interior_node, leaf_tuple, beaver_tuple, input_type>;
using dpf_values = std::tuple<correction_words_array, correction_advice_array, dpf_priv_values>;
#include <asio/yield.hpp>
return ::asio::async_compose<
CompletionToken, void(::asio::error_code, // error status
std::size_t, // bytes_read
std::size_t)>( // num_read
[
&dealer,
work_executor,
&output,
count,
dpf_data = std::make_shared<dpf_values>(),
num_read = std::size_t(0),
bytes_read = std::size_t(0),
coro = ::asio::coroutine()
]
(
auto & self,
const ::asio::error_code & error = {},
std::size_t bytes_just_read = 0
)
mutable
{
reenter (coro)
{
while (num_read++ < count)
{
yield ::asio::async_read(dealer, std::array<::asio::mutable_buffer, 6>{
::asio::buffer(&std::get<0>(*dpf_data), sizeof(correction_words_array)),
::asio::buffer(&std::get<1>(*dpf_data), sizeof(correction_advice_array)),
::asio::buffer(&std::get<0>(std::get<2>(*dpf_data)), sizeof(interior_node)),
::asio::buffer(&std::get<1>(std::get<2>(*dpf_data)), sizeof(leaf_tuple)),
::asio::buffer(&std::get<2>(std::get<2>(*dpf_data)), sizeof(beaver_tuple)),
::asio::buffer(&std::get<3>(std::get<2>(*dpf_data)), sizeof(input_type))}, std::move(self));
if constexpr(detail::has_lowest_layer_v<DealerT>)
{
if (do_quickack) dealer.get_lowest_layer().set_option(quickack_toggle);
}
bytes_read += bytes_just_read;
if (error)
{
self.complete(error, bytes_read, num_read);
break;
}
yield async_post(work_executor, [dpf_data, &output]() mutable
{
auto & [correction_words, correction_advice, priv]
= *dpf_data;
auto & [root, leaves, beavers, offset_share] = priv;
dpf_type::emplace_back(output, root, correction_words, correction_advice, leaves, beavers, offset_share);
}, std::move(self));
}
self.complete(error, bytes_read, count);
}
},
token, dealer, work_executor);
#include <asio/unyield.hpp>
}
template <typename DpfKey,
typename DealerT,
typename ExecutorT,
typename Emplaceable,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
auto async_read_dpf(DealerT & dealer, ExecutorT work_executor, Emplaceable & output, CompletionToken && token)
{
using dpf_type = DpfKey;
using correction_words_array = typename dpf_type::correction_words_array;
using correction_advice_array = typename dpf_type::correction_advice_array;
using interior_node = typename dpf_type::interior_node;
using leaf_tuple = typename dpf_type::leaf_tuple;
using beaver_tuple = typename dpf_type::beaver_tuple;
using input_type = typename dpf_type::input_type;
using dpf_priv_values = std::tuple<interior_node, leaf_tuple, beaver_tuple, input_type>;
using dpf_values = std::tuple<correction_words_array, correction_advice_array, dpf_priv_values>;
#include <asio/yield.hpp>
return ::asio::async_compose<
CompletionToken, void(::asio::error_code, // error status
std::size_t)>( // bytes_read
[
&dealer,
work_executor,
&output,
dpf_data = std::make_shared<dpf_values>(),
bytes_read = std::size_t(0),
coro = ::asio::coroutine()
]
(
auto & self,
const ::asio::error_code & error = {},
std::size_t bytes_just_read = 0
)
mutable
{
reenter (coro)
{
yield ::asio::async_read(dealer, std::array<::asio::mutable_buffer, 6>{
::asio::buffer(&std::get<0>(*dpf_data), sizeof(correction_words_array)),
::asio::buffer(&std::get<1>(*dpf_data), sizeof(correction_advice_array)),
::asio::buffer(&std::get<0>(std::get<2>(*dpf_data)), sizeof(interior_node)),
::asio::buffer(&std::get<1>(std::get<2>(*dpf_data)), sizeof(leaf_tuple)),
::asio::buffer(&std::get<2>(std::get<2>(*dpf_data)), sizeof(beaver_tuple)),
::asio::buffer(&std::get<3>(std::get<2>(*dpf_data)), sizeof(input_type))}, std::move(self));
if constexpr(detail::has_lowest_layer_v<DealerT>)
{
if (do_quickack) dealer.get_lowest_layer().set_option(quickack_toggle);
}
bytes_read = bytes_just_read;
if (error)
{
self.complete(error, bytes_read);
break;
}
yield async_post(work_executor, [dpf_data, &output]()
{
auto & [correction_words, correction_advice, priv]
= *dpf_data;
auto & [root, leaves, beavers, offset_share] = priv;
dpf_type::emplace(output,
std::get<0>(std::get<2>(*dpf_data)),
std::get<0>(*dpf_data),
std::get<1>(*dpf_data),
std::get<1>(std::get<2>(*dpf_data)),
std::get<2>(std::get<2>(*dpf_data)),
std::get<3>(std::get<2>(*dpf_data)));
}, std::move(self));
self.complete(error, bytes_read);
}
},
token, dealer, work_executor);
#include <asio/unyield.hpp>
}
template <typename DpfKey,
typename DealerT,
typename BackEmplaceable,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
HEDLEY_ALWAYS_INLINE
auto async_read_dpf(DealerT & dealer, BackEmplaceable & output, std::size_t count, CompletionToken && token)
{
auto work_executor = ::asio::system_executor();
return async_read_dpf<DpfKey>(dealer, work_executor, output, count, std::forward<CompletionToken>(token));
}
template <typename DpfKey,
typename DealerT,
typename Emplaceable,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
HEDLEY_ALWAYS_INLINE
auto async_read_dpf(DealerT & dealer, Emplaceable & output, CompletionToken && token)
{
auto work_executor = ::asio::system_executor();
return dpf::asio::async_read_dpf<DpfKey>(dealer, work_executor, output, std::forward<CompletionToken>(token));
}
//
// assign_wildcard_input
//
template <typename PeerT,
typename DpfKey,
typename InputType>
auto assign_wildcard_input(PeerT & peer_in, PeerT & peer_out, DpfKey & dpf,
InputType && input_share, ::asio::error_code & error)
{
using input_type = typename DpfKey::input_type;
static_assert(std::is_convertible_v<InputType, input_type>);
std::size_t bytes_written = 0, bytes_read = 0;
input_type offset_share = dpf.offset_x.compute_and_get_share(input_share);
bytes_written = ::asio::write(peer_out, ::asio::buffer(&offset_share, sizeof(input_type)), error);
if (error)
{
return std::make_tuple(offset_share, bytes_written, bytes_read);
}
bytes_read = ::asio::read(peer_in, ::asio::buffer(&offset_share, sizeof(input_type)), error);
if constexpr(detail::has_lowest_layer_v<PeerT>)
{
if (do_quickack) peer_in.get_lowest_layer().set_option(quickack_toggle);
}
if (!error)
{
offset_share = dpf.offset_x.reconstruct(offset_share);
}
return std::make_tuple(offset_share, bytes_written, bytes_read);
}
template <typename PeerT,
typename DpfKey,
typename InputType>
HEDLEY_ALWAYS_INLINE
auto assign_wildcard_input(PeerT & peer, DpfKey & dpf, InputType && input_share,
::asio::error_code & error)
{
return dpf::asio::assign_wildcard_input(peer, peer, dpf,
std::forward<InputType>(input_share), error);
}
template <typename PeerT,
typename DpfKey,
typename InputType>
HEDLEY_ALWAYS_INLINE
auto assign_wildcard_input(PeerT & peer_in, PeerT & peer_out, DpfKey & dpf,
InputType && input_share)
{
::asio::error_code error{};
auto ret = dpf::asio::assign_wildcard_input(peer_in, peer_out, dpf,
std::forward<InputType>(input_share), error);
if (error) throw error;
return ret;
}
template <typename PeerT,
typename DpfKey,
typename InputType>
HEDLEY_ALWAYS_INLINE
auto assign_wildcard_input(PeerT & peer, DpfKey & dpf, InputType && input_share)
{
::asio::error_code error{};
auto ret = dpf::asio::assign_wildcard_input(peer, dpf,
std::forward<InputType>(input_share), error);
if (error) throw error;
return ret;
}
//
// async_assign_wildcard_input
//
template <typename PeerT,
typename ExecutorT,
typename DpfKey,
typename InputType,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
auto async_assign_wildcard_input(PeerT & peer_in, PeerT & peer_out,
ExecutorT work_executor, DpfKey & dpf, InputType && input_share,
CompletionToken && token)
{
using input_type = typename DpfKey::input_type;
static_assert(std::is_convertible_v<InputType, input_type>);
#include <asio/yield.hpp>
return ::asio::async_compose<
CompletionToken, void(::asio::error_code, // error status
input_type, // offset
std::size_t, // bytes_written
std::size_t)>( // bytes_read
[
&peer_in,
&peer_out,
work_executor,
&dpf,
offset_share = std::make_shared<input_type>(input_share),
bytes_written = std::size_t(0),
bytes_read = std::size_t(0),
coro = ::asio::coroutine()
]
(
auto & self,
const ::asio::error_code & error = {},
std::size_t bytes_just_transmitted = 0
)
mutable
{
reenter (coro)
{
yield async_post(work_executor, [&dpf, offset_share]()
{
*offset_share = dpf.offset_x.compute_and_get_share(*offset_share);
}, std::move(self));
yield ::asio::async_write(peer_out,
::asio::buffer(offset_share.get(), sizeof(input_type)), std::move(self));
bytes_written = bytes_just_transmitted;
if (error)
{
self.complete(error, *offset_share, bytes_written, bytes_read);
break;
}
yield ::asio::async_read(peer_in,
::asio::buffer(offset_share.get(), sizeof(input_type)), std::move(self));
if constexpr(detail::has_lowest_layer_v<PeerT>)
{
if (do_quickack) peer_in.get_lowest_layer().set_option(quickack_toggle);
}
bytes_read = bytes_just_transmitted;
if (error)
{
self.complete(error, *offset_share, bytes_written, bytes_read);
break;
}
yield async_post(work_executor, [&dpf, offset_share]() mutable
{
*offset_share = dpf.offset_x.reconstruct(*offset_share);
}, std::move(self));
self.complete(error, *offset_share, bytes_written, bytes_read);
}
}, token, peer_in, peer_out, work_executor);
#include <asio/unyield.hpp>
}
template <typename PeerT,
typename ExecutorT,
typename DpfKey,
typename InputType,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
HEDLEY_ALWAYS_INLINE
auto async_assign_wildcard_input(PeerT & peer, ExecutorT work_executor,
DpfKey & dpf, InputType && input_share, CompletionToken && token)
{
return async_assign_wildcard_input(peer, peer, work_executor, dpf,
std::forward<InputType>(input_share),
std::forward<CompletionToken>(token));
}
template <typename PeerT,
typename DpfKey,
typename InputType,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
HEDLEY_ALWAYS_INLINE
auto async_assign_wildcard_input(PeerT & peer_in, PeerT & peer_out,
DpfKey & dpf, InputType && input_share, CompletionToken && token)
{
auto work_executor = ::asio::system_executor();
return async_assign_wildcard_input(peer_in, peer_out, work_executor, dpf,
std::forward<InputType>(input_share),
std::forward<CompletionToken>(token));
}
template <typename PeerT,
typename DpfKey,
typename InputType,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
HEDLEY_ALWAYS_INLINE
auto async_assign_wildcard_input(PeerT & peer, DpfKey & dpf, InputType && input_share, CompletionToken && token)
{
auto work_executor = ::asio::system_executor();
return async_assign_wildcard_input(peer, peer, work_executor, dpf,
std::forward<InputType>(input_share),
std::forward<CompletionToken>(token));
}
//
// assign_wildcard_output
//
template <std::size_t I = 0,
typename PeerT,
typename DpfKey,
typename OutputType>
auto assign_wildcard_output(PeerT & peer_in, PeerT & peer_out, DpfKey & dpf,
OutputType && output_share, ::asio::error_code & error)
{
using dpf_type = DpfKey;
using leaf_type = std::tuple_element_t<I, typename dpf_type::leaf_tuple>;
using output_type = typename dpf_type::concrete_output_type<I>;
static_assert(std::is_convertible_v<OutputType, output_type>);
std::size_t bytes_written = 0, bytes_read = 0;
leaf_type leaf_share;
constexpr bool is_packed = true;
auto & leaf_wrapper = utils::get<I>(dpf.leaf_nodes);
auto blinded_output = leaf_wrapper.compute_and_get_blinded_output_share(output_share);
bytes_written += ::asio::write(peer_out, ::asio::buffer(&blinded_output, sizeof(output_type)), error);
if (error)
{
return std::make_tuple(leaf_share, bytes_written, bytes_read);
}
bytes_read += ::asio::read(peer_in, ::asio::buffer(&blinded_output, sizeof(output_type)), error);
if constexpr(detail::has_lowest_layer_v<PeerT>)
{
if (do_quickack) peer_in.get_lowest_layer().set_option(quickack_toggle);
}
if (error)
{
return std::make_tuple(leaf_share, bytes_written, bytes_read);
}
leaf_share = leaf_wrapper.compute_and_get_leaf_share(blinded_output);
bytes_written += ::asio::write(peer_out, ::asio::buffer(&leaf_share, sizeof(leaf_type)), error);
if (error)
{
return std::make_tuple(leaf_share, bytes_written, bytes_read);
}
bytes_read += ::asio::read(peer_in, ::asio::buffer(&leaf_share, sizeof(leaf_type)), error);
if constexpr(detail::has_lowest_layer_v<PeerT>)
{
if (do_quickack) peer_in.get_lowest_layer().set_option(quickack_toggle);
}
if (!error)
{
leaf_share = leaf_wrapper.reconstruct_correction_word(leaf_share);
}
return std::make_tuple(leaf_share, bytes_written, bytes_read);
}
template <std::size_t I = 0,
typename PeerT,
typename DpfKey,
typename OutputType>
HEDLEY_ALWAYS_INLINE
auto assign_wildcard_output(PeerT & peer, DpfKey & dpf,
OutputType && output_share, ::asio::error_code & error)
{
return assign_wildcard_output<I>(peer, peer, dpf,
std::forward<OutputType>(output_share), error);
}
template <std::size_t I = 0,
typename PeerT,
typename DpfKey,
typename OutputType>
HEDLEY_ALWAYS_INLINE
auto assign_wildcard_output(PeerT & peer_in, PeerT & peer_out, DpfKey & dpf,
OutputType && output_share)
{
::asio::error_code error{};
auto ret = dpf::asio::assign_wildcard_output<I>(peer_in, peer_out, dpf,
std::forward<OutputType>(output_share), error);
if (error) throw error;
return ret;
}
template <std::size_t I = 0,
typename PeerT,
typename DpfKey,
typename OutputType>
HEDLEY_ALWAYS_INLINE
auto assign_wildcard_output(PeerT & peer, DpfKey & dpf, OutputType && output_share)
{
::asio::error_code error{};
auto ret = dpf::asio::assign_wildcard_output<I>(peer, peer, dpf,
std::forward<OutputType>(output_share), error);
if (error) throw error;
return ret;
}
//
// async_assign_wildcard_output
//
template <std::size_t I = 0,
typename PeerT,
typename ExecutorT,
typename DpfKey,
typename OutputType,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
auto async_assign_wildcard_output(PeerT & peer_in, PeerT & peer_out,
ExecutorT work_executor, DpfKey & dpf, OutputType && output_share,
CompletionToken && token)
{
using leaf_type = std::tuple_element_t<I, typename DpfKey::leaf_tuple>;
using output_type = typename DpfKey::concrete_output_type<I>;
static_assert(std::is_convertible_v<OutputType, output_type>);
#include <asio/yield.hpp>
return ::asio::async_compose<
CompletionToken, void(::asio::error_code, // error status
leaf_type, // assigned leaf
std::size_t, // bytes_written
std::size_t)>( // bytes_read
[
&peer_in,
&peer_out,
work_executor,
&leaf = utils::get<I>(dpf.leaf_nodes),
&dpf,
output_share = std::make_shared<output_type>(output_share),
leaf_share = std::make_shared<leaf_type>(),
bytes_written= std::size_t(0),
bytes_read= std::size_t(0),
coro = ::asio::coroutine()
]
(
auto & self,
const ::asio::error_code & error = {},
std::size_t bytes_just_transmitted = 0
)
mutable
{
reenter (coro)
{
yield async_post(work_executor, [&leaf, output_share]() mutable
{
*output_share = leaf.compute_and_get_blinded_output_share(*output_share);
}, std::move(self));
yield ::asio::async_write(peer_out,
::asio::buffer(output_share.get(), sizeof(output_type)), std::move(self));
bytes_written = bytes_just_transmitted;
if (error)
{
self.complete(error, *leaf_share, bytes_written, bytes_read);
break;
}
yield ::asio::async_read(peer_in,
::asio::buffer(output_share.get(), sizeof(output_type)), std::move(self));
if constexpr(detail::has_lowest_layer_v<PeerT>)
{
if (do_quickack) peer_in.get_lowest_layer().set_option(quickack_toggle);
}
bytes_read = bytes_just_transmitted;
if (error)
{
self.complete(error, *leaf_share, bytes_written, bytes_read);
break;
}
yield async_post(work_executor, [&leaf, leaf_share, output_share]() mutable
{
*leaf_share = leaf.compute_and_get_leaf_share(*output_share);
}, std::move(self));
yield ::asio::async_write(peer_out,
::asio::buffer(leaf_share.get(), sizeof(leaf_type)), std::move(self));
bytes_written += bytes_just_transmitted;
if (error)
{
self.complete(error, *leaf_share, bytes_written, bytes_read);
break;
}
yield ::asio::async_read(peer_in,
::asio::buffer(leaf_share.get(), sizeof(leaf_type)), std::move(self));
if constexpr(detail::has_lowest_layer_v<PeerT>)
{
if (do_quickack) peer_in.get_lowest_layer().set_option(quickack_toggle);
}
bytes_read += bytes_just_transmitted;
if (error)
{
self.complete(error, *leaf_share, bytes_written, bytes_read);
break;
}
yield async_post(work_executor, [&leaf, leaf_share]() mutable
{
*leaf_share = leaf.reconstruct_correction_word(*leaf_share);
}, std::move(self));
self.complete(error, *leaf_share, bytes_written, bytes_read);
}
},
token, peer_in, peer_out, work_executor);
#include <asio/unyield.hpp>
}
template <std::size_t I = 0,
typename PeerT,
typename ExecutorT,
typename DpfKey,
typename OutputType,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
HEDLEY_ALWAYS_INLINE
auto async_assign_wildcard_output(PeerT & peer, ExecutorT work_executor,
DpfKey & dpf, OutputType && output_share, CompletionToken && token)
{
return async_assign_wildcard_output<I>(peer, peer, work_executor, dpf,
std::forward<OutputType>(output_share),
std::forward<CompletionToken>(token));
}
template <std::size_t I = 0,
typename PeerT,
typename DpfKey,
typename OutputType,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
HEDLEY_ALWAYS_INLINE
auto async_assign_wildcard_output(PeerT & peer_in, PeerT & peer_out,
DpfKey & dpf, OutputType && output_share, CompletionToken && token)
{
auto work_executor = ::asio::system_executor();
return async_assign_wildcard_output<I>(peer_in, peer_out, work_executor, dpf,
std::forward<OutputType>(output_share),
std::forward<CompletionToken>(token));
}
template <std::size_t I = 0,
typename PeerT,
typename DpfKey,
typename OutputType,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
HEDLEY_ALWAYS_INLINE
auto async_assign_wildcard_output(PeerT & peer, DpfKey & dpf,
OutputType && output_share, CompletionToken && token)
{
auto work_executor = ::asio::system_executor();
return async_assign_wildcard_output<I>(peer, peer, dpf,
std::forward<OutputType>(output_share),
std::forward<CompletionToken>(token));
}
//
// make_interior_correction_word
//
template <typename DpfKey,
typename PeerT,
typename InteriorNode>
auto make_interior_correction_word(PeerT & peer, const InteriorNode & left,
const InteriorNode & right, dpf::bit dir, ::asio::error_code & error);
template <typename DpfKey,
typename PeerT,
typename InteriorNode>
HEDLEY_ALWAYS_INLINE
auto make_interior_correction_word(PeerT & peer, const InteriorNode & left,
const InteriorNode & right, dpf::bit dir)
{
::asio::error_code error{};
auto ret = dpf::asio::make_interior_correction_word(peer, left, right, dir, error);
if (error) throw error;
return ret;
}
//
// async_make_interior_correction_word
//
template <typename DpfKey,
typename PeerT,
typename ExecutorT,
typename InteriorNode,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
auto async_make_interior_correction_word(PeerT & peer, ExecutorT work_executor,
const InteriorNode & left, const InteriorNode & right, dpf::bit dir,
CompletionToken && token);
template <typename DpfKey,
typename PeerT,
typename ExecutorT,
typename InteriorNode,
typename CompletionToken>
HEDLEY_WARN_UNUSED_RESULT
HEDLEY_ALWAYS_INLINE
auto async_make_interior_correction_word(PeerT & peer, const InteriorNode & left,
const InteriorNode & right, dpf::bit dir, CompletionToken && token)
{
auto work_executor = ::asio::system_executor();
return async_make_interior_correction_word(peer, left, right, dir,
std::forward<CompletionToken>(token));
}
} // namespace asio
} // namespace dpf
#endif // LIBDPF_INCLUDE_DPF_ASIO_HPP__