Repository: SFDO-Tooling/CumulusCI
Pull request #3083 (2022-02-16 07:10:32)

Fix test by removing filesystem-related race conditioon
Fix a bug where too few records are processed due to a race condition where files are moved around while the controller process is looking at them.

Use a lock to avoid that. My previous solution to this problem had performance consequences.

Thousands of iterations of the flaky test have been run.


Patch:
diff --git a/cumulusci/tasks/bulkdata/snowfakery_utils/queue_manager.py b/cumulusci/tasks/bulkdata/snowfakery_utils/queue_manager.py
index ddd89f751..6edb6df3a 100644
--- a/cumulusci/tasks/bulkdata/snowfakery_utils/queue_manager.py
+++ b/cumulusci/tasks/bulkdata/snowfakery_utils/queue_manager.py
@@ -3,6 +3,7 @@ import shutil
 import time
 import typing as T
 from collections import defaultdict
+from multiprocessing import Lock
 from pathlib import Path
 
 import cumulusci.core.exceptions as exc
@@ -186,6 +187,7 @@ class Channel:
         self.run_until = subtask_configurator.run_until
         self.logger = logger
         self.results_reporter = results_reporter
+        self.filesystem_lock = Lock()
         self.job_counter = 0
         recipe_options = recipe_options or {}
         self._configure_queues(recipe_options)
@@ -218,7 +220,7 @@ class Channel:
             queue_size=0,
             num_workers=self.num_generator_workers,
         )
-        self.data_gen_q = WorkerQueue(data_gen_q_config)
+        self.data_gen_q = WorkerQueue(data_gen_q_config, self.filesystem_lock)
 
         load_data_q_config = WorkerQueueConfig(
             project_config=self.project_config,
@@ -234,7 +236,9 @@ class Channel:
             num_workers=self.num_loader_workers,
             rename_directory=self.data_loader_new_directory_name,
         )
-        self.load_data_q = WorkerQueue(load_data_q_config, self.results_reporter)
+        self.load_data_q = WorkerQueue(
+            load_data_q_config, self.filesystem_lock, self.results_reporter
+        )
 
         self.data_gen_q.feeds_data_to(self.load_data_q)
         return self.data_gen_q, self.load_data_q
@@ -290,31 +294,35 @@ class Channel:
         return data_dir
 
     def get_upload_status_for_channel(self):
-        def set_count_from_names(names):
-            return sum(int(name.split("_")[1]) for name in names)
-
-        return {
-            "sets_queued_to_be_generated": set_count_from_names(
-                self.data_gen_q.queued_jobs
-            ),
-            "sets_being_generated": set_count_from_names(
-                self.data_gen_q.inprogress_jobs
-            ),
-            "sets_queued_for_loading": set_count_from_names(
-                self.load_data_q.queued_jobs
-            ),
-            # note that these may count as already imported in the org
-            "sets_being_loaded": set_count_from_names(self.load_data_q.inprogress_jobs),
-            "max_num_loader_workers": self.num_loader_workers,
-            "max_num_generator_workers": self.num_generator_workers,
-            # todo: use row-level result from org load for better accuracy
-            "sets_finished": set_count_from_names(self.load_data_q.outbox_jobs),
-            "sets_failed": len(self.load_data_q.failed_jobs),
-            # TODO: are these two redundant?
-            "inprogress_generator_jobs": len(self.data_gen_q.inprogress_jobs),
-            "inprogress_loader_jobs": len(self.load_data_q.inprogress_jobs),
-            "data_gen_free_workers": self.data_gen_q.num_free_workers,
-        }
+        with self.filesystem_lock:
+
+            def set_count_from_names(names):
+                return sum(int(name.split("_")[1]) for name in names)
+
+            return {
+                "sets_queued_to_be_generated": set_count_from_names(
+                    self.data_gen_q.queued_jobs
+                ),
+                "sets_being_generated": set_count_from_names(
+                    self.data_gen_q.inprogress_jobs
+                ),
+                "sets_queued_for_loading": set_count_from_names(
+                    self.load_data_q.queued_jobs
+                ),
+                # note that these may count as already imported in the org
+                "sets_being_loaded": set_count_from_names(
+                    self.load_data_q.inprogress_jobs
+                ),
+                "max_num_loader_workers": self.num_loader_workers,
+                "max_num_generator_workers": self.num_generator_workers,
+                # todo: use row-level result from org load for better accuracy
+                "sets_finished": set_count_from_names(self.load_data_q.outbox_jobs),
+                "sets_failed": len(self.load_data_q.failed_jobs),
+                # TODO: are these two redundant?
+                "inprogress_generator_jobs": len(self.data_gen_q.inprogress_jobs),
+                "inprogress_loader_jobs": len(self.load_data_q.inprogress_jobs),
+                "data_gen_free_workers": self.data_gen_q.num_free_workers,
+            }
 
     def failure_descriptions(self) -> T.List[str]:
         """Log failures from sub-processes to main process"""
@@ -333,7 +341,18 @@ class Channel:
 
     def check_finished(self) -> bool:
         self.data_gen_q.tick()
-        still_running = len(self.data_gen_q.workers + self.load_data_q.workers) > 0
+        with self.filesystem_lock:
+            still_running = (
+                len(
+                    self.data_gen_q.workers
+                    + self.data_gen_q.queued_job_dirs
+                    + self.data_gen_q.inprogress_jobs
+                    + self.load_data_q.workers
+                    + self.load_data_q.inprogress_jobs
+                    + self.load_data_q.queued_job_dirs
+                )
+                > 0
+            )
         return not still_running
 
 
diff --git a/cumulusci/utils/parallel/task_worker_queues/parallel_worker.py b/cumulusci/utils/parallel/task_worker_queues/parallel_worker.py
index adc9815ab..7232f8808 100644
--- a/cumulusci/utils/parallel/task_worker_queues/parallel_worker.py
+++ b/cumulusci/utils/parallel/task_worker_queues/parallel_worker.py
@@ -97,10 +97,12 @@ def dotted_class_name(cls):
 class TaskWorker:
     """This class runs in a sub-thread or sub-process"""
 
-    def __init__(self, worker_dict, results_reporter):
+    def __init__(self, worker_dict, results_reporter, filesystem_lock):
         self.worker_config = WorkerConfig.from_dict(worker_dict)
         self.redirect_logging = worker_dict["redirect_logging"]
         self.results_reporter = results_reporter
+        self.filesystem_lock = filesystem_lock
+        assert filesystem_lock
 
     def __getattr__(self, name):
         """Easy access to names from the config"""
@@ -147,13 +149,15 @@ class TaskWorker:
                 self.save_exception(e)
                 self.failures_dir.mkdir(exist_ok=True)
                 logfile.close()
-                shutil.move(str(self.working_dir), str(self.failures_dir))
+                with self.filesystem_lock:
+                    shutil.move(str(self.working_dir), str(self.failures_dir))
                 self.results_reporter.put({"status": "error", "error": str(e)})
                 raise
 
         try:
-            self.outbox_dir.mkdir(exist_ok=True)
-            shutil.move(str(self.working_dir), str(self.outbox_dir))
+            with self.filesystem_lock:
+                self.outbox_dir.mkdir(exist_ok=True)
+                shutil.move(str(self.working_dir), str(self.outbox_dir))
         except BaseException as e:
             self.save_exception(e)
             raise
@@ -174,8 +178,9 @@ class TaskWorker:
             yield logger, f
 
 
-def run_task_in_worker(worker_dict: dict, results_reporter: Queue):
-    worker = TaskWorker(worker_dict, results_reporter)
+def run_task_in_worker(worker_dict: dict, results_reporter: Queue, filesystem_lock):
+    assert filesystem_lock
+    worker = TaskWorker(worker_dict, results_reporter, filesystem_lock)
     return worker.run()
 
 
@@ -188,11 +193,17 @@ class ParallelWorker:
     """Representation of the worker in the controller processs"""
 
     def __init__(
-        self, spawn_class, worker_config: WorkerConfig, results_reporter: Queue
+        self,
+        spawn_class,
+        worker_config: WorkerConfig,
+        results_reporter: Queue,
+        filesystem_lock,
     ):
         self.spawn_class = spawn_class
         self.worker_config = worker_config
         self.results_reporter = results_reporter
+        self.filesystem_lock = filesystem_lock
+        assert filesystem_lock
 
     def _validate_worker_config_is_simple(self, worker_config):
         assert json.dumps(worker_config, default=simplify)
@@ -205,7 +216,7 @@ class ParallelWorker:
         # under the covers, Python will pass this as Pickles.
         self.process = self.spawn_class(
             target=run_task_in_worker,
-            args=[dct, self.results_reporter],
+            args=[dct, self.results_reporter, self.filesystem_lock],
             # quit if the parent process decides to exit (e.g. after a timeout)
             daemon=True,
         )
diff --git a/cumulusci/utils/parallel/task_worker_queues/parallel_worker_queue.py b/cumulusci/utils/parallel/task_worker_queues/parallel_worker_queue.py
index 4c27c8596..ed4e60938 100644
--- a/cumulusci/utils/parallel/task_worker_queues/parallel_worker_queue.py
+++ b/cumulusci/utils/parallel/task_worker_queues/parallel_worker_queue.py
@@ -64,6 +64,7 @@ class WorkerQueue:
     def __init__(
         self,
         queue_config: WorkerQueueConfig,
+        filesystem_lock,
         results_reporter: Queue = None,
     ):
         self.config = queue_config
@@ -71,6 +72,7 @@ class WorkerQueue:
         self._create_dirs()
         self.workers = []
         self.results_reporter = results_reporter or self.context.Queue()
+        self.filesystem_lock = filesystem_lock
 
     def __getattr__(self, name):
         """Convenience proxy for config values
@@ -216,7 +218,10 @@ class WorkerQueue:
         )
 
         worker = ParallelWorker(
-            self.config.spawn_class, worker_config, self.results_reporter
+            self.config.spawn_class,
+            worker_config,
+            self.results_reporter,
+            self.filesystem_lock,
         )
         worker.start()
         self.workers.append(worker)


Test patch:
diff --git a/cumulusci/tasks/bulkdata/tests/test_snowfakery.py b/cumulusci/tasks/bulkdata/tests/test_snowfakery.py
index cee13aa76..25eb871b3 100644
--- a/cumulusci/tasks/bulkdata/tests/test_snowfakery.py
+++ b/cumulusci/tasks/bulkdata/tests/test_snowfakery.py
@@ -655,8 +655,9 @@ class TestSnowfakery:
         ]
 
         unique_values = [row.value for batchrows in all_rows for row in batchrows]
-        assert len(unique_values) == len(set(unique_values))
-        # See also W-10142031: Investigate unreliable test assertions
+        assert len(mock_load_data.mock_calls) == 6, len(mock_load_data.mock_calls)
+        assert len(unique_values) == 30, len(unique_values)
+        assert len(set(unique_values)) == 30, unique_values
 
     @mock.patch("cumulusci.tasks.bulkdata.snowfakery.MIN_PORTION_SIZE", 2)
     def test_two_channels(self, mock_load_data, create_task):
diff --git a/cumulusci/utils/parallel/task_worker_queues/tests/test_parallel_worker.py b/cumulusci/utils/parallel/task_worker_queues/tests/test_parallel_worker.py
index a5b522224..c33943b12 100644
--- a/cumulusci/utils/parallel/task_worker_queues/tests/test_parallel_worker.py
+++ b/cumulusci/utils/parallel/task_worker_queues/tests/test_parallel_worker.py
@@ -1,5 +1,6 @@
 from contextlib import contextmanager
 from logging import getLogger
+from multiprocessing import Lock
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from unittest import mock
@@ -63,7 +64,7 @@ class TestWorkerQueue:
             parent_dir=Path(parent_dir),
             **kwargs,
         )
-        q = WorkerQueue(config)
+        q = WorkerQueue(config, filesystem_lock=Lock())
 
         yield q
 
@@ -307,7 +308,7 @@ class TestParallelWorker:
                 outbox_dir=outbox_dir,
                 working_dir=working_dir,
             )
-            worker = ParallelWorker(DelaySpawner, config, None)
+            worker = ParallelWorker(DelaySpawner, config, None, Lock())
             worker.start()
             worker.terminate()
             assert "Alive: False" in repr(worker)
@@ -327,7 +328,7 @@ class TestTaskWorker:
                 outbox_dir=outbox_dir,
                 working_dir=working_dir,
             )
-            p = TaskWorker(config.as_dict(), None)
+            p = TaskWorker(config.as_dict(), None, Lock())
             with mock.patch("shutil.move", side_effect=AssertionError):
                 with pytest.raises(AssertionError):
                     p.run()