Distributed Spectral Analysis with FFTW-WASM (Web Browser)
This tutorial distributes a real signal-processing workload across the DCP network: a set of independent sensor readings, each analyzed with a real Fast Fourier Transform to find its dominant frequency, using FFTW compiled to WebAssembly as the transform engine inside each worker.
It assumes you’ve already seen the basic DCP job tutorial; this one reuses the same shape (input set → work function → compute.for → events → results) and adds one new ingredient: a compiled WASM module, running inside a sandboxed worker.
Note
A browser job author has no local job.requires() – that convenience (bundling a local file and everything it transitively require()s) only exists for Node.js. So instead of require()-ing a module inside the worker, this tutorial fetches the compiled module client-side and ships it as a job argument, the same value delivered to every invocation of the work function. See Building WASM modules for DCP jobs for how the module itself gets built – specifically why compiling with SINGLE_FILE=1 means no extra .wasm fetch and no custom instantiateWasm code is needed here at all.
Prerequisites
A modern web browser
A DCP Portal account
fftw-module.js, built as described in Building WASM modules for DCP jobs, served alongside your HTML page
The scenario
A predictive-maintenance system reads short vibration snapshots from several sensors on different pieces of machinery. For each sensor, the question is the same: what frequency dominates this snapshot, and is its magnitude large enough to flag for review? Each sensor’s reading is independent of every other sensor’s – exactly the shape distributed compute wants: one sensor’s snapshot is one unit of work, and a DCP worker can answer the question completely on its own with nothing but the samples it was handed, the compiled FFT module, and the anomaly threshold every worker shares.
Building it, concept by concept
Create an HTML file called spectralAnalysis.html with a button and a text area, and load the DCP client, the same page shape as the basic tutorial:
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8" />
<script src="https://scheduler.distributed.computer/dcp-client/dcp-client.js"></script>
</head>
<body>
<br />
<button id="deploy-btn" onclick="main()">Deploy Job</button>
<br /><br />
<textarea id="jobConsole" cols="120" rows="30"></textarea>
</body>
</html>
Everything below goes inside a <script> tag, in main().
Fetching the compiled module
Fetched once as text (not imported/executed) – it needs to travel to the worker as a plain string argument, not run on the page itself:
async function getFftwModuleSource() {
return fetch('./fftw-module.js').then((r) => r.text());
}
Input set
One element per sensor, same as the Node.js version of this tutorial:
function makeSignal(amplitude, bin, n = 16) {
return Array.from(
{ length: n },
(_, i) => amplitude * Math.sin((2 * Math.PI * bin * i) / n),
);
}
const inputSet = [
{ sensor: 'bearing-1', samples: makeSignal(1.0, 3) },
{ sensor: 'bearing-2', samples: makeSignal(3.0, 5) },
{ sensor: 'gearbox-1', samples: makeSignal(0.2, 1) },
];
Static arguments
The compiled module source, the sample rate, and the anomaly threshold are all shared facts, not per-sensor data – every worker needs the identical module and the identical thresholds. All three travel as static arguments, the third parameter to compute.for:
const SAMPLE_RATE_HZ = 1000;
const ANOMALY_MAGNITUDE_THRESHOLD = 15;
const fftwModuleSource = await getFftwModuleSource();
Work function
Without a local require() to bundle a wrapper file automatically, the worker rebuilds the module from the fetched source and talks to it directly via ccall/cwrap – the same calls a Node-side wrapper module would make on your behalf, just written out in the work function itself:
async function workFunction(
unit,
moduleSourceArg,
sampleRateHzArg,
anomalyThresholdArg,
) {
progress();
// Eval the fetched source as a CommonJS module body, recovering the
// FftwModule factory function it exports.
const moduleShim = { exports: {} };
new Function('module', 'exports', moduleSourceArg)(
moduleShim,
moduleShim.exports,
);
const FftwModule = moduleShim.exports;
// SINGLE_FILE=1 means the wasm bytes are already embedded in
// moduleSourceArg - no instantiateWasm override needed, no separate
// fetch of any kind.
const Module = await FftwModule();
const { sensor, samples } = unit;
const n = samples.length;
const nc = Math.floor(n / 2) + 1; // FFTW's r2c output size: the non-redundant half of the spectrum
const reInP = Module._malloc(n * 8);
Module.HEAPF64.set(samples, reInP / 8);
const reOutP = Module._malloc(nc * 8);
const imOutP = Module._malloc(nc * 8);
let re, im;
try {
const fn = Module.cwrap('fftw_r2c_1d_wasm', 'number', [
'number',
'number',
'number',
'number',
]);
const rc = fn(reInP, reOutP, imOutP, n);
if (rc !== 0) throw new Error(`fftw_r2c_1d_wasm failed, code ${rc}`);
re = Array.from(Module.HEAPF64.subarray(reOutP / 8, reOutP / 8 + nc));
im = Array.from(Module.HEAPF64.subarray(imOutP / 8, imOutP / 8 + nc));
} finally {
[reInP, reOutP, imOutP].forEach((p) => Module._free(p));
}
// Bin 0 is the DC component (the signal's average level) - skip it,
// only oscillating content is interesting here.
let peakBin = 0;
let peakMagnitude = 0;
for (let bin = 1; bin < re.length; bin++) {
const magnitude = Math.sqrt(re[bin] * re[bin] + im[bin] * im[bin]);
if (magnitude > peakMagnitude) {
peakMagnitude = magnitude;
peakBin = bin;
}
}
const dominantFrequencyHz = (peakBin * sampleRateHzArg) / n;
const isAnomaly = peakMagnitude >= anomalyThresholdArg;
return { sensor, dominantFrequencyHz, peakMagnitude, isAnomaly };
}
Job creation
The fetched module source, sample rate, and threshold all go in as static arguments – no job.requires() call anywhere in this version:
const job = compute.for(inputSet, workFunction, [
fftwModuleSource,
SAMPLE_RATE_HZ,
ANOMALY_MAGNITUDE_THRESHOLD,
]);
Job configuration
job.public = {
name: 'sensorSpectralAnalysis',
description: 'Per-sensor dominant-frequency analysis computed with FFTW-wasm',
link: 'https://distributive.network',
};
job.computeGroups = [{ joinKey: 'public' }];
Events
job.on('readystatechange', (ev) => log(`Ready state: ${ev}`));
job.on('accepted', () => log(` Job id: ${job.id}\n Awaiting results...`));
job.on('result', (ev) => log(` ${ev.result.sensor} done`));
job.on('error', (error) =>
log(` Job error: ${JSON.stringify(error, null, 2)}`),
);
job.on('nofunds', (ev) => log(`${JSON.stringify(ev, null, 2)}`));
Execution and result post-processing
const results = await job.exec();
for (const r of results) {
log(
`${r.sensor.padEnd(10)} dominant=${r.dominantFrequencyHz}Hz magnitude=${r.peakMagnitude} ${r.isAnomaly ? '** ANOMALY **' : ''}`,
);
}
const anomalies = results.filter((r) => r.isAnomaly);
log(
`\n${anomalies.length} sensor(s) flagged for review: ${anomalies.map((r) => r.sensor).join(', ') || 'none'}`,
);
Full listing
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8" />
<script src="https://scheduler.distributed.computer/dcp-client/dcp-client.js"></script>
<script>
async function getFftwModuleSource() {
return fetch('./fftw-module.js').then((r) => r.text());
}
function makeSignal(amplitude, bin, n = 16) {
return Array.from(
{ length: n },
(_, i) => amplitude * Math.sin((2 * Math.PI * bin * i) / n),
);
}
async function main() {
const compute = dcp.compute;
/* LOGGING */
const consoleEl = document.querySelector('#jobConsole');
const log = (msg) => {
consoleEl.value += msg + '\n';
consoleEl.scrollTop = consoleEl.scrollHeight;
};
/* INPUT SET */
const inputSet = [
{ sensor: 'bearing-1', samples: makeSignal(1.0, 3) },
{ sensor: 'bearing-2', samples: makeSignal(3.0, 5) },
{ sensor: 'gearbox-1', samples: makeSignal(0.2, 1) },
];
/* STATIC ARGUMENTS -- shared by every worker, not part of the input set */
const SAMPLE_RATE_HZ = 1000;
const ANOMALY_MAGNITUDE_THRESHOLD = 15;
const fftwModuleSource = await getFftwModuleSource();
/* WORK FUNCTION */
async function workFunction(
unit,
moduleSourceArg,
sampleRateHzArg,
anomalyThresholdArg,
) {
progress();
const moduleShim = { exports: {} };
new Function('module', 'exports', moduleSourceArg)(
moduleShim,
moduleShim.exports,
);
const FftwModule = moduleShim.exports;
const Module = await FftwModule();
const { sensor, samples } = unit;
const n = samples.length;
const nc = Math.floor(n / 2) + 1;
const reInP = Module._malloc(n * 8);
Module.HEAPF64.set(samples, reInP / 8);
const reOutP = Module._malloc(nc * 8);
const imOutP = Module._malloc(nc * 8);
let re, im;
try {
const fn = Module.cwrap('fftw_r2c_1d_wasm', 'number', [
'number',
'number',
'number',
'number',
]);
const rc = fn(reInP, reOutP, imOutP, n);
if (rc !== 0)
throw new Error(`fftw_r2c_1d_wasm failed, code ${rc}`);
re = Array.from(
Module.HEAPF64.subarray(reOutP / 8, reOutP / 8 + nc),
);
im = Array.from(
Module.HEAPF64.subarray(imOutP / 8, imOutP / 8 + nc),
);
} finally {
[reInP, reOutP, imOutP].forEach((p) => Module._free(p));
}
let peakBin = 0;
let peakMagnitude = 0;
for (let bin = 1; bin < re.length; bin++) {
const magnitude = Math.sqrt(re[bin] * re[bin] + im[bin] * im[bin]);
if (magnitude > peakMagnitude) {
peakMagnitude = magnitude;
peakBin = bin;
}
}
const dominantFrequencyHz = (peakBin * sampleRateHzArg) / n;
const isAnomaly = peakMagnitude >= anomalyThresholdArg;
return { sensor, dominantFrequencyHz, peakMagnitude, isAnomaly };
}
/* JOB CREATION */
const job = compute.for(inputSet, workFunction, [
fftwModuleSource,
SAMPLE_RATE_HZ,
ANOMALY_MAGNITUDE_THRESHOLD,
]);
/* JOB CONFIGURATION */
job.public = {
name: 'sensorSpectralAnalysis',
description:
'Per-sensor dominant-frequency analysis computed with FFTW-wasm',
link: 'https://distributive.network',
};
job.computeGroups = [{ joinKey: 'public' }];
/* EVENTS */
job.on('readystatechange', (ev) => log(`Ready state: ${ev}`));
job.on('accepted', () =>
log(` Job id: ${job.id}\n Awaiting results...`),
);
job.on('result', (ev) => log(` ${ev.result.sensor} done`));
job.on('error', (error) =>
log(` Job error: ${JSON.stringify(error, null, 2)}`),
);
job.on('nofunds', (ev) => log(`${JSON.stringify(ev, null, 2)}`));
/* EXECUTION */
const results = await job.exec();
/* RESULT POST-PROCESSING */
for (const r of results) {
log(
`${r.sensor.padEnd(10)} dominant=${r.dominantFrequencyHz}Hz magnitude=${r.peakMagnitude} ${r.isAnomaly ? '** ANOMALY **' : ''}`,
);
}
const anomalies = results.filter((r) => r.isAnomaly);
log(
`\n${anomalies.length} sensor(s) flagged for review: ${anomalies.map((r) => r.sensor).join(', ') || 'none'}`,
);
}
</script>
</head>
<body>
<br />
<button id="deploy-btn" onclick="main()">Deploy Job</button>
<br /><br />
<textarea id="jobConsole" cols="120" rows="30"></textarea>
</body>
</html>
Running it
1. With a web browser, open spectralAnalysis.html.
2. Click Deploy Job.
3. Example text area output:
Ready state: exec
...
Job id: ...
Awaiting results...
bearing-1 done
bearing-2 done
gearbox-1 done
bearing-1 dominant=187.5Hz magnitude=8
bearing-2 dominant=312.5Hz magnitude=24 ** ANOMALY **
gearbox-1 dominant=62.5Hz magnitude=1.6
1 sensor(s) flagged for review: bearing-2
Going further
Bigger, real data: swap
makeSignal’s generated arrays for real telemetry – an FFT this size doesn’t care whether its 16 samples came from a formula or a sensor.A friendlier wrapper, inlined: the
ccall/cwrapblock in the work function above is exactly what a Node-sidefftw-wasm.jswrapper (see the Node.js version of this tutorial) hides behind a plainfftw.r2c1d(samples)call. Nothing stops you from fetching and flattening a wrapper file the same way – concatenate its source with the module’s, in the style Building WASM modules for DCP jobs describes – if the raw pointer bookkeeping here is more than you want repeated across several work functions.More static arguments: the same pattern from
SAMPLE_RATE_HZ/ANOMALY_MAGNITUDE_THRESHOLDextends to anything every worker needs identically – a per-machine baseline spectrum to compare against, a set of frequency bands to ignore, whatever your real anomaly rule needs.