blazefl.core.ThreadPoolClientTrainer

blazefl.core.ThreadPoolClientTrainer#

class blazefl.core.ThreadPoolClientTrainer(*args, **kwargs)[source]#

Bases: BaseClientTrainer[UplinkPackage, DownlinkPackage], Protocol[UplinkPackage, DownlinkPackage]

Abstract base class for parallel client training using a thread pool.

This class enables parallel processing of clients within a processes.

num_parallels#

Number of parallel threads to use.

Type:

int

device#

Primary device for computation (e.g., “cpu”, “cuda”).

Type:

str

device_count#

Number of available CUDA devices for distribution.

Type:

int

cache#

Cache to store results from clients.

Type:

list[UplinkPackage]

stop_event#

Event to signal workers to stop.

Type:

threading.Event

Raises:

NotImplementedError – If the abstract methods are not implemented in a subclass.

__init__(*args, **kwargs)#

Methods

__init__(*args, **kwargs)

get_client_device(cid)

Retrieve the device to use for processing a given client.

local_process(payload, cid_list)

Manage the parallel processing of clients using threads.

progress_fn(it)

A no-op progress function that can be overridden to provide custom progress tracking.

uplink_package()

Prepare the data package to be sent from the client to the server.

worker(cid, device, payload, stop_event)

Process a single client's training task in a thread.

Attributes

num_parallels

device

device_count

cache

stop_event

cache: list[UplinkPackage]#
device: str#
device_count: int#
get_client_device(cid: int) → str[source]#

Retrieve the device to use for processing a given client.

Parameters:

cid (int) – Client ID.

Returns:

The device to use for processing the client.

Return type:

str

local_process(payload: DownlinkPackage, cid_list: list[int]) → None[source]#

Manage the parallel processing of clients using threads.

This method distributes the processing of multiple clients across a pool of threads.

Parameters:

  • payload (DownlinkPackage) – The data package received from the server.

  • cid_list (list[int]) – A list of client IDs to process.

num_parallels: int#
progress_fn(it: list[Future[UplinkPackage]]) → Iterable[Future[UplinkPackage]][source]#

A no-op progress function that can be overridden to provide custom progress tracking.

Parameters:

it (list[Future[UplinkPackage]]) – A list of Future objects representing the results of client processing.

Returns:

The original iterable.

Return type:

Iterable[Future[UplinkPackage]]

stop_event: Event#
worker(cid: int, device: str, payload: DownlinkPackage, stop_event: Event) → UplinkPackage[source]#

Process a single client’s training task in a thread.

Parameters:

  • cid (int) – The client ID.

  • device (str) – The device to use for processing this client.

  • payload (DownlinkPackage) – The data package received from the server.

  • stop_event (threading.Event) – Event to signal stopping the worker.

Returns:

The uplink package containing the client’s results.

Return type:

UplinkPackage

Contents