unit_tests/Engine/ScaleTests.cc

0001 #include "catch.hpp"
0002
0003 #include <JANA/JApplication.h>
0004 #include <JANA/Utils/JCpuInfo.h>
0005 #include <JANA/Engine/JExecutionEngine.h>
0006
0007 #include "ScaleTests.h"
0008
0009 TEST_CASE("NThreads") {
0010
0011     JApplication app;
0012     app.SetParameterValue("jana:nevents",3);
0013     app.SetParameterValue("jana:loglevel","warn");
0014     app.Add(new scaletest::DummySource);
0015
0016     SECTION("If nthreads not provided, default to 1") {
0017         app.Run(false);
0018         auto threads = app.GetNThreads();
0019         REQUIRE(threads == 1);
0020         app.Stop(true);
0021     }
0022
0023     SECTION("If nthreads=Ncores, use ncores") {
0024         auto ncores = JCpuInfo::GetNumCpus();
0025         app.SetParameterValue("nthreads", "Ncores");
0026         app.Run(false);
0027         auto threads = app.GetNThreads();
0028         REQUIRE(threads == ncores);
0029         app.Stop(true);
0030     }
0031
0032     SECTION("If nthreads is something else, use that") {
0033         app.SetParameterValue("nthreads", 17);
0034         app.Run(false);
0035         auto threads = app.GetNThreads();
0036         REQUIRE(threads == 17);
0037         app.Stop(true);
0038     }
0039 }
0040
0041 TEST_CASE("ScaleNWorkerUpdate") {
0042     JApplication app;
0043     app.SetParameterValue("nthreads",4);
0044     app.SetParameterValue("jana:loglevel","warn");
0045     app.Add(new scaletest::DummySource);
0046     app.Add(new scaletest::DummyProcessor);
0047     app.Add(new JFactoryGeneratorT<scaletest::DummyFactory>());
0048     app.Run(false);
0049     auto threads = app.GetNThreads();
0050     REQUIRE(threads == 4);
0051
0052     app.Stop(true, false);
0053
0054     app.Scale(8); // Scale blocks until workers have fired up
0055     threads = app.GetNThreads();
0056     REQUIRE(threads == 8);
0057     app.Stop(true);
0058 }
0059
0060 TEST_CASE("ScaleThroughputImprovement", "[.][performance]") {
0061
0062     JApplication app;
0063     app.SetParameterValue("jana:loglevel", "INFO");
0064     //app.SetTicker(false);
0065     app.Add(new scaletest::DummySource);
0066     app.Add(new scaletest::DummyProcessor);
0067     app.Add(new JFactoryGeneratorT<scaletest::DummyFactory>());
0068     app.Initialize();
0069     auto jee = app.GetService<JExecutionEngine>();
0070
0071     jee->ScaleWorkers(1);
0072     jee->RunTopology();
0073     std::this_thread::sleep_for(std::chrono::seconds(5));
0074     auto throughput_hz_1 = jee->GetPerf().throughput_hz;
0075     std::cout << "nthreads=1: throughput_hz=" << throughput_hz_1 << std::endl;
0076     jee->PauseTopology();
0077     jee->RunSupervisor();
0078
0079     jee->ScaleWorkers(2);
0080     jee->RunTopology();
0081     std::this_thread::sleep_for(std::chrono::seconds(5));
0082     auto throughput_hz_2 = jee->GetPerf().throughput_hz;
0083     std::cout << "nthreads=2: throughput_hz=" << throughput_hz_2 << std::endl;
0084     REQUIRE(jee->GetPerf().runstatus == JExecutionEngine::RunStatus::Running);
0085     jee->PauseTopology();
0086     jee->RunSupervisor();
0087
0088     jee->ScaleWorkers(4);
0089     jee->RunTopology();
0090     std::this_thread::sleep_for(std::chrono::seconds(5));
0091     auto throughput_hz_4 = jee->GetPerf().throughput_hz;
0092     jee->PauseTopology();
0093     jee->RunSupervisor();
0094     std::cout << "nthreads=4: throughput_hz=" << throughput_hz_4 << std::endl;
0095
0096     jee->FinishTopology();
0097
0098     REQUIRE(throughput_hz_2 > throughput_hz_1*1.5);
0099     REQUIRE(throughput_hz_4 > throughput_hz_2*1.25);
0100 }