Implement webworkers for parallel data parsing (#49)

Author: msLinuxNinja

package-lock.json

@@ -53,6 +53,7 @@
     ]
   },
   "devDependencies": {
+    "@craco/craco": "^7.1.0",
     "tailwindcss": "latest"
   }
 }

package.json

@@ -10,7 +10,6 @@ const { Dragger } = Upload;
 
 export function DropBox() {
   const { setCpuData, setMemoryData, setSwapData, setBlockData, setNetData, setNetErrData, setFileDetails, setDataLoaded } = useDataContext();
-  
 
   const props = { // props for antd upload component
     multiple: false,
@@ -25,7 +24,7 @@ export function DropBox() {
     try{
       if(fileContent.includes("Linux") && fileContent.includes("all")) { // Check if file is a sar file has the correct content
 
-        const dataObj = callParse(fileContent); // Object containing more objects (inception! 🤯)
+        const dataObj = await callParse(fileContent); // Object containing more objects (inception! 🤯)
         // Save data in context
         setCpuData(dataObj.cpuObject);
         setMemoryData(dataObj.memoryObject);

src/Components/Atoms/DropBox.js

@@ -1,17 +1,26 @@
-import { parseMemoryData, parseSwapData, parseCPUData, parseDiskIO, parseNetworkData, parseNetErrorData, parseFileDetails } from "./parseData";
+import readDataParallel from "./readDataParallel";
+import pparseAllParallel from "./parseAllParallel";
 
-import readData from "./readData";
+export async function callParse(fileContent) {
+  const sarFileData = await readDataParallel(fileContent); // Note these functions don't need to be async as data has already been read
+  const results = await pparseAllParallel(sarFileData);
+  const {
+    cpuObject,
+    memoryObject,
+    swapObject,
+    blockObject,
+    networkObject,
+    networkErrObject,
+    fileDetails,
+  } = results;
 
-export function callParse(fileContent) {
-
-    const sarFileData = readData(fileContent); // Note these functions don't need to be async as data has already been read
-    const cpuObject = parseCPUData(sarFileData);
-    const blockObject = parseDiskIO(sarFileData);
-    const memoryObject = parseMemoryData(sarFileData);
-    const swapObject = parseSwapData(sarFileData);
-    const networkObject = parseNetworkData(sarFileData);
-    const networkErrObject = parseNetErrorData(sarFileData);
-    const fileDetails = parseFileDetails(sarFileData);
-
-    return { cpuObject, memoryObject, swapObject, blockObject, networkObject, networkErrObject, fileDetails }
-}
+  return {
+    cpuObject,
+    memoryObject,
+    swapObject,
+    blockObject,
+    networkObject,
+    networkErrObject,
+    fileDetails,
+  };
+}

src/Utils/callParse.js

@@ -0,0 +1,45 @@
+export default function parseAllParallel(sarFileData) {
+  // List of parse tasks and their output keys
+  const tasks = [
+    "cpuObject",
+    "blockObject",
+    "memoryObject",
+    "swapObject",
+    "networkObject",
+    "networkErrObject",
+    "fileDetails",
+  ];
+
+  // Create one worker per task
+  const promises = tasks.map((key) => {
+    return new Promise((resolve, reject) => {
+      // Create a module worker (supports import statements)
+      const worker = new Worker(new URL("./parseWorker.js", import.meta.url), {
+        type: "module",
+      });
+
+      worker.onmessage = ({ data }) => {
+        const { funcKey, result, error } = data;
+        if (error) {
+          reject(new Error(`Worker ${funcKey} error: ${error}`));
+        } else {
+          resolve({ [funcKey]: result });
+        }
+        worker.terminate();
+      };
+
+      worker.onerror = (err) => {
+        reject(err);
+        worker.terminate();
+      };
+
+      // Start the worker
+      worker.postMessage({ funcKey: key, sarData: sarFileData });
+    });
+  });
+
+  // Combine all results into a single object
+  return Promise.all(promises).then((resultsArray) =>
+    Object.assign({}, ...resultsArray)
+  );
+}

src/Utils/parseAllParallel.js

@@ -0,0 +1,34 @@
+import {
+  parseCPUData,
+  parseDiskIO,
+  parseMemoryData,
+  parseSwapData,
+  parseNetworkData,
+  parseNetErrorData,
+  parseFileDetails,
+} from "./parseData";
+
+onmessage = ({ data }) => {
+  const { funcKey, sarData } = data;
+  const funcMap = {
+    cpuObject: parseCPUData,
+    blockObject: parseDiskIO,
+    memoryObject: parseMemoryData,
+    swapObject: parseSwapData,
+    networkObject: parseNetworkData,
+    networkErrObject: parseNetErrorData,
+    fileDetails: parseFileDetails,
+  };
+
+  if (!(funcKey in funcMap)) {
+    postMessage({ funcKey, error: "Unknown parsing function key." });
+    return;
+  }
+
+  try {
+    const result = funcMap[funcKey](sarData);
+    postMessage({ funcKey, result });
+  } catch (err) {
+    postMessage({ funcKey, error: err.message });
+  }
+};

src/Utils/parseWorker.js

@@ -0,0 +1,33 @@
+/**
+ * readDataParallel
+ * @param {string} data - raw text data (with \n-separated lines)
+ * @returns {Promise<Array>} - resolves to parsed and filtered SAR data
+ */
+export default function readDataParallel(data) {
+  const rows = data.split("\n");
+  const total = rows.length;
+  const numWorkers = navigator.hardwareConcurrency || 4; // Use available CPU cores or default to 4
+  const chunkSize = Math.ceil(total / numWorkers);
+
+  const promises = [];
+  for (let i = 0; i < numWorkers; i++) {
+    const start = i * chunkSize;
+    const end = Math.min(start + chunkSize, total);
+    const chunk = rows.slice(start, end);
+
+    promises.push(
+      new Promise((resolve, reject) => {
+        const worker = new Worker(new URL("./readWorker.js", import.meta.url));
+        worker.onmessage = (msg) => resolve(msg.data);
+        worker.onerror = (err) => reject(err);
+        worker.postMessage(chunk);
+      })
+    );
+  }
+
+  // Combine results from all workers
+  return Promise.all(promises).then((results) => {
+    // URL.revokeObjectURL(woker);
+    return results.flat();
+  });
+}

src/Utils/readData.js

@@ -0,0 +1,21 @@
+function parseRow(row) {
+  if (row.includes("RESTART")) return null;
+
+  const emptySplit = row.split(" ");
+  const commaSplit = emptySplit.toString().split(",");
+  let filtered = commaSplit.filter((e) => e.trim() !== "");
+
+  if (filtered[1] === "AM" || filtered[1] === "PM") {
+    filtered[0] = filtered[0] + " " + filtered[1];
+    filtered.splice(1, 1);
+  }
+  return filtered;
+}
+
+onmessage = function (e) {
+  const rows = e.data;
+  const parsed = rows.map(parseRow).filter((r) => r && String(r).trim());
+  postMessage(parsed);
+  // eslint-disable-next-line
+  close();
+};