scaling live time series data using simd vs thi.ng/geom transform

e96af198_b9a6_4445_9404_1ead6e0de9f6.ts

import * as rx from "@thi.ng/rstream";
import * as tx from "@thi.ng/transducers";
import * as g from "@thi.ng/geom";
import { $canvas } from "@thi.ng/rdom-canvas";
import { randomID, pickRandom } from "@thi.ng/random";
import { IObjectOf, Fn, identity } from "@thi.ng/api";
import { Vec, Vec2, safeDiv2, sub2 } from "@thi.ng/vectors";
import { scaleWithCenter23, translation23, mulM23 } from "@thi.ng/matrices";
import { binarySearch, bsLE } from "@thi.ng/arrays";
import { init } from "@thi.ng/simd";
import { equiv } from "@thi.ng/equiv";

interface KeyedRandomWalkState {
  state: IObjectOf<number>;
  cur: [string, number];
}
const keyedRndWalk = (n: number) =>
  tx.comp(
    tx.scan(
      tx.reducer(
        (): KeyedRandomWalkState => ({
          state: tx.assocObj(tx.map(() => [randomID(), 0], tx.range(n))),
          cur: ["", 0],
        }),
        (acc, _) => {
          const k = pickRandom(Object.keys(acc.state));
          const v = Math.random() - 0.5;
          acc.state[k] += v;
          acc.cur = [k, acc.state[k]];
          return acc;
        },
      ),
    ),
    tx.map(({ cur }) => cur),
  );

const numberOfSeries = 5;
const tsStream = rx
  .fromInterval(50)
  .transform(keyedRndWalk(numberOfSeries))
  .map((x): [number, string, number] => [+new Date(), ...x])
  .map(([ts, k, v]): [number, string, number] => [
    ts + Math.floor(Math.random() * 10),
    k,
    v,
  ]);
// tsStream.subscribe(rx.trace("tsStream"));

interface SparseTimeSeries {
  ts: number[];
  features: IObjectOf<Vec2[]>;
}
const binaryInsert = <A, B>(arr: A[], item: A, key?: Fn<A, B>) => {
  key = key ?? <Fn<A, B>>identity;
  const i = bsLE(binarySearch(arr, item, key));
  if (i === -1 || i === arr.length - 1) {
    arr.push(item);
  } else if (!equiv(key(arr[i]), key(item))) {
    arr.splice(i, 0, item);
  }
};
const tsData = tsStream.transform(
  tx.scan(
    tx.reducer(
      (): SparseTimeSeries => ({ ts: [], features: {} }),
      (acc, [ts, k, v]) => {
        if (!acc.features[k]) acc.features[k] = [];
        binaryInsert(acc.ts, ts);
        binaryInsert(acc.features[k], <Vec>[ts, v], (x) => x[0]);
        return acc;
      },
    ),
  ),
);
// tsData.subscribe(rx.trace("tsData"));

const simd = init(new WebAssembly.Memory({ initial: 1 }))!;
const root = document.getElementById("root")!;
const dims = [500, 500];
const canvasBody = tsData.map(({ features }) => {
  const [_, series] = tx.pairs(features).next().value;

  // SIMD Implementation
  const buf = simd.f32.subarray(0, series.length * 2);
  buf.set([...tx.flatten1(series)]);
  const points = Vec2.mapBuffer(buf);
  const lineSimd = g.polyline(points, { stroke: "#fff" });
  const src = g.bounds(lineSimd)!;
  const dst = g.rect([0, 0], dims);
  const tsl = translation23([], sub2([], dst.pos, src.pos));
  const scl = scaleWithCenter23([], dst.pos, safeDiv2([], dst.size, src.size));
  const tfm = mulM23([], scl, tsl);
  const tfmBuf = simd.f32.subarray(points.length + 8, points.length + 8 + tfm.length);
  tfmBuf.set(tfm);
  simd.mul_m23v2_aos(buf.byteOffset, tfmBuf.byteOffset, buf.byteOffset, series.length);

  // Geom Implementation
  {
    const linePre = g.polyline(series, { stroke: "#fff" });
    const src = g.bounds(linePre)!;
    const dst = g.rect([0, 0], dims);
    const tsl = translation23([], sub2([], dst.pos, src.pos));
    const scl = scaleWithCenter23([], dst.pos, safeDiv2([], dst.size, src.size));
    const tfm = mulM23([], scl, tsl);
    const lineGeom = g.transform(linePre, tfm);
    lineSimd;
    debugger;
  }

  return g.group({ __background: "#000" }, [lineSimd]);
});
$canvas(rx.syncRAF(canvasBody), dims).mount(root);