external-repos/squoosh
1
0
Fork 1
mirror of https://github.com/GoogleChromeLabs/squoosh.git synced 2025-11-12 16:57:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/dev' into banding-fix

Browse Source
This commit is contained in:
Surma
2021-01-05 15:30:07 +00:00
parent 36293d756b e7e205c326
commit dba8f097d3
2 changed files with 24 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
cli/src/index.js
View File

@@ -183,9 +183,13 @@ function progressTracker(results) {
return tracker; return tracker;
} }
async function checkInputFilesValid(files) { async function getInputFiles(paths) {
const validFiles = []; const validFiles = [];
for (const path of paths) {
const files = (await fsp.lstat(path)).isDirectory()
? (await fsp.readdir(path)).map(file => join(path, file))
: [path];
for (const file of files) { for (const file of files) {
try { try {
await fsp.stat(file); await fsp.stat(file);
@@ -202,12 +206,13 @@ async function checkInputFilesValid(files) {
validFiles.push(file); validFiles.push(file);
} }
}
return validFiles; return validFiles;
} }
async function processFiles(files) { async function processFiles(files) {
files = await checkInputFilesValid(files); files = await getInputFiles(files);
const parallelism = cpus().length; const parallelism = cpus().length;

18
cli/src/worker_pool.js
View File

@@ -24,7 +24,7 @@ function jobPromise(worker, msg) {
export default class WorkerPool { export default class WorkerPool {
constructor(numWorkers, workerFile) { constructor(numWorkers, workerFile) {
this.closing = false; this.numWorkers = numWorkers;
this.jobQueue = new TransformStream(); this.jobQueue = new TransformStream();
this.workerQueue = new TransformStream(); this.workerQueue = new TransformStream();
@@ -42,7 +42,6 @@ export default class WorkerPool {
while (true) { while (true) {
const { value, done } = await reader.read(); const { value, done } = await reader.read();
if (done) { if (done) {
this.workerQueue.writable.close();
await this._terminateAll(); await this._terminateAll();
return; return;
} }
@@ -50,12 +49,6 @@ export default class WorkerPool {
const worker = await this._nextWorker(); const worker = await this._nextWorker();
jobPromise(worker, msg).then((result) => { jobPromise(worker, msg).then((result) => {
resolve(result); resolve(result);
// If we are in the process of closing, `workerQueue` is
// already closed and we can’t requeue the worker.
if (this.closing) {
worker.terminate();
return;
}
const writer = this.workerQueue.writable.getWriter(); const writer = this.workerQueue.writable.getWriter();
writer.write(worker); writer.write(worker);
writer.releaseLock(); writer.releaseLock();
@@ -71,18 +64,15 @@ export default class WorkerPool {
} }
async _terminateAll() { async _terminateAll() {
while (true) { for (let n = 0; n < this.numWorkers; n++) {
const worker = await this._nextWorker(); const worker = await this._nextWorker();
if (!worker) {
return;
}
worker.terminate(); worker.terminate();
} }
this.workerQueue.writable.close();
} }
async join() { async join() {
this.closing = true; this.jobQueue.writable.getWriter().close();
this.jobQueue.writable.close();
await this.done; await this.done;
} }