fix: resolve DCP test hanging and distributed test errors (#330)

* fix(test): prevent dcp integration tests from hanging

In test_e2e_s3_file_system.py, multi_process_dcp_save_load() spawns
multiple processes to execute run(), and a race condition occurs when
faster processes destroyed shared process group in cleanup().

Isolated process group initialization with explicit TCP arguments, and
added barrier synchronization point to ensure all processes complete
before cleanup occurs.

* ci: run DCP and distributed integration tests sequentially

Removed -n auto flag from DCP and distributed training tests in CI
workflow. This change addresses stability issues in multi-process tests
by preventing potential race conditions, improving test reliability.

Author: jet-tong

.github/workflows/python-integration.yml

@@ -88,7 +88,14 @@ jobs:
           CI_REGION=${{ matrix.test-run.region }} \
           CI_BUCKET=${{ matrix.test-run.bucket }} \
           CI_STORAGE_CLASS=${{ matrix.test-run.storage-class }} \
-          pytest s3torchconnector/tst/e2e --ignore-glob '*/**/test_e2e_s3_lightning_checkpoint.py' --ignore-glob '*/**/dcp' -n auto
+          pytest s3torchconnector/tst/e2e --ignore-glob '*/**/test_e2e_s3_lightning_checkpoint.py' --ignore-glob '*/**/dcp' --ignore-glob '*/**/test_distributed_training.py' -n auto
+
+      - name: s3torchconnector ${{ matrix.test-run.name }} distributed training integration tests
+        run: |
+          CI_REGION=${{ matrix.test-run.region }} \
+          CI_BUCKET=${{ matrix.test-run.bucket }} \
+          CI_STORAGE_CLASS=${{ matrix.test-run.storage-class }} \
+          pytest s3torchconnector/tst/e2e/test_distributed_training.py
 
       - name: Install Lightning dependency
         run: |
@@ -110,7 +117,7 @@ jobs:
           CI_REGION=${{ matrix.test-run.region }} \
           CI_BUCKET=${{ matrix.test-run.bucket }} \
           CI_STORAGE_CLASS=${{ matrix.test-run.storage-class }} \
-          pytest s3torchconnector/tst/e2e/dcp -n auto
+          pytest s3torchconnector/tst/e2e/dcp
 
       - name: s3torchconnectorclient ${{ matrix.test-run.name }} integration tests
         run: |

s3torchconnector/tst/e2e/dcp/test_e2e_s3_file_system.py

@@ -24,13 +24,19 @@ def generate_random_port():
 
 
 def setup(rank, world_size, port):
-    os.environ["MASTER_ADDR"] = "localhost"
-    os.environ["MASTER_PORT"] = port
-    dist.init_process_group("gloo", rank=rank, world_size=world_size)
+    dist.init_process_group(
+        backend="gloo",
+        world_size=world_size,
+        rank=rank,
+        init_method=f"tcp://127.0.0.1:{port}",
+    )
 
 
 def cleanup():
-    dist.destroy_process_group()
+    # Synchronization point: Barrier ensures all process groups reach this point
+    dist.barrier()
+    if dist.is_initialized():
+        dist.destroy_process_group()
 
 
 def run(