feat: update event streaming UI to improve display of incremental torch progress

the initial live progress added a UI element for every torch step. this PR refines that work to pre-fill UI elements for all epochs and iterations. it marks epochs as InProgress while that epoch's iterations are being processed. etc.

this makes it appear more like a progress bar.

Author: starpit

plugins/plugin-codeflare/src/controller/events/Event.ts

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-export type State = "InProgress" | "Done" | "Error"
+export type State = "Pending" | "InProgress" | "Done" | "Error"
 
 type Event<T extends string, Detail> = Detail & {
   name: string

plugins/plugin-codeflare/src/controller/events/Events.tsx

@@ -19,6 +19,7 @@ import { join } from "path"
 import stripAnsi from "strip-ansi"
 import { Arguments } from "@kui-shell/core"
 
+import { GenericEvent } from "./Event"
 import parseKubeEvents, { collateEvent as collateKubeEvent, KubeEvent } from "./kube"
 import parseTorchEvents, { collateEvent as collateTorchEvent, TorchEvent } from "./torch"
 
@@ -32,7 +33,9 @@ interface EventState {
 
 type State = EventState & {
   nKubeEvents: number
+  nNotPendingKubeEvents: number
   nTorchEvents: number
+  nNotPendingTorchEvents: number
   catastrophicError?: Error
 }
 
@@ -57,6 +60,8 @@ class Events extends React.PureComponent<Props, State> {
       kubeEvents,
       torchEvents,
       nKubeEvents: kubeEvents.length,
+      nNotPendingKubeEvents: this.nNotPending(kubeEvents),
+      nNotPendingTorchEvents: this.nNotPending(torchEvents),
       nTorchEvents: torchEvents.length,
     }
 
@@ -83,6 +88,7 @@ class Events extends React.PureComponent<Props, State> {
               toBeProcessed.forEach((line) => collateKubeEvent(curState.kubeEvents, line))
               return {
                 nKubeEvents: curState.kubeEvents.length,
+                nNotPendingKubeEvents: this.nNotPending(curState.kubeEvents),
               }
             })
           }, queueFlushHysteresis)
@@ -108,6 +114,7 @@ class Events extends React.PureComponent<Props, State> {
               toBeProcessed.forEach((line) => collateTorchEvent(curState.torchEvents, line))
               return {
                 nTorchEvents: curState.torchEvents.length,
+                nNotPendingTorchEvents: this.nNotPending(curState.torchEvents),
               }
             })
           }, queueFlushHysteresis)
@@ -116,6 +123,10 @@ class Events extends React.PureComponent<Props, State> {
     }
   }
 
+  private nNotPending(events: GenericEvent[]) {
+    return events.reduce((N, _) => N + (_.state !== "Pending" ? 1 : 0), 0)
+  }
+
   public static getDerivedStateFromError(error: Error) {
     return { catastrophicError: error }
   }

plugins/plugin-codeflare/src/controller/events/torch.ts

@@ -20,18 +20,44 @@ type EventType = "Epoch" | "Iteration" | "Marker"
 type Detail = { epoch: number; step: number; nSteps: number; ip: string }
 export type TorchEvent = Event<EventType, Detail>
 
-function findPrevious(M: TorchEvent[], ip: TorchEvent["ip"], type: EventType) {
+function findPrevious(
+  M: TorchEvent[],
+  ip: TorchEvent["ip"],
+  type: EventType,
+  state: TorchEvent["state"],
+  step = -1,
+  epoch = -1
+) {
   for (let idx = M.length - 1; idx >= 0; idx--) {
     const evt = M[idx]
-    if (evt.type === type && evt.ip === ip) {
+    if (
+      evt.type === type &&
+      evt.ip === ip &&
+      evt.state === state &&
+      (step === -1 || evt.step === step) &&
+      (epoch === -1 || evt.epoch === epoch)
+    ) {
       return evt
     }
   }
 }
 
-function findEpoch(M: TorchEvent[], ip: TorchEvent["ip"]) {
-  const evt = findPrevious(M, ip, "Epoch")
-  return evt ? evt.step : -1
+function findEpoch(M: TorchEvent[], ip: TorchEvent["ip"], state: TorchEvent["state"] = "InProgress", step?: number) {
+  return findPrevious(M, ip, "Epoch", state, step)
+}
+
+class TorchEventImpl implements TorchEvent {
+  public constructor(
+    public readonly name: string,
+    public readonly ip: string,
+    public readonly type: EventType,
+    public readonly step: number,
+    public readonly nSteps: number,
+    public readonly epoch: number,
+    public readonly timestamp: number,
+    public state: TorchEvent["state"] = "InProgress",
+    public readonly message = `Epoch ${epoch}${type !== "Epoch" ? ` - ${type} ${step}` : ""} of ${nSteps}`
+  ) {}
 }
 
 export function collateEvent(M: TorchEvent[], line: string) {
@@ -46,7 +72,7 @@ export function collateEvent(M: TorchEvent[], line: string) {
     const epoch = -1
     const step = -1
     const nSteps = -1
-    const state = "InProgress"
+    const state = "Done"
     M.push({ ip, name, message, state, type, hidden, timestamp, epoch, step, nSteps })
     return M
   }
@@ -56,36 +82,63 @@ export function collateEvent(M: TorchEvent[], line: string) {
     const ip = match[1]
     const type = match[2] as EventType
     // const percentage = parseInt(match[3], 10)
-    const step = parseInt(match[4], 10)
+    const step = parseInt(match[4], 10) - (type === "Epoch" ? 0 : 1)
     const nSteps = parseInt(match[5], 10)
 
-    const epoch = type === "Epoch" ? step : findEpoch(M, ip)
-    const timestampMarker = findPrevious(M, ip, "Marker")
-
-    const event = {
-      name: `Torch Training on ${ip}`,
-      message: `Epoch ${epoch}${type !== "Epoch" ? ` - ${type} ${step}` : ""} of ${nSteps}`,
-      ip,
-      type,
-      step,
-      nSteps,
-      epoch,
-      timestamp: timestampMarker ? timestampMarker.timestamp : Date.now(),
-      state: "InProgress" as const,
+    const epoch =
+      type === "Epoch"
+        ? { step, nSteps, state: "InProgress" }
+        : findEpoch(M, ip) || { step: -1, nSteps: 0, state: "InProgress" }
+    const name = `Torch Training on ${ip}`
+    const timestampMarker = findPrevious(M, ip, "Marker", "Done")
+    const timestamp = timestampMarker ? timestampMarker.timestamp : Date.now()
+
+    if (type === "Iteration") {
+      epoch.state = "InProgress"
+    } else if (step > 0) {
+      const thisEpoch = findEpoch(M, ip, "Pending", step)
+      if (thisEpoch) {
+        thisEpoch.state = "InProgress"
+      }
     }
 
     // find previous by ip and mark it Done
-    const prev = findPrevious(M, ip, type)
-    if (prev) {
-      prev.state = "Done"
-
-      if (type === "Epoch" && prev.step === step) {
-        // strange, torch seems to repeat the e.g. Epoch 6/6 event...
-        return M
+    if (step > 0) {
+      const prev =
+        type === "Iteration"
+          ? findPrevious(M, ip, type, "Pending", step - 1, epoch.step) // previous iteration in this epoch
+          : findPrevious(M, ip, type, "InProgress", step - 1, epoch.step - 1) // previous epoch
+      if (prev) {
+        prev.state = "Done"
+      } else if (type === "Iteration" && step === nSteps - 1) {
+        // torch repeat the last step to indicate fully done
+        const prev = findPrevious(M, ip, type, "Pending", nSteps - 1, epoch.step) // last iteration
+        if (prev) {
+          prev.state = "Done"
+        }
       }
     }
 
-    M.push(event)
+    if (type === "Epoch" && step === 0) {
+      // first Epoch
+      M.push(new TorchEventImpl(name, ip, type, step, nSteps, epoch.step, timestamp))
+    } else if (type === "Iteration" && step === 0 && epoch.step === 0) {
+      // first Iteration of first Epoch: pre-fill the remaining Epochs and Iterations
+      M.push(new TorchEventImpl(name, ip, type, step, nSteps, epoch.step, timestamp, "Pending"))
+
+      // pre-fill remaining Iterations for epoch 0
+      for (let iterIdx = 1; iterIdx < nSteps; iterIdx++) {
+        M.push(new TorchEventImpl(name, ip, "Iteration", iterIdx, nSteps, 0, timestamp, "Pending"))
+      }
+
+      // now pre-fill the remaining Epochs
+      for (let epochIdx = 1; epochIdx < epoch.nSteps; epochIdx++) {
+        M.push(new TorchEventImpl(name, ip, "Epoch", epochIdx, epoch.nSteps, epochIdx, timestamp, "Pending"))
+        for (let iterIdx = 1; iterIdx < nSteps; iterIdx++) {
+          M.push(new TorchEventImpl(name, ip, "Iteration", iterIdx, nSteps, epochIdx, timestamp, "Pending"))
+        }
+      }
+    }
   }
 
   return M

plugins/plugin-codeflare/web/scss/components/Dashboard/Grid.scss

@@ -64,9 +64,9 @@ $fullWidth: 1em; /* $large * ($unit + $rgap) - $rgap */
   }
 }
 
-@mixin Color($color) {
+@mixin Color($color, $important: "") {
   @include CFCellContent {
-    background-color: $color;
+    background-color: $color $important;
   }
 }
 
@@ -91,7 +91,11 @@ $fullWidth: 1em; /* $large * ($unit + $rgap) - $rgap */
       @include Color(var(--color-base0B));
     }
     @include State("InProgress") {
-      @include Color(var(--color-base09));
+      @include Color(var(--color-base0A), !important);
+    }
+    @include State("Pending") {
+      @include Color(var(--color-base04), !important);
+      filter: saturate(0.3) brightness(0.95);
     }
 
     @include Color(var(--color-base0D));
@@ -117,7 +121,7 @@ $fullWidth: 1em; /* $large * ($unit + $rgap) - $rgap */
     grid-column: span $tiny;
     @include CFCellContent {
       background-color: var(--color-base0D);
-      filter: saturate(0.3) brightness(0.9);
+      filter: saturate(0.5) brightness(0.85);
       height: 80%;
     }
   }