diff --git a/lib/async-queue.js b/lib/async-queue.js index a9e671c7d0..dfdfadf17e 100644 --- a/lib/async-queue.js +++ b/lib/async-queue.js @@ -1,28 +1,72 @@ +class Task { + constructor(fn, parallel = true) { + this.fn = fn; + this.parallel = parallel; + this.promise = new Promise((resolve, reject) => { + this.resolve = resolve; + this.reject = reject; + }); + } + + async execute() { + try { + const value = await this.fn.call(undefined); + this.resolve(value); + } catch (err) { + this.reject(err); + } + } + + runsInParallel() { + return this.parallel; + } + + runsInSerial() { + return !this.parallel; + } + + getPromise() { + return this.promise; + } +} + export default class AsyncQueue { - constructor() { - this.commandQueue = []; - this.running = false; - } - - push(fn) { - const p = new Promise((resolve, reject) => this.commandQueue.push({fn, resolve, reject})); - if (!this.running) { this.processQueue(); } - return p; - } - - async processQueue() { - this.running = true; - let lastPromise = Promise.resolve(); - while (this.commandQueue.length) { - const {fn, resolve, reject} = this.commandQueue[0]; - const run = () => fn(); - lastPromise = lastPromise.then(run, run); - lastPromise.then(resolve, reject); - try { - await lastPromise; // eslint-disable-line babel/no-await-in-loop - } catch (e) { /* nothing */ } - this.commandQueue.shift(); + constructor(options = {}) { + this.parallelism = options.parallelism || 1; + this.nonParallelizableOperation = false; + this.tasksInProgress = 0; + this.queue = []; + } + + push(fn, {parallel} = {parallel: true}) { + const task = new Task(fn, parallel); + this.queue.push(task); + this.processQueue(); + return task.getPromise(); + } + + processQueue() { + if (!this.queue.length || this.nonParallelizableOperation) { return; } + + const task = this.queue[0]; + const canRunParallelOp = task.runsInParallel() && this.tasksInProgress < this.parallelism; + const canRunSerialOp = task.runsInSerial() && this.tasksInProgress === 0; + if (canRunSerialOp || canRunParallelOp) { + this.processTask(task, task.runsInParallel()); + this.queue.shift(); + this.processQueue(); } - this.running = false; + } + + async processTask(task, runsInParallel) { + this.tasksInProgress++; + if (!runsInParallel) { + this.nonParallelizableOperation = true; + } + + await task.execute(); + this.tasksInProgress--; + this.nonParallelizableOperation = false; + this.processQueue(); } } diff --git a/lib/controllers/git-panel-controller.js b/lib/controllers/git-panel-controller.js index 20d2177642..450cf97880 100644 --- a/lib/controllers/git-panel-controller.js +++ b/lib/controllers/git-panel-controller.js @@ -82,18 +82,28 @@ export default class GitPanelController { @autobind async fetchRepositoryData(repository) { + const dataPromises = { + mergeConflicts: repository.getMergeConflicts(), + lastCommit: repository.getLastCommit(), + isMerging: repository.isMerging(), + branchName: repository.getCurrentBranch(), + branches: repository.getBranches(), + remoteName: repository.getRemoteForBranch(this.branchName), + unstagedChanges: repository.getUnstagedChanges(), + stagedChanges: this.fetchStagedChanges(repository), + }; const data = { - unstagedChanges: await repository.getUnstagedChanges(), - stagedChanges: await this.fetchStagedChanges(repository), - mergeConflicts: await repository.getMergeConflicts(), - lastCommit: await repository.getLastCommit(), - isMerging: await repository.isMerging(), - branchName: await repository.getCurrentBranch(), - branches: await repository.getBranches(), + unstagedChanges: await dataPromises.unstagedChanges, + stagedChanges: await dataPromises.stagedChanges, + mergeConflicts: await dataPromises.mergeConflicts, + lastCommit: await dataPromises.lastCommit, + isMerging: await dataPromises.isMerging, + branchName: await dataPromises.branchName, + branches: await dataPromises.branches, mergeMessage: null, aheadCount: null, behindCount: null, - remoteName: await repository.getRemoteForBranch(this.branchName), + remoteName: await dataPromises.remoteName, }; if (data.remoteName) { diff --git a/lib/git-shell-out-strategy.js b/lib/git-shell-out-strategy.js index 5de778cc74..f47a7d59ed 100644 --- a/lib/git-shell-out-strategy.js +++ b/lib/git-shell-out-strategy.js @@ -1,4 +1,5 @@ import path from 'path'; +import os from 'os'; import {CompositeDisposable} from 'atom'; @@ -30,13 +31,16 @@ function withGpgScript(args) { } export default class GitShellOutStrategy { - constructor(workingDir) { + static defaultExecArgs = {stdin: null, useGitPromptServer: false, writeOperation: false} + + constructor(workingDir, options = {}) { this.workingDir = workingDir; - this.commandQueue = new AsyncQueue(); + const parallelism = options.parallelism || Math.max(3, os.cpus().length); + this.commandQueue = new AsyncQueue({parallelism}); } // Execute a command and read the output using the embedded Git environment - exec(args, stdin = null, useGitPromptServer = false) { + exec(args, {stdin, useGitPromptServer, writeOperation} = GitShellOutStrategy.defaultExecArgs) { /* eslint-disable no-console */ const subscriptions = new CompositeDisposable(); @@ -86,30 +90,35 @@ export default class GitShellOutStrategy { if (process.env.PRINT_GIT_TIMES) { console.time(`git:${formattedArgs}`); } - timingMarker.mark('execute'); - return GitProcess.exec(args, this.workingDir, options) - .then(({stdout, stderr, exitCode}) => { - timingMarker.finalize(); - if (process.env.PRINT_GIT_TIMES) { - console.timeEnd(`git:${formattedArgs}`); - } - if (gitPromptServer) { - gitPromptServer.terminate(); - } - subscriptions.dispose(); - if (exitCode) { - const err = new GitError( - `${formattedArgs} exited with code ${exitCode}\nstdout: ${stdout}\nstderr: ${stderr}`, - ); - err.code = exitCode; - err.stdErr = stderr; - err.stdOut = stdout; - err.command = formattedArgs; - return Promise.reject(err); - } - return stdout; + return new Promise(resolve => { + timingMarker.mark('nexttick'); + setImmediate(() => { + timingMarker.mark('execute'); + resolve(GitProcess.exec(args, this.workingDir, options) + .then(({stdout, stderr, exitCode}) => { + timingMarker.finalize(); + if (process.env.PRINT_GIT_TIMES) { + console.timeEnd(`git:${formattedArgs}`); + } + if (gitPromptServer) { + gitPromptServer.terminate(); + } + subscriptions.dispose(); + if (exitCode) { + const err = new GitError( + `${formattedArgs} exited with code ${exitCode}\nstdout: ${stdout}\nstderr: ${stderr}`, + ); + err.code = exitCode; + err.stdErr = stderr; + err.stdOut = stdout; + err.command = formattedArgs; + return Promise.reject(err); + } + return stdout; + })); }); - }); + }); + }, {parallel: !writeOperation}); /* eslint-enable no-console */ } @@ -117,12 +126,12 @@ export default class GitShellOutStrategy { * Execute a git command that may create a commit. If the command fails because the GPG binary was invoked and unable * to acquire a passphrase (because the pinentry program attempted to use a tty), retry with a `GitPromptServer`. */ - gpgExec(args, stdin = null) { + gpgExec(args, {stdin} = {stdin: null}) { const gpgArgs = withGpgScript(args); - return this.exec(gpgArgs, stdin).catch(err => { + return this.exec(gpgArgs, {stdin, writeOperation: true}).catch(err => { if (err.code === 128 && /gpg failed/.test(err.stdErr)) { // Retry with a GitPromptServer - return this.exec(gpgArgs, stdin, true); + return this.exec(gpgArgs, {stdin, useGitPromptServer: true, writeOperation: true}); } else { throw err; } @@ -144,33 +153,33 @@ export default class GitShellOutStrategy { stageFiles(paths) { if (paths.length === 0) { return null; } const args = ['add'].concat(paths); - return this.exec(args); + return this.exec(args, {writeOperation: true}); } unstageFiles(paths, commit = 'HEAD') { if (paths.length === 0) { return null; } const args = ['reset', commit, '--'].concat(paths); - return this.exec(args); + return this.exec(args, {writeOperation: true}); } applyPatch(patch, {index} = {}) { const args = ['apply', '-']; if (index) { args.splice(1, 0, '--cached'); } - return this.exec(args, patch); + return this.exec(args, {stdin: patch, writeOperation: true}); } commit(message, {allowEmpty, amend} = {}) { const args = ['commit', '-m', message]; if (amend) { args.push('--amend'); } if (allowEmpty) { args.push('--allow-empty'); } - return this.gpgExec(args); + return this.gpgExec(args, {writeOperation: true}); } /** * File Status and Diffs */ async getStatusesForChangedFiles() { - const output = await this.exec(['status', '--untracked-files=all', '-z']); + const output = await this.exec(['status', '--untracked-files=all', '-z'], {writeOperation: true}); const statusMap = { 'A': 'added', @@ -280,7 +289,7 @@ export default class GitShellOutStrategy { async isPartiallyStaged(filePath) { const args = ['status', '--short', '--', filePath]; - const output = await this.exec(args); + const output = await this.exec(args, {writeOperation: true}); const results = output.trim().split(LINE_ENDING_REGEX); if (results.length === 2) { return true; @@ -325,7 +334,7 @@ export default class GitShellOutStrategy { } abortMerge() { - return this.exec(['merge', '--abort']); + return this.exec(['merge', '--abort'], {writeOperation: true}); } /** @@ -334,7 +343,7 @@ export default class GitShellOutStrategy { clone(remoteUrl, options = {}) { const args = ['clone', '--no-local', remoteUrl, this.workingDir]; if (options.bare) { args.push('--bare'); } - return this.exec(args); + return this.exec(args, {writeOperation: true}); } async getRemoteForBranch(branchName) { @@ -348,12 +357,12 @@ export default class GitShellOutStrategy { async fetch(branchName) { const remote = await this.getRemoteForBranch(branchName); - return this.exec(['fetch', remote, branchName], null, true); + return this.exec(['fetch', remote, branchName], {useGitPromptServer: true, writeOperation: true}); } async pull(branchName) { const remote = await this.getRemoteForBranch(branchName); - return this.gpgExec(['pull', remote, branchName], null, true); + return this.gpgExec(['pull', remote, branchName], {useGitPromptServer: true, writeOperation: true}); } async push(branchName, options = {}) { @@ -361,7 +370,7 @@ export default class GitShellOutStrategy { const args = ['push', remote || 'origin', branchName]; if (options.setUpstream) { args.push('--set-upstream'); } if (options.force) { args.push('--force'); } - return this.exec(args, null, true); + return this.exec(args, {useGitPromptServer: true, writeOperation: true}); } async getAheadCount(branchName) { @@ -395,7 +404,7 @@ export default class GitShellOutStrategy { checkout(branchName, options = {}) { const args = ['checkout']; if (options.createNew) { args.push('-b'); } - return this.exec(args.concat(branchName)); + return this.exec(args.concat(branchName), {writeOperation: true}); } async checkoutFiles(paths, revision) { @@ -403,7 +412,7 @@ export default class GitShellOutStrategy { try { const args = ['checkout']; if (revision) { args.push(revision); } - return await this.exec(args.concat('--', paths)); + return await this.exec(args.concat('--', paths), {writeOperation: true}); } catch (error) { const matches = error.stdErr.match(/error: pathspec .* did not match any file\(s\) known to git\./g); if (matches.length) { @@ -449,7 +458,7 @@ export default class GitShellOutStrategy { let args = ['config']; if (replaceAll) { args.push('--replace-all'); } args = args.concat(option, value); - return this.exec(args); + return this.exec(args, {writeOperation: true}); } async getRemotes() { @@ -472,9 +481,9 @@ export default class GitShellOutStrategy { async createBlob({filePath, stdin} = {}) { let output; if (filePath) { - output = await this.exec(['hash-object', '-w', filePath]); + output = await this.exec(['hash-object', '-w', filePath], {writeOperation: true}); } else if (stdin) { - output = await this.exec(['hash-object', '-w', '--stdin'], stdin); + output = await this.exec(['hash-object', '-w', '--stdin'], {stdin, writeOperation: true}); } else { throw new Error('Must supply file path or stdin'); } diff --git a/lib/views/git-timings-view.js b/lib/views/git-timings-view.js index 2c1066ade1..b9faa0508b 100644 --- a/lib/views/git-timings-view.js +++ b/lib/views/git-timings-view.js @@ -94,6 +94,7 @@ class MarkerTooltip extends React.Component { const COLORS = { queued: 'red', prepare: 'cyan', + nexttick: 'yellow', execute: 'green', }; class MarkerSpan extends React.Component { diff --git a/test/async-queue.test.js b/test/async-queue.test.js index b27751cb15..988f88b3a8 100644 --- a/test/async-queue.test.js +++ b/test/async-queue.test.js @@ -1,47 +1,148 @@ +import {autobind} from 'core-decorators'; + import AsyncQueue from '../lib/async-queue'; -const timer = n => { - return new Promise(resolve => { - setTimeout(resolve, 10 - n); - }); -}; +class Task { + constructor(name, error) { + this.name = name; + this.error = error; + this.started = false; + this.finished = false; + } -describe('AsyncQueue', function() { - it('runs items in the order queued', async function() { - const queue = new AsyncQueue(); - - const expectedEvents = []; - const actualEvents = []; - const promises = []; - for (let i = 0; i < 10; i++) { - expectedEvents.push(i); - const promise = queue.push(() => timer(i)).then(() => actualEvents.push(i)); - promises.push(promise); + @autobind + run() { + this.started = true; + this.finished = false; + return new Promise((resolve, reject) => { + this.resolve = resolve; + this.reject = reject; + }); + } + + @autobind + finish() { + this.finished = true; + if (this.error) { + this.reject(new Error(this.name)); + } else { + this.resolve(this.name); } + } +} + +describe('AsyncQueue', function() { + it('runs items in parallel up to the set max', async function() { + const queue = new AsyncQueue({parallelism: 3}); + + const tasks = [ + new Task('task 1'), + new Task('task 2', true), + new Task('task 3'), + new Task('task 4'), + new Task('task 5'), + ]; + const results = [false, false, false, false, false]; + const errors = [false, false, false, false, false]; + + const p0 = queue.push(() => tasks[0].run()); + const p1 = queue.push(() => tasks[1].run()); + const p2 = queue.push(() => tasks[2].run()); + const p3 = queue.push(() => tasks[3].run()); + const p4 = queue.push(() => tasks[4].run()); + + p0.then(value => { results[0] = value; }).catch(err => { errors[0] = err; }); + p1.then(value => { results[1] = value; }).catch(err => { errors[1] = err; }); + p2.then(value => { results[2] = value; }).catch(err => { errors[2] = err; }); + p3.then(value => { results[3] = value; }).catch(err => { errors[3] = err; }); + p4.then(value => { results[4] = value; }).catch(err => { errors[4] = err; }); + + assert.isTrue(tasks[0].started); + assert.isTrue(tasks[1].started); + assert.isTrue(tasks[2].started); + assert.isFalse(tasks[3].started); + assert.isFalse(tasks[4].started); + + assert.isFalse(results[0]); + + tasks[0].finish(); + assert.isTrue(tasks[0].finished); + await assert.async.equal(results[0], 'task 1'); + assert.isFalse(tasks[1].finished); + assert.isFalse(results[1]); + + assert.isTrue(tasks[3].started); + assert.isFalse(tasks[4].started); - await Promise.all(promises); - assert.deepEqual(expectedEvents, actualEvents); + tasks[1].finish(); + assert.isTrue(tasks[1].finished); + assert.isFalse(tasks[2].finished); + await assert.async.equal(errors[1].message, 'task 2'); + + assert.isTrue(tasks[4].started); }); - it('continues running queued items when one fails', async function() { - const queue = new AsyncQueue(); - - const expectedEvents = []; - const actualEvents = []; - const promises = []; - for (let i = 0; i < 10; i++) { - expectedEvents.push(i === 5 ? 'error' : i); - const promise = queue.push(() => { - if (i === 5) { - throw new Error('omg'); - } else { - return timer(i); - } - }).then(() => actualEvents.push(i), () => actualEvents.push('error')); - promises.push(promise); - } + it('runs non-parallelizable tasks serially', async function() { + const queue = new AsyncQueue({parallelism: 3}); + + const tasks = [ + new Task('task 1'), + new Task('task 2'), + new Task('task 3'), + new Task('task 4'), + new Task('task 5'), + new Task('task 6'), + ]; + + const p0 = queue.push(() => tasks[0].run()); + const p1 = queue.push(() => tasks[1].run()); + const p2 = queue.push(() => tasks[2].run(), {parallel: false}); + const p3 = queue.push(() => tasks[3].run(), {parallel: false}); + queue.push(() => tasks[4].run()); + queue.push(() => tasks[5].run()); + + assert.isTrue(tasks[0].started); + assert.isTrue(tasks[1].started); + assert.isFalse(tasks[2].started); // not parallelizable!! + assert.isFalse(tasks[3].started); + assert.isFalse(tasks[4].started); + assert.isFalse(tasks[5].started); + + tasks[0].finish(); + await p0; + assert.isFalse(tasks[2].started); // still can't be started + assert.isFalse(tasks[3].started); + + tasks[1].finish(); + await p1; + await assert.async.isTrue(tasks[2].started); + assert.isFalse(tasks[3].started); // still can't be started + + tasks[2].finish(); + await p2; + await assert.async.isTrue(tasks[3].started); + assert.isFalse(tasks[4].started); // 3 is non-parallelizable so 4 can't start + + tasks[3].finish(); + await p3; + await assert.async.isTrue(tasks[4].started); // both can start since they are parallelizable + assert.isTrue(tasks[5].started); + }); + + it('continues to work when tasks throw synchronous errors', async function() { + const queue = new AsyncQueue({parallelism: 1}); + + const p1 = queue.push(() => { + throw new Error('error thrown from task 1'); + }); + const p2 = queue.push(() => { + return new Promise(res => res(2)); + }); - await Promise.all(promises); - assert.deepEqual(expectedEvents, actualEvents); + try { + await p1; + throw new Error('expected p1 to be rejectd'); + } catch (err) {} + assert.equal(await p2, 2); }); }); diff --git a/test/git-shell-out-strategy.test.js b/test/git-shell-out-strategy.test.js index 8d7fe1b97d..043c189bb9 100644 --- a/test/git-shell-out-strategy.test.js +++ b/test/git-shell-out-strategy.test.js @@ -14,42 +14,6 @@ import {cloneRepository, assertDeepPropertyVals, setUpLocalAndRemoteRepositories */ describe('Git commands', function() { - describe('exec', function() { - it('serializes operations', async function() { - const workingDirPath = await cloneRepository('three-files'); - const git = new GitShellOutStrategy(workingDirPath); - const expectedEvents = []; - const actualEvents = []; - const promises = []; - for (let i = 0; i < 10; i++) { - expectedEvents.push(i); - promises.push(git.getHeadCommit().then(() => actualEvents.push(i))); - } - - await Promise.all(promises); - assert.deepEqual(expectedEvents, actualEvents); - }); - - it('runs operations after one fails', async function() { - const workingDirPath = await cloneRepository('three-files'); - const git = new GitShellOutStrategy(workingDirPath); - const expectedEvents = []; - const actualEvents = []; - const promises = []; - for (let i = 0; i < 10; i++) { - expectedEvents.push(i); - if (i === 5) { - promises.push(git.exec(['fake', 'command']).catch(() => actualEvents.push(i))); - } else { - promises.push(git.exec(['status']).then(() => actualEvents.push(i))); - } - } - - await Promise.all(promises); - assert.deepEqual(expectedEvents, actualEvents); - }); - }); - describe('isGitRepository(directoryPath)', function() { it('returns true if the path passed is a valid repository, and false if not', async function() { const workingDirPath = await cloneRepository('three-files'); @@ -650,8 +614,8 @@ describe('Git commands', function() { const execArgs = callArgs[0]; assert.equal(execArgs[0], '-c'); assert.match(execArgs[1], /^gpg\.program=.*gpg-no-tty\.sh$/); - assert.isNull(callArgs[1]); - assert.isNotOk(callArgs[2]); + assert.isNotOk(callArgs[1].stdin); + assert.isNotOk(callArgs[1].useGitPromptServer); }); it(`retries a ${op.verbalNoun} with a GitPromptServer when GPG signing fails`, async () => { @@ -677,15 +641,15 @@ describe('Git commands', function() { const execArgs0 = callArgs0[0]; assert.equal(execArgs0[0], '-c'); assert.match(execArgs0[1], /^gpg\.program=.*gpg-no-tty\.sh$/); - assert.isNull(callArgs0[1]); - assert.isNotOk(callArgs0[2]); + assert.isNotOk(callArgs0[1].stdin); + assert.isNotOk(callArgs0[1].useGitPromptServer); const callArgs1 = execStub.getCall(callIndex + 1).args; const execArgs1 = callArgs1[0]; assert.equal(execArgs1[0], '-c'); assert.match(execArgs1[1], /^gpg\.program=.*gpg-no-tty\.sh$/); - assert.isNull(callArgs1[1]); - assert.isTrue(callArgs1[2]); + assert.isNotOk(callArgs1[1].stdin); + assert.isTrue(callArgs1[1].useGitPromptServer); }); }); }); diff --git a/test/runner.js b/test/runner.js index f74209a61d..02cec57b23 100644 --- a/test/runner.js +++ b/test/runner.js @@ -8,7 +8,7 @@ chai.use(chaiAsPromised); global.assert = chai.assert; // Give tests that rely on filesystem event delivery lots of breathing room. -until.setDefaultTimeout(5000); +until.setDefaultTimeout(3000); module.exports = createRunner({ htmlTitle: `GitHub Package Tests - pid ${process.pid}`,