remove unnecessary files

su-xingyu · Sep 8, 2022 · e70817b · e70817b
1 parent 5c5bfb6
commit e70817b
Show file tree

Hide file tree

Showing 6 changed files with 58 additions and 849 deletions.
diff --git a/eval/plot.py b/eval/plot.py
@@ -67,7 +67,7 @@ def plot_results(path, centralized, data_machine="machine0", data_node=0):
                 filepath = os.path.join(mf_path, f)
                 with open(filepath, "r") as inf:
                     results.append(json.load(inf))
-        if folder.startswith("FL"):
+        if folder.startswith("FL") or folder.startswith("Parameter Server"):
             data_node = -1
         else:
             data_node = 0
@@ -76,7 +76,8 @@ def plot_results(path, centralized, data_machine="machine0", data_node=0):
         main_data = [main_data]
         # Plot Training loss
         plt.figure(1)
-        means, stdevs, mins, maxs = get_stats([x["train_loss"] for x in results])
+        means, stdevs, mins, maxs = get_stats(
+            [x["train_loss"] for x in results])
         plot(means, stdevs, mins, maxs, "Training Loss", folder, "upper right")
         df = pd.DataFrame(
             {
@@ -93,9 +94,11 @@ def plot_results(path, centralized, data_machine="machine0", data_node=0):
         # Plot Testing loss
         plt.figure(2)
         if centralized:
-            means, stdevs, mins, maxs = get_stats([x["test_loss"] for x in main_data])
+            means, stdevs, mins, maxs = get_stats(
+                [x["test_loss"] for x in main_data])
         else:
-            means, stdevs, mins, maxs = get_stats([x["test_loss"] for x in results])
+            means, stdevs, mins, maxs = get_stats(
+                [x["test_loss"] for x in results])
         plot(means, stdevs, mins, maxs, "Testing Loss", folder, "upper right")
         df = pd.DataFrame(
             {
@@ -112,9 +115,11 @@ def plot_results(path, centralized, data_machine="machine0", data_node=0):
         # Plot Testing Accuracy
         plt.figure(3)
         if centralized:
-            means, stdevs, mins, maxs = get_stats([x["test_acc"] for x in main_data])
+            means, stdevs, mins, maxs = get_stats(
+                [x["test_acc"] for x in main_data])
         else:
-            means, stdevs, mins, maxs = get_stats([x["test_acc"] for x in results])
+            means, stdevs, mins, maxs = get_stats(
+                [x["test_acc"] for x in results])
         plot(means, stdevs, mins, maxs, "Testing Accuracy", folder, "lower right")
         df = pd.DataFrame(
             {
@@ -153,6 +158,7 @@ def plot_results(path, centralized, data_machine="machine0", data_node=0):
         means, stdevs, mins, maxs = get_stats(bytes_list)
         bytes_means[folder] = list(means.values())[0]
         bytes_stdevs[folder] = list(stdevs.values())[0]
+        print(bytes_list)
 
         meta_list = []
         for x in results:

diff --git a/eval/testingPeerSampler.py b/eval/testingPeerSampler.py
@@ -11,8 +11,6 @@
 from decentralizepy.node.DPSGDWithPeerSampler import DPSGDWithPeerSampler
 from decentralizepy.node.PeerSamplerDynamic import PeerSamplerDynamic
 from decentralizepy.node.PeerSampler import PeerSampler
-from decentralizepy.node.ParameterServer import ParameterServer
-from decentralizepy.node.DPSGDNodeWithParameterServer import DPSGDNodeWithParameterServer
 
 
 def read_ini(file_path):
@@ -62,8 +60,7 @@ def read_ini(file_path):
         processes.append(
             mp.Process(
                 # target=PeerSamplerDynamic,
-                target=ParameterServer,
-                # target=PeerSampler,
+                target=PeerSampler,
                 args=[
                     sr,
                     m_id,
@@ -80,8 +77,7 @@ def read_ini(file_path):
     for r in range(0, procs_per_machine):
         processes.append(
             mp.Process(
-                target=DPSGDNodeWithParameterServer,
-                # target=DPSGDWithPeerSampler,
+                target=DPSGDWithPeerSampler,
                 args=[
                     r,
                     m_id,

diff --git a/src/decentralizepy/node/DPSGDNodeFederated.py b/src/decentralizepy/node/DPSGDNodeFederated.py
@@ -1,4 +1,5 @@
 import importlib
+import json
 import logging
 import math
 import os
@@ -35,7 +36,7 @@ def run(self):
             del data["iteration"]
             del data["CHANNEL"]
 
-            self.model.load_state_dict(data)
+            self.model.load_state_dict(data["params"])
             self.sharing._post_step()
             self.sharing.communication_round += 1
 
@@ -60,6 +61,47 @@ def run(self):
             to_send["CHANNEL"] = "DPSGD"
             self.communication.send(self.parameter_server_uid, to_send)
 
+            if self.participated > 0:
+                with open(
+                    os.path.join(
+                        self.log_dir, "{}_results.json".format(self.rank)),
+                    "r",
+                ) as inf:
+                    results_dict = json.load(inf)
+            else:
+                results_dict = {
+                    "train_loss": {},
+                    "test_loss": {},
+                    "test_acc": {},
+                    "total_bytes": {},
+                    "total_meta": {},
+                    "total_data_per_n": {},
+                    "grad_mean": {},
+                    "grad_std": {},
+                }
+
+            results_dict["total_bytes"][iteration
+                                        + 1] = self.communication.total_bytes
+
+            if hasattr(self.communication, "total_meta"):
+                results_dict["total_meta"][
+                    iteration + 1
+                ] = self.communication.total_meta
+            if hasattr(self.communication, "total_data"):
+                results_dict["total_data_per_n"][
+                    iteration + 1
+                ] = self.communication.total_data
+            if hasattr(self.sharing, "mean"):
+                results_dict["grad_mean"][iteration + 1] = self.sharing.mean
+            if hasattr(self.sharing, "std"):
+                results_dict["grad_std"][iteration + 1] = self.sharing.std
+
+            with open(
+                os.path.join(
+                    self.log_dir, "{}_results.json".format(self.rank)), "w"
+            ) as of:
+                json.dump(results_dict, of)
+
             self.participated += 1
 
         # only if has participated in learning