Skip to content

climate_ref.executor.synchronous #

SynchronousExecutor #

Run a diagnostic synchronously, in-process.

This is mainly useful for debugging and testing. climate_ref.executor.LocalExecutor is a more general purpose executor.

Source code in packages/climate-ref/src/climate_ref/executor/synchronous.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
class SynchronousExecutor:
    """
    Run a diagnostic synchronously, in-process.

    This is mainly useful for debugging and testing.
    [climate_ref.executor.LocalExecutor][] is a more general purpose executor.
    """

    name = "synchronous"

    # ``join`` is a no-op, so skipping it (queue-and-exit) does not lose any work.
    collects_results_on_join = False

    # Diagnostics run on the solving host, so a health check there reflects where they run.
    runs_on_solve_host = True

    def __init__(
        self, *, database: Database | None = None, config: Config | None = None, **kwargs: Any
    ) -> None:
        if config is None:
            config = Config.default()
        if database is None:
            database = Database.from_config(config, run_migrations=False)
            # A database we opened is ours to close, so it does not outlive this executor.
            weakref.finalize(self, database.close)

        self.database = database
        self.config = config

        if self.config.executor.measure_resources:
            logger.warning(
                "The synchronous executor runs every execution in this one long-lived process, "
                "so a recorded peak memory can include the footprint of an earlier execution"
            )

    def run(
        self,
        definition: ExecutionDefinition,
        execution: Execution | None = None,
    ) -> None:
        """
        Run a diagnostic in process

        Parameters
        ----------
        definition
            A description of the information needed for this execution of the diagnostic
        execution
            A database model representing the execution of the diagnostic.
            If provided, the result will be updated in the database when completed.
        """
        result = execute_locally(
            definition,
            log_level=self.config.log_level,
            measure=self.config.executor.measure_resources,
            # One execution at a time in this one process,
            # so nothing else in the cgroup is competing for the memory being measured.
            exclusive=True,
        )

        # Solver now commits the Execution row before submitting to the executor,
        # so the instance handed to ``run`` is detached and DB writes inside
        # ``handle_execution_result`` would not persist on their own. Always own
        # the transaction here for real Execution rows. Test mocks (non-ORM
        # objects) keep the legacy in-process behaviour.
        if isinstance(execution, Execution):
            with self.database.session.begin():
                attached = self.database.session.merge(execution)
                process_result(self.config, self.database, result, attached, queue_seconds=0.0)
            return

        process_result(self.config, self.database, result, execution, queue_seconds=0.0)

    def join(self, timeout: float) -> None:
        """
        Wait for all diagnostics to finish

        This returns immediately because the executor runs diagnostics synchronously.

        Parameters
        ----------
        timeout
            Timeout in seconds (Not used)
        """
        pass

join(timeout) #

Wait for all diagnostics to finish

This returns immediately because the executor runs diagnostics synchronously.

Parameters:

Name Type Description Default
timeout float

Timeout in seconds (Not used)

 required
Source code in packages/climate-ref/src/climate_ref/executor/synchronous.py 
87
88
89
90
91
92
93
94
95
96
97
98
def join(self, timeout: float) -> None:
    """
    Wait for all diagnostics to finish

    This returns immediately because the executor runs diagnostics synchronously.

    Parameters
    ----------
    timeout
        Timeout in seconds (Not used)
    """
    pass

run(definition, execution=None) #

Run a diagnostic in process

Parameters:

Name Type Description Default
definition ExecutionDefinition

A description of the information needed for this execution of the diagnostic

 required
execution Execution | None

A database model representing the execution of the diagnostic. If provided, the result will be updated in the database when completed.

 None
Source code in packages/climate-ref/src/climate_ref/executor/synchronous.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def run(
    self,
    definition: ExecutionDefinition,
    execution: Execution | None = None,
) -> None:
    """
    Run a diagnostic in process

    Parameters
    ----------
    definition
        A description of the information needed for this execution of the diagnostic
    execution
        A database model representing the execution of the diagnostic.
        If provided, the result will be updated in the database when completed.
    """
    result = execute_locally(
        definition,
        log_level=self.config.log_level,
        measure=self.config.executor.measure_resources,
        # One execution at a time in this one process,
        # so nothing else in the cgroup is competing for the memory being measured.
        exclusive=True,
    )

    # Solver now commits the Execution row before submitting to the executor,
    # so the instance handed to ``run`` is detached and DB writes inside
    # ``handle_execution_result`` would not persist on their own. Always own
    # the transaction here for real Execution rows. Test mocks (non-ORM
    # objects) keep the legacy in-process behaviour.
    if isinstance(execution, Execution):
        with self.database.session.begin():
            attached = self.database.session.merge(execution)
            process_result(self.config, self.database, result, attached, queue_seconds=0.0)
        return

    process_result(self.config, self.database, result, execution, queue_seconds=0.0)