M25: WS FX realtime sang Web Worker + MessageChannel worker->worklet (het underrun u1/u10 cold-start + stall main)

2026-09-07 15:11:23 +07:00
parent a88fc5b3f1
commit 9307ecccec
5 changed files with 208 additions and 34 deletions
@@ -0,0 +1,85 @@
+// fx-rt-worker.js — M25: WebSocket FX realtime chạy trên Web Worker.
+// Động lực: main thread bị stall (rAF/React/render, longtask 375ms lúc cold
+// start) làm trễ cả WS receive lẫn postMessage → worklet queue cạn → underrun
+// nghe được (u1/u10). Chuyển WS + deinterleave sang worker; binary output đi
+// THẲNG worker→worklet qua MessageChannel (directPort) — delivery không phụ
+// thuộc event loop main → stall main không còn tạo hole âm thanh. Text
+// (latency/flush_done) vẫn forward main (PDC/UI). Input: worklet gửi thẳng
+// qua directPort ngược chiều (cùng channel, 2 chiều).
+//
+// Protocol main→worker:
+//   {kind:'init', wsUrl, port}   port = MessagePort tới worklet (transfer)
+//   {kind:'send', text|buf}      send 1 frame (text cmd hoặc binary input)
+//   {kind:'close'}
+// Protocol worker→main:
+//   {kind:'open'} | {kind:'close'} | {kind:'text', text}   (chỉ text; binary
+//   không bao giờ chạm main)
+let ws = null;
+let directPort = null;
+let flushing = false; // từ lúc gửi cmd 'flush' tới flush_done — drop output stale (V24, cùng logic fxRt.flushing phía main cũ)
+
+self.onmessage = (ev) => {
+  const d = ev.data;
+  if (!d) return;
+  if (d.kind === 'init') {
+    directPort = d.port || null;
+    if (directPort) {
+      // Input burst 2048 floats (4 block) — tách 4 WS frame 512 floats như
+      // fxRtSendFrames cũ (frame = 1 block 256 stereo; slice() copy đúng
+      // 2048B, KHÔNG subarray().buffer = gửi cả backing 8192B).
+      directPort.onmessage = (e2) => {
+        const d2 = e2.data;
+        if (!d2 || d2.type !== 'in' || !(d2.data instanceof ArrayBuffer)) return;
+        if (!ws || ws.readyState !== 1) return;
+        try {
+          const f = new Float32Array(d2.data);
+          for (let i = 0; i < f.length; i += 512) {
+            const c = new Float32Array(512);
+            c.set(f.subarray(i, i + 512));
+            ws.send(c.buffer);
+          }
+        } catch (e) {}
+      };
+    }
+    try { ws = new WebSocket(d.wsUrl); } catch (e) { self.postMessage({ kind: 'close' }); return; }
+    ws.binaryType = 'arraybuffer';
+    ws.onopen = () => self.postMessage({ kind: 'open' });
+    ws.onmessage = (mev) => {
+      if (typeof mev.data === 'string') {
+        try {
+          const c = JSON.parse(mev.data);
+          if (c && c.cmd === 'flush_done') {
+            flushing = false;
+            // Xóa cờ worklet NGAY trên cùng channel (trước mọi 'out' kế — FIFO)
+            // — không chờ main forward (main có thể bận/stall).
+            if (directPort) directPort.postMessage({ type: 'flush_done' });
+          }
+        } catch (e) {}
+        self.postMessage({ kind: 'text', text: mev.data });
+        return;
+      }
+      if (!(mev.data instanceof ArrayBuffer) || flushing || !directPort) return;
+      // Deinterleave stereo (giống fxRtWsOnMessage cũ) → gửi thẳng worklet.
+      const f = new Float32Array(mev.data);
+      const half = f.length / 2;
+      const l = new Float32Array(half);
+      const r = new Float32Array(half);
+      for (let i = 0, j = 0; i < half; i++) { l[i] = f[j++]; r[i] = f[j++]; }
+      directPort.postMessage({ type: 'out', l, r });
+    };
+    ws.onclose = () => self.postMessage({ kind: 'close' });
+    ws.onerror = () => { try { ws.close(); } catch (e) {} };
+    return;
+  }
+  if (d.kind === 'send') {
+    if (!ws || ws.readyState !== 1) return;
+    if (d.text !== undefined) {
+      try { const c = JSON.parse(d.text); if (c && c.cmd === 'flush') flushing = true; } catch (e) {}
+      try { ws.send(d.text); } catch (e) {}
+    } else if (d.buf !== undefined) {
+      try { ws.send(d.buf); } catch (e) {}
+    }
+    return;
+  }
+  if (d.kind === 'close') { try { if (ws) ws.close(); } catch (e) {} }
+};