multi core better error handling and race condition stsartup debugging

Author: [email protected]

package.json

@@ -1,7 +1,7 @@
 {
   "name": "jsonrpc-bidirectional",
   "description": "Bidirectional JSONRPC over web sockets or HTTP with extensive plugin support.",
-  "version": "10.0.1",
+  "version": "10.0.10",
   "scripts": {
     "build": "node --experimental-worker build.js",
     "prepublish": "node --experimental-worker build.js && node --expose-gc --max-old-space-size=1024 --experimental-worker tests/main.js",

src/NodeClusterBase/WorkerEndpoint.js

@@ -1,5 +1,6 @@
 const cluster = require("cluster");
 const NodeMultiCoreCPUBase = require("../NodeMultiCoreCPUBase");
+const sleep = require("sleep-promise");
 
 const JSONRPC = {
 	BidirectionalWorkerRouter: require("../BidirectionalWorkerRouter")
@@ -36,6 +37,33 @@ class WorkerEndpoint extends NodeMultiCoreCPUBase.WorkerEndpoint
 	 */
 	async _currentWorkerID()
 	{
+		// https://github.com/nodejs/node/issues/1269
+		if(
+			!this._bAlreadyDelayedReadingWorkerID
+			&& (
+				!cluster.worker 
+				|| cluster.worker.id === null 
+				|| cluster.worker.id === undefined
+			)
+		)
+		{
+			await sleep(2000);
+			this._bAlreadyDelayedReadingWorkerID = true;
+		}
+
+
+		if(
+			!cluster.worker
+			|| cluster.worker.id === null 
+			|| cluster.worker.id === undefined
+		)
+		{
+			console.error("cluster.worker: ", cluster.worker);
+			console.error("cluster.worker.id: ", cluster.worker ? cluster.worker.id : "");
+			console.error(`Returning 0 as cluster.worker.id.`);
+			return 0;
+		}
+
 		return cluster.worker.id;
 	}
 

src/NodeMultiCoreCPUBase/MasterEndpoint.js

@@ -49,6 +49,7 @@ class MasterEndpoint extends JSONRPC.EndpointBase
 		this._nNextAvailablePersistentWorkerID = 0;
 
 		this._bWorkersStarted = false;
+		this._promiseStart = null;
 		this._bWatchingForUpgrade = false;
 
 		this._nMaxWorkersCount = Number.MAX_SAFE_INTEGER;
@@ -189,29 +190,51 @@ class MasterEndpoint extends JSONRPC.EndpointBase
 	 */
 	async start()
 	{
-		if(this._bWorkersStarted)
+		if(this._promiseStart)
 		{
-			throw new Error("Workers have already been started.");
+			return this._promiseStart;
 		}
-		this._bWorkersStarted = true;
 
-		this._jsonrpcServer = new JSONRPC.Server();
-
-		// By default, JSONRPC.Server rejects all requests as not authenticated and not authorized.
-		this._jsonrpcServer.addPlugin(new JSONRPC.Plugins.Server.AuthenticationSkip());
-		this._jsonrpcServer.addPlugin(new JSONRPC.Plugins.Server.AuthorizeAll());
-		this._jsonrpcServer.registerEndpoint(this);
+		this._promiseStart = new Promise(async(fnResolve, fnReject) => {
+			try
+			{
+				if(this._jsonrpcServer)
+				{
+					this._jsonrpcServer.dispose();
+				}
+				
 
-		this._bidirectionalWorkerRouter = await this._makeBidirectionalRouter();
+				this._jsonrpcServer = new JSONRPC.Server();
+		
+				// By default, JSONRPC.Server rejects all requests as not authenticated and not authorized.
+				this._jsonrpcServer.addPlugin(new JSONRPC.Plugins.Server.AuthenticationSkip());
+				this._jsonrpcServer.addPlugin(new JSONRPC.Plugins.Server.AuthorizeAll());
+				this._jsonrpcServer.registerEndpoint(this);
+		
+				this._bidirectionalWorkerRouter = await this._makeBidirectionalRouter();
+		
+				await this._configureBeforeStart();
+		
+				await this._startServices();
+		
+				for (let i = 0; i < Math.min(Math.max(os.cpus().length, 1), this.maxWorkersCount); i++)
+				{
+					this._addWorker();
+				}
 
-		await this._configureBeforeStart();
+				this._bWorkersStarted = true;
 
-		await this._startServices();
+				fnResolve();
+			}
+			catch(error)
+			{
+				this._bWorkersStarted = false;
+				this._promiseStart = false;
+				fnReject(error);
+			}
+		});
 
-		for (let i = 0; i < Math.min(Math.max(os.cpus().length, 1), this.maxWorkersCount); i++)
-		{
-			this._addWorker();
-		}
+		return this._promiseStart;
 	}
 
 
@@ -283,6 +306,28 @@ class MasterEndpoint extends JSONRPC.EndpointBase
 	 */
 	async workerServicesReady(incomingRequest, nWorkerID)
 	{
+		if(!this.objWorkerIDToState[nWorkerID])
+		{
+			console.error(`[Master] .workerServicesReady() Could not find worker ID ${nWorkerID} key in this.objWorkerIDToState to set the .ready=true. Retrying after 10 seconds sleep in case race condition.`);
+			await sleep(10 * 1000);
+
+			if(!this.objWorkerIDToState[nWorkerID])
+			{
+				console.error(`[Master] .workerServicesReady() Could not find worker ID ${nWorkerID} key in this.objWorkerIDToState to set the .ready=true. Going berserk and marking all as ready. this.objWorkerIDToState: ${JSON.stringify(this.objWorkerIDToState, undefined, "    ")}`);
+
+				for(const _nWorkerID in this.objWorkerIDToState)
+				{
+					this.objWorkerIDToState[_nWorkerID].ready = true;
+				}
+
+				return;
+			}
+			else
+			{
+				console.error(`[Master] .workerServicesReady() Found worker ID ${nWorkerID} key in this.objWorkerIDToState to set the .ready=true. This indicates a race condition exists somewhere.`);
+			}
+		}
+
 		this.objWorkerIDToState[nWorkerID].ready = true;
 	}
 

src/NodeMultiCoreCPUBase/WorkerEndpoint.js

@@ -41,6 +41,8 @@ class WorkerEndpoint extends JSONRPC.EndpointBase
 		this._promiseStart = null;
 
 		this._nPersistentWorkerID = undefined;
+
+		this._bAlreadyDelayedReadingWorkerID = false;
 	}
 
 
@@ -72,7 +74,7 @@ class WorkerEndpoint extends JSONRPC.EndpointBase
 
 
 	/**
-	 * @returns {JSONRPC.Client}
+	 * @returns {JSONRPC.Client|null}
 	 */
 	get masterClient()
 	{
@@ -96,7 +98,8 @@ class WorkerEndpoint extends JSONRPC.EndpointBase
 				this.start().catch(console.error);
 			}
 
-			throw new Error("The .masterClient property was not initialized by .start().");
+			console.error("[jsornpc-bidirectional] [WorkerEndpoint] The .masterClient property was not initialized by .start(). Returning null.", new Error().stack);
+			return null;
 		}
 
 		return this._masterClient;
@@ -168,7 +171,13 @@ class WorkerEndpoint extends JSONRPC.EndpointBase
 					throw new Error("WorkerEndpoint.start() was already called.");
 				}
 				this._bWorkerStarted = true;
-				
+
+
+				if(this._jsonrpcServer)
+				{
+					this._jsonrpcServer.dispose();
+				}
+
 
 				this._jsonrpcServer = new JSONRPC.Server();
 				this._bidirectionalWorkerRouter = await this._makeBidirectionalRouter();
@@ -194,6 +203,8 @@ class WorkerEndpoint extends JSONRPC.EndpointBase
 			}
 			catch(error)
 			{
+				this._bWorkerStarted = false;
+				this._promiseStart = false;
 				fnReject(error);
 			}
 		});

src/NodeWorkerThreadsBase/WorkerEndpoint.js

@@ -4,6 +4,8 @@ const JSONRPC = {
 	BidirectionalWorkerThreadRouter: require("../BidirectionalWorkerThreadRouter")
 };
 
+const sleep = require("sleep-promise");
+
 let Threads;
 try
 {
@@ -41,6 +43,33 @@ class WorkerEndpoint extends NodeMultiCoreCPUBase.WorkerEndpoint
 	 */
 	async _currentWorkerID()
 	{
+		// https://github.com/nodejs/node/issues/1269
+		if(
+			!this._bAlreadyDelayedReadingWorkerID
+			&& (
+				!Threads
+				|| Threads.threadId === null 
+				|| Threads.threadId === undefined
+			)
+		)
+		{
+			await sleep(2000);
+			this._bAlreadyDelayedReadingWorkerID = true;
+		}
+
+
+		if(
+			!Threads
+			|| Threads.threadId === null 
+			|| Threads.threadId === undefined
+		)
+		{
+			console.error("Threads: ", Threads);
+			console.error("Threads.threadId: ", Threads ? Threads.threadId : "");
+			console.error(`Returning 0 as Threads.threadId.`);
+			return 0;
+		}
+
 		return Threads.threadId;
 	}