• June 19, 2024
• in examples
• 2 min read

Streamline Debugging with Faster Issue Detection

The new execution option return_as=generator_unordered can be set via options to facilitate quicker issue detection in your runs. This feature allows you to identify and resolve failures without waiting for all executions to complete, enhancing the efficiency of diagnostics and debugging.

Streamline debugging with faster issue detection

import subprocess
import sys
import time

import mltraq
from mltraq.opts import options
from mltraq.run import RunException
from mltraq.utils.logging import logging_ctx


def faulty_step(run: mltraq.Run):
    """
    Faulty step: it's slow if a==1, and raises an exception if a==5.
    """

    class TestException(Exception):
        pass

    if run.params.a == 5:
        raise TestException("test error")
    elif run.params.a == 1:
        # make this task slower (faster than the one with a==5)
        time.sleep(0.1)


def experiment():
    """
    Execute the experiment.
    """

    # Define experiment with 10 runs
    s = mltraq.create_session()
    e = s.create_experiment("test")
    e.add_runs(a=range(10))

    with options().ctx(
        {
            "execution.return_as": "generator_unordered",
            "execution.exceptions.compact_message": True,
            "execution.exceptions.report_basenames": True,
        }
    ):
        try:
            # Run using only two parallel running jobs
            e.execute(faulty_step, n_jobs=2)
        except RunException as ex:
            print(ex)


def local(args: list[str]):
    """
    Run command with argv `args` and return output as a string, suppressing stderr.
    """
    return subprocess.check_output(args, stderr=subprocess.DEVNULL).decode(
        "utf-8"
    )  # noqa: S603


if sys.argv[0] == "./failfast.py":
    # If executed directly ...
    if sys.argv[-1] != "failfast_experiment":
        # If executing the script with no parameters, run the same script as a separate process,
        # returning only the standard output. This avoids unnecessariy debug information from
        # other threads and processes handled by Joblib, on which we cannot control the output.
        out = local(
            ["python", "./failfast.py", "failfast_experiment"]
        ).strip()
        print(f"\n--\n{out}\n--\n")
    else:
        # Run the experiment if this script is executed by passing "failfast_experiment" as last argument.
        with logging_ctx(level_name="DEBUG"):
            experiment()
else:
    # If executed from mkdocs scripts ...
    print(local(["python", "./failfast.py"]))

Output

--
DEBUG     2024-10-24 12:48:57  Logging level set to DEBUG
DEBUG     2024-10-24 12:48:57  Created DB link: 'sqlite:///:memory:'
DEBUG     2024-10-24 12:48:57  Using backend: loky
DEBUG     2024-10-24 12:48:57  Executing 10 tasks on 2 workers (backend:loky)
DEBUG     2024-10-24 12:48:59  Encountered exception in task, propagating to main process
DEBUG     2024-10-24 12:48:59  Executed 4 of 10 tasks (40%) with return_as=generator_unordered
TestException at failfast.py:20::faulty_step "<unkown>": test error
--