eval/testing_secure.py

+import logging
+from pathlib import Path
+from shutil import copy
+
+from localconfig import LocalConfig
+from torch import multiprocessing as mp
+
+from decentralizepy import utils
+from decentralizepy.graphs.Graph import Graph
+from decentralizepy.mappings.Linear import Linear
+from decentralizepy.node.SecureCompressedAggregation import SecureCompressedAggregation
+
+
+def read_ini(file_path):
+    config = LocalConfig(file_path)
+    for section in config:
+        print("Section: ", section)
+        for key, value in config.items(section):
+            print((key, value))
+    print(dict(config.items("DATASET")))
+    return config
+
+
+if __name__ == "__main__":
+    args = utils.get_args()
+
+    Path(args.log_dir).mkdir(parents=True, exist_ok=True)
+
+    log_level = {
+        "INFO": logging.INFO,
+        "DEBUG": logging.DEBUG,
+        "WARNING": logging.WARNING,
+        "ERROR": logging.ERROR,
+        "CRITICAL": logging.CRITICAL,
+    }
+
+    config = read_ini(args.config_file)
+    my_config = dict()
+    for section in config:
+        my_config[section] = dict(config.items(section))
+
+    copy(args.config_file, args.log_dir)
+    copy(args.graph_file, args.log_dir)
+    utils.write_args(args, args.log_dir)
+
+    g = Graph()
+    g.read_graph_from_file(args.graph_file, args.graph_type)
+    n_machines = args.machines
+    procs_per_machine = args.procs_per_machine
+    l = Linear(n_machines, procs_per_machine)
+    m_id = args.machine_id
+
+    processes = []
+    for r in range(procs_per_machine):
+        processes.append(
+            mp.Process(
+                target=SecureCompressedAggregation,
+                args=[
+                    r,
+                    m_id,
+                    l,
+                    g,
+                    my_config,
+                    args.iterations,
+                    args.log_dir,
+                    args.weights_store_dir,
+                    log_level[args.log_level],
+                    args.test_after,
+                    args.train_evaluate_after,
+                    args.reset_optimizer,
+                ],
+            )
+        )
+
+    for p in processes:
+        p.start()
+
+    for p in processes:
+        p.join()

src/decentralizepy/graphs/Graph.py

+from multiprocessing.sharedctypes import Value
 import networkx as nx
 import numpy as np
 
@@ -152,6 +153,33 @@ class Graph:
         """
         return self.adj_list[uid]
 
+    def nodes_at_distance(self, uid, distance):
+        """
+        Returns the set of all vertices at the given
+        distance from a node.
+        
+        Parameters
+        ----------
+        uid : int
+            globally unique identifier of the node
+            
+        Returns
+        -------
+        set(int)
+            a set of nodes at the given distance from uid
+        
+        """
+        if distance == 1:
+            return self.adj_list[uid]
+        elif distance == 0:
+            return set(uid)
+        elif distance < 0:
+            raise ValueError('Negative distance')
+        vertex_set = set()
+        for neighbor in self.adj_list[uid]:
+            vertex_set |= self.nodes_at_distance(neighbor, distance - 1)
+        return vertex_set
+
     def centr(self):
         my_adj = {x: list(adj) for x, adj in enumerate(self.adj_list)}
         nxGraph = nx.Graph(my_adj)

src/decentralizepy/random.py

+import random
+import contextlib
+
+import torch
+import numpy as np
+
+@contextlib.contextmanager
+def temp_seed(seed):
+    """
+    Creates a context with seeds set to given value. Returns to the
+    previous seed afterwards. 
+
+    Note: Based on torch implementation there might be issues with CUDA
+        causing troubles with the correctness of this function. Function
+        torch.rand() work fine from testing as their results are generated
+        on CPU regardless if CUDA is used for other things.
+        
+    """
+    random_state    = random.getstate()
+    np_old_state    = np.random.get_state()
+    torch_old_state = torch.random.get_rng_state()
+    random.seed(seed)
+    np.random.seed(seed)
+    torch.random.manual_seed(seed)
+    try:
+        yield
+    finally:
+        random.setstate(random_state)
+        np.random.set_state(np_old_state)
+        torch.random.set_rng_state(torch_old_state)
+
+
+class RandomState:
+    """
+    Creates a state that affects random number generation on 
+    torch and numpy and whose context can be activated at will
+
+    """
+    def __init__(self, seed):
+        with temp_seed(seed):
+            self.__refresh_states()
+
+    def __refresh_states(self):
+        self.__random_state = random.getstate()
+        self.__np_state     = np.random.get_state()
+        self.__torch_state  = torch.random.get_rng_state()
+
+    def __set_states(self):
+        random.setstate(self.__random_state)
+        np.random.set_state(self.__np_state)
+        torch.random.set_rng_state(self.__torch_state)
+
+    @contextlib.contextmanager
+    def activate(self):
+        """
+        Activates this state in the given context for torch and
+        numpy. The previous state is restored when the context
+        is finished
+
+        """
+        random_state    = random.getstate()
+        np_old_state    = np.random.get_state()
+        torch_old_state = torch.random.get_rng_state()
+        self.__set_states()
+        try:
+            yield
+        finally:
+            self.__refresh_states()
+            random.setstate(random_state)
+            np.random.set_state(np_old_state)
+            torch.random.set_rng_state(torch_old_state)
\ No newline at end of file

src/decentralizepy/sharing/Choco.py

@@ -310,9 +310,9 @@ class Choco(Sharing):
             model,
             dataset,
             log_dir,
-            compress=False,
-            compression_package=None,
-            compression_class=None
+            compress,
+            compression_package,
+            compression_class
         )
         self.step_size = step_size
         self.alpha = alpha