Example: PowerOfChoice

This section describes how to use FLGo to implement algorithms that make changes during the user sampling phase. The example used here is PowerOfChoice, an algorithm proposed by Cho et al. in 2020 (link to paper). Compared with the traditional unbiased sampling strategy, this method uses a biased but faster convergence sampling method, that is, it preferentially samples those users with large losses of local datasets. To achieve this, its sampling steps are summarized as follows:

1. The server does not put back the sampled \(d\) candidate users from all \(K\) users according to the size ratio of the dataset (\(m<=d<=K\), \(m\) is the actual number of sampled users in the current round of the server);
2. The server broadcasts the current global model \(\theta^t\) to \(d\) candidate users, evaluates their local dataset loss, and these users send back the loss value \(F_k(\theta^t)\);
3. The server sorts according to the loss value sent back by \(d\) candidate users, and preferentially selects the first \(m\) users with the largest loss to participate in this round of training

The following describes how to implement this sampling strategy with FLGo.

import numpy as np
import flgo.algorithm.fedavg as fedavg
from flgo.algorithm.fedbase import BasicServer
import flgo.system_simulator.base as ss
import os
import flgo

class Server(BasicServer):
    def initialize(self, *args, **kwargs):
        self.init_algo_para({'d': self.num_clients})

    def sample(self):
        # create candidate set A
        num_candidate = min(self.d, len(self.available_clients))
        p_candidate = np.array([len(self.clients[cid].train_data) for cid in self.available_clients])
        candidate_set = np.random.choice(self.available_clients, num_candidate, p=p_candidate / p_candidate.sum(), replace=False)
        candidate_set = sorted(candidate_set)
        # communicate with the candidates for their local loss
        losses = []
        for cid in candidate_set:
            losses.append(self.clients[cid].test(self.model, dataflag='train')['loss'])
        # sort candidate set according to their local loss value, and choose the top-M highest ones
        sort_id = np.array(losses).argsort().tolist()
        sort_id.reverse()
        num_selected = min(self.clients_per_round, len(self.available_clients))
        selected_clients = np.array(self.available_clients)[sort_id][:num_selected]
        return selected_clients.tolist()

class powerofchoice:
    Server=Server
    Client=fedavg.Client

First, the algorithm has a hyperparameter d to control the number of candidates, so implement the hyperparameter in the initialization method initialize and set the default value to the total number of users;

Then, the number of candidates is determined to be the smaller value of the current active user and d, and the candidate_set of the candidate set is obtained by sampling according to the size ratio of its dataset.

Then, for convenience, instead of rewriting communication-related content, directly call the candidates' test functions to obtain their local dataset loss (the two effects are equivalent, and rewriting communication-related code is more troublesome). The set of candidates is then sorted based on loss.

Finally, the top self.clients_per_round users with the most losses are selected and their IDs are returned.

Note: The decorator of the sample method is ss.with_availability to instantly refresh the user's usability, and this function is to achieve system heterogeneity, which will be explained in subsequent chapters