diff --git a/imports/startup/server/calcUnionAsync.js b/imports/startup/server/calcUnionAsync.js index ff7af18..1cb1b24 100644 --- a/imports/startup/server/calcUnionAsync.js +++ b/imports/startup/server/calcUnionAsync.js @@ -1,54 +1,28 @@ -import tcircle from '@turf/circle'; -import tunion from '@turf/union'; -import ttrunc from '@turf/truncate'; +import path from 'path'; +import { Worker } from 'worker_threads'; -const truncOptions = { precision: 6, coordinates: 2 }; -// How many polygons to process before yielding to the event loop once. -const YIELD_EVERY = 100; - -const yieldToEventLoop = () => new Promise((resolve) => { setImmediate(resolve); }); +const WORKER_PATH = path.resolve(process.cwd(), 'assets/app/workers/unionWorker.js'); /** - * Circle-union of subscriptions, equivalent to map-common-utils' calcUnion(typeCircle=true), - * but split into a `for` loop that yields to the event loop every YIELD_EVERY polygons. - * The vendored calcUnion runs the same turf.union chain fully synchronously, which for - * thousands of subscriptions blocks Meteor's single-threaded DDP/HTTP handling for minutes - * (see plan-modernizacion notes on the subsUnion hang). This keeps the exact same geometry - * output while letting other requests interleave between chunks. + * Circle-union of already-validated, already-decorated subscriptions, run in a + * worker thread so the main event loop (DDP/HTTP) stays free regardless of + * how long any single turf.union call takes. subs must be an array of plain + * { location: { lat, lon }, distance } objects. */ -const calcUnionAsync = async (subs, decorated) => { - const unionGroup = []; - for (let i = 0; i < subs.length; i += 1) { - const osub = subs[i]; - try { - if (osub.location && osub.location.lat && osub.location.lon && osub.distance) { - const dsub = decorated(osub); - unionGroup.push(tcircle( - [dsub.location.lon, dsub.location.lat], - dsub.distance, - { units: 'kilometers', steps: 144 } - )); - } else { - // Expected for old telegram subscriptions predating the distance field - // (e.g. distance: null) — logged, not an error, safe to skip. - console.info(`Wrong element to do union ${JSON.stringify(osub)}`); - } - } catch (e) { - console.error(e, `Wrong element trying to make union ${JSON.stringify(osub)}`); +const calcUnionAsync = subs => new Promise((resolve, reject) => { + const worker = new Worker(WORKER_PATH, { workerData: { subs } }); + worker.once('message', (msg) => { + worker.terminate(); + if (msg.error) { + reject(new Error(msg.error)); + } else { + resolve(msg.union); } - // Deliberate: yield to the event loop periodically instead of blocking it. - // eslint-disable-next-line no-await-in-loop - if (i % YIELD_EVERY === 0) await yieldToEventLoop(); - } - - let unionTemp = unionGroup.length > 0 ? ttrunc(unionGroup[0], truncOptions) : null; - for (let i = 1; i < unionGroup.length; i += 1) { - unionTemp = ttrunc(tunion(unionTemp, unionGroup[i]), truncOptions); - // Deliberate: yield to the event loop periodically instead of blocking it. - // eslint-disable-next-line no-await-in-loop - if (i % YIELD_EVERY === 0) await yieldToEventLoop(); - } - return unionTemp; -}; + }); + worker.once('error', (err) => { + worker.terminate(); + reject(err); + }); +}); export default calcUnionAsync; diff --git a/imports/startup/server/subsUnion.js b/imports/startup/server/subsUnion.js index 087a1e5..9fe55ec 100644 --- a/imports/startup/server/subsUnion.js +++ b/imports/startup/server/subsUnion.js @@ -37,7 +37,20 @@ Meteor.startup(async () => { const process = async (isPublic) => { const subscribers = await Subscriptions.find().fetchAsync(); - const union = await calcUnionAsync(subscribers, isPublic ? addNoisy : noNoisy); + const decorate = isPublic ? addNoisy : noNoisy; + const validSubs = subscribers.filter((osub) => { + const valid = osub.location && osub.location.lat && osub.location.lon && osub.distance; + if (!valid) console.info(`Wrong element to do union ${JSON.stringify(osub)}`); + return valid; + }); + const decorated = validSubs.map(decorate); + + let union = null; + try { + union = await calcUnionAsync(decorated); + } catch (e) { + console.error('subsUnion worker failed', e); + } const bounds = union === null ? null : L.geoJSON(union).getBounds(); const publicl = isPublic ? 'public' : 'private'; diff --git a/private/workers/unionWorker.js b/private/workers/unionWorker.js new file mode 100644 index 0000000..9f6bce2 --- /dev/null +++ b/private/workers/unionWorker.js @@ -0,0 +1,40 @@ +// Plain CommonJS worker_threads entry point, NOT part of the Meteor imports +// graph (lives under private/ so `meteor build` copies it verbatim into +// programs/server/assets/app/workers/unionWorker.js instead of compiling it). +// +// Runs the CPU-heavy part of the subscriptions geo-union (circle generation + +// sequential turf.union chain) off the main thread. A single tunion() call +// against an already-complex merged polygon can itself take seconds once +// there are thousands of subscriptions — yielding between iterations on the +// main thread doesn't help when one iteration alone blocks that long, so this +// needs a real worker thread, not just cooperative scheduling. +// +// workerData.subs must already be validated (location.lat/lon/distance +// present) and decorated (addNoisy/noNoisy already applied) by the caller. +const { parentPort, workerData } = require('worker_threads'); +const tcircle = require('@turf/circle').default; +const tunion = require('@turf/union').default; +const ttrunc = require('@turf/truncate').default; + +const truncOptions = { precision: 6, coordinates: 2 }; + +const run = () => { + const { subs } = workerData; + const unionGroup = subs.map(s => tcircle( + [s.location.lon, s.location.lat], + s.distance, + { units: 'kilometers', steps: 144 } + )); + + let unionTemp = unionGroup.length > 0 ? ttrunc(unionGroup[0], truncOptions) : null; + for (let i = 1; i < unionGroup.length; i += 1) { + unionTemp = ttrunc(tunion(unionTemp, unionGroup[i]), truncOptions); + } + return unionTemp; +}; + +try { + parentPort.postMessage({ union: run() }); +} catch (e) { + parentPort.postMessage({ error: e.message }); +}