Skip to content

Commit 7d4f8b0

Browse files
committed
Remove restriction on number of threads for framework_graph
1 parent de46956 commit 7d4f8b0

File tree

3 files changed

+23
-8
lines changed

3 files changed

+23
-8
lines changed

phlex/core/framework_graph.cpp

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -57,14 +57,6 @@ namespace phlex::experimental {
5757
}},
5858
multiplexer_{graph_}
5959
{
60-
// FIXME: This requirement is in place so that the yielding driver can be used.
61-
// At least 2 threads are required for that to work.
62-
// It would be better if the specified concurrency would be applied to an
63-
// arena in which the user-facing work is done.
64-
if (max_parallelism < 2) {
65-
throw std::runtime_error("Must choose concurrency level of at least 2.");
66-
}
67-
6860
// FIXME: Should the loading of env levels happen in the phlex app only?
6961
spdlog::cfg::load_env_levels();
7062
spdlog::info("Number of worker threads: {}", max_allowed_parallelism::active_value());

test/CMakeLists.txt

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,7 @@ cet_test(
9292
LIBRARIES
9393
phlex::core
9494
Boost::json
95+
layer_generator
9596
)
9697
cet_test(
9798
function_registration

test/framework_graph.cpp

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
#include "phlex/core/framework_graph.hpp"
2+
#include "plugins/layer_generator.hpp"
23

34
#include "catch2/catch_test_macros.hpp"
45

@@ -17,3 +18,24 @@ TEST_CASE("Catch other exceptions", "[graph]")
1718
framework_graph g{[](framework_driver&) { throw 2.5; }};
1819
CHECK_THROWS_AS(g.execute(), double);
1920
}
21+
22+
TEST_CASE("Make progress with one thread", "[graph]")
23+
{
24+
layer_generator gen;
25+
gen.add_layer("spill", {"job", 1000});
26+
27+
framework_graph g{driver_for_test(gen), 1};
28+
g.provide(
29+
"provide_number",
30+
[](data_cell_index const& index) -> unsigned int { return index.number(); },
31+
concurrency::unlimited)
32+
.output_product("number"_in("spill"));
33+
g.observe(
34+
"observe_number", [](unsigned int const /*number*/) {}, concurrency::unlimited)
35+
.input_family("number"_in("spill"));
36+
g.execute();
37+
38+
CHECK(gen.emitted_cells("/job/spill") == 1000);
39+
CHECK(g.execution_counts("provide_number") == 1000);
40+
CHECK(g.execution_counts("observe_number") == 1000);
41+
}

0 commit comments

Comments
 (0)