diff --git a/CHANGELOG.md b/CHANGELOG.md index dc7cbf0a..bc3b9d26 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -51,6 +51,10 @@ compatibility changes and a short test procedure. These changes are not yet a ta - `PVForecastPVNode` — native 15-minute forecasts from the pvnode.com API. - `PVForecastForecastSolar` — forecasts from the free Forecast.Solar API. - `PVForecastSolcast` — forecasts from the Solcast rooftop-site API. +- Parallel fitness evaluation for GENETIC: `optimization.genetic.workers` (unset = automatic: + at most two worker processes, one CPU core and a container CPU limit stay free; `1` = one + process as before) and `optimization.genetic.pin_workers` (workers pinned to the last + cores, default on). For a fixed seed the plan is identical to a single-process run. ### Changed / compatibility diff --git a/docs/_generated/configexample.md b/docs/_generated/configexample.md index 179cb312..a4d96d69 100644 --- a/docs/_generated/configexample.md +++ b/docs/_generated/configexample.md @@ -259,6 +259,8 @@ "generations": 400, "seed": null, "fitness_cache_max_entries": null, + "workers": null, + "pin_workers": true, "measurement_max_age_seconds": 300, "tail_horizon_hours": 48, "terminal_value_mode": "AUTO", diff --git a/docs/_generated/configoptimization.md b/docs/_generated/configoptimization.md index 41e2a822..0fa95e78 100644 --- a/docs/_generated/configoptimization.md +++ b/docs/_generated/configoptimization.md @@ -32,6 +32,8 @@ "generations": 400, "seed": null, "fitness_cache_max_entries": null, + "workers": null, + "pin_workers": true, "measurement_max_age_seconds": 300, "tail_horizon_hours": 48, "terminal_value_mode": "AUTO", @@ -72,6 +74,8 @@ "generations": 400, "seed": null, "fitness_cache_max_entries": null, + "workers": null, + "pin_workers": true, "measurement_max_age_seconds": 300, "tail_horizon_hours": 48, "terminal_value_mode": "AUTO", @@ -186,11 +190,13 @@ | interval_sec | `Literal[900, 3600]` | `rw` | `3600` | The optimization interval [sec]. Defaults to 3600 seconds (1 hour) | | measurement_max_age_seconds | `int` | `rw` | `300` | Maximum age of SoC measurements for configuration-based optimization [s]. | | penalties | `dict[str, Union[float, int, str]]` | `rw` | `required` | Penalty parameters used in fitness evaluation. | +| pin_workers | `bool` | `rw` | `True` | Pin the evaluation workers to the last 'workers' CPU cores. The server process stays unpinned and keeps a core that is not busy with the optimization, so it answers requests during a run, and the run never uses more than 'workers' cores. Needs at least one core more than workers; Linux only. | | seed | `Optional[int]` | `rw` | `None` | Random seed for reproducibility. None = random. | | tail_horizon_hours | `int` | `rw` | `48` | Forecast lookahead after the control horizon [h]. No tail commands are issued. Set 0 to disable. | | terminal_value_euro_per_kwh | `float` | `rw` | `0.0` | Value assigned to usable battery energy remaining at the end of the optimization horizon [EUR/kWh]. This terminal value is independent of the battery LCOS. Only used with terminal_value_mode = FIXED. Defaults to 0 EUR/kWh. | | terminal_value_mode | `` | `rw` | `AUTO` | How to value the energy left in the battery at the end of the control horizon. AUTO solves the forecast tail with an AUTO continuation proxy at its end (or only the proxy if tail is zero); FIXED uses 'terminal_value_euro_per_kwh'. Defaults to AUTO. | | terminal_value_window_hours | `int` | `rw` | `24` | Length of the trailing window at the effective tail end the AUTO continuation curve is derived from [h]. One day covers a full load and PV cycle. Defaults to 24 hours. | +| workers | `Optional[int]` | `rw` | `None` | Processes that evaluate candidate solutions in parallel. None = automatic: one CPU core always stays free and at most two are used (1-2 cores -> 1, 3 or more cores -> 2), capped by a container CPU limit. 1 = evaluate in the server process only. The result for a fixed seed is the same with any value. | ::: @@ -209,6 +215,8 @@ "generations": 400, "seed": null, "fitness_cache_max_entries": null, + "workers": null, + "pin_workers": true, "measurement_max_age_seconds": 300, "tail_horizon_hours": 48, "terminal_value_mode": "AUTO", @@ -238,6 +246,8 @@ "generations": 400, "seed": null, "fitness_cache_max_entries": null, + "workers": null, + "pin_workers": true, "measurement_max_age_seconds": 300, "tail_horizon_hours": 48, "terminal_value_mode": "AUTO", diff --git a/openapi.json b/openapi.json index b7358420..416adb21 100644 --- a/openapi.json +++ b/openapi.json @@ -8819,6 +8819,34 @@ 5000 ] }, + "workers": { + "anyOf": [ + { + "type": "integer", + "minimum": 1.0 + }, + { + "type": "null" + } + ], + "title": "Workers", + "description": "Processes that evaluate candidate solutions in parallel. None = automatic: one CPU core always stays free and at most two are used (1-2 cores -> 1, 3 or more cores -> 2), capped by a container CPU limit. 1 = evaluate in the server process only. The result for a fixed seed is the same with any value.", + "examples": [ + null, + 1, + 2 + ] + }, + "pin_workers": { + "type": "boolean", + "title": "Pin Workers", + "description": "Pin the evaluation workers to the last 'workers' CPU cores. The server process stays unpinned and keeps a core that is not busy with the optimization, so it answers requests during a run, and the run never uses more than 'workers' cores. Needs at least one core more than workers; Linux only.", + "default": true, + "examples": [ + true, + false + ] + }, "measurement_max_age_seconds": { "type": "integer", "exclusiveMinimum": 0.0, @@ -8985,6 +9013,34 @@ 5000 ] }, + "workers": { + "anyOf": [ + { + "type": "integer", + "minimum": 1.0 + }, + { + "type": "null" + } + ], + "title": "Workers", + "description": "Processes that evaluate candidate solutions in parallel. None = automatic: one CPU core always stays free and at most two are used (1-2 cores -> 1, 3 or more cores -> 2), capped by a container CPU limit. 1 = evaluate in the server process only. The result for a fixed seed is the same with any value.", + "examples": [ + null, + 1, + 2 + ] + }, + "pin_workers": { + "type": "boolean", + "title": "Pin Workers", + "description": "Pin the evaluation workers to the last 'workers' CPU cores. The server process stays unpinned and keeps a core that is not busy with the optimization, so it answers requests during a run, and the run never uses more than 'workers' cores. Needs at least one core more than workers; Linux only.", + "default": true, + "examples": [ + true, + false + ] + }, "measurement_max_age_seconds": { "type": "integer", "exclusiveMinimum": 0.0, diff --git a/src/akkudoktoreos/optimization/genetic/genetic.py b/src/akkudoktoreos/optimization/genetic/genetic.py index 544bbb9d..6c3ad56e 100644 --- a/src/akkudoktoreos/optimization/genetic/genetic.py +++ b/src/akkudoktoreos/optimization/genetic/genetic.py @@ -4,6 +4,8 @@ import ctypes import ctypes.util import gc import math +import multiprocessing +import os import random import sys import time @@ -162,6 +164,111 @@ def _release_freed_memory() -> None: _LIBC.malloc_trim(0) +# CPU share of a container/systemd limit kept free for the server process: +# while the workers compute, it still has to answer API requests (a client +# pushing forecasts timed out when two workers used a 2-core limit fully). +SERVER_CPU_RESERVE = 0.25 + +# Parallel fitness evaluation. Workers are forked once per optimization run, +# after the run is fully prepared, so each one owns a copy of the optimizer with +# the run's forecasts, devices and terminal-value curve. Only genomes and +# results cross the process boundary. +MAX_EVALUATION_WORKERS = 2 +EVALUATION_BATCH_TIMEOUT_SECONDS = 120.0 +_WORKER_OPTIMIZER: Optional[Any] = None + + +def _cpu_cores() -> int: + """Return the number of CPU cores this process may run on.""" + try: + return max(1, len(os.sched_getaffinity(0))) + except (AttributeError, OSError): + return max(1, os.cpu_count() or 1) + + +def _cgroup_cpu_limit(cgroup_root: str = "/sys/fs/cgroup") -> Optional[float]: + """Return the CPU limit (cgroup v2 ``cpu.max``) in cores, if any. + + Checks this process's own cgroup and every parent, so a container limit + and a systemd ``CPUQuota=`` on a native service both count. + """ + try: + with open("/proc/self/cgroup") as own: + path = next( + (line.split("::", 1)[1].strip() for line in own if line.startswith("0::")), "/" + ) + except OSError: + path = "/" + limits: list[float] = [] + while True: + try: + with open(os.path.join(cgroup_root, path.lstrip("/"), "cpu.max")) as cpu_max: + quota, period = cpu_max.read().split()[:2] + if quota != "max": + limits.append(int(quota) / int(period)) + except (OSError, ValueError, ZeroDivisionError): + pass + if path in ("", "/"): + break + path = os.path.dirname(path.rstrip("/")) + return min(limits) if limits else None + + +def auto_evaluation_workers(cores: Optional[int] = None, cpu_limit: Optional[float] = None) -> int: + """Number of evaluation processes when ``workers`` is not configured. + + One core always stays free for the rest of the system, and at most + ``MAX_EVALUATION_WORKERS`` are used: 1-2 cores -> 1, 3 or more -> 2. + A CPU limit caps the result further and keeps ``SERVER_CPU_RESERVE`` of + it for the server process (limit 2 -> 1 worker, 2.5 -> 2). + """ + if cores is None: + cores = _cpu_cores() + if cpu_limit is None: + cpu_limit = _cgroup_cpu_limit() + workers = min(MAX_EVALUATION_WORKERS, max(1, cores - 1)) + if cpu_limit is not None: + workers = min(workers, max(1, math.floor(cpu_limit - SERVER_CPU_RESERVE))) + return workers + + +def worker_cpu_set(workers: int, allowed: Optional[set[int]] = None) -> Optional[set[int]]: + """CPU cores the evaluation workers are pinned to, or None for no pinning. + + The workers get the last ``workers`` cores this process may use; the + server process itself stays unpinned. It therefore always finds a core + that is not busy with the optimization and keeps answering requests, and + the optimization never spreads beyond ``workers`` cores (no CPU quota + needed). Without a spare core there is nothing to gain: no pinning. + """ + if allowed is None: + try: + allowed = set(os.sched_getaffinity(0)) + except (AttributeError, OSError): + return None + if workers < 1 or len(allowed) <= workers: + return None + return set(sorted(allowed)[-workers:]) + + +def _evaluation_worker_init(cpus: Optional[set[int]] = None) -> None: + """Evaluate without the run cache in a worker; the parent owns the cache.""" + if _WORKER_OPTIMIZER is not None: + _WORKER_OPTIMIZER._fitness_cache_enabled = False + if cpus: + try: + os.sched_setaffinity(0, cpus) + except (AttributeError, OSError): + pass + + +def _evaluation_worker(genome: list[int]) -> tuple[list[int], tuple[float], Any]: + """Evaluate one genome in a worker and return the (repaired) genome and results.""" + individual = creator.Individual(genome) + fitness = _WORKER_OPTIMIZER.toolbox.evaluate(individual) # type: ignore[union-attr] + return list(individual), tuple(fitness), getattr(individual, "extra_data", None) + + @dataclass(frozen=True) class FitnessCacheEntry: """One canonical, successful fitness evaluation within an optimization run.""" @@ -847,6 +954,9 @@ class GeneticOptimization(OptimizationBase): self._fitness_cache_max_entries: Optional[int] = None self._fitness_cache_hits = 0 self._fitness_cache_misses = 0 + # Evaluation worker processes, forked per run in optimize(). + self._evaluation_pool: Optional[Any] = None + self._evaluation_workers = 1 # Appliance genome layout, built once per optimization run in # optimize_ems(). Empty by default so setup_deap_environment() can be @@ -2467,11 +2577,132 @@ class GeneticOptimization(OptimizationBase): def _evaluate_invalid(self, population: list[Any]) -> int: """Evaluate invalid individuals and return the number of cache lookups.""" invalid = [individual for individual in population if not individual.fitness.valid] - fitnesses = self.toolbox.map(self.toolbox.evaluate, invalid) - for individual, fitness in zip(invalid, fitnesses): + pool = getattr(self, "_evaluation_pool", None) + if pool is not None and len(invalid) > 1: + try: + self._evaluate_parallel(pool, invalid) + return len(invalid) + except Exception as exc: + logger.warning( + "Parallel fitness evaluation failed ({}); evaluating serially for the " + "rest of this run.", + exc, + ) + self._stop_evaluation_pool() + pending = [individual for individual in invalid if not individual.fitness.valid] + fitnesses = self.toolbox.map(self.toolbox.evaluate, pending) + for individual, fitness in zip(pending, fitnesses): individual.fitness.values = fitness return len(invalid) + def _evaluate_parallel(self, pool: Any, invalid: list[Any]) -> None: + """Evaluate individuals in the worker processes, same results as evaluate(). + + Cache lookups and stores stay in this process. With the cache active, + identical genomes are evaluated once, as a serial run would hit the + cache for the repeats. Results are applied only when the whole batch + has returned, so a failed batch leaves nothing half-assigned. + """ + cache_enabled = getattr(self, "_fitness_cache_enabled", False) + groups: list[list[Any]] = [] + keys: list[Optional[PackedGenes]] = [] + by_key: dict[PackedGenes, list[Any]] = {} + for individual in invalid: + if not cache_enabled: + groups.append([individual]) + keys.append(None) + continue + key = self._fitness_key(individual) + cached = self._fitness_cache.get(key) + if cached is not None: + individual.fitness.values = self._apply_cached_fitness(individual, cached) + continue + group = by_key.get(key) + if group is not None: + group.append(individual) + self._fitness_cache_hits += 1 + continue + by_key[key] = [individual] + groups.append(by_key[key]) + keys.append(key) + self._fitness_cache_misses += 1 + if not groups: + return + + genomes = [[int(value) for value in group[0]] for group in groups] + chunksize = max(1, len(genomes) // (self._evaluation_workers * 4)) + results = pool.map_async(_evaluation_worker, genomes, chunksize=chunksize).get( + timeout=EVALUATION_BATCH_TIMEOUT_SECONDS + ) + for group, cache_key, (genome, fitness, extra_data) in zip(groups, keys, results): + for individual in group: + individual[:] = genome + if extra_data is None: + if hasattr(individual, "extra_data"): + del individual.extra_data + else: + individual.extra_data = extra_data + individual.fitness.values = fitness + if cache_key is not None and extra_data is not None: + self._store_fitness_cache(cache_key, group[0], fitness) + + def _configured_evaluation_workers(self) -> int: + """Configured number of evaluation processes; automatic when unset.""" + try: + workers = self._genetic_cfg.workers + except Exception: + workers = None + if workers is None: + return auto_evaluation_workers() + return max(1, int(workers)) + + def _start_evaluation_pool(self) -> None: + """Fork the evaluation workers for this run (none when one process is enough).""" + global _WORKER_OPTIMIZER + self._evaluation_pool = None + self._evaluation_workers = self._configured_evaluation_workers() + if self._evaluation_workers < 2: + return + try: + context = multiprocessing.get_context("fork") + except ValueError: + logger.warning("Parallel fitness evaluation needs fork(); evaluating serially.") + return + cpus = worker_cpu_set(self._evaluation_workers) if self._pin_workers() else None + _WORKER_OPTIMIZER = self + try: + self._evaluation_pool = context.Pool( + self._evaluation_workers, + initializer=_evaluation_worker_init, + initargs=(cpus,), + ) + except Exception as exc: + _WORKER_OPTIMIZER = None + logger.warning("Could not start evaluation workers ({}); evaluating serially.", exc) + return + logger.info( + "Genetic evaluation: {} worker processes{}.", + self._evaluation_workers, + f" on CPU cores {sorted(cpus)}" if cpus else "", + ) + + def _pin_workers(self) -> bool: + """Whether evaluation workers are pinned to fixed CPU cores.""" + try: + return bool(self._genetic_cfg.pin_workers) + except Exception: + return True + + def _stop_evaluation_pool(self) -> None: + """Stop the evaluation workers of this run.""" + global _WORKER_OPTIMIZER + pool = getattr(self, "_evaluation_pool", None) + self._evaluation_pool = None + _WORKER_OPTIMIZER = None + if pool is not None: + pool.terminate() + pool.join() + def _fresh_population(self, count: int, *, educated_fraction: float) -> list[Any]: """Create a mixed set of current educated guesses and random immigrants.""" if count <= 0: @@ -2918,19 +3149,30 @@ class GeneticOptimization(OptimizationBase): original_key = self._fitness_key(individual) cached = self._fitness_cache.get(original_key) if cached is not None: - individual[:] = _unpack_genes(cached.genome) - individual.extra_data = cached.extra_data # type: ignore[attr-defined] - self._fitness_cache_hits += 1 - return cached.fitness + return self._apply_cached_fitness(individual, cached) self._fitness_cache_misses += 1 fitness = self._evaluate_uncached(individual, parameters, start_hour, worst_case) + self._store_fitness_cache(original_key, individual, fitness) + return fitness + + def _apply_cached_fitness(self, individual: Any, cached: FitnessCacheEntry) -> tuple[float]: + """Give an individual the cached canonical genome and results.""" + individual[:] = _unpack_genes(cached.genome) + individual.extra_data = cached.extra_data + self._fitness_cache_hits += 1 + return cached.fitness + + def _store_fitness_cache( + self, original_key: PackedGenes, individual: Any, fitness: tuple[float] + ) -> None: + """Remember a successful evaluation under its original and canonical key.""" extra_data = getattr(individual, "extra_data", None) if extra_data is None: # Failed evaluations use the sentinel fitness and are intentionally # not cached: an unexpected transient failure must never become a # persistent result for the remainder of the run. - return fitness + return canonical_key = self._fitness_key(individual) extra_value1, extra_value2, extra_value3 = extra_data @@ -2948,7 +3190,6 @@ class GeneticOptimization(OptimizationBase): max_entries = getattr(self, "_fitness_cache_max_entries", None) if max_entries is not None: self._evict_fitness_cache(max_entries) - return fitness def _evict_fitness_cache(self, max_entries: int) -> None: """Drop the oldest cache keys until at most ``max_entries`` remain. @@ -3375,6 +3616,7 @@ class GeneticOptimization(OptimizationBase): self._fitness_cache_hits = 0 self._fitness_cache_misses = 0 self._fitness_cache_enabled = max_entries != 0 + self._start_evaluation_pool() local_evaluations = 0 local_improvements = 0 local_initial_fitness = float("nan") @@ -3408,6 +3650,7 @@ class GeneticOptimization(OptimizationBase): _release_freed_memory() raise finally: + self._stop_evaluation_pool() self._fitness_cache_enabled = False if local_improvements: diff --git a/src/akkudoktoreos/optimization/genetic/geneticsettings.py b/src/akkudoktoreos/optimization/genetic/geneticsettings.py index f692e768..b4d7ae61 100644 --- a/src/akkudoktoreos/optimization/genetic/geneticsettings.py +++ b/src/akkudoktoreos/optimization/genetic/geneticsettings.py @@ -123,6 +123,34 @@ class GeneticCommonSettings(SettingsBaseModel): }, ) + workers: Optional[int] = Field( + default=None, + ge=1, + json_schema_extra={ + "description": ( + "Processes that evaluate candidate solutions in parallel. None = automatic: " + "one CPU core always stays free and at most two are used (1-2 cores -> 1, " + "3 or more cores -> 2), capped by a container CPU limit. 1 = evaluate in the " + "server process only. The result for a fixed seed is the same with any value." + ), + "examples": [None, 1, 2], + }, + ) + + pin_workers: bool = Field( + default=True, + json_schema_extra={ + "description": ( + "Pin the evaluation workers to the last 'workers' CPU cores. The server " + "process stays unpinned and keeps a core that is not busy with the " + "optimization, so it answers requests during a run, and the run never " + "uses more than 'workers' cores. Needs at least one core more than " + "workers; Linux only." + ), + "examples": [True, False], + }, + ) + measurement_max_age_seconds: int = Field( default=300, gt=0, diff --git a/tests/test_genetic_parallel_evaluation.py b/tests/test_genetic_parallel_evaluation.py new file mode 100644 index 00000000..bb6e19fe --- /dev/null +++ b/tests/test_genetic_parallel_evaluation.py @@ -0,0 +1,205 @@ +"""Parallel fitness evaluation gives the same result as evaluating in one process.""" + +import multiprocessing + +import pytest + +from akkudoktoreos.config.config import ConfigEOS +from akkudoktoreos.core.cache import CacheEnergyManagementStore +from akkudoktoreos.core.coreabc import get_ems +from akkudoktoreos.optimization.genetic import genetic as genetic_module +from akkudoktoreos.optimization.genetic.genetic import ( + GeneticOptimization, + auto_evaluation_workers, +) +from akkudoktoreos.optimization.genetic.geneticparams import ( + GeneticOptimizationParameters, +) +from akkudoktoreos.utils.datetimeutil import to_datetime + +ems_eos = get_ems(init=True) # init once + +HOURS = 24 + + +@pytest.mark.parametrize( + "cores,cpu_limit,expected", + [ + (1, None, 1), + (2, None, 1), # one core stays free + (3, None, 2), + (4, None, 2), + (16, None, 2), # at most two + (4, 1, 1), # container CPU limit + (4, 2, 1), # a quarter core stays free for the server process + (4, 2.5, 2), + (4, 3, 2), + (2, 4, 1), + ], +) +def test_auto_workers_keep_one_core_free(cores, cpu_limit, expected): + assert auto_evaluation_workers(cores=cores, cpu_limit=cpu_limit) == expected + + +def _parameters( + config_eos: ConfigEOS, workers: int, cache_entries +) -> GeneticOptimizationParameters: + config_eos.merge_settings_from_dict( + { + "prediction": {"hours": HOURS}, + "optimization": { + "genetic": { + "individuals": 30, + "generations": 10, + "tail_horizon_hours": 0, + "horizon_hours": HOURS, + "interval_sec": 3600, + "workers": workers, + "fitness_cache_max_entries": cache_entries, + } + }, + "feedintariff": {"direct_marketing_enabled": True}, + "devices": { + "max_batteries": 1, + "batteries": { + "battery1": {"device_id": "battery1", "grid_export_rates": [0.5, 1.0]} + }, + }, + } + ) + ems_eos.set_start_datetime(to_datetime().set(hour=0, minute=0)) + CacheEnergyManagementStore().clear() + return GeneticOptimizationParameters.model_validate( + dict( + ems={ + "pv_prognose_wh": [0.0] * 8 + [3000.0] * 8 + [0.0] * 8, + "strompreis_euro_pro_wh": [0.0003] * 6 + [0.0001] * 6 + [0.0004] * 12, + "einspeiseverguetung_euro_pro_wh": [0.0001] * 12 + [0.0009] * 12, + "preis_euro_pro_wh_akku": 0.0, + "gesamtlast": [400.0] * HOURS, + }, + pv_battery={ + "device_id": "battery1", + "capacity_wh": 10000, + "initial_soc_percentage": 40, + "min_soc_percentage": 5, + "max_charge_power_w": 5000, + }, + inverter={ + "device_id": "inverter1", + "max_power_wh": 10000, + "battery_id": "battery1", + }, + ev=None, + ) + ) + + +def _run(config_eos: ConfigEOS, workers: int, cache_entries): + parameters = _parameters(config_eos, workers, cache_entries) + optimization = GeneticOptimization(fixed_seed=7) + used: list[int] = [] + original_start = optimization._start_evaluation_pool + + def start_and_record(): + original_start() + used.append( + 0 if optimization._evaluation_pool is None else optimization._evaluation_workers + ) + + optimization._start_evaluation_pool = start_and_record # type: ignore[method-assign] + solution = optimization.optimize_ems(parameters=parameters, start_hour=0, ngen=6) + assert optimization._evaluation_pool is None # stopped after the run + return solution, used + + +@pytest.mark.parametrize("cache_entries", [None, 0]) +@pytest.mark.skipif("fork" not in multiprocessing.get_all_start_methods(), reason="requires fork") +def test_parallel_evaluation_matches_serial(config_eos: ConfigEOS, cache_entries): + serial, serial_used = _run(config_eos, 1, cache_entries) + parallel, parallel_used = _run(config_eos, 2, cache_entries) + + assert serial_used == [0] + assert parallel_used == [2] + assert parallel.start_solution == serial.start_solution + assert parallel.result.total_balance == serial.result.total_balance + assert parallel.fitness_history == serial.fitness_history + assert parallel.ac_charge == serial.ac_charge + assert parallel.dc_charge == serial.dc_charge + assert parallel.discharge_allowed == serial.discharge_allowed + assert parallel.battery_grid_export_factor == serial.battery_grid_export_factor + + +@pytest.mark.skipif("fork" not in multiprocessing.get_all_start_methods(), reason="requires fork") +def test_failed_batch_falls_back_to_serial(config_eos: ConfigEOS, monkeypatch): + serial, _ = _run(config_eos, 1, None) + + def broken_worker(genome): + raise RuntimeError("worker lost") + + monkeypatch.setattr(genetic_module, "_evaluation_worker", broken_worker) + fallback, used = _run(config_eos, 2, None) + + assert used == [2] + assert fallback.start_solution == serial.start_solution + assert fallback.ac_charge == serial.ac_charge + + +def test_unsupported_fork_falls_back_to_serial(config_eos: ConfigEOS, monkeypatch): + serial, _ = _run(config_eos, 1, None) + + def unsupported_context(method): + assert method == "fork" + raise ValueError("fork is unavailable") + + monkeypatch.setattr(genetic_module.multiprocessing, "get_context", unsupported_context) + fallback, used = _run(config_eos, 2, None) + + assert used == [0] + assert fallback.start_solution == serial.start_solution + assert fallback.result.total_balance == serial.result.total_balance + assert fallback.ac_charge == serial.ac_charge + assert fallback.dc_charge == serial.dc_charge + + +def test_cgroup_cpu_limit_reads_own_and_parent_cgroups(tmp_path, monkeypatch): + from akkudoktoreos.optimization.genetic.genetic import _cgroup_cpu_limit + + service = tmp_path / "system.slice" / "eos.service" + service.mkdir(parents=True) + (tmp_path / "cpu.max").write_text("max 100000\n") + (tmp_path / "system.slice" / "cpu.max").write_text("max 100000\n") + (service / "cpu.max").write_text("150000 100000\n") + + real_open = open + + def fake_open(path, *args, **kwargs): + if path == "/proc/self/cgroup": + import io + + return io.StringIO("0::/system.slice/eos.service\n") + return real_open(path, *args, **kwargs) + + monkeypatch.setattr("builtins.open", fake_open) + assert _cgroup_cpu_limit(str(tmp_path)) == 1.5 + (service / "cpu.max").write_text("max 100000\n") + assert _cgroup_cpu_limit(str(tmp_path)) is None + (tmp_path / "cpu.max").write_text("100000 100000\n") + assert _cgroup_cpu_limit(str(tmp_path)) == 1 # a parent limit counts too + + +@pytest.mark.parametrize( + "workers,allowed,expected", + [ + (2, {0, 1, 2, 3}, {2, 3}), # core 0/1 stay free for the server and the system + (3, {0, 1, 2, 3}, {1, 2, 3}), + (1, {0, 1}, {1}), + (2, {0, 1}, None), # no spare core -> no pinning + (2, {4, 5, 6, 7}, {6, 7}), # restricted container cpuset + (0, {0, 1, 2, 3}, None), + ], +) +def test_worker_cpu_set(workers, allowed, expected): + from akkudoktoreos.optimization.genetic.genetic import worker_cpu_set + + assert worker_cpu_set(workers, allowed) == expected