Skip to content

Commit e024e80

Browse files
frs69wqfrs69wq
committed
add tests of reduction with staging engine
1 parent 948c53b commit e024e80

2 files changed

Lines changed: 267 additions & 1 deletion

File tree

‎test/dtl_reduction.cpp‎

Lines changed: 133 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -453,6 +453,139 @@ TEST_F(DTLReductionTest, CompressionWithDerivedRatio)
453453
});
454454
}
455455

456+
TEST_F(DTLReductionTest, DecimationStagingEngine)
457+
{
458+
DO_TEST_WITH_FORK([this]() {
459+
// Build a two-host platform with a network link (required for staging transport)
460+
auto* zone = sg4::Engine::get_instance()->get_netzone_root()->add_netzone_star("zone");
461+
auto* pub_host = zone->add_host("pub_host", "6Gf");
462+
auto* sub_host = zone->add_host("sub_host", "6Gf");
463+
auto* backbone = zone->add_link("backbone", "10Gbps")->set_latency("10us");
464+
auto* link_pub = zone->add_link("link_pub", "10Gbps")->set_latency("10us");
465+
auto* link_sub = zone->add_link("link_sub", "10Gbps")->set_latency("10us");
466+
zone->add_route(pub_host, nullptr, std::vector<const sg4::Link*>{link_pub, backbone});
467+
zone->add_route(sub_host, nullptr, std::vector<const sg4::Link*>{link_sub, backbone});
468+
zone->seal();
469+
dtlmod::DTL::create();
470+
471+
pub_host->add_actor("Publisher", []() {
472+
auto dtl = dtlmod::DTL::connect();
473+
auto stream = dtl->add_stream("my-output");
474+
stream->set_engine_type(dtlmod::Engine::Type::Staging);
475+
stream->set_transport_method(dtlmod::Transport::Method::MQ);
476+
XBT_INFO("Create a 2D variable with 10kx10k doubles");
477+
auto var = stream->define_variable("var", {10000, 10000}, {0, 0}, {10000, 10000}, sizeof(double));
478+
auto decimator = stream->define_reduction_method("decimation");
479+
auto engine = stream->open("my-output", dtlmod::Stream::Mode::Publish);
480+
sg4::this_actor::sleep_for(0.5);
481+
482+
XBT_INFO("Assign decimation with stride 2,2");
483+
ASSERT_NO_THROW(var->set_reduction_operation(decimator, {{"stride", "2,2"}}));
484+
ASSERT_TRUE(var->is_reduced());
485+
ASSERT_TRUE(var->is_reduced_by_publisher());
486+
487+
XBT_INFO("Verify reduced shape: 5000x5000");
488+
auto shape = decimator->get_reduced_variable_shape(*var);
489+
ASSERT_EQ(shape[0], 5000u);
490+
ASSERT_EQ(shape[1], 5000u);
491+
492+
engine->begin_transaction();
493+
ASSERT_NO_THROW(engine->put(var));
494+
engine->end_transaction();
495+
sg4::this_actor::sleep_for(1);
496+
engine->close();
497+
dtlmod::DTL::disconnect();
498+
});
499+
500+
sub_host->add_actor("Subscriber", []() {
501+
auto dtl = dtlmod::DTL::connect();
502+
auto stream = dtl->add_stream("my-output");
503+
auto engine = stream->open("my-output", dtlmod::Stream::Mode::Subscribe);
504+
auto var = stream->inquire_variable("var");
505+
506+
XBT_INFO("Get the decimated variable");
507+
engine->begin_transaction();
508+
ASSERT_NO_THROW(engine->get(var));
509+
engine->end_transaction();
510+
511+
engine->close();
512+
dtlmod::DTL::disconnect();
513+
});
514+
515+
ASSERT_NO_THROW(sg4::Engine::get_instance()->run());
516+
});
517+
}
518+
519+
TEST_F(DTLReductionTest, CompressionStagingEngine)
520+
{
521+
DO_TEST_WITH_FORK([this]() {
522+
// Build a two-host platform with a network link (required for staging transport)
523+
auto* zone = sg4::Engine::get_instance()->get_netzone_root()->add_netzone_star("zone");
524+
auto* pub_host = zone->add_host("pub_host", "6Gf");
525+
auto* sub_host = zone->add_host("sub_host", "6Gf");
526+
auto* backbone = zone->add_link("backbone", "10Gbps")->set_latency("10us");
527+
auto* link_pub = zone->add_link("link_pub", "10Gbps")->set_latency("10us");
528+
auto* link_sub = zone->add_link("link_sub", "10Gbps")->set_latency("10us");
529+
zone->add_route(pub_host, nullptr, std::vector<const sg4::Link*>{link_pub, backbone});
530+
zone->add_route(sub_host, nullptr, std::vector<const sg4::Link*>{link_sub, backbone});
531+
zone->seal();
532+
dtlmod::DTL::create();
533+
534+
pub_host->add_actor("Publisher", []() {
535+
auto dtl = dtlmod::DTL::connect();
536+
auto stream = dtl->add_stream("my-output");
537+
stream->set_engine_type(dtlmod::Engine::Type::Staging);
538+
stream->set_transport_method(dtlmod::Transport::Method::MQ);
539+
XBT_INFO("Create a 2D variable with 10kx10k doubles");
540+
auto var = stream->define_variable("var", {10000, 10000}, {0, 0}, {10000, 10000}, sizeof(double));
541+
auto compressor = stream->define_reduction_method("compression");
542+
auto engine = stream->open("my-output", dtlmod::Stream::Mode::Publish);
543+
sg4::this_actor::sleep_for(0.5);
544+
545+
XBT_INFO("Assign compression with ratio 5 and explicit costs");
546+
ASSERT_NO_THROW(var->set_reduction_operation(compressor, {{"compression_ratio", "5"},
547+
{"compression_cost_per_element", "3"},
548+
{"decompression_cost_per_element", "1"}}));
549+
ASSERT_TRUE(var->is_reduced());
550+
ASSERT_TRUE(var->is_reduced_by_publisher());
551+
552+
XBT_INFO("Verify compressed sizes");
553+
size_t expected_reduced = static_cast<size_t>(std::ceil(sizeof(double) * 10000.0 * 10000.0 / 5.0));
554+
ASSERT_EQ(compressor->get_reduced_variable_global_size(*var), expected_reduced);
555+
ASSERT_EQ(compressor->get_reduced_variable_local_size(*var), expected_reduced);
556+
557+
XBT_INFO("Verify shape is unchanged (compression preserves shape)");
558+
auto shape = compressor->get_reduced_variable_shape(*var);
559+
ASSERT_EQ(shape[0], 10000u);
560+
ASSERT_EQ(shape[1], 10000u);
561+
562+
engine->begin_transaction();
563+
ASSERT_NO_THROW(engine->put(var));
564+
engine->end_transaction();
565+
sg4::this_actor::sleep_for(1);
566+
engine->close();
567+
dtlmod::DTL::disconnect();
568+
});
569+
570+
sub_host->add_actor("Subscriber", []() {
571+
auto dtl = dtlmod::DTL::connect();
572+
auto stream = dtl->add_stream("my-output");
573+
auto engine = stream->open("my-output", dtlmod::Stream::Mode::Subscribe);
574+
auto var = stream->inquire_variable("var");
575+
576+
XBT_INFO("Get the compressed variable (decompression cost should be applied on subscriber)");
577+
engine->begin_transaction();
578+
ASSERT_NO_THROW(engine->get(var));
579+
engine->end_transaction();
580+
581+
engine->close();
582+
dtlmod::DTL::disconnect();
583+
});
584+
585+
ASSERT_NO_THROW(sg4::Engine::get_instance()->run());
586+
});
587+
}
588+
456589
TEST_F(DTLReductionTest, DoubleReductionForbidden)
457590
{
458591
DO_TEST_WITH_FORK([this]() {

‎test/python/dtl_reduction.py‎

Lines changed: 134 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
import sys
99
import multiprocessing
1010

11-
from simgrid import Engine, this_actor
11+
from simgrid import Engine, Host, this_actor
1212
from fsmod import FileSystem, OneDiskStorage
1313
from dtlmod import (DTL, Engine as DTLEngine, Stream, Transport,
1414
UnknownReductionMethodException,
@@ -429,6 +429,137 @@ def test_actor():
429429
e.run()
430430

431431

432+
def setup_staging_platform():
433+
"""Set up a two-host platform with network links for staging engine tests."""
434+
e = Engine(sys.argv)
435+
e.set_log_control("no_loc")
436+
e.set_log_control("root.thresh:critical")
437+
438+
zone = e.netzone_root.add_netzone_star("zone")
439+
pub_host = zone.add_host("pub_host", "6Gf")
440+
sub_host = zone.add_host("sub_host", "6Gf")
441+
backbone = zone.add_link("backbone", "10Gbps").set_latency("10us")
442+
link_pub = zone.add_link("link_pub", "10Gbps").set_latency("10us")
443+
link_sub = zone.add_link("link_sub", "10Gbps").set_latency("10us")
444+
zone.add_route(pub_host, None, [link_pub, backbone])
445+
zone.add_route(sub_host, None, [link_sub, backbone])
446+
zone.seal()
447+
448+
DTL.create()
449+
return e
450+
451+
452+
def run_test_decimation_staging_engine():
453+
e = setup_staging_platform()
454+
pub_host = Host.by_name("pub_host")
455+
sub_host = Host.by_name("sub_host")
456+
457+
def publisher():
458+
dtl = DTL.connect()
459+
stream = dtl.add_stream("my-output")
460+
stream.set_engine_type(DTLEngine.Type.Staging)
461+
stream.set_transport_method(Transport.Method.MQ)
462+
this_actor.info("Create a 2D variable with 10kx10k doubles")
463+
var = stream.define_variable("var", (10000, 10000), (0, 0), (10000, 10000), ctypes.sizeof(ctypes.c_double))
464+
decimator = stream.define_reduction_method("decimation")
465+
engine = stream.open("my-output", Stream.Mode.Publish)
466+
this_actor.sleep_for(0.5)
467+
468+
this_actor.info("Assign decimation with stride 2,2")
469+
var.set_reduction_operation(decimator, {"stride": "2,2"})
470+
assert var.is_reduced
471+
assert var.is_reduced_by_publisher
472+
473+
this_actor.info("Verify reduced shape: 5000x5000")
474+
shape = decimator.get_reduced_variable_shape(var)
475+
assert shape[0] == 5000
476+
assert shape[1] == 5000
477+
478+
engine.begin_transaction()
479+
engine.put(var)
480+
engine.end_transaction()
481+
this_actor.sleep_for(1)
482+
engine.close()
483+
DTL.disconnect()
484+
485+
def subscriber():
486+
dtl = DTL.connect()
487+
stream = dtl.add_stream("my-output")
488+
engine = stream.open("my-output", Stream.Mode.Subscribe)
489+
var = stream.inquire_variable("var")
490+
491+
this_actor.info("Get the decimated variable")
492+
engine.begin_transaction()
493+
engine.get(var)
494+
engine.end_transaction()
495+
496+
engine.close()
497+
DTL.disconnect()
498+
499+
pub_host.add_actor("Publisher", publisher)
500+
sub_host.add_actor("Subscriber", subscriber)
501+
e.run()
502+
503+
504+
def run_test_compression_staging_engine():
505+
e = setup_staging_platform()
506+
pub_host = Host.by_name("pub_host")
507+
sub_host = Host.by_name("sub_host")
508+
509+
def publisher():
510+
dtl = DTL.connect()
511+
stream = dtl.add_stream("my-output")
512+
stream.set_engine_type(DTLEngine.Type.Staging)
513+
stream.set_transport_method(Transport.Method.MQ)
514+
this_actor.info("Create a 2D variable with 10kx10k doubles")
515+
var = stream.define_variable("var", (10000, 10000), (0, 0), (10000, 10000), ctypes.sizeof(ctypes.c_double))
516+
compressor = stream.define_reduction_method("compression")
517+
engine = stream.open("my-output", Stream.Mode.Publish)
518+
this_actor.sleep_for(0.5)
519+
520+
this_actor.info("Assign compression with ratio 5 and explicit costs")
521+
var.set_reduction_operation(compressor, {"compression_ratio": "5",
522+
"compression_cost_per_element": "3",
523+
"decompression_cost_per_element": "1"})
524+
assert var.is_reduced
525+
assert var.is_reduced_by_publisher
526+
527+
this_actor.info("Verify compressed sizes")
528+
expected_reduced = math.ceil(ctypes.sizeof(ctypes.c_double) * 10000.0 * 10000.0 / 5.0)
529+
assert compressor.get_reduced_variable_global_size(var) == expected_reduced
530+
assert compressor.get_reduced_variable_local_size(var) == expected_reduced
531+
532+
this_actor.info("Verify shape is unchanged")
533+
shape = compressor.get_reduced_variable_shape(var)
534+
assert shape[0] == 10000
535+
assert shape[1] == 10000
536+
537+
engine.begin_transaction()
538+
engine.put(var)
539+
engine.end_transaction()
540+
this_actor.sleep_for(1)
541+
engine.close()
542+
DTL.disconnect()
543+
544+
def subscriber():
545+
dtl = DTL.connect()
546+
stream = dtl.add_stream("my-output")
547+
engine = stream.open("my-output", Stream.Mode.Subscribe)
548+
var = stream.inquire_variable("var")
549+
550+
this_actor.info("Get the compressed variable (decompression cost should be applied)")
551+
engine.begin_transaction()
552+
engine.get(var)
553+
engine.end_transaction()
554+
555+
engine.close()
556+
DTL.disconnect()
557+
558+
pub_host.add_actor("Publisher", publisher)
559+
sub_host.add_actor("Subscriber", subscriber)
560+
e.run()
561+
562+
432563
if __name__ == '__main__':
433564
tests = [
434565
run_test_bogus_decimation_setting,
@@ -438,6 +569,8 @@ def test_actor():
438569
run_test_simple_compression_file_engine,
439570
run_test_compression_with_derived_ratio,
440571
run_test_double_reduction_forbidden,
572+
run_test_decimation_staging_engine,
573+
run_test_compression_staging_engine,
441574
]
442575

443576
for test in tests:

0 commit comments

Comments
 (0)