Skip to content

Commit ee1f291

Browse files
shivansh023023shivansh023023
committed
test(execution): fix thread_pool_scheduler tests and deadlock in future_sender_test
Signed-off-by: Shivansh Singh <singhshivansh023@gmail.com>
1 parent a240b29 commit ee1f291

2 files changed

Lines changed: 95 additions & 71 deletions

File tree

‎libs/core/executors/tests/unit/thread_pool_scheduler.cpp‎

Lines changed: 90 additions & 68 deletions
Original file line numberDiff line numberDiff line change
@@ -554,7 +554,7 @@ void test_bulk_starts_on()
554554

555555
// Test starts_on pattern: bulk operation with scheduler in environment
556556
// Use start_on to provide scheduler through environment
557-
auto bulk_sender = ex::continues_on(
557+
auto bulk_sender = ex::starts_on(
558558
ex::thread_pool_scheduler{}, ex::just() | ex::bulk(n, [&](int i) {
559559
++v[i];
560560
HPX_TEST_NEQ(parent_id, hpx::this_thread::get_id());
@@ -865,7 +865,7 @@ void test_future_sender()
865865
}
866866

867867
{
868-
auto s = ex::just(ex::thread_pool_scheduler{}, 3);
868+
auto s = ex::starts_on(ex::thread_pool_scheduler{}, ex::just(3));
869869
auto f = ex::make_future(std::move(s));
870870
HPX_TEST_EQ(f.get(), 3);
871871
}
@@ -876,7 +876,8 @@ void test_future_sender()
876876
}
877877

878878
{
879-
auto f = ex::just(ex::thread_pool_scheduler{}, 3) | ex::make_future();
879+
auto f = ex::starts_on(ex::thread_pool_scheduler{}, ex::just(3)) |
880+
ex::make_future();
880881
HPX_TEST_EQ(f.get(), 3);
881882
}
882883

@@ -890,9 +891,11 @@ void test_future_sender()
890891
}
891892

892893
{
893-
auto s1 = ex::just(ex::thread_pool_scheduler{}, std::size_t(42));
894-
auto s2 = ex::just(ex::thread_pool_scheduler{}, 3.14);
895-
auto s3 = ex::just(ex::thread_pool_scheduler{}, std::string("hello"));
894+
auto s1 = ex::starts_on(
895+
ex::thread_pool_scheduler{}, ex::just(std::size_t(42)));
896+
auto s2 = ex::starts_on(ex::thread_pool_scheduler{}, ex::just(3.14));
897+
auto s3 = ex::starts_on(
898+
ex::thread_pool_scheduler{}, ex::just(std::string("hello")));
896899
auto f = ex::make_future(ex::then(
897900
ex::when_all(std::move(s1), std::move(s2), std::move(s3)),
898901
[](std::size_t x, double, std::string z) { return z.size() + x; }));
@@ -901,8 +904,9 @@ void test_future_sender()
901904

902905
// mixing senders and futures
903906
{
904-
HPX_TEST_EQ(hpx::get<0>(*tt::sync_wait(ex::as_sender(ex::make_future(
905-
ex::just(ex::thread_pool_scheduler{}, 42))))),
907+
HPX_TEST_EQ(
908+
hpx::get<0>(*tt::sync_wait(ex::as_sender(ex::make_future(
909+
ex::starts_on(ex::thread_pool_scheduler{}, ex::just(42)))))),
906910
42);
907911
}
908912

@@ -916,9 +920,11 @@ void test_future_sender()
916920
}
917921

918922
{
919-
auto s1 = ex::just(ex::thread_pool_scheduler{}, std::size_t(42));
920-
auto s2 = ex::just(ex::thread_pool_scheduler{}, 3.14);
921-
auto s3 = ex::just(ex::thread_pool_scheduler{}, std::string("hello"));
923+
auto s1 = ex::starts_on(
924+
ex::thread_pool_scheduler{}, ex::just(std::size_t(42)));
925+
auto s2 = ex::starts_on(ex::thread_pool_scheduler{}, ex::just(3.14));
926+
auto s3 = ex::starts_on(
927+
ex::thread_pool_scheduler{}, ex::just(std::string("hello")));
922928
auto f = ex::make_future(ex::then(
923929
ex::when_all(std::move(s1), std::move(s2), std::move(s3)),
924930
[](std::size_t x, double, std::string z) { return z.size() + x; }));
@@ -945,18 +951,19 @@ void test_ensure_started()
945951
}
946952

947953
{
948-
auto s = ex::just(sched, 42) | ex::ensure_started();
954+
auto s = ex::starts_on(sched, ex::just(42)) | ex::ensure_started();
949955
HPX_TEST_EQ(hpx::get<0>(*tt::sync_wait(std::move(s))), 42);
950956
}
951957

952958
{
953-
auto s = ex::just(sched, 42) | ex::ensure_started() |
959+
auto s = ex::starts_on(sched, ex::just(42)) | ex::ensure_started() |
954960
ex::continues_on(sched);
955961
HPX_TEST_EQ(hpx::get<0>(*tt::sync_wait(std::move(s))), 42);
956962
}
957963

958964
{
959-
auto s = ex::just(sched, 42) | ex::ensure_started() | ex::split();
965+
auto s = ex::starts_on(sched, ex::just(42)) | ex::ensure_started() |
966+
ex::split();
960967
HPX_TEST_EQ(hpx::get<0>(*tt::sync_wait(s)), 42);
961968
HPX_TEST_EQ(hpx::get<0>(*tt::sync_wait(s)), 42);
962969
HPX_TEST_EQ(hpx::get<0>(*tt::sync_wait(s)), 42);
@@ -1081,17 +1088,18 @@ void test_split()
10811088
}
10821089

10831090
{
1084-
auto s = ex::just(sched, 42) | ex::split();
1091+
auto s = ex::starts_on(sched, ex::just(42)) | ex::split();
10851092
HPX_TEST_EQ(hpx::get<0>(*tt::sync_wait(std::move(s))), 42);
10861093
}
10871094

10881095
{
1089-
auto s = ex::just(sched, 42) | ex::split() | ex::continues_on(sched);
1096+
auto s = ex::starts_on(sched, ex::just(42)) | ex::split() |
1097+
ex::continues_on(sched);
10901098
HPX_TEST_EQ(hpx::get<0>(*tt::sync_wait(std::move(s))), 42);
10911099
}
10921100

10931101
{
1094-
auto s = ex::just(sched, 42) | ex::split();
1102+
auto s = ex::starts_on(sched, ex::just(42)) | ex::split();
10951103
HPX_TEST_EQ(hpx::get<0>(*tt::sync_wait(s)), 42);
10961104
HPX_TEST_EQ(hpx::get<0>(*tt::sync_wait(s)), 42);
10971105
HPX_TEST_EQ(hpx::get<0>(*tt::sync_wait(s)), 42);
@@ -1183,50 +1191,59 @@ void test_let_value()
11831191
}
11841192

11851193
{
1186-
auto result = hpx::get<0>(*(tt::sync_wait(ex::schedule(sched) |
1187-
ex::let_value([=]() { return ex::just(sched, 42); }))));
1194+
auto result = hpx::get<0>(
1195+
*(tt::sync_wait(ex::schedule(sched) | ex::let_value([=]() {
1196+
return ex::starts_on(sched, ex::just(42));
1197+
}))));
11881198
HPX_TEST_EQ(result, 42);
11891199
}
11901200

11911201
{
1192-
auto result = hpx::get<0>(*tt::sync_wait((ex::just() |
1193-
ex::let_value([=]() { return ex::just(sched, 42); }))));
1202+
auto result =
1203+
hpx::get<0>(*tt::sync_wait((ex::just() | ex::let_value([=]() {
1204+
return ex::starts_on(sched, ex::just(42));
1205+
}))));
11941206
HPX_TEST_EQ(result, 42);
11951207
}
11961208

11971209
// int predecessor, value ignored
11981210
{
1199-
auto result = hpx::get<0>(*(tt::sync_wait(ex::just(sched, 43) |
1200-
ex::let_value([](int&) { return ex::just(42); }))));
1211+
auto result =
1212+
hpx::get<0>(*(tt::sync_wait(ex::starts_on(sched, ex::just(43)) |
1213+
ex::let_value([](int&) { return ex::just(42); }))));
12011214
HPX_TEST_EQ(result, 42);
12021215
}
12031216

12041217
{
1205-
auto result = hpx::get<0>(*(tt::sync_wait(ex::just(sched, 43) |
1206-
ex::let_value([=](int&) { return ex::just(sched, 42); }))));
1218+
auto result = hpx::get<0>(*(tt::sync_wait(
1219+
ex::starts_on(sched, ex::just(43)) | ex::let_value([=](int&) {
1220+
return ex::starts_on(sched, ex::just(42));
1221+
}))));
12071222
HPX_TEST_EQ(result, 42);
12081223
}
12091224

12101225
{
1211-
auto result = hpx::get<0>(*(tt::sync_wait(ex::just(43) |
1212-
ex::let_value([=](int&) { return ex::just(sched, 42); }))));
1226+
auto result =
1227+
hpx::get<0>(*(tt::sync_wait(ex::just(43) | ex::let_value([=](int&) {
1228+
return ex::starts_on(sched, ex::just(42));
1229+
}))));
12131230
HPX_TEST_EQ(result, 42);
12141231
}
12151232

12161233
// int predecessor, value used
12171234
{
1218-
auto result = hpx::get<0>(
1219-
*(tt::sync_wait(ex::just(sched, 43) | ex::let_value([](int& x) {
1235+
auto result = hpx::get<0>(*(tt::sync_wait(
1236+
ex::starts_on(sched, ex::just(43)) | ex::let_value([](int& x) {
12201237
return ex::just(42) | ex::then([&](int y) { return x + y; });
12211238
}))));
12221239

12231240
HPX_TEST_EQ(result, 85);
12241241
}
12251242

12261243
{
1227-
auto result = hpx::get<0>(
1228-
*(tt::sync_wait(ex::just(sched, 43) | ex::let_value([=](int& x) {
1229-
return ex::just(sched, 42) |
1244+
auto result = hpx::get<0>(*(tt::sync_wait(
1245+
ex::starts_on(sched, ex::just(43)) | ex::let_value([=](int& x) {
1246+
return ex::starts_on(sched, ex::just(42)) |
12301247
ex::then([&](int y) { return x + y; });
12311248
}))));
12321249
HPX_TEST_EQ(result, 85);
@@ -1235,7 +1252,7 @@ void test_let_value()
12351252
{
12361253
auto result = hpx::get<0>(
12371254
*(tt::sync_wait(ex::just(43) | ex::let_value([=](int& x) {
1238-
return ex::just(sched, 42) |
1255+
return ex::starts_on(sched, ex::just(42)) |
12391256
ex::then([&](int y) { return x + y; });
12401257
}))));
12411258
HPX_TEST_EQ(result, 85);
@@ -1247,13 +1264,15 @@ void test_let_value()
12471264

12481265
try
12491266
{
1250-
tt::sync_wait(ex::just(sched, 43) | ex::then([](int x) {
1251-
throw std::runtime_error("error");
1252-
return x;
1253-
}) | ex::let_value([](int&) {
1254-
HPX_TEST(false);
1255-
return ex::just(0);
1256-
}));
1267+
tt::sync_wait(ex::starts_on(sched, ex::just(43)) |
1268+
ex::then([](int x) {
1269+
throw std::runtime_error("error");
1270+
return x;
1271+
}) |
1272+
ex::let_value([](int&) {
1273+
HPX_TEST(false);
1274+
return ex::just(0);
1275+
}));
12571276
HPX_TEST(false);
12581277
}
12591278
catch (std::runtime_error const& e)
@@ -1306,7 +1325,7 @@ void test_let_error()
13061325
}) | ex::let_error([=, &called](std::exception_ptr& ep) {
13071326
called = true;
13081327
check_exception_ptr_message(ep, "error");
1309-
return ex::just(sched);
1328+
return ex::starts_on(sched, ex::just());
13101329
}));
13111330
HPX_TEST(called);
13121331
}
@@ -1318,7 +1337,7 @@ void test_let_error()
13181337
}) | ex::let_error([=, &called](std::exception_ptr& ep) {
13191338
called = true;
13201339
check_exception_ptr_message(ep, "error");
1321-
return ex::just(sched);
1340+
return ex::starts_on(sched, ex::just());
13221341
}));
13231342
HPX_TEST(called);
13241343
}
@@ -1343,7 +1362,7 @@ void test_let_error()
13431362
return 43;
13441363
}) | ex::let_error([=](std::exception_ptr& ep) {
13451364
check_exception_ptr_message(ep, "error");
1346-
return ex::just(sched, 42);
1365+
return ex::starts_on(sched, ex::just(42));
13471366
}))));
13481367
HPX_TEST_EQ(result, 42);
13491368
}
@@ -1354,35 +1373,37 @@ void test_let_error()
13541373
return 43;
13551374
}) | ex::let_error([=](std::exception_ptr& ep) {
13561375
check_exception_ptr_message(ep, "error");
1357-
return ex::just(sched, 42);
1376+
return ex::starts_on(sched, ex::just(42));
13581377
}))));
13591378
HPX_TEST_EQ(result, 42);
13601379
}
13611380

13621381
// predecessor doesn't throw, let sender is ignored
13631382
{
1364-
auto result = hpx::get<0>(*(tt::sync_wait(
1365-
ex::just(sched, 42) | ex::let_error([](std::exception_ptr) {
1366-
HPX_TEST(false);
1367-
return ex::just(43);
1368-
}))));
1383+
auto result =
1384+
hpx::get<0>(*(tt::sync_wait(ex::starts_on(sched, ex::just(42)) |
1385+
ex::let_error([](std::exception_ptr) {
1386+
HPX_TEST(false);
1387+
return ex::just(43);
1388+
}))));
13691389
HPX_TEST_EQ(result, 42);
13701390
}
13711391

13721392
{
1373-
auto result = hpx::get<0>(*(tt::sync_wait(
1374-
ex::just(sched, 42) | ex::let_error([=](std::exception_ptr) {
1375-
HPX_TEST(false);
1376-
return ex::just(sched, 43);
1377-
}))));
1393+
auto result =
1394+
hpx::get<0>(*(tt::sync_wait(ex::starts_on(sched, ex::just(42)) |
1395+
ex::let_error([=](std::exception_ptr) {
1396+
HPX_TEST(false);
1397+
return ex::starts_on(sched, ex::just(43));
1398+
}))));
13781399
HPX_TEST_EQ(result, 42);
13791400
}
13801401

13811402
{
13821403
auto result = hpx::get<0>(*(
13831404
tt::sync_wait(ex::just(42) | ex::let_error([=](std::exception_ptr) {
13841405
HPX_TEST(false);
1385-
return ex::just(sched, 43);
1406+
return ex::starts_on(sched, ex::just(43));
13861407
}))));
13871408
HPX_TEST_EQ(result, 42);
13881409
}
@@ -1683,12 +1704,12 @@ void test_bulk()
16831704
std::vector<int> v(n, -1);
16841705
hpx::thread::id parent_id = hpx::this_thread::get_id();
16851706

1686-
auto v_out = hpx::get<0>(*(
1687-
tt::sync_wait(ex::just(ex::thread_pool_scheduler{}, std::move(v)) |
1688-
ex::bulk(n, [&parent_id](int i, std::vector<int>& v) {
1689-
v[i] = i;
1690-
HPX_TEST_NEQ(parent_id, hpx::this_thread::get_id());
1691-
}))));
1707+
auto v_out = hpx::get<0>(*(tt::sync_wait(
1708+
ex::starts_on(ex::thread_pool_scheduler{}, ex::just(std::move(v))) |
1709+
ex::bulk(n, [&parent_id](int i, std::vector<int>& v) {
1710+
v[i] = i;
1711+
HPX_TEST_NEQ(parent_id, hpx::this_thread::get_id());
1712+
}))));
16921713

16931714
// In chunked mode, only chunk begin indices are processed
16941715
// So we check that at least some elements were set correctly
@@ -1736,7 +1757,8 @@ void test_bulk()
17361757

17371758
try
17381759
{
1739-
tt::sync_wait(ex::just(ex::thread_pool_scheduler{}) |
1760+
tt::sync_wait(
1761+
ex::starts_on(ex::thread_pool_scheduler{}, ex::just()) |
17401762
ex::bulk(n, [&v, i_fail](int i) {
17411763
if (i == i_fail)
17421764
{
@@ -1804,7 +1826,7 @@ void test_stdexec_domain_queries()
18041826
// 4. Verify transform_sender produces thread_pool_bulk_sender for
18051827
// bulk_chunked (proves the domain customization is picked up)
18061828
{
1807-
auto env = ex::env{ex::prop{ex::get_scheduler, scheduler}};
1829+
auto env = ex::make_env(ex::prop{ex::get_scheduler, scheduler});
18081830

18091831
auto chunked_sndr = ex::bulk_chunked(
18101832
ex::schedule(scheduler), ex::par, 10, [](int, int) {});
@@ -1827,7 +1849,7 @@ void test_stdexec_domain_queries()
18271849
// 5. Verify transform_sender produces thread_pool_bulk_sender for
18281850
// bulk_unchunked (proves the domain customization is picked up)
18291851
{
1830-
auto env = ex::env{ex::prop{ex::get_scheduler, scheduler}};
1852+
auto env = ex::make_env(ex::prop{ex::get_scheduler, scheduler});
18311853

18321854
auto unchunked_sndr = ex::bulk_unchunked(
18331855
ex::schedule(scheduler), ex::par, 10, [](int) {});
@@ -2090,7 +2112,7 @@ void test_completion_scheduler()
20902112
}
20912113

20922114
{
2093-
auto sender = ex::just(ex::thread_pool_scheduler{}, 42);
2115+
auto sender = ex::starts_on(ex::thread_pool_scheduler{}, ex::just(42));
20942116
auto completion_scheduler =
20952117
ex::get_completion_scheduler<ex::set_value_t>(ex::get_env(sender));
20962118
static_assert(
@@ -2112,7 +2134,7 @@ void test_completion_scheduler()
21122134

21132135
{
21142136
auto sender = ex::then(
2115-
ex::bulk(ex::just(ex::thread_pool_scheduler{}, 42), 10,
2137+
ex::bulk(ex::starts_on(ex::thread_pool_scheduler{}, ex::just(42)), 10,
21162138
[](int, int) {}),
21172139
[](int) {});
21182140
auto completion_scheduler =
@@ -2136,7 +2158,7 @@ void test_completion_scheduler()
21362158

21372159
{
21382160
auto sender = ex::then(
2139-
ex::bulk(ex::just(ex::thread_pool_scheduler{}, 42),
2161+
ex::bulk(ex::starts_on(ex::thread_pool_scheduler{}, ex::just(42)),
21402162
ex::par, 10, [](int, int) {}),
21412163
[](int) {});
21422164
auto completion_scheduler =
@@ -2149,7 +2171,7 @@ void test_completion_scheduler()
21492171

21502172
{
21512173
auto sender = ex::bulk(
2152-
ex::then(ex::just(ex::thread_pool_scheduler{}, 42),
2174+
ex::then(ex::starts_on(ex::thread_pool_scheduler{}, ex::just(42)),
21532175
[](int i) { return i; }),
21542176
ex::par, 10, [](int idx, int val) {});
21552177
auto completion_scheduler =

0 commit comments

Comments
 (0)