From e807e2c56b0bae4afab6205ba9057a689b321943 Mon Sep 17 00:00:00 2001 From: Erik Boasson Date: Mon, 12 Aug 2024 21:27:26 +0200 Subject: [PATCH] Add bidirectional pairs of processes This adds cwpl where processes communicate in pairs, each process publishing N topics in partition Pk and subscribing in Pj, where k is the specified id and j=k+1-2(k%2), so 0 and 1 form a pair, 2 and 3 form a pair, etc. A number of different types are available. Each writer has its own thread, readers record latency and optional write it to a file on termination. Signed-off-by: Erik Boasson --- examples/hop/CMakeLists.txt | 3 + examples/hop/cwpl.cpp | 266 ++++++++++++++++++++++++++++++++++++ 2 files changed, 269 insertions(+) create mode 100644 examples/hop/cwpl.cpp diff --git a/examples/hop/CMakeLists.txt b/examples/hop/CMakeLists.txt index fd5b1291..f6d2e5ab 100644 --- a/examples/hop/CMakeLists.txt +++ b/examples/hop/CMakeLists.txt @@ -27,3 +27,6 @@ target_link_libraries(hop CycloneDDS-CXX::ddscxx hop_type) add_executable(mop mop.cpp) target_link_libraries(mop CycloneDDS-CXX::ddscxx mop_type) + +add_executable(cwpl cwpl.cpp) +target_link_libraries(cwpl CycloneDDS-CXX::ddscxx mop_type) diff --git a/examples/hop/cwpl.cpp b/examples/hop/cwpl.cpp new file mode 100644 index 00000000..9cba9ce8 --- /dev/null +++ b/examples/hop/cwpl.cpp @@ -0,0 +1,266 @@ +/* + * Copyright(c) 2024 ZettaScale Technology and others + * + * This program and the accompanying materials are made available under the + * terms of the Eclipse Public License v. 2.0 which is available at + * http://www.eclipse.org/legal/epl-2.0, or the Eclipse Distribution License + * v. 1.0 which is available at + * http://www.eclipse.org/org/documents/edl-v10.php. + * + * SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "dds/dds.hpp" +#include "mop_type.hpp" + +using namespace org::eclipse::cyclonedds; +using namespace std::chrono_literals; +using namespace std::chrono; + +using CLK = high_resolution_clock; + +enum class Type { T8, T128, T1k, T8k, T128k }; + +static Type type = Type::T128; +static uint32_t pairid = 0; +static uint32_t ntopics = 10; +static std::optional datafile; + +static dds::core::Time mkDDSTime (const time_point x) +{ + int64_t t = duration_cast(x.time_since_epoch()).count(); + return dds::core::Time(t / 1000000000, static_cast(t % 1000000000)); +} + +static volatile std::atomic interrupted = false; +static void sigh(int sig) +{ + static_cast(sig); + interrupted = true; +} + +template +static dds::sub::DataReader make_reader(dds::topic::Topic tp) +{ + dds::domain::DomainParticipant dp = tp.domain_participant(); + std::vector spart{"P" + std::to_string(pairid + 1 - 2 * (pairid % 2))}; + dds::sub::qos::SubscriberQos sqos = dp.default_subscriber_qos() << dds::core::policy::Partition(spart); + dds::sub::Subscriber sub{dp, sqos}; + return dds::sub::DataReader{sub, tp, tp.qos()}; +} + +template +static dds::pub::DataWriter make_writer(dds::topic::Topic tp) +{ + dds::domain::DomainParticipant dp = tp.domain_participant(); + std::vector ppart{"P" + std::to_string(pairid)}; + dds::pub::qos::PublisherQos pqos = dp.default_publisher_qos() << dds::core::policy::Partition(ppart); + dds::pub::Publisher pub{dp, pqos}; + return dds::pub::DataWriter{pub, tp, tp.qos()}; +} + +template +static void source(dds::pub::DataWriter wr) +{ + T sample{}; + sample.k(0); + auto now = CLK::now(); + while (!interrupted) + { + wr.write(sample, mkDDSTime(CLK::now())); + ++sample.seq(); + now += 10ms; + std::this_thread::sleep_until(now); + } +} + +// t = reception time, l = latency, i = topic index, k = source key +struct TLK { int64_t t; double l; uint32_t k; }; +struct TLIK { int64_t t; double l; size_t i; uint32_t k; }; +struct LIK { double l; size_t i; uint32_t k; }; + +template +class Sink : public dds::sub::NoOpDataReaderListener { +public: + Sink() = default; + + const std::vector& lats() const { + return lats_; + }; + +private: + void on_data_available(dds::sub::DataReader& rd) + { + const auto now_clk = CLK::now(); + const int64_t now = duration_cast(now_clk.time_since_epoch()).count(); + auto xs = rd.take(); + for (const auto& x : xs) { + if (x.info().valid()) { + const auto lat = now - (x.info().timestamp().sec() * 1000000000 + x.info().timestamp().nanosec()); + lats_.push_back(TLK{now, lat / 1e3, x.data().k()}); + } else { + interrupted = true; + } + }; + } + + std::vector lats_; +}; + +template +static void run() +{ + dds::domain::DomainParticipant dp{0}; + auto tpqos = dp.default_topic_qos() + << dds::core::policy::Reliability::Reliable(dds::core::Duration::infinite()) + << dds::core::policy::History::KeepLast(1); + std::vector> tps; + std::vector> wrs; + for (uint32_t i = 0; i < ntopics; i++) + tps.push_back(dds::topic::Topic{dp, "Mop" + std::to_string(i), tpqos}); + for (auto& tp : tps) + wrs.push_back(make_writer(tp)); + std::vector> rds; + std::vector> ls; + for (size_t i = 0; i < tps.size(); i++) + ls.push_back(Sink{}); + for (size_t i = 0; i < tps.size(); i++) + rds.push_back(make_reader(tps[i])); + for (size_t i = 0; i < tps.size(); i++) + rds[i].listener(&ls[i], dds::core::status::StatusMask::data_available()); + + signal(SIGINT, sigh); + signal(SIGTERM, sigh); + std::vector threads; + for (auto wr : wrs) + threads.push_back(std::thread(source, wr)); + + // latencies in microseconds + std::vector lats; + while (!interrupted) + std::this_thread::sleep_for(103ms); + for (auto& t : threads) + t.join(); + for (auto rd : rds) + rd.close(); + for (auto wr : wrs) + wr.close(); + // collect latencies for all topics and sort by reception time + std::vector tlats; + for (size_t i = 0; i < ls.size(); i++) + for (const auto& x : ls[i].lats()) + tlats.push_back(TLIK{x.t, x.l, i, x.k}); + std::sort(tlats.begin(), tlats.end(), [](const TLIK& a, const TLIK& b) -> bool { return a.t < b.t; }); + // then reduce to just latency, topic and key + for (const auto& x : tlats) + lats.push_back(LIK{x.l, x.i, x.k}); + + if (datafile.has_value()) + { + std::ofstream f; + f.open(datafile.value()); + for (const auto& l : lats) + f << l.l << " " << l.i << " " << l.k << std::endl; + f.close(); + } + const size_t n = lats.size(); + if (n < 2) { + std::cout << "insufficient data" << std::endl; + } else { + std::sort(lats.begin(), lats.end(), [](const LIK& a, const LIK& b) -> bool { return a.l < b.l; }); + std::cout + << "received " << n + << " samples; min " << lats[0].l + << " max-1 " << lats[n-2].l + << " max " << lats[n-1].l << std::endl; + } +} + +[[noreturn]] +static void usage() +{ + std::cout + << "usage: cwpl [OPTIONS] id" << std::endl + << std::endl + << "OPTIONS:" << std::endl + << "-tTYPE type to use one of 8, 128 (def), 1k, 8k, 128k" << std::endl + << "-nNTPS use N (def = 10) topics in parallel" << std::endl + << "-oFILE write latencies to FILE" << std::endl + << std::endl + << "id = 0 writes in partition P0, reads from P1" << std::endl + << "id = 1 writes in partition P1, reads from P0" << std::endl + << "id = 2 writes in partition P2, reads from P3" << std::endl + << "etc." << std::endl; + std::exit(1); +} + +static Type convert_typestr (const std::string& typestr) +{ + if (typestr == "8") { + return Type::T8; + } else if (typestr == "128") { + return Type::T128; + } else if (typestr == "1k") { + return Type::T1k; + } else if (typestr == "8k") { + return Type::T8k; + } else if (typestr == "128k") { + return Type::T128k; + } else { + std::cout << "invalid type, should be 8, 128, 1k, 8k, 128k" << std::endl; + std::exit(1); + return Type::T128; + } +} + +int main (int argc, char **argv) +{ + if (argc < 1) + usage(); + + int opt; + while ((opt = getopt (argc, argv, "n:o:t:")) != EOF) + { + switch (opt) + { + case 'n': + ntopics = static_cast(std::atoi(optarg)); + break; + case 'o': + datafile = std::string(optarg); + break; + case 't': + type = convert_typestr(std::string(optarg)); + break; + default: + usage(); + } + } + if (argc - optind != 1) + { + usage(); + } + pairid = static_cast(std::atoi(argv[optind])); + switch (type) + { + case Type::T8: run(); break; + case Type::T128: run(); break; + case Type::T1k: run(); break; + case Type::T8k: run(); break; + case Type::T128k: run(); break; + } + return 0; +}