lriggs commented on code in PR #53160:
URL: https://github.com/apache/arrow/pull/53160#discussion_r4353673396
##########
cpp/src/arrow/compute/kernels/scalar_string_test.cc:
##########
@@ -1703,6 +1706,41 @@ TYPED_TEST(TestBaseBinaryKernels, MatchSubstring) {
&options_empty);
}
+TYPED_TEST(TestBaseBinaryKernels, MatchSubstringConcurrentFirstUse) {
+ auto in_schema = schema({field("x", this->type())});
+ MatchSubstringOptions options{"ab"};
+ auto expr = call("match_substring", {field_ref("x")}, options);
+ ASSERT_OK_AND_ASSIGN(auto bound_expr, expr.Bind(*in_schema));
+
+ Datum input =
+ ArrayFromJSON(this->type(), R"(["abc", "acb", "cab", null, "bac",
"AB"])");
+ ExecBatch batch({input}, 6);
+
+ constexpr int kNumThreads = 16;
+ std::vector<std::thread> threads;
+ std::vector<Result<Datum>> results(kNumThreads);
+ std::atomic<bool> ready{false};
+
+ for (int i = 0; i < kNumThreads; ++i) {
Review Comment:
I think there might be an issue here with the create/run initialization and
ordering.
This uses memory_order_relaxed and the ready.store(true) fires immediately
after thread construction.
Thread creation and OS scheduling are not instantaneous. Some threads may
not have yet entered (or even started) the spin-wait when ready becomes true.
In that case they run sequentially after the first-use is already complete, and
the test never actually exercises a concurrent first-init race.
You could fix this by using a std::atomic<int> countdown latch: have each
thread decrement a counter when it's ready, spin until the counter reaches
zero, then proceed. This guarantees all threads are spinning before any
executes. (C++20 std::latch or std::barrier work too)
std::atomic<int> waiting{0};
std::atomic<bool> go{false};
for (int i = 0; i < kNumThreads; ++i) {
threads.emplace_back([&, i]() {
waiting.fetch_add(1, std::memory_order_relaxed);
while (!go.load(std::memory_order_acquire)) {}
results[i] = ExecuteScalarExpression(bound_expr, batch);
});
}
// Wait for all threads to reach the barrier
while (waiting.load(std::memory_order_relaxed) < kNumThreads) {}
go.store(true, std::memory_order_release);
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
