🌐 US-Proxy
class="logged-out env-production page-responsive" style="word-wrap: break-word;" >
Skip to content

Commit e4e8f84

Browse files
lowsferlowsfer
committed
[None][fix] Harden the KVCM2 concurrency tests against their own failure modes
Three problems in the tests added earlier in this PR, all of which made them look stronger than they were. Fatal assertions ran while worker threads were still joinable. ASSERT_* returns from the function immediately, so a firing assertion would have destroyed a joinable std::thread and terminated the process instead of reporting which check failed. The three sites now use EXPECT_* and guard the dependent checks, so the join always runs. ProbeReuseInterleavesWithExclusiveWork evaluated its loop condition before the first iteration, so if the prober had already cleared kMinProbes between sampling the baseline and the check, the writer performed no exclusive work at all and the test passed having exercised no interleaving. It is now a do/while, so at least one exclusive mutation always happens. The same test could also hang. glibc's rwlock prefers readers, so a continuous stream of probeReuse() shared acquisitions can starve the writer, and getAndResetIterationStats() would then block with no bound. The prober now stops on its own deadline and yields between probes, so starvation ends the run and fails the assertion rather than hanging CI. Tests: 84 KVCM2 C++ tests and 230 python tests pass; both concurrency binaries run 5x without flaking. Signed-off-by: Yao Yao <lowsfer@users.noreply.github.com>
1 parent bf075c3 commit e4e8f84

2 files changed

Lines changed: 49 additions & 22 deletions

File tree

‎cpp/tests/unit_tests/batch_manager/kvCacheManagerV2ConcurrencyTest.cpp‎

Lines changed: 32 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -162,43 +162,59 @@ TEST(KvCacheManagerV2ConcurrencyTest, ProbeReuseInterleavesWithExclusiveWork)
162162

163163
std::atomic<bool> stop{false};
164164
std::atomic<long long> probeCount{0};
165+
// The prober also stops on its own deadline. glibc's rwlock prefers readers, so a continuous
166+
// stream of shared acquisitions could in principle starve the writer; without the deadline the
167+
// writer's next exclusive call would block forever and hang CI instead of failing.
168+
auto const proberDeadline = std::chrono::steady_clock::now() + std::chrono::seconds{30};
165169
std::thread prober(
166170
[&]
167171
{
168-
while (!stop.load(std::memory_order_relaxed))
172+
while (!stop.load(std::memory_order_relaxed) && std::chrono::steady_clock::now() < proberDeadline)
169173
{
170174
manager->probeReuse({}, toSpan(tokens), /*knownNoDigest=*/true);
171175
probeCount.fetch_add(1, std::memory_order_relaxed);
176+
std::this_thread::yield(); // leave the writer a window to acquire
172177
}
173178
});
174179

175180
// Do not start the writer until the prober is demonstrably running, so "probes progressed
176-
// alongside exclusive work" cannot be satisfied by the writer finishing first. Bounded, so a
177-
// prober that never starts fails the test instead of hanging.
178-
auto const proberDeadline = std::chrono::steady_clock::now() + std::chrono::seconds{5};
179-
while (probeCount.load(std::memory_order_relaxed) == 0 && std::chrono::steady_clock::now() < proberDeadline)
181+
// alongside exclusive work" cannot be satisfied by the writer finishing first.
182+
auto const startDeadline = std::chrono::steady_clock::now() + std::chrono::seconds{5};
183+
while (probeCount.load(std::memory_order_relaxed) == 0 && std::chrono::steady_clock::now() < startDeadline)
180184
{
181185
std::this_thread::yield();
182186
}
183-
ASSERT_GT(probeCount.load(), 0) << "prober never ran";
187+
// EXPECT, not ASSERT: `prober` is joinable, and returning here would terminate on its destructor.
188+
bool const proberRan = probeCount.load(std::memory_order_relaxed) > 0;
189+
EXPECT_TRUE(proberRan) << "prober never ran";
184190

185-
// Now require further progress while the writer holds the exclusive lock repeatedly.
186191
long long const probesBeforeWriter = probeCount.load(std::memory_order_relaxed);
187192
constexpr long long kMinProbes = 1000;
188-
for (int iteration = 0;
189-
iteration < 100000 && probeCount.load(std::memory_order_relaxed) - probesBeforeWriter < kMinProbes;
190-
++iteration)
193+
int writerIterations = 0;
194+
if (proberRan)
191195
{
192-
// Exclusive-locked, and re-entrant into other locked methods.
193-
manager->getAndResetIterationStats();
194-
manager->markStatsDirty(std::nullopt);
195-
manager->clearStatsDirty(std::nullopt);
196+
// do/while so at least one exclusive mutation always happens: if the prober had already
197+
// cleared the threshold by the time the loop was entered, a while-loop would run zero
198+
// iterations and the test would assert nothing about interleaving.
199+
do
200+
{
201+
// Exclusive-locked, and re-entrant into other locked methods.
202+
manager->getAndResetIterationStats();
203+
manager->markStatsDirty(std::nullopt);
204+
manager->clearStatsDirty(std::nullopt);
205+
++writerIterations;
206+
} while (
207+
writerIterations < 100000 && probeCount.load(std::memory_order_relaxed) - probesBeforeWriter < kMinProbes);
196208
}
197209

198210
stop.store(true, std::memory_order_relaxed);
199211
prober.join();
200-
EXPECT_GE(probeCount.load() - probesBeforeWriter, kMinProbes)
201-
<< "probes made no progress against concurrent exclusive work";
212+
213+
if (proberRan)
214+
{
215+
EXPECT_GE(probeCount.load() - probesBeforeWriter, kMinProbes)
216+
<< "probes made no progress against " << writerIterations << " exclusive mutations";
217+
}
202218
}
203219

204220
} // namespace

‎cpp/tests/unit_tests/batch_manager/kvCacheManagerV2ReentrantSharedMutexTest.cpp‎

Lines changed: 17 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -112,9 +112,15 @@ TEST(ReentrantSharedMutexTest, NestingDoesNotLeakToOtherThreads)
112112
// about to acquire. The remaining window between that store and the lock() call itself is
113113
// inherent, so still allow a brief grace period -- but the test now fails if the thread
114114
// never runs at all, which a fixed sleep alone would have passed.
115-
ASSERT_TRUE(waitFor([&] { return otherAttempting.load(); })) << "contending thread never ran";
116-
std::this_thread::sleep_for(std::chrono::milliseconds(20));
117-
EXPECT_FALSE(otherEntered.load()) << "another thread entered while the lock was held";
115+
// EXPECT, not ASSERT: `other` is joinable here, and returning early would run its
116+
// destructor on a joinable thread, which terminates instead of reporting the failure.
117+
bool const attempted = waitFor([&] { return otherAttempting.load(); });
118+
EXPECT_TRUE(attempted) << "contending thread never ran";
119+
if (attempted)
120+
{
121+
std::this_thread::sleep_for(std::chrono::milliseconds(20));
122+
EXPECT_FALSE(otherEntered.load()) << "another thread entered while the lock was held";
123+
}
118124
} // both guards released here
119125
other.join();
120126
EXPECT_TRUE(otherEntered.load());
@@ -179,9 +185,14 @@ TEST(ReentrantSharedMutexTest, WriterWaitsForReaders)
179185
});
180186
// As above: confirm the writer actually reached its acquisition point before asserting
181187
// that it is still outside.
182-
ASSERT_TRUE(waitFor([&] { return writerAttempting.load(); })) << "writer thread never ran";
183-
std::this_thread::sleep_for(std::chrono::milliseconds(20));
184-
EXPECT_FALSE(writerIn.load()) << "writer entered while a reader held the lock";
188+
// EXPECT, not ASSERT: `writer` is joinable here (see above).
189+
bool const attempted = waitFor([&] { return writerAttempting.load(); });
190+
EXPECT_TRUE(attempted) << "writer thread never ran";
191+
if (attempted)
192+
{
193+
std::this_thread::sleep_for(std::chrono::milliseconds(20));
194+
EXPECT_FALSE(writerIn.load()) << "writer entered while a reader held the lock";
195+
}
185196
} // reader released here
186197
writer.join();
187198
EXPECT_TRUE(writerIn.load());

0 commit comments

Comments
 (0)