Multiprocessing

opengate/base.py

@@ -690,7 +690,7 @@ def process_dynamic_parametrisation(self, params):
         extra_params = {}
         extra_params["auto_changer"] = params.pop(
             "auto_changer", True
-        )  # True of key not found (default)
+        )  # True if key not found (default)
         if extra_params["auto_changer"] not in (False, True):
             fatal(
                 f"Received wrong value type for 'auto_changer': got {type(extra_params['auto_changer'])}, "
@@ -750,6 +750,14 @@ def add_dynamic_parametrisation(self, name=None, **params):
             s += f"{k}: {v}\n"
         log.debug(s)
 
+    def reassign_subset_of_dynamic_params(self, subset):
+        # loop over all dynamic parametrisations of this object,
+        for param in self.user_info["dynamic_params"].values():
+            for k, v in param.items():
+                # extract the subset of entries to the list that are relevant to this process
+                if k in self.dynamic_user_info:
+                    param[k] = [v[i] for i in subset]
+
     def create_changers(self):
         # this base class implementation is here to keep inheritance intact.
         return []

opengate/engines.py

@@ -1004,6 +1004,7 @@ def __init__(self, simulation, new_process=False):
         # this is only for info.
         # Process handling is done in Simulation class, not in SimulationEngine!
         self.new_process = new_process
+        self.process_index = None
 
         # LATER : option to wait the end of completion or not
 

opengate/managers.py

@@ -7,6 +7,7 @@
 import os
 from pathlib import Path
 import weakref
+import multiprocessing
 
 import opengate_core as g4
 
@@ -1460,6 +1461,7 @@ def __init__(self, name="simulation", **kwargs):
         self._current_random_seed = None
 
         self.expected_number_of_events = None
+        self.mapping_run_timing_intervals = {}
 
     def __str__(self):
         s = (
@@ -1653,7 +1655,7 @@ def add_filter(self, filter_type, name):
     def multithreaded(self):
         return self.number_of_threads > 1 or self.force_multithread_mode
 
-    def _run_simulation_engine(self, start_new_process):
+    def _run_simulation_engine(self, start_new_process, process_index=None):
         """Method that creates a simulation engine in a context (with ...) and runs a simulation.
 
         Args:
@@ -1668,17 +1670,74 @@ def _run_simulation_engine(self, start_new_process):
         with SimulationEngine(self) as se:
             se.new_process = start_new_process
             se.init_only = self.init_only
+            se.process_index = process_index
             output = se.run_engine()
         return output
 
-    def run(self, start_new_process=False):
+    def generate_run_timing_interval_map(self, number_of_processes):
+        if number_of_processes % len(self.run_timing_intervals) != 0:
+            fatal(
+                "number_of_sub_processes must be a multiple of the number of run_timing_intervals, \n"
+                f"but I received {number_of_processes}, while there are {len(self.run_timing_intervals)}."
+            )
+
+        number_of_processes_per_run = int(
+            number_of_processes / len(self.run_timing_intervals)
+        )
+        run_timing_interval_map = {}
+        process_index = 0
+        for i, rti in enumerate(self.run_timing_intervals):
+            t_start, t_end = rti
+            duration_original = t_end - t_start
+            duration_in_process = duration_original / number_of_processes_per_run
+            t_intermediate = [
+                t_start + (j + 1) * duration_in_process
+                for j in range(number_of_processes_per_run - 1)
+            ]
+            t_all = [t_start] + t_intermediate + [t_end]
+            for t_s, t_e in zip(t_all[:-1], t_all[1:]):
+                run_timing_interval_map[process_index] = {
+                    "run_timing_intervals": [[t_s, t_e]],
+                    "lut_original_rti": [i],
+                }
+                process_index += 1
+        return run_timing_interval_map
+
+    def run_in_process(self, process_index, run_timing_intervals, lut_original_rti):
+        # Important: this method is intended to run in a processes spawned off the main process.
+        # Therefore, self is actually a separate instance from the original simulation
+        # and we can safely adapt it in this process.
+
+        # adapt the output_dir
+        self.output_dir = str(Path(self.output_dir) / f"process_{process_index}")
+        print("self.output_dir = ", self.output_dir)
+
+        # adapt the run timing intervals in
+        self.run_timing_intervals = run_timing_intervals
+        # adapt all dynamic volumes
+        for vol in self.volume_manager.dynamic_volumes:
+            vol.reassign_subset_of_dynamic_params(lut_original_rti)
+            print(process_index)
+            print(f"Volume {vol.name}:")
+            print(vol.user_info["dynamic_params"])
+
+        output = self._run_simulation_engine(False, process_index=process_index)
+        print(
+            process_index, os.getpid(), id(self), run_timing_intervals, lut_original_rti
+        )
+        return output
+
+    def run(self, start_new_process=False, number_of_sub_processes=0):
         # if windows and MT -> fail
         if os.name == "nt" and self.multithreaded:
             fatal(
                 "Error, the multi-thread option is not available for Windows now. "
                 "Run the simulation with one thread."
             )
 
+        if number_of_sub_processes == 1:
+            start_new_process = True
+
         # prepare sub process
         if start_new_process is True:
             """Important: put:
@@ -1705,6 +1764,50 @@ def run(self, start_new_process=False):
                     source.fTotalSkippedEvents = s.user_info.fTotalSkippedEvents
                     source.fTotalZeroEvents = s.user_info.fTotalZeroEvents
 
+        elif number_of_sub_processes > 1:
+            run_timing_interval_map = self.generate_run_timing_interval_map(
+                number_of_sub_processes
+            )
+            try:
+                multiprocessing.set_start_method("spawn")
+            except RuntimeError:
+                print("Could not set start method 'spawn'.")
+                pass
+            # q = multiprocessing.Queue()
+            with multiprocessing.Pool(len(run_timing_interval_map)) as pool:
+                print("pool._outqueue: ", pool._outqueue)  # DEMO
+                results = [
+                    pool.apply_async(
+                        self.run_in_process,
+                        (
+                            k,
+                            v["run_timing_intervals"],
+                            v["lut_original_rti"],
+                        ),
+                    )
+                    for k, v in run_timing_interval_map.items()
+                ]
+                # `.apply_async()` immediately returns AsyncResult (ApplyResult) object
+                print(results[0])  # DEMO
+                list_of_output = [res.get() for res in results]
+                print(f"list_of_output: {list_of_output}")
+            return list_of_output
+            # processes = []
+            # for k, v in run_timing_interval_map.items():
+            #     p = multiprocessing.Process(
+            #         target=target_func,
+            #         args=(q, self.run_in_process, k, v['run_timing_intervals'], v['lut_original_rti'])
+            #     )
+            #     p.start()
+            #     processes.append(p)
+            # for p in processes:
+            #     p.join()  # (timeout=10)  # timeout might be needed
+            #
+            # try:
+            #     output = q.get(block=False)
+            # except queue.Empty:
+            #     fatal("The queue is empty. The spawned process probably died.")
+            # return output
         else:
             # Nothing special to do if the simulation engine ran in the native python process
             # because everything is already in place.

opengate/tests/src/test030_dose_motion_dynamic_param_multiproc.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import opengate as gate
+from scipy.spatial.transform import Rotation
+from opengate.tests import utility
+
+if __name__ == "__main__":
+    paths = utility.get_default_test_paths(
+        __file__, "gate_test029_volume_time_rotation", "test030"
+    )
+
+    # create the simulation
+    sim = gate.Simulation()
+
+    # main options
+    sim.g4_verbose = False
+    sim.visu = False
+    sim.random_seed = 983456
+    sim.output_dir = paths.output
+
+    # units
+    m = gate.g4_units.m
+    mm = gate.g4_units.mm
+    cm = gate.g4_units.cm
+    um = gate.g4_units.um
+    nm = gate.g4_units.nm
+    MeV = gate.g4_units.MeV
+    Bq = gate.g4_units.Bq
+    sec = gate.g4_units.second
+
+    #  change world size
+    sim.world.size = [1 * m, 1 * m, 1 * m]
+
+    # add a simple fake volume to test hierarchy
+    # translation and rotation like in the Gate macro
+    fake = sim.add_volume("Box", "fake")
+    fake.size = [40 * cm, 40 * cm, 40 * cm]
+    fake.translation = [1 * cm, 2 * cm, 3 * cm]
+    fake.material = "G4_AIR"
+    fake.color = [1, 0, 1, 1]
+
+    # waterbox
+    waterbox = sim.add_volume("Box", "waterbox")
+    waterbox.mother = fake
+    waterbox.size = [20 * cm, 20 * cm, 20 * cm]
+    waterbox.translation = [-3 * cm, -2 * cm, -1 * cm]
+    waterbox.rotation = Rotation.from_euler("y", -20, degrees=True).as_matrix()
+    waterbox.material = "G4_WATER"
+    waterbox.color = [0, 0, 1, 1]
+
+    # physics
+    sim.physics_manager.set_production_cut("world", "all", 700 * um)
+
+    # default source for tests
+    # the source is fixed at the center, only the volume will move
+    source = sim.add_source("GenericSource", "mysource")
+    source.energy.mono = 150 * MeV
+    source.particle = "proton"
+    source.position.type = "disc"
+    source.position.radius = 5 * mm
+    source.direction.type = "momentum"
+    source.direction.momentum = [0, 0, 1]
+    source.activity = 30000 * Bq
+
+    # add dose actor
+    dose = sim.add_actor("DoseActor", "dose")
+    dose.output_filename = "test030.mhd"
+    dose.attached_to = waterbox
+    dose.size = [99, 99, 99]
+    mm = gate.g4_units.mm
+    dose.spacing = [2 * mm, 2 * mm, 2 * mm]
+    dose.translation = [2 * mm, 3 * mm, -2 * mm]
+    dose.edep.keep_data_per_run = True
+    dose.edep.auto_merge = True
+    dose.edep_uncertainty.active = True
+
+    # add stat actor
+    stats = sim.add_actor("SimulationStatisticsActor", "Stats")
+
+    # motion
+    n = 3
+    interval_length = 1 * sec / n
+    sim.run_timing_intervals = [
+        (i * interval_length, (i + 1) * interval_length) for i in range(n)
+    ]
+    gantry_angles_deg = [i * 20 for i in range(n)]
+    (
+        dynamic_translations,
+        dynamic_rotations,
+    ) = gate.geometry.utility.get_transform_orbiting(
+        initial_position=fake.translation, axis="Y", angle_deg=gantry_angles_deg
+    )
+    fake.add_dynamic_parametrisation(
+        translation=dynamic_translations, rotation=dynamic_rotations
+    )
+
+    # start simulation
+    sim.run(number_of_sub_processes=3 * len(sim.run_timing_intervals))
+
+    # # print results at the end
+    # print(stats)
+    #
+    # # tests
+    # stats_ref = utility.read_stat_file(paths.output_ref / "stats030.txt")
+    # is_ok = utility.assert_stats(stats, stats_ref, 0.11)
+    #
+    # print()
+    # gate.exception.warning("Difference for EDEP")
+    # is_ok = (
+    #     utility.assert_images(
+    #         paths.output_ref / "test030-edep.mhd",
+    #         dose.edep.get_output_path(),
+    #         stats,
+    #         tolerance=30,
+    #         ignore_value=0,
+    #     )
+    #     and is_ok
+    # )
+    #
+    # print("\nDifference for uncertainty")
+    # is_ok = (
+    #     utility.assert_images(
+    #         paths.output_ref / "test030-edep_uncertainty.mhd",
+    #         dose.edep_uncertainty.get_output_path(),
+    #         stats,
+    #         tolerance=15,
+    #         ignore_value=1,
+    #     )
+    #     and is_ok
+    # )
+    #
+    # utility.test_ok(is_ok)

opengate/tests/src/test080_multiprocessing_1.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from opengate.utility import g4_units
+import opengate as gate
+from opengate.tests.utility import get_default_test_paths
+
+
+if __name__ == "__main__":
+    paths = get_default_test_paths(__file__, output_folder="test080")
+
+    s = g4_units.s
+
+    sim = gate.Simulation()
+    sim.run_timing_intervals = [[0 * s, 1 * s], [1 * s, 3 * s], [10 * s, 15 * s]]
+    sim.output_dir = paths.output
+
+    box1 = sim.add_volume("BoxVolume", "box1")
+    box1.add_dynamic_parametrisation(
+        translation=[[i, i, i] for i in range(len(sim.run_timing_intervals))]
+    )
+
+    n_proc = 4 * len(sim.run_timing_intervals)
+    run_timing_interval_map = sim.generate_run_timing_interval_map(n_proc)
+    print(run_timing_interval_map)
+
+    output = sim.run(number_of_sub_processes=n_proc)
+
+    print("*** output ***")
+    for o in output:
+        print(o)
+
+    print(f"ID of the main sim: {id(sim)}")
+
+    ids = [e[2] for e in output]
+    assert id(sim) not in ids