feat: Remove global (cross-worker) stopping criteria (#252)

Author: nastaran78

docs/reference/evaluate_pipeline.md

@@ -132,7 +132,7 @@ neps.save_pipeline_results(
 
 ### 3.4 Common pitfalls
 
-* When using async approach, one worker, may create as many trials as possible, of course that in `Slurm` or other workload managers it's impossible to overload the system because of limitations set for each user, but if you want to control resources used for optimization, it's crucial to set `max_evaluations_per_run` when calling `neps.run`.
+* When using async approach, one worker, may create as many trials as possible, of course that in `Slurm` or other workload managers it's impossible to overload the system because of limitations set for each user, but if you want to control resources used for optimization, it's crucial to set `evaluations_to_spend` when calling `neps.run`.
 
 ## 4  Extra injected arguments
 

docs/reference/neps_run.md

@@ -167,8 +167,7 @@ Any new workers that come online will automatically pick up work and work togeth
         evaluate_pipeline=...,
         pipeline_space=...,
         root_directory="some/path",
-        evaluations_to_spend=100,
-        max_evaluations_per_run=10, # (1)!
+        evaluations_to_spend=100, # (1)!
         continue_until_max_evaluation_completed=True, # (2)!
         overwrite_root_directory=False, #!!!
     )
@@ -220,7 +219,7 @@ neps.run(
 
 !!! note
 
-    Any runs that error will still count towards the total `evaluations_to_spend` or `max_evaluations_per_run`.
+    Any runs that error will still count towards the total `evaluations_to_spend`.
 
 ### Re-running Failed Configurations
 

neps/api.py

@@ -28,7 +28,7 @@
 logger = logging.getLogger(__name__)
 
 
-def run(  # noqa: C901, D417, PLR0913
+def run(  # noqa: D417, PLR0913
     evaluate_pipeline: Callable[..., EvaluatePipelineReturn] | str,
     pipeline_space: (
         Mapping[str, dict | str | int | float | Parameter]
@@ -39,11 +39,9 @@ def run(  # noqa: C901, D417, PLR0913
     root_directory: str | Path = "neps_results",
     overwrite_root_directory: bool = False,
     evaluations_to_spend: int | None = None,
-    max_evaluations_total: int | None = None,
     max_evaluations_per_run: int | None = None,
     continue_until_max_evaluation_completed: bool = False,
     cost_to_spend: int | float | None = None,
-    max_cost_total: int | float | None = None,
     fidelities_to_spend: int | float | None = None,
     ignore_errors: bool = False,
     objective_value_on_error: float | None = None,
@@ -104,7 +102,6 @@ def evaluate_pipeline(some_parameter: float) -> float:
         },
         root_directory="usage_example",
         evaluations_to_spend=5,
-        max_evaluations_per_run=10,
     )
     ```
 
@@ -200,18 +197,15 @@ def evaluate_pipeline(some_parameter: float) -> float:
         overwrite_root_directory: If true, delete the working directory at the start of
             the run. This is, e.g., useful when debugging a evaluate_pipeline function.
 
-        max_evaluations_per_run: Number of evaluations this specific call should do.
+        evaluations_to_spend: Number of evaluations this specific call/worker should do.
         ??? note "Limitation on Async mode"
             Currently, there is no specific number to control number of parallel evaluations running with
             the same worker, so in case you want to limit the number of parallel evaluations,
-            it's crucial to limit the number of evaluations per run.
-
-        evaluations_to_spend: Number of evaluations after which to terminate.
-            This is shared between all workers operating in the same `root_directory`.
+            it's crucial to limit the `evaluations_to_spend` accordingly.
 
         continue_until_max_evaluation_completed:
-            If true, only stop after evaluations_to_spend have been completed.
-            This is only relevant in the parallel setting.
+            If true, stop only after evaluations_to_spend have fully completed. In other words,
+            pipelines that are still running do not count toward the stopping criterion.
 
         cost_to_spend: No new evaluations will start when this cost is exceeded. Requires
             returning a cost in the evaluate_pipeline function, e.g.,
@@ -423,54 +417,24 @@ def __call__(
                 runtime to run your optimizer.
 
     """  # noqa: E501
-    if (
-        evaluations_to_spend is None
-        and max_evaluations_total is None
-        and max_evaluations_per_run is None
-        and cost_to_spend is None
-        and max_cost_total is None
-        and fidelities_to_spend is None
-    ):
-        warnings.warn(
-            "None of the following were set, this will run idefinitely until the worker"
-            " process is stopped."
-            f"\n * {evaluations_to_spend=}"
-            f"\n * {max_evaluations_per_run=}"
-            f"\n * {cost_to_spend=}"
-            f"\n * {fidelities_to_spend}",
-            UserWarning,
-            stacklevel=2,
-        )
-
-    if max_evaluations_total is not None:
-        warnings.warn(
-            "`max_evaluations_total` is deprecated and will be removed in"
-            " a future release. Please use `evaluations_to_spend` instead.",
-            DeprecationWarning,
-            stacklevel=2,
-        )
-        evaluations_to_spend = max_evaluations_total
-
-    if max_cost_total is not None:
-        warnings.warn(
-            "`max_cost_total` is deprecated and will be removed in a future release. "
-            "Please use `cost_to_spend` instead.",
-            DeprecationWarning,
-            stacklevel=2,
+    if max_evaluations_per_run is not None:
+        raise ValueError(
+            "`max_evaluations_per_run` is deprecated, please use "
+            "`evaluations_to_spend` for limiting the number of evaluations for this run.",
         )
-        cost_to_spend = max_cost_total
 
-    criteria = {
+    controling_params = {
         "evaluations_to_spend": evaluations_to_spend,
-        "max_evaluations_per_run": max_evaluations_per_run,
         "cost_to_spend": cost_to_spend,
         "fidelities_to_spend": fidelities_to_spend,
     }
-    set_criteria = [k for k, v in criteria.items() if v is not None]
-    if len(set_criteria) > 1:
-        raise ValueError(
-            f"Multiple stopping criteria specified: {', '.join(set_criteria)}. "
-            "Only one is allowed."
+    if all(x is None for x in controling_params.values()):
+        warnings.warn(
+            "None of the following were set, this will run idefinitely until the worker"
+            " process is stopped."
+            f"{', '.join(list(controling_params.keys()))}.",
+            UserWarning,
+            stacklevel=2,
         )
 
     logger.info(f"Starting neps.run using root directory {root_directory}")
@@ -491,13 +455,7 @@ def __call__(
 
     is_multi_fidelity = _optimizer_info["name"] in multi_fidelity_optimizers
 
-    if is_multi_fidelity:
-        if evaluations_to_spend is not None:
-            raise ValueError(
-                "`evaluations_to_spend` is not allowed for multi-fidelity optimizers. "
-                "Only `fidelities_to_spend` or `cost_to_spend`"
-            )
-    elif fidelities_to_spend is not None:
+    if not is_multi_fidelity and fidelities_to_spend is not None:
         raise ValueError(
             "`fidelities_to_spend` is not allowed for non-multi-fidelity optimizers."
         )
@@ -527,7 +485,6 @@ def __call__(
         fidelities_to_spend=fidelities_to_spend,
         optimization_dir=Path(root_directory),
         evaluations_to_spend=evaluations_to_spend,
-        max_evaluations_for_worker=max_evaluations_per_run,
         continue_until_max_evaluation_completed=continue_until_max_evaluation_completed,
         objective_value_on_error=objective_value_on_error,
         cost_value_on_error=cost_value_on_error,

neps/runtime.py

@@ -181,15 +181,6 @@ class DefaultWorker:
     worker_id: str
     """The id of the worker."""
 
-    worker_cumulative_eval_count: int = 0
-    """The number of evaluations done by this worker."""
-
-    worker_cumulative_eval_cost: float = 0.0
-    """The cost of the evaluations done by this worker."""
-
-    worker_cumulative_evaluation_time_seconds: float = 0.0
-    """The time spent evaluating configurations by this worker."""
-
     _GRACE: ClassVar = FS_SYNC_GRACE_BASE
 
     @classmethod
@@ -251,46 +242,13 @@ def _check_worker_local_settings(
                 raise WorkerRaiseError(msg) from error_from_this_worker
             return msg
 
-        if (
-            self.settings.max_evaluations_for_worker is not None
-            and self.worker_cumulative_eval_count
-            >= self.settings.max_evaluations_for_worker
-        ):
-            return (
-                "Worker has reached the maximum number of evaluations it is allowed to do"
-                f" as given by `{self.settings.max_evaluations_for_worker=}`."
-                "\nTo allow more evaluations, increase this value or use a different"
-                " stopping criterion."
-            )
-
-        if (
-            self.settings.max_cost_for_worker is not None
-            and self.worker_cumulative_eval_cost >= self.settings.max_cost_for_worker
-        ):
-            return (
-                "Worker has reached the maximum cost it is allowed to spend"
-                f" which is given by `{self.settings.max_cost_for_worker=}`."
-                f" This worker has spend '{self.worker_cumulative_eval_cost}'."
-                "\n To allow more evaluations, increase this value or use a different"
-                " stopping criterion."
-            )
-
-        if self.settings.max_wallclock_time_for_worker_seconds is not None and (
+        if self.settings.max_wallclock_time_seconds is not None and (
             time.monotonic() - time_monotonic_start
-            >= self.settings.max_wallclock_time_for_worker_seconds
+            >= self.settings.max_wallclock_time_seconds
         ):
             return (
                 "Worker has reached the maximum wallclock time it is allowed to spend"
-                f", given by `{self.settings.max_wallclock_time_for_worker_seconds=}`."
-            )
-
-        if self.settings.max_evaluation_time_for_worker_seconds is not None and (
-            self.worker_cumulative_evaluation_time_seconds
-            >= self.settings.max_evaluation_time_for_worker_seconds
-        ):
-            return (
-                "Worker has reached the maximum evaluation time it is allowed to spend"
-                f", given by `{self.settings.max_evaluation_time_for_worker_seconds=}`."
+                f", given by `{self.settings.max_wallclock_time_seconds=}`."
             )
 
         return False
@@ -328,23 +286,30 @@ def _check_global_stopping_criterion(
         self,
         trials: Mapping[str, Trial],
     ) -> str | Literal[False]:
+        # worker related stopping criterion
+        worker_trials = {
+            _id: trial
+            for _id, trial in trials.items()
+            if trial.metadata.evaluating_worker_id == self.worker_id
+        }
         if self.settings.evaluations_to_spend is not None:
             if self.settings.include_in_progress_evaluations_towards_maximum:
                 count = sum(
                     1
-                    for _, trial in trials.items()
-                    if trial.metadata.state
-                    not in (Trial.State.PENDING, Trial.State.SUBMITTED)
+                    for _, trial in worker_trials.items()
+                    if trial.metadata.state != Trial.State.PENDING
                 )
             else:
                 # This indicates they have completed.
-                count = sum(1 for _, trial in trials.items() if trial.report is not None)
+                count = sum(
+                    1 for _, trial in worker_trials.items() if trial.report is not None
+                )
 
             if count >= self.settings.evaluations_to_spend:
                 return (
-                    "The total number of evaluations has reached the maximum allowed of"
-                    f" `{self.settings.evaluations_to_spend=}`."
-                    " To allow more evaluations, increase this value or use a different"
+                    "Worker has reached the maximum number of evaluations it is allowed"
+                    f" to do as given by `{self.settings.evaluations_to_spend=}`."
+                    "\nTo allow more evaluations, increase this value or use a different"
                     " stopping criterion."
                 )
 
@@ -354,7 +319,7 @@ def _check_global_stopping_criterion(
             fidelity_name = next(iter(self.optimizer.space.fidelities.keys()))
             count = sum(
                 trial.config[fidelity_name]
-                for _, trial in trials.items()
+                for _, trial in worker_trials.items()
                 if trial.report is not None and trial.config[fidelity_name] is not None
             )
             if count >= self.settings.fidelities_to_spend:
@@ -368,20 +333,22 @@ def _check_global_stopping_criterion(
         if self.settings.cost_to_spend is not None:
             cost = sum(
                 trial.report.cost
-                for _, trial in trials.items()
+                for _, trial in worker_trials.items()
                 if trial.report is not None and trial.report.cost is not None
             )
             if cost >= self.settings.cost_to_spend:
                 return (
-                    f"The maximum cost `{self.settings.cost_to_spend=}` has been"
-                    " reached by all of the evaluated trials. To allow more evaluations,"
-                    " increase this value or use a different stopping criterion."
+                    "Worker has reached the maximum cost it is allowed to spend"
+                    f" which is given by `{self.settings.cost_to_spend=}`."
+                    f" This worker has spend '{cost}'."
+                    "\n To allow more evaluations, increase this value or use a different"
+                    " stopping criterion."
                 )
 
         if self.settings.max_evaluation_time_total_seconds is not None:
             time_spent = sum(
                 trial.report.evaluation_duration
-                for _, trial in trials.items()
+                for _, trial in worker_trials.items()
                 if trial.report is not None
                 if trial.report.evaluation_duration is not None
             )
@@ -658,13 +625,6 @@ def run(self) -> None:  # noqa: C901, PLR0912, PLR0915
                     evaluation_fn=self.evaluation_fn,
                     default_report_values=self.settings.default_report_values,
                 )
-                evaluation_duration = evaluated_trial.metadata.evaluation_duration
-                assert (evaluation_duration is not None) | (report is None)
-                self.worker_cumulative_evaluation_time_seconds += (
-                    evaluation_duration if evaluation_duration else 0
-                )
-
-            self.worker_cumulative_eval_count += 1
 
             if report is None:
                 logger.info(
@@ -681,9 +641,6 @@ def run(self) -> None:  # noqa: C901, PLR0912, PLR0915
                 evaluated_trial.metadata.state,
             )
 
-            if report.cost is not None:
-                self.worker_cumulative_eval_cost += report.cost
-
             if report.err is not None:
                 logger.error(
                     f"Error during evaluation of '{evaluated_trial.id}'"
@@ -960,7 +917,6 @@ def _launch_runtime(  # noqa: PLR0913
     overwrite_optimization_dir: bool,
     evaluations_to_spend: int | None,
     fidelities_to_spend: int | float | None,
-    max_evaluations_for_worker: int | None,
     sample_batch_size: int | None,
     worker_id: str | None = None,
 ) -> None:
@@ -1034,11 +990,8 @@ def _launch_runtime(  # noqa: PLR0913
             not continue_until_max_evaluation_completed
         ),
         cost_to_spend=cost_to_spend,
-        max_evaluations_for_worker=max_evaluations_for_worker,
         max_evaluation_time_total_seconds=None,  # TODO: User can't specify yet
-        max_wallclock_time_for_worker_seconds=None,  # TODO: User can't specify yet
-        max_evaluation_time_for_worker_seconds=None,  # TODO: User can't specify yet
-        max_cost_for_worker=None,  # TODO: User can't specify yet
+        max_wallclock_time_seconds=None,  # TODO: User can't specify yet
     )
 
     # HACK: Due to nfs file-systems, locking with the default `flock()` is not reliable.