diff --git a/core/src/main/java/org/testcontainers/utility/RyukResourceReaper.java b/core/src/main/java/org/testcontainers/utility/RyukResourceReaper.java index 999b1cbbffb..76e85c01132 100644 --- a/core/src/main/java/org/testcontainers/utility/RyukResourceReaper.java +++ b/core/src/main/java/org/testcontainers/utility/RyukResourceReaper.java @@ -33,6 +33,8 @@ class RyukResourceReaper extends ResourceReaper { private final AtomicBoolean started = new AtomicBoolean(false); + private volatile boolean ready = false; + private final RyukContainer ryukContainer = new RyukContainer(); @Override @@ -70,66 +72,81 @@ public CreateContainerCmd register(GenericContainer container, CreateContaine } @SneakyThrows(InterruptedException.class) - private synchronized void maybeStart() { - if (!started.compareAndSet(false, true)) { + private void maybeStart() { + if (ready) { return; } - ryukContainer.start(); - - CountDownLatch ryukScheduledLatch = new CountDownLatch(1); - - String host = ryukContainer.getHost(); - Integer ryukPort = ryukContainer.getFirstMappedPort(); - Thread kiraThread = new Thread( - DockerClientFactory.TESTCONTAINERS_THREAD_GROUP, - () -> { - while (true) { - RYUK_ACK_RATE_LIMITER.doWhenReady(() -> { - int index = 0; - // not set the read timeout, as Ryuk would not send anything unless a new filter is submitted, meaning that we would get a timeout exception pretty quick - try (Socket clientSocket = new Socket()) { - clientSocket.connect(new InetSocketAddress(host, ryukPort), 5 * 1000); - ResourceReaper.FilterRegistry registry = new ResourceReaper.FilterRegistry( - clientSocket.getInputStream(), - clientSocket.getOutputStream() - ); - - synchronized (ResourceReaper.DEATH_NOTE) { - while (true) { - if (ResourceReaper.DEATH_NOTE.size() <= index) { - try { - ResourceReaper.DEATH_NOTE.wait(1_000); - continue; - } catch (InterruptedException e) { - throw new RuntimeException(e); + DockerClientFactory.instance().client(); + + synchronized (this) { + if (ready) { + return; + } + if (!started.compareAndSet(false, true)) { + return; + } + + ryukContainer.start(); + + CountDownLatch ryukScheduledLatch = new CountDownLatch(1); + + String host = ryukContainer.getHost(); + Integer ryukPort = ryukContainer.getFirstMappedPort(); + Thread kiraThread = new Thread( + DockerClientFactory.TESTCONTAINERS_THREAD_GROUP, + () -> { + while (true) { + RYUK_ACK_RATE_LIMITER.doWhenReady(() -> { + int index = 0; + // not set the read timeout, as Ryuk would not send anything unless a new filter is submitted, meaning that we would get a timeout exception pretty quick + try (Socket clientSocket = new Socket()) { + clientSocket.connect(new InetSocketAddress(host, ryukPort), 5 * 1000); + ResourceReaper.FilterRegistry registry = new ResourceReaper.FilterRegistry( + clientSocket.getInputStream(), + clientSocket.getOutputStream() + ); + + synchronized (ResourceReaper.DEATH_NOTE) { + while (true) { + if (ResourceReaper.DEATH_NOTE.size() <= index) { + try { + ResourceReaper.DEATH_NOTE.wait(1_000); + continue; + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + List> filters = ResourceReaper.DEATH_NOTE.get(index); + boolean isAcknowledged = registry.register(filters); + if (isAcknowledged) { + log.debug("Received 'ACK' from Ryuk"); + ryukScheduledLatch.countDown(); + index++; + } else { + log.debug("Didn't receive 'ACK' from Ryuk. Will retry to send filters."); } - } - List> filters = ResourceReaper.DEATH_NOTE.get(index); - boolean isAcknowledged = registry.register(filters); - if (isAcknowledged) { - log.debug("Received 'ACK' from Ryuk"); - ryukScheduledLatch.countDown(); - index++; - } else { - log.debug("Didn't receive 'ACK' from Ryuk. Will retry to send filters."); } } + } catch (IOException e) { + log.warn("Can not connect to Ryuk at {}:{}", host, ryukPort, e); } - } catch (IOException e) { - log.warn("Can not connect to Ryuk at {}:{}", host, ryukPort, e); - } - }); - } - }, - "testcontainers-ryuk" - ); - kiraThread.setDaemon(true); - kiraThread.start(); - // We need to wait before we can start any containers to make sure that we delete them - if (!ryukScheduledLatch.await(TestcontainersConfiguration.getInstance().getRyukTimeout(), TimeUnit.SECONDS)) { - log.error("Timed out waiting for Ryuk container to start. Ryuk's logs:\n{}", ryukContainer.getLogs()); - throw new IllegalStateException(String.format("Could not connect to Ryuk at %s:%s", host, ryukPort)); + }); + } + }, + "testcontainers-ryuk" + ); + kiraThread.setDaemon(true); + kiraThread.start(); + // We need to wait before we can start any containers to make sure that we delete them + if ( + !ryukScheduledLatch.await(TestcontainersConfiguration.getInstance().getRyukTimeout(), TimeUnit.SECONDS) + ) { + log.error("Timed out waiting for Ryuk container to start. Ryuk's logs:\n{}", ryukContainer.getLogs()); + throw new IllegalStateException(String.format("Could not connect to Ryuk at %s:%s", host, ryukPort)); + } + + ready = true; } } } diff --git a/core/src/test/java/org/testcontainers/RyukResourceReaperDeadlockTest.java b/core/src/test/java/org/testcontainers/RyukResourceReaperDeadlockTest.java new file mode 100644 index 00000000000..dbd00c29c78 --- /dev/null +++ b/core/src/test/java/org/testcontainers/RyukResourceReaperDeadlockTest.java @@ -0,0 +1,85 @@ +package org.testcontainers; + +import org.junit.jupiter.api.Test; +import org.testcontainers.utility.ResourceReaper; + +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class RyukResourceReaperDeadlockTest { + + @Test + void shouldNotHoldReaperLockWhenCallingDockerClientFactory() { + ResourceReaper reaper = ResourceReaper.instance(); + AtomicBoolean heldLock = new AtomicBoolean(true); + + DockerClientFactory dockerClientFactory = DockerClientFactory.instance(); + RuntimeException simulatedFailure = new RuntimeException("Simulated failure to intercept call") { + @Override + public synchronized Throwable fillInStackTrace() { + heldLock.set(Thread.holdsLock(reaper)); + return this; + } + }; + + dockerClientFactory.cachedClientFailure = simulatedFailure; + try { + assertThatThrownBy(() -> reaper.registerLabelsFilterForCleanup(Collections.singletonMap("k", "v"))) + .isSameAs(simulatedFailure); + + assertThat(heldLock.get()) + .as("RyukResourceReaper monitor must NOT be held when calling DockerClientFactory to prevent deadlocks") + .isFalse(); + } finally { + dockerClientFactory.cachedClientFailure = null; + } + } + + @Test + void concurrentCallersShouldNotDeadlockUnderContention() throws InterruptedException { + ResourceReaper reaper = ResourceReaper.instance(); + DockerClientFactory dockerClientFactory = DockerClientFactory.instance(); + RuntimeException simulatedFailure = new RuntimeException("Simulated client error"); + + dockerClientFactory.cachedClientFailure = simulatedFailure; + try { + int threadCount = 8; + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(threadCount); + + for (int i = 0; i < threadCount; i++) { + final int id = i; + executor.submit(() -> { + try { + startLatch.await(2, TimeUnit.SECONDS); + if (id % 2 == 0) { + reaper.registerLabelsFilterForCleanup(Collections.singletonMap("test", "val" + id)); + } else { + reaper.getLabels(); + } + } catch (Throwable ignored) {} finally { + doneLatch.countDown(); + } + }); + } + + startLatch.countDown(); + boolean finished = doneLatch.await(5, TimeUnit.SECONDS); + executor.shutdownNow(); + + assertThat(finished) + .as("Concurrent calls to RyukResourceReaper must finish within timeout and never deadlock") + .isTrue(); + } finally { + dockerClientFactory.cachedClientFailure = null; + } + } +}