Implement a concurrency cap in loading GNATcoverage report

Author: eliericha

integration/vscode/ada/src/gnatcov.ts

@@ -3,6 +3,8 @@ import * as fs from 'fs';
 import * as path from 'path';
 import * as vscode from 'vscode';
 import { CancellationToken } from 'vscode-languageclient';
+import { parallelize, staggerProgress } from './helpers';
+import { cpus } from 'os';
 
 /**
  * TypeScript types to represent data from GNATcoverage XML reports
@@ -291,10 +293,16 @@ export async function addCoverageData(run: vscode.TestRun, covDir: string) {
         async (progress, token) => {
             const array = data.coverage_report.coverage_summary!.file;
             let done: number = 0;
+            let lastProgress = 0;
             const totalFiles = array.length;
+            progress.report({
+                message: `${done} / ${totalFiles} source files`,
+            });
             const fileCovs = (
-                await Promise.all(
-                    array.map(async (file) => {
+                await parallelize(
+                    array,
+                    Math.min(cpus().length, 8),
+                    async (file) => {
                         if (token.isCancellationRequested) {
                             throw new vscode.CancellationError();
                         }
@@ -333,13 +341,46 @@ export async function addCoverageData(run: vscode.TestRun, covDir: string) {
                             total,
                         });
 
-                        progress.report({
-                            message: `${++done} / ${totalFiles} source files`,
-                            increment: (100 * 1) / totalFiles,
-                        });
+                        /**
+                         * Do we need a lock to increment this counter given
+                         * that we are processing with threads?
+                         *
+                         * The answer is no.
+                         *
+                         * In JavaScript semantics each function runs to
+                         * completion uninterrupted. 'async' doesn't mean that
+                         * functions run concurrently. It means that something
+                         * will be executed later.
+                         *
+                         * When a function flow encounters 'await', it hands
+                         * off processing to another async operation.
+                         *
+                         * So at any one time, only one function is executing
+                         * and accessing the counter. It is safe to increment
+                         * without locking.
+                         */
+                        ++done;
+
+                        /**
+                         * Reporting progress too often can cause the UI to freeze.
+                         * Instead we use staggerProgress to report progress only
+                         * when the percentage has increased by 1%.
+                         */
+                        lastProgress = staggerProgress(
+                            done,
+                            totalFiles,
+                            lastProgress,
+                            (increment) => {
+                                progress.report({
+                                    message: `${done} / ${totalFiles} source files`,
+                                    increment: increment,
+                                });
+                            },
+                        );
 
                         return fileCov;
-                    }),
+                    },
+                    token,
                 )
             ).filter((v) => !!v);
             fileCovs.map((fileCov) => {

integration/vscode/ada/src/helpers.ts

@@ -520,3 +520,89 @@ export async function showErrorMessageWithOpenLogButton(
         mainOutputChannel.show();
     }
 }
+
+/**
+ *
+ * @param data - input data to process
+ * @param threads - number of async functions processing the data. Must be strictly positive 0.
+ * @param op - the function to process the data
+ * @returns the processed
+ */
+export async function parallelize<T, Result>(
+    data: T[],
+    threads: number,
+    op: (arg: T) => Result | Promise<Result>,
+    token?: CancellationToken,
+): Promise<Result[]> {
+    assert(threads > 0);
+    const len = data.length;
+    let start = 0;
+
+    /**
+     * Initialize an array with the expected length.
+     */
+    const result = new Array<Result>(len);
+    while (start < len) {
+        if (token?.isCancellationRequested) {
+            throw new CancellationError();
+        }
+
+        /**
+         * Take a chunk corresponding to the number of threads.
+         */
+        const chunk = data.slice(start, start + threads);
+
+        /**
+         * Process the chunk as async promises which may be evaluated concurrently.
+         */
+        const chunkRes = await Promise.all(chunk.map(op));
+
+        /**
+         * Since the array is initialized with a length, its content is already
+         * allocated with empty values. So we need a splice operation where we
+         * delete the existing empty values and replace them with the results.
+         */
+        result.splice(start, chunkRes.length, ...chunkRes);
+
+        start += threads;
+    }
+
+    return result;
+}
+
+/**
+ * When using {@link vscode.window.withProgress}, reporting progress too often
+ * can cause the UI to freeze. This function helps stagger progress reports
+ * such that progress is reported only when a minimum increment is reached,
+ * e.g. every 1%.
+ *
+ * Here's how the function should be used:
+ *
+ *    TODO
+ *
+ *
+ * @param done - the number of items processed since the beginning of the work
+ * @param total - the total number of items to be processed
+ * @param lastProgress - the last 'done' value at which progress was reported
+ * @param progress - a callback that is called when the minimum increment is reached
+ * @param minimumIncrement - the minimum increment to wait for before calling
+ * the 'progress' callback.
+ * @returns `lastProgress` if no progress was reported, and returns `done` when
+ * progress was reported.
+ */
+export function staggerProgress(
+    done: number,
+    total: number,
+    lastProgress: number,
+    progress: (increment: number) => void,
+    minimumIncrement: number = 1,
+): number {
+    const sinceLastReport = done - lastProgress;
+    const increment = Math.floor((sinceLastReport * 100) / total);
+    if (increment >= minimumIncrement) {
+        progress(increment);
+        return done;
+    } else {
+        return lastProgress;
+    }
+}

integration/vscode/ada/test/general/helpers.test.ts

@@ -1,6 +1,22 @@
 import assert from 'assert';
-import { envHasExec, findAdaMain, getSymbols, which } from '../../src/helpers';
-import { DocumentSymbol, SymbolKind, Uri, commands, workspace } from 'vscode';
+import {
+    envHasExec,
+    findAdaMain,
+    getSymbols,
+    parallelize,
+    staggerProgress,
+    which,
+} from '../../src/helpers';
+import {
+    CancellationError,
+    DocumentSymbol,
+    ProgressLocation,
+    SymbolKind,
+    Uri,
+    commands,
+    window,
+    workspace,
+} from 'vscode';
 import { rangeToStr } from '../utils';
 import { adaExtState } from '../../src/extension';
 
@@ -134,3 +150,197 @@ suite('getSymbols', function () {
         };
     }
 });
+
+suite('parallelize', function () {
+    const cases = [
+        {
+            size: 4,
+            threads: 1,
+            workDuration: 200,
+        },
+        {
+            size: 4,
+            threads: 2,
+            workDuration: 200,
+        },
+        {
+            size: 4,
+            threads: 3,
+            workDuration: 200,
+        },
+        {
+            size: 4,
+            threads: 4,
+            workDuration: 200,
+        },
+        {
+            size: 4,
+            threads: 5,
+            workDuration: 200,
+        },
+        {
+            size: 2 ** 10,
+            threads: 3,
+            workDuration: 0,
+        },
+        {
+            size: 2 ** 20,
+            threads: 3,
+            workDuration: 0,
+        },
+    ];
+
+    for (const tc of cases) {
+        testCase(tc.size, tc.threads, tc.workDuration);
+    }
+
+    function testCase(dataSize: number, threads: number, singleWorkDurationMs: number) {
+        test(`Process ${dataSize} elements with ${threads} threads`, async function () {
+            const data: number[] = Array.from({ length: dataSize }, (_, i) => i);
+            let done = 0;
+            let lastProgressDone = 0;
+            const start = Date.now();
+            const result = await window.withProgress(
+                {
+                    location: ProgressLocation.Notification,
+                    cancellable: true,
+                    title: this.currentTest?.fullTitle(),
+                },
+                async (progress, token) => {
+                    return await parallelize(
+                        data,
+                        threads,
+                        async (i) => {
+                            if (token.isCancellationRequested) {
+                                throw new CancellationError();
+                            }
+
+                            /**
+                             * Do we need a lock to increment this counter given
+                             * that we are processing with threads?
+                             *
+                             * The answer is no.
+                             *
+                             * In JavaScript semantics each function runs to
+                             * completion uninterrupted. 'async' doesn't mean that
+                             * functions run concurrently. It means that something
+                             * will be executed later.
+                             *
+                             * When a function flow encounters 'await', it hands
+                             * off processing to another async operation.
+                             *
+                             * So at any one time, only one function is executing
+                             * and accessing the counter. It is safe to increment
+                             * without locking.
+                             */
+                            ++done;
+
+                            const reportEveryProgress = false;
+                            if (reportEveryProgress) {
+                                progress.report({
+                                    message: `${done} / ${data.length}`,
+                                    increment: (1 * 100) / data.length,
+                                });
+                            } else {
+                                lastProgressDone = staggerProgress(
+                                    done,
+                                    data.length,
+                                    lastProgressDone,
+                                    (increment) => {
+                                        progress.report({
+                                            message: `${done} / ${data.length}`,
+                                            increment: increment,
+                                        });
+                                    },
+                                );
+                            }
+
+                            /**
+                             * Short-circuit the timing emulation if the test case doesn't want it.
+                             */
+                            if (singleWorkDurationMs > 0) {
+                                /**
+                                 * Create artificial delay in the processing of each item.
+                                 */
+                                await new Promise((resolve) => {
+                                    setTimeout(resolve, singleWorkDurationMs);
+                                });
+                            }
+
+                            return i + 1;
+                        },
+                        token,
+                    );
+                },
+            );
+            const duration = Date.now() - start;
+
+            assert.equal(result.length, data.length);
+
+            /**
+             * Don't check timing when the test cases doesn't use timing emulation.
+             */
+            if (singleWorkDurationMs > 0) {
+                /**
+                 * Given the number of threads, the overal duration should
+                 * not go beyond a certain limit. For example, if the data
+                 * size is 4, and we requested 4 threads to do the
+                 * operation, then all the data is processed in one go. So
+                 * overall the computation shouldn't take more than the
+                 * time needed to process one data item (with a margin).
+                 */
+                const margin = singleWorkDurationMs * 0.2;
+                /**
+                 * When processing 4 items with 3 threads, the first parallel
+                 * batch processes 3 items and a second batch processes 1 item.
+                 * So there are 2 sequential iterations, hence the use of
+                 * Math.ceil.
+                 */
+                const sequentialChunks = Math.ceil(dataSize / threads);
+                const expectedTotalWorkDuration = sequentialChunks * singleWorkDurationMs + margin;
+
+                assert.ok(
+                    duration < expectedTotalWorkDuration,
+                    `The computation took ${duration}ms when we expected` +
+                        ` no more than ${expectedTotalWorkDuration}`,
+                );
+            }
+        });
+    }
+});
+
+suite('staggerProgress', function () {
+    test('staggerProgress', function () {
+        const total = 1000;
+        let lastProgress = 0;
+
+        lastProgress = staggerProgress(5, total, lastProgress, () => {
+            /**
+             * progress is less than 1% so this shouldn't be called.
+             */
+            assert.fail('Progress reporting was called unexpectedly');
+        });
+        assert.equal(lastProgress, 0);
+
+        let called = false;
+        lastProgress = staggerProgress(10, total, lastProgress, (increment) => {
+            /**
+             * Progress is 1%. Reporting should be called.
+             */
+            called = true;
+            assert.equal(increment, 1);
+        });
+        assert.ok(called, 'Progress reporting function was unexpectedly not called');
+        assert.equal(lastProgress, 10);
+
+        lastProgress = staggerProgress(50, total, lastProgress, (increment) => {
+            /**
+             * Progress is 1%. Reporting should be called.
+             */
+            called = true;
+            assert.equal(increment, 4);
+        });
+        assert.ok(called, 'Progress reporting function was unexpectedly not called');
+        assert.equal(lastProgress, 50);
+    });
+});