Skip to content

Commit 29415a8

Browse files
kristoferbaxterkristoferbaxter
authored
Increase concurrency (#77)
* Fixes in place, updated API * Slightly cleaner implementation * Cleanup JSDoc
1 parent b11213c commit 29415a8

File tree

17 files changed

+688
-392
lines changed

17 files changed

+688
-392
lines changed

‎src/api.ts

Lines changed: 10 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -16,45 +16,16 @@
1616

1717
import Project from './validation/Project';
1818
import Config from './validation/Config';
19-
import { Context, SizeMap, FileModifier } from './validation/Condition';
20-
import compress from './compress';
19+
import { Context, FileModifier, SizeMap } from './validation/Condition';
20+
import compress, { CompressionItem, findItemsToCompress } from './compress';
21+
import { Report } from './log/report';
22+
export { Report } from './log/report';
2123

22-
export async function* report(projectPath: string, fileModifier: FileModifier): AsyncGenerator<[SizeMap, SizeMap]> {
23-
const conditions = [Project, Config];
24-
let context: Context = {
25-
projectPath,
26-
packagePath: '',
27-
packageContent: '',
28-
silent: true,
29-
originalPaths: new Map(),
30-
// Stores the result of compression <path, [...results]>
31-
compressed: new Map(),
32-
// Stores the basis of comparison.
33-
comparison: new Map(),
34-
fileModifier,
35-
fileContents: new Map(),
36-
};
37-
38-
for (const condition of conditions) {
39-
const message = await condition(context)();
40-
if (message !== null) {
41-
throw message;
42-
}
43-
}
44-
45-
const compressResults = compress(context, false);
46-
let nextResult = await compressResults.next();
47-
while (!nextResult.done) {
48-
yield [context.compressed, context.comparison];
49-
nextResult = await compressResults.next();
50-
}
51-
return [context.compressed, context.comparison];
52-
}
53-
54-
export async function* serialReport(
24+
export async function report(
5525
projectPath: string,
5626
fileModifier: FileModifier,
57-
): AsyncGenerator<[string, number, number, number]> {
27+
report?: typeof Report,
28+
): Promise<SizeMap> {
5829
const conditions = [Project, Config];
5930
let context: Context = {
6031
projectPath,
@@ -77,18 +48,7 @@ export async function* serialReport(
7748
}
7849
}
7950

80-
const compressResults = compress(context, false);
81-
const paths: Set<string> = new Set(Array.from(context.compressed.keys()));
82-
let next = await compressResults.next();
83-
while (!next.done) {
84-
for (const filePath of paths) {
85-
const sizes = context.compressed.get(filePath);
86-
if (sizes !== undefined && sizes?.[0][0] && sizes?.[1][0] && sizes?.[2][0]) {
87-
yield [filePath, sizes?.[0][0], sizes?.[1][0], sizes?.[2][0]];
88-
paths.delete(filePath);
89-
}
90-
}
91-
next = await compressResults.next();
92-
}
93-
return;
51+
const toCompress: Array<CompressionItem> = await findItemsToCompress(context, true);
52+
await compress(context, toCompress, report || null);
53+
return context.compressed;
9454
}

‎src/compress.ts

Lines changed: 36 additions & 101 deletions
Original file line numberDiff line numberDiff line change
@@ -14,97 +14,20 @@
1414
* limitations under the License.
1515
*/
1616

17-
import { Context, Compression, OrderedCompressionValues, maxSize } from './validation/Condition';
1817
import { cpus } from 'os';
19-
import { constants as brotliConstants, brotliCompress, gzip, ZlibOptions } from 'zlib';
18+
import { Context, Compression, OrderedCompressionValues, maxSize } from './validation/Condition';
2019
import { readFile } from './helpers/fs';
21-
import { LogError } from './log/helpers/error';
2220
import { Report } from './log/report';
23-
import { TTYReport } from './log/tty-report';
24-
import { stdout } from 'process';
21+
import { compressor } from './compressor';
2522

2623
const COMPRESSION_CONCURRENCY = cpus().length;
27-
const BROTLI_OPTIONS = {
28-
params: {
29-
[brotliConstants.BROTLI_PARAM_MODE]: brotliConstants.BROTLI_DEFAULT_MODE,
30-
[brotliConstants.BROTLI_PARAM_QUALITY]: brotliConstants.BROTLI_MAX_QUALITY,
31-
[brotliConstants.BROTLI_PARAM_SIZE_HINT]: 0,
32-
},
33-
};
34-
const GZIP_OPTIONS: ZlibOptions = {
35-
level: 9,
36-
};
3724

38-
interface CompressionItem {
25+
export interface CompressionItem {
3926
path: string;
4027
compression: Compression;
4128
maxSize: maxSize;
4229
}
4330

44-
/**
45-
* Use the given configuration and actual size to report item filesize.
46-
* @param report Optional reporter to update with this value
47-
* @param item Configuration for an Item
48-
* @param error Error from compressing an Item
49-
* @param size actual size for this comparison
50-
*/
51-
function store(
52-
report: Report | null,
53-
context: Context,
54-
item: CompressionItem,
55-
error: Error | null,
56-
size: number,
57-
): boolean {
58-
if (error !== null) {
59-
LogError(`Could not compress '${item.path}' with '${item.compression}'.`);
60-
return false;
61-
}
62-
63-
// Store the size of the item in the compression map.
64-
const sizeMap = context.compressed.get(item.path);
65-
if (sizeMap === undefined) {
66-
LogError(`Could not find item '${item.path}' with '${item.compression}' in compression map.`);
67-
return false;
68-
}
69-
sizeMap[OrderedCompressionValues.indexOf(item.compression)][0] = size;
70-
71-
report?.update(context);
72-
if (item.maxSize === undefined) {
73-
return true;
74-
}
75-
return size < item.maxSize;
76-
}
77-
78-
/**
79-
* Compress an Item and report status to the console.
80-
* @param item Configuration for an Item.
81-
*/
82-
async function compressor(report: Report | null, context: Context, item: CompressionItem): Promise<boolean> {
83-
const contents = context.fileContents.get(item.path);
84-
if (contents) {
85-
const buffer = Buffer.from(contents, 'utf8');
86-
87-
switch (item.compression) {
88-
case 'brotli':
89-
return new Promise(resolve =>
90-
brotliCompress(buffer, BROTLI_OPTIONS, (error: Error | null, result: Buffer) =>
91-
resolve(store(report, context, item, error, result.byteLength)),
92-
),
93-
);
94-
case 'gzip':
95-
return new Promise(resolve =>
96-
gzip(buffer, GZIP_OPTIONS, (error: Error | null, result: Buffer) =>
97-
resolve(store(report, context, item, error, result.byteLength)),
98-
),
99-
);
100-
default:
101-
return store(report, context, item, null, buffer.byteLength);
102-
}
103-
}
104-
105-
return false;
106-
}
107-
10831
/**
10932
* Store the original content so it isn't retrieved from FileSystem for each compression.
11033
* @param context
@@ -125,15 +48,15 @@ async function storeOriginalFileContents(context: Context, path: string): Promis
12548
* @param context
12649
* @param findDefaultSize
12750
*/
128-
async function findItemsToCompress(context: Context, findDefaultSize: boolean): Promise<Array<CompressionItem>> {
51+
export async function findItemsToCompress(context: Context, findDefaultSize: boolean): Promise<Array<CompressionItem>> {
12952
const toCompress: Array<CompressionItem> = [];
13053
for (const [path, sizeMapValue] of context.compressed) {
13154
for (let iterator: number = 0; iterator < OrderedCompressionValues.length; iterator++) {
13255
const compression: Compression = OrderedCompressionValues[iterator] as Compression;
13356
const [size, maxSize] = sizeMapValue[iterator];
13457
await storeOriginalFileContents(context, path);
13558
if (findDefaultSize && compression === 'none') {
136-
await compressor(null, context, { path, compression, maxSize });
59+
await compressor(context, null, { path, compression, maxSize });
13760
}
13861
if (size !== undefined) {
13962
toCompress.push({
@@ -152,28 +75,40 @@ async function findItemsToCompress(context: Context, findDefaultSize: boolean):
15275
* Given a context, compress all Items within splitting work eagly per cpu core to achieve some concurrency.
15376
* @param context Finalized Valid Context from Configuration
15477
*/
155-
export default async function* compress(context: Context, outputReport: boolean): AsyncGenerator<boolean, boolean> {
156-
const toCompress: Array<CompressionItem> = await findItemsToCompress(context, true);
157-
const report: Report | null = outputReport
158-
? null
159-
: stdout.isTTY && toCompress.length < 30
160-
? new TTYReport(context)
161-
: new Report(context);
162-
let success: boolean = true;
78+
export default async function compress(
79+
context: Context,
80+
toCompress: Array<CompressionItem>,
81+
report: typeof Report | null,
82+
): Promise<boolean> {
83+
if (toCompress.length === 0) {
84+
return true;
85+
}
16386

164-
for (let iterator: number = 0; iterator < toCompress.length; iterator += COMPRESSION_CONCURRENCY) {
165-
if (iterator === 0) {
166-
report?.update(context);
167-
}
168-
let itemsSuccessful = await Promise.all(
169-
toCompress.slice(iterator, iterator + COMPRESSION_CONCURRENCY).map(item => compressor(report, context, item)),
170-
);
171-
if (itemsSuccessful.includes(false)) {
172-
success = false;
87+
const returnable: Array<Promise<boolean>> = [];
88+
const executing: Array<Promise<boolean>> = [];
89+
const reportInstance: Report | null = report ? new report(context) : null;
90+
let success = true;
91+
for (const item of toCompress) {
92+
const promise: Promise<boolean> = Promise.resolve(item).then((item) => compressor(context, reportInstance, item));
93+
returnable.push(promise);
94+
95+
if (COMPRESSION_CONCURRENCY <= toCompress.length) {
96+
const execute: any = promise.then((successful) => {
97+
if (!successful) {
98+
success = successful;
99+
}
100+
executing.splice(executing.indexOf(execute), 1);
101+
});
102+
executing.push(execute);
103+
if (executing.length >= COMPRESSION_CONCURRENCY) {
104+
await Promise.race(executing);
105+
}
173106
}
174-
yield success;
107+
}
108+
if ((await Promise.all(returnable)).includes(false)) {
109+
success = false;
175110
}
176111

177-
report?.end();
112+
reportInstance?.end();
178113
return success;
179114
}

‎src/compressor.ts

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
/**
2+
* Copyright 2020 The AMP HTML Authors. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS-IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
import { constants as brotliConstants, brotliCompress, gzip } from 'zlib';
18+
import { Report } from './log/report';
19+
import { Context, Compression, maxSize, OrderedCompressionValues } from './validation/Condition';
20+
import { LogError } from './log/helpers/error';
21+
22+
type CompressionMethod = (buffer: Buffer, options: {}, callback: (error: Error | null, result: Buffer) => void) => void;
23+
24+
interface CompressionItem {
25+
path: string;
26+
compression: Compression;
27+
maxSize: maxSize;
28+
}
29+
30+
const SUPPORTED_COMPRESSION: Map<string, [CompressionMethod, Object]> = new Map([
31+
[
32+
'brotli',
33+
[
34+
brotliCompress,
35+
{
36+
params: {
37+
[brotliConstants.BROTLI_PARAM_MODE]: brotliConstants.BROTLI_DEFAULT_MODE,
38+
[brotliConstants.BROTLI_PARAM_QUALITY]: brotliConstants.BROTLI_MAX_QUALITY,
39+
[brotliConstants.BROTLI_PARAM_SIZE_HINT]: 0,
40+
},
41+
},
42+
],
43+
],
44+
[
45+
'gzip',
46+
[
47+
gzip,
48+
{
49+
level: 9,
50+
},
51+
],
52+
],
53+
]);
54+
55+
/**
56+
* Use the given configuration and actual size to report item filesize.
57+
* @param report Optional reporter to update with this value
58+
* @param item Configuration for an Item
59+
* @param error Error from compressing an Item
60+
* @param size actual size for this comparison
61+
*/
62+
function store(
63+
report: Report | null,
64+
context: Context,
65+
item: CompressionItem,
66+
error: Error | null,
67+
size: number,
68+
): boolean {
69+
if (error !== null) {
70+
LogError(`Could not compress '${item.path}' with '${item.compression}'.`);
71+
return false;
72+
}
73+
74+
// Store the size of the item in the compression map.
75+
const sizeMap = context.compressed.get(item.path);
76+
if (sizeMap === undefined) {
77+
LogError(`Could not find item '${item.path}' with '${item.compression}' in compression map.`);
78+
return false;
79+
}
80+
sizeMap[OrderedCompressionValues.indexOf(item.compression)][0] = size;
81+
82+
report?.update(context);
83+
if (item.maxSize === undefined) {
84+
return true;
85+
}
86+
return size < item.maxSize;
87+
}
88+
89+
export function compressor(context: Context, report: Report | null, item: CompressionItem): Promise<boolean> {
90+
const contents = context.fileContents.get(item.path);
91+
if (contents) {
92+
return new Promise((resolve) => {
93+
const buffer = Buffer.from(contents, 'utf8');
94+
const compression = SUPPORTED_COMPRESSION.get(item.compression);
95+
if (compression) {
96+
compression[0](buffer, compression[1], (error: Error | null, result: Buffer) =>
97+
resolve(store(report, context, item, error, result.byteLength)),
98+
);
99+
} else {
100+
resolve(store(report, context, item, null, buffer.byteLength));
101+
}
102+
});
103+
}
104+
105+
return Promise.resolve(false);
106+
}

0 commit comments

Comments
 (0)