Coverage for moptipy / algorithms / luby_restarts.py: 54%
101 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-24 08:49 +0000
1"""
2Perform restarts of an algorithm based on the Luby sequence.
4Luby et al. showed that, for Las Vegas algorithms with known run time
5distribution, there is an optimal stopping time in order to minimize the
6expected running time. Even if the distribution is unknown, there is a
7universal sequence of running times given by (1,1,2,1,1,2,4,1,1,2,1,1,2,4,
88,...), which is the optimal restarting strategy up to constant factors.
9While this only holds for Las Vegas algorithms, this restart length may also
10be used for optimization, e.g., if we aim to find a globally optimal
11solution.
131. M. Luby, A. Sinclair, and S. Zuckerman. Optimal Speedup of Las Vegas
14 Algorithms. *Information Processing Letters* 47(4):173-180. September 1993.
15 https://doi.org/10.1016/0020-0190(93)90029-9
17"""
18from typing import Callable, Final, TypeVar, cast
20import numba # type: ignore
21from pycommons.types import check_int_range, type_error
23from moptipy.api.algorithm import Algorithm
24from moptipy.api.mo_algorithm import MOAlgorithm
25from moptipy.api.mo_process import MOProcess
26from moptipy.api.process import Process
27from moptipy.api.subprocesses import for_fes
28from moptipy.utils.logger import CSV_SEPARATOR, KeyValueLogSection
31@numba.njit(cache=True, inline="always", fastmath=True, boundscheck=False)
32def luby(i: int) -> int:
33 """
34 Compute the Luby sequence.
36 >>> [luby(ii) for ii in range(1, 65)]
37 [1, 1, 2, 1, 1, 2, 4, 1, 1, 2, 1, 1, 2, 4, 8, 1, 1, 2, 1, 1, 2, 4, 1, \
381, 2, 1, 1, 2, 4, 8, 16, 1, 1, 2, 1, 1, 2, 4, 1, 1, 2, 1, 1, 2, 4, 8, \
391, 1, 2, 1, 1, 2, 4, 1, 1, 2, 1, 1, 2, 4, 8, 16, 32, 1]
40 """
41 two_by_k: int = 1
42 while True:
43 two_by_k_minus_one = two_by_k
44 two_by_k *= 2
45 if two_by_k < two_by_k_minus_one:
46 return two_by_k_minus_one
47 if i == (two_by_k - 1):
48 return two_by_k_minus_one
49 if i >= two_by_k:
50 continue
51 i = (i - two_by_k_minus_one) + 1
52 two_by_k = 1
55class __LubyAlgorithm(Algorithm):
56 """A wrapper around an existing algorithm."""
58 def __init__(self, algorithm: Algorithm, base_fes: int,
59 log_restarts: bool = False) -> None:
60 """
61 Create the algorithm wrapper.
63 :param algorithm: the algorithm to wrap
64 :param base_fes: the base fes
65 :param log_restarts: should we log the restarts?
66 """
67 super().__init__()
68 #: the algorithm
69 self._algo: Final[Algorithm] = algorithm
70 #: the base FEs
71 self._base_fes: Final[int] = base_fes
72 self.initialize = algorithm.initialize # type: ignore # fast call
73 #: should we log the restarts?
74 self._log_restarts: Final[bool] = log_restarts
76 def __str__(self) -> str:
77 """
78 Get the string representation of this algorithm.
80 :returns: `luby_` + sub-algorithm name
81 """
82 return f"luby{self._base_fes}_{self._algo}"
84 def solve(self, process: Process) -> None:
85 """
86 Solve a single-objective problem.
88 :param process: the process
89 """
90 st: Final[Callable[[], bool]] = process.should_terminate
91 rst: Final[Callable[[], None]] = self.initialize
92 sv: Final[Callable[[Process], None]] = self._algo.solve
93 restarts: Final[list[tuple[int, int]] | None] = \
94 [] if self._log_restarts and process.has_log() else None
95 base: Final[int] = self._base_fes
96 index: int = 0
97 while not st():
98 if (index > 1) and (restarts is not None):
99 restarts.append((process.get_consumed_fes(),
100 process.get_consumed_time_millis()))
101 index += 1
102 with for_fes(process, base * luby(index)) as prc:
103 als: Callable[[str, str], None] = prc.add_log_section
105 def __als(t: str, c: str, _a=als, _i=index) -> None:
106 _a(f"{t}_{_i}", c)
108 prc.add_log_section = __als # type: ignore
109 rst()
110 sv(prc)
111 if (restarts is not None) and (len(restarts) > 0):
112 log: Final[list[str]] = [f"fes{CSV_SEPARATOR}timeMillis"]
113 for row in restarts:
114 log.append(CSV_SEPARATOR.join(map(
115 str, (x for x in row))))
116 del restarts
117 process.add_log_section("LUBY_RESTARTS", "\n".join(log))
118 del log
120 def log_parameters_to(self, logger: KeyValueLogSection) -> None:
121 """
122 Store the parameters of this luby algorithm wrapper to the logger.
124 :param logger: the logger
125 """
126 super().log_parameters_to(logger)
127 logger.key_value("baseFEs", self._base_fes)
128 logger.key_value("logRestarts", self._log_restarts)
129 with logger.scope("a") as scope:
130 self._algo.log_parameters_to(scope)
133class __LubyMOAlgorithm(__LubyAlgorithm, MOAlgorithm):
134 """A wrapper around an existing multi-objective algorithm."""
136 def solve_mo(self, process: MOProcess) -> None:
137 """
138 Solve a multi-objective problem.
140 :param process: the process
141 """
142 st: Final[Callable[[], bool]] = process.should_terminate
143 rst: Final[Callable[[], None]] = self.initialize
144 sv: Final[Callable[[MOProcess], None]] = cast(
145 "MOAlgorithm", self._algo).solve_mo
146 restarts: Final[list[tuple[int, int]] | None] = \
147 [] if self._log_restarts and process.has_log() else None
148 base: Final[int] = self._base_fes
149 index: int = 0
150 while not st():
151 if (index > 1) and (restarts is not None):
152 restarts.append((process.get_consumed_fes(),
153 process.get_consumed_time_millis()))
154 index += 1
155 with for_fes(process, base * luby(index)) as prc:
156 als: Callable[[str, str], None] = prc.add_log_section
158 def __als(t: str, c: str, _a=als, _i=index) -> None:
159 _a(f"{t}_{_i}", c)
161 prc.add_log_section = __als # type: ignore
162 rst()
163 sv(prc)
164 if (restarts is not None) and (len(restarts) > 0):
165 log: Final[list[str]] = [f"fes{CSV_SEPARATOR}timeMillis"]
166 for row in restarts:
167 log.append(CSV_SEPARATOR.join(map(
168 str, (x for x in row))))
169 del restarts
170 process.add_log_section("LUBY_RESTARTS", "\n".join(log))
171 del log
173 def __str__(self) -> str:
174 """
175 Get the string representation of this algorithm.
177 :returns: `moluby(base_fes)_` + sub-algorithm name
178 """
179 return f"moluby{self._base_fes}_{self._algo}"
182#: the type variable for single- and multi-objective algorithms.
183T = TypeVar("T", Algorithm, MOAlgorithm)
186def luby_restarts(algorithm: T, base_fes: int = 64,
187 log_restarts: bool = False) -> T:
188 """
189 Perform restarts of an algorithm in the Luby fashion.
191 The restart run length in FEs is determined by the Luby sequence (1, 1,
192 2, 1, 1, 2, 4, 1, 1, 2, 1, 1, 2, 4, 8, 1, 1, 2, 1, 1, 2, 4, 1, ...)
193 multiplied by `base_fes`. Restarts are performed until a termination
194 criterion is met.
196 :param algorithm: the algorithm
197 :param base_fes: the basic objective function evaluations
198 :param log_restarts: should we log the restarts?
199 """
200 check_int_range(base_fes, "base_fes", 1, 1_000_000_000_000)
201 if not isinstance(log_restarts, bool):
202 raise type_error(log_restarts, "log_restarts", bool)
203 if isinstance(algorithm, MOAlgorithm):
204 return __LubyMOAlgorithm(algorithm, base_fes, log_restarts)
205 if isinstance(algorithm, Algorithm):
206 return __LubyAlgorithm(algorithm, base_fes, log_restarts)
207 raise type_error(algorithm, "algorithm", (Algorithm, MOAlgorithm))