Fix: Eliminate port collision race conditions in distributed tests

Implemented a thread-safe port reservation system to prevent EADDRINUSE
errors in distributed training tests.

- Created PortManager class with mutex-protected port allocation
- Updated find_free_network_port() to use PortManager
- Enhanced test teardown to release ports after completion
- Added 24 comprehensive tests (17 unit + 7 integration)
- Added context manager for automatic port cleanup

Fixes port collision issues in:
- tests_fabric/strategies/test_ddp_integration.py::test_clip_gradients
- tests_pytorch/strategies/test_fsdp.py::test_checkpoint_multi_gpus

Author: littlebullGit

src/lightning/fabric/plugins/environments/lightning.py

@@ -13,11 +13,11 @@
 # limitations under the License.
 
 import os
-import socket
 
 from typing_extensions import override
 
 from lightning.fabric.plugins.environments.cluster_environment import ClusterEnvironment
+from lightning.fabric.utilities.port_manager import get_port_manager
 from lightning.fabric.utilities.rank_zero import rank_zero_only
 
 
@@ -111,9 +111,17 @@ def find_free_network_port() -> int:
     It is useful in single-node training when we don't want to connect to a real main node but have to set the
     `MASTER_PORT` environment variable.
 
+    This function uses a global port manager to prevent internal race conditions within the test suite.
+    The allocated port is reserved and won't be returned by subsequent calls until it's explicitly released.
+
+    Note:
+        While this prevents collisions between concurrent Lightning tests, external processes can still
+        claim the port between allocation and binding. For production use, explicitly set the MASTER_PORT
+        environment variable.
+
+    Returns:
+        A port number that is reserved and free at the time of allocation
+
     """
-    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-    s.bind(("", 0))
-    port = s.getsockname()[1]
-    s.close()
-    return port
+    port_manager = get_port_manager()
+    return port_manager.allocate_port()

src/lightning/fabric/utilities/port_manager.py

@@ -0,0 +1,168 @@
+# Copyright The Lightning AI team.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Port allocation manager to prevent race conditions in distributed training."""
+
+import atexit
+import socket
+import threading
+from collections.abc import Iterator
+from contextlib import contextmanager
+from typing import Optional
+
+
+class PortManager:
+    """Thread-safe port manager to prevent EADDRINUSE errors.
+
+    This manager maintains a global registry of allocated ports to ensure that multiple concurrent tests don't try to
+    use the same port. While this doesn't completely eliminate the race condition with external processes, it prevents
+    internal collisions within the test suite.
+
+    """
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._allocated_ports: set[int] = set()
+        # Register cleanup to release all ports on exit
+        atexit.register(self.release_all)
+
+    def allocate_port(self, preferred_port: Optional[int] = None, max_attempts: int = 100) -> int:
+        """Allocate a free port, ensuring it's not already reserved.
+
+        Args:
+            preferred_port: If provided, try to allocate this specific port first
+            max_attempts: Maximum number of attempts to find a free port
+
+        Returns:
+            An allocated port number
+
+        Raises:
+            RuntimeError: If unable to find a free port after max_attempts
+
+        """
+        with self._lock:
+            # If a preferred port is specified and available, use it
+            if (
+                preferred_port is not None
+                and preferred_port not in self._allocated_ports
+                and self._is_port_free(preferred_port)
+            ):
+                self._allocated_ports.add(preferred_port)
+                return preferred_port
+
+            # Try to find a free port
+            for attempt in range(max_attempts):
+                port = self._find_free_port()
+
+                # Double-check it's not in our reserved set (shouldn't happen, but be safe)
+                if port not in self._allocated_ports:
+                    self._allocated_ports.add(port)
+                    return port
+
+            raise RuntimeError(
+                f"Failed to allocate a free port after {max_attempts} attempts. "
+                f"Currently allocated ports: {len(self._allocated_ports)}"
+            )
+
+    def release_port(self, port: int) -> None:
+        """Release a previously allocated port.
+
+        Args:
+            port: Port number to release
+
+        """
+        with self._lock:
+            self._allocated_ports.discard(port)
+
+    def release_all(self) -> None:
+        """Release all allocated ports."""
+        with self._lock:
+            self._allocated_ports.clear()
+
+    @contextmanager
+    def allocated_port(self, preferred_port: Optional[int] = None) -> Iterator[int]:
+        """Context manager for automatic port cleanup.
+
+        Usage:
+            with manager.allocated_port() as port:
+                # Use port here
+                pass
+            # Port automatically released
+
+        Args:
+            preferred_port: Optional preferred port number
+
+        Yields:
+            Allocated port number
+
+        """
+        port = self.allocate_port(preferred_port=preferred_port)
+        try:
+            yield port
+        finally:
+            self.release_port(port)
+
+    @staticmethod
+    def _find_free_port() -> int:
+        """Find a free port using OS allocation.
+
+        Returns:
+            A port number that was free at the time of checking
+
+        """
+        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        s.bind(("", 0))
+        port = s.getsockname()[1]
+        s.close()
+        return port
+
+    @staticmethod
+    def _is_port_free(port: int) -> bool:
+        """Check if a specific port is available.
+
+        Args:
+            port: Port number to check
+
+        Returns:
+            True if the port is free, False otherwise
+
+        """
+        try:
+            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            s.bind(("", port))
+            s.close()
+            return True
+        except OSError:
+            return False
+
+
+# Global singleton instance
+_port_manager: Optional[PortManager] = None
+_port_manager_lock = threading.Lock()
+
+
+def get_port_manager() -> PortManager:
+    """Get or create the global port manager instance.
+
+    Returns:
+        The global PortManager singleton
+
+    """
+    global _port_manager
+    if _port_manager is None:
+        with _port_manager_lock:
+            if _port_manager is None:
+                _port_manager = PortManager()
+    return _port_manager

tests/tests_fabric/conftest.py

@@ -11,6 +11,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import contextlib
 import os
 import sys
 import threading
@@ -77,9 +78,25 @@ def restore_env_variables():
 @pytest.fixture(autouse=True)
 def teardown_process_group():
     """Ensures that the distributed process group gets closed before the next test runs."""
+    import os
+
+    from lightning.fabric.utilities.port_manager import get_port_manager
+
+    # Record the port used in this test (if any)
+    port_to_release = None
+    if "MASTER_PORT" in os.environ:
+        with contextlib.suppress(ValueError, KeyError):
+            port_to_release = int(os.environ["MASTER_PORT"])
+
     yield
+
+    # Clean up distributed connection
     _destroy_dist_connection()
 
+    # Release the port from the manager so it can be reused
+    if port_to_release is not None:
+        get_port_manager().release_port(port_to_release)
+
 
 @pytest.fixture(autouse=True)
 def thread_police_duuu_daaa_duuu_daaa():