refactor: async classes

mlr-org · Apr 16, 2024 · 08ef4a7 · 08ef4a7
1 parent b54eb47
commit 08ef4a7
Show file tree

Hide file tree

Showing 37 changed files with 1,607 additions and 648 deletions.
diff --git a/DESCRIPTION b/DESCRIPTION
@@ -85,13 +85,17 @@ Collate:
     'ObjectiveRFunDt.R'
     'ObjectiveRFunMany.R'
     'OptimInstance.R'
+    'OptimInstanceAsync.R'
+    'OptimInstanceAsyncMultiCrit.R'
+    'OptimInstanceAsyncSingleCrit.R'
     'OptimInstanceMultiCrit.R'
-    'OptimInstanceRush.R'
-    'OptimInstanceRushMultiCrit.R'
-    'OptimInstanceRushSingleCrit.R'
     'OptimInstanceSingleCrit.R'
     'mlr_optimizers.R'
     'Optimizer.R'
+    'OptimizerAsync.R'
+    'OptimizerAsyncDesignPoints.R'
+    'OptimizerAsyncGridSearch.R'
+    'OptimizerAsyncRandomSearch.R'
     'OptimizerCmaes.R'
     'OptimizerDesignPoints.R'
     'OptimizerFocusSearch.R'
@@ -100,7 +104,6 @@ Collate:
     'OptimizerIrace.R'
     'OptimizerNLoptr.R'
     'OptimizerRandomSearch.R'
-    'OptimizerRandomSearchV2.R'
     'Progressor.R'
     'mlr_terminators.R'
     'Terminator.R'

diff --git a/NAMESPACE b/NAMESPACE
@@ -7,12 +7,8 @@ S3method(as.data.table,DictionaryTerminator)
 S3method(as_terminator,Terminator)
 S3method(as_terminators,default)
 S3method(as_terminators,list)
-S3method(assign_result_default,OptimInstance)
-S3method(assign_result_default,OptimInstanceRush)
 S3method(bb_optimize,"function")
 S3method(bb_optimize,Objective)
-S3method(optimize_default,OptimInstance)
-S3method(optimize_default,OptimInstanceRush)
 export(Archive)
 export(ArchiveBest)
 export(ArchiveRush)
@@ -24,12 +20,16 @@ export(ObjectiveRFun)
 export(ObjectiveRFunDt)
 export(ObjectiveRFunMany)
 export(OptimInstance)
+export(OptimInstanceAsync)
+export(OptimInstanceAsyncMultiCrit)
+export(OptimInstanceAsyncSingleCrit)
 export(OptimInstanceMultiCrit)
-export(OptimInstanceRush)
-export(OptimInstanceRushMultiCrit)
-export(OptimInstanceRushSingleCrit)
 export(OptimInstanceSingleCrit)
 export(Optimizer)
+export(OptimizerAsync)
+export(OptimizerAsyncDesignPoints)
+export(OptimizerAsyncGridSearch)
+export(OptimizerAsyncRandomSearch)
 export(OptimizerCmaes)
 export(OptimizerDesignPoints)
 export(OptimizerFocusSearch)
@@ -38,7 +38,6 @@ export(OptimizerGridSearch)
 export(OptimizerIrace)
 export(OptimizerNLoptr)
 export(OptimizerRandomSearch)
-export(OptimizerRandomSearchV2)
 export(Terminator)
 export(TerminatorClockTime)
 export(TerminatorCombo)
@@ -66,22 +65,24 @@ export(branin_wu)
 export(callback_optimization)
 export(clbk)
 export(clbks)
+export(finish_async_optimize)
 export(is_dominated)
 export(mlr_callbacks)
 export(mlr_optimizers)
 export(mlr_terminators)
 export(nds_selection)
 export(oi)
 export(opt)
-export(optimize_decentralized)
 export(optimize_default)
 export(opts)
 export(shrink_ps)
+export(start_async_optimize)
 export(terminated_error)
 export(trafo_xs)
 export(transform_xdt_to_xss)
 export(trm)
 export(trms)
+export(wait_for_async_optimize)
 import(checkmate)
 import(data.table)
 import(mlr3misc)

diff --git a/R/ContextOptimization.R b/R/ContextOptimization.R
@@ -22,7 +22,7 @@ ContextOptimization = R6Class("ContextOptimization",
     #' @param instance ([OptimInstance]).
     #' @param optimizer ([Optimizer]).
     initialize = function(instance, optimizer) {
-      self$instance = assert_multi_class(instance, c("OptimInstance", "OptimInstanceRush"))
+      self$instance = assert_multi_class(instance, c("OptimInstance", "OptimInstanceAsync"))
       self$optimizer = optimizer
     }
   ),

diff --git a/R/OptimInstance.R b/R/OptimInstance.R
@@ -262,7 +262,7 @@ objective_error = function(x, inst, maximization_to_minimization) {
     contains numeric values")
 }
 
-# used by OptimInstance and OptimInstanceRush
+# used by OptimInstance and OptimInstanceAsync
 choose_search_space = function(objective, search_space) {
   # create search space
   domain_search_space = objective$domain$search_space()

diff --git a/R/OptimInstanceRush.R → R/OptimInstanceAsync.R b/R/OptimInstanceRush.R → R/OptimInstanceAsync.R
@@ -3,17 +3,19 @@
 #' @include OptimInstance.R
 #'
 #' @description
-#' Abstract base class for [OptimInstanceRushSingleCrit] and [OptimInstanceRushMultiCrit].
+#'
+#'
+#' Abstract base class for [OptimInstanceAsyncSingleCrit] and [OptimInstanceAsyncMultiCrit].
 #' The optimization instances specify an optimization problem for [Optimizer]s.
 #' Points are evaluated asynchronously with the `rush` package.
-#' The function [oi()] creates an [OptimInstanceRushSingleCrit] or [OptimInstanceRushMultiCrit] and the function [bb_optimize()] creates an instance internally.
+#' The function [oi()] creates an [OptimInstanceAsyncSingleCrit] or [OptimInstanceAsyncMultiCrit] and the function [bb_optimize()] creates an instance internally.
 #'
 #' @template param_objective
 #' @template param_search_space
 #' @template param_terminator
-#' @template param_rush
 #' @template param_callbacks
 #' @template param_archive
+#' @template param_rush
 #'
 #' @template field_objective
 #' @template field_search_space
@@ -23,7 +25,7 @@
 #' @template field_archive_rush
 #'
 #' @export
-OptimInstanceRush = R6Class("OptimInstanceRush",
+OptimInstanceAsync = R6Class("OptimInstanceAsync",
   public = list(
 
     objective = NULL,
@@ -52,7 +54,7 @@ OptimInstanceRush = R6Class("OptimInstanceRush",
       self$search_space = choose_search_space(self$objective, search_space)
       self$terminator = assert_terminator(terminator, self)
       self$callbacks = assert_callbacks(as_callbacks(callbacks))
-      self$rush = rush %??% rsh()
+      self$rush = assert_rush(rush, null_ok = TRUE) %??% rsh()
 
       # archive is passed when a downstream packages creates a new archive class
       self$archive = if (is.null(archive)) {
@@ -89,59 +91,27 @@ OptimInstanceRush = R6Class("OptimInstanceRush",
       }
     },
 
-    #' @description
-    #' Adds points in `xdt` to the queue.
-    #' The points are evaluated by calling the [Objective] asynchronously.
-    #'
-    #' @param xdt (`data.table::data.table()`)\cr
-    #' x values as `data.table()` with one point per row.
-    #' Contains the value in  the *search space* of the [OptimInstance] object.
-    #' Can contain additional columns for extra information.
-    #' @param wait (`logical(1)`)\cr
-    #' If `TRUE`, wait for all evaluations to finish.
-    eval_async = function(xdt, wait = FALSE) {
-      assert_data_table(xdt)
-      assert_names(colnames(xdt), must.include = self$search_space$ids())
-
-      if (self$is_terminated) stop(terminated_error(self))
-
-      lg$info("Sending %i configuration(s) to workers:", max(1, nrow(xdt)))
-      lg$info(capture.output(print(xdt, class = FALSE, row.names = FALSE, print.keys = FALSE)))
-
-      xss = transpose_list(xdt[, self$search_space$ids(), with = FALSE])
-      xdt[, timestamp_xs := Sys.time()]
-      extra = transpose_list(xdt[, !self$search_space$ids(), with = FALSE])
-
-      # push to shared queue or priority queues
-      if (!is.null(xdt$priority_id)) {
-        keys = self$rush$push_priority_tasks(xss, extra, priority = xdt$priority_id)
-      } else {
-        keys = self$rush$push_tasks(xss, extra)
-      }
-
-      # optimizer can request to wait for all evaluations to finish
-      if (wait) {
-        self$rush$wait_for_tasks(keys, detect_lost_workers = TRUE) # private$.detect_lost_tasks
-      }
-
-      # terminate optimization if all workers crashed
-      if (!self$rush$n_running_workers)  {
-        lg$warn("Optimization terminated because all workers crashed.")
-        stop(terminated_error(self))
-      }
-
-      # if (private$.detect_lost_tasks) self$rush$detect_lost_tasks()
-
-      return(invisible(keys))
-    },
-
     #' @description
     #' Reset terminator and clear all evaluation results from archive and results.
     clear = function() {
       self$rush$reset()
       self$archive$clear()
       private$.result = NULL
       invisible(self)
+    },
+
+    #' @description
+    #' The [Optimizer] object writes the best found point and estimated performance value here.
+    #' For internal use.
+    #'
+    #' @param xdt (`data.table::data.table()`)\cr
+    #'  x values as `data.table::data.table()` with one row.
+    #' Contains the value in the  *search space* of the [OptimInstance] object.
+    #' Can contain additional columns for extra information.
+    #' @param y (`numeric(1)`)\cr
+    #'   Optimal outcome.
+    assign_result = function(xdt, y) {
+      stop("Abstract class")
     }
   ),
 

diff --git a/R/OptimInstanceRushMultiCrit.R → R/OptimInstanceAsyncMultiCrit.R b/R/OptimInstanceRushMultiCrit.R → R/OptimInstanceAsyncMultiCrit.R
@@ -1,20 +1,22 @@
 #' @title Multi Criterion Optimization Instance
 #'
 #' @description
-#' The [OptimInstanceRushMultiCrit] specifies an optimization problem for [Optimizer]s.
+#' The [OptimInstanceAsyncMultiCrit] specifies an optimization problem for [OptimizerAsync]s.
 #' Points are evaluated asynchronously with the `rush` package.
-#' The function [oi()] creates an [OptimInstanceRushMultiCrit] and the function [bb_optimize()] creates an instance internally.
+#' The function [oi()] creates an [OptimInstanceAsyncMultiCrit] and the function [bb_optimize()] creates an instance internally.
 #'
 #' @template param_objective
 #' @template param_search_space
 #' @template param_terminator
-#' @template param_rush
 #' @template param_callbacks
 #' @template param_archive
+#' @template param_rush
+#'
+#' @template param_xdt
 #'
 #' @export
-OptimInstanceRushMultiCrit = R6Class("OptimInstanceRushMultiCrit",
-  inherit = OptimInstanceRush,
+OptimInstanceAsyncMultiCrit = R6Class("OptimInstanceAsyncMultiCrit",
+  inherit = OptimInstanceAsync,
   public = list(
 
     #' @description
@@ -23,9 +25,9 @@ OptimInstanceRushMultiCrit = R6Class("OptimInstanceRushMultiCrit",
       objective,
       search_space = NULL,
       terminator,
-      rush,
       callbacks = list(),
-      archive = NULL
+      archive = NULL,
+      rush = NULL
       ) {
       if (objective$codomain$target_length == 1) {
         stop("Codomain length must be greater than 1.")
@@ -37,10 +39,14 @@ OptimInstanceRushMultiCrit = R6Class("OptimInstanceRushMultiCrit",
         rush = rush,
         callbacks = callbacks,
         archive = archive)
-    }
-  ),
+    },
 
-  private = list(
+    #' @description
+    #' The [Optimizer] object writes the best found points and estimated performance values here (probably the Pareto set / front).
+    #' For internal use.
+    #'
+    #' @param ydt (`numeric(1)`)\cr
+    #'  Optimal outcomes, e.g. the Pareto front.
     assign_result = function(xdt, ydt) {
       # FIXME: We could have one way that just lets us put a 1xn DT as result directly.
       assert_data_table(xdt)

diff --git a/R/OptimInstanceRushSingleCrit.R → R/OptimInstanceAsyncSingleCrit.R b/R/OptimInstanceRushSingleCrit.R → R/OptimInstanceAsyncSingleCrit.R
@@ -1,20 +1,22 @@
 #' @title Single Criterion Optimization Instance with Rush
 #'
 #' @description
-#' The [OptimInstanceRushSingleCrit] specifies an optimization problem for [Optimizer]s.
+#' The [OptimInstanceAsyncSingleCrit] specifies an optimization problem for [OptimizerAsync]s.
 #' Points are evaluated asynchronously with the `rush` package.
-#' The function [oi()] creates an [OptimInstanceRushSingleCrit] and the function [bb_optimize()] creates an instance internally.
+#' The function [oi()] creates an [OptimInstanceAsyncSingleCrit] and the function [bb_optimize()] creates an instance internally.
 #'
 #' @template param_objective
 #' @template param_search_space
 #' @template param_terminator
-#' @template param_rush
 #' @template param_callbacks
 #' @template param_archive
+#' @template param_rush
+#'
+#' @template param_xdt
 #'
 #' @export
-OptimInstanceRushSingleCrit = R6Class("OptimInstanceRushSingleCrit",
-  inherit = OptimInstanceRush,
+OptimInstanceAsyncSingleCrit = R6Class("OptimInstanceAsyncSingleCrit",
+  inherit = OptimInstanceAsync,
   public = list(
 
     #' @description
@@ -38,11 +40,15 @@ OptimInstanceRushSingleCrit = R6Class("OptimInstanceRushSingleCrit",
         callbacks = callbacks,
         archive = archive,
         rush = rush)
-    }
-  ),
+    },
 
-  private = list(
-    .assign_result = function(xdt, y) {
+    #' @description
+    #' The [OptimizerAsync] object writes the best found point and estimated performance value here.
+    #' For internal use.
+    #'
+    #' @param y (`numeric(1)`)\cr
+    #' Optimal outcome.
+    assign_result = function(xdt, y) {
       # FIXME: We could have one way that just lets us put a 1xn DT as result directly.
       assert_data_table(xdt)
       assert_names(names(xdt), must.include = self$search_space$ids())

diff --git a/R/OptimInstanceMultiCrit.R b/R/OptimInstanceMultiCrit.R
@@ -16,6 +16,7 @@
 #' @template param_search_space
 #' @template param_keep_evals
 #' @template param_callbacks
+#' @template param_archive
 #'
 #' @export
 OptimInstanceMultiCrit = R6Class("OptimInstanceMultiCrit",
@@ -54,9 +55,6 @@ OptimInstanceMultiCrit = R6Class("OptimInstanceMultiCrit",
     #' The [Optimizer] object writes the best found points
     #' and estimated performance values here (probably the Pareto set / front).
     #' For internal use.
-    #'
-    #' @param ydt (`numeric(1)`)\cr
-    #'   Optimal outcomes, e.g. the Pareto front.
     assign_result = function(xdt, ydt) {
       # FIXME: We could have one way that just lets us put a 1xn DT as result directly.
       assert_data_table(xdt)