• 基于cross_silo做联邦学习编程的学习

    框架是fedml

    我是一名初学者，若是有也研究联邦学习的朋友看见这篇博文，欢迎私信或者加我好友，一起讨论一起学习。

    我从client开始学习

    先是客户端初始化，clientinitialize

    1. from fedml.constants import FEDML_CROSS_SILO_SCENARIO_HIERARCHICAL, FEDML_CROSS_SILO_SCENARIO_HORIZONTAL
    2. from .fedml_client_master_manager import ClientMasterManager
    3. from .fedml_trainer_dist_adapter import TrainerDistAdapter
    4.  
    5.  
    6. def init_client(
    7.         args,
    8.         device,
    9.         comm,
    10.         client_rank,
    11.         client_num,
    12.         model,
    13.         train_data_num,
    14.         train_data_local_num_dict,
    15.         train_data_local_dict,
    16.         test_data_local_dict,
    17.         model_trainer=None,
    18. ):
    19.     backend=args.backend
    20.     trainer_dist_adapter=get_trainer_dist_adapter(
    21.         args,
    22.         device,
    23.         client_rank,
    24.         model,
    25.         train_data_num,
    26.         train_data_local_num_dict,
    27.         train_data_local_dict,
    28.         test_data_local_dict,
    29.         model_trainer,
    30.  
    31.     )
    32.     if args.scenario == FEDML_CROSS_SILO_SCENARIO_HIERARCHICAL:#垂直划分
    33.         if args.proc_rank_in_silo == 0:
    34.             client_manager=get_client_manager_master(
    35.                 args,trainer_dist_adapter,comm,client_rank, client_num, backend
    36.             )
    37.         else:
    38.             client_manager=get_client_manager_salve(args, trainer_dist_adapter)
    39.  
    40.     elif args.scenario == FEDML_CROSS_SILO_SCENARIO_HORIZONTAL:#水平划分
    41.         client_manager=get_client_manager_master(
    42.             args,trainer_dist_adapter,comm,client_rank,client_num,backend
    43.         )
    44.     else:
    45.         raise Exception(
    46.             "we do not support {}. Please check whether this is typo.".format(
    47.                 args.scenario
    48.             )
    49.         )
    50.     client_manager.run()#配置好了客户端的管理，开始运行
    51. def get_trainer_dist_adapter(
    52.         args,
    53.         device,
    54.         client_rank,
    55.         model,
    56.         train_data_num,
    57.         train_data_local_num_dict,
    58.         train_data_local_dict,
    59.         test_data_local_dict,
    60.         model_trainer,
    61. ):
    62.     return TrainDistAdapter(
    63.         args,
    64.         device,
    65.         client_rank,
    66.         model,
    67.         train_data_num,
    68.         train_data_local_num_dict,
    69.         train_data_local_dict,
    70.         test_data_local_dict,
    71.         model_trainer,
    72.     )
    73. def get_client_manager_master(
    74.         args,trainer_dist_adapter,comm,client_rank,client_num,backend
    75. ):
    76.     return ClientMasterManager(#这个函数后文还有出现
    77.         args,trainer_dist_adapter,comm,client_rank,client_num,backend
    78.     )
    79. def get_client_manager_salve(args,trainer_dist_adapter):
    80.     from .fedml_client_slave_manager import ClientSlaveManager
    81.     return ClientSlaveManager(args,trainer_dist_adapter)
    82. #这两个函数后续都有说明

    2.fedml_client_master_manager.py

    这个用于横向联邦学习，另外一个用于纵向，因为我要做的是横向，所以我需要这个，要求后台可以支持mpi多机并行运算，有一个参数comm之前一直不知道他是什么意思，现在这个文件，感觉应该是comm_round全局迭代轮数

    这个是我认为比较重要的文件之一，在这个上面做一些改动

    1. import json
    2. import logging
    3. import platform
    4. import time
    5. import torch.distributed as dist
    6.  
    7. from fedml.constants import FEDML_CROSS_SILO_SCENARIO_HIERARCHICAL
    8. from .message_define import MyMessage
    9. from .utils import convert_model_params_from_ddp, convert_model_params_to_ddp
    10. from ...core.distributed.client.client_manager import ClientManager
    11. from ...core.distributed.communication.message import Message
    12. from ...core.mlops.mlops_metrics import MLOpsMetrics
    13. from ...core.mlops.mlops_profiler_event import MLOpsProfilerEvent
    14.  
    15. class ClientMasterManager(ClientManager):
    16.     def __init__(
    17.             self,args,trainer_dist_adapter,comm=None,rank=0,size=0,backend="MPI"
    18.     ):
    19.         super().__init__(args,comm,rank,size,backend)
    20.         self.trainer_dist_adapter=trainer_dist_adapter
    21.         self.args=args
    22.         self.num_rounds=args.comm_round
    23.         self.round_idx=0
    24.         self.rank=rank
    25.         self.client_real_ids=json.loads(args.client_id_list)
    26.         #读取客户端的id
    27.         logging.info("self.client_real_ids = {}".format(self.client_real_ids))
    28.         # for the client, len(self.client_real_ids)==1: we only specify its client id in the list, not including others.
    29.         self.client_real_id = self.client_real_ids[0]
    30.         if hasattr(self.args, "using_mlops") and self.args.using_mlops:
    31.             self.mlops_metrics = MLOpsMetrics()
    32.             self.mlops_metrics.set_messenger(self.com_manager_status, args)
    33.             self.mlops_event = MLOpsProfilerEvent(self.args)
    34.         #判断是否适用mlops
    35.  
    36. #登记收信的操作者
    37.     def register_message_receive_handlers(self):
    38.         self.register_message_receive_handler(
    39.             MyMessage.MSG_TYPE_CONNECTION_IS_READY, self.handle_message_connection_ready
    40.         )
    41.  
    42.         self.register_message_receive_handler(
    43.             MyMessage.MSG_TYPE_S2C_CHECK_CLIENT_STATUS, self.handle_message_check_status
    44.         )
    45.  
    46.         self.register_message_receive_handler(
    47.             MyMessage.MSG_TYPE_S2C_INIT_CONFIG, self.handle_message_init
    48.         )
    49.         self.register_message_receive_handler(
    50.             MyMessage.MSG_TYPE_S2C_SYNC_MODEL_TO_CLIENT,
    51.             self.handle_message_receive_model_from_server,
    52.         )
    53.  
    54.         self.register_message_receive_handler(
    55.             MyMessage.MSG_TYPE_S2C_FINISH, self.handle_message_finish,
    56.         )
    57.     def handle_message_connection_ready(self, msg_params):
    58.         logging.info("Connection is ready!")
    59.         if not self.has_sent_online_msg:
    60.             self.has_sent_online_msg = True
    61.             self.send_client_status(0)
    62.  
    63.             if hasattr(self.args, "using_mlops") and self.args.using_mlops:
    64.                 # Notify MLOps with training status.
    65.                 self.report_training_status(
    66.                     MyMessage.MSG_MLOPS_CLIENT_STATUS_INITIALIZING
    67.                 )
    68.  
    69.                 # Open new process for report system performances to MQTT server
    70.                 MLOpsMetrics.report_sys_perf(self.args)
    71.  
    72.     def handle_message_check_status(self, msg_params):
    73.         self.send_client_status(0)
    74.     def handle_messsage_init(self,msg_params):
    75.         global_model_params=msg_params.get(MyMessage.MSG_ARG_KEY_MODEL_PARAMS)
    76.         data_silo_index = msg_params.get(MyMessage.MSG_ARG_KEY_CLIENT_INDEX)
    77.         logging.info("data_silo_index = %s" % str(data_silo_index))
    78.         self.report_training_status(MyMessage.MSG_MLOPS_CLIENT_STATUS_TRAINING)
    79.  
    80.         if self.args.scenario == FEDML_CROSS_SILO_SCENARIO_HIERARCHICAL:
    81.             global_model_params = convert_model_params_to_ddp(global_model_params)
    82.             self.sync_process_group(0, global_model_params, data_silo_index)
    83.  
    84.         self.trainer_dist_adapter.update_model(global_model_params)
    85.         self.trainer_dist_adapter.update_dataset(int(data_silo_index))
    86.         self.round_idx = 0
    87.  
    88.         self.__train()
    89.     def  handle_message_receive_model_from_server(self, msg_params):
    90.         logging.info("handle_message_receive_model_from_server.")
    91.         model_params = msg_params.get(MyMessage.MSG_ARG_KEY_MODEL_PARAMS)
    92.         client_index = msg_params.get(MyMessage.MSG_ARG_KEY_CLIENT_INDEX)
    93.         if self.args.scenario == FEDML_CROSS_SILO_SCENARIO_HIERARCHICAL:
    94.             model_params = convert_model_params_to_ddp(model_params)
    95.             self.sync_process_group(self.round_idx, model_params, client_index)
    96.  
    97.         self.trainer_dist_adapter.update_model(model_params)
    98.         self.trainer_dist_adapter.update_dataset(int(client_index))
    99.         if self.round_idx == self.num_rounds - 1:
    100.             # 这里可能需要动
    101.  
    102.             # Notify MLOps with the finished message
    103.             if hasattr(self.args, "using_mlops") and self.args.using_mlops:
    104.                 self.mlops_metrics.report_client_id_status(
    105.                     self.args.run_id,
    106.                     self.client_real_id,
    107.                     MyMessage.MSG_MLOPS_CLIENT_STATUS_FINISHED,
    108.                 )
    109.             return
    110.         self.round_idx += 1
    111.         self.__train()
    112.     def handle_message_finish(self, msg_params):
    113.         logging.info(" ====================cleanup ====================")
    114.         self.cleanup()
    115.     def cleanup(self):
    116.         if hasattr(self.args, "using_mlops") and self.args.using_mlops:
    117.             # mlops_metrics = MLOpsMetrics()
    118.             # mlops_metrics.set_sys_reporting_status(False)
    119.             pass
    120.         self.finish()
    121.     def send_model_to_sever(self,receive_id,weights,local_sample_num):
    122.         tick=time.time()
    123.         if hasattr(self.args, "using_mlops") and self.args.using_mlops:
    124.             self.mlops_event.log_event_started(
    125.                 "comm_c2s", event_value=str(self.round_idx)
    126.             )
    127.         message = Message(
    128.             MyMessage.MSG_TYPE_C2S_SEND_MODEL_TO_SERVER,
    129.             self.client_real_id,
    130.             receive_id,
    131.         )
    132.         message.add_params(MyMessage.MSG_ARG_KEY_MODEL_PARAMS, weights)
    133.         message.add_params(MyMessage.MSG_ARG_KEY_NUM_SAMPLES, local_sample_num)
    134.         self.send_message(message)
    135.         MLOpsProfilerEvent.log_to_wandb(
    136.             {"Communication/Send_Total": time.time() - tick}
    137.         )
    138.         # Report client model to MLOps
    139.         if hasattr(self.args, "using_mlops") and self.args.using_mlops:
    140.             model_url = message.get(MyMessage.MSG_ARG_KEY_MODEL_PARAMS_URL)
    141.             model_info = {
    142.                 "run_id": self.args.run_id,
    143.                 "edge_id": self.client_real_id,
    144.                 "round_idx": self.round_idx + 1,
    145.                 "client_model_s3_address": model_url,
    146.             }
    147.             self.mlops_metrics.report_client_model_info(model_info)
    148.  
    149.     #
    150.     
    151.     def send_client_status(self, receive_id, status="ONLINE"):
    152.         logging.info("send_client_status")
    153.         message = Message(
    154.             MyMessage.MSG_TYPE_C2S_CLIENT_STATUS, self.client_real_id, receive_id
    155.         )
    156.         sys_name = platform.system()
    157.         if sys_name == "Darwin":
    158.             sys_name = "Mac"
    159.         # Debug for simulation mobile system
    160.         # sys_name = MyMessage.MSG_CLIENT_OS_ANDROID
    161.  
    162.         message.add_params(MyMessage.MSG_ARG_KEY_CLIENT_STATUS, status)
    163.         message.add_params(MyMessage.MSG_ARG_KEY_CLIENT_OS, sys_name)
    164.         self.send_message(message)
    165.     def report_training_status(self, status):
    166.         if hasattr(self.args, "using_mlops") and self.args.using_mlops:
    167.             self.mlops_metrics.set_messenger(self.com_manager_status, self.args)
    168.             self.mlops_metrics.report_client_training_status(
    169.                 self.client_real_id, status
    170.             )
    171.  
    172.     def sync_process_group(
    173.             self, round_idx, model_params=None, client_index=None, src=0
    174.     ):
    175.         logging.info("sending round number to pg")
    176.         round_number = [round_idx, model_params, client_index]
    177.         dist.broadcast_object_list(
    178.             round_number,
    179.             src=src,
    180.             group=self.trainer_dist_adapter.process_group_manager.get_process_group(),
    181.         )
    182.         logging.info("round number %d broadcast to process group" % round_number[0])
    183.  
    184.     def __train(self):
    185.         logging.info("#######training########### round_id = %d" % self.round_idx)
    186.         if hasattr(self.args, "using_mlops") and self.args.using_mlops:
    187.             self.mlops_event.log_event_started("train", event_value=str(self.round_idx))
    188.  
    189.         weights, local_sample_num = self.trainer_dist_adapter.train(self.round_idx)
    190.  
    191.         if hasattr(self.args, "using_mlops") and self.args.using_mlops:
    192.             self.mlops_event.log_event_ended("train", event_value=str(self.round_idx))
    193.  
    194.         # the current model is still DDP-wrapped under cross-silo-hi setting
    195.         if self.args.scenario == FEDML_CROSS_SILO_SCENARIO_HIERARCHICAL:
    196.             weights = convert_model_params_from_ddp(weights)
    197.  
    198.         self.send_model_to_server(0, weights, local_sample_num)
    199.  
    200.     def run(self):
    201.         super().run()

    client_launch.py

    这个感觉基本不用动吧

    1. import os
    2. import subprocess
    3. import torch
    4. from fedml.arguments import load_arguments
    5. from fedml.constants import (
    6.     FEDML_CROSS_SILO_SCENARIO_HIERARCHICAL,
    7.     FEDML_TRAINING_PLATFORM_CROSS_SILO,
    8.     FEDML_CROSS_SILO_SCENARIO_HORIZONTAL,
    9. )
    10. from fedml.device import get_device_type
    11.  
    12. class CrossSiloLauncher:
    13.     def launch_dist_trainer(torch_client_filename,inputs):
    14.         args=load_arguments(FEDML_TRAINING_PLATFORM_CROSS_SILO)
    15.         if args.scenario == FEDML_CROSS_SILO_SCENARIO_HIERARCHICAL:
    16.             CrossSiloLauncher._run_cross_silo_hierarchical(
    17.                 args, torch_client_filename, inputs
    18.             )
    19.         elif args.scenarios == FEDML_CROSS_SILO_SCENARIO_HORIZONTAL:
    20.             CrossSiloLauncher._run_cross_silo_horizontal(
    21.                 args, torch_client_filename, inputs
    22.             )
    23.         else:
    24.             raise Exception(
    25.                 "we do not support {}, check whether this is typo in args.scenario".format(
    26.                     args.scenario
    27.                 )
    28.             )
    29.     def _run_cross_silo_horizontal(args, torch_client_filename, inputs):
    30.         python_path = subprocess.run(
    31.             ["which", "python"], capture_output=True, text=True
    32.         ).stdout.strip()
    33.         process_arguments = [python_path, torch_client_filename] + inputs
    34.         subprocess.run(process_arguments)
    35.     def _run_cross_silo_hierarchical(args, torch_client_filename, inputs):
    36.         def get_torchrun_arguments(node_rank):
    37.             torchrun_path = subprocess.run(
    38.                 ["which", "torchrun"], capture_output=True, text=True
    39.             ).stdout.strip()
    40.  
    41.             return [
    42.                 torchrun_path,
    43.                 f"--nnodes={args.n_node_in_silo}",
    44.                 f"--nproc_per_node={args.n_proc_per_node}",
    45.                 # "--rdzv_backend=c10d",
    46.                 f"--rdzv_endpoint={args.master_address}:{args.launcher_rdzv_port}",
    47.                 f"--node_rank={node_rank}",
    48.                 "--rdzv_id=hi_fl",
    49.                 torch_client_filename,
    50.             ] + inputs
    51.  
    52.         network_interface = (
    53.             None if not hasattr(args, "network_interface") else args.network_interface
    54.         )
    55.         print(
    56.             f"Using network interface {network_interface} for process group and TRPC communication"
    57.         )
    58.         env_variables = {
    59.             "OMP_NUM_THREADS": "4",
    60.         }
    61.         if network_interface:
    62.             env_variables = {
    63.                 **env_variables,
    64.                 "NCCL_SOCKET_IFNAME": network_interface,
    65.                 "GLOO_SOCKET_IFNAME": network_interface,
    66.             }
    67.  
    68.         if args.n_node_in_silo == 1:
    69.             args.node_rank = 0
    70.             args.manual_launch = True
    71.             if not (hasattr(args, "n_proc_per_node") and args.n_proc_per_node):
    72.                 print("Number of processes per node not specified.")
    73.                 device_type = get_device_type(args)
    74.                 if torch.cuda.is_available() and device_type == "gpu":
    75.                     gpu_count = torch.cuda.device_count()
    76.                     print(f"Using number of GPUs ({gpu_count}) as number of processeses.")
    77.                     args.n_proc_per_node = gpu_count
    78.                 else: 
    79.                     print(f"Using number 1 as number of processeses.")
    80.                     args.n_proc_per_node = 1
    81.  
    82.         if hasattr(args, "manual_launch") and args.manual_launch:
    83.             print(f"Manual Client Launcher")
    84.             node_rank = args.node_rank
    85.             torchrun_cmd_arguments = get_torchrun_arguments(node_rank)
    86.             process_args = torchrun_cmd_arguments
    87.             print(f"Launching node {node_rank} of silo {args.rank}")
    88.             subprocess.run(process_args, env=dict(os.environ, **env_variables))
    89.  
    90.         else:
    91.             print(f"Automatic Client Launcher")
    92.  
    93.             which_pdsh = subprocess.run(
    94.                 ["which", "pdsh"], capture_output=True, text=True
    95.             ).stdout.strip()
    96.  
    97.             if not which_pdsh:
    98.                 raise Exception(
    99.                     f"Silo {args.rank} has {args.n_node_in_silo} nodes. Automatic Client Launcher for more than 1 nodes requires PSDH."
    100.                 )
    101.  
    102.             print(f"Launching nodes using pdsh")
    103.  
    104.             os.environ["PDSH_RCMD_TYPE"] = "ssh"
    105.             node_addresses = ",".join(args.node_addresses)
    106.             pdsh_cmd_aruments = ["pdsh", "-w", node_addresses]
    107.  
    108.             exports = ""
    109.             for key, val in env_variables.items():
    110.                 exports += "export {}={}; ".format(key, val)
    111.             prerun_args = [
    112.                 exports,
    113.                 f"cd {os.path.abspath('.')};",
    114.             ]
    115.  
    116.             node_rank = "%n"
    117.             torchrun_cmd_arguments = get_torchrun_arguments(node_rank)
    118.             process_args = pdsh_cmd_aruments + prerun_args + torchrun_cmd_arguments
    119.             subprocess.run(process_args)

    fedml_trainer.py

    这里有一个train函数可能需要改

    1. import time
    2.  
    3. from ...constants import FEDML_CROSS_SILO_SCENARIO_HIERARCHICAL
    4. from ...core.mlops.mlops_profiler_event import MLOpsProfilerEvent
    5. from fedml.data import split_data_for_dist_trainers
    6.  
    7. class FedMLTrainer(object):
    8.     def __init__(
    9.             self,
    10.             client_index,
    11.             train_data_local_dict,
    12.             train_data_local_num_dict,
    13.             test_data_local_dict,
    14.             train_data_num,
    15.             device,
    16.             args,
    17.             model_trainer,
    18.     ):
    19.         self.trainer=model_trainer
    20.         self.client_index=client_index
    21.         if args.scenario == FEDML_CROSS_SILO_SCENARIO_HIERARCHICAL:
    22.             self.train_data_local_dict = split_data_for_dist_trainers(
    23.                 train_data_local_dict, args.n_proc_in_silo
    24.             )
    25.         else:
    26.             self.train_data_local_dict = train_data_local_dict
    27.         self.train_data_local_num_dict=train_data_local_num_dict
    28.         self.test_data_local_dict=test_data_local_dict
    29.         self.all_train_data_num=train_data_num
    30.         self.train_local=None
    31.         self.local_sample_number=None
    32.         self.test_local=None
    33.         self.device=device
    34.         self.args=args
    35.  
    36.     def update_model(self,weights):
    37.         self.trainer.set_model_params(weights)
    38.  
    39.     def update_dataset(self,client_index):
    40.         self.client_index=client_index
    41.         if self.args.scenario == FEDML_CROSS_SILO_SCENARIO_HIERARCHICAL:
    42.             self.train_local = self.train_data_local_dict[client_index][
    43.                 self.args.proc_rank_in_silo
    44.             ]
    45.         else:
    46.             self.train_local = self.train_data_local_dict[client_index]
    47.         self.local_sample_number = self.train_data_local_num_dict[client_index]
    48.         self.test_local = self.test_data_local_dict[client_index]
    49.  
    50.  
    51.     def train(self,round_idx=None):
    52.         self.args.round_idx=round_idx
    53.         tick=time.time()
    54.         self.trainer.train(self.train_local, self.device, self.args)
    55.         MLOpsProfilerEvent.log_to_wandb(
    56.             {"Train/Time": time.time() - tick, "round": round_idx}
    57.         )
    58.         weights=self.trainer.get_model_params()
    59.         return weights,self.local_sample_number
    60.  
    61.  
    62.     def test(self):
    63.         train_metrics=self.train.test(self.train_local, self.device, self.args)
    64.         train_tot_correct,train_num_sample,train_loss=(
    65.             train_metrics["test_correct"],
    66.             train_metrics["test_total"],
    67.             train_metrics["test_loss"],
    68.         )
    69.  
    70.         test_metrics = self.trainer.test(self.test_local, self.device, self.args)
    71.         test_tot_correct, test_num_sample, test_loss = (
    72.             test_metrics["test_correct"],
    73.             test_metrics["test_total"],
    74.             test_metrics["test_loss"],
    75.         )
    76.  
    77.         return(
    78.             train_tot_correct,
    79.             train_loss,
    80.             train_num_sample,
    81.             test_tot_correct,
    82.             test_loss,
    83.             test_num_sample,
    84.         )

    fedml_trainer_dist_adapter.py

    1. import logging
    2.  
    3. from fedml.constants import FEDML_CROSS_SILO_SCENARIO_HIERARCHICAL
    4. from .fedml_trainer import FedMLTrainer
    5. from .trainer.trainer_creator import create_model_trainer
    6.  
    7. class TrainaDistAdapter:
    8.     def __init__(
    9.             self,
    10.             args,
    11.             device,
    12.             client_rank,
    13.             model,
    14.             train_data_num,
    15.             train_data_local_num_dict,
    16.             train_data_local_dict,
    17.             test_data_local_dict,
    18.             model_trainer,
    19.     ):
    20.         model.to(device)
    21.         if args.scenario == FEDML_CROSS_SILO_SCENARIO_HIERARCHICAL:
    22.             from torch.nn.parallel import DistributedDataParallel as DDP
    23.             from .process_group_manager import ProcessGroupManager
    24.  
    25.             only_gpu = args.using_gpu
    26.             self.process_group_manager = ProcessGroupManager(
    27.                 args.proc_rank_in_silo,
    28.                 args.n_proc_in_silo,
    29.                 args.pg_master_address,
    30.                 args.pg_master_port,
    31.                 only_gpu,
    32.             )
    33.             model = DDP(model, device_ids=[device] if only_gpu else None)
    34.  
    35.         if model_trainer is None:
    36.             model_trainer = create_model_trainer(args, model)
    37.         else:
    38.             model_trainer.model = model
    39.  
    40.         client_index = client_rank - 1
    41.  
    42.         model_trainer.set_id(client_index)
    43.  
    44.         logging.info("Initiating Trainer")
    45.         trainer = self.get_trainer(
    46.             client_index,
    47.             train_data_local_dict,
    48.             train_data_local_num_dict,
    49.             test_data_local_dict,
    50.             train_data_num,
    51.             device,
    52.             args,
    53.             model_trainer,
    54.         )
    55.         self.client_index=client_index
    56.         self.client_rank=client_rank
    57.         self.device=device
    58.         self.trainer=trainer
    59.         self.args=args
    60.  
    61.  
    62.     def get_trainer(
    63.             self,
    64.             client_index,
    65.             train_data_local_dict,
    66.             train_data_local_num_dict,
    67.             test_data_local_dict,
    68.             train_data_num,
    69.             device,
    70.             args,
    71.             model_trainer,
    72.     ):
    73.         return FedMLTrainer(
    74.             client_index,
    75.             train_data_local_dict,
    76.             train_data_local_num_dict,
    77.             test_data_local_dict,
    78.             train_data_num,
    79.             device,
    80.             args,
    81.             model_trainer,
    82.         )
    83.  
    84.  
    85.     def train(self,round_idx):
    86.         weights,local_sample_num = self.trainer.train(round_idx)
    87.         return weights,local_sample_num
    88.     def update_model(self,model_params):
    89.         self.trainer.update_model(model_params)
    90.     def update_dataset(self,client_index=None):
    91.         _client_index=client_index or self.client_index
    92.         self.trainer.update_model(int(_client_index))
    93.  
    94.     def cleanup_pg(self):
    95.         if self.args.scenario == FEDML_CROSS_SILO_SCENARIO_HIERARCHICAL:
    96.             logging.info(
    97.                 "Cleaningup process group for client %s in silo %s"
    98.                 % (self.args.proc_rank_in_silo, self.args.rank_in_node)
    99.             )
    100.             self.process_group_manager.cleanup()

    其他的文件我觉得是不需要动的

  • 相关阅读:
    Redis 缓存预热、预热数据选取策略、缓存保温、性能边界
    19.前端笔记-CSS-显示隐藏元素
    RK3568 安卓12 EC20模块NOCONN没有ip的问题（已解决）
    Redis7.2.3集群安装，新增节点，删除节点，分配哈希槽，常见问题
    项目问题-常见错误
    从文件里一次读取一行
    vscode git 拉取报错 在签出前，请清理存储库工作树
    基于Java毕业设计中小企业人力资源管理系统源码+系统+mysql+lw文档+部署软件
    【CNN-SVM回归预测】基于CNN-SVM实现数据回归预测附matlab代码
    香山杯-2023-Crypto
  • 原文地址：https://blog.csdn.net/kling_bling/article/details/126729854