Skip to content

Commit 6e5476a

Browse files
committed
Make FileReaderManager clear admission tests deterministic
1 parent 2b010ff commit 6e5476a

1 file changed

Lines changed: 34 additions & 6 deletions

File tree

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/control/FileReaderManagerConcurrencyTest.java‎

Lines changed: 34 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -361,7 +361,7 @@ public void clearDrainsReleasePinsEvenWhenInterrupted() throws Exception {
361361
entry = pin.invoke(manager, registry, a.getTsFileID(), false, false);
362362
assertNotNull(entry);
363363
gate.owner.get().interrupt();
364-
Future<?> acquire = workers.submit(() -> get(a, true));
364+
Future<?> acquire = acquireDuringClear(a, true);
365365
gate.close();
366366
awaitClearDrain(gate.owner.get());
367367
assertFalse(clear.isDone());
@@ -385,18 +385,46 @@ public void clearDrainsReleasePinsEvenWhenInterrupted() throws Exception {
385385
}
386386
}
387387

388+
private Future<TsFileSequenceReader> acquireDuringClear(TsFileResource resource, boolean closed)
389+
throws Exception {
390+
AtomicReference<Thread> thread = new AtomicReference<>();
391+
CountDownLatch started = new CountDownLatch(1);
392+
Future<TsFileSequenceReader> acquire =
393+
workers.submit(
394+
() -> {
395+
thread.set(Thread.currentThread());
396+
started.countDown();
397+
return get(resource, closed);
398+
});
399+
assertTrue(started.await(10, TimeUnit.SECONDS));
400+
// An unfinished Future alone does not prove that the worker has reached the admission gate.
401+
awaitRegistryWait(thread.get(), "pin");
402+
return acquire;
403+
}
404+
388405
private void awaitClearDrain(Thread thread) throws Exception {
406+
awaitRegistryWait(thread, "drainPins");
407+
}
408+
409+
private void awaitRegistryWait(Thread thread, String method) throws Exception {
410+
Object registryLock = field("registryLock");
389411
long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
390412
while (System.nanoTime() < deadline) {
391413
ThreadInfo info = ManagementFactory.getThreadMXBean().getThreadInfo(thread.getId(), 30);
392-
for (StackTraceElement frame : info.getStackTrace()) {
393-
if (frame.getMethodName().equals("drainPins")) {
394-
return;
414+
if (info != null
415+
&& info.getThreadState() == Thread.State.WAITING
416+
&& info.getLockInfo() != null
417+
&& info.getLockInfo().getIdentityHashCode() == System.identityHashCode(registryLock)) {
418+
for (StackTraceElement frame : info.getStackTrace()) {
419+
if (frame.getClassName().equals(FileReaderManager.class.getName())
420+
&& frame.getMethodName().equals(method)) {
421+
return;
422+
}
395423
}
396424
}
397425
Thread.sleep(1);
398426
}
399-
fail("clear did not reach its final pin drain");
427+
fail("worker did not wait on registryLock in " + method);
400428
}
401429

402430
@Test
@@ -566,7 +594,7 @@ public void interruptedClearWaitsForExistingOperation() throws Exception {
566594
assertTrue(started.await(10, TimeUnit.SECONDS));
567595
awaitClearDrain(clearThread.get());
568596
clearThread.get().interrupt();
569-
Future<?> acquire = workers.submit(() -> get(a, false));
597+
Future<?> acquire = acquireDuringClear(a, false);
570598
assertFalse(clear.isDone());
571599
assertFalse(acquire.isDone());
572600
gate.close();

0 commit comments

Comments
 (0)