From ebcaef4a4d66b06208cf66039c3e0b775cfb4a5b Mon Sep 17 00:00:00 2001 From: luoyuan Date: Tue, 29 Sep 2026 20:44:46 +0800 Subject: [PATCH] YARN-11997. Serialize concurrent Docker image pulls by image name --- .../runtime/DockerLinuxContainerRuntime.java | 41 +++++ .../runtime/TestDockerContainerRuntime.java | 149 ++++++++++++++++++ 2 files changed, 190 insertions(+) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/DockerLinuxContainerRuntime.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/DockerLinuxContainerRuntime.java index 14f5ffeefe0d3f..ad8f83cddd5eb8 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/DockerLinuxContainerRuntime.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/DockerLinuxContainerRuntime.java @@ -80,6 +80,8 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -212,6 +214,13 @@ public class DockerLinuxContainerRuntime extends OCIContainerRuntime { private static final String DEFAULT_PROCFS = "/proc"; + private final ConcurrentMap imagePullLocks = + new ConcurrentHashMap<>(); + + private static final class ImagePullLock { + private int referenceCount; + } + @InterfaceAudience.Private private static final String RUNTIME_TYPE = "DOCKER"; @@ -1219,6 +1228,38 @@ public static void validateImageName(String imageName) public void pullImageFromRemote(String containerIdStr, String imageName) throws ContainerExecutionException { + ImagePullLock imagePullLock = acquireImagePullLock(imageName); + try { + synchronized (imagePullLock) { + doPullImageFromRemote(containerIdStr, imageName); + } + } finally { + releaseImagePullLock(imageName); + } + } + + private ImagePullLock acquireImagePullLock(String imageName) { + return imagePullLocks.compute(imageName, (key, lock) -> { + ImagePullLock imagePullLock = lock == null ? new ImagePullLock() : lock; + imagePullLock.referenceCount++; + return imagePullLock; + }); + } + + private void releaseImagePullLock(String imageName) { + imagePullLocks.computeIfPresent(imageName, (key, lock) -> { + lock.referenceCount--; + return lock.referenceCount == 0 ? null : lock; + }); + } + + @VisibleForTesting + int getImagePullLockCount() { + return imagePullLocks.size(); + } + + private void doPullImageFromRemote(String containerIdStr, String imageName) + throws ContainerExecutionException { long start = System.currentTimeMillis(); DockerPullCommand dockerPullCommand = new DockerPullCommand(imageName); LOG.debug("now pulling docker image. image name: {}, container: {}", diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/TestDockerContainerRuntime.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/TestDockerContainerRuntime.java index 84f262438341f4..28fc3b81618327 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/TestDockerContainerRuntime.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/TestDockerContainerRuntime.java @@ -28,6 +28,7 @@ import org.apache.hadoop.io.DataOutputBuffer; import org.apache.hadoop.registry.client.api.RegistryConstants; import org.apache.hadoop.security.Credentials; +import org.apache.hadoop.test.GenericTestUtils; import org.apache.hadoop.util.Shell; import org.apache.hadoop.util.StringUtils; import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; @@ -58,6 +59,7 @@ import org.apache.hadoop.yarn.server.nodemanager.containermanager.runtime.ContainerRuntimeConstants; import org.apache.hadoop.yarn.server.nodemanager.containermanager.runtime.ContainerRuntimeContext; import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.MethodSource; @@ -89,6 +91,10 @@ import java.util.Random; import java.util.Set; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import static org.apache.hadoop.test.MockitoUtil.verifyZeroInteractions; import static org.apache.hadoop.yarn.conf.YarnConfiguration.NM_DOCKER_DEFAULT_RO_MOUNTS; @@ -119,6 +125,8 @@ import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.OCIContainerRuntime.RUN_PRIVILEGED_CONTAINER_SUFFIX; import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.OCIContainerRuntime.formatOciEnvKey; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; @@ -696,6 +704,147 @@ public void testDockerContainerLaunchWithDefaultImageUpdate(boolean pHttps) dockerCommands.get(counter)); } + @Test + public void testConcurrentPullsForSameImageAreSerialized() + throws Exception { + initHttps(false); + DockerLinuxContainerRuntime runtime = + new DockerLinuxContainerRuntime(mockExecutor, mockCGroupsHandler); + runtime.initialize(conf, nmContext); + + CountDownLatch firstPullStarted = new CountDownLatch(1); + CountDownLatch releaseFirstPull = new CountDownLatch(1); + AtomicInteger pullInvocations = new AtomicInteger(); + AtomicReference firstFailure = new AtomicReference<>(); + AtomicReference secondFailure = new AtomicReference<>(); + + Mockito.doAnswer(invocation -> { + int count = pullInvocations.incrementAndGet(); + if (count == 1) { + firstPullStarted.countDown(); + if (!releaseFirstPull.await(5, TimeUnit.SECONDS)) { + throw new AssertionError("Timed out waiting to release first pull"); + } + } + return ""; + }).when(mockExecutor).executePrivilegedOperation(any(), + any(PrivilegedOperation.class), any(), any(), anyBoolean(), + anyBoolean()); + + Thread firstPullThread = new Thread(() -> { + try { + runtime.pullImageFromRemote(containerIdStr, image); + } catch (Throwable t) { + firstFailure.set(t); + } + }); + Thread secondPullThread = new Thread(() -> { + try { + runtime.pullImageFromRemote(containerIdStr, image); + } catch (Throwable t) { + secondFailure.set(t); + } + }); + + firstPullThread.start(); + assertTrue(firstPullStarted.await(5, TimeUnit.SECONDS)); + secondPullThread.start(); + GenericTestUtils.waitFor( + () -> secondPullThread.getState() == Thread.State.BLOCKED, + 10, 5000); + assertEquals(1, pullInvocations.get()); + + releaseFirstPull.countDown(); + firstPullThread.join(5000); + secondPullThread.join(5000); + + assertFalse(firstPullThread.isAlive()); + assertFalse(secondPullThread.isAlive()); + assertNull(firstFailure.get()); + assertNull(secondFailure.get()); + assertEquals(2, pullInvocations.get()); + assertEquals(0, runtime.getImagePullLockCount()); + } + + @Test + public void testConcurrentPullsForDifferentImagesRunInParallel() + throws Exception { + initHttps(false); + DockerLinuxContainerRuntime runtime = + new DockerLinuxContainerRuntime(mockExecutor, mockCGroupsHandler); + runtime.initialize(conf, nmContext); + + CountDownLatch firstPullStarted = new CountDownLatch(1); + CountDownLatch releaseFirstPull = new CountDownLatch(1); + CountDownLatch secondPullInvoked = new CountDownLatch(1); + AtomicInteger pullInvocations = new AtomicInteger(); + AtomicReference firstFailure = new AtomicReference<>(); + AtomicReference secondFailure = new AtomicReference<>(); + + Mockito.doAnswer(invocation -> { + int count = pullInvocations.incrementAndGet(); + if (count == 1) { + firstPullStarted.countDown(); + if (!releaseFirstPull.await(5, TimeUnit.SECONDS)) { + throw new AssertionError("Timed out waiting to release first pull"); + } + } else if (count == 2) { + secondPullInvoked.countDown(); + } + return ""; + }).when(mockExecutor).executePrivilegedOperation(any(), + any(PrivilegedOperation.class), any(), any(), anyBoolean(), + anyBoolean()); + + Thread firstPullThread = new Thread(() -> { + try { + runtime.pullImageFromRemote(containerIdStr, image); + } catch (Throwable t) { + firstFailure.set(t); + } + }); + Thread secondPullThread = new Thread(() -> { + try { + runtime.pullImageFromRemote(containerIdStr, "alpine:latest"); + } catch (Throwable t) { + secondFailure.set(t); + } + }); + + firstPullThread.start(); + assertTrue(firstPullStarted.await(5, TimeUnit.SECONDS)); + secondPullThread.start(); + assertTrue(secondPullInvoked.await(5, TimeUnit.SECONDS)); + assertEquals(2, pullInvocations.get()); + + releaseFirstPull.countDown(); + firstPullThread.join(5000); + secondPullThread.join(5000); + + assertFalse(firstPullThread.isAlive()); + assertFalse(secondPullThread.isAlive()); + assertNull(firstFailure.get()); + assertNull(secondFailure.get()); + assertEquals(0, runtime.getImagePullLockCount()); + } + + @Test + public void testImagePullLockIsReleasedAfterFailure() throws Exception { + initHttps(false); + DockerLinuxContainerRuntime runtime = + new DockerLinuxContainerRuntime(mockExecutor, mockCGroupsHandler); + runtime.initialize(conf, nmContext); + + Mockito.doThrow(new PrivilegedOperationException("pull failed")) + .when(mockExecutor).executePrivilegedOperation(any(), + any(PrivilegedOperation.class), any(), any(), anyBoolean(), + anyBoolean()); + + assertThrows(ContainerExecutionException.class, + () -> runtime.pullImageFromRemote(containerIdStr, image)); + assertEquals(0, runtime.getImagePullLockCount()); + } + @ParameterizedTest(name = "https={0}") @MethodSource("data") public void testContainerLaunchWithUserRemapping(boolean pHttps)