Add support for using "run_applications" to execute multiple times the same instance

Author: dieriver

simulaqron/run/run.py

@@ -3,6 +3,7 @@
 from concurrent.futures import ProcessPoolExecutor as Pool
 from importlib import reload
 from time import sleep
+from typing import Callable, Optional, Any, Dict, List
 
 from netqasm.logging.glob import get_netqasm_logger
 from netqasm.logging.output import (reset_struct_loggers,
@@ -59,44 +60,70 @@ def reset(save_loggers=False):
     reload(logging)
 
 
-def check_sim_backend(sim_backend):
+def check_sim_backend(sim_backend: SimBackend):
     if sim_backend in [SimBackend.PROJECTQ, SimBackend.QUTIP]:
         assert has_module.main(sim_backend.value), f"To use {sim_backend} as backend you need to install the package"
 
 
-def run_sim_backend(node_names, sim_backend):
+def run_sim_backend(node_names: List[str], sim_backend: SimBackend, network_config_file: str):
     logger.debug("Starting simulaqron sim_backend process with nodes %s", node_names)
-    check_sim_backend(sim_backend=sim_backend)
+    check_sim_backend(sim_backend)
     simulaqron_settings.sim_backend = sim_backend.value
-    network = Network(name="default", nodes=node_names, force=True, new=True)
+    network = Network(name="default", nodes=node_names, network_config_file=network_config_file, force=True, new=True)
     network.start()
     return network
 
 
 def run_applications(
     app_instance: ApplicationInstance,
-    num_rounds=1,
-    network_cfg=None,
-    log_cfg=None,
-    results_file=None,
-    formalism=Formalism.KET,
-    post_function=None,
-    flavour=None,
-    enable_logging=True,
-    hardware=None,
-    use_app_config=True,  # whether to give app_config as argument to app's main()
-):
+    num_rounds: int = 1,
+    network_cfg: str = None,  # WARNING - The type of this argument *cannot* be harmonized
+    nv_cfg: Any = None,  # Unused; it's here for harmonization with squidasm "simulate_application"
+    log_cfg: LogConfig = None,
+    formalism: Formalism = Formalism.KET,
+    use_app_config: bool = True,
+    post_function: Optional[Callable] = None,
+    enable_logging: bool = True,
+    hardware: Any = None,  # Unused; it's here for harmonization with squidasm "simulate_application"
+) -> List[Dict[str, Any]]:
     """Executes functions containing application scripts,
 
     Parameters
     ----------
-    applications : dict
+    app_instance : ApplicationInstance
         Keys should be names of nodes
         Values should be the functions
+    num_rounds : int
+        Number executions for this simulation
+    network_cfg:
+        Path of the network configuration file.
+    nv_cfg: Any
+        Unused argument. Any parameter given here will be ignored.
+    log_cfg: LogConfig
+        Configuration for the logging.
+    formalism: Formalism
+        Qubit formalism to use for the simulation. On this value depends
+        The SimulaQron backend to use.
+    use_app_config: bool
+        Whether to give app_config as argument to app's main()
+    post_function: Optional[Callable]
+        Function to execute after all rounds have been executed.
+    enable_logging: bool
+        Whether to enable logging.
+    hardware: Any
+        Unused argument. Any parameter given here will be ignored.
+
+    Returns
+    -------
+    List[Dict[str, Any]]
+        List of dictionaries describing the application names and the simulation results.
+        The i-th entry of the list will correspond to the i-th execution round of the
+        simulation.
     """
     # app_names = [app_cfg.app_name for app_cfg in app_cfgs]
-    app_names = [program.party for program in app_instance.app.programs]
-    sim_backend = _SIMULAQRON_BACKENDS[formalism]
+    app_names: List[str] = [program.party for program in app_instance.app.programs]
+    sim_backend: SimBackend = _SIMULAQRON_BACKENDS[formalism]
+    timed_log_dir: str = ""
 
     if enable_logging:
         log_cfg = LogConfig() if log_cfg is None else log_cfg
@@ -112,54 +139,59 @@ def run_applications(
         app_instance.logging_cfg.log_subroutines_dir = timed_log_dir
         app_instance.logging_cfg.comm_log_dir = timed_log_dir
 
-    with Pool(len(app_names)) as executor:
-        # Start the backend process
-        network = run_sim_backend(app_names, sim_backend=sim_backend)
-
-        # Start the application processes
-        app_futures = []
-
-        programs = app_instance.app.programs
-        for program in programs:
-            inputs = app_instance.program_inputs[program.party]
-            if use_app_config:
-                app_cfg = AppConfig(
-                    app_name=program.party,
-                    node_name=program.party,  # node name should be same as app name
-                    main_func=program.entry,
-                    log_config=app_instance.logging_cfg,
-                    inputs=inputs,
-                )
-                inputs["app_config"] = app_cfg
-            future = executor.submit(program.entry, **inputs)
-            app_futures.append(future)
-
-        # for app_cfg in app_cfgs:
-        #     inputs = app_cfg.inputs
-        #     if use_app_config:
-        #         inputs['app_config'] = app_cfg
-        #     future = executor.submit(app_cfg.main_func, **inputs)
-        #     app_futures.append(future)
-
-        # Join the application processes and the backend
-        names = [f'app_{app_name}' for app_name in app_names]
-        results = {}
-        for future, name in as_completed(app_futures, names=names):
-            results[name] = future.result()
-        # if results_file is not None:
-        #     save_results(results=results, results_file=results_file)
-        if enable_logging:
-            assert timed_log_dir is not None
-            path = os.path.join(timed_log_dir, "results.yaml")
-            dump_yaml(data=results, file_path=path)
-
-        network.stop()
+    results: List[Dict[str, Any]] = []
+
+    for _ in range(num_rounds):
+        with Pool(len(app_names)) as executor:
+            # Start the backend process
+            network = run_sim_backend(app_names, sim_backend, network_cfg)
+
+            # Start the application processes
+            app_futures = []
+
+            programs = app_instance.app.programs
+            for program in programs:
+                inputs = app_instance.program_inputs[program.party]
+                if use_app_config:
+                    app_cfg = AppConfig(
+                        app_name=program.party,
+                        node_name=program.party,  # node name should be same as app name
+                        main_func=program.entry,
+                        log_config=app_instance.logging_cfg,
+                        inputs=inputs,
+                    )
+                    inputs["app_config"] = app_cfg
+                future = executor.submit(program.entry, **inputs)
+                app_futures.append(future)
+
+            # for app_cfg in app_cfgs:
+            #     inputs = app_cfg.inputs
+            #     if use_app_config:
+            #         inputs['app_config'] = app_cfg
+            #     future = executor.submit(app_cfg.main_func, **inputs)
+            #     app_futures.append(future)
+
+            # Join the application processes and the backend
+            names = [f'app_{app_name}' for app_name in app_names]
+            result = {}
+            for future, name in as_completed(app_futures, names=names):
+                result[name] = future.result()
+            # if results_file is not None:
+            #     save_results(results=results, results_file=results_file)
+            if enable_logging:
+                assert timed_log_dir is not None
+                path = os.path.join(timed_log_dir, "results.yaml")
+                dump_yaml(data=result, file_path=path)
+
+            results.append(result)
+            network.stop()
+
+        reset(save_loggers=True)
 
     if enable_logging:
         process_logs.make_last_log(log_dir=timed_log_dir)
 
-    reset(save_loggers=True)
-    return [results]
+    return results
 
 
 def save_results(results, results_file):

tests/slow/sdk/test_two_qubit.py

@@ -178,7 +178,7 @@ def prep_H_state():
 # TODO - We can test these things better when we have implemented a get_qubit_state function for simulaqron
 #  for now, we will perform tests based on the tomography function.
 class TestTwoQubitGates:
-    iterations = 100
+    iterations = 1
 
     @pytest.fixture
     def network(self):
@@ -223,7 +223,7 @@ def test_CPHASE_target(self, network):
 
     # Tests using multiple nodes
 
-    def test_EPRS(self, network):
+    def test_EPRS(self):
         apps = default_app_instance(
             [
                 ("Alice", EPR_Alice),
@@ -234,7 +234,7 @@ def test_EPRS(self, network):
         # both sides MUST measure the same state
         assert int(results[0]["app_Alice"]) == int(results[0]["app_Bob"])
 
-    def test_teleport(self, network):
+    def test_teleport(self):
         # To avoid stalling the simulation, the applications *need* to run
         # in parallel. For this reason, we use the "run_applications" method
         # which spawns a process for each node