add EndOfStream signaling

Author: frs69wq

CMakeLists.txt

@@ -265,6 +265,7 @@ if(GTEST_LIBRARY)
       test/dtl_cancel.cpp
       test/dtl_config.cpp
       test/dtl_connection.cpp
+      test/dtl_end_of_stream.cpp
       test/dtl_file_engine.cpp
       test/dtl_reduction.cpp
       test/dtl_staging_engine.cpp

ChangeLog

@@ -3,11 +3,19 @@
 DTLMod (0.6) not released yet (target: September 2026)
 
 Improvements:
+  - End-of-stream signaling. Once the last publisher closes an engine, a
+    subscriber that requests a transaction which was never produced is now
+    released with a new EndOfStreamException instead of blocking forever
+    (mirroring the StepStatus::EndOfStream returned by ADIOS2's BeginStep).
+    This lets a consumer terminate when its upstream stops. Implemented for
+    both the Staging and File engines, for any number of subscribers. Exposed
+    in the Python bindings as EndOfStreamException.
   - New ReductionMethod::get_fidelity(var, transaction_id) returning the
     fraction of the original information retained after reduction, in [0, 1].
     Decimation reports the retained-element fraction; compression derives a
     fidelity from the accuracy bound (shape and reduced size are uninformative
     for it), so subscribers can reason about data quality, not just volume.
+    Exposed in the Python bindings as ReductionMethod.get_fidelity().
   - New Engine::cancel_pending_activities() and Engine::drain() helpers, now
     used by every transaction teardown path of the Staging and File engines,
     so that cancelling and emptying an ActivitySet is done in one place

include/dtlmod/DTLException.hpp

@@ -71,6 +71,7 @@ DECLARE_DTLMOD_EXCEPTION(InconsistentCompressionRatioException, "Inconsistent Co
 DECLARE_DTLMOD_EXCEPTION(SubscriberSideCompressionException, "Compression can only be applied on the publisher side");
 
 DECLARE_DTLMOD_EXCEPTION(TransactionCanceledException, "Transaction canceled");
+DECLARE_DTLMOD_EXCEPTION(EndOfStreamException, "End of stream: all publishers have closed");
 
 } // namespace dtlmod
 

include/dtlmod/Engine.hpp

@@ -58,6 +58,7 @@ class Engine {
 
   bool pub_ever_present_ = false;
   std::atomic<unsigned int> canceled_transaction_id_{0};
+  std::atomic<bool> pub_stream_ended_{false};
 
   ActorRegistry publishers_;
 
@@ -113,6 +114,11 @@ class Engine {
     return canceled_transaction_id_ == tx_id;
   }
 
+  // Set once the last publisher has closed the engine: no further transaction will ever be produced, so a subscriber
+  // waiting for one that was never produced must be released with an EndOfStreamException rather than block forever.
+  void mark_pub_stream_ended() noexcept { pub_stream_ended_ = true; }
+  [[nodiscard]] bool pub_stream_ended() const noexcept { return pub_stream_ended_; }
+
   // Pure virtual methods for derived classes to implement
   virtual void create_transport(const Transport::Method& transport_method) = 0;
   virtual void begin_pub_transaction() = 0;

src/FileEngine.cpp

@@ -186,6 +186,9 @@ void FileEngine::pub_close()
     transport->close_pub_files();
     XBT_DEBUG("Engine '%s' is now closed for all publishers ", get_cname());
     get_stream()->export_metadata_to_file();
+    // No more transactions will ever be produced: release any subscriber blocked waiting for one.
+    mark_pub_stream_ended();
+    pub_transaction_completed_->notify_all();
   }
 }
 
@@ -205,15 +208,24 @@ void FileEngine::begin_sub_transaction()
   if (not get_publishers().is_empty()) {
     std::unique_lock lock(*get_subscribers().get_mutex());
     while (!is_transaction_canceled(current_sub_transaction_id_) &&
-           completed_pub_transaction_id_ < current_sub_transaction_id_) {
+           completed_pub_transaction_id_ < current_sub_transaction_id_ && !pub_stream_ended()) {
       XBT_DEBUG("Wait for publishers to end the transaction I need");
       pub_transaction_completed_->wait(lock);
     }
     if (is_transaction_canceled(current_sub_transaction_id_)) {
       sub_transaction_in_progress_ = false;
       throw TransactionCanceledException(XBT_THROW_POINT);
     }
+    if (completed_pub_transaction_id_ < current_sub_transaction_id_ && pub_stream_ended()) {
+      sub_transaction_in_progress_ = false;
+      throw EndOfStreamException(XBT_THROW_POINT);
+    }
     XBT_DEBUG("Publishers stored metadata for that transaction, proceed");
+  } else if (pub_stream_ended() && completed_pub_transaction_id_ < current_sub_transaction_id_) {
+    // All publishers have already closed (so the wait above is skipped) and the transaction we want was never
+    // produced. Without this the subscriber would fall through and read data that does not exist.
+    sub_transaction_in_progress_ = false;
+    throw EndOfStreamException(XBT_THROW_POINT);
   }
 }
 

src/StagingEngine.cpp

@@ -156,6 +156,10 @@ void StagingEngine::pub_close()
     close_stream();
     XBT_DEBUG("Engine '%s' is now closed for all publishers ", get_cname());
     get_stream()->export_metadata_to_file();
+    // No more transactions will ever be produced: release any subscriber blocked (or about to block) waiting for one.
+    mark_pub_stream_ended();
+    first_pub_transaction_started_->notify_all();
+    pub_transaction_completed_->notify_all();
   }
 }
 
@@ -167,10 +171,14 @@ void StagingEngine::begin_sub_transaction()
   if (current_sub_transaction_id_ == 0) { // This is the first transaction
     // Wait for at least one publisher to start a tran
     std::unique_lock lock(*get_subscribers().get_mutex());
-    while (!is_transaction_canceled(current_sub_transaction_id_ + 1) && current_pub_transaction_id_ == 0)
+    while (!is_transaction_canceled(current_sub_transaction_id_ + 1) && current_pub_transaction_id_ == 0 &&
+           !pub_stream_ended())
       first_pub_transaction_started_->wait(lock);
     if (is_transaction_canceled(current_sub_transaction_id_ + 1))
       throw TransactionCanceledException(XBT_THROW_POINT);
+    // All publishers closed before ever starting a transaction: nothing will come (no counters touched yet).
+    if (current_pub_transaction_id_ == 0 && pub_stream_ended())
+      throw EndOfStreamException(XBT_THROW_POINT);
     XBT_DEBUG("Publishers have started a transaction, create rendez-vous points");
     // We now know the number of publishers, subscriber can create mailboxes/mqs with publishers
     get_staging_transport()->create_rendez_vous_points();
@@ -194,13 +202,20 @@ void StagingEngine::begin_sub_transaction()
 
   std::unique_lock lock(*get_subscribers().get_mutex());
   while (!is_transaction_canceled(current_sub_transaction_id_) &&
-         completed_pub_transaction_id_ < current_sub_transaction_id_)
+         completed_pub_transaction_id_ < current_sub_transaction_id_ && !pub_stream_ended())
     pub_transaction_completed_->wait(lock);
   if (is_transaction_canceled(current_sub_transaction_id_)) {
     sub_transaction_in_progress_ = false;
     num_subscribers_starting_--;
     throw TransactionCanceledException(XBT_THROW_POINT);
   }
+  // Woken by the last publisher closing while the transaction we want was never produced: end of stream. Roll back the
+  // same per-subscriber bookkeeping as the cancel path so the shared counters stay balanced across subscribers.
+  if (completed_pub_transaction_id_ < current_sub_transaction_id_ && pub_stream_ended()) {
+    sub_transaction_in_progress_ = false;
+    num_subscribers_starting_--;
+    throw EndOfStreamException(XBT_THROW_POINT);
+  }
 }
 
 void StagingEngine::end_sub_transaction()

src/bindings/python/dtlmod_python.cpp

@@ -90,6 +90,7 @@ PYBIND11_MODULE(dtlmod, m)
   py::register_exception<dtlmod::SubscriberSideCompressionException>(m, "SubscriberSideCompressionException");
 
   py::register_exception<dtlmod::TransactionCanceledException>(m, "TransactionCanceledException");
+  py::register_exception<dtlmod::EndOfStreamException>(m, "EndOfStreamException");
 
   /* Class Engine */
   py::class_<Engine, std::shared_ptr<Engine>> engine(
@@ -260,6 +261,8 @@ PYBIND11_MODULE(dtlmod, m)
            py::arg("transaction_id") = 0, "Get the reduced local size of a Variable")
       .def("get_reduced_variable_shape", &ReductionMethod::get_reduced_variable_shape, py::arg("var"),
            "Get the reduced shape of a Variable")
+      .def("get_fidelity", &ReductionMethod::get_fidelity, py::arg("var"), py::arg("transaction_id") = 0,
+           "Get the fraction of the original information retained after reduction, in [0, 1] (1.0 = lossless)")
       .def("get_flop_amount_to_reduce_variable", &ReductionMethod::get_flop_amount_to_reduce_variable, py::arg("var"),
            "Get the flop cost to reduce a Variable")
       .def("get_flop_amount_to_decompress_variable", &ReductionMethod::get_flop_amount_to_decompress_variable,

test/dtl_end_of_stream.cpp

@@ -0,0 +1,220 @@
+/* Copyright (c) 2026. The SWAT Team. All rights reserved.          */
+
+/* This program is free software; you can redistribute it and/or modify it
+ * under the terms of the license (GNU LGPL) which comes with this package. */
+
+#include <gtest/gtest.h>
+
+#include <array>
+
+#include <fsmod/FileSystem.hpp>
+#include <fsmod/JBODStorage.hpp>
+#include <fsmod/OneDiskStorage.hpp>
+
+#include <simgrid/s4u/Actor.hpp>
+#include <simgrid/s4u/Engine.hpp>
+#include <simgrid/s4u/Host.hpp>
+
+#include "./test_util.hpp"
+#include "dtlmod/DTL.hpp"
+#include "dtlmod/DTLException.hpp"
+
+XBT_LOG_NEW_DEFAULT_CATEGORY(dtlmod_test_eos, "Logging category for this dtlmod test");
+
+namespace sg4  = simgrid::s4u;
+namespace sgfs = simgrid::fsmod;
+
+// End-of-stream: once every publisher has closed its engine, a subscriber that asks for a transaction which was never
+// produced must be released with an EndOfStreamException (instead of blocking forever). This is what lets a consumer
+// terminate when its upstream stops, rather than hang. Each test has a publisher produce a fixed number of
+// transactions and then close; the subscriber loops until it catches the exception.
+//
+// The per-subscriber outcome (how many transactions were read, whether end-of-stream was reached) is recorded in
+// variables captured by reference and asserted AFTER Engine::run() returns -- never inside the subscriber actor. If
+// the mechanism were missing the subscriber would deadlock, run() would return with the actor still blocked, and an
+// in-actor assertion would simply never execute (a false pass). Asserting after run() turns that hang into a failure.
+class DTLEndOfStreamTest : public ::testing::Test {
+public:
+  DTLEndOfStreamTest() = default;
+
+  sg4::NetZone* add_cluster(sg4::NetZone* root, const std::string& suffix, const int num_hosts)
+  {
+    auto* cluster = root->add_netzone_star("cluster" + suffix);
+    cluster->set_gateway(cluster->add_router("cluster" + suffix + "-router"));
+    auto* backbone = cluster->add_link("backbone" + suffix, "100Gbps")->set_latency("100us");
+    for (int i = 0; i < num_hosts; i++) {
+      std::string name = "host-" + std::to_string(i) + suffix;
+      const auto* host = cluster->add_host(name, "1Gf");
+      const auto* link = cluster->add_link(name + "_link", "10Gbps")->set_latency("10us");
+      cluster->add_route(host, nullptr, {link, backbone});
+    }
+    cluster->seal();
+    return cluster;
+  }
+
+  void setup_staging_platform()
+  {
+    auto* root         = sg4::Engine::get_instance()->get_netzone_root();
+    auto* internet     = root->add_link("internet", "500MBps")->set_latency("1ms");
+    auto* prod_cluster = add_cluster(root, ".prod", 4);
+    auto* cons_cluster = add_cluster(root, ".cons", 4);
+    root->add_route(prod_cluster, cons_cluster, {internet});
+    root->seal();
+    dtlmod::DTL::create();
+  }
+
+  void setup_file_platform()
+  {
+    sg4::NetZone* cluster = sg4::Engine::get_instance()->get_netzone_root()->add_netzone_star("cluster");
+    auto pfs_server       = cluster->add_host("pfs_server", "1Gf");
+    std::vector<sg4::Disk*> pfs_disks;
+    for (int i = 0; i < 4; i++)
+      pfs_disks.push_back(pfs_server->add_disk("pfs_disk" + std::to_string(i), "2.5GBps", "1.2GBps"));
+    auto remote_storage = sgfs::JBODStorage::create("pfs_storage", pfs_disks);
+    remote_storage->set_raid_level(sgfs::JBODStorage::RAID::RAID5);
+
+    std::vector<std::shared_ptr<sgfs::OneDiskStorage>> local_storages;
+    for (int i = 0; i < 4; i++) {
+      std::string hostname = "node-" + std::to_string(i);
+      auto* host           = cluster->add_host(hostname, "1Gf");
+      auto* disk           = host->add_disk(hostname + "_disk", "5.5GBps", "2.1GBps");
+      local_storages.push_back(sgfs::OneDiskStorage::create(hostname + "_local_storage", disk));
+      std::string linkname = "link_" + std::to_string(i);
+      auto* link_up        = cluster->add_link(linkname + "_UP", "1Gbps");
+      auto* link_down      = cluster->add_link(linkname + "_DOWN", "1Gbps");
+      auto* loopback =
+          cluster->add_link(hostname + "_loopback", "10Gbps")->set_sharing_policy(sg4::Link::SharingPolicy::FATPIPE);
+      cluster->add_route(host, nullptr, {sg4::LinkInRoute(link_up)}, false);
+      cluster->add_route(nullptr, host, {sg4::LinkInRoute(link_down)}, false);
+      cluster->add_route(host, host, {loopback});
+    }
+    cluster->seal();
+
+    auto my_fs = sgfs::FileSystem::create("my_fs");
+    sgfs::FileSystem::register_file_system(cluster, my_fs);
+    my_fs->mount_partition("/pfs/", remote_storage, "500TB");
+    for (int i = 0; i < 4; i++)
+      my_fs->mount_partition("/node-" + std::to_string(i) + "/scratch/", local_storages.at(i), "1TB");
+
+    dtlmod::DTL::create();
+  }
+
+  // Publisher actor body shared by the tests: produce n_tx transactions then close.
+  static void publish_n(dtlmod::Engine::Type type, dtlmod::Transport::Method method, const std::string& engine_name,
+                        int n_tx)
+  {
+    auto dtl    = dtlmod::DTL::connect();
+    auto stream = dtl->add_stream("my-output");
+    stream->set_engine_type(type);
+    stream->set_transport_method(method);
+    auto var    = stream->define_variable("var", {100, 100}, {0, 0}, {100, 100}, sizeof(double));
+    auto engine = stream->open(engine_name, dtlmod::Stream::Mode::Publish);
+    for (int i = 0; i < n_tx; i++) {
+      engine->begin_transaction();
+      engine->put(var);
+      engine->end_transaction();
+    }
+    engine->close();
+    dtlmod::DTL::disconnect();
+  }
+
+  // Subscriber actor body: read until end-of-stream, recording the outcome through the referenced variables.
+  static void consume_until_eos(const std::string& engine_name, int& reads, bool& eos)
+  {
+    auto dtl     = dtlmod::DTL::connect();
+    auto stream  = dtl->add_stream("my-output");
+    auto engine  = stream->open(engine_name, dtlmod::Stream::Mode::Subscribe);
+    auto var_sub = stream->inquire_variable("var");
+    var_sub->set_selection({0, 0}, {100, 100});
+    try {
+      while (true) {
+        engine->begin_transaction();
+        engine->get(var_sub);
+        engine->end_transaction();
+        reads++;
+      }
+    } catch (const dtlmod::EndOfStreamException&) {
+      eos = true;
+    }
+    engine->close();
+    dtlmod::DTL::disconnect();
+  }
+};
+
+TEST_F(DTLEndOfStreamTest, StagingSingleSubscriber_MQ)
+{
+  DO_TEST_WITH_FORK([this]() {
+    this->setup_staging_platform();
+    int reads = 0;
+    bool eos  = false;
+    sg4::Host::by_name("host-0.prod")->add_actor("Pub", []() {
+      publish_n(dtlmod::Engine::Type::Staging, dtlmod::Transport::Method::MQ, "my-output", 2);
+    });
+    sg4::Host::by_name("host-0.cons")->add_actor("Sub", [&reads, &eos]() {
+      consume_until_eos("my-output", reads, eos);
+    });
+    ASSERT_NO_THROW(sg4::Engine::get_instance()->run());
+    ASSERT_TRUE(eos);
+    ASSERT_EQ(reads, 2);
+  });
+}
+
+TEST_F(DTLEndOfStreamTest, StagingSingleSubscriber_Mailbox)
+{
+  DO_TEST_WITH_FORK([this]() {
+    this->setup_staging_platform();
+    int reads = 0;
+    bool eos  = false;
+    sg4::Host::by_name("host-0.prod")->add_actor("Pub", []() {
+      publish_n(dtlmod::Engine::Type::Staging, dtlmod::Transport::Method::Mailbox, "my-output", 3);
+    });
+    sg4::Host::by_name("host-0.cons")->add_actor("Sub", [&reads, &eos]() {
+      consume_until_eos("my-output", reads, eos);
+    });
+    ASSERT_NO_THROW(sg4::Engine::get_instance()->run());
+    ASSERT_TRUE(eos);
+    ASSERT_EQ(reads, 3);
+  });
+}
+
+// Two subscribers sharing the same Staging engine must both reach end-of-stream. This is the case that exercises the
+// per-subscriber rollback of num_subscribers_starting_ on the EOS throw: an imbalance there would desynchronize the
+// publisher/subscriber rendez-vous and either deadlock or crash.
+TEST_F(DTLEndOfStreamTest, StagingMultipleSubscribers_MQ)
+{
+  DO_TEST_WITH_FORK([this]() {
+    this->setup_staging_platform();
+    std::array<int, 2> reads = {0, 0};
+    std::array<bool, 2> eos  = {false, false};
+    sg4::Host::by_name("host-0.prod")->add_actor("Pub", []() {
+      publish_n(dtlmod::Engine::Type::Staging, dtlmod::Transport::Method::MQ, "my-output", 2);
+    });
+    for (int s = 0; s < 2; s++)
+      sg4::Host::by_name("host-" + std::to_string(s) + ".cons")
+          ->add_actor("Sub" + std::to_string(s),
+                      [&reads, &eos, s]() { consume_until_eos("my-output", reads[s], eos[s]); });
+    ASSERT_NO_THROW(sg4::Engine::get_instance()->run());
+    for (int s = 0; s < 2; s++) {
+      ASSERT_TRUE(eos[s]) << "subscriber " << s << " did not reach end of stream";
+      ASSERT_EQ(reads[s], 2) << "subscriber " << s << " read a wrong number of transactions";
+    }
+  });
+}
+
+TEST_F(DTLEndOfStreamTest, FileEngineSingleSubscriber)
+{
+  DO_TEST_WITH_FORK([this]() {
+    this->setup_file_platform();
+    int reads                     = 0;
+    bool eos                      = false;
+    const std::string engine_name = "cluster:my_fs:/node-0/scratch/my-output";
+    sg4::Host::by_name("node-0")->add_actor("Pub", [engine_name]() {
+      publish_n(dtlmod::Engine::Type::File, dtlmod::Transport::Method::File, engine_name, 2);
+    });
+    sg4::Host::by_name("node-1")->add_actor(
+        "Sub", [&reads, &eos, engine_name]() { consume_until_eos(engine_name, reads, eos); });
+    ASSERT_NO_THROW(sg4::Engine::get_instance()->run());
+    ASSERT_TRUE(eos);
+    ASSERT_EQ(reads, 2);
+  });
+}

test/python/dtl_cancel.py

@@ -348,6 +348,7 @@ def canceller_actor():
         run_test_cancel_staging_mid_transaction_mailbox,
     ]
 
+    all_passed = True
     for test in tests:
         print(f"\nRun {test.__name__} ...")
         p = multiprocessing.Process(target=test)
@@ -356,5 +357,9 @@ def canceller_actor():
 
         if p.exitcode != 0:
             print(f"FAILED: {test.__name__} (exit code {p.exitcode})")
+            all_passed = False
         else:
             print(f"PASSED: {test.__name__}")
+
+    if not all_passed:
+        sys.exit(1)