diff --git a/pex/atomic_directory.py b/pex/atomic_directory.py
new file mode 100644
index 000000000..bfd3dc945
--- /dev/null
+++ b/pex/atomic_directory.py
@@ -0,0 +1,187 @@
+# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
+# Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+from __future__ import absolute_import
+
+import errno
+import fcntl
+import os
+import threading
+from contextlib import contextmanager
+
+from pex import pex_warnings
+from pex.common import safe_mkdir, safe_rmtree
+from pex.enum import Enum
+from pex.typing import TYPE_CHECKING, cast
+
+if TYPE_CHECKING:
+    from typing import Callable, Iterator, Optional
+
+
+class AtomicDirectory(object):
+    def __init__(self, target_dir):
+        # type: (str) -> None
+        self._target_dir = target_dir
+        self._work_dir = "{}.workdir".format(target_dir)
+
+    @property
+    def work_dir(self):
+        # type: () -> str
+        return self._work_dir
+
+    @property
+    def target_dir(self):
+        # type: () -> str
+        return self._target_dir
+
+    def is_finalized(self):
+        # type: () -> bool
+        return os.path.exists(self._target_dir)
+
+    def finalize(self, source=None):
+        # type: (Optional[str]) -> None
+        """Rename `work_dir` to `target_dir` using `os.rename()`.
+
+        :param source: An optional source offset into the `work_dir`` to use for the atomic update
+                       of `target_dir`. By default, the whole `work_dir` is used.
+
+        If a race is lost and `target_dir` already exists, the `target_dir` dir is left unchanged and
+        the `work_dir` directory will simply be removed.
+        """
+        if self.is_finalized():
+            return
+
+        source = os.path.join(self._work_dir, source) if source else self._work_dir
+        try:
+            # Perform an atomic rename.
+            #
+            # Per the docs: https://docs.python.org/2.7/library/os.html#os.rename
+            #
+            #   The operation may fail on some Unix flavors if src and dst are on different
+            #   filesystems. If successful, the renaming will be an atomic operation (this is a
+            #   POSIX requirement).
+            #
+            # We have satisfied the single filesystem constraint by arranging the `work_dir` to be a
+            # sibling of the `target_dir`.
+            os.rename(source, self._target_dir)
+        except OSError as e:
+            if e.errno not in (errno.EEXIST, errno.ENOTEMPTY):
+                raise e
+        finally:
+            self.cleanup()
+
+    def cleanup(self):
+        # type: () -> None
+        safe_rmtree(self._work_dir)
+
+
+class FileLockStyle(Enum["FileLockStyle.Value"]):
+    class Value(Enum.Value):
+        pass
+
+    BSD = Value("bsd")
+    POSIX = Value("posix")
+
+
+@contextmanager
+def atomic_directory(
+    target_dir,  # type: str
+    lock_style=FileLockStyle.POSIX,  # type: FileLockStyle.Value
+    source=None,  # type: Optional[str]
+):
+    # type: (...) -> Iterator[AtomicDirectory]
+    """A context manager that yields an exclusively locked AtomicDirectory.
+
+    :param target_dir: The target directory to atomically update.
+    :param lock_style: By default, a POSIX fcntl lock will be used to ensure exclusivity.
+    :param source: An optional source offset into the work directory to use for the atomic update
+                   of the target directory. By default, the whole work directory is used.
+
+    If the `target_dir` already exists the enclosed block will be yielded an AtomicDirectory that
+    `is_finalized` to signal there is no work to do.
+
+    If the enclosed block fails the `target_dir` will not be created if it does not already exist.
+
+    The new work directory will be cleaned up regardless of whether the enclosed block succeeds.
+    """
+
+    # We use double-checked locking with the check being target_dir existence and the lock being an
+    # exclusive blocking file lock.
+
+    atomic_dir = AtomicDirectory(target_dir=target_dir)
+    if atomic_dir.is_finalized():
+        # Our work is already done for us so exit early.
+        yield atomic_dir
+        return
+
+    head, tail = os.path.split(atomic_dir.target_dir)
+    if head:
+        safe_mkdir(head)
+    lockfile = os.path.join(head, ".{}.atomic_directory.lck".format(tail or "here"))
+
+    # N.B.: We don't actually write anything to the lock file but the fcntl file locking
+    # operations only work on files opened for at least write.
+    lock_fd = os.open(lockfile, os.O_CREAT | os.O_WRONLY)
+
+    lock_api = cast(
+        "Callable[[int, int], None]",
+        fcntl.flock if lock_style is FileLockStyle.BSD else fcntl.lockf,
+    )
+
+    def unlock():
+        # type: () -> None
+        try:
+            lock_api(lock_fd, fcntl.LOCK_UN)
+        finally:
+            os.close(lock_fd)
+
+    # N.B.: Since lockf and flock operate on an open file descriptor and these are
+    # guaranteed to be closed by the operating system when the owning process exits,
+    # this lock is immune to staleness.
+    lock_api(lock_fd, fcntl.LOCK_EX)  # A blocking write lock.
+    if atomic_dir.is_finalized():
+        # We lost the double-checked locking race and our work was done for us by the race
+        # winner so exit early.
+        try:
+            yield atomic_dir
+        finally:
+            unlock()
+        return
+
+    # If there is an error making the work_dir that means that either file-locking guarantees have
+    # failed somehow and another process has the lock and has made the work_dir already or else a
+    # process holding the lock ended abnormally.
+    try:
+        os.mkdir(atomic_dir.work_dir)
+    except OSError as e:
+        ident = "[pid:{pid}, tid:{tid}, cwd:{cwd}]".format(
+            pid=os.getpid(), tid=threading.current_thread().ident, cwd=os.getcwd()
+        )
+        pex_warnings.warn(
+            "{ident}: After obtaining an exclusive lock on {lockfile}, failed to establish a work "
+            "directory at {workdir} due to: {err}".format(
+                ident=ident,
+                lockfile=lockfile,
+                workdir=atomic_dir.work_dir,
+                err=e,
+            ),
+        )
+        if e.errno != errno.EEXIST:
+            raise
+        pex_warnings.warn(
+            "{ident}: Continuing to forcibly re-create the work directory at {workdir}.".format(
+                ident=ident,
+                workdir=atomic_dir.work_dir,
+            )
+        )
+        safe_mkdir(atomic_dir.work_dir, clean=True)
+
+    try:
+        yield atomic_dir
+    except Exception:
+        atomic_dir.cleanup()
+        raise
+    else:
+        atomic_dir.finalize(source=source)
+    finally:
+        unlock()
diff --git a/pex/common.py b/pex/common.py
index e3adca82a..5df9a3192 100644
--- a/pex/common.py
+++ b/pex/common.py
@@ -6,7 +6,6 @@
 import atexit
 import contextlib
 import errno
-import fcntl
 import itertools
 import os
 import re
@@ -18,12 +17,10 @@
 import time
 import zipfile
 from collections import defaultdict, namedtuple
-from contextlib import contextmanager
 from datetime import datetime
 from uuid import uuid4
 
-from pex.enum import Enum
-from pex.typing import TYPE_CHECKING, cast
+from pex.typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from typing import (
@@ -37,7 +34,6 @@
         Set,
         Sized,
         Tuple,
-        Union,
     )
 
 # We use the start of MS-DOS time, which is what zipfiles use (see section 4.4.6 of
@@ -333,152 +329,6 @@ def safe_sleep(seconds):
             current_time = time.time()
 
 
-class AtomicDirectory(object):
-    def __init__(self, target_dir):
-        # type: (str) -> None
-        self._target_dir = target_dir
-        self._work_dir = "{}.{}".format(target_dir, uuid4().hex)
-
-    @property
-    def work_dir(self):
-        # type: () -> str
-        return self._work_dir
-
-    @property
-    def target_dir(self):
-        # type: () -> str
-        return self._target_dir
-
-    def is_finalized(self):
-        # type: () -> bool
-        return os.path.exists(self._target_dir)
-
-    def finalize(self, source=None):
-        # type: (Optional[str]) -> None
-        """Rename `work_dir` to `target_dir` using `os.rename()`.
-
-        :param source: An optional source offset into the `work_dir`` to use for the atomic update
-                       of `target_dir`. By default the whole `work_dir` is used.
-
-        If a race is lost and `target_dir` already exists, the `target_dir` dir is left unchanged and
-        the `work_dir` directory will simply be removed.
-        """
-        if self.is_finalized():
-            return
-
-        source = os.path.join(self._work_dir, source) if source else self._work_dir
-        try:
-            # Perform an atomic rename.
-            #
-            # Per the docs: https://docs.python.org/2.7/library/os.html#os.rename
-            #
-            #   The operation may fail on some Unix flavors if src and dst are on different filesystems.
-            #   If successful, the renaming will be an atomic operation (this is a POSIX requirement).
-            #
-            # We have satisfied the single filesystem constraint by arranging the `work_dir` to be a
-            # sibling of the `target_dir`.
-            os.rename(source, self._target_dir)
-        except OSError as e:
-            if e.errno not in (errno.EEXIST, errno.ENOTEMPTY):
-                raise e
-        finally:
-            self.cleanup()
-
-    def cleanup(self):
-        # type: () -> None
-        safe_rmtree(self._work_dir)
-
-
-class FileLockStyle(Enum["FileLockStyle.Value"]):
-    class Value(Enum.Value):
-        pass
-
-    BSD = Value("bsd")
-    POSIX = Value("posix")
-
-
-@contextmanager
-def atomic_directory(
-    target_dir,  # type: str
-    exclusive,  # type: Union[bool, FileLockStyle.Value]
-    source=None,  # type: Optional[str]
-):
-    # type: (...) -> Iterator[AtomicDirectory]
-    """A context manager that yields a potentially exclusively locked AtomicDirectory.
-
-    :param target_dir: The target directory to atomically update.
-    :param exclusive: If `True`, its guaranteed that only one process will be yielded a non `None`
-                      workdir; otherwise two or more processes might be yielded unique non-`None`
-                      workdirs with the last process to finish "winning". By default, a POSIX fcntl
-                      lock will be used to ensure exclusivity. To change this, pass an explicit
-                      `LockStyle` instead of `True`.
-    :param source: An optional source offset into the work directory to use for the atomic update
-                   of the target directory. By default the whole work directory is used.
-
-    If the `target_dir` already exists the enclosed block will be yielded an AtomicDirectory that
-    `is_finalized` to signal there is no work to do.
-
-    If the enclosed block fails the `target_dir` will be undisturbed.
-
-    The new work directory will be cleaned up regardless of whether or not the enclosed block
-    succeeds.
-
-    If the contents of the resulting directory will be subsequently mutated it's probably correct to
-    pass `exclusive=True` to ensure mutations that race the creation process are not lost.
-    """
-    atomic_dir = AtomicDirectory(target_dir=target_dir)
-    if atomic_dir.is_finalized():
-        # Our work is already done for us so exit early.
-        yield atomic_dir
-        return
-
-    lock_fd = None  # type: Optional[int]
-    lock_api = cast(
-        "Callable[[int, int], None]",
-        fcntl.flock if exclusive is FileLockStyle.BSD else fcntl.lockf,
-    )
-
-    def unlock():
-        # type: () -> None
-        if lock_fd is None:
-            return
-        try:
-            lock_api(lock_fd, fcntl.LOCK_UN)
-        finally:
-            os.close(lock_fd)
-
-    if exclusive:
-        head, tail = os.path.split(atomic_dir.target_dir)
-        if head:
-            safe_mkdir(head)
-        # N.B.: We don't actually write anything to the lock file but the fcntl file locking
-        # operations only work on files opened for at least write.
-        lock_fd = os.open(
-            os.path.join(head, ".{}.atomic_directory.lck".format(tail or "here")),
-            os.O_CREAT | os.O_WRONLY,
-        )
-        # N.B.: Since lockf and flock operate on an open file descriptor and these are
-        # guaranteed to be closed by the operating system when the owning process exits,
-        # this lock is immune to staleness.
-        lock_api(lock_fd, fcntl.LOCK_EX)  # A blocking write lock.
-        if atomic_dir.is_finalized():
-            # We lost the double-checked locking race and our work was done for us by the race
-            # winner so exit early.
-            try:
-                yield atomic_dir
-            finally:
-                unlock()
-            return
-
-    try:
-        os.makedirs(atomic_dir.work_dir)
-        yield atomic_dir
-        atomic_dir.finalize(source=source)
-    finally:
-        unlock()
-        atomic_dir.cleanup()
-
-
 def chmod_plus_x(path):
     # type: (str) -> None
     """Equivalent of unix `chmod a+x path`"""
diff --git a/pex/interpreter.py b/pex/interpreter.py
index 4262bcce0..fe4f03340 100644
--- a/pex/interpreter.py
+++ b/pex/interpreter.py
@@ -828,12 +828,13 @@ def create_interpreter(
                         import os
                         import sys
 
-                        from pex.common import atomic_directory, safe_open
+                        from pex.atomic_directory import atomic_directory
+                        from pex.common import safe_open
                         from pex.interpreter import PythonIdentity
 
 
                         encoded_identity = PythonIdentity.get(binary={binary!r}).encode()
-                        with atomic_directory({cache_dir!r}, exclusive=False) as cache_dir:
+                        with atomic_directory({cache_dir!r}) as cache_dir:
                             if not cache_dir.is_finalized():
                                 with safe_open(
                                     os.path.join(cache_dir.work_dir, {info_file!r}), 'w'
diff --git a/pex/layout.py b/pex/layout.py
index b11c84588..0923c7853 100644
--- a/pex/layout.py
+++ b/pex/layout.py
@@ -8,7 +8,8 @@
 from abc import abstractmethod
 from contextlib import contextmanager
 
-from pex.common import atomic_directory, is_script, open_zip, safe_copy, safe_mkdir
+from pex.atomic_directory import atomic_directory
+from pex.common import is_script, open_zip, safe_copy, safe_mkdir
 from pex.enum import Enum
 from pex.tracer import TRACER
 from pex.typing import TYPE_CHECKING
@@ -104,7 +105,7 @@ def _install(
     with TRACER.timed("Laying out {}".format(layout)):
         pex = layout.path
         install_to = unzip_dir(pex_root=pex_root, pex_hash=pex_hash)
-        with atomic_directory(install_to, exclusive=True) as chroot:
+        with atomic_directory(install_to) as chroot:
             if not chroot.is_finalized():
                 with TRACER.timed("Installing {} to {}".format(pex, install_to)):
                     from pex.pex_info import PexInfo
@@ -124,7 +125,7 @@ def _install(
                         )
 
                     with atomic_directory(
-                        bootstrap_cache, source=layout.bootstrap_strip_prefix(), exclusive=True
+                        bootstrap_cache, source=layout.bootstrap_strip_prefix()
                     ) as bootstrap_zip_chroot:
                         if not bootstrap_zip_chroot.is_finalized():
                             layout.extract_bootstrap(bootstrap_zip_chroot.work_dir)
@@ -139,7 +140,6 @@ def _install(
                         with atomic_directory(
                             spread_dest,
                             source=layout.dist_strip_prefix(location),
-                            exclusive=True,
                         ) as spread_chroot:
                             if not spread_chroot.is_finalized():
                                 layout.extract_dist(spread_chroot.work_dir, dist_relpath)
@@ -154,7 +154,7 @@ def _install(
                         )
 
                     code_dest = os.path.join(pex_info.zip_unsafe_cache, code_hash)
-                    with atomic_directory(code_dest, exclusive=True) as code_chroot:
+                    with atomic_directory(code_dest) as code_chroot:
                         if not code_chroot.is_finalized():
                             layout.extract_code(code_chroot.work_dir)
                     for path in os.listdir(code_dest):
diff --git a/pex/pex_bootstrapper.py b/pex/pex_bootstrapper.py
index e06824051..7f5337446 100644
--- a/pex/pex_bootstrapper.py
+++ b/pex/pex_bootstrapper.py
@@ -8,7 +8,8 @@
 import sys
 
 from pex import pex_warnings
-from pex.common import atomic_directory, die, pluralize
+from pex.atomic_directory import atomic_directory
+from pex.common import die, pluralize
 from pex.environment import ResolveError
 from pex.inherit_path import InheritPath
 from pex.interpreter import PythonInterpreter
@@ -482,7 +483,7 @@ def ensure_venv(
             "The PEX_VENV environment variable was set, but this PEX was not built with venv "
             "support (Re-build the PEX file with `pex --venv ...`)"
         )
-    with atomic_directory(venv_dir, exclusive=True) as venv:
+    with atomic_directory(venv_dir) as venv:
         if not venv.is_finalized():
             from pex.venv.pex import populate_venv
             from pex.venv.virtualenv import Virtualenv
@@ -505,7 +506,7 @@ def ensure_venv(
             for chars in range(8, len(venv_hash) + 1):
                 entropy = venv_hash[:chars]
                 short_venv_dir = os.path.join(pex_info.pex_root, "venvs", "s", entropy)
-                with atomic_directory(short_venv_dir, exclusive=True) as short_venv:
+                with atomic_directory(short_venv_dir) as short_venv:
                     if short_venv.is_finalized():
                         collisions.append(short_venv_dir)
                         if entropy == venv_hash:
diff --git a/pex/pex_builder.py b/pex/pex_builder.py
index e655acabb..2ee6ca3ea 100644
--- a/pex/pex_builder.py
+++ b/pex/pex_builder.py
@@ -9,9 +9,9 @@
 import shutil
 
 from pex import pex_warnings
+from pex.atomic_directory import atomic_directory
 from pex.common import (
     Chroot,
-    atomic_directory,
     chmod_plus_x,
     filter_pyc_files,
     is_pyc_temporary_file,
@@ -724,9 +724,7 @@ def zip_cache_dir(path):
         cached_bootstrap_zip_dir = zip_cache_dir(
             os.path.join(pex_info.pex_root, "bootstrap_zips", bootstrap_hash)
         )
-        with atomic_directory(
-            cached_bootstrap_zip_dir, exclusive=False
-        ) as atomic_bootstrap_zip_dir:
+        with atomic_directory(cached_bootstrap_zip_dir) as atomic_bootstrap_zip_dir:
             if not atomic_bootstrap_zip_dir.is_finalized():
                 self._chroot.zip(
                     os.path.join(atomic_bootstrap_zip_dir.work_dir, pex_info.bootstrap),
@@ -750,9 +748,7 @@ def zip_cache_dir(path):
                 cached_installed_wheel_zip_dir = zip_cache_dir(
                     os.path.join(pex_info.pex_root, "installed_wheel_zips", fingerprint)
                 )
-                with atomic_directory(
-                    cached_installed_wheel_zip_dir, exclusive=False
-                ) as atomic_zip_dir:
+                with atomic_directory(cached_installed_wheel_zip_dir) as atomic_zip_dir:
                     if not atomic_zip_dir.is_finalized():
                         self._chroot.zip(
                             os.path.join(atomic_zip_dir.work_dir, location),
diff --git a/pex/pip/installation.py b/pex/pip/installation.py
index 068bdd8db..a931fec78 100644
--- a/pex/pip/installation.py
+++ b/pex/pip/installation.py
@@ -7,7 +7,7 @@
 from textwrap import dedent
 
 from pex import pex_warnings, third_party
-from pex.common import atomic_directory
+from pex.atomic_directory import atomic_directory
 from pex.interpreter import PythonInterpreter
 from pex.orderedset import OrderedSet
 from pex.pex import PEX
@@ -38,7 +38,7 @@ def _pip_venv(
     path = os.path.join(ENV.PEX_ROOT, "pip-{version}.pex".format(version=version))
     pip_interpreter = interpreter or PythonInterpreter.get()
     pip_pex_path = os.path.join(path, isolated().pex_hash)
-    with atomic_directory(pip_pex_path, exclusive=True) as chroot:
+    with atomic_directory(pip_pex_path) as chroot:
         if not chroot.is_finalized():
             from pex.pex_builder import PEXBuilder
 
diff --git a/pex/platforms.py b/pex/platforms.py
index 05c5ceba2..f4ce3c311 100644
--- a/pex/platforms.py
+++ b/pex/platforms.py
@@ -9,7 +9,8 @@
 from textwrap import dedent
 
 from pex import compatibility
-from pex.common import atomic_directory, safe_open, safe_rmtree
+from pex.atomic_directory import atomic_directory
+from pex.common import safe_open, safe_rmtree
 from pex.pep_425 import CompatibilityTags
 from pex.third_party.packaging import tags
 from pex.tracer import TRACER
@@ -265,7 +266,7 @@ def supported_tags(self, manylinux=None):
         if manylinux:
             components.append(manylinux)
         disk_cache_key = os.path.join(ENV.PEX_ROOT, "platforms", self.SEP.join(components))
-        with atomic_directory(target_dir=disk_cache_key, exclusive=False) as cache_dir:
+        with atomic_directory(target_dir=disk_cache_key) as cache_dir:
             if not cache_dir.is_finalized():
                 # Missed both caches - spawn calculation.
                 plat_info = attr.asdict(self)
diff --git a/pex/resolve/downloads.py b/pex/resolve/downloads.py
index 72003ea31..57cc0b911 100644
--- a/pex/resolve/downloads.py
+++ b/pex/resolve/downloads.py
@@ -4,7 +4,8 @@
 import shutil
 
 from pex import hashing
-from pex.common import atomic_directory, safe_mkdir, safe_mkdtemp
+from pex.atomic_directory import atomic_directory
+from pex.common import safe_mkdir, safe_mkdtemp
 from pex.compatibility import unquote, urlparse
 from pex.fetcher import URLFetcher
 from pex.hashing import Sha256
@@ -71,7 +72,7 @@ def _fingerprint_and_move(path):
         hashing.file_hash(path, digest)
         fingerprint = digest.hexdigest()
         target_dir = os.path.join(get_downloads_dir(), fingerprint)
-        with atomic_directory(target_dir, exclusive=True) as atomic_dir:
+        with atomic_directory(target_dir) as atomic_dir:
             if not atomic_dir.is_finalized():
                 shutil.move(path, os.path.join(atomic_dir.work_dir, os.path.basename(path)))
         return Fingerprint(algorithm=fingerprint.algorithm, hash=fingerprint)
diff --git a/pex/resolve/lock_resolver.py b/pex/resolve/lock_resolver.py
index 9a88a6fcc..b1fb8abd1 100644
--- a/pex/resolve/lock_resolver.py
+++ b/pex/resolve/lock_resolver.py
@@ -10,8 +10,9 @@
 from multiprocessing.pool import ThreadPool
 
 from pex import resolver
+from pex.atomic_directory import FileLockStyle
 from pex.auth import PasswordDatabase, PasswordEntry
-from pex.common import FileLockStyle, pluralize
+from pex.common import pluralize
 from pex.compatibility import cpu_count
 from pex.network_configuration import NetworkConfiguration
 from pex.orderedset import OrderedSet
diff --git a/pex/resolve/lockfile/download_manager.py b/pex/resolve/lockfile/download_manager.py
index d8946019e..a17a67432 100644
--- a/pex/resolve/lockfile/download_manager.py
+++ b/pex/resolve/lockfile/download_manager.py
@@ -8,7 +8,8 @@
 import os
 
 from pex import hashing
-from pex.common import FileLockStyle, atomic_directory, safe_rmtree
+from pex.atomic_directory import FileLockStyle, atomic_directory
+from pex.common import safe_rmtree
 from pex.pep_503 import ProjectName
 from pex.resolve.downloads import get_downloads_dir
 from pex.resolve.locked_resolve import Artifact
@@ -127,7 +128,7 @@ def store(
         download_dir = os.path.join(
             get_downloads_dir(pex_root=self._pex_root), artifact.fingerprint.hash
         )
-        with atomic_directory(download_dir, exclusive=self._file_lock_style) as atomic_dir:
+        with atomic_directory(download_dir, lock_style=self._file_lock_style) as atomic_dir:
             if atomic_dir.is_finalized():
                 TRACER.log("Using cached artifact at {} for {}".format(download_dir, artifact))
             else:
diff --git a/pex/resolver.py b/pex/resolver.py
index 4d2c44fb7..d73e6aaa9 100644
--- a/pex/resolver.py
+++ b/pex/resolver.py
@@ -13,8 +13,9 @@
 from collections import OrderedDict, defaultdict
 
 from pex import targets
+from pex.atomic_directory import AtomicDirectory, atomic_directory
 from pex.auth import PasswordEntry
-from pex.common import AtomicDirectory, atomic_directory, safe_mkdir, safe_mkdtemp
+from pex.common import safe_mkdir, safe_mkdtemp
 from pex.dist_metadata import Distribution, Requirement
 from pex.fingerprinted_distribution import FingerprintedDistribution
 from pex.jobs import Raise, SpawnedJob, execute_parallel
@@ -449,7 +450,7 @@ def finalize_install(self, install_requests):
         #
         wheel_dir_hash = fingerprint_path(self.install_chroot)
         runtime_key_dir = os.path.join(self._installation_root, wheel_dir_hash)
-        with atomic_directory(runtime_key_dir, exclusive=False) as atomic_dir:
+        with atomic_directory(runtime_key_dir) as atomic_dir:
             if not atomic_dir.is_finalized():
                 # Note: Create a relative path symlink between the two directories so that the
                 # PEX_ROOT can be used within a chroot environment where the prefix of the path may
diff --git a/pex/testing.py b/pex/testing.py
index 50b006327..4752d4f39 100644
--- a/pex/testing.py
+++ b/pex/testing.py
@@ -10,27 +10,18 @@
 import random
 import subprocess
 import sys
-from collections import OrderedDict
 from contextlib import contextmanager
 from textwrap import dedent
 
 import pytest
 
-from pex.common import (
-    atomic_directory,
-    open_zip,
-    safe_mkdir,
-    safe_mkdtemp,
-    safe_rmtree,
-    safe_sleep,
-    temporary_dir,
-)
+from pex.atomic_directory import atomic_directory
+from pex.common import open_zip, safe_mkdir, safe_mkdtemp, safe_rmtree, safe_sleep, temporary_dir
 from pex.compatibility import to_unicode
 from pex.dist_metadata import Distribution
 from pex.enum import Enum
 from pex.executor import Executor
 from pex.interpreter import PythonInterpreter
-from pex.orderedset import OrderedSet
 from pex.pex import PEX
 from pex.pex_builder import PEXBuilder
 from pex.pex_info import PexInfo
@@ -493,13 +484,11 @@ def ensure_python_distribution(version):
 
     pip = os.path.join(interpreter_location, "bin", "pip")
 
-    with atomic_directory(target_dir=os.path.join(pyenv_root), exclusive=True) as target_dir:
+    with atomic_directory(target_dir=os.path.join(pyenv_root)) as target_dir:
         if not target_dir.is_finalized():
             bootstrap_python_installer(target_dir.work_dir)
 
-    with atomic_directory(
-        target_dir=interpreter_location, exclusive=True
-    ) as interpreter_target_dir:
+    with atomic_directory(target_dir=interpreter_location) as interpreter_target_dir:
         if not interpreter_target_dir.is_finalized():
             subprocess.check_call(
                 [
diff --git a/pex/third_party/__init__.py b/pex/third_party/__init__.py
index 6b7793c16..05d4fa1a6 100644
--- a/pex/third_party/__init__.py
+++ b/pex/third_party/__init__.py
@@ -447,7 +447,7 @@ def isolated():
     global _ISOLATED
     if _ISOLATED is None:
         from pex import layout, vendor
-        from pex.common import atomic_directory
+        from pex.atomic_directory import atomic_directory
         from pex.util import CacheHelper
         from pex.variables import ENV
 
@@ -491,7 +491,7 @@ def isolated():
 
         isolated_dir = os.path.join(ENV.PEX_ROOT, "isolated", pex_hash)
         with _tracer().timed("Isolating pex"):
-            with atomic_directory(isolated_dir, exclusive=True) as chroot:
+            with atomic_directory(isolated_dir) as chroot:
                 if not chroot.is_finalized():
                     with _tracer().timed("Extracting pex to {}".format(isolated_dir)):
                         if pex_zip_paths:
diff --git a/pex/venv/virtualenv.py b/pex/venv/virtualenv.py
index c1c770c2d..2544ed453 100644
--- a/pex/venv/virtualenv.py
+++ b/pex/venv/virtualenv.py
@@ -12,7 +12,8 @@
 from contextlib import closing
 from fileinput import FileInput
 
-from pex.common import AtomicDirectory, atomic_directory, is_exe, safe_mkdir, safe_open
+from pex.atomic_directory import AtomicDirectory, atomic_directory
+from pex.common import is_exe, safe_mkdir, safe_open
 from pex.compatibility import commonpath, get_stdout_bytes_buffer
 from pex.dist_metadata import Distribution, find_distributions
 from pex.executor import Executor
@@ -366,7 +367,7 @@ def install_pip(self):
                 url_rel_path = get_pip_script
                 dst_rel_path = os.path.join("default", get_pip_script)
             get_pip = os.path.join(ENV.PEX_ROOT, "get-pip", dst_rel_path)
-            with atomic_directory(os.path.dirname(get_pip), exclusive=True) as atomic_dir:
+            with atomic_directory(os.path.dirname(get_pip)) as atomic_dir:
                 if not atomic_dir.is_finalized():
                     with URLFetcher().get_body_stream(
                         "https://bootstrap.pypa.io/pip/" + url_rel_path
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index 52e4fe0a2..c6859d24b 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -10,7 +10,8 @@
 
 import pytest
 
-from pex.common import atomic_directory, safe_mkdtemp, temporary_dir
+from pex.atomic_directory import atomic_directory
+from pex.common import safe_mkdtemp, temporary_dir
 from pex.testing import PY310, ensure_python_venv, make_env, run_pex_command
 from pex.typing import TYPE_CHECKING
 
@@ -48,7 +49,7 @@ def pex_bdist(
     # type: (...) -> str
     pex_bdist_chroot = os.path.join(shared_integration_test_tmpdir, "pex_bdist_chroot")
     wheels_dir = os.path.join(pex_bdist_chroot, "wheels_dir")
-    with atomic_directory(pex_bdist_chroot, exclusive=True) as chroot:
+    with atomic_directory(pex_bdist_chroot) as chroot:
         if not chroot.is_finalized():
             pex_pex = os.path.join(chroot.work_dir, "pex.pex")
             run_pex_command(
diff --git a/tests/resolve/lockfile/test_download_manager.py b/tests/resolve/lockfile/test_download_manager.py
index 5e1a9bfb0..1e1bbb98a 100644
--- a/tests/resolve/lockfile/test_download_manager.py
+++ b/tests/resolve/lockfile/test_download_manager.py
@@ -128,9 +128,12 @@ def test_storage_version_upgrade(
 
     downloaded_artifact2 = download_manager.store(artifact, project_name)
     assert downloaded_artifact1 == downloaded_artifact2
-    assert 2 == len(set(download_manager.save_calls)), (
-        "Expected two save calls, each with a different atomic directory work dir signalling a "
-        "re-build of the artifact storage"
+    assert 2 == len(
+        download_manager.save_calls
+    ), "Expected two save calls signalling a re-build of the artifact storage."
+    assert 1 == len(set(download_manager.save_calls)), (
+        "Expected each save call is with the same atomic directory work dir signalling a re-build "
+        "of the same artifact storage."
     )
 
 
diff --git a/tests/test_common.py b/tests/test_common.py
index 93273ae88..fa6cbb109 100644
--- a/tests/test_common.py
+++ b/tests/test_common.py
@@ -8,11 +8,10 @@
 
 import pytest
 
+from pex.atomic_directory import AtomicDirectory, atomic_directory
 from pex.common import (
-    AtomicDirectory,
     Chroot,
     PermPreservingZipFile,
-    atomic_directory,
     can_write_dir,
     chmod_plus_x,
     is_exe,
@@ -40,7 +39,8 @@ def maybe_raises(exception=None):
     def noop():
         yield
 
-    with (noop() if exception is None else pytest.raises(exception)):
+    context = noop() if exception is None else pytest.raises(exception)
+    with context:
         yield
 
 
@@ -73,7 +73,7 @@ def test_atomic_directory_empty_workdir_finalize():
         target_dir = os.path.join(sandbox, "target_dir")
         assert not os.path.exists(target_dir)
 
-        with atomic_directory(target_dir, exclusive=False) as atomic_dir:
+        with atomic_directory(target_dir) as atomic_dir:
             assert not atomic_dir.is_finalized()
             assert target_dir == atomic_dir.target_dir
             assert os.path.exists(atomic_dir.work_dir)
@@ -96,7 +96,7 @@ class SimulatedRuntimeError(RuntimeError):
     with temporary_dir() as sandbox:
         target_dir = os.path.join(sandbox, "target_dir")
         with pytest.raises(SimulatedRuntimeError):
-            with atomic_directory(target_dir, exclusive=False) as atomic_dir:
+            with atomic_directory(target_dir) as atomic_dir:
                 assert not atomic_dir.is_finalized()
                 touch(os.path.join(atomic_dir.work_dir, "created"))
                 raise SimulatedRuntimeError()
@@ -113,7 +113,7 @@ class SimulatedRuntimeError(RuntimeError):
 def test_atomic_directory_empty_workdir_finalized():
     # type: () -> None
     with temporary_dir() as target_dir:
-        with atomic_directory(target_dir, exclusive=False) as work_dir:
+        with atomic_directory(target_dir) as work_dir:
             assert (
                 work_dir.is_finalized()
             ), "When the target_dir exists no work_dir should be created."
diff --git a/tests/test_enum.py b/tests/test_enum.py
index 87c61389d..195b15619 100644
--- a/tests/test_enum.py
+++ b/tests/test_enum.py
@@ -4,7 +4,8 @@
 
 import pytest
 
-from pex.common import AtomicDirectory, PermPreservingZipFile
+from pex.atomic_directory import AtomicDirectory
+from pex.common import PermPreservingZipFile
 from pex.compatibility import PY2
 from pex.enum import Enum, qualified_name
 
@@ -104,10 +105,10 @@ def test_qualified_name():
         qualified_name
     ), "Expected functions to be handled"
 
-    assert "pex.common.AtomicDirectory" == qualified_name(
+    assert "pex.atomic_directory.AtomicDirectory" == qualified_name(
         AtomicDirectory
     ), "Expected custom types to be handled."
-    expected_prefix = "pex.common." if PY2 else "pex.common.AtomicDirectory."
+    expected_prefix = "pex.atomic_directory." if PY2 else "pex.atomic_directory.AtomicDirectory."
     assert expected_prefix + "finalize" == qualified_name(
         AtomicDirectory.finalize
     ), "Expected methods to be handled."