Remove FileAllocator#groupPathsByWorker and always use WorkerQueue. Optimize WorkerQueue.
MODIFY
packages/@romejs/cli-reporter/Reporter.ts
packages/@romejs/cli-reporter/Reporter.ts
@@ -853,6 +853,10 @@ export default class Reporter {
}
inspect(value: unknown) {
+ if (!this.isEnabled()) {
+ return;
+ }
+
let formatted = value;
if (typeof formatted !== 'number' && typeof formatted !== 'string') {
@@ -1064,6 +1068,10 @@ export default class Reporter {
args: Array<unknown>,
opts: LogCategoryOptions,
) {
+ if (!this.isEnabled(opts.stderr)) {
+ return;
+ }
+
const inner = markupTag(opts.markupTag, rawInner);
const unicodeSuffix = opts.unicodeSuffix === undefined
? ''

MODIFY
packages/@romejs/core/master/WorkerQueue.ts
packages/@romejs/core/master/WorkerQueue.ts
@@ -22,45 +22,31 @@ type Callback<M> = (
) => undefined | Promise<void>;
export default class WorkerQueue<M> {
- constructor(master: Master) {
+ constructor(master: Master, maxPer: number = 2) {
this.master = master;
- this.queue = [];
this.callbacks = [];
- this.promises = [];
+ this.runningWorkers = [];
this.workers = new Map();
this.open = true;
+ this.maxPer = maxPer;
}
master: Master;
- queue: Queue<M>;
- promises: Array<Promise<void>>;
+ maxPer: number;
+ runningWorkers: Array<Promise<void>>;
callbacks: Array<Callback<M>>;
workers: Map<WorkerContainer, WorkerQueueItem<M>>;
open: boolean;
- pushQueue(path: AbsoluteFilePath, metadata: M) {
+ async pushQueue(path: AbsoluteFilePath, metadata: M) {
if (!this.open) {
throw new Error('WorkerQueue has already closed');
}
- this.queue.push([path, metadata]);
- }
-
- addCallback(callback: Callback<M>) {
- this.callbacks.push(callback);
- }
-
- // Take all the root queue items, assign them to a worker, and start the worker queue if it's not running
- async updateWorkerQueues() {
- const {queue} = this;
-
- while (queue.length > 0) {
- const item = queue.shift();
- if (item === undefined) {
- throw new Error('Already validated queue.length above');
+ if (this.callbacks.length === 0) {
+ throw new Error('No callbacks attached to queue');
}
- const path = item[0];
const workerContainer = await this.master.fileAllocator.getOrAssignOwner(
path,
);
@@ -74,67 +60,61 @@ export default class WorkerQueue<M> {
};
this.workers.set(workerContainer, worker);
}
- worker.queue.push(item);
+ worker.queue.push([path, metadata]);
// Start this worker if it isn't already
if (worker.running === false) {
const promise = this.processWorker(worker);
// Add a `catch` so that we aren't considered an unhandled promise if it rejects before a handler is attached
promise.catch(() => {});
- this.promises.push(promise);
+ this.runningWorkers.push(promise);
}
}
+
+ addCallback(callback: Callback<M>) {
+ this.callbacks.push(callback);
}
- async processWorker(item: WorkerQueueItem<M>) {
- item.running = true;
+ async processWorker(worker: WorkerQueueItem<M>) {
+ worker.running = true;
- const {queue} = item;
+ const {queue} = worker;
- while (queue.length > 0) {
+ const next = async () => {
const item = queue.shift();
if (item === undefined) {
- throw new Error('Already validated queue.length above');
+ // Exhausted queue
+ return;
}
const [filename, metadata] = item;
for (const callback of this.callbacks) {
await callback(filename, metadata);
}
- await this.updateWorkerQueues();
+ await next();
+ };
+
+ while (queue.length > 0) {
+ // "threads"
+ const threads = [];
+ for (let i = 0; i < this.maxPer; i++) {
+ threads.push(next());
+ }
+ await Promise.all(threads);
}
- item.running = false;
+ worker.running = false;
}
async spin() {
- const {queue} = this;
-
- // Create the initial queue
- await this.updateWorkerQueues();
-
- // Keep consuming all the promises until we're exhausted
- while (this.promises.length > 0) {
- const {promises} = this;
- this.promises = [];
- await Promise.all(promises);
+ while ( // Keep consuming all the promises until we're exhausted
+ this.runningWorkers.length > 0) {
+ const {runningWorkers} = this;
+ this.runningWorkers = [];
+ await Promise.all(runningWorkers);
}
// Ensure we never receive anymore queue items
this.open = false;
-
- // Ensure main queue has been drained
- if (queue.length > 0) {
- throw new Error('Expected no queue items to remain');
- }
-
- // Ensure worker queues have been drained
- for (const [worker, {queue}] of this.workers) {
- if (queue.length > 0) {
- throw new Error(
- `Expected no queue items to remain for worker ${worker.id}`,
- );
- }
- }
}
}

MODIFY
packages/@romejs/core/master/bundler/BundleRequest.ts
packages/@romejs/core/master/bundler/BundleRequest.ts
@@ -28,6 +28,7 @@ import crypto = require('crypto');
import {Dict} from '@romejs/typescript-helpers';
import {Reporter} from '@romejs/cli-reporter';
+import WorkerQueue from '../WorkerQueue';
export type BundleOptions = {
prefix?: string;
@@ -124,16 +125,21 @@ export default class BundleRequest {
});
compilingSpinner.setTotal(paths.length);
- const groupedPaths = await master.fileAllocator.groupPathsByWorker(paths);
- await Promise.all(groupedPaths.map(async (paths) => {
- for (const path of paths) {
+ const queue: WorkerQueue<void> = new WorkerQueue(master);
+
+ queue.addCallback(async (path) => {
const progressText = `<filelink target="${path.join()}" />`;
compilingSpinner.pushText(progressText);
await this.compileJS(path);
compilingSpinner.tick();
compilingSpinner.popText(progressText);
+ });
+
+ for (const path of paths) {
+ await queue.pushQueue(path);
}
- }));
+
+ await queue.spin();
compilingSpinner.end();
}

MODIFY
packages/@romejs/core/master/dependencies/DependencyGraph.ts
packages/@romejs/core/master/dependencies/DependencyGraph.ts
@@ -320,7 +320,7 @@ export default class DependencyGraph {
const subAncestry = [...ancestry, filename];
for (const path of node.getAbsoluteDependencies()) {
const dep = node.getDependencyInfoFromAbsolute(path).analyze;
- opts.workerQueue.pushQueue(path, {
+ await opts.workerQueue.pushQueue(path, {
all: dep.all,
async: dep.async,
type: dep.type,

MODIFY
packages/@romejs/core/master/fs/FileAllocator.ts
packages/@romejs/core/master/fs/FileAllocator.ts
@@ -9,7 +9,7 @@ import {Master} from '@romejs/core';
import {Stats} from './MemoryFileSystem';
import {WorkerContainer} from '../WorkerManager';
import Locker from '../../common/utils/Locker';
-import {AbsoluteFilePath, AbsoluteFilePathSet} from '@romejs/path';
+import {AbsoluteFilePath} from '@romejs/path';
import {Event} from '@romejs/events';
export default class FileAllocator {
@@ -88,26 +88,6 @@ export default class FileAllocator {
}
}
- async groupPathsByWorker(
- paths: AbsoluteFilePathSet | Array<AbsoluteFilePath>,
- ): Promise<Array<Array<AbsoluteFilePath>>> {
- const pathsByWorker: Map<number, Array<AbsoluteFilePath>> = new Map();
-
- // Populate our queues
- await Promise.all(Array.from(paths, async (path) => {
- const worker = await this.getOrAssignOwner(path);
-
- let queue = pathsByWorker.get(worker.id);
- if (queue === undefined) {
- queue = [];
- pathsByWorker.set(worker.id, queue);
- }
- queue.push(path);
- }));
-
- return Array.from(pathsByWorker.values());
- }
-
async evict(path: AbsoluteFilePath) {
// Find owner
const workerId = this.getOwnerId(path);

MODIFY
packages/@romejs/core/master/linter/Linter.ts
packages/@romejs/core/master/linter/Linter.ts
@@ -24,6 +24,7 @@ import {ReporterProgressOptions, ReporterProgress} from '@romejs/cli-reporter';
import DependencyNode from '../dependencies/DependencyNode';
import {areAnalyzeDependencyResultsEqual} from '@romejs/js-compiler';
import {markup} from '@romejs/string-markup';
+import WorkerQueue from '../WorkerQueue';
type LintWatchChanges = Array<{
filename: undefined | string;
@@ -155,15 +156,13 @@ class LintRunner {
}: LintRunOptions): Promise<{fixedCount: number}> {
let fixedCount = 0;
const {master} = this.request;
- const pathsByWorker = await master.fileAllocator.groupPathsByWorker(
- changedPaths,
- );
+
+ const queue: WorkerQueue<void> = new WorkerQueue(master);
const progress = this.events.createProgress({title: 'Linting'});
progress.setTotal(changedPaths.size);
- await Promise.all(pathsByWorker.map(async (paths) => {
- for (const path of paths) {
+ queue.addCallback(async (path) => {
const text = markup`<filelink target="${path.join()}" />`;
progress.pushText(text);
@@ -181,9 +180,13 @@ class LintRunner {
progress.popText(text);
progress.tick();
+ });
+
+ for (const path of changedPaths) {
+ await queue.pushQueue(path);
}
- }));
+ await queue.spin();
progress.end();
return {fixedCount};

Remove FileAllocator#groupPathsByWorker and always use WorkerQueue. Optimize WorkerQueue.