Refactor client-side AI into multiple web workers

client_src/js/Helpers/PromiseUnwind.mjs

@@ -0,0 +1,16 @@
+"use strict";
+
+/**
+ * Unwinds a recursive promise.
+ * @param  {Promise} promise       The recursive promise to unwind.
+ * @param  {Function} callback_step A function to exexcute at each step of the unwinding process
+ */
+async function promise_unwind(promise) {
+	let current_promise = promise;
+	while(current_promise !== null) {
+		// Unwind this Promise and obtain it's return value
+		current_promise = await current_promise;
+	}
+}
+
+export default promise_unwind;

client_src/js/LayerAI.mjs

@@ -6,6 +6,7 @@ import chroma from 'chroma-js';
 import WorkerWrapper from './WorkerWrapper.mjs';
 import GetFromUrl from './Helpers/GetFromUrl.mjs';
 import Config from './ClientConfig.mjs';
+import promise_unwind from './Helpers/PromiseUnwind.mjs';
 
 
 class LayerAI {
@@ -37,9 +38,16 @@ class LayerAI {
 	 * @param {[type]} map [description]
 	 */
 	constructor(map) {
+		// The Leaflet map instance to attach to
 		this.map = map;
-		this.worker = new WorkerWrapper();
+		// The array of web workers that do the computation
+		this.workers = [];
+		// The array the GeoJSON is stored in as it's generated
+		this.geojson = [];
 		
+		this.stats = { rssi_min: Infinity, rssi_max: -Infinity };
+		
+		// The bounding box of the map we're generating
 		this.map_bounds = null;
 	}
 	
@@ -65,8 +73,23 @@ class LayerAI {
 		this.map_bounds.east += Config.border.lng;
 		this.map_bounds.west -= Config.border.lng;
 		
-		// Setup the web worker
+		// Setup the colour scale
-		await this.worker.setup(this.map_bounds, this.index);
+		this.colour_scale = chroma.scale([
+			Config.colour_scale.min,
+			Config.colour_scale.max
+		]).domain(
+			this.index.properties.rssi_min,
+			this.index.properties.rssi_max
+		);
+		
+		// Setup the web worker army
+		for(let i = 0; i < (navigator.hardwareConcurrency || 4); i++) {
+			console.info(`Starting worker ${i}`);
+			let next_worker = new WorkerWrapper();
+			
+			await next_worker.setup(this.map_bounds, this.index);
+			this.workers.push(next_worker);
+		}
 		
 		// Generate the Leaflet layer
 		this.layer = await this.generate_layer();
@@ -101,34 +124,50 @@ class LayerAI {
 	 * @return	{Promise}	A Promise that resolves to a GeoJSON array representing the map.
 	 */
 	async render_map() {
-		let coverage = [],
+		let work_generator = (function*() {
-			colour_scale = chroma.scale([
-				Config.colour_scale.min,
-				Config.colour_scale.max
-			]).domain(
-				this.index.properties.rssi_min,
-				this.index.properties.rssi_max
-			);
-		
-		let stats = {
-			rssi_min: Infinity,
-			rssi_max: -Infinity
-		};
-		
 			for(let lat = this.map_bounds.south; lat < this.map_bounds.north; lat += Config.step.lat) {
-			let next_result = await this.worker.predict_row(lat);
+				yield lat;
+			}
+		}).bind(this)();
+		
+		let unwinders = [];
+		for(let i = 0; i < this.workers.length; i++) {
+			let next_latitude = work_generator.next().value;
+			let next_promise =  this.workers[i].predict_row(next_latitude)
+				.then(this.process_result_row.bind(this, next_latitude, i, work_generator));
+			
+			unwinders.push(promise_unwind(next_promise));
+		}
+		
+		// Wait for all the workers to finish
+		await Promise.all(unwinders);
+		
+		this.workers.length = 0; // Empty the array of workers
+		
+		return this.geojson;
+	}
+	
+	/**
+	 * Processes a single row of results from a web worker.
+	 * @param	{number}	lat				The latitude value that was processed by the web worker.
+	 * @param	{number}	worker_index	The index of the worker that returned the results.
+	 * @param	{Generator}	work_generator	The work generator to use to queue the next work item.
+ 	 * @param	{Object}	next_result		The result object returned by the web worker.
+	 * @return	{Promise<Object>|null}		Returns a Promise that resolves to the next work item that need processing, or null if allt he work hass been completed.
+	 */
+	process_result_row(lat, worker_index, work_generator, next_result) {
 		let lng = this.map_bounds.west;
 		
 		// Keep up with the statistics
-			if(next_result.stats.rssi_max > stats.rssi_max)
+		if(next_result.stats.rssi_max > this.stats.rssi_max)
-				stats.rssi_max = next_result.stats.rssi_max;
+			this.stats.rssi_max = next_result.stats.rssi_max;
-			if(next_result.stats.rssi_min < stats.rssi_min)
+		if(next_result.stats.rssi_min < this.stats.rssi_min)
-				stats.rssi_min = next_result.stats.rssi_min;
+			this.stats.rssi_min = next_result.stats.rssi_min;
 		
 		
 		for(let value of next_result.result) {
 			// Generate the GeoJSON feature for this cell of the map
-				coverage.push({
+			this.geojson.push({
 				type: "Feature",
 				geometry: {
 					type: "Polygon",
@@ -143,19 +182,24 @@ class LayerAI {
 					]
 				},
 				properties: {
-						colour: colour_scale(value).toString()
+					colour: this.colour_scale(value).toString()
 				}
 			});
 			
 			lng += Config.step.lng;
 		}
+		
+		let next_work_item = work_generator.next();
+		if(!next_work_item.done) {
+			return this.workers[worker_index].predict_row(next_work_item.value)
+				.then(this.process_result_row.bind(this, next_work_item.value, worker_index, work_generator));
+		}
+		else {
+			console.log(`Ending worker ${worker_index}`);
+			// Shut down the worker - we're done!
+			this.workers[worker_index].end();
+			return null;
 		}
-		
-		this.worker.end(this.worker);
-		
-		console.log("Stats:", stats);
-		
-		return coverage;
 	}
 }