Drop remaining aggregated data stream code

Author: ch1bo

ui/public/scenarios.json

@@ -4,29 +4,7 @@
       "name": "Small",
       "topology": "topologies/example.yaml",
       "duration": 120,
-      "trace": "traces/example-nocpu.jsonl",
-      "aggregated": false
-    },
-    {
-      "name": "1x",
-      "topology": "topologies/thousand.yaml",
-      "duration": 300,
-      "trace": "traces/1xsummary.jsonl.gz",
-      "aggregated": true
-    },
-    {
-      "name": "10x",
-      "topology": "topologies/thousand.yaml",
-      "duration": 300,
-      "trace": "traces/10xsummary.jsonl.gz",
-      "aggregated": true
-    },
-    {
-      "name": "100x",
-      "topology": "topologies/thousand.yaml",
-      "duration": 300,
-      "trace": "traces/100xsummary.jsonl.gz",
-      "aggregated": true
+      "trace": "traces/example-nocpu.jsonl"
     }
   ]
 }

ui/public/traces/100xsummary.jsonl.gz

@@ -5,7 +5,7 @@ import StreamWorker from "./worker?worker";
 
 export const useStreamMessagesHandler = () => {
   const {
-    state: { tracePath, aggregated, speedMultiplier },
+    state: { tracePath },
     dispatch,
   } = useSimContext();
   const [streaming, setStreaming] = useState(false);
@@ -26,12 +26,10 @@ export const useStreamMessagesHandler = () => {
       worker.postMessage({
         type: "START",
         tracePath,
-        aggregated,
-        speedMultiplier,
         includeTransactions,
       });
     },
-    [worker, tracePath, aggregated, speedMultiplier, dispatch],
+    [worker, tracePath, dispatch],
   );
 
   const stopStream = useCallback(() => {

ui/public/traces/10xsummary.jsonl.gz

@@ -1,6 +1,3 @@
-import {
-  ISimulationAggregatedDataState,
-} from "@/contexts/SimContext/types";
 import * as cbor from "cborg";
 import type { ReadableStream } from "stream/web";
 import { IServerMessage, EServerMessageType } from "../types";
@@ -9,13 +6,6 @@ export type TWorkerRequest =
   | {
       type: "START";
       tracePath: string;
-      aggregated: true;
-      speedMultiplier: number;
-    }
-  | {
-      type: "START";
-      tracePath: string;
-      aggregated?: false;
       includeTransactions?: boolean;
     }
   | { type: "STOP" };
@@ -172,35 +162,6 @@ const consumeStream = async (
   postMessage({ type: "DONE", tracePath });
 };
 
-// TODO: unused
-const consumeAggregateStream = async (
-  stream: ReadableStream<ISimulationAggregatedDataState>,
-  tracePath: string,
-  speedMultiplier: number,
-) => {
-  let lastTimestamp = 0;
-  for await (const aggregatedData of stream) {
-    const nodes = new Map();
-    for (const [id, stats] of Object.entries(aggregatedData.nodes)) {
-      nodes.set(id, stats);
-    }
-    aggregatedData.nodes = nodes;
-
-    const elapsedMs = (aggregatedData.progress - lastTimestamp) * 1000;
-    lastTimestamp = aggregatedData.progress;
-    await new Promise((resolve) =>
-      setTimeout(resolve, elapsedMs / speedMultiplier),
-    );
-
-    // TODO: re-create when needed
-    // postMessage({
-    //   type: "EVENT",
-    //   tracePath,
-    //   aggregatedData,
-    // } as TWorkerResponse);
-  }
-  postMessage({ type: "DONE", tracePath });
-};
 
 let controller = new AbortController();
 onmessage = (e: MessageEvent<TWorkerRequest>) => {
@@ -211,32 +172,13 @@ onmessage = (e: MessageEvent<TWorkerRequest>) => {
   }
 
   controller = new AbortController();
-  if (request.aggregated) {
-    createEventStream<ISimulationAggregatedDataState>(
-      request.tracePath,
-      controller.signal,
+  createEventStream<IServerMessage>(request.tracePath, controller.signal)
+    .then((stream) =>
+      consumeStream(stream, request.tracePath, request.includeTransactions),
     )
-      .then((stream) =>
-        consumeAggregateStream(
-          stream,
-          request.tracePath,
-          request.speedMultiplier,
-        ),
-      )
-      .catch((err) => {
-        if (err.name !== "AbortError") {
-          throw err;
-        }
-      });
-  } else {
-    createEventStream<IServerMessage>(request.tracePath, controller.signal)
-      .then((stream) =>
-        consumeStream(stream, request.tracePath, request.includeTransactions),
-      )
-      .catch((err) => {
-        if (err.name !== "AbortError") {
-          throw err;
-        }
-      });
-  }
+    .catch((err) => {
+      if (err.name !== "AbortError") {
+        throw err;
+      }
+    });
 };

ui/public/traces/1xsummary.jsonl.gz

@@ -24,7 +24,6 @@ export const defaultAggregatedData: ISimulationAggregatedDataState = {
 export const defaultState: ISimContextState = {
   allScenarios: [],
   activeScenario: "",
-  aggregated: true,
   graph: {
     canvasRef: { current: null },
     canvasOffsetX: 0,

ui/src/components/Sim/hooks/useStreamMessagesHandler.ts

@@ -16,7 +16,6 @@ export const reducer = (
         activeScenario: scenario.name,
         maxTime: scenario.duration,
         tracePath: scenario.trace,
-        aggregated: scenario.aggregated,
         topologyPath: scenario.topology,
       };
     }
@@ -34,7 +33,6 @@ export const reducer = (
         activeScenario: scenario.name,
         maxTime: scenario.duration,
         tracePath: scenario.trace,
-        aggregated: scenario.aggregated,
         topologyPath: scenario.topology,
         topologyLoaded:
           state.topologyLoaded && scenario.topology === state.topologyPath,
@@ -60,9 +58,6 @@ export const reducer = (
     }
 
 
-    case "SET_AGGREGATED_DATA": {
-      return { ...state, aggregatedData: action.payload };
-    }
 
     case "SET_CANVAS_PROPS": {
       return {

ui/src/components/Sim/hooks/worker.ts

@@ -75,7 +75,6 @@ export interface IScenario {
   topology: string;
   duration: number;
   trace: string;
-  aggregated: boolean;
 }
 
 export interface ISimContextState {
@@ -84,7 +83,6 @@ export interface ISimContextState {
   graph: IGraphContextState;
   aggregatedData: ISimulationAggregatedDataState;
   tracePath: string;
-  aggregated: boolean;
   topography: ITransformedNodeMap;
   topologyPath: string;
   topologyLoaded: boolean;
@@ -113,8 +111,6 @@ export type TSimContextActions =
         canvasOffsetY: ((prev: number) => number) | number;
       }>;
     }
-  // TODO: unused
-  | { type: "SET_AGGREGATED_DATA"; payload: ISimulationAggregatedDataState }
   | {
       type: "BATCH_UPDATE";
       payload: Partial<ISimContextState>;