thread pool well behaved under stress

Author: tysonrm

src/domain/domain-events.js

@@ -3,7 +3,7 @@
 /**
  * @param {import('./event-broker').EventBroker} broker
  */
-export function registerEvents(broker) {
+export function registerEvents (broker) {
   //
   broker.on('shutdown', signal => {
     console.log('shutdown requested')
@@ -51,6 +51,7 @@ const domainEvents = {
   poolAbort: modelName => `poolAbort_${String(modelName).toUpperCase()}`,
   badUserRoute: error => `badUserRoute error: ${error}`,
   reload: 'reload',
+  portEvent: name => `externalPortEvent_${name}`
 }
 
 export default domainEvents

src/domain/event-router.js

@@ -1,6 +1,8 @@
 'use strict'
 
 import { workerData, BroadcastChannel } from 'worker_threads'
+import domainEvents from './domain-events'
+const { portEvent } = domainEvents
 
 /**
  * Port events controls event-driven, distributed workflow and
@@ -16,6 +18,8 @@ export class PortEventRouter {
     this.models = models
     this.broker = broker
     this._localSpec = this.models.getModelSpec(workerData.poolName)
+    if (!this._localSpec)
+      this.models.getModelSpecs().find(i => i.domain === workerData.poolName)
     this._threadRemotePorts = this.threadRemotePorts()
     this._threadLocalPorts = this.threadLocalPorts()
   }
@@ -103,13 +107,34 @@ export class PortEventRouter {
   }
 
   handleBroadcastEvent (msg) {
-    if (msg?.data?.eventName) this.broker.notify(msg.data.eventName, msg.data)
+    if (msg?.data?.eventName)
+      this.broker.notify(this.internalizeEvent(msg.data.eventName), msg.data)
     else {
       console.log('missing eventName', msg.data)
       this.broker.notify('missingEventName', msg.data)
     }
   }
 
+  externalizeEvent (event) {
+    if (typeof event === 'object') {
+      return {
+        ...event,
+        eventName: portEvent(event.eventName)
+      }
+    }
+    if (typeof event === 'string') return portEvent(event)
+  }
+
+  internalizeEvent (event) {
+    if (typeof event === 'object') {
+      return {
+        ...event,
+        eventName: event.eventName.replace(portEvent(), '')
+      }
+    }
+    if (typeof event === 'string') return event.replace(portEvent(), '')
+  }
+
   /**
    * Listen for producer events coming from other thread pools and
    * fire the events in the local pool for ports to consume them.
@@ -139,7 +164,9 @@ export class PortEventRouter {
     // listen for producer events and dispatch them to local pools
     publisherPorts.forEach(port =>
       this.broker.on(port.producesEvent, event =>
-        channels.get(port.domain).postMessage(JSON.parse(JSON.stringify(event)))
+        channels
+          .get(port.domain)
+          .postMessage(JSON.parse(JSON.stringify(this.externalizeEvent(event))))
       )
     )
 
@@ -154,16 +181,18 @@ export class PortEventRouter {
       this.broker.on(port.producesEvent, event => {
         this.broker.notify('to_main', {
           ...event,
-          eventName: port.producesEvent,
+          eventName: this.externalizeEvent(port.producesEvent),
           route: 'balanceEventConsumer' // mesh routing algo
         })
       })
     })
 
-    // listen to this model's channel
-    new BroadcastChannel(workerData.poolName).onmessage = msg => {
-      console.log('onmessage', msg.data)
-      this.handleBroadcastEvent(msg)
+    if (!channels.has(this.localSpec.domain)) {
+      // listen to this model's channel
+      new BroadcastChannel(this.localSpec.domain).onmessage = msg => {
+        console.log('onmessage', msg.data)
+        this.handleBroadcastEvent(msg)
+      }
     }
   }
 }

src/domain/make-relations.js

@@ -57,7 +57,7 @@ export const relationType = {
    * @param {*} rel
    * @returns
    */
-  oneToOne(model, ds, rel) {
+  oneToOne (model, ds, rel) {
     return this.manyToOne(model, ds, rel)
   },
 
@@ -87,14 +87,14 @@ export const relationType = {
     const rds = ds.factory.getRestrictedDataSource(model.modelName)
     const relRds = ds.factory.getRestrictedDataSource(rel.modelName, {
       isCached: true,
-      ephemeral: true,
+      ephemeral: true
     })
 
     // if relRds is in the same domain but is remote, this fails
     if (rds.namespace !== relRds.namespace) return null
 
     return rds[rel.name]({ args, model, ds: relRds, relation: rel })
-  },
+  }
 }
 
 /**
@@ -108,39 +108,39 @@ const updateForeignKeys = {
    * @param {import('./index').relations[x]} relation
    * @param {import('./model-factory').Datasource} ds
    */
-  async [relationType.manyToOne.name](fromModel, toModels, relation, ds) {
+  async [relationType.manyToOne.name] (fromModel, toModels, relation, ds) {
     return fromModel.update(
       { [relation.foreignKey]: toModels[0].getId() },
       false
     )
   },
 
-  async [relationType.oneToOne.name](fromModel, toModels, relation, ds) {
+  async [relationType.oneToOne.name] (fromModel, toModels, relation, ds) {
     return this[relationType.manyToOne.name](fromModel, toModels, relation, ds)
   },
 
-  async [relationType.oneToMany.name](fromModel, toModels, relation, ds) {
+  async [relationType.oneToMany.name] (fromModel, toModels, relation, ds) {
     return Promise.all(
       toModels.map(async m => {
         const model = await ds.find(m.id || m.getId())
         return ds.save(m.id, {
           ...model,
-          [relation.foreignKey]: fromModel.getId(),
+          [relation.foreignKey]: fromModel.getId()
         })
       })
     )
   },
 
-  async [relationType.containsMany.name](fromModel, toModels, relation, ds) {
+  async [relationType.containsMany.name] (fromModel, toModels, relation, ds) {
     toModels.map(model =>
       model.update({ [relation.foreignKey]: fromModel.getId() })
     )
   },
 
-  async [relationType.custom.name](fromModel, toModels, relation, ds) {
+  async [relationType.custom.name] (fromModel, toModels, relation, ds) {
     const customFn = fromModel[`${relation}UpdateForeignKeys`]
     if (customFn === 'function') customFn(toModels, relation, ds)
-  },
+  }
 }
 
 /**
@@ -151,7 +151,7 @@ const updateForeignKeys = {
  * @param {import('./datasource').default} ds
  * @returns
  */
-async function createModels(args, fromModel, relation, ds) {
+async function createModels (args, fromModel, relation, ds) {
   if (args.length > 0) {
     const { UseCaseService } = require('.')
     const service = UseCaseService(relation.modelName)
@@ -162,18 +162,18 @@ async function createModels(args, fromModel, relation, ds) {
   }
 }
 
-async function createNewModels(args, rel, datasource) {
+async function createNewModels (args, rel, datasource) {
   if (args.length > 0 && rel.type !== 'custom') {
     // fetch the local ds and create the models
     const ds = datasource.factory.getDataSource(rel.modelName)
     return {
       yes: true,
-      create: async () => await createModels(args, this, rel, ds),
+      create: async () => await createModels(args, this, rel, ds)
     }
   }
   return {
     yes: false,
-    create: () => null,
+    create: () => null
   }
 }
 
@@ -188,7 +188,7 @@ async function createNewModels(args, rel, datasource) {
  * @param {import("./event-broker").EventBroker} broker
  * @returns {Promise<import(".").Event>} source model
  */
-export function requireRemoteObject(model, relation, broker, ...args) {
+export function requireRemoteObject (model, relation, broker, ...args) {
   const request = internalCacheRequest(relation.modelName)
   const response = internalCacheResponse(relation.modelName)
   const name = (model ? model.getName() : relation.modelName).toUpperCase()
@@ -207,7 +207,7 @@ export function requireRemoteObject(model, relation, broker, ...args) {
     modelId: id,
     relation,
     model,
-    args,
+    args
   }
 
   return new Promise(async function (resolve) {
@@ -223,7 +223,7 @@ export function requireRemoteObject(model, relation, broker, ...args) {
  * @param {import('.').relations[x]} relation
  * @returns {boolean}
  */
-function isRelatedModelLocal(relation) {
+function isRelatedModelLocal (relation) {
   return require('.')
     .default.getModelSpecs()
     .filter(spec => !spec.isCached)
@@ -236,7 +236,7 @@ function isRelatedModelLocal(relation) {
  * @param {import("./index").relations} relations
  * @param {import("./datasource").default} datasource
  */
-export default function makeRelations(relations, datasource, broker) {
+export default function makeRelations (relations, datasource, broker) {
   if (Object.getOwnPropertyNames(relations).length < 1) return
 
   return Object.keys(relations)
@@ -246,7 +246,7 @@ export default function makeRelations(relations, datasource, broker) {
       const rel = {
         ...relations[relation],
         modelName: relModelName,
-        name: relation,
+        name: relation
       }
 
       try {
@@ -258,18 +258,24 @@ export default function makeRelations(relations, datasource, broker) {
 
         return {
           // the relation function
-          async [relation](...args) {
+          async [relation] (...args) {
             const local = isRelatedModelLocal(relModelName)
             if (!local) await importModelCache(relModelName)
 
+            console.log({ relModelName })
+            
             if (local) {
               const result = createNewModels(args, rel, datasource)
               if (result.yes) return result.create()
             }
 
+            const importedSpec = require('.').default.getModelSpec(relModelName)
             // If object is remote, we should have its code by now.
             // Recreate its datasource, including any customization
-            const ds = datasource.factory.getDataSource(relModelName)
+            const ds = datasource.factory.getDataSource(
+              relModelName,
+              importedSpec.domain
+            )
 
             const models = await relationType[rel.type](this, ds, rel, args)
 
@@ -289,7 +295,7 @@ export default function makeRelations(relations, datasource, broker) {
             }
 
             return models
-          },
+          }
         }
       } catch (error) {
         console.error({ fn: makeRelations.name, error })

src/domain/thread-pool.js

@@ -440,27 +440,10 @@ export class ThreadPool extends EventEmitter {
 
   incrementJobsQueued () {
     this.jobsQueued++
-    if (
-      this.jobQueueRate() > this.jobQueueThreshold() &&
-      this.jobsRequested > 20
-    ) {
-      console.warn('job queue threshold exceeded')
-      this.emit(this.growPool.name)
-    }
     return this
   }
 
-  async growPool () {
-    if (this.capacityAvailable()) {
-      const thread = await this.startThread()
-      if (this.capacityAvailable())
-        broker.on(this.growPool.name, this.growPool, {
-          singleton: true,
-          once: true
-        })
-      return thread
-    }
-  }
+  async growPool () {}
 
   jobQueueRate () {
     return Math.round((this.jobsQueued / this.jobsRequested) * 100)
@@ -473,13 +456,6 @@ export class ThreadPool extends EventEmitter {
   jobTime (millisec) {
     this.totJobTime += millisec
     this.avgJobTime = Math.round(this.totJobTime / this.jobsRequested)
-    if (
-      this.avgJobTime > this.jobDurationThreshold() &&
-      this.jobsRequested > 20
-    ) {
-      this.emit(this.growPool.name)
-      console.log('job duration threshold exceeded', this)
-    }
     return this
   }
 
@@ -493,10 +469,6 @@ export class ThreadPool extends EventEmitter {
 
   incrementErrorCount () {
     this.errors++
-    if (this.errorRate() > this.errorRateThreshold() && this.errors > 10) {
-      this.emit(this.growPool.name)
-      console.log('job duration threshold exceeded', this)
-    }
     return this
   }
 
@@ -597,10 +569,8 @@ export class ThreadPool extends EventEmitter {
       return thread
     }
     // none free, try to allocate
-    const thread = this.allocate()
-    // pop this one
-    if (thread) this.freeThreads.pop()
-    // dont return tyhread, let queued jobs go
+    this.allocate()
+    // dont return thread, let queued jobs go
   }
 
   checkin (thread) {

src/domain/use-cases/load-models.js

@@ -3,7 +3,6 @@
 import { isMainThread } from 'worker_threads'
 import { Writable, Transform } from 'node:stream'
 import { hostConfig } from '../../config'
-import e from 'express'
 
 function nextModelFn (port, mf) {
   mf