fix thread limit

Author: tysonrm

src/domain/make-relations.js

@@ -12,6 +12,14 @@ const { internalCacheRequest, internalCacheResponse, externalCacheRequest } =
 
 const maxwait = process.env.REMOTE_OBJECT_MAXWAIT || 1000
 
+class RelError extends Error {
+  constructor (error, code, fn) {
+    super(error)
+    this.code = code
+    console.log({ fn, code, error })
+  }
+}
+
 export const relationType = {
   /**
    * Search memory and external storage
@@ -108,7 +116,7 @@ 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] (fromModels, toModels, relation, ds) {
     return fromModel.update(
       { [relation.foreignKey]: toModels[0].getId() },
       false
@@ -228,7 +236,7 @@ function isRelatedModelLocal (relation) {
     .default.getModelSpecs()
     .filter(spec => !spec.isCached)
     .map(spec => spec.modelName)
-    .includes(relation.modelName)
+    .includes(relation.modelName.toUpperCase())
 }
 
 /**
@@ -259,26 +267,25 @@ export default function makeRelations (relations, datasource, broker) {
         return {
           // the relation function
           async [relation] (...args) {
+            const createNew = args?.length > 0 ? true : false
             const local = isRelatedModelLocal(relModelName)
             if (!local) await importModelCache(relModelName)
 
-            console.log({ relModelName })
-            
-            if (local) {
+            if (local && createNew) {
               const result = createNewModels(args, rel, datasource)
               if (result.yes) return result.create()
+              throw new Error('cannot create new models')
             }
 
-            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,
-              importedSpec.domain
-            )
+            const ds = datasource.factory.getDataSource(relModelName)
+
+            if (!ds) throw new RelError(`missing ds for ${relModelName}`, 500)
 
+            // call the relation's implementation
             const models = await relationType[rel.type](this, ds, rel, args)
 
+            // still failing, try service mesh
             if (!models || models.length < 1) {
               // find remote instance(s)
               const event = await requireRemoteObject(
@@ -288,7 +295,8 @@ export default function makeRelations (relations, datasource, broker) {
                 ...args
               )
 
-              if (args.length > 0)
+              if (createNew)
+                // args mean models were created and need new fk's
                 updateForeignKeys[rel.type](this, event.model, rel, ds)
 
               return await relationType[rel.type](this, ds, rel, args)

src/domain/thread-pool.js

@@ -9,6 +9,7 @@ import domainEvents from './domain-events'
 import assert from 'assert'
 import path from 'path'
 import os from 'os'
+import { setServers } from 'dns/promises'
 
 const { poolOpen, poolClose, poolDrain, poolAbort } = domainEvents
 const broker = EventBrokerFactory.getInstance()
@@ -159,6 +160,7 @@ export class ThreadPool extends EventEmitter {
     this.aborting = false
     this.jobsRequested = this.jobsQueued = 0
     this.broadcastChannel = options.broadcast
+    this.updateLocks()
 
     if (options?.preload) {
       console.info('preload enabled for', this.name)
@@ -220,6 +222,7 @@ export class ThreadPool extends EventEmitter {
               console.warn('shutdown timeout')
               resolve(await worker.terminate())
             }, 6000)
+
             worker.once('exit', () => {
               clearTimeout(timerId)
               console.log('orderly shutdown')
@@ -539,33 +542,33 @@ export class ThreadPool extends EventEmitter {
 
   threadsInUse () {
     const diff = this.threads.length - this.freeThreads.length
-    assert.ok(
-      diff >= 0 && diff <= this.maxThreads,
-      'thread mgmt issue: in use: ' +
-        diff +
-        ', total:' +
-        this.threads.length +
-        ', free:' +
-        this.freeThreads.length +
-        ', min:' +
-        this.minThreads +
-        ', max:' +
-        this.maxThreads
-    )
+    assert.ok(diff >= 0 && diff <= this.maxThreads, 'thread mgmt issue')
     return diff
   }
 
+  updateLocks () {
+    this.locks = []
+    this.threads.forEach(t => this.locks.push(t.id))
+  }
+
+  canCheckout () {
+    return true
+    // return this.threads.some(t => t.id === this.locks.pop())
+  }
+
   checkout () {
-    if (this.freeThreads.length > 0) {
-      const thread = this.freeThreads.shift()
-      console.debug(
-        `thread checked out, total in use now ${this.threadsInUse()}`
-      )
-      return thread
-    }
-    // none free, try to allocate
-    this.allocate()
-    // dont return thread, let queued jobs go
+    if (!this.canCheckout()) return
+
+    try {
+      if (this.freeThreads.length > 0) {
+        const thread = this.freeThreads.shift()
+        console.debug(
+          `thread checked out, total in use now ${this.threadsInUse()}`
+        )
+        return thread
+      }
+      this.allocate()
+    } catch (err) {}
   }
 
   checkin (thread) {
@@ -575,7 +578,6 @@ export class ThreadPool extends EventEmitter {
         `thread checked in, total in use now ${this.threadsInUse()}`
       )
     }
-    return this
   }
 
   enqueue (job) {

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

@@ -1,11 +1,11 @@
 'use strict'
 
 import { isMainThread } from 'worker_threads'
-import { Writable, Transform } from 'node:stream'
+import { Writable } from 'node:stream'
 import { hostConfig } from '../../config'
 
-function nextModelFn (port, mf) {
-  mf
+function nextModelFn (port, modelFactory) {
+  modelFactory
     .getModelSpecs()
     .filter(spec => spec.ports)
     .map(spec =>
@@ -15,20 +15,21 @@ function nextModelFn (port, mf) {
     )[0]
 }
 
-function startWorkflow (model, mf) {
+function startWorkflow (model, modelFactory) {
   const history = model.getPortFlow()
   const ports = model.getPorts()
 
   if (history?.length > 0 && !model.compensate) {
     const lastPort = history.length - 1
     const nextPort = ports[history[lastPort]]?.producesEvent
-    const nextModel = nextModelFn(nextPort, mf)
+    const nextModel = nextModelFn(nextPort, modelFactory)
     if (nextPort) model.emit(nextPort, nextModel)
   }
 }
 
 /**
- * deserialize and unmarshall from stream.
+ * Add models to cache and run workflow.
+ *
  * @typedef {import('..').Model}
  * @param {{
  *  models:import('../model-factory').ModelFactory,
@@ -38,7 +39,7 @@ function startWorkflow (model, mf) {
  * }} options
  * @returns {function():Promise<void>}
  */
-export default async function ({ modelName, repository, models }) {
+export default async function ({ repository, models }) {
   if (isMainThread) return
 
   const cacheLoadDisabled =
@@ -47,39 +48,17 @@ export default async function ({ modelName, repository, models }) {
 
   if (cacheLoadDisabled) return
 
-  const resumeWorkflowDisabled =
-    /false/i.test(process.env.RESUMEWORKFLOW) ||
-    hostConfig.adapters.resumeWorkflow === false
-
-  console.log('loading cache ', modelName)
-
-  // const serializers = models.getModelSpec(modelName).serializers
-
-  // const deserializers = serializers
-  //   ? serializers.filter(s => !s.disabled && s.on === 'deserialize')
-  //   : null
-
-  // if (deserializers) deserializers.forEach(des => Serializer.addSerializer(des))
-
   const loadCache = new Writable({
     objectMode: true,
 
     write (chunk, _, next) {
-      if (chunk?.id) repository.saveSync(chunk.id, chunk)
-      next()
-    }
-  })
-
-  const resumeWorkflow = new Transform({
-    objectMode: true,
-
-    transform (chunk, _, next) {
-      startWorkflow(chunk, models)
+      if (chunk?.id) {
+        repository.saveSync(chunk.id, chunk)
+        startWorkflow(chunk, models)
+      }
       next()
     }
   })
 
-  if (resumeWorkflowDisabled)
-    repository.list({ writable: loadCache, transform: resumeWorkflow })
-  else repository.list({ writable: loadCache })
+  repository.list({ writable: loadCache })
 }

start.sh

@@ -1,3 +1,4 @@
+export LOADCACHE=false
 cd ../aegis-app
 nohup node repo.js &
 nohup node repo.js 8001 cache &