fedn.network.combiner package

The FEDn Combiner package is responsible for combining models from multiple clients. It’s the acting gRPC server for the federated network.

Subpackages

• fedn.network.combiner.aggregators package
  • Submodules
  • fedn.network.combiner.aggregators.aggregatorbase module
    • AggregatorBase
      • AggregatorBase.combine_models()
      • AggregatorBase.get_state()
      • AggregatorBase.next_model_update()
      • AggregatorBase.on_model_update()
    • get_aggregator()
  • fedn.network.combiner.aggregators.fedavg module
    • Aggregator
      • Aggregator.combine_models()

Submodules

fedn.network.combiner.connect module

class fedn.network.combiner.connect.ConnectorCombiner(host, port, myhost, fqdn, myport, token, name, secure=False, verify=False)

Bases: object

Connector for annnouncing combiner to the FEDn network.

Parameters:

• host (str) – host of discovery service

• port (int) – port of discovery service

• myhost (str) – host of combiner

• fqdn (str) – fully qualified domain name of combiner

• myport (int) – port of combiner

• token (str) – token for authentication

• name (str) – name of combiner

• secure (bool) – True if https is used, False if http

• verify (bool) – True if certificate is verified, False if not

announce()

Announce combiner to FEDn network via discovery service (REST-API).

Returns:

Tuple with announcement Status, FEDn network configuration if sucessful, else None.

Return type:

fedn.network.combiner.connect.Status, str

class fedn.network.combiner.connect.Status(value)

Bases: Enum

Enum for representing the status of a combiner announcement.

Assigned = 1

TryAgain = 2

UnAuthorized = 3

UnMatchedConfig = 4

Unassigned = 0

fedn.network.combiner.interfaces module

class fedn.network.combiner.interfaces.Channel(address, port, certificate=None)

Bases: object

Wrapper for a gRPC channel.

Parameters:

• address (str) – The address for the gRPC server.

• port (int) – The port for connecting to the gRPC server.

• certificate (str) – The certificate for connecting to the gRPC server (optional)

get_channel()

Get a channel.

Returns:

An instance of a gRPC channel

Return type:

grpc.Channel

class fedn.network.combiner.interfaces.CombinerInterface(parent, name, address, fqdn, port, certificate=None, key=None, ip=None, config=None)

Bases: object

Interface for the Combiner (aggregation server).

Abstraction on top of the gRPC server servicer.

Parameters:

• parent (fedn.network.api.interfaces.API) – The parent combiner (controller)

• name (str) – The name of the combiner.

• address (str) – The address of the combiner.

• fqdn (str) – The fully qualified domain name of the combiner.

• port (int) – The port of the combiner.

• certificate (str) – The certificate of the combiner (optional).

• key (str) – The key of the combiner (optional).

• ip (str) – The ip of the combiner (optional).

• config (dict) – The configuration of the combiner (optional).

allowing_clients()

Check if the combiner is allowing additional client connections.

Returns:

True if accepting, else False.

Return type:

bool

configure(config=None)

Configure the combiner. Set the parameters in config at the server.

Parameters:

config (dict) – A dictionary containing parameters.

flush_model_update_queue()

Reset the model update queue on the combiner.

classmethod from_json()

Initialize the combiner config from a json document.

Parameters:

combiner_config (dict) – The combiner configuration.

Returns:

An instance of the combiner interface.

Return type:

fedn.network.combiner.interfaces.CombinerInterface

get_certificate()

Get combiner certificate.

Returns:

The combiner certificate.

Return type:

str, None if no certificate is set.

get_key()

Get combiner key.

Returns:

The combiner key.

Return type:

str, None if no key is set.

get_model(id)

Download a model from the combiner server.

Parameters:

id (str) – The model id.

Returns:

A file-like object containing the model.

Return type:

io.BytesIO, None if the model is not available.

list_active_clients()

List active clients.

Returns:

A list of active clients.

Return type:

json

report()

Recieve a status report from the combiner.

Returns:

A dictionary describing the combiner state.

Return type:

dict

Raises:

CombinerUnavailableError – If the combiner is unavailable.

submit(config)

Submit a compute plan to the combiner.

Parameters:

config (dict) – The job configuration.

Returns:

Server ControlResponse object.

Return type:

fedn.common.net.grpc.fedn_pb2.ControlResponse

to_dict()

Export combiner configuration to a dictionary.

Returns:

A dictionary with the combiner configuration.

Return type:

dict

to_json()

Export combiner configuration to json.

Returns:

A json document with the combiner configuration.

Return type:

str

exception fedn.network.combiner.interfaces.CombinerUnavailableError

Bases: Exception

fedn.network.combiner.modelservice module

class fedn.network.combiner.modelservice.ModelService

Bases: ModelServiceServicer

Service for handling download and upload of models to the server.

Download(request, context)

RPC endpoints for downloading a model.

Parameters:

• request (fedn.common.net.grpc.fedn_pb2.ModelRequest) – The model request object.

• context (grpc._server._Context) – The context object (unused)

Returns:

A model response iterator.

Return type:

fedn.common.net.grpc.fedn_pb2.ModelResponse

Upload(request_iterator, context)

RPC endpoints for uploading a model.

Parameters:

• request_iterator (fedn.common.net.grpc.fedn_pb2.ModelRequest) – The model request iterator.

• context (grpc._server._Context) – The context object (unused)

Returns:

A model response object.

Return type:

fedn.common.net.grpc.fedn_pb2.ModelResponse

exist(model_id)

Check if a model exists on the server.

Parameters:

model_id – The model id.

Returns:

True if the model exists, else False.

get_model(id)

Download model with id ‘id’ from server.

Parameters:

id (str) – The model id.

Returns:

A BytesIO object containing the model.

Return type:

io.BytesIO, None if model does not exist.

get_tmp_path()

Return a temporary output path compatible with save_model, load_model.

load_model_from_BytesIO(model_bytesio, helper)

Load a model from a BytesIO object.

Parameters:

• model_bytesio (io.BytesIO) – A BytesIO object containing the model.

• helper (fedn.utils.helperbase.HelperBase) – The helper object for the model.

Returns:

The model object.

Return type:

return type of helper.load

serialize_model_to_BytesIO(model, helper)

Serialize a model to a BytesIO object.

Parameters:

• model (return type of helper.load) – The model object.

• helper (fedn.utils.helperbase.HelperBase) – The helper object for the model.

Returns:

A BytesIO object containing the model.

Return type:

io.BytesIO

set_model(model, id)

Upload model to server.

Parameters:

• model (io.BytesIO) – A model object (BytesIO)

• id (str) – The model id.

fedn.network.combiner.round module

exception fedn.network.combiner.round.ModelUpdateError

Bases: Exception

class fedn.network.combiner.round.RoundController(aggregator_name, storage, server, modelservice)

Bases: object

Round controller.

The round controller recieves round configurations from the global controller and coordinates model updates and aggregation, and model validations.

Parameters:

• aggregator_name (str) – The name of the aggregator plugin module.

• storage (class: fedn.common.storage.s3.s3repo.S3ModelRepository) – Model repository for :class: fedn.network.combiner.Combiner

• server (class: fedn.network.combiner.Combiner) – A handle to the Combiner class :class: fedn.network.combiner.Combiner

• modelservice (class: fedn.network.combiner.modelservice.ModelService) – A handle to the model service :class: fedn.network.combiner.modelservice.ModelService

execute_training_round(config)

Coordinates clients to execute training tasks.

Parameters:

config (dict) – The round config object.

Returns:

metadata about the training round.

Return type:

dict

execute_validation_round(round_config)

Coordinate validation rounds as specified in config.

Parameters:

round_config (dict) – The round config object.

load_model_update(helper, model_id)

Load model update in its native format.

Parameters:

• helper (class: fedn.utils.helpers.HelperBase) – An instance of :class: fedn.utils.helpers.HelperBase, ML framework specific helper, defaults to None

• model_id (str) – The ID of the model update, UUID in str format

load_model_update_str(model_id, retry=3)

Load model update object and return it as BytesIO.

Parameters:

• model_id (str) – The ID of the model

• retry (int, optional) – number of times retrying load model update, defaults to 3

Returns:

Updated model

Return type:

class: io.BytesIO

push_round_config(round_config)

Add a round_config (job description) to the inbox.

Parameters:

round_config (dict) – A dict containing the round configuration (from global controller).

Returns:

A job id (universally unique identifier) for the round.

Return type:

str

run(polling_interval=1.0)

Main control loop. Execute rounds based on round config on the queue.

Parameters:

polling_interval (float) – The polling interval in seconds for checking if a new job/config is available.

stage_model(model_id, timeout_retry=3, retry=2)

Download a model from persistent storage and set in modelservice.

Parameters:

• model_id (str) – ID of the model update object to stage.

• timeout_retry (int, optional) – Sleep before retrying download again(sec), defaults to 3

• retry (int, optional) – Number of retries, defaults to 2

waitforit(config, buffer_size=100, polling_interval=0.1)

Defines the policy for how long the server should wait before starting to aggregate models.

The policy is as follows:

1. Wait a maximum of time_window time until the round times out.

2. Terminate if a preset number of model updates (buffer_size) are in the queue.

Parameters:

• config (dict) – The round config object

• buffer_size (int, optional) – The number of model updates to wait for before starting aggregation, defaults to 100

• polling_interval (float, optional) – The polling interval, defaults to 0.1

fedn.network.combiner.server module

class fedn.network.combiner.server.Combiner(config)

Bases: CombinerServicer, ReducerServicer, ConnectorServicer, ControlServicer

Combiner gRPC server.

Parameters:

config (dict) – configuration for the combiner

AcceptingClients(request: ConnectionRequest, context)

RPC endpoint that returns a ConnectionResponse indicating whether the server is accepting clients or not.

Parameters:

• request (fedn.common.net.grpc.fedn_pb2.ConnectionRequest) – the request (unused)

• context (grpc._server._Context) – the context (unused)

Returns:

the response

Return type:

fedn.common.net.grpc.fedn_pb2.ConnectionResponse

Configure(control: ControlRequest, context)

Configure the Combiner.

Parameters:

• control (fedn.common.net.grpc.fedn_pb2.ControlRequest) – the control request

• context (grpc._server._Context) – the context (unused)

Returns:

the control response

Return type:

fedn.common.net.grpc.fedn_pb2.ControlResponse

FlushAggregationQueue(control: ControlRequest, context)

Flush the queue.

Parameters:

• control (fedn.common.net.grpc.fedn_pb2.ControlRequest) – the control request

• context (grpc._server._Context) – the context (unused)

Returns:

the control response

Return type:

fedn.common.net.grpc.fedn_pb2.ControlResponse

ListActiveClients(request: ListClientsRequest, context)

RPC endpoint that returns a ClientList containing the names of all active clients.

An active client has sent a status message / responded to a heartbeat request in the last 10 seconds.

Parameters:

• request (fedn.common.net.grpc.fedn_pb2.ListClientsRequest) – the request

• context (grpc._server._Context) – the context (unused)

Returns:

the client list

Return type:

fedn.common.net.grpc.fedn_pb2.ClientList

ModelUpdateRequestStream(response, context)

A server stream RPC endpoint (Update model). Messages from client stream.

Parameters:

• response (fedn.common.net.grpc.fedn_pb2.ModelUpdateRequest) – the response

• context (grpc._server._Context) – the context

ModelUpdateStream(update, context)

Model update stream RPC endpoint. Update status for client is connecting to stream.

Parameters:

• update (fedn.common.net.grpc.fedn_pb2.ModelUpdate) – the update message

• context (grpc._server._Context) – the context

ModelValidationRequestStream(response, context)

A server stream RPC endpoint (Validation). Messages from client stream.

Parameters:

• response (fedn.common.net.grpc.fedn_pb2.ModelValidationRequest) – the response

• context (grpc._server._Context) – the context

ModelValidationStream(update, context)

Model validation stream RPC endpoint. Update status for client is connecting to stream.

Parameters:

• update (fedn.common.net.grpc.fedn_pb2.ModelValidation) – the update message

• context (grpc._server._Context) – the context

Report(control: ControlRequest, context)

Describe current state of the Combiner.

Parameters:

• control (fedn.common.net.grpc.fedn_pb2.ControlRequest) – the control request

• context (grpc._server._Context) – the context (unused)

Returns:

the control response

Return type:

fedn.common.net.grpc.fedn_pb2.ControlResponse

SendHeartbeat(heartbeat: Heartbeat, context)

RPC that lets clients send a hearbeat, notifying the server that

the client is available.

Parameters:

• heartbeat (fedn.common.net.grpc.fedn_pb2.Heartbeat) – the heartbeat

• context (grpc._server._Context) – the context (unused)

Returns:

the response

Return type:

fedn.common.net.grpc.fedn_pb2.Response

SendModelUpdate(request, context)

Send a model update response.

Parameters:

• request (fedn.common.net.grpc.fedn_pb2.ModelUpdate) – the request

• context (grpc._server._Context) – the context

Returns:

the response

Return type:

fedn.common.net.grpc.fedn_pb2.Response

SendModelUpdateRequest(request, context)

Send a model update request.

Parameters:

• request (fedn.common.net.grpc.fedn_pb2.ModelUpdateRequest) – the request

• context (grpc._server._Context) – the context

Returns:

the response

Return type:

fedn.common.net.grpc.fedn_pb2.Response

SendModelValidation(request, context)

Send a model validation response.

Parameters:

• request (fedn.common.net.grpc.fedn_pb2.ModelValidation) – the request

• context (grpc._server._Context) – the context

Returns:

the response

Return type:

fedn.common.net.grpc.fedn_pb2.Response

SendModelValidationRequest(request, context)

Send a model validation request.

Parameters:

• request (fedn.common.net.grpc.fedn_pb2.ModelValidationRequest) – the request

• context (grpc._server._Context) – the context

Returns:

the response

Return type:

fedn.common.net.grpc.fedn_pb2.Response

SendStatus(status: Status, context)

A client stream RPC endpoint that accepts status messages.

Parameters:

• status (fedn.common.net.grpc.fedn_pb2.Status) – the status message

• context (grpc._server._Context) – the context (unused)

Returns:

the response

Return type:

fedn.common.net.grpc.fedn_pb2.Response

Start(control: ControlRequest, context)

Start a round of federated learning”

Parameters:

• control (fedn.common.net.grpc.fedn_pb2.ControlRequest) – the control request

• context (grpc._server._Context) – the context (unused)

Returns:

the control response

Return type:

fedn.common.net.grpc.fedn_pb2.ControlResponse

Stop(control: ControlRequest, context)

TODO: Not yet implemented.

Parameters:

• control (fedn.common.net.grpc.fedn_pb2.ControlRequest) – the control request

• context (grpc._server._Context) – the context (unused)

Returns:

the control response

Return type:

fedn.common.net.grpc.fedn_pb2.ControlResponse

flush_model_update_queue()

Clear the model update queue (aggregator).

get_active_trainers()

Get a list of active trainers.

Returns:

the list of active trainers

Return type:

list

get_active_validators()

Get a list of active validators.

Returns:

the list of active validators

Return type:

list

nr_active_trainers()

Get the number of active trainers.

Returns:

the number of active trainers

Return type:

int

nr_active_validators()

Get the number of active validators.

Returns:

the number of active validators

Return type:

int

register_model_validation(validation)

Register a model validation.

Parameters:

validation (fedn.common.net.grpc.fedn_pb2.ModelValidation) – the model validation

report_status(msg, log_level=0, type=None, request=None, flush=True)

Report status of the combiner.

Parameters:

• msg (str) – the message to report

• log_level (fedn.common.net.grpc.fedn_pb2.Status) – the log level to report at

• type (fedn.common.net.grpc.fedn_pb2.Status.Type) – the type of status to report

• request (fedn.common.net.grpc.fedn_pb2.Request) – the request to report status for

• flush (bool) – whether to flush the message to stdout

request_model_update(config, clients=[])

Ask clients to update the current global model.

Parameters:

• config (dict) – the model configuration to send to clients

• clients (list) – the clients to send the request to

request_model_validation(model_id, config, clients=[])

Ask clients to validate the current global model.

Parameters:

• model_id (str) – the model id to validate

• config (dict) – the model configuration to send to clients

• clients (list) – the clients to send the request to

run()

Start the server.

class fedn.network.combiner.server.Role(value)

Bases: Enum

Enum for combiner roles.

COMBINER = 2

OTHER = 4

REDUCER = 3

WORKER = 1

fedn.network.combiner.server.role_to_proto_role(role)

Convert a Role to a proto Role.

Parameters:

role (fedn.network.combiner.server.Role) – the role to convert

Returns:

proto role

Return type:

fedn.common.net.grpc.fedn_pb2.Role