Fix UI issues

datasets/populate

@@ -10,10 +10,8 @@ curl 'https://storage.googleapis.com/deai-313515.appspot.com/example_training_da
 
 # lungs ultrasound
 mkdir -p lus_covid
-curl 'https://drive.switch.ch/index.php/s/zM5ZrUWK3taaIly/download' > archive.zip
-ln -fs lus_covid DeAI-testimages # redirect top level dir
-unzip -u archive.zip
-rm archive.zip DeAI-testimages
+curl 'https://storage.googleapis.com/deai-313515.appspot.com/lus_covid.tar.gz'|
+	tar -xz
 
 # wikitext
 mkdir -p wikitext

discojs/src/client/client.ts

@@ -167,6 +167,8 @@ export abstract class Client<N extends Network> extends EventEmitter<{
         if (this.#previousStatus !== undefined)
           this.emit("status", this.#previousStatus);
         this.nbOfParticipants = event.nbOfParticipants;
+        // Make sure to set the promise back to undefined
+        this.promiseForMoreParticipants = undefined;
         resolve();
       });
     });
@@ -181,8 +183,6 @@ export abstract class Client<N extends Network> extends EventEmitter<{
       );
       this.emit("status", "not enough participants");
       await this.promiseForMoreParticipants;
-      // Make sure to set the promise back to undefined once resolved
-      this.promiseForMoreParticipants = undefined;
     }
   }
   /**

discojs/src/client/decentralized/README.md

@@ -0,0 +1,59 @@
+# Decentralized Event flow
+
+```mermaid
+sequenceDiagram
+    autonumber
+    participant T as Trainer
+    participant C as DecentralizedClient
+    participant S as Server
+    participant P as Peers
+
+    rect rgb(240,240,240)
+        Note over T,S: 1. Connecting
+        T->>C: connect()
+        C->>S: WebSocket connect + ClientConnected
+        S-->>C: NewDecentralizedNodeInfo { id, nbOfParticipants, waitForMoreParticipants }
+        C-->>T: base model
+    end
+
+    rect rgb(240,240,240)
+        Note over T,S: 2. Round begin
+        T->>C: onRoundBeginCommunication()
+        C->>S: JoinRound
+        Note over C: status "local training"
+        T->>T: local training
+    end
+
+    rect rgb(240,240,240)
+        Note over T,P: 3. Round end, server barrier
+        T->>C: onRoundEndCommunication(weights)
+        Note over C: status "waiting for peers to share weights"
+        C->>S: PeerIsReady
+        S-->>C: PeersForRound { peers, aggregationRound }
+        Note over C: status "connecting to peers"
+    end
+
+    rect rgb(240,240,240)
+        Note over C,P: 4. Peer connections
+        C->>S: SignalForPeer { peer, offer/answer/candidate }
+        S->>P: SignalForPeer (forwarded)
+        P-->>C: SignalForPeer (forwarded back)
+        Note over C,P: WebRTC data channel open
+    end
+
+    rect rgb(240,240,240)
+        Note over C,P: 5. Weight exchange
+        Note over C: status "updating model"
+        C->>P: Payload { aggregationRound, communicationRound, weights }
+        P-->>C: Payload from each peer
+        Note over C: aggregator aggregates once full
+        C-->>T: aggregated weights
+    end
+
+    opt participants drop below the minimum, at any point
+        S-->>C: WaitingForMoreParticipants
+        Note over C: status "not enough participants", block before sending weights
+        S-->>C: EnoughParticipants
+        Note over C: resume, re-emit the previous status
+    end
+```

discojs/src/client/decentralized/decentralized_client.ts

@@ -25,6 +25,8 @@ const debug = createDebug("discojs:client:decentralized");
  * help of the network's server, yet only exchange payloads between each other. Communication
  * with the server is based off regular WebSockets, whereas peer-to-peer communication uses
  * WebRTC for Node.js.
+ *
+ * See decentralized README.md for schema of the event flow.
  */
 export class DecentralizedClient extends Client<"decentralized"> {
   /**
@@ -163,7 +165,8 @@ export class DecentralizedClient extends Client<"decentralized"> {
     }
     // Save the status in case participants leave and we switch to waiting for more participants
     // Once enough new participants join we can display the previous status again
-    this.saveAndEmit("connecting to peers");
+    // We are done with our round and now wait for the peers to be done with theirs
+    this.saveAndEmit("waiting for peers to share weights");
     // First we check if we are waiting for more participants before sending our weight update
     await this.waitForParticipantsIfNeeded();
     // Create peer-to-peer connections with all peers for the round
@@ -204,6 +207,8 @@ export class DecentralizedClient extends Client<"decentralized"> {
         this.server,
         MType.PeersForRound,
       );
+      // every peer is ready to share weights, we can now connect to them
+      this.saveAndEmit("connecting to peers");
 
       const peers = Set(receivedMessage.peers);
 

discojs/src/client/federated/README.md

@@ -0,0 +1,59 @@
+# Federated Event flow
+
+```mermaid
+sequenceDiagram
+    autonumber
+    participant T as Trainer
+    participant C as FederatedClient
+    participant S as Server
+    participant O as Other clients
+
+    rect rgb(240,240,240)
+        Note over T,S: 1. Connecting
+        T->>C: connect()
+        C->>S: WebSocket connect + ClientConnected
+        S-->>C: NewFederatedNodeInfo { id, payload, round, nbOfParticipants, waitForMoreParticipants }
+        C-->>T: base model with the latest global weights
+    end
+
+    rect rgb(240,240,240)
+        Note over T,S: 2. Round begin
+        T->>C: onRoundBeginCommunication()
+        Note over C: status "local training"
+        T->>T: local training
+    end
+
+    rect rgb(240,240,240)
+        Note over T,S: 3. Round end, sending the local update
+        T->>C: onRoundEndCommunication(weights)
+        Note over C: status "updating model"
+        C->>S: SendPayload { payload, round }
+    end
+
+    rect rgb(240,240,240)
+        Note over C,O: 4. Server aggregation
+        O->>S: SendPayload from the other clients
+        Note over S: MeanAggregator waits for all<br/>registered clients of the round
+        Note over S: aggregate, save as the latest global weights
+    end
+
+    rect rgb(240,240,240)
+        Note over T,O: 5. Global update
+        S-->>C: ReceiveServerPayload { payload, round, nbOfParticipants }
+        S-->>O: ReceiveServerPayload
+        Note over C: aggregator.setRound(round)
+        C-->>T: global weights
+    end
+
+    opt stale or invalid contribution
+        Note over S: contribution dropped, no aggregation
+        S-->>C: ReceiveServerPayload with the previous round's global weights
+    end
+
+    opt participants drop below the minimum, at any point
+        S-->>C: WaitingForMoreParticipants
+        Note over C: status "not enough participants", block before sending weights
+        S-->>C: EnoughParticipants
+        Note over C: resume, re-emit the previous status
+    end
+```

discojs/src/client/federated/federated_client.ts

@@ -21,6 +21,9 @@ const SERVER_NODE_ID = "federated-server-node-id";
 /**
  * Client class that communicates with a centralized, federated server, when training
  * a specific task in the federated setting.
+ *
+ * See federated README.md for schema of the event flow.
+ *
  */
 export class FederatedClient extends Client<"federated"> {
   /**

discojs/src/training/types.ts

@@ -17,6 +17,7 @@ export type SummaryLogs = {
 
 export type RoundStatus =
   | "not enough participants" // Server notification to wait for more participants
+  | "waiting for peers to share weights" // for decentralized only, the other peers are still training their round
   | "updating model" // fetching/aggregating local updates into a global model
   | "local training" // Training the model locally
-  | "connecting to peers"; // for decentralized only, fetch the server's list of participating peers
+  | "connecting to peers"; // for decentralized only, establishing the peer-to-peer connections

server/tests/e2e/decentralized.spec.ts

@@ -160,7 +160,11 @@ describe("end-to-end decentralized", { timeout: 50_000 }, () => {
     await reachConsensus(url, "secure", 3);
   });
 
-  it("peers emit expected events", { timeout: 100_000 }, async () => {
+  /** The LUS COVID task, decentralized between at least two participants */
+  async function lusCovidDecentralized(): Promise<{
+    task: Task<"image", "decentralized">;
+    taskProvider: TaskProvider<"image", "decentralized">;
+  }> {
     const baseTask = await defaultTasks.lusCovid.getTask();
     const task: Task<"image", "decentralized"> = {
       ...baseTask,
@@ -172,10 +176,17 @@ describe("end-to-end decentralized", { timeout: 50_000 }, () => {
         minNbOfParticipants: 2,
       },
     };
-    const taskProvider = {
-      ...defaultTasks.lusCovid,
-      getTask: () => Promise.resolve(task),
+    return {
+      task,
+      taskProvider: {
+        ...defaultTasks.lusCovid,
+        getTask: () => Promise.resolve(task),
+      },
     };
+  }
+
+  it("peers emit expected events", { timeout: 100_000 }, async () => {
+    const { task, taskProvider } = await lusCovidDecentralized();
     const url = await startServer(defaultModels.LUSClassifier, taskProvider);
     const dataset = await datasets.loadLusCOVID();
 
@@ -187,11 +198,11 @@ describe("end-to-end decentralized", { timeout: 50_000 }, () => {
      * (without waiting for a server answer)
      * b) local training (the status remains "local training")
      * c) During onRoundEndCommunication
-     *   1. the peer notifies the server that they are ready to share weights
-     *      set status to "connecting to peers"
+     *   1. the peer sets its status to "waiting for peers to share weights"
+     *      and notifies the server that they are ready to share weights
      *   2. wait for the server to answer with the current round's peers list
      *      this is where the nb of participants is updated
-     *   3. establish peer-to-peer connections
+     *   3. set status to "connecting to peers" and establish the connections
      *   4. set status to "updating model" and exchange weight updates
      *
      * Given this, it is important to note that calling disco.trainByRound().next()
@@ -235,7 +246,9 @@ describe("end-to-end decentralized", { timeout: 50_000 }, () => {
     // Calling next() a 2nd time makes User 1 go to c) where the peer should
     // stay stuck awaiting until another participant joins
     const logUser1Round2Promise = generatorUser1.next();
-    expect(await statusUser1.next()).equal("connecting to peers"); // tries to connect to peers
+    expect(await statusUser1.next()).equal(
+      "waiting for peers to share weights",
+    ); // ready to share
     expect(await statusUser1.next()).equal("not enough participants"); // but has to wait for more participants
 
     /* USER 2 JOINS */
@@ -264,7 +277,9 @@ describe("end-to-end decentralized", { timeout: 50_000 }, () => {
     // User 2 did a) and b)
     expect(await statusUser2.next()).equal("local training");
     // User 1 is still in c) now waiting for user 2 to be ready to exchange weight updates
-    expect(await statusUser1.next()).equal("connecting to peers");
+    expect(await statusUser1.next()).equal(
+      "waiting for peers to share weights",
+    );
 
     /* ROUND 2 */
 
@@ -282,10 +297,14 @@ describe("end-to-end decentralized", { timeout: 50_000 }, () => {
     expect(await nbParticipantsUser2.next()).equal(2);
     expect(await nbParticipantsUser1.next()).equal(2);
     // User 1 and 2 did c), a) and b)
+    expect(await statusUser1.next()).equal("connecting to peers");
     expect(await statusUser1.next()).equal("updating model"); // second to last
     expect(await statusUser1.next()).equal("local training");
 
-    expect(await statusUser2.next()).equal("connecting to peers"); // back to connecting when user 1 joins
+    expect(await statusUser2.next()).equal(
+      "waiting for peers to share weights",
+    );
+    expect(await statusUser2.next()).equal("connecting to peers");
     expect(await statusUser2.next()).equal("updating model");
     expect(await statusUser2.next()).equal("local training");
 
@@ -302,7 +321,9 @@ describe("end-to-end decentralized", { timeout: 50_000 }, () => {
     const logUser2Round3Promise = generatorUser2.next();
     // await new Promise((res, _) => setTimeout(res, statusUpdateTime)) // Wait some time for the status to update
     // starts c) and waits for user 3 to join
-    expect(await statusUser2.next()).equal("connecting to peers");
+    expect(await statusUser2.next()).equal(
+      "waiting for peers to share weights",
+    );
     expect(await statusUser2.next()).equal("not enough participants");
 
     /* USER 3 JOINS */
@@ -333,7 +354,9 @@ describe("end-to-end decentralized", { timeout: 50_000 }, () => {
     // User 3 did a) and b)
     expect(await statusUser3.next()).equal("local training");
     // User 2 is still in c) waiting for user 3 to be ready to exchange waits
-    expect(await statusUser2.next()).equal("connecting to peers");
+    expect(await statusUser2.next()).equal(
+      "waiting for peers to share weights",
+    );
 
     /* ROUND 3 */
 
@@ -350,9 +373,13 @@ describe("end-to-end decentralized", { timeout: 50_000 }, () => {
     expect(await nbParticipantsUser2.next()).equal(2);
 
     // both user 2 and 3 did c), a) and are now in b)
+    expect(await statusUser2.next()).equal("connecting to peers");
     expect(await statusUser2.next()).equal("updating model");
     expect(await statusUser2.next()).equal("local training");
 
+    expect(await statusUser3.next()).equal(
+      "waiting for peers to share weights",
+    );
     expect(await statusUser3.next()).equal("connecting to peers");
     expect(await statusUser3.next()).equal("updating model");
     expect(await statusUser3.next()).equal("local training");
@@ -365,4 +392,64 @@ describe("end-to-end decentralized", { timeout: 50_000 }, () => {
 
     await discoUser3.close();
   });
+
+  // regression test, peer used to display missing participants when
+  // it was not the case
+  it(
+    "peer sharing its weights doesn't report missing participants",
+    { timeout: 100_000 },
+    async () => {
+      const { task, taskProvider } = await lusCovidDecentralized();
+      const url = await startServer(defaultModels.LUSClassifier, taskProvider);
+      const dataset = await datasets.loadLusCOVID();
+
+      /**
+       * The timeline is:
+       * - User 1 joins the task by themselves and trains locally
+       * - User 2 joins while User 1 is still training
+       * - User 1 is done training and waits for User 2 to share its weights
+       *
+       * User 1 has to wait for User 2 to be ready but shouldn't be told that
+       * participants are missing: User 2 is here, only still training.
+       */
+
+      /* USER 1 JOINS */
+
+      const discoUser1 = new Disco(task, url, { preprocessOnce: true });
+      const statusUser1 = new Queue<RoundStatus>();
+      discoUser1.on("status", (status) => {
+        statusUser1.put(status);
+      });
+      const generatorUser1 = discoUser1.trainByRound(dataset);
+
+      await generatorUser1.next(); // a) and b)
+      expect(await statusUser1.next()).equal("local training");
+
+      /* USER 2 JOINS, WHILE USER 1 IS STILL TRAINING */
+
+      const discoUser2 = new Disco(task, url, { preprocessOnce: true });
+      const generatorUser2 = discoUser2.trainByRound(dataset);
+      await generatorUser2.next(); // a) and b)
+
+      // there are enough participants now, User 1 keeps on training
+      expect(await statusUser1.next()).equal("local training");
+
+      /* USER 1 IS DONE TRAINING */
+
+      const logUser1Round2 = generatorUser1.next(); // c)
+      expect(await statusUser1.next()).equal(
+        "waiting for peers to share weights",
+      );
+
+      /* USER 2 IS DONE TRAINING TOO */
+
+      await generatorUser2.next();
+      await logUser1Round2;
+      expect(await statusUser1.next()).equal("connecting to peers");
+      expect(await statusUser1.next()).equal("updating model");
+
+      await discoUser1.close();
+      await discoUser2.close();
+    },
+  );
 });