from __future__ import annotations
import concurrent.futures
import getpass
import json
import logging
import multiprocessing
import pathlib
import queue
import threading
import time
import traceback
import typing
import uuid
import warnings
from datetime import UTC, datetime
import dill
import numpy as np # noqa: F401
import pandas as pd # noqa: F401
from passengersim.callbacks import CallbackMixin
from passengersim.config import Config
from passengersim.core import SimulationEngine
from passengersim.diagnostics.collection import RunDiagnostics, summary_metadata
from passengersim.diagnostics.monitor import _RunMonitor
from passengersim.driver._base_sim import BaseSimulation
from passengersim.driver._singleproc import Simulation, check_summarizer
from passengersim.driver._trial_database import TrialDatabaseStore, TrialResult, finalize_result, finish_database
from passengersim.mp_executor import JobExecutor
from passengersim.rm.systems import export_registered_rm_systems, restore_registered_rm_systems
from passengersim.summaries import GenericSimulationTables, SimulationTables
from passengersim.utils.tempdir import MaybeTemporaryDirectory
if typing.TYPE_CHECKING:
from rich.progress import Progress
class _QueueLike(typing.Protocol):
"""Minimal destination interface for trial progress telemetry.
Parallel trials send messages through multiprocessing queues, while
sequential trials can use a bounded local queue or ``_DirectProgress``
to deliver messages synchronously to the display. Producers only need
``put_nowait``; requiring a concrete queue would exclude the direct adapter.
This protocol is used for static typing, without runtime checks or explicit
inheritance by destinations. Queue implementations may raise ``queue.Full``;
``_offer_progress`` treats that as dropped telemetry rather than waiting for
space. The direct adapter invokes its callback immediately without buffering.
"""
# Positional-only allows implementations to call this argument "item" or
# "obj" instead of "message", as the standard queue implementations do.
def put_nowait(self, message: tuple[int, str, typing.Any], /) -> None:
"""Accept a progress message without waiting for queue capacity.
Parameters
----------
message : tuple
Trial identifier, event name, and event-specific payload.
Raises
------
queue.Full
If a bounded destination cannot accept another message.
"""
...
def _offer_progress(progress_queue: _QueueLike, message: tuple[int, str, typing.Any]) -> bool:
# The Future carries authoritative final state; telemetry must never wait
# for a slow terminal/notebook to consume a full queue.
try:
progress_queue.put_nowait(message)
return True
except queue.Full:
return False
def _resolve_run_id(run_id: str | None, scenario: str | None) -> str:
"""Return the display and log key for a run, shared by both execution modes.
Parameters
----------
run_id : str or None
Explicit identifier. If None, the scenario name is used.
scenario : str or None
Configured scenario name, used when ``run_id`` is None.
Returns
-------
str
The identifier with one trailing space for progress-bar formatting. A
short random identifier is used when neither input gives a name, so
unnamed runs sharing a display do not collide.
"""
if run_id is None:
run_id = scenario
if not run_id:
run_id = str(uuid.uuid4().hex[:4])
run_id = str(run_id)
if not run_id.endswith(" "):
run_id = run_id + " "
return run_id
class _DirectProgress:
"""Deliver sequential telemetry immediately without allocating a queue."""
def __init__(self, callback: typing.Callable[[tuple[int, str, typing.Any]], None]) -> None:
self.callback = callback
def put_nowait(self, message: tuple[int, str, typing.Any]) -> None:
self.callback(message)
class ThrottledUpdater:
"""Offer bounded progress snapshots without waiting for a display consumer."""
def __init__(
self,
progress_queue: _QueueLike,
trial_id: int,
throttle: float = 1.0,
*,
handles_diagnostic_logging: bool = False,
on_snapshot: typing.Callable[[dict[str, typing.Any]], None] | None = None,
) -> None:
self.progress_queue = progress_queue
self.trial_id = trial_id
self.throttle = throttle
self.last_time = 0
self.n = 0
self.completed = 0
self.handles_diagnostic_logging = handles_diagnostic_logging
self.on_snapshot = on_snapshot
self.diagnostics = None
def attach_diagnostics(self, diagnostics: RunDiagnostics | None) -> None:
"""Connect a collector, which then drives progress through its snapshots.
With diagnostics off (``None``), progress falls back to plain sample-count
updates. Otherwise cumulative snapshots carry the completed sample count,
so separate update messages are unnecessary.
"""
self.diagnostics = diagnostics
if diagnostics is not None:
diagnostics.publisher = self.publish
diagnostics.publish_interval = 1.0
diagnostics.next_publish = diagnostics.clock() # first snapshot is available immediately
def publish(self, snapshot: dict[str, typing.Any]) -> None:
"""Offer a cumulative snapshot, dropping it if the queue is full."""
if self.on_snapshot is not None:
self.on_snapshot(snapshot)
return
_offer_progress(self.progress_queue, (self.trial_id, "diagnostics", snapshot))
def __call__(self, completed: int, total: int) -> None:
# The callback includes an initial zero; count samples, not callbacks.
self.n += completed - self.completed
self.completed = completed
if self.diagnostics is not None:
self.diagnostics.publish()
elif self.n and time.monotonic() - self.last_time > self.throttle:
if _offer_progress(self.progress_queue, (self.trial_id, "update", self.n)):
self.n = 0
self.last_time = time.monotonic()
def flush(self) -> None:
"""Offer final progress; the worker Future remains authoritative."""
if self.diagnostics is not None:
self.publish(self.diagnostics.snapshot())
self.n = 0
elif _offer_progress(self.progress_queue, (self.trial_id, "update", self.n)):
self.n = 0
self.last_time = time.monotonic()
def _subprocess_run_trial(
trial_id: int,
cfg_json: str,
progress_queue: _QueueLike | None = None,
output_dir: pathlib.Path | None = None,
*,
summarizer=SimulationTables,
callbacks=None,
exported_rm_systems=None,
database_path: pathlib.Path | None = None,
diagnostic_callback: typing.Callable[[dict[str, typing.Any]], None] | None = None,
):
trial_started = time.perf_counter()
sim = None
parent_logging = progress_queue is not None or diagnostic_callback is not None
if progress_queue is None:
# Sequential runs have no consumer; they must not accumulate telemetry.
progress_queue = queue.Queue(maxsize=1)
try:
if exported_rm_systems:
restore_registered_rm_systems(exported_rm_systems)
cfg = Config.model_validate(json.loads(cfg_json))
# Never write basic outputs to disk from a subprocess, except possibly for sim_state
cfg.outputs.disk = False
cfg.outputs.html.filename = False
cfg.outputs.pickle = False
cfg.outputs.excel = False
# Writing to database is not a "output" and so is still allowed
# but we need to make sure that each subprocess has a unique database file
# because SQLite does not like multiple processes writing to the same database file.
if database_path is not None:
cfg.db.filename = database_path
cfg.db.result_directory = None
cfg.db.result_storage = "persistent"
output_dir = MaybeTemporaryDirectory(output_dir)
sim = Simulation(cfg, output_dir.joinpath(f"passengersim-trial-{trial_id:02}"))
if callbacks is not None:
for cb in callbacks.get("begin_sample_callbacks", []):
sim.begin_sample_callback(cb)
for cb in callbacks.get("end_sample_callbacks", []):
sim.end_sample_callback(cb)
for cb in callbacks.get("daily_callbacks", []):
sim.daily_callback(cb)
sim.sample_done_callback = ThrottledUpdater(
progress_queue,
trial_id,
handles_diagnostic_logging=parent_logging,
on_snapshot=diagnostic_callback,
)
summary = sim.run(single_trial=trial_id, summarizer=summarizer)
final_diagnostics = sim._diagnostics.snapshot() if sim._diagnostics is not None else None
simulation_finished = time.perf_counter()
sim.sample_done_callback.flush()
_offer_progress(progress_queue, (trial_id, "finalizing", None))
# Passing a database connection between processes is not allowed,
# so we need to delete it before returning the summary. But first
# we will run any queries that were requested as output reports
if isinstance(summary, GenericSimulationTables) and summary.db is not None:
summary.run_queries(items=cfg.outputs.reports)
summary.__dict__.pop("_trial_database_store", None)
try:
summary.db = None
except AttributeError as err:
print(err)
reports_finished = time.perf_counter()
descriptor = None
if database_path is not None:
descriptor = finish_database(sim.db, trial_id)
else:
sim.db.close()
_offer_progress(progress_queue, (trial_id, "done", None))
published = time.perf_counter()
return TrialResult(
summary,
descriptor,
{
"simulation_and_writes_s": simulation_finished - trial_started,
"eager_reports_s": reports_finished - simulation_finished,
"worker_finalization_s": published - reports_finished,
},
published,
diagnostics=final_diagnostics,
)
except Exception as e:
# Capture the subprocess's full traceback text so it can be preserved
# when the exception is pickled and re-raised in the parent process.
# Exception traceback objects (``__traceback__``) are not picklable, so
# we attach the formatted traceback string via ``add_note`` (PEP 678).
# Notes survive pickling and are automatically displayed by Python's
# default traceback printer when the exception is eventually raised.
tb_str = traceback.format_exc()
note = f"Subprocess traceback (trial {trial_id}):\n{tb_str.rstrip()}"
try:
e.add_note(note)
except AttributeError:
# Fallback for very old Python versions lacking ``add_note``.
existing = getattr(e, "__notes__", [])
e.__notes__ = [*existing, note]
# Also store the formatted traceback as a plain attribute so the
# parent process can display it directly (e.g., in the rich console)
# without needing to re-raise the exception.
try:
e._passengersim_subprocess_tb = tb_str
except Exception:
pass
_offer_progress(progress_queue, (trial_id, "exception", None))
return e
finally:
if sim is not None and sim.db.connection is not None:
# Failed trials are never advertised as complete. Roll back pending
# work and release handles before the parent can clean up files.
sim.db.disconnect_from_simulation_engine()
sim.db.connection.rollback()
sim.db.close()
def _executor_run_trial(
trial_id: int,
cfg_json: str,
progress_queue,
output_dir: pathlib.Path | None,
dill_kwargs_blob: bytes,
database_path: pathlib.Path | None = None,
):
"""Worker function for use with ProcessPoolExecutor.
Deserializes dill-encoded kwargs and delegates to _subprocess_run_trial.
Exceptions are re-raised so the Future captures them.
"""
kwargs = dill.loads(dill_kwargs_blob)
result = _subprocess_run_trial(
trial_id, cfg_json, progress_queue, output_dir, database_path=database_path, **kwargs
)
if isinstance(result, Exception):
raise result
return result
[docs]
class MultiSimulation(BaseSimulation, CallbackMixin):
def __init__(
self,
config: Config,
output_dir: pathlib.Path | None = None,
*,
database_initializer=None,
):
self._database_initializer = database_initializer
super().__init__(config, output_dir)
self.config = config
self._placeholder_sim = None
[docs]
def close(self):
"""Close the current result and every database alias; preserve persistent files."""
store = getattr(self, "_trial_database_store", None)
if store is not None:
store.close()
def __enter__(self):
return self
def __exit__(self, *args):
self.close()
@property
def db(self):
"""The shared consolidated database, opened on the consuming thread."""
store = getattr(self, "_trial_database_store", None)
return store.database if store is not None else None
@db.setter
def db(self, value):
# BaseSimulation initializes this attribute before a run has an owner.
if value is not None:
raise AttributeError("MultiSimulation.db is managed by its result owner")
@property
def database_path(self):
"""Absolute consolidated result path, without opening its connection."""
store = getattr(self, "_trial_database_store", None)
return store.consolidated_path if store is not None else None
@property
def database_available(self):
"""Whether this simulation has retained detail data."""
store = getattr(self, "_trial_database_store", None)
return bool(store is not None and store.available)
@property
def database_status(self):
"""Completed, failed, and cancelled trials, plus whole-run completeness."""
store = getattr(self, "_trial_database_store", None)
return store.status if store is not None else None
@property
def trial_databases(self):
"""Completed trial descriptors; temporary trials share the consolidated path."""
store = getattr(self, "_trial_database_store", None)
return dict(store.completed) if store is not None else {}
[docs]
def sequential_run(
self,
*,
summarizer: typing.Any = SimulationTables,
fail_fast: bool = True,
rich_progress: Progress | None = None,
run_id: str | None = None,
) -> GenericSimulationTables:
"""Run trials sequentially using the same closed-file handoff as workers.
Parameters
----------
summarizer : type, optional
Summary class used to collect and aggregate results.
fail_fast : bool, default True
Stop after the first failed trial.
rich_progress : rich.progress.Progress, optional
Shared display for trial progress and optional live diagnostics.
run_id : str, optional
Diagnostic display key. Defaults to the configured scenario name,
or a short random id if that is empty, as in :meth:`run`. Use
distinct keys for runs sharing a progress display.
"""
run_start = time.time()
summarizer = check_summarizer(summarizer)
from passengersim.config.storage import resolve_storage, validate_database_reports
validate_database_reports(self.config, resolve_storage(self.config.db, orchestrated=True), summarizer)
store = self._trial_database_store = TrialDatabaseStore(self.config, initializer=self._database_initializer)
results = []
cfg_json = self._dump_config()
paths = [store.consolidated_path, *(store.path_for(tid) for tid in store.expected)]
monitor = _RunMonitor(
self.config.simulation_controls.diagnostics,
_resolve_run_id(run_id, self.config.scenario),
sorted(store.expected),
self.config.simulation_controls.num_samples,
[path for path in paths if path is not None],
rich_progress,
)
status = "failed"
def record_snapshot(snapshot: dict[str, typing.Any]) -> None:
monitor.receive((trial_id, "diagnostics", snapshot))
try:
monitor.start()
for trial_id in range(self.config.simulation_controls.num_trials):
try:
value = _subprocess_run_trial(
trial_id,
cfg_json,
progress_queue=_DirectProgress(monitor.receive) if rich_progress is not None else None,
output_dir=self.output_dir,
summarizer=summarizer,
callbacks=self.callback_functions(),
database_path=store.path_for(trial_id),
diagnostic_callback=record_snapshot
if monitor.log is not None and rich_progress is None
else None,
)
if isinstance(value, Exception):
raise value
monitor.worker_completed(trial_id, value.diagnostics, database=value.database is not None)
store.accept(trial_id, value)
monitor.accepted(trial_id, database=value.database is not None)
results.append(value.summary)
except Exception as error:
store.failed[trial_id] = str(error)
monitor.failed(trial_id, error)
if fail_fast:
store.cancelled.update(store.expected - store.merged - store.failed.keys())
store.finalized = True
raise
store.finalized = True
result = summarizer.aggregate(results)
if self.config.simulation_controls.diagnostics.mode != "off":
result._metadata["diagnostics"] = summary_metadata(monitor.snapshot(), aggregated=True)
self._finalize_trial_summaries(result, self.config, run_start, store)
result._trial_database_store = store
result = finalize_result(result)
status = "failed" if store.failed else "completed"
return result
except BaseException as error:
# fail_fast controls ordinary trial failures, never Ctrl-C or exit
# signals. Preserve accepted results and retire unfinished trials.
status = "cancelled" if isinstance(error, KeyboardInterrupt) else "failed"
store.interrupted = True
store.cancelled.update(store.expected - store.merged - store.failed.keys())
store.finalized = True
raise
finally:
try:
for tid in sorted(store.cancelled):
monitor.cancelled(tid)
finally:
monitor.close(
status,
failed_trials=dict(store.failed),
cancelled_trials=sorted(store.cancelled),
database_timings=store.metrics,
)
def _finalize_trial_summaries(self, result, config, run_start, store):
"""Preserve parent output/report behavior for both trial execution modes."""
result._metadata["database.timings"] = store.metrics
result.config = config.model_copy(deep=True)
result._metadata["time.started"] = datetime.fromtimestamp(run_start, UTC).isoformat()
run_finished = time.time()
result._metadata["time.runtime"] = run_finished - run_start
result._metadata["time.finished"] = datetime.fromtimestamp(run_finished, UTC).isoformat()
# write output files if designated
output_filenames = {}
output_filenames["html"] = config.outputs.get_output_filename(
"html",
timestamp=self._timestamp,
)
output_filenames["disk"] = config.outputs.get_output_filename(
"disk",
timestamp=self._timestamp,
)
if output_filenames["html"]:
result._metadata["outputs.html_filename"] = output_filenames["html"]
if output_filenames["disk"]:
result._metadata["outputs.disk_filename"] = output_filenames["disk"]
if output_filenames["html"]:
try:
result.to_html(output_filenames["html"], cfg=config, make_dirs=True, add_timestamp=False)
except Exception as err:
warnings.warn(f"Error writing HTML file: {err}", stacklevel=2)
if output_filenames["disk"]:
try:
result.to_file(
output_filenames["disk"],
make_dirs=True,
add_timestamp_ext=False,
)
except Exception as err:
warnings.warn(f"Error writing PXSIM file: {err}", stacklevel=2)
if filename := config.outputs.get_output_filename("pickle", timestamp=self._timestamp):
result._metadata["outputs.pickle_filename"] = result.to_pickle(filename)
if filename := config.outputs.get_output_filename("excel", timestamp=self._timestamp):
result.to_xlsx(filename)
return result
@property
def _eng(self) -> SimulationEngine:
if self._placeholder_sim is None:
self._placeholder_sim = Simulation(self.config, self.output_dir)
return self._placeholder_sim
def _dump_config(self) -> str:
"""
Dump the configuration to a JSON string suitable for a subprocess.
License info is added to the config if available. Output file
reporting is removed.
Returns
-------
str
"""
if self.config.raw_license_certificate is None:
try:
from passengersim_license import raw_license_certificate
except ImportError:
raw_license_certificate = None
self.config.raw_license_certificate = raw_license_certificate
return self.config.model_dump_json(exclude={"outputs": {"html", "pickle", "excel", "log_reports", "disk"}})
def _run_asynchronously(
self,
*,
summarizer: type | None = None,
output_dir: pathlib.Path | None = None,
fail_fast: bool = True,
run_id: str | None = None,
jobber: JobExecutor,
rich_progress: Progress | None = None,
) -> concurrent.futures.Future:
"""
Run the simulation using multiple processes.
Returns immediately with a Future that resolves to the aggregated
result once all trials are complete.
Parameters
----------
summarizer : type[SimulationTables]
output_dir : pathlib.Path, optional
The directory to write output files to. If not provided, a temporary
directory will be created.
fail_fast : bool, default True
If True, if any subprocess fails by raising an exception, then cancel
pending tasks and set the exception on the returned Future.
run_id : str, optional
An optional identifier for this run, which will be included in log
messages and progress bars. Set to `True` to generate a random id.
jobber : JobExecutor, required
A mp_executor.JobExecutor to which work will be submitted.
Returns
-------
concurrent.futures.Future[SimulationTables]
A Future that resolves to the aggregated simulation results.
"""
run_start = time.time()
summarizer = check_summarizer(summarizer)
from passengersim.config.storage import resolve_storage, validate_database_reports
validate_database_reports(self.config, resolve_storage(self.config.db, orchestrated=True), summarizer)
run_id = _resolve_run_id(run_id, self.config.scenario)
task_ids = list(range(self.config.simulation_controls.num_trials))
num_samples = self.config.simulation_controls.num_samples
cfg_json = self._dump_config()
serialized_rm_systems = export_registered_rm_systems()
dill_kwargs_blob = dill.dumps(
{
"summarizer": summarizer,
"callbacks": self.callback_functions(),
"exported_rm_systems": serialized_rm_systems,
}
)
# Use a Manager queue so the proxy object is picklable with standard
# pickle (required by ProcessPoolExecutor).
manager = multiprocessing.Manager()
try:
progress_queue = manager.Queue(maxsize=max(1, len(task_ids)))
# Reserve storage only after serialization and manager setup succeed;
# failures before worker submission must not consume a destination.
store = self._trial_database_store = TrialDatabaseStore(self.config, initializer=self._database_initializer)
except BaseException:
manager.shutdown()
raise
store._running = True
# Submit all trials to the executor
trial_futures: dict[int, concurrent.futures.Future] = {}
try:
for task_id in task_ids:
fut = jobber.executor.submit(
_executor_run_trial,
task_id,
cfg_json,
progress_queue,
output_dir,
dill_kwargs_blob,
store.path_for(task_id),
)
store.track(fut)
trial_futures[task_id] = fut
except BaseException:
store.interrupted = True
# Submission itself may fail after earlier workers have started.
# Keep their progress queue and files alive until they finish.
for future in trial_futures.values():
future.cancel()
concurrent.futures.wait(trial_futures.values())
store.wait()
for tid in task_ids:
future = trial_futures.get(tid)
if future is None or future.cancelled():
store.cancelled.add(tid)
continue
try:
value = future.result()
store.accept(tid, value)
except BaseException as error:
store.failed[tid] = str(error)
store.finalized = True
store._running = False
manager.shutdown()
raise
result_future: concurrent.futures.Future = concurrent.futures.Future()
config = self.config
# use provided rich_progress if available, otherwise use the one from the jobber
if rich_progress is None:
rich_progress = jobber.rich_progress
def _monitor_and_finalize():
results = {}
first_error = None
pending = set(task_ids)
monitor: _RunMonitor | None = None
log_status = "failed"
try:
paths = [store.consolidated_path, *(store.path_for(tid) for tid in task_ids)]
monitor = _RunMonitor(
config.simulation_controls.diagnostics,
run_id,
task_ids,
num_samples,
[path for path in paths if path is not None],
rich_progress,
)
monitor.start()
while pending:
if result_future.cancelled():
store.interrupted = True
if result_future.cancelled() or (first_error is not None and fail_fast):
for tid in pending:
trial_futures[tid].cancel()
# Only the monitor merges. Inspect Futures independently of
# queue messages, which may precede actual file publication.
for tid in sorted(pending.copy()):
future = trial_futures[tid]
if not future.done():
continue
pending.remove(tid)
if future.cancelled():
store.cancelled.add(tid)
monitor.cancelled(tid)
continue
try:
value = future.result()
monitor.worker_completed(tid, value.diagnostics, database=value.database is not None)
store.accept(tid, value)
monitor.accepted(tid, database=value.database is not None)
results[tid] = value.summary
except BaseException as error:
store.failed[tid] = str(error)
monitor.failed(tid, error)
if first_error is None:
first_error = error
# Worker traceback notes survive Future transport.
sub_tb = getattr(error, "_passengersim_subprocess_tb", None)
rich_progress.console.print(
f"Exception in Trial {tid}: {sub_tb or error}", style="red", markup=False
)
monitor.refresh()
if pending:
try:
tid, message, payload = progress_queue.get(timeout=0.05)
except queue.Empty:
continue
monitor.receive((tid, message, payload))
store.wait()
store.finalized = True
if result_future.cancelled():
return
if first_error is not None and fail_fast:
result_future.set_exception(first_error)
return
result = summarizer.aggregate([v for k, v in sorted(results.items())])
if config.simulation_controls.diagnostics.mode != "off":
result._metadata["diagnostics"] = summary_metadata(monitor.snapshot(), aggregated=True)
self._finalize_trial_summaries(result, config, run_start, store)
# Publish the owner only after monitor-thread output generation;
# no retained connection may be opened on this thread.
result._trial_database_store = store
log_status = "failed" if store.failed else "cancelled" if store.cancelled else "completed"
result_future.set_result(result)
except BaseException as error:
store.interrupted = True
for tid in pending:
trial_futures[tid].cancel()
if not result_future.done():
result_future.set_exception(error)
finally:
# Cancellation cannot stop a running worker. Keep its queue and
# storage alive until all file users and callbacks finish.
concurrent.futures.wait(trial_futures.values())
store.wait()
try:
# Preserve successful workers even after the normal collection
# loop fails. Database acceptance remains owned by this runner.
def observe_recovery(method: str, *args: typing.Any, **kwargs: typing.Any) -> None:
# A broken output must not stop recovery of other workers'
# files. The original monitor failure is already on the Future.
# Construction may have failed before a monitor existed.
if monitor is None:
return
try:
getattr(monitor, method)(*args, **kwargs)
except BaseException:
logging.getLogger(__name__).warning("Diagnostic recovery failed", exc_info=True)
for tid in sorted(pending):
future = trial_futures[tid]
if future.cancelled():
store.cancelled.add(tid)
observe_recovery("cancelled", tid)
continue
try:
value = future.result()
observe_recovery(
"worker_completed",
tid,
value.diagnostics,
database=value.database is not None,
)
store.accept(tid, value)
except BaseException as error:
store.failed[tid] = str(error)
observe_recovery("failed", tid, error)
else:
observe_recovery("accepted", tid, database=value.database is not None)
finally:
if result_future.cancelled():
store.interrupted = True
store.finalized = True
store._running = False
try:
if monitor is not None:
monitor.close(
"cancelled"
if result_future.cancelled() or (store.cancelled and not store.failed)
else log_status,
failed_trials=dict(store.failed),
cancelled_trials=sorted(store.cancelled),
database_timings=store.metrics,
)
finally:
try:
manager.shutdown()
except Exception:
pass
monitor_thread = threading.Thread(target=_monitor_and_finalize, daemon=True)
result_future._monitor_thread = monitor_thread
monitor_thread.start()
return result_future
[docs]
def log_config(self):
user_name = getpass.getuser()
trials = self.config.simulation_controls.num_trials
samples = self.config.simulation_controls.num_samples
msg = f"user={user_name}, trials={trials}, samples={samples}"
# Send a basic informational log message
try:
import syslog
except ImportError:
logging.getLogger(__name__).info(msg)
else:
syslog.openlog(ident="PassegerSim", facility=syslog.LOG_USER)
syslog.syslog(syslog.LOG_INFO, msg)
syslog.closelog()
[docs]
def run[SimulationTablesT: GenericSimulationTables](
self,
*,
summarizer: type[SimulationTablesT] | SimulationTablesT | None = None,
output_dir: pathlib.Path | None = None,
max_processes: int | None = None,
fail_fast: bool = True,
rich_progress: Progress | None = None,
cache_dir: pathlib.Path | None = None,
run_id: str | None = None,
):
summarizer = check_summarizer(summarizer)
from passengersim.config.storage import resolve_storage, validate_database_reports
validate_database_reports(self.config, resolve_storage(self.config.db, orchestrated=True), summarizer)
# self.log_config()
if cache_dir is not None:
from ._cache_run import cache_run
return cache_run(
self,
cache_dir=cache_dir,
summarizer=summarizer,
output_dir=output_dir,
max_processes=max_processes,
fail_fast=fail_fast,
rich_progress=rich_progress,
run_id=run_id,
)
jobber = JobExecutor(max_workers=max_processes)
with jobber:
future = self._run_asynchronously(
summarizer=summarizer,
output_dir=output_dir,
rich_progress=rich_progress,
run_id=run_id,
fail_fast=fail_fast,
# executor=executor,
jobber=jobber,
)
try:
result = future.result()
future._monitor_thread.join()
return finalize_result(result)
finally:
# Synchronous completion includes releasing worker queues and
# monitor references, not merely publishing the summary.
future._monitor_thread.join()