Remove tracker and tracker_worker

src/ert/cli/main.py

@@ -1,7 +1,10 @@
 #!/usr/bin/env python
+from __future__ import annotations
+
 import contextlib
 import logging
 import os
+import queue
 import sys
 from typing import Any, TextIO
 
@@ -19,8 +22,9 @@
 from ert.cli.workflow import execute_workflow
 from ert.config import ErtConfig, QueueSystem
 from ert.enkf_main import EnKFMain
-from ert.ensemble_evaluator import EvaluatorServerConfig, EvaluatorTracker
+from ert.ensemble_evaluator import EvaluatorServerConfig
 from ert.namespace import Namespace
+from ert.run_models.base_run_model import StatusEvents
 from ert.storage import open_storage
 from ert.storage.local_storage import local_storage_set_ert_config
 
@@ -76,11 +80,13 @@ def run_cli(args: Namespace, _: Any = None) -> None:
         execute_workflow(ert, storage, args.name)
         return
 
+    status_queue: queue.SimpleQueue[StatusEvents] = queue.SimpleQueue()
     try:
         model = create_model(
             ert_config,
             storage,
             args,
+            status_queue,
         )
     except ValueError as e:
         raise ErtCliError(e) from e
@@ -106,11 +112,6 @@ def run_cli(args: Namespace, _: Any = None) -> None:
         target=model.start_simulations_thread,
         args=(evaluator_server_config,),
     )
-    thread.start()
-
-    tracker = EvaluatorTracker(
-        model, ee_con_info=evaluator_server_config.get_connection_info()
-    )
 
     with contextlib.ExitStack() as exit_stack:
         out: TextIO
@@ -121,13 +122,12 @@ def run_cli(args: Namespace, _: Any = None) -> None:
         else:
             out = sys.stderr
         monitor = Monitor(out=out, color_always=args.color_always)
-
+        thread.start()
         try:
-            monitor.monitor(tracker.track())
+            monitor.monitor(status_queue)
         except (SystemExit, KeyboardInterrupt, OSError):
-            # _base_service.py translates CTRL-c to OSError
             print("\nKilling simulations...")
-            tracker.request_termination()
+            model.cancel()
 
     thread.join()
     storage.close()

src/ert/cli/model_factory.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+from queue import SimpleQueue
 from typing import TYPE_CHECKING, Tuple
 
 import numpy as np
@@ -38,13 +39,15 @@
     import numpy.typing as npt
 
     from ert.namespace import Namespace
+    from ert.run_models.base_run_model import StatusEvents
     from ert.storage import Storage
 
 
 def create_model(
     config: ErtConfig,
     storage: Storage,
     args: Namespace,
+    status_queue: SimpleQueue[StatusEvents],
 ) -> BaseRunModel:
     logger = logging.getLogger(__name__)
     logger.info(
@@ -57,26 +60,33 @@ def create_model(
     update_settings = config.analysis_config.observation_settings
 
     if args.mode == TEST_RUN_MODE:
-        return _setup_single_test_run(config, storage, args)
+        return _setup_single_test_run(config, storage, args, status_queue)
     elif args.mode == ENSEMBLE_EXPERIMENT_MODE:
-        return _setup_ensemble_experiment(config, storage, args)
+        return _setup_ensemble_experiment(config, storage, args, status_queue)
     elif args.mode == EVALUATE_ENSEMBLE_MODE:
-        return _setup_evaluate_ensemble(config, storage, args)
+        return _setup_evaluate_ensemble(config, storage, args, status_queue)
     elif args.mode == ENSEMBLE_SMOOTHER_MODE:
-        return _setup_ensemble_smoother(config, storage, args, update_settings)
+        return _setup_ensemble_smoother(
+            config, storage, args, update_settings, status_queue
+        )
     elif args.mode == ES_MDA_MODE:
-        return _setup_multiple_data_assimilation(config, storage, args, update_settings)
+        return _setup_multiple_data_assimilation(
+            config, storage, args, update_settings, status_queue
+        )
     elif args.mode == ITERATIVE_ENSEMBLE_SMOOTHER_MODE:
         return _setup_iterative_ensemble_smoother(
-            config, storage, args, update_settings
+            config, storage, args, update_settings, status_queue
         )
 
     else:
         raise NotImplementedError(f"Run type not supported {args.mode}")
 
 
 def _setup_single_test_run(
-    config: ErtConfig, storage: Storage, args: Namespace
+    config: ErtConfig,
+    storage: Storage,
+    args: Namespace,
+    status_queue: SimpleQueue[StatusEvents],
 ) -> SingleTestRun:
     return SingleTestRun(
         SingleTestRunArguments(
@@ -89,11 +99,15 @@ def _setup_single_test_run(
         ),
         config,
         storage,
+        status_queue,
     )
 
 
 def _setup_ensemble_experiment(
-    config: ErtConfig, storage: Storage, args: Namespace
+    config: ErtConfig,
+    storage: Storage,
+    args: Namespace,
+    status_queue: SimpleQueue[StatusEvents],
 ) -> EnsembleExperiment:
     min_realizations_count = config.analysis_config.minimum_required_realizations
     active_realizations = _realizations(args, config.model_config.num_realizations)
@@ -124,11 +138,15 @@ def _setup_ensemble_experiment(
         config,
         storage,
         config.queue_config,
+        status_queue=status_queue,
     )
 
 
 def _setup_evaluate_ensemble(
-    config: ErtConfig, storage: Storage, args: Namespace
+    config: ErtConfig,
+    storage: Storage,
+    args: Namespace,
+    status_queue: SimpleQueue[StatusEvents],
 ) -> EvaluateEnsemble:
     min_realizations_count = config.analysis_config.minimum_required_realizations
     active_realizations = _realizations(args, config.model_config.num_realizations)
@@ -154,6 +172,7 @@ def _setup_evaluate_ensemble(
         config,
         storage,
         config.queue_config,
+        status_queue=status_queue,
     )
 
 
@@ -162,6 +181,7 @@ def _setup_ensemble_smoother(
     storage: Storage,
     args: Namespace,
     update_settings: UpdateSettings,
+    status_queue: SimpleQueue[StatusEvents],
 ) -> EnsembleSmoother:
     return EnsembleSmoother(
         ESRunArguments(
@@ -181,6 +201,7 @@ def _setup_ensemble_smoother(
         config.queue_config,
         es_settings=config.analysis_config.es_module,
         update_settings=update_settings,
+        status_queue=status_queue,
     )
 
 
@@ -208,6 +229,7 @@ def _setup_multiple_data_assimilation(
     storage: Storage,
     args: Namespace,
     update_settings: UpdateSettings,
+    status_queue: SimpleQueue[StatusEvents],
 ) -> MultipleDataAssimilation:
     restart_run, prior_ensemble = _determine_restart_info(args)
 
@@ -231,6 +253,7 @@ def _setup_multiple_data_assimilation(
         config.queue_config,
         es_settings=config.analysis_config.es_module,
         update_settings=update_settings,
+        status_queue=status_queue,
     )
 
 
@@ -239,6 +262,7 @@ def _setup_iterative_ensemble_smoother(
     storage: Storage,
     args: Namespace,
     update_settings: UpdateSettings,
+    status_queue: SimpleQueue[StatusEvents],
 ) -> IteratedEnsembleSmoother:
     return IteratedEnsembleSmoother(
         SIESRunArguments(
@@ -260,6 +284,7 @@ def _setup_iterative_ensemble_smoother(
         config.queue_config,
         config.analysis_config.ies_module,
         update_settings=update_settings,
+        status_queue=status_queue,
     )
 
 

src/ert/cli/monitor.py

@@ -1,7 +1,10 @@
 # -*- coding: utf-8 -*-
+from __future__ import annotations
+
 import sys
 from datetime import datetime, timedelta
-from typing import Dict, Iterator, Optional, TextIO, Tuple, Union
+from queue import SimpleQueue
+from typing import Dict, Optional, TextIO, Tuple
 
 from tqdm import tqdm
 
@@ -18,6 +21,7 @@
     FORWARD_MODEL_STATE_FAILURE,
     REAL_STATE_TO_COLOR,
 )
+from ert.run_models.base_run_model import StatusEvents
 from ert.shared.status.utils import format_running_time
 
 Color = Tuple[int, int, int]
@@ -57,13 +61,15 @@ def __init__(self, out: TextIO = sys.stdout, color_always: bool = False) -> None
 
             # The dot adds no value without color, so remove it.
             self.dot = ""
+        self.done = False
 
     def monitor(
         self,
-        events: Iterator[Union[FullSnapshotEvent, SnapshotUpdateEvent, EndEvent]],
+        event_queue: SimpleQueue[StatusEvents],
     ) -> None:
         self._start_time = datetime.now()
-        for event in events:
+        while True:
+            event = event_queue.get()
             if isinstance(event, FullSnapshotEvent):
                 if event.snapshot is not None:
                     self._snapshots[event.iteration] = event.snapshot

src/ert/ensemble_evaluator/__init__.py

@@ -7,7 +7,6 @@
 )
 from .config import EvaluatorServerConfig
 from .evaluator import EnsembleEvaluator
-from .evaluator_tracker import EvaluatorTracker
 from .event import EndEvent, FullSnapshotEvent, SnapshotUpdateEvent
 from .monitor import Monitor
 from .snapshot import PartialSnapshot, Snapshot
@@ -18,7 +17,6 @@
     "EnsembleBuilder",
     "EnsembleEvaluator",
     "EvaluatorServerConfig",
-    "EvaluatorTracker",
     "ForwardModel",
     "FullSnapshotEvent",
     "Monitor",

src/ert/ensemble_evaluator/evaluator.py

@@ -391,7 +391,7 @@ def _run_server(self, loop: asyncio.AbstractEventLoop) -> None:
         loop.run_until_complete(self.evaluator_server())
         logger.debug("Server thread exiting.")
 
-    def _start_running(self) -> None:
+    def start_running(self) -> None:
         self._ws_thread.start()
         self._ensemble.evaluate(self._config)
 
@@ -419,11 +419,11 @@ def _signal_cancel(self) -> None:
             logger.debug("Stopping current ensemble")
             self._loop.call_soon_threadsafe(self._stop)
 
-    def run_and_get_successful_realizations(self) -> List[int]:
-        self._start_running()
-        logger.debug("Started evaluator, joining until shutdown")
+    def join(self) -> None:
         self._ws_thread.join()
         logger.debug("Evaluator is done")
+
+    def get_successful_realizations(self) -> List[int]:
         return self._ensemble.get_successful_realizations()
 
     @staticmethod