diff --git a/.gitignore b/.gitignore
index 51247532..c0596393 100644
--- a/.gitignore
+++ b/.gitignore
@@ -4,8 +4,12 @@ expt_dump_old/
di_test/
imgs/
pascal/
+<<<<<<< HEAD
data/
!src/inversefed/data/
+=======
+# data/
+>>>>>>> b73ed37 (mia so far)
notes.txt
removeme*.png
diff --git a/src/algos/base_class.py b/src/algos/base_class.py
index c10ac0ac..fbfca210 100644
--- a/src/algos/base_class.py
+++ b/src/algos/base_class.py
@@ -1,5 +1,5 @@
"""Add docstring here"""
-
+import pickle
from abc import ABC, abstractmethod
import sys
import torch
@@ -44,11 +44,14 @@
)
from utils.types import ConfigType, TorchModelType
from utils.dropout_utils import NodeDropout
+from utils.gias import gia_main, LaplacianGossipMatrix, gia_from_disambiguate, compute_param_delta, ReconstructOptim, ParameterTracker
import torchvision.transforms as T # type: ignore
import os
from yolo import YOLOLoss
+from collections import defaultdict
+from algos.topologies.collections import select_topology
def set_seed(seed: int) -> None:
@@ -161,6 +164,8 @@ def setup_logging(self, config: ConfigType) -> None:
"""
try:
config["log_path"] = f"{config['log_path']}/node_{self.node_id}" # type: ignore
+ # print(f"Log path: {config['log_path']}")
+ # raise
os.makedirs(config["log_path"])
except FileExistsError:
color_code = "\033[91m" # Red color
@@ -480,9 +485,18 @@ def __init__(
"""Add docstring here"""
super().__init__(config, comm_utils)
self.server_node = 0
- self.set_parameters(config)
- if "gia" in config:
- if int(self.node_id) in self.config["gia_attackers"]:
+ self.set_model_parameters(config)
+
+ if "mia" in config:
+ # self-attack on train and test labels
+ from utils.mias import LOSSMIA, MIA
+ self.mia = True
+ # self.mia_rounds = set(config["mia_rounds"]) # we are just attacking after every aggregation round
+ self.mia_attacker = LOSSMIA(device=self.device)
+ self.mia_func = MIA
+
+ if "gia" in config and self.node_id in config["gia_attackers"]:
+ if int(self.node_id) in config["gia_attackers"]:
self.gia_attacker = True
self.params_s = dict()
self.params_t = dict()
@@ -490,8 +504,139 @@ def __init__(
self.neighbor_updates = defaultdict(list)
# Track which neighbors we've already attacked
self.attacked_neighbors = set()
-
self.base_params = [key for key, _ in self.model.named_parameters()]
+ self.base_params_vals = [val for _, val in self.model.named_parameters()]
+ self.base_model = copy.deepcopy(OrderedDict(self.model.named_parameters()))
+ self.gia_disaggregate = False
+ self.n_iter = config["rounds"]
+
+ if "disaggregate" in config and self.node_id in config["gia_attackers"]:
+ self.gia_disaggregate = True # flag for launching the Mrini attack
+ self.recorded_params = {
+ 'initial_params': {}, # Store initial parameters for each node
+ 'params_by_round': {}, # Store parameters for each node by round
+ 'actual_images': {}, # Store initial images for each node
+ 'actual_labels': {}, # Store initial labels for each node
+ }
+
+ # repeat code
+ self.topology = select_topology(config, self.node_id)
+ self.topology.initialize()
+ self.G = self.topology.graph
+ self.N = self.G.number_of_nodes()
+ self.attackers = config["gia_attackers"] # a list of node IDs for attackers
+ self.W = LaplacianGossipMatrix(self.G)
+ self.Wt = torch.tensor(self.W)
+ self.R = ReconstructOptim(self.G, self.n_iter, self.attackers)
+ self.num_clients = config["num_users"]
+
+ # TODO client mapping here is implicit based on sorted order of nodes
+ # this might need to be changed to explicit mapping of node_id to index as we introduce more attackers
+ self.parameter_tracker = ParameterTracker(self.node_id,
+ self.Wt,
+ self.num_clients,
+ self.R.n_neighbors,
+ config["gia_attackers"],
+ self.base_params,
+ self.base_params_vals,
+ self.n_iter)
+
+ # save network information
+ network_stats = self.topology.calculate_graph_metrics(self.node_id)
+ self.log_utils.log_network_stats(network_stats)
+
+ def reconstruct_disaggregated(self, x_hat: torch.Tensor) -> None:
+ """
+ Reconstructs the disaggregated model parameters and updates the model.
+ saves reconstructed image to reconstruction path and error rates to the log
+ TODO finish implementing
+ """
+ for matrix_idx in range(self.N-1):
+ print(f"DEBUG: matrix_idx is {matrix_idx}, iter range is {self.N-1}")
+ # Convert matrix index back to original client ID
+ target_id = self.R.get_original_client_id(matrix_idx)
+
+ target_images, target_labels = self.log_utils.retrieve_base_image(target_id)
+
+ # Launch the attack
+
+ if result := gia_from_disambiguate(torch.from_numpy(x_hat[matrix_idx]), # indices rather than actual ID
+ self.base_params_vals,
+ self.model,
+ target_labels,
+ target_images,
+ target_id,
+ self.config["model_lr"],
+ self.device,
+ mean=self.dset_obj.mean,
+ std=self.dset_obj.std,
+ dataset=self.config["dset"]):
+ output, test_mse, test_psnr, feat_mse = result
+ # log output and stats as image
+ stats_dict = {
+ # "output": output,
+ "test_mse": test_mse,
+ "test_psnr": test_psnr,
+ "feat_mse": feat_mse
+ }
+ assert stats_dict is not None, f"stats_dict is None"
+ self.log_utils.log_gia_image(target_images, target_labels, target_id, label=f"client_{target_id}_base_images") # log original image
+ self.log_utils.log_gia_image(output, target_labels, target_id, label=f"round_{round}_disambiguation_reconstruction")
+ self.log_utils.log_gia_stats(stats_dict, target_id)
+ else:
+ self.log_utils.log_summary(f"Client {self.node_id} failed to attack {target_id} in round {round}!")
+ self.log_utils.log_gia_stats(dict(), target_id)
+ print(f"Client {self.node_id} failed to attack {target_id}!")
+ continue
+
+ def receive_disambiguate_attack_and_aggregate(self, neighbors: List[int], current_round: int, total_rounds: int, num_neighbors: int) -> None:
+ if self.is_working:
+ assert neighbors == self.topology.get_all_neighbours(), f"Neighbors mismatch: {neighbors} vs {self.topology.get_all_neighbours()}"
+
+ current_model = OrderedDict((key, value) for key, value in self.model.state_dict().items() if key in self.base_params)
+
+ if self.node_id not in self.recorded_params['initial_params']:
+ self.recorded_params['initial_params'][self.node_id] = copy.deepcopy(current_model)
+ self.recorded_params['params_by_round'].setdefault(self.node_id, {})[current_round] = copy.deepcopy(current_model)
+
+ model_updates = self.comm_utils.receive(node_ids=neighbors)
+ sorted_model_updates = sorted(model_updates, key=lambda x: x["sender"])
+ # with open(f"sorted_updates_{current_round}.pkl", 'wb') as f:
+ # pickle.dump(sorted_model_updates, f)
+
+ self.parameter_tracker.record_update(self.node_id - 1, current_model, current_round, is_attacker=True)
+
+ for neighbor_info in sorted_model_updates:
+ neighbor_id = neighbor_info["sender"]
+ neighbor_index = neighbor_id - 1 # the key used to track parameters is 0-indexed
+ neighbor_model = OrderedDict((key, value) for key, value in neighbor_info["model"].items() if key in self.base_params)
+
+ # assert self.round == neighbor_round, f"Round mismatch: {self.round} vs {neighbor_round}"
+ # record neighbor updates
+ self.parameter_tracker.record_update(neighbor_index, neighbor_model, current_round)
+
+ if neighbor_id not in self.recorded_params['initial_params']:
+ print(neighbor_id, " not in recorded_params")
+ self.recorded_params['initial_params'][neighbor_id] = copy.deepcopy(neighbor_model)
+ else:
+ print(f"neighbor id {neighbor_id} in recorded_params {self.recorded_params['initial_params'].keys()}")
+ self.recorded_params['params_by_round'].setdefault(neighbor_id, {})[current_round] = copy.deepcopy(neighbor_model)
+
+ if neighbor_id not in self.recorded_params['actual_images']:
+ self.recorded_params['actual_images'][neighbor_id] = neighbor_info["images"]
+ self.recorded_params['actual_labels'][neighbor_id] = neighbor_info["labels"]
+
+ # update params0
+ self.parameter_tracker.update_params0(round=current_round)
+
+ if current_round == (total_rounds - 1): # at the final round
+ sent_params, attackers_params = self.parameter_tracker.generate_param_tensors()
+
+ if sent_params is not None and attackers_params is not None:
+ x_hat = self.R.reconstruct_LS_target_only(sent_params, attackers_params)
+ self.reconstruct_disaggregated(x_hat)
+
+ self.aggregate(model_updates, keys_to_ignore=self.model_keys_to_ignore)
def set_parameters(self, config: Dict[str, Any]) -> None:
"""
@@ -506,33 +651,23 @@ def set_parameters(self, config: Dict[str, Any]) -> None:
self.set_model_parameters(config)
self.set_shared_exp_parameters(config)
self.set_data_parameters(config)
-
- # after setting data loaders, save client dataset
- # TODO verify this .data and .labels fields are correct
- if "gia" in config:
- # Extract data and labels
- train_data = torch.stack([data[0] for data in self.train_dset])
- train_labels = torch.tensor([data[1] for data in self.train_dset])
-
- self.log_utils.log_gia_image(train_data,
- train_labels,
- self.node_id)
+ assert self.dloader, f"Data loader not set in baseClient {self.node_id}"
def set_data_parameters(self, config: ConfigType) -> None:
-
# Train set and test set from original dataset
train_dset = self.dset_obj.train_dset
test_dset = self.dset_obj.test_dset
# Handle GIA case first, before any other modifications
if "gia" in config:
- # Select 10 random labels and exactly one image per label for both train and test
+ self.num_labels = 5 if not config.get("disaggregate", False) else 1
+
train_dset, test_dset, classes, train_indices = gia_client_dataset(
- train_dset, test_dset, num_labels=10
+ train_dset, test_dset, num_labels=self.num_labels, node_id=self.node_id
)
- assert len(train_dset) == 10, "GIA should have exactly 10 samples in train set"
- assert len(test_dset) == 10, "GIA should have exactly 10 samples in test set"
+ assert len(train_dset) == self.num_labels, f"GIA should have exactly {self.num_labels} samples in train set but has {len(train_dset)}"
+ assert len(test_dset) == self.num_labels, f"GIA should have exactly {self.num_labels} samples in test set but has {len(test_dset)}"
# Store the images and labels in tensors, matching the format from your example
self.images = []
@@ -552,10 +687,18 @@ def set_data_parameters(self, config: ConfigType) -> None:
self.classes_of_interest = classes
self.train_indices = train_indices
self.train_dset = train_dset
- self.dloader: DataLoader[Any] = DataLoader(train_dset, batch_size=len(train_dset), shuffle=False)
- self._test_loader: DataLoader[Any] = DataLoader(test_dset, batch_size=len(test_dset), shuffle=False)
- print("Using GIA data setup")
- print(self.labels)
+
+ self.dloader = DataLoader(train_dset, batch_size=1, shuffle=False)
+ self._test_loader = DataLoader(test_dset, batch_size=1, shuffle=False)
+
+ # save to log folder:
+ train_data = torch.stack([data[0] for data in self.train_dset])
+ train_labels = torch.tensor([data[1] for data in self.train_dset])
+
+ self.log_utils.log_base_image(train_data,
+ train_labels,
+ self.node_id)
+
else:
if config.get("test_samples_per_class", None) is not None:
test_dset, _ = balanced_subset(test_dset, config["test_samples_per_class"])
@@ -566,7 +709,6 @@ def set_data_parameters(self, config: ConfigType) -> None:
# Support user specific dataset
if isinstance(config["dset"], dict):
-
def is_same_dest(dset):
# Consider all variations of cifar10 as the same dataset
# To avoid having exactly same original dataset (without
@@ -733,12 +875,19 @@ def local_train(self, round: int, epochs: int = 1, **kwargs: Any) -> Tuple[float
start_time = time.time()
self.is_working = self.get_and_set_working(round)
-
if self.is_working:
avg_loss, avg_acc = 0, 0
for _ in range(epochs):
tr_loss, tr_acc = self.model_utils.train(
- self.model, self.optim, self.dloader, self.loss_fn, self.device, malicious_type=self.config.get("malicious_type", "normal"), config=self.config, node_id=self.node_id, gia=self.config.get("gia", False)
+ self.model,
+ self.optim,
+ self.dloader,
+ self.loss_fn,
+ self.device,
+ malicious_type=self.config.get("malicious_type", "normal"),
+ config=self.config,
+ node_id=self.node_id,
+ gia=self.config.get("gia", False)
)
avg_loss += tr_loss
avg_acc += tr_acc
@@ -771,6 +920,9 @@ def local_train(self, round: int, epochs: int = 1, **kwargs: Any) -> Tuple[float
self.stats["train_loss"], self.stats["train_acc"], self.stats["train_time"] = avg_loss, avg_acc, time_taken
+ if avg_loss == float('nan'):
+ print(f"Client {self.node_id} has NaN loss in round {round}")
+ raise ValueError("Loss is NaN")
return avg_loss, avg_acc, time_taken
def local_test(self, **kwargs: Any) -> float | Tuple[float, float] | Tuple[float, float, float] | None:
@@ -779,25 +931,11 @@ def local_test(self, **kwargs: Any) -> float | Tuple[float, float] | Tuple[float
"""
raise NotImplementedError
- def receive_and_aggregate(self):
- """
- Receive the model weights from the server and aggregate them
- """
- if self.is_working:
- repr = self.comm_utils.receive([self.server_node])[0]
- if "round" in repr:
- round = repr["round"]
- if "sender" in repr:
- sender = repr["sender"]
- assert "model" in repr, "Model not found in the received message"
- self.set_model_weights(repr["model"])
-
def receive_attack_and_aggregate(self, neighbors: List[int], round: int, num_neighbors: int) -> None:
"""
Receives updates, launches GIA attack when second update is seen from a neighbor
"""
from utils.gias import gia_main
-
if self.is_working:
# Receive the model updates from the neighbors
model_updates = self.comm_utils.receive(node_ids=neighbors)
@@ -821,13 +959,29 @@ def receive_attack_and_aggregate(self, neighbors: List[int], round: int, num_nei
"labels": neighbor_labels
})
- # Check if we have 2 updates from this neighbor and haven't attacked them yet
- if len(self.neighbor_updates[neighbor_id]) == 2 and neighbor_id not in self.attacked_neighbors:
- print(f"Client {self.node_id} attacking {neighbor_id}!")
+ # Check if we are at the final round and haven't attacked this neighbor yet
+ if round == (self.n_iter - 1) and neighbor_id not in self.attacked_neighbors:
+ print(f"Client {self.node_id} attacking {neighbor_id} in final round!")
- # Get the two parameter sets for the attack
- p_s = self.neighbor_updates[neighbor_id][0]["model"]
- p_t = self.neighbor_updates[neighbor_id][1]["model"]
+ # Get the first parameter set for the attack
+ p_s = self.base_model
+
+ # Find the first non-zero parameter difference after the first set
+ p_t = None
+ for idx in range(0, len(self.neighbor_updates[neighbor_id])):
+ candidate_p_t = self.neighbor_updates[neighbor_id][idx]["model"]
+ param_diff = compute_param_delta(p_s, candidate_p_t, self.base_params)
+ # print(f"DEBUG: param_diff: {param_diff}")
+ if any((diff != 0).any() for diff in param_diff):
+ p_t = candidate_p_t
+ p_t_index = idx
+ break
+
+ if p_t is None:
+ # Handle attack failure case
+ self.log_utils.log_summary(f"Client {self.node_id} failed to find a suitable p_t to attack {neighbor_id} in round {round}!")
+ print(f"Client {self.node_id} failed to find a suitable p_t to attack {neighbor_id} in round {round}!")
+ continue
# Launch the attack
if result := gia_main(p_s,
@@ -836,22 +990,30 @@ def receive_attack_and_aggregate(self, neighbors: List[int], round: int, num_nei
self.model,
neighbor_labels,
neighbor_images,
- self.node_id):
- output, stats = result
-
- # log output and stats as image
+ self.node_id,
+ self.device,
+ mean=self.dset_obj.mean,
+ std=self.dset_obj.std):
+ output, test_mse, test_psnr, feat_mse = result
+ # Log output and stats as image
+ self.log_utils.log_gia_image(neighbor_images, neighbor_labels, neighbor_id, label=f"client_{neighbor_id}_base_images") # log base
self.log_utils.log_gia_image(output, neighbor_labels, neighbor_id, label=f"round_{round}_reconstruction")
- self.log_utils.log_summary(f"round {round} gia targeting {neighbor_id} stats: {stats}")
+ self.log_utils.log_summary(f"round {round} gia targeting {neighbor_id} stats: test_mse: {test_mse}, test_psnr: {test_psnr}, feat_mse: {feat_mse}")
+ self.log_utils.log_gia_stats(f"GIA result- test_mse: {test_mse}, test_psnr: {test_psnr}, feat_mse: {feat_mse}", neighbor_id)
else:
self.log_utils.log_summary(f"Client {self.node_id} failed to attack {neighbor_id} in round {round}!")
- print(f"Client {self.node_id} failed to attack {neighbor_id}!")
+ print(f"Client {self.node_id} failed to attack {neighbor_id} in round {round}!")
continue
# Mark this neighbor as attacked
self.attacked_neighbors.add(neighbor_id)
- # Optionally, clear the stored updates to save memory
+ # clear the stored updates to save memory
del self.neighbor_updates[neighbor_id]
+ else:
+ self.log_utils.log_summary(f"Client {self.node_id} failed to attack {neighbor_id} in round {round}!")
+ print("(DEBUG): neighbor updates that I cannot recover: ", [len(self.neighbor_updates[client]) for client in self.neighbor_updates])
+ print(f"Client {self.node_id} failed to attack {neighbor_id} in round {round}!")
self.aggregate(model_updates, keys_to_ignore=self.model_keys_to_ignore)
@@ -941,8 +1103,8 @@ def set_data_parameters(self, config: Dict[str, Any]) -> None:
if "gia" not in config:
self._test_loader = DataLoader(test_dset, batch_size=batch_size)
else:
- _, test_data, labels, indices = gia_client_dataset(self.dset_obj.train_dset, test_dset)
- self._test_loader = DataLoader(test_data, batch_size=10)
+ _, test_data, labels, indices = gia_client_dataset(self.dset_obj.train_dset, test_dset, num_labels=5) # only using five images
+ self._test_loader = DataLoader(test_data, batch_size=1)
def aggregate(
self, representation_list: List[OrderedDict[str, Any]], **kwargs: Any
) -> OrderedDict[str, Tensor]:
@@ -1003,7 +1165,6 @@ def __init__(
keys = self.model_utils.get_last_layer_keys(self.get_model_weights())
self.model_keys_to_ignore.extend(keys)
-
def local_test(self, **kwargs: Any) -> Tuple[float, float]:
"""
Test the model locally, not to be used in the traditional FedAvg
@@ -1177,9 +1338,38 @@ def receive_and_aggregate_streaming(self, neighbors: List[int]) -> None:
# Update the model with the aggregated weights
self.set_model_weights(agg_wts)
- def receive_and_aggregate(self, neighbors: List[int]) -> None:
- if hasattr(self, "gia_attacker"):
+ def receive_mia_and_aggregate(self, neighbors: List[int], current_round: int, num_neighbors: int) -> None:
+ """
+ performs MIA attack on the received models
+ """
+ if self.is_working:
+ # first perform self-attack and log the results
+ in_size = len(self.dloader.dataset)
+ out_size = len(self._test_loader.dataset)
+
+ attack_vals, metadata = self.mia_func(self.model, self.dloader, self._test_loader, device=self.device)
+
+ for metric, score in attack_vals.items():
+ print(f"{metric}: {score:.4f}")
+
+ # Log all metrics
+ self.log_utils.log_mia_stats(attack_vals, current_round)
+ self.log_utils.log_mia_stats(metadata, current_round, metadata=True)
+ self.log_utils.log_tb_mia_stats(attack_vals, current_round, self.node_id)
+
+ model_updates = self.comm_utils.receive(node_ids=neighbors)
+ self.aggregate(model_updates, keys_to_ignore=self.model_keys_to_ignore)
+
+
+ def receive_and_aggregate(self, neighbors: List[int], it:int=0) -> None:
+ if hasattr(self, "gia_attacker") and not self.gia_disaggregate:
self.receive_attack_and_aggregate(neighbors, it, len(neighbors))
+ elif hasattr(self, "gia_attacker") and self.gia_disaggregate:
+ self.receive_disambiguate_attack_and_aggregate(neighbors, it, self.n_iter, len(neighbors))
+ elif hasattr(self, "mia"):
+ # if it in self.mia_rounds:
+ print(f"Client {self.node_id} performing MIA attack in round {it}")
+ self.receive_mia_and_aggregate(neighbors, it, len(neighbors))
if self.streaming_aggregation:
self.receive_and_aggregate_streaming(neighbors)
else:
diff --git a/src/algos/fl_static.py b/src/algos/fl_static.py
index 4a19f936..fa61db28 100644
--- a/src/algos/fl_static.py
+++ b/src/algos/fl_static.py
@@ -1,7 +1,7 @@
"""
Module for FedStaticClient and FedStaticServer in Federated Learning.
"""
-from typing import Any, Dict, OrderedDict, List
+from typing import Any, Dict, OrderedDict, List, Tuple
from collections import OrderedDict, defaultdict
from utils.communication.comm_utils import CommunicationManager
@@ -38,7 +38,8 @@ def get_neighbors(self) -> List[int]:
self.stats["neighbors"] = neighbors # type: ignore, where the hell self.stats is coming from
return neighbors
-
+
+
def run_protocol(self) -> None:
"""
Runs the federated learning protocol for the client.
@@ -64,12 +65,12 @@ def run_protocol(self) -> None:
neighbors = self.get_neighbors()
# TODO: Log the neighbors
- self.receive_and_aggregate(neighbors)
+ self.receive_and_aggregate(neighbors, it)
# evaluate the model on the test data
# Inside FedStaticNode.run_protocol()
self.local_test()
-
self.round_finalize()
+ # self.local_round_done()
class FedStaticServer(BaseFedAvgClient):
"""
diff --git a/src/algos/topologies/base.py b/src/algos/topologies/base.py
index 086c67df..44a49e6d 100644
--- a/src/algos/topologies/base.py
+++ b/src/algos/topologies/base.py
@@ -45,11 +45,16 @@ def initialize(self) -> None:
self.generate_graph()
self._convert_labels_to_int()
+ # if graph is not fully conncted, print warning in red
+ if not nx.is_connected(self.graph):
+ print("\033[91m" + "Warning: Graph is not fully connected" + "\033[0m")
+
def get_all_neighbours(self) -> List[int]:
"""
Returns the list of neighbours of the current node
"""
# get all neighbours of the current node using the self.graph attribute
+ # NOTE: graph is 1-indexed, but our node IDs are 0-indexed
if self.graph is None:
raise ValueError("Graph not initialized")
return list(self.graph.neighbors(self.rank)) # type: ignore
@@ -104,3 +109,30 @@ def get_neighbourhood_size(self) -> int:
if self.graph is None:
raise ValueError("Graph not initialized")
return len(self.get_all_neighbours())
+
+ def calculate_graph_metrics(self, target_node:int) -> dict[str, int | dict[int, int]]:
+ G = self.graph
+ # Calculate graph density
+ density = nx.density(G)
+ print(f"Graph Density: {density}")
+
+ # Calculate shortest path lengths from every node to the target node
+ shortest_paths = nx.shortest_path_length(G, target=target_node)
+ print(f"Shortest Path Lengths to Target Node {target_node}: {shortest_paths}")
+
+ # Calculate centrality metrics
+ eigenvector_centrality = nx.eigenvector_centrality(G)
+ closeness_centrality = nx.closeness_centrality(G)
+ degree_centrality = nx.degree_centrality(G)
+
+ print("Eigenvector Centrality:", eigenvector_centrality)
+ print("Closeness Centrality:", closeness_centrality)
+ print("Degree Centrality:", degree_centrality)
+
+ return {
+ "density": density,
+ "shortest_paths": shortest_paths,
+ "eigenvector_centrality": eigenvector_centrality,
+ "closeness_centrality": closeness_centrality,
+ "degree_centrality": degree_centrality
+ }
diff --git a/src/algos/topologies/collections.py b/src/algos/topologies/collections.py
index b7cc03ef..fe3e4002 100644
--- a/src/algos/topologies/collections.py
+++ b/src/algos/topologies/collections.py
@@ -117,9 +117,15 @@ def __init__(self, config: ConfigType, rank: int):
self.p = p
self.seed = config["seed"]
- def generate_graph(self) -> None:
- self.graph = nx.erdos_renyi_graph(self.num_users, self.p, self.seed)
-
+ def generate_graph(self, full:bool=False) -> None:
+ """ if full is True, generate a fully connected graph """
+ if full:
+ while True:
+ self.graph = nx.erdos_renyi_graph(self.num_users, self.p, seed=self.seed)
+ if nx.is_connected(self.graph):
+ break
+ else:
+ self.graph = nx.erdos_renyi_graph(self.num_users, self.p, seed=self.seed)
class WattsStrogatzTopology(BaseTopology):
def __init__(self, config: ConfigType, rank: int):
@@ -145,6 +151,17 @@ def __init__(self, config: ConfigType, rank: int):
def generate_graph(self) -> None:
self.graph = nx.random_regular_graph(self.d, self.num_users, self.seed) # type: ignore
+class LineTopology(BaseTopology):
+ """ test topology for debugging gradient disambiguation attack """
+ def __init__(self, config: ConfigType, rank: int):
+ super().__init__(config, rank)
+
+ def generate_graph(self) -> None:
+ self.graph = nx.Graph()
+ self.graph.add_node(0)
+ for i in range(1, self.num_users):
+ self.graph.add_node(i)
+ self.graph.add_edge(i-1, i)
class DynamicGraph(BaseTopology):
def __init__(self, config: ConfigType, rank: int):
@@ -277,6 +294,8 @@ def select_topology(config: ConfigType, rank: int) -> BaseTopology:
return RandomRegularTopology(config, rank)
if topology_name == "barbell":
return BarbellTopology(config, rank)
+ if topology_name == "line":
+ return LineTopology(config, rank)
if topology_name == "one_peer_exponential":
return OnePeerExponentialTopology(config, rank)
if topology_name == "hyper_hypercube":
diff --git a/src/configs/algo_config.py b/src/configs/algo_config.py
index b84ce565..a0a3f4ad 100644
--- a/src/configs/algo_config.py
+++ b/src/configs/algo_config.py
@@ -31,8 +31,7 @@ def get_malicious_types(malicious_config_list: List[ConfigType]) -> Dict[str, st
traditional_fl: ConfigType = {
# Collaboration setup
"algo": "fedavg",
- "rounds": 2,
-
+ "rounds": 5,
# Model parameters
"model": "resnet10",
"model_lr": 3e-4,
@@ -197,7 +196,7 @@ def get_malicious_types(malicious_config_list: List[ConfigType]) -> Dict[str, st
"average_last_layer": True,
"mask_finetune_last_layer": False,
# params for model
- "position": 0,
+ "position": 0,
}
fedstatic: ConfigType = {
@@ -207,9 +206,9 @@ def get_malicious_types(malicious_config_list: List[ConfigType]) -> Dict[str, st
# "topology": {"name": "base_graph", "max_degree": 2}, # type: ignore
"rounds": 3,
# Model parameters
- "optimizer": "sgd", # TODO comment out for real training
+ "optimizer": "sgd",
"model": "resnet10",
- "model_lr": 3e-4,
+ "model_lr": 0.1, # lr for sgd
"batch_size": 256,
}
diff --git a/src/configs/sys_config.py b/src/configs/sys_config.py
index b305d3d4..c2f75fd0 100644
--- a/src/configs/sys_config.py
+++ b/src/configs/sys_config.py
@@ -158,6 +158,8 @@ def get_digit_five_support(num_users: int, domains: List[str] = DIGIT_FIVE):
CIFAR10_DSET = "cifar10"
CIAR10_DPATH = "./datasets/imgs/cifar10/"
+MNIST_DEST = "mnist"
+MNIST_DPATH = "./datasets/imgs/mnist/"
NUM_COLLABORATORS = 3
DUMP_DIR = "/tmp/new_sonar/"
@@ -340,7 +342,7 @@ def get_digit_five_support(num_users: int, domains: List[str] = DIGIT_FIVE):
"num_collaborators": NUM_COLLABORATORS,
"comm": {"type": "GRPC", "synchronous": True, "peer_ids": ["localhost:32048"]}, # The super-node
"dset": CIFAR10_DSET,
- "dump_dir": DUMP_DIR,
+ "dump_dir": f"{DUMP_DIR}test/ring_",
"dpath": CIAR10_DPATH,
"seed": 2,
"device_ids": get_device_ids(num_users, gpu_ids),
@@ -361,21 +363,63 @@ def get_digit_five_support(num_users: int, domains: List[str] = DIGIT_FIVE):
"num_users": num_users,
"num_collaborators": NUM_COLLABORATORS,
"comm": {"type": "GRPC", "synchronous": True, "peer_ids": ["localhost:50048"]}, # The super-node
- "dset": CIFAR10_DSET,
- "dump_dir": DUMP_DIR,
- "dpath": CIAR10_DPATH,
+ "dset": MNIST_DEST,
+ "dump_dir": f"{DUMP_DIR}disambiguate/torus_",
+ # "dpath": CIAR10_DPATH,
+ "dpath": MNIST_DPATH,
"seed": 2,
"device_ids": get_device_ids(num_users, gpu_ids),
- # "algos": get_algo_configs(num_users=num_users, algo_configs=default_config_list), # type: ignore
"algos": get_algo_configs(num_users=num_users, algo_configs=[fedstatic]), # type: ignore
"samples_per_user": 50000 // num_users, # distributed equally
"train_label_distribution": "iid",
"test_label_distribution": "iid",
"exp_keys": [],
+ "assign_based_on_host": True,
+ "hostname_to_device_ids": {
+ "matlaber1": [2, 3, 4, 5, 6, 7],
+ "matlaber12": [0, 1, 2, 3],
+ "matlaber3": [0, 1, 2, 3],
+ "matlaber4": [0, 2, 3, 4, 5, 6, 7],
+ },
"dropout_dicts": dropout_dicts,
"gia":True,
- "gia_attackers":[1]
+ "gia_attackers":[1],
+ "disaggregate": True
+}
+
+grpc_system_config_mia: ConfigType = {
+ "exp_id": "static",
+ "num_users": num_users,
+ "num_collaborators": NUM_COLLABORATORS,
+ "comm": {"type": "GRPC", "synchronous": True, "peer_ids": ["localhost:50048"]}, # The super-node
+ "dset": CIFAR10_DSET,
+ "dump_dir": f"{DUMP_DIR}mia/SGD/1.0/er_",
+ "dpath": CIFAR10_DSET,
+ "seed": 2,
+ "device_ids": get_device_ids(num_users, gpu_ids),
+ "algos": get_algo_configs(num_users=num_users, algo_configs=[fedstatic]), # type: ignore
+ "samples_per_user": 50000 // num_users, # distributed equally
+ "train_label_distribution": "non_iid",
+ "test_label_distribution": "iid",
+ "alpha_data": 1.0,
+ "test_samples_per_user": 500,
+ "exp_keys": [],
+ "assign_based_on_host": True,
+ "hostname_to_device_ids": {
+ "matlaber1": [2, 3, 4, 5, 6, 7],
+ "matlaber12": [0, 1, 2, 3],
+ "matlaber3": [0, 1, 2, 3],
+ "matlaber4": [0, 2, 3, 4, 5, 6, 7],
+ "matlaberp4": [0,1],
+ "matlaberp1": [0,1]
+
+ },
+ "dropout_dicts": dropout_dicts,
+ "mia":True,
}
-current_config = grpc_system_config
-# current_config = mpi_system_config
+
+# current_config = grpc_system_config
+current_config = mpi_system_config
+# current_config = grpc_system_config_gia
+# current_config = grpc_system_config_mia
diff --git a/src/data_loaders/mnist.py b/src/data_loaders/mnist.py
index 7fdc08b7..f43ddca7 100644
--- a/src/data_loaders/mnist.py
+++ b/src/data_loaders/mnist.py
@@ -8,6 +8,7 @@ class MNISTDataset:
"""
def __init__(self, dpath: str) -> None:
+ print(f"DEBUG dp: {dpath}")
self.image_size = 28
self.num_cls = 10
self.mean = 0.1307
@@ -27,8 +28,8 @@ def __init__(self, dpath: str) -> None:
]
)
self.train_dset = MNIST(
- root=dpath, train=True, download=True, transform=self.train_transform
+ root=dpath, train=True, download=False, transform=self.train_transform
)
self.test_dset = MNIST(
- root=dpath, train=False, download=True, transform=self.test_transform
+ root=dpath, train=False, download=False, transform=self.test_transform
)
diff --git a/src/inversefed/data/data_processing.py b/src/inversefed/data/data_processing.py
index f6e36e3a..2b622538 100644
--- a/src/inversefed/data/data_processing.py
+++ b/src/inversefed/data/data_processing.py
@@ -206,4 +206,4 @@ def _get_meanstd(dataset):
cc = torch.cat([trainset[i][0].reshape(3, -1) for i in range(len(trainset))], dim=1)
data_mean = torch.mean(cc, dim=1).tolist()
data_std = torch.std(cc, dim=1).tolist()
- return data_mean, data_std
\ No newline at end of file
+ return data_mean, data_std
diff --git a/src/inversefed/data/loss.py b/src/inversefed/data/loss.py
index f43ce933..f01b936c 100644
--- a/src/inversefed/data/loss.py
+++ b/src/inversefed/data/loss.py
@@ -111,4 +111,4 @@ def metric(self, x=None, y=None):
return name, format
else:
value = (x.data.argmax(dim=1) == y).sum().float() / y.shape[0]
- return value.detach(), name, format
\ No newline at end of file
+ return value.detach(), name, format
diff --git a/src/inversefed/reconstruction_algorithms.py b/src/inversefed/reconstruction_algorithms.py
index 63965d85..9b452efb 100644
--- a/src/inversefed/reconstruction_algorithms.py
+++ b/src/inversefed/reconstruction_algorithms.py
@@ -86,7 +86,7 @@ def loss_fn(pred, labels):
return torch.mean(torch.sum(- labels * torch.nn.functional.log_softmax(pred, dim=-1), 1))
self.loss_fn = loss_fn
else:
- assert labels.shape[0] == self.num_images
+ assert labels.shape[0] == self.num_images, f"Labels shape {labels.shape} does not match num_images {self.num_images}."
self.reconstruct_label = False
try:
@@ -109,6 +109,10 @@ def loss_fn(pred, labels):
else:
print('Choosing optimal result ...')
scores = scores[torch.isfinite(scores)] # guard against NaN/-Inf scores?
+ if scores.numel() == 0:
+ print("All scores are NaN or -Inf. Cannot determine the optimal result.")
+ return None # or handle the situation as needed
+
optimal_index = torch.argmin(scores)
print(f'Optimal result score: {scores[optimal_index]:2.4f}')
stats['opt'] = scores[optimal_index].item()
@@ -128,6 +132,13 @@ def _init_images(self, img_shape):
raise ValueError()
def _run_trial(self, x_trial, input_data, labels, dryrun=False):
+ # Print data types
+ print(f"Input data type: {type(input_data)}")
+ if isinstance(input_data, torch.Tensor):
+ print(f"Input data tensor dtype: {input_data.dtype}, shape: {input_data.shape}")
+ if isinstance(labels, torch.Tensor):
+ print(f"Labels tensor dtype: {labels.dtype}, shape: {labels.shape}, labels value: {labels}")
+
x_trial.requires_grad = True
if self.reconstruct_label:
output_test = self.model(x_trial)
diff --git a/src/main.py b/src/main.py
index fe4aaa7b..c045b550 100644
--- a/src/main.py
+++ b/src/main.py
@@ -45,10 +45,10 @@
)
args: argparse.Namespace = parser.parse_args()
-
scheduler: Scheduler = Scheduler()
-
# Assign the configuration from the file paths provided via arguments
+print(f"arg s {args.s}")
+print(f"all args {args}")
scheduler.assign_config_by_path(args.s, args.b, args.super, args.host)
print("Config loaded")
diff --git a/src/main_grpc.py b/src/main_grpc.py
index e3678431..eb4dd5f5 100644
--- a/src/main_grpc.py
+++ b/src/main_grpc.py
@@ -35,10 +35,10 @@
# Command for opening each process
command_list: List[str] = ["python", "main.py", "-host", args.host]
if args.dev == True:
- command_list: List[str] = ["python", "main.py", "-b", "./configs/algo_config_test.py", "-s", "./configs/sys_config_test.py", "-host", args.host]
+ print("here!")
+ command_list: List[str] = ["python", "main.py", "-b", "./configs/algo_config_2.py", "-s", "./configs/sys_config_2.py", "-host", args.host]
# Start process for each user
for i in range(args.n):
- print(f"Starting process for user {i}")
# start a Popen process
subprocess.Popen(command_list)
diff --git a/src/plot.ipynb b/src/plot.ipynb
new file mode 100644
index 00000000..aa2b6d7c
--- /dev/null
+++ b/src/plot.ipynb
@@ -0,0 +1,576 @@
+{
+ "cells": [
+ {
+ "cell_type": "code",
+ "execution_count": 1,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "import pandas as pd\n",
+ "import json\n",
+ "import os"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 2,
+ "metadata": {},
+ "outputs": [
+ {
+ "data": {
+ "text/plain": [
+ "'/u/yshi23/sonar/src'"
+ ]
+ },
+ "execution_count": 2,
+ "metadata": {},
+ "output_type": "execute_result"
+ }
+ ],
+ "source": [
+ "os.getcwd()"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 3,
+ "metadata": {},
+ "outputs": [
+ {
+ "name": "stdout",
+ "output_type": "stream",
+ "text": [
+ "expt_dump/disambiguate/0.6mnist_36users_1388_static_seed2\n",
+ "expt_dump/disambiguate/ring_mnist_36users_1388_static_seed2\n",
+ "expt_dump/disambiguate/0.2mnist_36users_1388_static_seed2\n",
+ "expt_dump/disambiguate/0.4mnist_36users_1388_static_seed2\n",
+ "expt_dump/disambiguate/torus_mnist_36users_1388_static_seed2\n",
+ "expt_dump/disambiguate/1.0mnist_36users_1388_static_seed2\n",
+ "expt_dump/disambiguate/0.9mnist_36users_1388_static_seed2\n",
+ "expt_dump/disambiguate/fc_mnist_36users_1388_static_seed2\n",
+ "expt_dump/disambiguate/0.7mnist_36users_1388_static_seed2\n",
+ "expt_dump/disambiguate/0.3mnist_36users_1388_static_seed2\n",
+ "expt_dump/disambiguate/0.8mnist_36users_1388_static_seed2\n",
+ "expt_dump/disambiguate/0.5mnist_36users_1388_static_seed2\n"
+ ]
+ }
+ ],
+ "source": [
+ "def extract_data_from_subfolders(base_dir):\n",
+ " all_data = []\n",
+ "\n",
+ " # Iterate over each subfolder in the base directory\n",
+ " for subfolder in os.listdir(base_dir):\n",
+ " subfolder_path = os.path.join(base_dir, subfolder)\n",
+ " \n",
+ " if os.path.isdir(subfolder_path):\n",
+ " print(subfolder_path)\n",
+ " try:\n",
+ " # Extract the value of p from the subfolder name\n",
+ " p_value = float(subfolder.split('mnist')[0])\n",
+ " except ValueError:\n",
+ " # Skip the subfolder if the p value is not a float\n",
+ " continue\n",
+ "\n",
+ " # Load the JSON data\n",
+ " gia_stat_path = os.path.join(subfolder_path, 'logs/node_1/gia_stat_summary.json')\n",
+ " network_stats_path = os.path.join(subfolder_path, 'logs/node_1/network_stats.json')\n",
+ "\n",
+ " if os.path.exists(gia_stat_path) and os.path.exists(network_stats_path):\n",
+ " with open(gia_stat_path, 'r') as f:\n",
+ " gia_data = json.load(f)\n",
+ "\n",
+ " with open(network_stats_path, 'r') as f:\n",
+ " network_data = json.load(f)\n",
+ "\n",
+ " # Extract node IDs from network_stats.json\n",
+ " node_ids = list(network_data['shortest_paths'].keys())\n",
+ "\n",
+ " # Create a DataFrame with node_id as a column\n",
+ " data = []\n",
+ " for node_id in node_ids:\n",
+ " # Get PSNR value from gia_data, defaulting to 0 if node_id doesn't exist\n",
+ " psnr = gia_data.get(node_id, {}).get('test_psnr', 0)\n",
+ "\n",
+ " # Append the row data\n",
+ " row = {\n",
+ " 'node_id': int(node_id), # Ensure node_id is an integer\n",
+ " 'PSNR': psnr,\n",
+ " 'density': network_data['density'],\n",
+ " 'shortest_paths': network_data['shortest_paths'][node_id],\n",
+ " 'eigenvector_centrality': network_data['eigenvector_centrality'][node_id],\n",
+ " 'closeness_centrality': network_data['closeness_centrality'][node_id],\n",
+ " 'degree_centrality': network_data['degree_centrality'][node_id],\n",
+ " 'p': p_value\n",
+ " }\n",
+ " data.append(row)\n",
+ "\n",
+ " # Create a DataFrame from the collected data\n",
+ " df = pd.DataFrame(data)\n",
+ "\n",
+ " # Append the DataFrame to the list\n",
+ " all_data.append(df)\n",
+ "\n",
+ " # Concatenate all DataFrames\n",
+ " final_df = pd.concat(all_data, ignore_index=True)\n",
+ "\n",
+ " return final_df\n",
+ "\n",
+ "# Example usage\n",
+ "base_dir = 'expt_dump/disambiguate'\n",
+ "final_df = extract_data_from_subfolders(base_dir)\n",
+ "\n",
+ "# Save the final DataFrame to a CSV file\n",
+ "final_df.to_csv('gia_stat.csv')"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 4,
+ "metadata": {},
+ "outputs": [
+ {
+ "name": "stdout",
+ "output_type": "stream",
+ "text": [
+ " node_id eigenvector_centrality RCR_eigenvector closeness_centrality \\\n",
+ "0 1 0.174753 0.174753 0.729167 \n",
+ "1 4 0.161298 0.161298 0.700000 \n",
+ "2 5 0.191581 0.191581 0.760870 \n",
+ "3 9 0.174332 0.174332 0.729167 \n",
+ "4 12 0.191878 0.191878 0.760870 \n",
+ "\n",
+ " RCR_closeness degree_centrality RCR_degree \n",
+ "0 0.729167 0.628571 0.628571 \n",
+ "1 0.700000 0.571429 0.571429 \n",
+ "2 0.760870 0.685714 0.685714 \n",
+ "3 0.729167 0.628571 0.628571 \n",
+ "4 0.760870 0.685714 0.685714 \n"
+ ]
+ }
+ ],
+ "source": [
+ "# Calculate attacker centrality (node_id = 1)\n",
+ "attacker_eigenvector_centrality = final_df[final_df['node_id'] == 1]['eigenvector_centrality'].values[0]\n",
+ "attacker_closeness_centrality = final_df[final_df['node_id'] == 1]['closeness_centrality'].values[0]\n",
+ "attacker_degree_centrality = final_df[final_df['node_id'] == 1]['degree_centrality'].values[0]\n",
+ "\n",
+ "# Calculate Relative Centrality Ratio (RCR) for each centrality measure\n",
+ "final_df['RCR_eigenvector'] = final_df['eigenvector_centrality'] \n",
+ "final_df['RCR_closeness'] = final_df['closeness_centrality'] \n",
+ "final_df['RCR_degree'] = final_df['degree_centrality']\n",
+ "# final_df['RCR_eigenvector'] = final_df['eigenvector_centrality'] / attacker_eigenvector_centrality\n",
+ "# final_df['RCR_closeness'] = final_df['closeness_centrality'] / attacker_closeness_centrality\n",
+ "# final_df['RCR_degree'] = final_df['degree_centrality'] / attacker_degree_centrality\n",
+ "\n",
+ "# Display the updated DataFrame\n",
+ "print(final_df[['node_id', 'eigenvector_centrality', 'RCR_eigenvector', 'closeness_centrality', 'RCR_closeness', 'degree_centrality', 'RCR_degree']].head())"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 5,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "import seaborn as sns\n",
+ "import matplotlib.pyplot as plt\n",
+ "\n"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 6,
+ "metadata": {},
+ "outputs": [
+ {
+ "data": {
+ "text/html": [
+ "\n",
+ "\n",
+ "
\n",
+ " \n",
+ " \n",
+ " | \n",
+ " node_id | \n",
+ " PSNR | \n",
+ " density | \n",
+ " shortest_paths | \n",
+ " eigenvector_centrality | \n",
+ " closeness_centrality | \n",
+ " degree_centrality | \n",
+ " p | \n",
+ " RCR_eigenvector | \n",
+ " RCR_closeness | \n",
+ " RCR_degree | \n",
+ "
\n",
+ " \n",
+ " \n",
+ " \n",
+ " | 0 | \n",
+ " 1 | \n",
+ " 0.000000 | \n",
+ " 0.593651 | \n",
+ " 0 | \n",
+ " 0.174753 | \n",
+ " 0.729167 | \n",
+ " 0.628571 | \n",
+ " 0.6 | \n",
+ " 0.174753 | \n",
+ " 0.729167 | \n",
+ " 0.628571 | \n",
+ "
\n",
+ " \n",
+ " | 1 | \n",
+ " 4 | \n",
+ " 6.713571 | \n",
+ " 0.593651 | \n",
+ " 1 | \n",
+ " 0.161298 | \n",
+ " 0.700000 | \n",
+ " 0.571429 | \n",
+ " 0.6 | \n",
+ " 0.161298 | \n",
+ " 0.700000 | \n",
+ " 0.571429 | \n",
+ "
\n",
+ " \n",
+ " | 2 | \n",
+ " 5 | \n",
+ " 12.390618 | \n",
+ " 0.593651 | \n",
+ " 1 | \n",
+ " 0.191581 | \n",
+ " 0.760870 | \n",
+ " 0.685714 | \n",
+ " 0.6 | \n",
+ " 0.191581 | \n",
+ " 0.760870 | \n",
+ " 0.685714 | \n",
+ "
\n",
+ " \n",
+ " | 3 | \n",
+ " 9 | \n",
+ " 9.810331 | \n",
+ " 0.593651 | \n",
+ " 1 | \n",
+ " 0.174332 | \n",
+ " 0.729167 | \n",
+ " 0.628571 | \n",
+ " 0.6 | \n",
+ " 0.174332 | \n",
+ " 0.729167 | \n",
+ " 0.628571 | \n",
+ "
\n",
+ " \n",
+ " | 4 | \n",
+ " 12 | \n",
+ " 11.033689 | \n",
+ " 0.593651 | \n",
+ " 1 | \n",
+ " 0.191878 | \n",
+ " 0.760870 | \n",
+ " 0.685714 | \n",
+ " 0.6 | \n",
+ " 0.191878 | \n",
+ " 0.760870 | \n",
+ " 0.685714 | \n",
+ "
\n",
+ " \n",
+ "
\n",
+ "