fix: enable parallel test execution with pytest-xdist in CI workflow

.github/workflows/ci-testing.yml

@@ -42,7 +42,9 @@ jobs:
           pip list
 
       - name: Tests
-        run: coverage run --source litdata -m pytest tests -v --durations=100
+        run: |
+          pip install pytest-xdist
+          coverage run --source litdata -m pytest tests -v --durations=100 -n auto
 
       - name: Statistics
         run: |

src/litdata/processing/data_processor.py

@@ -556,6 +556,8 @@ def _terminate(self) -> None:
         if self.remover and self.remover.is_alive():
             self.remover.join()
 
+        sleep(5)  # Give some buffer time for file creation/deletion
+
     def _loop(self) -> None:
         """The main loop of the worker.
 
@@ -1339,6 +1341,9 @@ def run(self, data_recipe: DataRecipe) -> None:
             pbar.close()
 
         print("Workers are finished.")
+
+        sleep(5)  # Give some buffer time for file creation/deletion
+
         size = len(workers_user_items) if workers_user_items is not None else None
         result = data_recipe._done(size, self.delete_cached_files, self.output_dir)
 

tests/conftest.py

@@ -2,6 +2,7 @@
 import shutil
 import sys
 import threading
+import uuid
 from collections import OrderedDict
 from types import ModuleType
 from unittest.mock import Mock
@@ -25,9 +26,12 @@ def teardown_process_group():
 
 
 @pytest.fixture(autouse=True)
-def set_env():
+def set_env(monkeypatch):
     # Set environment variable before each test to configure BaseWorker's maximum wait time
     os.environ["DATA_OPTIMIZER_TIMEOUT"] = "20"
+    uuid_str = uuid.uuid4().hex
+    monkeypatch.setenv("DATA_OPTIMIZER_DATA_CACHE_FOLDER", f"/tmp/{uuid_str}/")  # noqa: S108
+    monkeypatch.setenv("DATA_OPTIMIZER_CACHE_FOLDER", f"/tmp/{uuid_str}/")  # noqa: S108
 
 
 @pytest.fixture

tests/streaming/test_parquet.py

@@ -2,6 +2,7 @@
 import os
 import sys
 import tempfile
+import uuid
 from contextlib import nullcontext
 from fnmatch import fnmatch
 from unittest.mock import Mock, patch
@@ -135,7 +136,8 @@ def test_get_parquet_indexer_cls(pq_url, tmp_path, cls, expectation, monkeypatch
 @pytest.mark.parametrize(("pre_load_chunk"), [False, True])
 @pytest.mark.parametrize(("low_memory"), [False, True])
 def test_stream_hf_parquet_dataset(monkeypatch, huggingface_hub_fs_mock, pq_data, pre_load_chunk, low_memory):
-    hf_url = "hf://datasets/some_org/some_repo/some_path"
+    uuid_str = uuid.uuid4().hex  # Unique identifier for the dataset
+    hf_url = f"hf://datasets/some_org/some_repo/{uuid_str}"
 
     # Test case 1: Invalid item_loader
     with pytest.raises(ValueError, match="Invalid item_loader for hf://datasets."):