From 986b7bad42e5937334c6be7b59fa297a92ba77b9 Mon Sep 17 00:00:00 2001 From: Michael Harms Date: Mon, 13 Jul 2026 12:46:44 -0700 Subject: [PATCH 1/2] pin python <3.14 to avoid conda resolution problems on linux --- environment.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/environment.yml b/environment.yml index 372c0ed..0548fc4 100644 --- a/environment.yml +++ b/environment.yml @@ -4,6 +4,7 @@ channels: - bioconda - defaults dependencies: + - python<3.14 - pip - numpy - pandas From 9399f03f02677413406e88ad221066c4c999403f Mon Sep 17 00:00:00 2001 From: Michael Harms Date: Fri, 31 Jul 2026 17:39:42 -0700 Subject: [PATCH 2/2] fixing bug that caused hang with mpi in bootstrap reconcile --- docs/badges/coverage-badge.svg | 2 +- docs/badges/tests-badge.svg | 2 +- src/topiary/_private/threads.py | 6 +- src/topiary/generax/_reconcile_bootstrap.py | 536 +++++++++++++++--- src/topiary/generax/reconcile.py | 10 +- src/topiary/pipeline/bootstrap_reconcile.py | 54 +- .../generax/test__reconcile_bootstrap.py | 452 +++++++++++++++ .../generax/test_reconcile_bootstrap.py | 124 ++-- .../pipeline/test_bootstrap_reconcile.py | 6 +- 9 files changed, 1019 insertions(+), 173 deletions(-) diff --git a/docs/badges/coverage-badge.svg b/docs/badges/coverage-badge.svg index 01e2389..ac480a6 100644 --- a/docs/badges/coverage-badge.svg +++ b/docs/badges/coverage-badge.svg @@ -1 +1 @@ -coverage: 93.12%coverage93.12% \ No newline at end of file +coverage: 93.18%coverage93.18% \ No newline at end of file diff --git a/docs/badges/tests-badge.svg b/docs/badges/tests-badge.svg index 1c47a13..860282d 100644 --- a/docs/badges/tests-badge.svg +++ b/docs/badges/tests-badge.svg @@ -1 +1 @@ -tests: 326tests326 \ No newline at end of file +tests: 338tests338 \ No newline at end of file diff --git a/src/topiary/_private/threads.py b/src/topiary/_private/threads.py index a7fdd96..bf51220 100644 --- a/src/topiary/_private/threads.py +++ b/src/topiary/_private/threads.py @@ -18,8 +18,10 @@ class MockLock(): def __init__(self): pass - def acquire(self): - pass + def acquire(self,blocking=True,timeout=None): + # Mirror the multiprocessing.Lock proxy interface: return True to + # indicate the (non-existent) lock was acquired. + return True def release(self): pass diff --git a/src/topiary/generax/_reconcile_bootstrap.py b/src/topiary/generax/_reconcile_bootstrap.py index 170c2bd..142034f 100644 --- a/src/topiary/generax/_reconcile_bootstrap.py +++ b/src/topiary/generax/_reconcile_bootstrap.py @@ -30,12 +30,233 @@ import tarfile import random import string +import signal import subprocess import time import pathlib import multiprocessing as mp +# Default configuration for per-replicate timeouts and the failure circuit +# breaker. All times are in seconds. These can be overridden by the caller via +# the ``timeout_config`` argument threaded down from the pipeline entry point. +# +# factor : a replicate is killed if it runs longer than +# factor * (longest replicate seen so far). +# ceiling : timeout to use before we have enough completed replicates +# to estimate a runtime (i.e. for the very first block of +# replicates). Also the maximum a first-block replicate is +# allowed to run before we give up on the whole calculation. +# floor : minimum timeout, so that fast replicates are not killed by +# filesystem/scheduler/MPI-startup jitter on a busy cluster. +# max_failed_fraction : abort the whole calculation if more than this fraction +# of replicates fail (after `max_failed_floor` failures). +# max_failed_floor : never trip the circuit breaker until at least this many +# replicates have failed (keeps small runs from aborting on +# one or two transient failures). +_DEFAULT_TIMEOUT_CONFIG = {"factor":3.0, + "ceiling":24*60*60.0, + "floor":300.0, + "max_failed_fraction":0.1, + "max_failed_floor":5} + +# How long (seconds) to wait on the shared multiprocessing lock before treating +# it as orphaned (e.g. a sibling worker was hard-killed while holding it). This +# is generously long relative to the sub-second file operations the lock +# protects, so honest contention never trips it. +_LOCK_TIMEOUT = 300.0 + + +class _LocalValue: + """ + Minimal stand-in for a ``multiprocessing.Manager().Value`` with a ``.value`` + attribute. Used when the calculation runs single-threaded (in-process) and a + real shared value would be needless overhead. + """ + def __init__(self,value=0): + self.value = value + + +def _get_timeout_config(timeout_config): + """ + Merge a (possibly partial or None) timeout_config dict on top of the + defaults. + + Parameters + ---------- + timeout_config : dict or None + user-supplied overrides for `_DEFAULT_TIMEOUT_CONFIG` + + Returns + ------- + config : dict + complete configuration dictionary + """ + + config = dict(_DEFAULT_TIMEOUT_CONFIG) + if timeout_config is not None: + for k in timeout_config: + if k not in _DEFAULT_TIMEOUT_CONFIG: + err = f"\nunrecognized timeout_config key '{k}'. allowed keys:\n" + err += f"{list(_DEFAULT_TIMEOUT_CONFIG.keys())}\n\n" + raise ValueError(err) + config[k] = timeout_config[k] + + return config + + +def _compute_replicate_timeout(durations, + sample_threshold, + config): + """ + Decide how long to allow a single generax replicate to run before killing + it. + + Parameters + ---------- + durations : list + wall-clock run times (seconds) of replicates that have completed + successfully so far. + sample_threshold : int or None + number of completed replicates required before we trust the observed + runtimes and switch from the (long) ceiling to an adaptive timeout. If + None, always use the ceiling. + config : dict + timeout configuration (see `_DEFAULT_TIMEOUT_CONFIG`). + + Returns + ------- + timeout : float + number of seconds to allow the replicate to run. + """ + + # Not enough data yet: fall back to the (long) ceiling. This covers the very + # first block of replicates, where we have no runtime estimate. + if sample_threshold is None or len(durations) < sample_threshold: + return config["ceiling"] + + # Adaptive: some multiple of the slowest replicate observed so far, but never + # less than the floor (protects fast replicates from cluster jitter). + return max(config["floor"], config["factor"] * max(durations)) + + +def _should_abort(num_failed, + num_total, + config): + """ + Decide whether so many replicates have failed that the whole calculation + should be aborted (rather than silently producing supports from a broken + run). + + Parameters + ---------- + num_failed : int + number of replicates that have failed so far. + num_total : int + total number of replicates. + config : dict + timeout configuration (see `_DEFAULT_TIMEOUT_CONFIG`). + + Returns + ------- + abort : bool + whether or not to abort. + """ + + # Never abort until we have accumulated a meaningful number of failures. + if num_failed < config["max_failed_floor"]: + return False + + if num_total is None or num_total <= 0: + return False + + return num_failed > config["max_failed_fraction"] * num_total + + +def _kill_process_group(proc): + """ + Kill an entire process group spawned by `proc`. `proc` must have been + launched with ``start_new_session=True`` so that it (and its mpirun/generax + children) form their own process group. This makes sure a timed-out generax + run does not leave orphaned MPI ranks holding onto the node's slots. + + Parameters + ---------- + proc : subprocess.Popen + process whose group should be killed. + """ + + try: + pgid = os.getpgid(proc.pid) + os.killpg(pgid,signal.SIGKILL) + except (ProcessLookupError,PermissionError,OSError): + # Group is already gone or we cannot signal it; fall back to killing the + # direct child. + try: + proc.kill() + except (ProcessLookupError,OSError): + pass + + +def _launch_replicate(cmd,timeout,stdout_path,stderr_path): + """ + Run a single replicate command as a subprocess, capturing stdout/stderr to + files and enforcing a timeout. + + stdout/stderr are redirected to files rather than captured via pipes. This + avoids the classic MPI deadlock in which mpirun exits but an orphaned rank + keeps the stdout/stderr pipe open, leaving ``subprocess`` blocked forever + waiting for EOF. With file redirection the call only waits on the direct + child (mpirun) exiting. + + Parameters + ---------- + cmd : list + subprocess-style command to run. + timeout : float + number of seconds to allow the command to run before killing it. + stdout_path : str + file to which stdout is written. + stderr_path : str + file to which stderr is written. + + Returns + ------- + returncode : int or None + return code of the process (None if it could not be reaped). + timed_out : bool + whether or not the process was killed because it hit the timeout. + """ + + stdout_f = open(stdout_path,"w") + stderr_f = open(stderr_path,"w") + try: + proc = subprocess.Popen(cmd, + stdout=stdout_f, + stderr=stderr_f, + env=mpi.get_mpi_env(), + start_new_session=True) + finally: + # The child has its own duplicated file descriptors; we can close ours. + stdout_f.close() + stderr_f.close() + + try: + proc.wait(timeout=timeout) + return proc.returncode, False + + except subprocess.TimeoutExpired: + + # Take out the whole process group (mpirun + generax ranks), then reap. + _kill_process_group(proc) + try: + proc.wait(timeout=30) + except subprocess.TimeoutExpired: + pass + + return proc.returncode, True + + def _progress_bar(replicate_dir): """ Check to see how far along the calculation is. @@ -98,8 +319,12 @@ def _check_convergence(replicate_dir, lock = threads.MockLock() # Grab copy of the newick file with lock to avoid collisons with threads - # that might be writing to it. - lock.acquire() + # that might be writing to it. Use a bounded wait so an orphaned lock (e.g. + # a sibling worker hard-killed while holding it) cannot wedge the manager + # thread forever; if we cannot get the lock, simply report not-converged and + # try again after the next replicate finishes. + if not lock.acquire(timeout=_LOCK_TIMEOUT): + return False, None try: shutil.copy(os.path.join(replicate_dir,"bs-trees.newick"),"tmp.newick") finally: @@ -133,31 +358,35 @@ def _check_convergence(replicate_dir, num_running = 0 # Lock to prevent other threads from starting a calculation or updating - # the progress bar while this is running. - lock.acquire() - try: + # the progress bar while this is running. Use a bounded wait so an + # orphaned lock cannot wedge the manager thread; if we cannot get it, + # skip marking (the remaining workers will still terminate naturally as + # they exhaust the directories). + got_lock = lock.acquire(timeout=_LOCK_TIMEOUT) + if got_lock: + try: - # Go through all directories - for d in dirs: + # Go through all directories + for d in dirs: - # Is calculation done? - if os.path.isfile(os.path.join(d,"completed")): - num_completed += 1 + # Is calculation done? + if os.path.isfile(os.path.join(d,"completed")): + num_completed += 1 - # if not... - else: + # if not... + else: - # If calculation running? - if os.path.isfile(os.path.join(d,"running")): - num_running += 1 + # If calculation running? + if os.path.isfile(os.path.join(d,"running")): + num_running += 1 - # if not, write file telling other threads to skip - else: - pathlib.Path(os.path.join(d,"skipped")).touch() + # if not, write file telling other threads to skip + else: + pathlib.Path(os.path.join(d,"skipped")).touch() - # Release lock - finally: - lock.release() + # Release lock + finally: + lock.release() return converged, df @@ -166,7 +395,12 @@ def _generax_thread_function(replicate_dir, converge_cutoff, is_manager, hosts, - lock=None): + lock=None, + durations=None, + fail_count=None, + total_replicates=None, + sample_threshold=None, + timeout_config=None): """ Run a generax calculation in parallel, checking for and avoiding collisions with other workers. @@ -186,10 +420,31 @@ def _generax_thread_function(replicate_dir, --hosts ",".join(hosts) lock : multiprocessing.Manager().Lock() lock to allow multiple threads to access files + durations : list or multiprocessing.Manager().list(), optional + shared list to which the wall-clock runtime of each successful replicate + is appended. Used to set adaptive timeouts. If None, a private list is + used (no cross-worker sharing). + fail_count : multiprocessing.Manager().Value or object with a `.value`, optional + shared counter of failed replicates, used for the failure circuit + breaker. If None, a private counter is used. + total_replicates : int, optional + total number of replicates in the calculation. Used by the failure + circuit breaker. If None, the circuit breaker never trips. + sample_threshold : int, optional + number of successful replicates required before switching from the + ceiling timeout to an adaptive timeout. If None, always use the ceiling. + timeout_config : dict, optional + overrides for `_DEFAULT_TIMEOUT_CONFIG`. """ if lock is None: lock = threads.MockLock() + if durations is None: + durations = [] + if fail_count is None: + fail_count = _LocalValue() + + config = _get_timeout_config(timeout_config) # Change into replicate_directory and get sorted list of bootstrap # replicates. @@ -262,76 +517,113 @@ def _generax_thread_function(replicate_dir, cmd = base_cmd[:] cmd.extend(bash_cmd) - # Launch as a subprocess + # Run the replicate and decide whether it succeeded. `failure_reason` is + # None on success and a human-readable string on any failure. Everything + # here is wrapped so that no error -- expected or not -- can leave the + # directory flagged `running` and wedge the calculation. + failure_reason = None + elapsed = None try: - # Run job. We no longer strip SLURM environment variables, but - # instead rely on MCA flags in the mpirun command to avoid - # rank collisions. - ret = subprocess.run(cmd,capture_output=True,env=mpi.get_mpi_env()) - - # Write stdout and stderr - f = open("stdout.log","w") - f.write(ret.stdout.decode()) - f.close() - - f = open("stderr.log","w") - f.write(ret.stderr.decode()) - f.close() - - # If failure (non-zero return code), check if result tree was actually - # generated. MPI can sometimes throw a warning and a non-zero exit code - # even if the underlying program finished successfully. - if ret.returncode != 0: - if not os.path.isfile(result_tree): - err = f"\ngenerax crashed in directory {d}. Writing stderr and\n" - err += "stdout there.\n\n" - raise RuntimeError(err) - - # Grab result tree - f = open(result_tree,'r') - tree = f.read().strip() - f.close() - - # Update status bar and final tree file - lock.acquire() - try: - # Update replicates tree file - f = open(os.path.join("..","bs-trees.newick"),"a") - f.write(f"{tree}\n") + # Decide how long to let this replicate run. Snapshot the durations + # seen so far (list() works for a plain list or a manager list proxy). + timeout = _compute_replicate_timeout(list(durations), + sample_threshold, + config) + + # Launch the replicate. Output is redirected to files (rather than + # captured via pipes) to avoid the MPI deadlock where mpirun exits but + # an orphaned rank keeps the stdout/stderr pipe open. The timeout + # guarantees we can never wedge on a single hung replicate. + start_time = time.time() + returncode, timed_out = _launch_replicate(cmd, + timeout=timeout, + stdout_path="stdout.log", + stderr_path="stderr.log") + elapsed = time.time() - start_time + + # MPI can throw a non-zero exit code even when generax finished fine, + # so the presence of the result tree is the real arbiter of success. + if timed_out: + failure_reason = (f"replicate exceeded its timeout of " + f"{timeout:.0f} s and was killed.") + elif not os.path.isfile(result_tree): + failure_reason = (f"generax exited with code {returncode} and " + f"did not produce a result tree.") + else: + + # Success: read the tree and append it to the shared bs-trees + # file under the lock. Use a bounded wait so an orphaned lock + # cannot wedge this worker; if we cannot get it, fail this + # replicate rather than block forever. + f = open(result_tree,'r') + tree = f.read().strip() f.close() - finally: - lock.release() + if lock.acquire(timeout=_LOCK_TIMEOUT): + try: + f = open(os.path.join("..","bs-trees.newick"),"a") + f.write(f"{tree}\n") + f.close() + finally: + lock.release() + else: + failure_reason = ("could not acquire the shared lock to " + "record the result (a sibling worker may " + "have died holding it).") - # Create a completed file, nuke running file, move to next directory. + except Exception as e: + failure_reason = f"unexpected error running replicate: {e}" + + # Success: record the runtime and flag the directory complete. + if failure_reason is None: + + durations.append(elapsed) pathlib.Path("completed").touch() - os.remove("running") + if os.path.exists("running"): + os.remove("running") + os.chdir("..") - except Exception as e: - - # If we were in the directory, make sure we get out even if we crash - if os.path.split(os.getcwd())[-1] == d: - - # If we have a stderr.log, it might already have info. If not, - # write the exception. - if not os.path.isfile("stderr.log"): - f = open("stderr.log","w") - f.write(str(e)) - f.close() - - pathlib.Path("failed").touch() - if os.path.exists("running"): - os.remove("running") - - os.chdir("..") - - w = f"\nWARNING: bootstrap replicate {d} failed. Check\n" - w += f"{os.path.abspath(os.path.join(d,'stderr.log'))} for details.\n" + # Failure (crash, timeout, missing tree, or lost lock): annotate the + # directory, drop the claim, and move on. We are still inside `d` here. + else: + + try: + with open("stderr.log","a") as f: + f.write(f"\ntopiary: {failure_reason}\n") + except OSError: + pass + + pathlib.Path("failed").touch() + if os.path.exists("running"): + os.remove("running") + os.chdir("..") + + # Track the failure and, if too many replicates have failed, abort + # the whole calculation rather than produce garbage supports. + if lock.acquire(timeout=_LOCK_TIMEOUT): + try: + fail_count.value += 1 + num_failed = fail_count.value + finally: + lock.release() + else: + num_failed = fail_count.value + + w = f"\nWARNING: bootstrap replicate {d} failed ({failure_reason})\n" + w += f"Check {os.path.abspath(os.path.join(d,'stderr.log'))} for details.\n" print(w, flush=True) - + + if _should_abort(num_failed,total_replicates,config): + err = f"\n{num_failed} bootstrap replicates have failed, which\n" + err += "exceeds the allowed failure fraction " + err += f"({config['max_failed_fraction']}). Aborting. This usually\n" + err += "indicates a systemic problem (a bad node, an MPI\n" + err += "misconfiguration, or an input problem) rather than\n" + err += "isolated replicate failures.\n\n" + raise RuntimeError(err) + continue - os.chdir("..") # For the manager thread, check for convergence if is_manager: @@ -509,7 +801,11 @@ def _clean_replicate_dir(replicate_dir): def _construct_args(replicate_dir, converge_cutoff, num_threads, - threads_per_rep): + threads_per_rep, + durations=None, + fail_count=None, + total_replicates=None, + timeout_config=None): """ Construct a list of arguments to pass to each thread in the pool. @@ -523,6 +819,14 @@ def _construct_args(replicate_dir, total number of mpi slots to use for the calculation threads_per_rep : int number of slots to use per replicate. + durations : list or multiprocessing.Manager().list(), optional + shared list of successful-replicate runtimes (for adaptive timeouts). + fail_count : object with a `.value` attribute, optional + shared counter of failed replicates (for the circuit breaker). + total_replicates : int, optional + total number of replicate directories (for the circuit breaker). + timeout_config : dict, optional + overrides for `_DEFAULT_TIMEOUT_CONFIG`. Returns ------- @@ -558,13 +862,33 @@ def _construct_args(replicate_dir, "hosts":this_hosts, "converge_cutoff":converge_cutoff}) - return kwargs_list, len(kwargs_list) + num_workers = len(kwargs_list) + + # Only start trusting observed runtimes (and switch from the ceiling to an + # adaptive timeout) once a full first block of workers has each completed at + # least one replicate. Cap by the total number of replicates so tiny runs + # still eventually adapt. + if total_replicates is not None: + sample_threshold = min(num_workers,total_replicates) + else: + sample_threshold = num_workers + + # Inject shared state / timeout configuration into every worker's kwargs. + for kwargs in kwargs_list: + kwargs["durations"] = durations + kwargs["fail_count"] = fail_count + kwargs["total_replicates"] = total_replicates + kwargs["sample_threshold"] = sample_threshold + kwargs["timeout_config"] = timeout_config + + return kwargs_list, num_workers def _run_bootstrap_calculations(replicate_dir, converge_cutoff, num_threads, - threads_per_rep): + threads_per_rep, + timeout_config=None): """ Run generax in parallel using mpirun for all directories. @@ -578,10 +902,29 @@ def _run_bootstrap_calculations(replicate_dir, number of parallel jobs to start threads_per_rep : int number of threads to use per replicate. only used if bootstrap = True + timeout_config : dict, optional + overrides for `_DEFAULT_TIMEOUT_CONFIG` (per-replicate timeout and + failure circuit breaker). """ print("\nGenerating reconciliation bootstraps.\n",flush=True) + # Total number of replicate directories (used by the failure circuit breaker) + total_replicates = len(glob.glob(os.path.join(replicate_dir,"0*"))) + + # Shared state for adaptive timeouts and the failure circuit breaker. When + # single-threaded, the workers run in-process, so plain objects suffice and + # we avoid spinning up a manager. When multi-threaded, use manager proxies + # so the separate worker processes share the same state. + manager = None + if num_threads == 1: + durations = [] + fail_count = _LocalValue(0) + else: + manager = mp.Manager() + durations = manager.list() + fail_count = manager.Value("i",0) + # This is a status bar that we spawn on its own thread that will spew onto # stderr as the calculations are completed in each directory. status_bar = mp.Process(target=_progress_bar,args=(replicate_dir,)) @@ -599,7 +942,11 @@ def _run_bootstrap_calculations(replicate_dir, kwargs_list, num_threads = _construct_args(replicate_dir, converge_cutoff, num_threads, - threads_per_rep) + threads_per_rep, + durations=durations, + fail_count=fail_count, + total_replicates=total_replicates, + timeout_config=timeout_config) # Launch calculation. try: @@ -613,6 +960,8 @@ def _run_bootstrap_calculations(replicate_dir, # calculation crashes. except Exception as e: status_bar.kill() + if manager is not None: + manager.shutdown() raise e # If we get here, the job is done whether the status bar is or not. Wait @@ -634,6 +983,9 @@ def _run_bootstrap_calculations(replicate_dir, converged, df = _check_convergence(replicate_dir=replicate_dir, converge_cutoff=converge_cutoff) + if manager is not None: + manager.shutdown() + return converged, df @run_cleanly @@ -652,7 +1004,8 @@ def reconcile_bootstrap(df, num_threads, threads_per_rep, generax_binary, - raxml_binary): + raxml_binary, + timeout_config=None): """ Reconcile gene and species trees using generax with bootstrap replicates of the gene tree and alignments. @@ -702,6 +1055,10 @@ def reconcile_bootstrap(df, number of threads to use per replicate. generax_binary : str, optional what generax binary to use + timeout_config : dict, optional + overrides for `_DEFAULT_TIMEOUT_CONFIG`, controlling the per-replicate + timeout (keys "factor", "ceiling", "floor") and the failure circuit + breaker (keys "max_failed_fraction", "max_failed_floor"). Returns ------- @@ -751,7 +1108,8 @@ def reconcile_bootstrap(df, converged, df = _run_bootstrap_calculations(replicate_dir, converge_cutoff, num_threads, - threads_per_rep) + threads_per_rep, + timeout_config=timeout_config) # Write convergence report and whether this converged or not df.to_csv("bootstrap-convergence-report.csv") diff --git a/src/topiary/generax/reconcile.py b/src/topiary/generax/reconcile.py index bfec20f..63d7ab8 100644 --- a/src/topiary/generax/reconcile.py +++ b/src/topiary/generax/reconcile.py @@ -32,7 +32,8 @@ def reconcile(prev_calculation=None, num_threads=-1, threads_per_rep=1, generax_binary=GENERAX_BINARY, - raxml_binary=RAXML_BINARY): + raxml_binary=RAXML_BINARY, + timeout_config=None): """ Reconcile the gene tree to the species tree using generax. @@ -93,6 +94,10 @@ def reconcile(prev_calculation=None, what generax binary to use raxml_binary : str, optional what raxml binary to use + timeout_config : dict, optional + overrides for the per-replicate timeout / failure circuit breaker used + during bootstrap reconciliation (only used if bootstrap = True). See + `topiary.generax._reconcile_bootstrap._DEFAULT_TIMEOUT_CONFIG`. Returns ------- @@ -225,4 +230,5 @@ def reconcile(prev_calculation=None, num_threads=num_threads, threads_per_rep=threads_per_rep, generax_binary=generax_binary, - raxml_binary=raxml_binary) + raxml_binary=raxml_binary, + timeout_config=timeout_config) diff --git a/src/topiary/pipeline/bootstrap_reconcile.py b/src/topiary/pipeline/bootstrap_reconcile.py index e228827..a290b02 100644 --- a/src/topiary/pipeline/bootstrap_reconcile.py +++ b/src/topiary/pipeline/bootstrap_reconcile.py @@ -26,6 +26,10 @@ def bootstrap_reconcile(previous_run_dir, num_threads=None, threads_per_replicate=None, converge_cutoff=0.03, + replicate_timeout_factor=3.0, + replicate_max_hours=24.0, + replicate_min_seconds=300.0, + max_failed_fraction=0.1, restart=False, overwrite=False, raxml_binary=RAXML_BINARY, @@ -51,8 +55,26 @@ def bootstrap_reconcile(previous_run_dir, the number of slots on each compute node to avoid wasting slots. If you have 24 slots per node, you could choose 2, 3, 4, 6, 8, 12, or 24. converge_cutoff : float, default=0.03 - bootstrap convergence criterion. This is RAxML-NG default, passed + bootstrap convergence criterion. This is RAxML-NG default, passed to --bs-cutoff. + replicate_timeout_factor : float, default=3.0 + a bootstrap replicate is killed (and dropped) if it runs longer than + this factor times the longest replicate observed so far. This is what + prevents a single hung generax/MPI replicate from wedging the whole + calculation. + replicate_max_hours : float, default=24.0 + maximum time (hours) a replicate may run before we have enough completed + replicates to estimate a runtime. Also the longest a first-block + replicate is allowed to run before the whole calculation is aborted with + an error. + replicate_min_seconds : float, default=300.0 + minimum per-replicate timeout (seconds), so fast replicates are not + killed by filesystem/scheduler/MPI-startup jitter on a busy cluster. + max_failed_fraction : float, default=0.1 + abort the whole calculation if more than this fraction of replicates + fail. This catches systemic problems (bad node, MPI misconfiguration) + rather than letting the run silently produce supports from a broken + calculation. restart : bool, default=False restart job from where it stopped in output directory. incompatible with overwrite @@ -120,6 +142,30 @@ def bootstrap_reconcile(previous_run_dir, err = "overwrite and restart flags are incompatible.\n" raise ValueError(err) + # -------------------------------------------------------------------------- + # Assemble per-replicate timeout / failure circuit-breaker configuration + + replicate_timeout_factor = check.check_float(replicate_timeout_factor, + "replicate_timeout_factor", + minimum_allowed=1.0) + replicate_max_hours = check.check_float(replicate_max_hours, + "replicate_max_hours", + minimum_allowed=0, + minimum_inclusive=False) + replicate_min_seconds = check.check_float(replicate_min_seconds, + "replicate_min_seconds", + minimum_allowed=0, + minimum_inclusive=False) + max_failed_fraction = check.check_float(max_failed_fraction, + "max_failed_fraction", + minimum_allowed=0, + maximum_allowed=1) + + timeout_config = {"factor":replicate_timeout_factor, + "ceiling":replicate_max_hours*60*60, + "floor":replicate_min_seconds, + "max_failed_fraction":max_failed_fraction} + # -------------------------------------------------------------------------- # Validate software stack required for this pipeline @@ -282,7 +328,8 @@ def bootstrap_reconcile(previous_run_dir, num_threads=num_threads, threads_per_rep=threads_per_replicate, generax_binary=generax_binary, - raxml_binary=raxml_binary) + raxml_binary=raxml_binary, + timeout_config=timeout_config) else: @@ -294,7 +341,8 @@ def bootstrap_reconcile(previous_run_dir, num_threads=num_threads, threads_per_rep=threads_per_replicate, raxml_binary=raxml_binary, - generax_binary=generax_binary) + generax_binary=generax_binary, + timeout_config=timeout_config) os.chdir('..') diff --git a/tests/topiary/generax/test__reconcile_bootstrap.py b/tests/topiary/generax/test__reconcile_bootstrap.py index 0dfc95b..eb8cd0d 100644 --- a/tests/topiary/generax/test__reconcile_bootstrap.py +++ b/tests/topiary/generax/test__reconcile_bootstrap.py @@ -1,6 +1,7 @@ import pytest import topiary +import topiary.generax._reconcile_bootstrap as _rb from topiary.generax._reconcile_bootstrap import _progress_bar from topiary.generax._reconcile_bootstrap import _check_convergence from topiary.generax._reconcile_bootstrap import _generax_thread_function @@ -9,20 +10,31 @@ from topiary.generax._reconcile_bootstrap import _construct_args from topiary.generax._reconcile_bootstrap import _run_bootstrap_calculations from topiary.generax._reconcile_bootstrap import reconcile_bootstrap +from topiary.generax._reconcile_bootstrap import _LocalValue +from topiary.generax._reconcile_bootstrap import _get_timeout_config +from topiary.generax._reconcile_bootstrap import _compute_replicate_timeout +from topiary.generax._reconcile_bootstrap import _should_abort +from topiary.generax._reconcile_bootstrap import _kill_process_group +from topiary.generax._reconcile_bootstrap import _launch_replicate +from topiary.generax._reconcile_bootstrap import _DEFAULT_TIMEOUT_CONFIG from topiary.generax._generax import GENERAX_BINARY from topiary.raxml import RAXML_BINARY from topiary._private import Supervisor from topiary._private import mpi +from topiary._private.threads import MockLock import ete4 as ete import pandas as pd import os +import sys import glob import shutil import copy import pathlib +import signal +import subprocess import time import multiprocessing as mp @@ -714,3 +726,443 @@ def test_reconcile_bootstrap(small_phylo,tmpdir): print(n.support) os.chdir(current_dir) + + +# ----------------------------------------------------------------------------- +# Tests for the per-replicate timeout / failure-handling machinery. +# ----------------------------------------------------------------------------- + +def test_mocklock_acquire(): + + # MockLock.acquire should mirror the multiprocessing lock proxy interface: + # accept blocking/timeout and return True. + lock = MockLock() + assert lock.acquire() is True + assert lock.acquire(timeout=5) is True + assert lock.acquire(blocking=False) is True + assert lock.acquire(True,10) is True + assert lock.release() is None + + +def test__LocalValue(): + + v = _LocalValue() + assert v.value == 0 + + v = _LocalValue(7) + assert v.value == 7 + + v.value += 3 + assert v.value == 10 + + +def test__get_timeout_config(): + + # None -> a copy of the defaults (not the same object) + config = _get_timeout_config(None) + assert config == _DEFAULT_TIMEOUT_CONFIG + assert config is not _DEFAULT_TIMEOUT_CONFIG + + # Partial override merges on top of defaults + config = _get_timeout_config({"factor":10.0}) + assert config["factor"] == 10.0 + assert config["ceiling"] == _DEFAULT_TIMEOUT_CONFIG["ceiling"] + assert config["floor"] == _DEFAULT_TIMEOUT_CONFIG["floor"] + + # Full override + override = {"factor":2.0, + "ceiling":10.0, + "floor":1.0, + "max_failed_fraction":0.5, + "max_failed_floor":1} + config = _get_timeout_config(override) + assert config == override + + # Unrecognized key raises + with pytest.raises(ValueError): + _get_timeout_config({"not_a_key":1}) + + +def test__compute_replicate_timeout(): + + config = {"factor":3.0,"ceiling":1000.0,"floor":10.0, + "max_failed_fraction":0.1,"max_failed_floor":5} + + # No sample threshold -> always the ceiling, even with data + assert _compute_replicate_timeout([],None,config) == 1000.0 + assert _compute_replicate_timeout([1,2,3],None,config) == 1000.0 + + # Not enough samples yet -> ceiling + assert _compute_replicate_timeout([],3,config) == 1000.0 + assert _compute_replicate_timeout([5.0,5.0],3,config) == 1000.0 + + # Enough samples, factor dominates the floor + # max = 20, factor*max = 60 > floor (10) -> 60 + assert _compute_replicate_timeout([10.0,20.0,5.0],3,config) == 60.0 + + # Enough samples, floor dominates + # max = 1, factor*max = 3 < floor (10) -> 10 + assert _compute_replicate_timeout([1.0,1.0,1.0],3,config) == 10.0 + + +def test__should_abort(): + + config = {"factor":3.0,"ceiling":1000.0,"floor":10.0, + "max_failed_fraction":0.1,"max_failed_floor":5} + + # Below the failure floor -> never abort, regardless of fraction + assert _should_abort(0,100,config) is False + assert _should_abort(4,10,config) is False + + # num_total is None or non-positive -> never abort + assert _should_abort(100,None,config) is False + assert _should_abort(100,0,config) is False + + # At/above floor and above the fraction -> abort + # 6 > 0.1 * 10 = 1.0 -> True + assert _should_abort(6,10,config) is True + + # At/above floor but not above the fraction -> do not abort + # 6 failures out of 100 -> 6 > 10.0 is False + assert _should_abort(6,100,config) is False + + # Exactly equal to the fraction is not "more than" -> do not abort + # 10 == 0.1 * 100 -> not > -> False + assert _should_abort(10,100,config) is False + + +def test__kill_process_group(tmpdir,monkeypatch): + + monkeypatch.chdir(tmpdir) + + # Launch a shell that spawns a sleeping grandchild, all in a new session so + # they form their own process group. + proc = subprocess.Popen(["sh","-c","sleep 60"],start_new_session=True) + + # Let the group come up and grab its id. + time.sleep(0.3) + pgid = os.getpgid(proc.pid) + + # signal 0 -> group currently exists + os.killpg(pgid,0) + + _kill_process_group(proc) + proc.wait(timeout=10) + + # The direct child is dead + assert proc.poll() is not None + + # The whole group (including the sleep grandchild) is gone + time.sleep(0.3) + with pytest.raises(ProcessLookupError): + os.killpg(pgid,0) + + +def test__launch_replicate(tmpdir,monkeypatch): + + monkeypatch.chdir(tmpdir) + + # ------------------------------------------------------------------------- + # Normal completion: stdout/stderr captured to files, returncode reported, + # not timed out. + + cmd = [sys.executable,"-c", + "import sys; sys.stdout.write('hello out'); sys.stderr.write('hello err')"] + returncode, timed_out = _launch_replicate(cmd, + timeout=30, + stdout_path="stdout.log", + stderr_path="stderr.log") + assert returncode == 0 + assert timed_out is False + + with open("stdout.log") as f: + assert "hello out" in f.read() + with open("stderr.log") as f: + assert "hello err" in f.read() + + # ------------------------------------------------------------------------- + # Non-zero exit code is reported (but not treated as a timeout) + + cmd = [sys.executable,"-c","import sys; sys.exit(3)"] + returncode, timed_out = _launch_replicate(cmd, + timeout=30, + stdout_path="stdout2.log", + stderr_path="stderr2.log") + assert returncode == 3 + assert timed_out is False + + # ------------------------------------------------------------------------- + # Timeout: a long-running process is killed and flagged as timed out. The + # call must return well before the process would have finished on its own. + + cmd = [sys.executable,"-c","import time; time.sleep(60)"] + start = time.time() + returncode, timed_out = _launch_replicate(cmd, + timeout=0.5, + stdout_path="stdout3.log", + stderr_path="stderr3.log") + elapsed = time.time() - start + assert timed_out is True + assert elapsed < 30 + + +def test__launch_replicate_env(tmpdir,monkeypatch): + + monkeypatch.chdir(tmpdir) + + # The environment handed to the subprocess must come from mpi.get_mpi_env. + monkeypatch.setattr(_rb.mpi,"get_mpi_env", + lambda: {"TOPIARY_MARKER":"present"}) + + cmd = [sys.executable,"-c", + "import os; print(os.environ.get('TOPIARY_MARKER','missing'))"] + returncode, timed_out = _launch_replicate(cmd, + timeout=30, + stdout_path="stdout.log", + stderr_path="stderr.log") + assert returncode == 0 + assert timed_out is False + with open("stdout.log") as f: + assert "present" in f.read() + + +def _build_fake_replicate_dir(repdir,names): + """ + Build a minimal replicate directory tree with `run_generax.sh` files for + each replicate in `names`. + """ + + os.mkdir(repdir) + for name in names: + d = os.path.join(repdir,name) + os.mkdir(d) + with open(os.path.join(d,"run_generax.sh"),"w") as f: + f.write("generax --families control.txt &> topiary.log\n") + + +def _make_fake_launch(plan): + """ + Build a fake `_launch_replicate` that behaves according to `plan`, a dict + mapping replicate directory name -> one of "success", "timeout", "notree". + """ + + result_tree = os.path.join("result","results","reconcile","geneTree.newick") + + def fake_launch(cmd,timeout,stdout_path,stderr_path): + + # We are inside the replicate directory when this is called. + with open(stdout_path,"w") as f: + f.write("fake stdout\n") + with open(stderr_path,"w") as f: + f.write("fake stderr\n") + + outcome = plan[os.path.basename(os.getcwd())] + + if outcome == "success": + os.makedirs(os.path.dirname(result_tree),exist_ok=True) + with open(result_tree,"w") as f: + f.write("(A:1,B:1);\n") + return 0, False + + if outcome == "timeout": + return None, True + + if outcome == "raise": + raise RuntimeError("boom") + + # "notree" -- exited (non-zero) without producing a result tree + return 1, False + + return fake_launch + + +def test__generax_thread_function_failure_handling(tmpdir,monkeypatch): + + monkeypatch.chdir(tmpdir) + + names = ["00001","00002","00003","00004","00005"] + _build_fake_replicate_dir("replicates",names) + + # 00002 times out, 00003 produces no tree; the rest succeed. + plan = {"00001":"success", + "00002":"timeout", + "00003":"notree", + "00004":"success", + "00005":"success"} + monkeypatch.setattr(_rb,"_launch_replicate",_make_fake_launch(plan)) + + durations = [] + fail_count = _LocalValue(0) + + out = _generax_thread_function(replicate_dir="replicates", + converge_cutoff=0.5, + is_manager=False, + hosts=["localhost"], + lock=None, + durations=durations, + fail_count=fail_count, + total_replicates=len(names), + sample_threshold=None, + timeout_config=None) + assert out is None + + # Successful replicates are marked completed (not failed) + for name in ["00001","00004","00005"]: + assert os.path.isfile(os.path.join("replicates",name,"completed")) + assert not os.path.isfile(os.path.join("replicates",name,"failed")) + assert not os.path.isfile(os.path.join("replicates",name,"running")) + + # Failed replicates (timeout + missing tree) are marked failed (not completed) + for name in ["00002","00003"]: + assert os.path.isfile(os.path.join("replicates",name,"failed")) + assert not os.path.isfile(os.path.join("replicates",name,"completed")) + assert not os.path.isfile(os.path.join("replicates",name,"running")) + + # Runtime recorded only for the three successes + assert len(durations) == 3 + + # Failure counter incremented for the two failures + assert fail_count.value == 2 + + # bs-trees.newick has exactly the three successful trees + with open(os.path.join("replicates","bs-trees.newick")) as f: + lines = [line for line in f if line.strip() != ""] + assert len(lines) == 3 + + # Failure reason annotated into the stderr log + with open(os.path.join("replicates","00002","stderr.log")) as f: + assert "timeout" in f.read() + with open(os.path.join("replicates","00003","stderr.log")) as f: + assert "result tree" in f.read() + + +def test__generax_thread_function_unexpected_exception(tmpdir,monkeypatch): + + # An unexpected exception inside the launch must be caught and turned into a + # normal replicate failure (directory flagged `failed`, not left `running`), + # so it can never wedge the whole calculation. + + monkeypatch.chdir(tmpdir) + + names = ["00001","00002"] + _build_fake_replicate_dir("replicates",names) + + plan = {"00001":"raise","00002":"success"} + monkeypatch.setattr(_rb,"_launch_replicate",_make_fake_launch(plan)) + + durations = [] + fail_count = _LocalValue(0) + + out = _generax_thread_function(replicate_dir="replicates", + converge_cutoff=0.5, + is_manager=False, + hosts=["localhost"], + lock=None, + durations=durations, + fail_count=fail_count, + total_replicates=len(names), + sample_threshold=None, + timeout_config=None) + assert out is None + + # The exception replicate is failed (and not stuck running) + assert os.path.isfile(os.path.join("replicates","00001","failed")) + assert not os.path.isfile(os.path.join("replicates","00001","running")) + assert not os.path.isfile(os.path.join("replicates","00001","completed")) + + # The following replicate still ran to completion + assert os.path.isfile(os.path.join("replicates","00002","completed")) + + assert fail_count.value == 1 + with open(os.path.join("replicates","00001","stderr.log")) as f: + assert "unexpected error" in f.read() + + +def test__generax_thread_function_circuit_breaker(tmpdir,monkeypatch): + + monkeypatch.chdir(tmpdir) + + names = ["00001","00002","00003","00004","00005"] + _build_fake_replicate_dir("replicates",names) + repdir = os.path.abspath("replicates") + + # Everything fails. + plan = {name:"notree" for name in names} + monkeypatch.setattr(_rb,"_launch_replicate",_make_fake_launch(plan)) + + durations = [] + fail_count = _LocalValue(0) + + # floor of 2 failures, 10% fraction of 5 replicates -> abort on the 2nd + # failure (2 >= 2 and 2 > 0.5). + timeout_config = {"max_failed_floor":2,"max_failed_fraction":0.1} + + with pytest.raises(RuntimeError): + _generax_thread_function(replicate_dir="replicates", + converge_cutoff=0.5, + is_manager=False, + hosts=["localhost"], + lock=None, + durations=durations, + fail_count=fail_count, + total_replicates=len(names), + sample_threshold=None, + timeout_config=timeout_config) + + # Aborting mid-run can leave the working directory changed; restore it so + # relative-path bookkeeping in later tests is unaffected. + monkeypatch.chdir(tmpdir) + + # Aborted after the second failure; later replicates never ran. + assert fail_count.value == 2 + assert os.path.isfile(os.path.join(repdir,"00001","failed")) + assert os.path.isfile(os.path.join(repdir,"00002","failed")) + assert not os.path.isfile(os.path.join(repdir,"00003","failed")) + + +def test__generax_thread_function_adaptive_timeout(tmpdir,monkeypatch): + + monkeypatch.chdir(tmpdir) + + names = ["00001","00002","00003"] + _build_fake_replicate_dir("replicates",names) + + plan = {name:"success" for name in names} + + # Record the timeout handed to each replicate so we can verify the adaptive + # behavior. The first replicate should get the (long) ceiling; once we have + # a sample, subsequent replicates get factor*max(durations), floored. + seen_timeouts = [] + base_fake = _make_fake_launch(plan) + + def recording_fake(cmd,timeout,stdout_path,stderr_path): + seen_timeouts.append(timeout) + return base_fake(cmd,timeout,stdout_path,stderr_path) + + monkeypatch.setattr(_rb,"_launch_replicate",recording_fake) + + durations = [] + fail_count = _LocalValue(0) + timeout_config = {"factor":3.0,"ceiling":9999.0,"floor":1.0, + "max_failed_fraction":0.1,"max_failed_floor":5} + + _generax_thread_function(replicate_dir="replicates", + converge_cutoff=0.5, + is_manager=False, + hosts=["localhost"], + lock=None, + durations=durations, + fail_count=fail_count, + total_replicates=len(names), + sample_threshold=1, + timeout_config=timeout_config) + + # First replicate: no samples yet -> ceiling. + assert seen_timeouts[0] == 9999.0 + + # Subsequent replicates: adaptive. With a sample_threshold of 1, once one + # replicate has completed we switch to factor*max(durations), floored at 1. + assert len(seen_timeouts) == 3 + for t in seen_timeouts[1:]: + assert t != 9999.0 + assert t >= 1.0 diff --git a/tests/topiary/generax/test_reconcile_bootstrap.py b/tests/topiary/generax/test_reconcile_bootstrap.py index c7053cd..7a516b1 100644 --- a/tests/topiary/generax/test_reconcile_bootstrap.py +++ b/tests/topiary/generax/test_reconcile_bootstrap.py @@ -1,6 +1,5 @@ import pytest import topiary.generax._reconcile_bootstrap as rb -from unittest.mock import MagicMock import os def test__construct_args(mocker): @@ -38,96 +37,75 @@ def test__construct_args(mocker): assert kwargs_list[0]["hosts"] == ["n1", "n1", "n2"] assert kwargs_list[1]["hosts"] == ["n2"] -def test__generax_thread_function(mocker): - # Mock subprocess.run - mock_run = mocker.patch("topiary.generax._reconcile_bootstrap.subprocess.run") - mock_run.return_value = MagicMock(returncode=0, stdout=b"out", stderr=b"err") - - # Mock other things to avoid file IO - mocker.patch("os.chdir") - mocker.patch("os.listdir", return_value=["00001"]) - mocker.patch("os.path.isdir", return_value=True) - mocker.patch("os.path.isfile", side_effect=[False, False, False, False, True, True]) # completed, running, skipped, failed, run_generax.sh, result_tree - mocker.patch("pathlib.Path.touch") - mocker.patch("os.remove") - - # Define a custom side-effect for open to handle different files - def side_effect_open(filename, mode='r'): - mock = MagicMock() - if 'r' in mode: - if "run_generax.sh" in filename: - mock.__enter__.return_value.read.return_value = "generax --args\n" - elif "geneTree.newick" in filename: - mock.__enter__.return_value.read.return_value = "((a,b),c);" - else: - mock.__enter__.return_value.read.return_value = "" - return mock +def test__generax_thread_function(mocker, tmpdir): + # The subprocess launch now happens inside _launch_replicate; mock that seam + # and verify the mpirun command is constructed with the right hosts. - mocker.patch("builtins.open", side_effect=side_effect_open) - - # Mock mpi._get_mpi_oversubscribe - mocker.patch("topiary.generax._reconcile_bootstrap.mpi._get_mpi_oversubscribe", return_value=False) - mock_get_mpi_env = mocker.patch("topiary.generax._reconcile_bootstrap.mpi.get_mpi_env", return_value={"STAY": "STAY"}) + rep_dir = os.path.join(tmpdir, "replicates") + os.makedirs(os.path.join(rep_dir, "00001")) + with open(os.path.join(rep_dir, "00001", "run_generax.sh"), "w") as f: + f.write("generax --args &> topiary.log\n") - # Mock lock - mock_lock = MagicMock() - - # Run thread function - rb._generax_thread_function("replicate_dir", 0.03, False, ["n1", "n1"], lock=mock_lock) - - # Check if mpirun was called with correct hosts - found_mpirun = False - for call_args in mock_run.call_args_list: - args = call_args[0][0] - if "mpirun" in args: - found_mpirun = True - assert "--host" in args - assert "n1,n1" in args - # Check if env was passed correctly - assert call_args[1]["env"] == {"STAY": "STAY"} - - assert found_mpirun + mocker.patch("topiary.generax._reconcile_bootstrap.mpi._get_mpi_oversubscribe", + return_value=False) + mocker.patch("topiary.generax._reconcile_bootstrap.mpi.get_mpi_flags", + return_value=[]) + + captured = {} + def fake_launch(cmd, timeout, stdout_path, stderr_path): + captured["cmd"] = cmd + # Create a result tree so the replicate "succeeds". + os.makedirs(os.path.join("result", "results", "reconcile"), exist_ok=True) + with open(os.path.join("result", "results", "reconcile", "geneTree.newick"), "w") as f: + f.write("((a,b),c);\n") + return 0, False + + mocker.patch("topiary.generax._reconcile_bootstrap._launch_replicate", + side_effect=fake_launch) + + original_dir = os.getcwd() + try: + rb._generax_thread_function(rep_dir, 0.03, False, ["n1", "n1"], lock=None) + finally: + os.chdir(original_dir) + + # mpirun command constructed with the correct hosts + cmd = captured["cmd"] + assert "mpirun" in cmd + assert "--host" in cmd + assert "n1,n1" in cmd + + # Replicate ran to completion + assert os.path.isfile(os.path.join(rep_dir, "00001", "completed")) + assert not os.path.exists(os.path.join(rep_dir, "00001", "running")) def test_generax_thread_function_failure(mocker, tmpdir): - # Mock subprocess.run to fail - mock_run = mocker.patch("topiary.generax._reconcile_bootstrap.subprocess.run", - side_effect=RuntimeError("MPI Crash!")) - + # An error out of the launch must be caught and turned into a `failed` + # replicate (never left `running`). + mocker.patch("topiary.generax._reconcile_bootstrap._launch_replicate", + side_effect=RuntimeError("MPI Crash!")) + # Create a mock directory structure in tmpdir rep_dir = os.path.join(tmpdir, "replicates") os.makedirs(rep_dir) os.makedirs(os.path.join(rep_dir, "00001")) with open(os.path.join(rep_dir, "00001", "run_generax.sh"), "w") as f: - f.write("generax --args\n") - - # Mock chdir to actually change directory within tmpdir - original_chdir = os.chdir - def mock_chdir(path): - if path == ".." : - original_chdir(os.path.dirname(os.getcwd())) - elif os.path.isabs(path): - original_chdir(path) - else: - original_chdir(os.path.join(os.getcwd(), path)) + f.write("generax --args &> topiary.log\n") - mocker.patch("os.chdir", side_effect=mock_chdir) - - # Mock mpi._get_mpi_oversubscribe and get_mpi_env - mocker.patch("topiary.generax._reconcile_bootstrap.mpi._get_mpi_oversubscribe", return_value=False) - mocker.patch("topiary.generax._reconcile_bootstrap.mpi.get_mpi_env", return_value={"STAY": "STAY"}) + # Mock mpi._get_mpi_oversubscribe and get_mpi_flags + mocker.patch("topiary.generax._reconcile_bootstrap.mpi._get_mpi_oversubscribe", + return_value=False) + mocker.patch("topiary.generax._reconcile_bootstrap.mpi.get_mpi_flags", + return_value=[]) # Run thread function original_dir = os.getcwd() try: - os.chdir(rep_dir) rb._generax_thread_function(rep_dir, 0.03, False, ["localhost"], lock=None) finally: os.chdir(original_dir) - + # Check if 'failed' file was created assert os.path.isfile(os.path.join(rep_dir, "00001", "failed")) # Check if 'running' file was removed (or never existed/was cleaned up) assert not os.path.exists(os.path.join(rep_dir, "00001", "running")) - - # Verify that the Exception was caught and printed as a WARNING - # (We can check stdout if needed, but the 'failed' file is the main indicator) diff --git a/tests/topiary/pipeline/test_bootstrap_reconcile.py b/tests/topiary/pipeline/test_bootstrap_reconcile.py index 6813323..d68db70 100644 --- a/tests/topiary/pipeline/test_bootstrap_reconcile.py +++ b/tests/topiary/pipeline/test_bootstrap_reconcile.py @@ -62,7 +62,8 @@ def side_effect_isdir(path): num_threads=2, threads_per_rep=8, # Closest factor of 56 to 10 raxml_binary=mocker.ANY, - generax_binary=mocker.ANY + generax_binary=mocker.ANY, + timeout_config=mocker.ANY ) mock_pipeline_report.assert_called() @@ -139,5 +140,6 @@ def side_effect_isdir(path): num_threads=56, threads_per_rep=8, raxml_binary=mocker.ANY, - generax_binary=mocker.ANY + generax_binary=mocker.ANY, + timeout_config=mocker.ANY )