Merge pull request #282 from module-federation/release-0-1-0

thread limit and scaling fixed and optimal

Author: tysonrm

src/domain/event-router.js

@@ -1,7 +1,8 @@
 'use strict'
 
-import { workerData, BroadcastChannel } from 'worker_threads'
+import { workerData } from 'worker_threads'
 import domainEvents from './domain-events'
+
 const { portEvent } = domainEvents
 
 /**
@@ -18,8 +19,10 @@ export class PortEventRouter {
     this.models = models
     this.broker = broker
     this._localSpec = this.models.getModelSpec(workerData.poolName)
+
     if (!this._localSpec)
       this.models.getModelSpecs().find(s => s.domain === workerData.poolName)
+      
     this._threadRemotePorts = this.threadRemotePorts()
     this._threadLocalPorts = this.threadLocalPorts()
   }
@@ -106,35 +109,17 @@ export class PortEventRouter {
     )
   }
 
-  handleBroadcastEvent (msg) {
-    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) {
+  externalizeEvent (event, route = 'broadcast') {
     if (typeof event === 'object') {
       return {
         ...event,
-        eventName: portEvent(event.eventName)
+        eventName: portEvent(event.eventName),
+        route
       }
     }
     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.
@@ -143,56 +128,33 @@ export class PortEventRouter {
    * dispatch them to pools that consume them.
    */
   listen () {
-    const services = new Set()
-    const channels = new Map()
     const subscriberPorts = this.subscriberPorts()
     const publisherPorts = this.publisherPorts()
     const unhandledPorts = this.unhandledPorts()
 
-    publisherPorts.forEach(port => services.add(port.domain))
-    subscriberPorts.forEach(port => services.add(port.domain))
-
-    services.forEach(service =>
-      channels.set(service, new BroadcastChannel(service))
-    )
-
     console.log('publisherPorts', publisherPorts)
     console.log('subscriberPorts', subscriberPorts)
     console.log('unhandledPorts', unhandledPorts)
-    console.log('channels', channels)
 
     // 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(this.externalizeEvent(event))))
+        this.broker.notify('broadcast', this.externalizeEvent(event))
       )
     )
 
     // listen for incoming consumer events from local pools
-    subscriberPorts.forEach(port => {
-      channels.get(port.domain).onmessage = msg =>
-        this.handleBroadcastEvent(msg)
-    })
+    subscriberPorts.forEach(port =>
+      this.broker.on(this.externalizeEvent(port.consumesEvent), event =>
+        this.notify(port.consumesEvent, event)
+      )
+    )
 
     // dispatch events not handled by local pools to the service mesh
-    unhandledPorts.forEach(port => {
-      this.broker.on(port.producesEvent, event => {
-        this.broker.notify('to_main', {
-          ...event,
-          eventName: this.externalizeEvent(port.producesEvent),
-          route: 'balanceEventConsumer' // mesh routing algo
-        })
-      })
-    })
-
-    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)
-      }
-    }
+    unhandledPorts.forEach(port =>
+      this.broker.on(port.producesEvent, event =>
+        this.broker.notify('to_main', this.externalizeEvent(event))
+      )
+    )
   }
 }

src/domain/thread-pool.js

@@ -553,14 +553,13 @@ export class ThreadPool extends EventEmitter {
   /**
    * Spin up a new thread if needed and available.
    */
-  async allocate (cb = null) {
+  async allocate () {
     if (this.poolCanGrow()) {
       console.debug('allocating thread')
       const thread = await this.startThread()
       if (!thread) {
         throw new Error('cannot allocate thread')
       }
-      if (cb) cb(thread)
       return this.freeThreads.shift()
     }
   }
@@ -572,25 +571,22 @@ export class ThreadPool extends EventEmitter {
   }
 
   async checkout () {
+    if (this.threads.length >= this.maxThreads) {
+      console.warn('no threads available')
+      return
+    }
     const thread = this.freeThreads.shift()
     if (thread) {
-      console.debug(
-        `thread checked out, total in use now ${this.threadsInUse()}`
-      )
+      console.debug(`checkout: threads in use now ${this.threadsInUse()}`)
       return thread
     }
-    if (this.threads.length == this.maxThreads) return
-    if (this.threads.length > this.maxThreads)
-      throw new Error('too many threads')
     return this.allocate()
   }
 
   checkin (thread) {
     if (thread) {
       this.freeThreads.push(thread)
-      console.debug(
-        `thread checked in, total in use now ${this.threadsInUse()}`
-      )
+      console.debug(`checkin: threads in use now ${this.threadsInUse()}`)
     }
   }