84 lines
2.6 KiB
C++
84 lines
2.6 KiB
C++
|
/// @file examples/protocol/stream_app_smoke.cpp
|
||
|
|
/// @brief Thin app-runtime smoke: prep files or memory dealer + compose on streams.
|
||
|
|
/// @details Drives a composed open via `drive_both_on_streams` (the stream
|
||
|
|
/// framework). The same schedule runs unchanged over the event-driven
|
||
|
|
/// backends (`dpf::net::async_stream_array` + `async_round_sink` /
|
||
|
|
/// `dpf::async::overlapped_byte_protocol`) and, on Linux, over real
|
||
|
|
/// SCTP (`async_sctp_stream_array`). The `experiment_bench` harness
|
||
|
|
/// selects the transport with
|
||
|
|
/// `DPF_TRANSPORT=memory|stream|async|mux|parallel|sctp`.
|
||
|
|
#include <cstdint>
|
||
|
|
#include <cstring>
|
||
|
|
#include <iostream>
|
||
|
|
#include <vector>
|
||
|
|
|
||
|
|
#include "dpf/app_runtime.hpp"
|
||
|
|
#include "dpf/compose.hpp"
|
||
|
|
#include "dpf/prep_source.hpp"
|
||
|
|
#include "dpf/protocol_roles.hpp"
|
||
|
|
|
||
|
|
namespace
|
||
|
|
{
|
||
|
|
|
||
|
|
void put_raw(std::vector<std::vector<std::uint8_t>> & values,
|
||
|
|
dpf::protocol::node n, const void * src, std::size_t nbyte)
|
||
|
|
{
|
||
|
|
values[n.id].assign(nbyte, 0);
|
||
|
|
std::memcpy(values[n.id].data(), src, nbyte);
|
||
|
|
}
|
||
|
|
|
||
|
|
} // namespace
|
||
|
|
|
||
|
|
int main()
|
||
|
|
{
|
||
|
|
// Prep: file dealer round-trip (roles helper + app_runtime reader).
|
||
|
|
{
|
||
|
|
const std::string base = "/tmp/libdpf_stream_app_smoke_prep";
|
||
|
|
dpf::prep::demand d{};
|
||
|
|
d.ring_triples = 1;
|
||
|
|
dpf::roles::dealer_write_files(base, d);
|
||
|
|
auto c0 = dpf::app::open_file_prep_cursor(base, 0);
|
||
|
|
auto c1 = dpf::app::open_file_prep_cursor(base, 1);
|
||
|
|
if (c0.limb() != 8 || c1.limb() != 8)
|
||
|
|
{
|
||
|
|
std::cerr << "prep cursor\n";
|
||
|
|
return 1;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// Online: memory dealer + single-wave exchange plan on stream arrays.
|
||
|
|
dpf::protocol::composer c0(0);
|
||
|
|
dpf::protocol::composer c1(1);
|
||
|
|
auto x0 = c0.input(dpf::protocol::domain::a, 8);
|
||
|
|
auto x1 = c1.input(dpf::protocol::domain::a, 8);
|
||
|
|
auto e0 = c0.exchange(x0);
|
||
|
|
auto e1 = c1.exchange(x1);
|
||
|
|
auto p0 = c0.schedule();
|
||
|
|
auto p1 = c1.schedule();
|
||
|
|
|
||
|
|
dpf::prep::demand d{};
|
||
|
|
d.ring_triples = 1;
|
||
|
|
auto [d0, d1] = dpf::prep::deal_memory_pair(d);
|
||
|
|
(void)d0;
|
||
|
|
(void)d1;
|
||
|
|
|
||
|
|
std::vector<std::vector<std::uint8_t>> v0(p0.nodes().size());
|
||
|
|
std::vector<std::vector<std::uint8_t>> v1(p1.nodes().size());
|
||
|
|
const std::uint64_t a = 5, b = 9;
|
||
|
|
put_raw(v0, x0, &a, 8);
|
||
|
|
put_raw(v1, x1, &b, 8);
|
||
|
|
dpf::protocol::drive_both_on_streams(p0, p1, v0, v1);
|
||
|
|
|
||
|
|
std::uint64_t open0 = 0, open1 = 0;
|
||
|
|
std::memcpy(&open0, v0[e0.id].data(), 8);
|
||
|
|
std::memcpy(&open1, v1[e1.id].data(), 8);
|
||
|
|
if (open0 != 14u || open1 != 14u)
|
||
|
|
{
|
||
|
|
std::cerr << "exchange " << open0 << " " << open1 << "\n";
|
||
|
|
return 1;
|
||
|
|
}
|
||
|
|
|
||
|
|
std::cout << "ok\n";
|
||
|
|
return 0;
|
||
|
|
}
|