basic_server#

Module Contents#

SyncServerHandler

Synchronous Parameter Server Handler.

AsyncServerHandler

Asynchronous Parameter Server Handler
class SyncServerHandler(model: torch.nn.Module, global_round: int, num_clients: int = 0, sample_ratio: float = 1, cuda: bool = False, device: str = None, sampler: fedlab.contrib.client_sampler.base_sampler.FedSampler = None, logger: fedlab.utils.Logger = None)#

Bases: fedlab.core.server.handler.ServerHandler

Synchronous Parameter Server Handler.

Backend of synchronous parameter server: this class is responsible for backend computing in synchronous server.

Synchronous parameter server will wait for every client to finish local training process before the next FL round.

Details in paper: http://proceedings.mlr.press/v54/mcmahan17a.html

Parameters:

  • model (torch.nn.Module) – model trained by federated learning.

  • global_round (int) – stop condition. Shut down FL system when global round is reached.

  • num_clients (int) – number of clients in FL. Default: 0 (initialized external).

  • sample_ratio (float) – the result of sample_ratio * num_clients is the number of clients for every FL round.

  • cuda (bool) – use GPUs or not. Default: False.

  • device (str, optional) – assign model/data to the given GPUs. E.g., ‘device:0’ or ‘device:0,1’. Defaults to None. If device is None and cuda is True, FedLab will set the gpu with the largest memory as default.

  • sampler (FedSampler, optional) – assign a sampler to define the client sampling strategy. Default: random sampling with FedSampler.

  • logger (Logger, optional) – object of Logger.

Property for manager layer. Server manager will call this property when activates clients.

property num_clients_per_round#
property if_stop#

NetworkManager keeps monitoring this attribute, and it will stop all related processes and threads when True returned.

sample_clients(num_to_sample=None)#

Return a list of client rank indices selected randomly. The client ID is from 0 to self.num_clients -1.

global_update(buffer)#
load(payload: List[torch.Tensor]) → bool#

Update global model with collected parameters from clients.

Note

Server handler will call this method when its client_buffer_cache is full. User can overwrite the strategy of aggregation to apply on model_parameters_list, and use SerializationTool.deserialize_model() to load serialized parameters after aggregation into self._model.

Parameters:

payload (list[torch.Tensor]) – A list of tensors passed by manager layer.

class AsyncServerHandler(model: torch.nn.Module, global_round: int, num_clients: int, cuda: bool = False, device: str = None, logger: fedlab.utils.Logger = None)#

Bases: fedlab.core.server.handler.ServerHandler

Asynchronous Parameter Server Handler

Update global model immediately after receiving a ParameterUpdate message Paper: https://arxiv.org/abs/1903.03934

Parameters:

  • model (torch.nn.Module) – Global model in server

  • global_round (int) – stop condition. Shut down FL system when global round is reached.

  • num_clients (int) – number of clients in FL.

  • cuda (bool) – Use GPUs or not.

  • device (str, optional) – Assign model/data to the given GPUs. E.g., ‘device:0’ or ‘device:0,1’. Defaults to None. If device is None and cuda is True, FedLab will set the gpu with the largest memory as default.

  • logger (Logger, optional) – Object of Logger.

property if_stop#

NetworkManager keeps monitoring this attribute, and it will stop all related processes and threads when True returned.

Property for manager layer. Server manager will call this property when activates clients.

setup_optim(alpha, strategy='constant', a=10, b=4)#

Setup optimization configuration.

Parameters:

  • alpha (float) – Weight used in async aggregation.

  • strategy (str, optional) – Adaptive strategy. constant, hinge and polynomial is optional. Default: constant.. Defaults to ‘constant’.

  • a (int, optional) – Parameter used in async aggregation.. Defaults to 10.

  • b (int, optional) – Parameter used in async aggregation.. Defaults to 4.

global_update(buffer)#
load(payload: List[torch.Tensor]) → bool#

Override this function to define how to update global model (aggregation or optimization).

adapt_alpha(receive_model_time)#

update the alpha according to staleness