Merge pull request #84 from franchuterivera/refactor_development_loggermsg

Reduce Deadlock Probability

Author: ravinkohli

.github/workflows/pytest.yml

@@ -29,7 +29,7 @@ jobs:
     - name: Run tests
       run: |
         if [ ${{ matrix.code-cov }} ]; then codecov='--cov=autoPyTorch --cov-report=xml'; fi
-        python -m pytest --durations=20 --timeout=500 --timeout-method=thread -v $codecov test
+        python -m pytest --durations=20 --timeout=300 --timeout-method=thread -v $codecov test
     - name: Check for files left behind by test
       if: ${{ always() }}
       run: |

autoPyTorch/api/base_task.py

@@ -4,6 +4,7 @@
 import math
 import multiprocessing
 import os
+import platform
 import sys
 import tempfile
 import time
@@ -186,6 +187,8 @@ def __init__(
 
         self.stop_logging_server = None  # type: Optional[multiprocessing.synchronize.Event]
 
+        self._dask_client = None
+
         self.search_space_updates = search_space_updates
         if search_space_updates is not None:
             if not isinstance(self.search_space_updates,
@@ -504,7 +507,7 @@ def _do_dummy_prediction(self, num_run: int) -> None:
             backend=self._backend,
             seed=self.seed,
             metric=self._metric,
-            logger=self._logger,
+            logger_port=self._logger_port,
             cost_for_crash=get_cost_of_crash(self._metric),
             abort_on_first_run_crash=False,
             initial_num_run=num_run,
@@ -550,8 +553,11 @@ def _do_dummy_prediction(self, num_run: int) -> None:
 
     def _do_traditional_prediction(self, num_run: int, time_for_traditional: int) -> int:
 
+        # Mypy Checkings -- Traditional prediction is only called for search
+        # where the following objects are created
         assert self._metric is not None
         assert self._logger is not None
+        assert self._dask_client is not None
 
         self._logger.info("Starting to create dummy predictions.")
 
@@ -573,7 +579,7 @@ def _do_traditional_prediction(self, num_run: int, time_for_traditional: int) ->
                 backend=self._backend,
                 seed=self.seed,
                 metric=self._metric,
-                logger=self._logger,
+                logger_port=self._logger_port,
                 cost_for_crash=get_cost_of_crash(self._metric),
                 abort_on_first_run_crash=False,
                 initial_num_run=num_run,
@@ -720,6 +726,9 @@ def _search(
 
         self._backend.save_datamanager(dataset)
 
+        # Print debug information to log
+        self._print_debug_info_to_log()
+
         self._metric = get_metrics(
             names=[optimize_metric], dataset_properties=dataset_properties)[0]
 
@@ -737,7 +746,14 @@ def _search(
         if self.task_type is None:
             raise ValueError("Cannot interpret task type from the dataset")
 
-        self._create_dask_client()
+        # If no dask client was provided, we create one, so that we can
+        # start a ensemble process in parallel to smbo optimize
+        if (
+            self._dask_client is None and (self.ensemble_size > 0 or self.n_jobs is not None and self.n_jobs > 1)
+        ):
+            self._create_dask_client()
+        else:
+            self._is_dask_client_internally_created = False
 
         # ============> Run dummy predictions
         num_run = 1
@@ -794,7 +810,7 @@ def _search(
                 ensemble_memory_limit=self._memory_limit,
                 random_state=self.seed,
                 precision=precision,
-                logger_port=self._logger_port
+                logger_port=self._logger_port,
             )
             self._stopwatch.stop_task(ensemble_task_name)
 
@@ -854,19 +870,21 @@ def _search(
         if proc_ensemble is not None:
             self.ensemble_performance_history = list(proc_ensemble.history)
 
+            if len(proc_ensemble.futures) > 0:
+                # Also add ensemble runs that did not finish within smac time
+                # and add them into the ensemble history
+                self._logger.info("Ensemble script still running, waiting for it to finish.")
+                result = proc_ensemble.futures.pop().result()
+                if result:
+                    ensemble_history, _, _, _ = result
+                    self.ensemble_performance_history.extend(ensemble_history)
+                self._logger.info("Ensemble script finished, continue shutdown.")
+
             # save the ensemble performance history file
             if len(self.ensemble_performance_history) > 0:
                 pd.DataFrame(self.ensemble_performance_history).to_json(
                     os.path.join(self._backend.internals_directory, 'ensemble_history.json'))
 
-            if len(proc_ensemble.futures) > 0:
-                future = proc_ensemble.futures.pop()
-                # Now we need to wait for the future to return as it cannot be cancelled while it
-                # is running: https://stackoverflow.com/a/49203129
-                self._logger.info("Ensemble script still running, waiting for it to finish.")
-                future.result()
-                self._logger.info("Ensemble script finished, continue shutdown.")
-
         self._logger.info("Closing the dask infrastructure")
         self._close_dask_client()
         self._logger.info("Finished closing the dask infrastructure")
@@ -1123,3 +1141,32 @@ def get_incumbent_config(
             self
     ):
         pass
+
+    def get_models_with_weights(self) -> List:
+        if self.models_ is None or len(self.models_) == 0 or \
+                self.ensemble_ is None:
+            self._load_models()
+
+        assert self.ensemble_ is not None
+        return self.ensemble_.get_models_with_weights(self.models_)
+
+    def show_models(self) -> str:
+        df = []
+        for weight, model in self.get_models_with_weights():
+            representation = model.get_pipeline_representation()
+            representation.update({'Weight': weight})
+            df.append(representation)
+        return pd.DataFrame(df).to_markdown()
+
+    def _print_debug_info_to_log(self) -> None:
+        """
+        Prints to the log file debug information about the current estimator
+        """
+        assert self._logger is not None
+        self._logger.debug("Starting to print environment information")
+        self._logger.debug('  Python version: %s', sys.version.split('\n'))
+        self._logger.debug('  System: %s', platform.system())
+        self._logger.debug('  Machine: %s', platform.machine())
+        self._logger.debug('  Platform: %s', platform.platform())
+        for key, value in vars(self).items():
+            self._logger.debug(f"\t{key}->{value}")

autoPyTorch/configs/default_pipeline_options.json

@@ -6,7 +6,7 @@
             "runtime": 3600,
             "torch_num_threads": 1,
             "early_stopping": 20,
-            "use_tensorboard_logger": "True",
+            "use_tensorboard_logger": "False",
             "use_pynisher": "False",
             "metrics_during_training": "True"
 }

autoPyTorch/ensemble/abstract_ensemble.py

@@ -1,5 +1,5 @@
 from abc import ABCMeta, abstractmethod
-from typing import List, Tuple, Union
+from typing import Any, Dict, List, Tuple, Union
 
 import numpy as np
 
@@ -49,7 +49,7 @@ def predict(self, base_models_predictions: Union[np.ndarray, List[np.ndarray]])
         self
 
     @abstractmethod
-    def get_models_with_weights(self, models: BasePipeline) -> List[Tuple[float, BasePipeline]]:
+    def get_models_with_weights(self, models: Dict[Any, BasePipeline]) -> List[Tuple[float, BasePipeline]]:
         """Return a list of (weight, model) pairs
 
         Args:

autoPyTorch/ensemble/ensemble_builder.py

@@ -4,6 +4,7 @@
 import logging
 import logging.handlers
 import math
+import multiprocessing
 import numbers
 import os
 import pickle
@@ -157,7 +158,12 @@ def __call__(
     ) -> None:
         self.build_ensemble(smbo.tae_runner.client)
 
-    def build_ensemble(self, dask_client: dask.distributed.Client) -> None:
+    def build_ensemble(
+        self,
+        dask_client: dask.distributed.Client,
+        pynisher_context: str = 'spawn',
+        unit_test: bool = False
+    ) -> None:
 
         # The second criteria is elapsed time
         elapsed_time = time.time() - self.start_time
@@ -227,11 +233,13 @@ def build_ensemble(self, dask_client: dask.distributed.Client) -> None:
                     memory_limit=self.ensemble_memory_limit,
                     read_at_most=self.read_at_most,
                     random_state=self.seed,
-                    logger_port=self.logger_port,
                     end_at=self.start_time + self.time_left_for_ensembles,
                     iteration=self.iteration,
                     return_predictions=False,
                     priority=100,
+                    pynisher_context=pynisher_context,
+                    logger_port=self.logger_port,
+                    unit_test=unit_test,
                 ))
 
                 logger.info(
@@ -267,10 +275,12 @@ def fit_and_return_ensemble(
     memory_limit: Optional[int],
     read_at_most: int,
     random_state: int,
-    logger_port: int,
     end_at: float,
     iteration: int,
     return_predictions: bool,
+    pynisher_context: str,
+    logger_port: int = logging.handlers.DEFAULT_TCP_LOGGING_PORT,
+    unit_test: bool = False,
 ) -> Tuple[
         List[Dict[str, float]],
         int,
@@ -317,13 +327,20 @@ def fit_and_return_ensemble(
             memory limit in mb. If ``None``, no memory limit is enforced.
         read_at_most: int
             read at most n new prediction files in each iteration
-        logger_port: int
-            port in localhost where to publish msg
         end_at: float
             At what time the job must finish. Needs to be the endtime and not the time left
             because we do not know when dask schedules the job.
         iteration: int
             The current iteration
+        pynisher_context: str
+            Context to use for multiprocessing, can be either fork, spawn or forkserver.
+        logger_port: int
+            The port where the logging server is listening to.
+        unit_test: bool
+            Turn on unit testing mode. This currently makes fit_ensemble raise a MemoryError.
+            Having this is very bad coding style, but I did not find a way to make
+            unittest.mock work through the pynisher with all spawn contexts. If you know a
+            better solution, please let us know by opening an issue.
     Returns
     -------
         List[Tuple[int, float, float, float]]
@@ -346,33 +363,36 @@ def fit_and_return_ensemble(
         read_at_most=read_at_most,
         random_state=random_state,
         logger_port=logger_port,
+        unit_test=unit_test,
     ).run(
         end_at=end_at,
         iteration=iteration,
         return_predictions=return_predictions,
+        pynisher_context=pynisher_context,
     )
     return result
 
 
 class EnsembleBuilder(object):
     def __init__(
-            self,
-            backend: Backend,
-            dataset_name: str,
-            task_type: int,
-            output_type: int,
-            metrics: List[autoPyTorchMetric],
-            opt_metric: str,
-            ensemble_size: int = 10,
-            ensemble_nbest: int = 100,
-            max_models_on_disc: Union[float, int] = 100,
-            performance_range_threshold: float = 0,
-            seed: int = 1,
-            precision: int = 32,
-            memory_limit: Optional[int] = 1024,
-            read_at_most: int = 5,
-            random_state: Optional[Union[int, np.random.RandomState]] = None,
-            logger_port: int = logging.handlers.DEFAULT_TCP_LOGGING_PORT,
+        self,
+        backend: Backend,
+        dataset_name: str,
+        task_type: int,
+        output_type: int,
+        metrics: List[autoPyTorchMetric],
+        opt_metric: str,
+        ensemble_size: int = 10,
+        ensemble_nbest: int = 100,
+        max_models_on_disc: Union[float, int] = 100,
+        performance_range_threshold: float = 0,
+        seed: int = 1,
+        precision: int = 32,
+        memory_limit: Optional[int] = 1024,
+        read_at_most: int = 5,
+        random_state: Optional[Union[int, np.random.RandomState]] = None,
+        logger_port: int = logging.handlers.DEFAULT_TCP_LOGGING_PORT,
+        unit_test: bool = False,
     ):
         """
             Constructor
@@ -420,7 +440,12 @@ def __init__(
             read_at_most: int
                 read at most n new prediction files in each iteration
             logger_port: int
-                port where to publish messages
+                port that receives logging records
+            unit_test: bool
+                Turn on unit testing mode. This currently makes fit_ensemble raise a MemoryError.
+                Having this is very bad coding style, but I did not find a way to make
+                unittest.mock work through the pynisher with all spawn contexts. If you know a
+                better solution, please let us know by opening an issue.
         """
 
         super(EnsembleBuilder, self).__init__()
@@ -461,6 +486,7 @@ def __init__(
         self.memory_limit = memory_limit
         self.read_at_most = read_at_most
         self.random_state = check_random_state(random_state)
+        self.unit_test = unit_test
 
         # Setup the logger
         self.logger_port = logger_port
@@ -564,6 +590,7 @@ def run(
         end_at: Optional[float] = None,
         time_buffer: int = 5,
         return_predictions: bool = False,
+        pynisher_context: str = 'spawn',  # only change for unit testing!
     ) -> Tuple[
         List[Dict[str, float]],
         int,
@@ -625,12 +652,16 @@ def run(
             else:
                 raise NotImplementedError()
 
-            if time_left - time_buffer < 1:
+            wall_time_in_s = int(time_left - time_buffer)
+            if wall_time_in_s < 1:
                 break
+            context = multiprocessing.get_context(pynisher_context)
+
             safe_ensemble_script = pynisher.enforce_limits(
-                wall_time_in_s=int(time_left - time_buffer),
+                wall_time_in_s=wall_time_in_s,
                 mem_in_mb=self.memory_limit,
-                logger=self.logger
+                logger=self.logger,
+                context=context,
             )(self.main)
             safe_ensemble_script(time_left, iteration, return_predictions)
             if safe_ensemble_script.exit_status is pynisher.MemorylimitException:
@@ -1216,6 +1247,10 @@ def fit_ensemble(self, selected_keys: List[str]) -> Optional[EnsembleSelection]:
             ensemble: EnsembleSelection
                 trained Ensemble
         """
+
+        if self.unit_test:
+            raise MemoryError()
+
         predictions_train = [self.read_preds[k][Y_ENSEMBLE] for k in selected_keys]
         include_num_runs = [
             (

autoPyTorch/ensemble/ensemble_selection.py

@@ -189,7 +189,7 @@ def __str__(self) -> str:
 
     def get_models_with_weights(
         self,
-        models: BasePipeline
+        models: Dict[Any, BasePipeline]
     ) -> List[Tuple[float, BasePipeline]]:
         output = []
         for i, weight in enumerate(self.weights_):

autoPyTorch/ensemble/singlebest_ensemble.py

@@ -1,5 +1,5 @@
 import os
-from typing import List, Tuple, Union
+from typing import Any, Dict, List, Tuple, Union
 
 import numpy as np
 
@@ -97,7 +97,7 @@ def __str__(self) -> str:
                           enumerate(self.identifiers_)
                           if self.weights_[idx] > 0]))
 
-    def get_models_with_weights(self, models: BasePipeline
+    def get_models_with_weights(self, models: Dict[Any, BasePipeline]
                                 ) -> List[Tuple[float, BasePipeline]]:
         output = []
         for i, weight in enumerate(self.weights_):