Merge pull request #20 from Financial-Times/matth/depends-on

Add a check in the task runner to wait for dependents to finish

readme.md

@@ -34,21 +34,21 @@ _Please note:_ Before executing a command Athloi will sort the packages [topolog
 
 ### exec
 
-Runs an arbitrary command in the scope of each package.
+Runs an arbitrary command within the scope of each package.
 
 ```sh
 athloi exec npm install
 ```
 
-A double-dash (`--`) is necessary to pass any dashed arguments to the script being executed.
+A double-dash (`--`) is necessary to pass any dashed arguments to the command being executed.
 
 ```sh
-athloi exec -- npm i -D
+athloi exec -- npm i -D lodash
 ```
 
 ### run
 
-Runs an [npm script] in each package that contains that script.
+Runs an [npm script] in each package that defines that script.
 
 ```sh
 athloi run build
@@ -66,7 +66,7 @@ athloi script path/to/task.js
 
 ### version
 
-Updates the release number for all packages and writes the new data back to `package.json`. The given tag must parseable as a valid semver number.
+Updates the release number for all public packages and writes the new data back to `package.json`. The given tag must parseable as a valid semver number.
 
 ```sh
 athloi version v1.0.0
@@ -93,18 +93,25 @@ athloi publish -- --access=public
 
 ### concurrency
 
-A global concurrency option which can be used to execute multiple tasks in parallel. By default only one task will run at a time.
+A global option which will execute up to the given number of tasks concurrently. By default one task will be run at a time.
 
 ```sh
-# run a build script 3 packages at a time
-athloi run build --concurrency 3
+# run a lint script in up to 3 packages at a time
+athloi run lint --concurrency 3
 ```
 
-_Please note:_ using a concurrency value higher than 1 no longer ensures that tasks will finish for packages which are dependencies of other packages.
+### preserve-order
+
+A global flag which will ensure tasks maintain topological sort order. When used with a concurrency value higher than 1 this option will force queued tasks to wait for any still running tasks in cross-dependent packages to finish first.
+
+```sh
+# run a concurrent build script but ensure dependencies are built first
+athloi run build --concurrency 5 --preserve-order
+```
 
 ### filter
 
-A global filter option which can be used for all tasks. It can filter packages based on the value of a field within their package manifest.
+A global option which can be used for all tasks. It filters packages based on the value of a field within their package manifest or the package name.
 
 ```sh
 # Run a build script in only the packages marked as private

src/bin/cli

@@ -12,7 +12,11 @@ program
 	.option(
 		'-C, --concurrency <number>',
 		'Number of tasks to be run concurrently',
-		(arg) => /\d/.test(arg) ? parseInt(arg, 10) : undefined
+		(arg) => /^\d$/.test(arg) ? parseInt(arg, 10) : null
+	)
+	.option(
+		'-P, --preserve-order',
+		'Preserve topological sort order when running tasks concurrently'
 	)
 	.option(
 		'-R, --reverse',

src/cli-task.js

@@ -39,7 +39,7 @@ module.exports = (task) => {
 			logger.info(`Running ${tasks.length} tasks`);
 
 			// 6. execute all tasks
-			await runParallel(tasks, globals.concurrency);
+			await runParallel(tasks, globals.concurrency, globals.preserveOrder);
 
 			timer.stop();
 

src/evented-queue.js

@@ -0,0 +1,39 @@
+const EventEmitter = require('events');
+
+class EventedQueue extends EventEmitter {
+	constructor() {
+		super();
+		this.queue = new Set();
+	}
+
+	add(item) {
+		this.queue.add(item);
+		this.emit('add', item);
+		return this;
+	}
+
+	delete(item) {
+		this.queue.delete(item);
+		this.emit('delete', item);
+		return this;
+	}
+
+	waitFor(items = []) {
+		return new Promise((resolve) => {
+			const callback = () => {
+				const itemsRunning = items.some((item) => this.queue.has(item));
+
+				if (!itemsRunning) {
+					this.removeListener('delete', callback);
+					resolve();
+				}
+			};
+
+			this.on('delete', callback);
+
+			callback(null);
+		});
+	}
+}
+
+module.exports = EventedQueue;

src/package.js

@@ -31,6 +31,15 @@ class Package {
 		return path.relative(process.cwd(), this.location);
 	}
 
+	get allDependencies () {
+		return Object.keys({
+			...this.manifest.dependencies,
+			...this.manifest.devDependencies,
+			...this.manifest.peerDependencies,
+			...this.manifest.optionalDependencies
+		});
+	};
+
 	async writeManifest (manifest) {
 		const json = JSON.stringify(manifest, null, 2);
 		await writeFile(this.manifestLocation, json);

src/run-parallel.js

@@ -1,14 +1,30 @@
+const Semaphore = require('async-sema');
 const logger = require('./logger');
-const Sema = require('async-sema');
+const EventedQueue = require('./evented-queue');
 
-module.exports = (tasks = [], concurrency = 1) => {
-	const sema = new Sema(concurrency);
+module.exports = (tasks = [], concurrency = 1, preserveOrder = false) => {
+	const semaphore = new Semaphore(concurrency);
+	const queue = new EventedQueue();
 
 	logger.info(`Executing up to ${concurrency} tasks at a time`);
 
 	return Promise.all(
-		tasks.map((task) => {
-			return sema.acquire().then(task).then(() => sema.release());
+		tasks.map(({ pkg, apply }) => {
+			queue.add(pkg.name);
+
+			return semaphore
+				.acquire()
+				.then(() => {
+					// wait for any dependencies still in the queue to finish
+					return preserveOrder ? queue.waitFor(pkg.allDependencies) : null;
+				})
+				.then(() => {
+					return apply();
+				})
+				.then(() => {
+					queue.delete(pkg.name);
+					return semaphore.release();
+				});
 		})
 	);
 };

src/sort-packages.js

@@ -1,21 +1,10 @@
 const toposort = require('toposort');
 
-const collateDependencies = (manifest) => {
-	return Object.keys({
-		...manifest.dependencies,
-		...manifest.devDependencies,
-		...manifest.peerDependencies,
-		...manifest.optionalDependencies
-	});
-};
-
 module.exports = (reverse = false, packages = []) => {
 	const packageNames = new Set(packages.map((pkg) => pkg.name));
 
 	const edges = packages.reduce((edges, pkg) => {
-		const dependencyNames = collateDependencies(pkg.manifest);
-
-		const localDependencies = dependencyNames.filter((dependency) => {
+		const localDependencies = pkg.allDependencies.filter((dependency) => {
 			return packageNames.has(dependency);
 		});
 

src/tasks/exec.js

@@ -3,7 +3,8 @@ const runPackage = require('../run-package');
 
 function exec (packages = [], command, args = []) {
 	return packages.map((pkg) => {
-		return () => runPackage(command, args, pkg.location);
+		const apply = () => runPackage(command, args, pkg.location);
+		return { pkg, apply };
 	});
 };
 

src/tasks/publish.js

@@ -10,7 +10,8 @@ function publish (packages = [], args = []) {
 
 	// create a queue of tasks to run
 	return filteredPackages.map((pkg) => {
-		return () => runPackage('npm', ['publish', ...args], pkg.location);
+		const apply = () => runPackage('npm', ['publish', ...args], pkg.location);
+		return { pkg, apply };
 	});
 };
 

src/tasks/run.js

@@ -12,7 +12,8 @@ function run (packages = [], script) {
 
 	// create a queue of tasks to run
 	return filteredPackages.map((pkg) => {
-		return () => runPackage('npm', ['run', script], pkg.location);
+		const apply = () => runPackage('npm', ['run', script], pkg.location);
+		return { pkg, apply };
 	});
 };
 

src/tasks/script.js

@@ -6,7 +6,8 @@ function script (packages = [], scriptPath) {
 	const resolvedScript = path.resolve(process.cwd(), scriptPath);
 
 	return packages.map((pkg) => {
-		return () => runPackage('node', [resolvedScript], pkg.location);
+		const apply = () => runPackage('node', [resolvedScript], pkg.location);
+		return { pkg, apply };
 	});
 }
 

src/tasks/version.js

@@ -16,10 +16,12 @@ function version (packages = [], tag) {
 	const packageNames = new Set(packages.map((pkg) => pkg.name));
 
 	return packages.map((pkg) => {
-		return () => {
+		const apply = () => {
 			const newManifest = updateVersions(pkg.manifest, number, packageNames);
 			return pkg.writeManifest(newManifest);
 		};
+
+		return { pkg, apply };
 	});
 };
 

test/helpers/create-package.js

@@ -0,0 +1,10 @@
+const Package = require('../../src/package');
+
+module.exports = (name, options = {}) => {
+	const manifest = { name, ...options };
+	const instance = new Package(manifest, `/Path/to/${name}`);
+
+	instance.writeManifest = jest.fn();
+
+	return instance;
+};

test/src/evented-queue.spec.js

@@ -0,0 +1,63 @@
+const Subject = require('../../src/evented-queue');
+
+describe('src/evented-queue', () => {
+	let instance;
+
+	beforeEach(() => {
+		instance = new Subject();
+	});
+
+	describe('#add', () => {
+		it('adds the given item to the queue', () => {
+			instance.add('foo');
+			expect(instance.queue.size).toEqual(1);
+		});
+
+		it('emits an event when items are added to the queue', (done) => {
+			instance.on('add', () => {
+				done();
+			});
+
+			instance.add('foo');
+		});
+	});
+
+	describe('#delete', () => {
+		it('removes the given item to the queue', () => {
+			instance.add('foo');
+			expect(instance.queue.size).toEqual(1);
+
+			instance.delete('foo');
+			expect(instance.queue.size).toEqual(0);
+		});
+
+		it('emits an event when items are removed from the queue', (done) => {
+			instance.add('foo');
+
+			instance.on('delete', () => {
+				done();
+			});
+
+			instance.delete('foo');
+		});
+	});
+
+	describe('#waitFor', () => {
+		it('resolves when the queue no longer contains any of the given items', (done) => {
+			instance
+				.add('foo')
+				.add('bar')
+				.add('baz');
+
+			instance.waitFor(['foo', 'bar', 'baz']).then(() => {
+				expect(instance.queue.size).toEqual(0);
+				done();
+			});
+
+			instance
+				.delete('foo')
+				.delete('bar')
+				.delete('baz');
+		});
+	});
+});

test/src/package.spec.js

@@ -6,6 +6,14 @@ const Subject = require('../../src/package');
 const fixture = Object.freeze({
 	name: 'my-package',
 	version: '0.0.0',
+	dependencies: {
+		lodash: '^3.0.0',
+		hyperons: '^0.5.0'
+	},
+	devDependencies: {
+		jest: '^16.0.0',
+		prettier: '^12.0.0'
+	}
 });
 
 describe('src/package', () => {
@@ -53,6 +61,14 @@ describe('src/package', () => {
 		});
 	});
 
+	describe('get #allDependencies', () => {
+		it('returns a list of all dependencies', () => {
+			const instance = factory(fixture);
+			expect(instance.allDependencies).toBeInstanceOf(Array);
+			expect(instance.allDependencies.length).toEqual(4);
+		});
+	});
+
 	describe('#writeManifest', () => {
 		beforeEach(() => {
 			// The final arg is a callback that needs calling!
@@ -65,7 +81,7 @@ describe('src/package', () => {
 
 			expect(fs.writeFile).toHaveBeenCalledWith(
 				'/root/path/to/package/package.json',
-				JSON.stringify({ name : 'my-package', version: '1.0.0' }, null, 2),
+				JSON.stringify({ ...fixture, version: '1.0.0' }, null, 2),
 				expect.any(Function)
 			);
 		});