Merge pull request #811 from maresb/double-check-on-lookup-cache

Refactor lookup cache test to use multiprocessing instead of threading

Author: maresb

conda_lock/conda_lock.py

@@ -1193,7 +1193,6 @@ def run_lock(
     _conda_exe = determine_conda_executable(
         conda_exe, mamba=mamba, micromamba=micromamba
     )
-    logger.debug(f"Using conda executable: {_conda_exe}")
     version_info = subprocess.check_output(
         [_conda_exe, "--version"], encoding="utf-8"
     ).strip()

conda_lock/invoke_conda.py

@@ -62,6 +62,7 @@ def determine_conda_executable(
             if is_micromamba(candidate):
                 if determine_micromamba_version(str(candidate)) < Version("0.17"):
                     mamba_root_prefix()
+            logger.debug(f"Found conda executable: {candidate}")
             return candidate
     raise RuntimeError("Could not find conda (or compatible) executable")
 

conda_lock/lookup_cache.py

@@ -57,14 +57,17 @@ def cached_download_file(
     cache.mkdir(parents=True, exist_ok=True)
     clear_old_files_from_cache(cache, max_age_seconds=max_age_seconds)
 
-    destination_lock = (cache / cached_filename_for_url(url)).with_suffix(".lock")
+    destination = cache / cached_filename_for_url(url)
+    destination_lock = destination.with_suffix(".lock")
 
     # Wait for any other process to finish downloading the file.
     # This way we can use the result from the current download without
     # spawning multiple concurrent downloads.
     while True:
         try:
+            logger.debug(f"Attempting to acquire lock on {destination_lock}")
             with FileLock(str(destination_lock), timeout=5):
+                logger.debug(f"Successfully acquired lock on {destination_lock}")
                 return _download_to_or_read_from_cache(
                     url,
                     cache=cache,
@@ -77,6 +80,33 @@ def cached_download_file(
             )
 
 
+def _is_cached_file_fresh(
+    destination: Path, dont_check_if_newer_than_seconds: float
+) -> bool:
+    """Check if a cached file exists and is fresh enough to use without checking.
+
+    (In this context, "checking" means that later, beyond the scope of this function,
+    we will query the server with an ETag to see if the file has changed or if we
+    get a 304 Not Modified response.)
+
+    A file is "fresh" if its age is positive and less than
+    `dont_check_if_newer_than_seconds`.
+
+    Returns True if the file is fresh, False otherwise.
+    """
+    if destination.is_file():
+        age_seconds = get_age_seconds(destination)
+        if age_seconds is None:
+            raise RuntimeError(f"Error checking age of {destination}")
+        if 0 <= age_seconds < dont_check_if_newer_than_seconds:
+            logger.debug(
+                f"Using cached file {destination} of age {age_seconds}s "
+                f"without checking for updates"
+            )
+            return True
+    return False
+
+
 def _download_to_or_read_from_cache(
     url: str, *, cache: Path, dont_check_if_newer_than_seconds: float
 ) -> bytes:
@@ -93,22 +123,14 @@ def _download_to_or_read_from_cache(
     destination_etag = destination.with_suffix(".etag")
     request_headers = {"User-Agent": "conda-lock"}
     # Return the contents immediately if the file is fresh
-    if destination.is_file():
-        age_seconds = get_age_seconds(destination)
-        if age_seconds is None:
-            raise RuntimeError(f"Error checking age of {destination}")
-        if 0 <= age_seconds < dont_check_if_newer_than_seconds:
-            logger.debug(
-                f"Using cached mapping {destination} of age {age_seconds}s "
-                f"without checking for updates"
-            )
-            return destination.read_bytes()
-        # Add the ETag from the last download, if it exists, to the headers.
-        # The ETag is used to avoid downloading the file if it hasn't changed remotely.
-        # Otherwise, download the file and cache the contents and ETag.
-        if destination_etag.is_file():
-            old_etag = destination_etag.read_text().strip()
-            request_headers["If-None-Match"] = old_etag
+    if _is_cached_file_fresh(destination, dont_check_if_newer_than_seconds):
+        return destination.read_bytes()
+    # Add the ETag from the last download, if it exists, to the headers.
+    # The ETag is used to avoid downloading the file if it hasn't changed remotely.
+    # Otherwise, download the file and cache the contents and ETag.
+    if destination.is_file() and destination_etag.is_file():
+        old_etag = destination_etag.read_text().strip()
+        request_headers["If-None-Match"] = old_etag
     # Download the file and cache the result.
     logger.debug(f"Requesting {url}")
     res = requests.get(url, headers=request_headers)

tests/test_lookup_cache.py

@@ -1,6 +1,6 @@
+import multiprocessing
 import os
-import queue
-import threading
+import random
 import time
 
 from pathlib import Path
@@ -17,6 +17,44 @@
 )
 
 
+def _concurrent_download_worker(
+    url,
+    cache_root,
+    result_queue,
+    process_names_emitting_lock_warnings,
+    process_names_calling_requests_get,
+    request_count,
+):
+    """Download the file in a process and store the result in a queue."""
+
+    def mock_get(*args, **kwargs):
+        time.sleep(6)
+        response = MagicMock()
+        response.content = b"content"
+        response.status_code = 200
+        process_name = multiprocessing.current_process().name
+        process_names_calling_requests_get.append(process_name)
+        request_count.value += 1
+        return response
+
+    def mock_warning(msg, *args, **kwargs):
+        if "Failed to acquire lock" in msg:
+            process_names_emitting_lock_warnings.append(
+                multiprocessing.current_process().name
+            )
+
+    # Randomize which process calls cached_download_file first
+    time.sleep(random.uniform(0, 0.1))
+
+    with patch("conda_lock.lookup_cache.requests.get", side_effect=mock_get), patch(
+        "conda_lock.lookup_cache.logger.warning", side_effect=mock_warning
+    ):
+        result = cached_download_file(
+            url, cache_subdir_name="test_cache", cache_root=cache_root
+        )
+        result_queue.put(result)
+
+
 @pytest.fixture
 def mock_cache_dir(tmp_path):
     cache_dir = tmp_path / "cache" / "test_cache"
@@ -252,71 +290,66 @@ def wrapped_get(*args, **kwargs):
 
 
 def test_concurrent_cached_download_file(tmp_path):
-    """Test concurrent access to cached_download_file with 5 threads."""
+    """Test concurrent access to cached_download_file with 5 processes."""
     url = "https://example.com/test.json"
-    results: queue.Queue[bytes] = queue.Queue()
-    thread_names_emitting_lock_warnings: queue.Queue[str] = queue.Queue()
-    thread_names_calling_requests_get: queue.Queue[str] = queue.Queue()
 
-    def mock_get(*args, **kwargs):
-        time.sleep(6)
-        response = MagicMock()
-        response.content = b"content"
-        response.status_code = 200
-        thread_name = threading.current_thread().name
-        thread_names_calling_requests_get.put(thread_name)
-        return response
-
-    def download_file(result_queue):
-        """Download the file in a thread and store the result in a queue."""
-        import random
-
-        # Randomize which thread calls cached_download_file first
-        time.sleep(random.uniform(0, 0.1))
-        result = cached_download_file(
-            url, cache_subdir_name="test_cache", cache_root=tmp_path
-        )
-        result_queue.put(result)
-
-    with patch("requests.get", side_effect=mock_get) as mock_get, patch(
-        "conda_lock.lookup_cache.logger"
-    ) as mock_logger:
-        # Set up the logger to record which threads emit warnings
-        def mock_warning(msg, *args, **kwargs):
-            if "Failed to acquire lock" in msg:
-                thread_names_emitting_lock_warnings.put(threading.current_thread().name)
-
-        mock_logger.warning.side_effect = mock_warning
-
-        # Create and start 5 threads
-        thread_names = [f"CachedDownloadFileThread-{i}" for i in range(5)]
-        threads = [
-            threading.Thread(target=download_file, args=(results,), name=thread_name)
-            for thread_name in thread_names
+    # Use multiprocessing Manager to share state between processes
+    with multiprocessing.Manager() as manager:
+        results = manager.Queue()
+        process_names_emitting_lock_warnings = manager.list()
+        process_names_calling_requests_get = manager.list()
+        request_count = manager.Value("i", 0)
+
+        # Create and start 5 processes
+        process_names = [f"CachedDownloadFileProcess-{i}" for i in range(5)]
+        processes = [
+            multiprocessing.Process(
+                target=_concurrent_download_worker,
+                args=(
+                    url,
+                    tmp_path,
+                    results,
+                    process_names_emitting_lock_warnings,
+                    process_names_calling_requests_get,
+                    request_count,
+                ),
+                name=process_name,
+            )
+            for process_name in process_names
         ]
-        for thread in threads:
-            thread.start()
-        for thread in threads:
-            thread.join()
+        for process in processes:
+            process.start()
+        for process in processes:
+            process.join()
 
         # Collect results from the queue
-        assert results.qsize() == len(threads)
-        assert all(result == b"content" for result in results.queue)
+        assert results.qsize() == len(processes)
+        results_list = []
+        while not results.empty():
+            results_list.append(results.get())
+        assert all(result == b"content" for result in results_list)
 
-        # We expect one thread to have made the request and the other four
+        # We expect one process to have made the request and the other four
         # to have emitted warnings.
+        process_names_calling_requests_get_list = list(
+            process_names_calling_requests_get
+        )
+        process_names_emitting_lock_warnings_list = list(
+            process_names_emitting_lock_warnings
+        )
+
         assert (
-            thread_names_calling_requests_get.qsize()
+            len(process_names_calling_requests_get_list)
             == 1
-            == len(set(thread_names_calling_requests_get.queue))
-            == mock_get.call_count
-        ), f"{thread_names_calling_requests_get.queue=}"
+            == len(set(process_names_calling_requests_get_list))
+            == request_count.value
+        ), f"{process_names_calling_requests_get_list=}"
         assert (
-            thread_names_emitting_lock_warnings.qsize()
+            len(process_names_emitting_lock_warnings_list)
             == 4
-            == len(set(thread_names_emitting_lock_warnings.queue))
-        ), f"{thread_names_emitting_lock_warnings.queue=}"
-        assert set(thread_names) == set(
-            thread_names_calling_requests_get.queue
-            + thread_names_emitting_lock_warnings.queue
+            == len(set(process_names_emitting_lock_warnings_list))
+        ), f"{process_names_emitting_lock_warnings_list=}"
+        assert set(process_names) == set(
+            process_names_calling_requests_get_list
+            + process_names_emitting_lock_warnings_list
         )