server#

hierarchical
- connector
- scheduler

Package Contents#

`SyncParameterServerHandler`	Synchronous Parameter Server Handler.
`AsyncParameterServerHandler`	Asynchronous Parameter Server Handler
`SynchronousServerManager`	Synchronous communication
`AsynchronousServerManager`	Asynchronous communication network manager for server

class SyncParameterServerHandler(model, global_round, sample_ratio, cuda=False, logger=None)#

Bases: ParameterServerBackendHandler

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 used in this federation.
global_round (int) – stop condition. Shut down FL system when global round is reached.
sample_ratio (float) – The result of sample_ratio * client_num is the number of clients for every FL round.
cuda (bool) – Use GPUs or not. Default: False.
logger (Logger, optional) – object of Logger.

property downlink_package(self)#: Property for manager layer. Server manager will call this property when activates clients.

property if_stop(self)#: NetworkManager keeps monitoring this attribute, and it will stop all related processes and threads when True returned.

property client_num_per_round(self)#

sample_clients(self)#: Return a list of client rank indices selected randomly. The client ID is from 1 to self.client_num_in_total + 1.

_update_global_model(self, payload)#

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 AsyncParameterServerHandler(model, alpha, total_time, strategy='constant', cuda=False, logger=None)#

Bases: ParameterServerBackendHandler

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
alpha (float) – Weight used in async aggregation.
total_time (int) – Stop condition. Shut down FL system when total_time is reached.
strategy (str) – Adaptive strategy. constant, hinge and polynomial is optional. Default: constant.
cuda (bool) – Use GPUs or not.
logger (Logger, optional) – Object of Logger.

property if_stop(self)#: NetworkManager keeps monitoring this attribute, and it will stop all related processes and threads when True returned.

property downlink_package(self)#: Property for manager layer. Server manager will call this property when activates clients.

_update_global_model(self, payload)#: Override this function to define how to update global model (aggregation or optimization).

_adapt_alpha(self, receive_model_time)#: update the alpha according to staleness

class SynchronousServerManager(network, handler, logger=None)#

Bases: ServerManager

Synchronous communication

This is the top class in our framework which is mainly responsible for network communication of SERVER!. Synchronously communicate with clients following agreements defined in main_loop().

Parameters

network (DistNetwork) – Network configuration and interfaces.
handler (ParameterServerBackendHandler) – Backend calculation handler for parameter server.
logger (Logger, optional) – Object of Logger.

setup(self)#

Initialization Stage.

Server accept local client num report from client manager.
Init a coordinator for client_id -> rank mapping.

main_loop(self)#

Actions to perform in server when receiving a package from one client.

Server transmits received package to backend computation handler for aggregation or others manipulations.

Loop:

activate clients for current training round.
listen for message from clients -> transmit received parameters to server backend.

Note

Communication agreements related: user can overwrite this function to customize communication agreements. This method is key component connecting behaviors of ParameterServerBackendHandler and NetworkManager.

Raises: Exception – Unexpected MessageCode.

shutdown(self)#: Shutdown stage.

activate_clients(self)#

Activate subset of clients to join in one FL round

Manager will start a new thread to send activation package to chosen clients’ process rank. The id of clients are obtained from handler.sample_clients(). And their communication ranks are are obtained via coordinator.

shutdown_clients(self)#

Shutdown all clients.

Send package to each client with MessageCode.Exit.

Note

Communication agreements related: User can overwrite this function to define package for exiting information.

class AsynchronousServerManager(network, handler, logger=None)#

Bases: ServerManager

Asynchronous communication network manager for server

This is the top class in our framework which is mainly responsible for network communication of SERVER!. Asynchronously communicate with clients following agreements defined in mail_loop().

Parameters

network (DistNetwork) – Network configuration and interfaces.
handler (ParameterServerBackendHandler) – Backend computation handler for parameter server.
logger (Logger, optional) – Object of Logger.

setup(self)#

Initialization Stage.

Server accept local client num report from client manager.
Init a coordinator for client_id -> rank mapping.

main_loop(self)#

Communication agreements of asynchronous FL.

Server receive ParameterRequest from client. Send model parameter to client.
Server receive ParameterUpdate from client. Transmit parameters to queue waiting for aggregation.

Raises: ValueError – invalid message code.

shutdown(self)#

Shutdown stage.

Close the network connection in the end.

updater_thread(self)#: Asynchronous communication maintain a message queue. A new thread will be started to keep monitoring message queue.

shutdown_clients(self)#

Shutdown all clients.

Send package to clients with MessageCode.Exit.