add support for inbound ports to wasm

Author: tysonrm

nohup.out

@@ -0,0 +1,6 @@
+yarn run v1.22.19
+error Command "startLocalRepo" not found.
+info Visit https://yarnpkg.com/en/docs/cli/run for documentation about this command.
+yarn run v1.22.19
+error Command "startLocalRepo" not found.
+info Visit https://yarnpkg.com/en/docs/cli/run for documentation about this command.

src/adapters/webassembly/index.js

@@ -4,7 +4,7 @@ import {
   wrapWasmAdapter,
   wrapWasmModelSpec,
   wrapWasmService
-} from './wasm-wrappers'
+} from './wasm-wrap'
 
 export * from './wasm-interop'
 export * from './wasm-import'
@@ -22,7 +22,7 @@ export const adapterTypes = {
 export const wasmAdapters = {
   /**
    * @param {WebAssembly.Exports} wasm
-   * @returns {import('./wasm-wrappers').ModelSpecification}
+   * @returns {import('./wasm-wrap').ModelSpecification}
    */
   [adapterTypes.model]: wasm => wrapWasmModelSpec(wasm),
   /**

src/adapters/webassembly/repo-client.js

@@ -40,10 +40,10 @@ export const RepoClient = {
           })
         })
         .then(function (rest) {
-          const buf = Buffer.from(rest.data.content, 'base64')
+          const buf = Buffer.from(rest.data.content)
           resolve({
             toString: () => buf.toString('utf-8'),
-            asBase64Buffer: () => buf,
+            toArrayBuffer: () => buf,
             toUint16Array: () =>
               new Uint16Array(
                 buf.buffer,

src/adapters/webassembly/wasm-import.js

@@ -1,7 +1,6 @@
-import { callbackify } from 'node:util'
 import { wasmAdapters } from '.'
-
 import { EventBrokerFactory } from '../../../lib/domain'
+import { RepoClient } from './repo-client'
 
 /**@type {import('../../domain/event-broker').EventBroker} */
 const broker = EventBrokerFactory.getInstance()
@@ -366,19 +365,19 @@ exports.importWebAssembly = function (remoteEntry) {
     return adaptedExports
   }
 
-  async function compileStream (url) {
+  async function compileStream (remoteEntry) {
     try {
       return await globalThis.WebAssembly.compileStreaming(
-        globalThis.fetch(url)
+        globalThis.fetch(remoteEntry.url)
       )
     } catch {
       return globalThis.WebAssembly.compile(
-        await (await import('node:fs/promises')).readFile(url)
+        (await RepoClient.fetch(remoteEntry)).toArrayBuffer()
       )
     }
   }
 
-  const initWasm = async () => instantiate(await compileStream(remoteEntry.url))
+  const initWasm = async () => instantiate(await compileStream(remoteEntry))
 
   return initWasm().then(wasmExports =>
     wasmAdapters[remoteEntry.type](wasmExports)

src/adapters/webassembly/wasm-interop.js

@@ -7,11 +7,11 @@
 
 /**
  * WASM interop functions
- * - find exported functions
- * - call exported functions
- * - import command configuration
+ * - call any exported function
+ *   (no js glue code required)
  * - import port configuration
- * - decode memory addresses
+ * - import inbound port functions
+ * - import commands
  * @param {WebAssembly.Exports} wasmExports
  * @returns adapter functions
  */
@@ -29,11 +29,23 @@ exports.WasmInterop = function (wasmExports) {
     _exports
   } = wasmExports
 
+  function parse (v) {
+    return !isNaN(parseInt(v))
+      ? parseInt(v)
+      : !isNaN(parseFloat(v))
+      ? parseFloat(v)
+      : /true/i.test(v)
+      ? true
+      : /false/i.test(v)
+      ? false
+      : v
+  }
+
   /**
    *
    * @param {string} fn function name
    * @param {string[][]} kv key-value pairs
-   * @returns
+   * @returns {object}
    */
   function lift (fn, kv) {
     return liftArray(
@@ -46,14 +58,14 @@ exports.WasmInterop = function (wasmExports) {
       2,
       _exports[fn](kv) >>> 0
     )
-      .map(([k, v]) => ({ [k]: v }))
+      .map(([k, v]) => ({ [k]: parse(v) }))
       .reduce((a, b) => ({ ...a, ...b }))
   }
 
   /**
    *
    * @param {string[][]} kv key-value pairs in a 2 dimensional string array
-   * @returns
+   * @returns {string[][]}
    */
   function lower (kv) {
     return (
@@ -63,7 +75,7 @@ exports.WasmInterop = function (wasmExports) {
             pointer,
             lowerArray(
               (pointer, value) => {
-                store_ref(pointer, lowerString(value?.toString()) || notnull())
+                store_ref(pointer, lowerString(value) || notnull())
               },
               4,
               2,
@@ -78,10 +90,15 @@ exports.WasmInterop = function (wasmExports) {
     )
   }
 
-  function cleanse (obj) {
-    return Object.entries(obj)
-      .filter(([k, v]) => ['string', 'number', 'boolean'].includes(typeof v))
-      .map(([k, v]) => [k, v.toString()])
+  function clean (obj) {
+    const convert = obj =>
+      Object.entries(obj)
+        .filter(([k, v]) => ['string', 'number', 'boolean'].includes(typeof v))
+        .map(([k, v]) => [k, v.toString()])
+
+    // handle custom port format
+    if (obj.port && obj.args) return convert(obj.args)
+    return convert(obj)
   }
 
   /**
@@ -98,9 +115,7 @@ exports.WasmInterop = function (wasmExports) {
    * @returns {object|number} object
    */
   function callWasmFunction (fn, obj) {
-    const props = cleanse(obj)
-    const kv = lower(props)
-    return lift(fn, kv)
+    return lift(fn, lower(clean(obj)))
   }
 
   return Object.freeze({
@@ -135,7 +150,6 @@ exports.WasmInterop = function (wasmExports) {
      * wasm function.
      */
     importWasmPorts () {
-      /** @type {import("../../domain").ports} */
       const ports = getPorts()
       return Object.keys(ports)
         .map(port => {
@@ -149,32 +163,34 @@ exports.WasmInterop = function (wasmExports) {
             inbound
           ] = ports[port].split(',')
           return {
-            /** @type {import("../../domain").ports[x]} */
             [port]: {
               service,
               type,
               consumesEvent,
               producesEvent,
               callback: data => callWasmFunction(callback, data),
               undo: data => callWasmFunction(undo, data),
-              inbound: function inbound (port, args, id) {
+              inbound (port, args, id) {
                 callWasmFunction(inbound, { port, ...args, id })
               }
             }
           }
         })
-        .reduce((p, c) => ({ ...p, ...c }))
+        .reduce((p, c) => ({ ...p, ...c }), {})
     },
 
+    /**
+     *
+     * @returns {{[x: string]:(x) => any}}
+     */
     importWasmPortFunctions () {
-      const ports = getPorts()
-      return Object.values(ports)
-        .filter(v => v.inbound)
-        .reduce((a, b) => [...a, ...b], [])
+      return Object.entries(getPorts())
+        .map(([k, v]) => [k, v.split(',')[1]])
+        .filter(([k, v]) => v === 'inbound')
+        .map(([k, v]) => ({ [k]: x => callWasmFunction(k, x) }))
+        .reduce((a, b) => ({ ...a, ...b }), {})
     },
 
-    constructObject (ptr) {
-      return constructObject(ptr, false)
-    }
+    callWasmFunction
   })
 }

src/adapters/webassembly/wasm-wrappers.js src/adapters/webassembly/wasm-wrap.jssrc/adapters/webassembly/wasm-wrappers.js renamed to src/adapters/webassembly/wasm-wrap.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const { dependencies } = require('webpack')
+const exports = require('webpack')
 const { WasmInterop } = require('./wasm-interop')
 
 /**@typedef {import("../../domain").ModelSpecification} ModelSpecification */
@@ -57,31 +57,33 @@ exports.wrapWasmModelSpec = function (wasmExports) {
 
 /**
  *
- * @param {WebAssembly.Instance} instance
+ * @param {WebAssembly.Exports} exports
  * @returns {Adapter}
  */
-exports.wrapWasmAdapter = function (instance) {
-  const { invoke } = instance.exports
-  const interop = WasmInterop(instance)
+exports.wrapWasmAdapter = function (exports) {
+  const interop = WasmInterop(exports)
 
-  return function (service) {
-    return async function (options) {
-      const { model } = options
-      const adapter = interop.callWasmFunction(invoke, model)
-
-      if (service) {
-        interop.callWasmFunction(service[adapter.serviceFn], model)
-      }
-    }
-  }
+  return Object.keys(exports)
+    .filter(k => typeof exports[k] === 'function' && /adapter/i.test(k))
+    .map(k => ({
+      [k.replace('adapter', '')]: service => (model, args = []) =>
+        service
+          ? interop.callWasmFunction(service[k], args[0] || model)
+          : interop.callWasmFunction(k, args[0] || model)
+    }))
+    .reduce((a, b) => ({ ...a, ...b }))
 }
 
 /**
  *
- * @param {WebAssembly.Instance} instance
- * @returns {Service}ww w
+ * @param {WebAssembly.Exports} exports
+ * @returns {Service}
  */
-exports.wrapWasmService = function (instance) {
-  const { makeService } = instance.exports
-  return WasmInterop(instance).callWasmFunction(makeService)
+exports.wrapWasmService = function (exports) {
+  return {
+    [exports.getName()]: Object.keys(exports)
+      .filter(k => typeof exports[k] === 'function' && /service/i.test(k))
+      .map(k => ({ [k]: x => callWasmFunction(k, x) }))
+      .reduce((a, b) => ({ ...a, ...b }))
+  }
 }

src/domain/circuit-breaker.js

@@ -38,9 +38,6 @@ const DefaultThreshold = {
  */
 const logs = new Map()
 
-/** @type {{[x: string]: number[]}} */
-let counters = {}
-
 /**
  *
  * @param {*} id
@@ -255,13 +252,12 @@ const Switch = function (id, thresholds) {
         error
       })
     },
-
-    incrementInvocationCounter () {
-      if (!counters[id]) counters[id] = [Date.Now()]
-      else counters[id].push(Date.now())
-      counters[id].filter(time => time > Date.now() - 5000).length > 999
-    },
-
+    /**
+     *
+     * @param {*} error
+     * @param {*} arg
+     * @returns
+     */
     async fallbackFn (error, arg) {
       try {
         return getThreshold(error, thresholds).fallbackFn.apply(this, arg)

src/domain/datasource-factory.js

@@ -128,7 +128,7 @@ const DsCoreExtensions = superclass =>
                 ModelFactory.loadModel(broker, this, model, this.name)
               )
       } catch (error) {
-        console.error({ fn: this.list.name })
+        console.error({ fn: this.list.name, error })
         throw error
       }
     }
@@ -254,8 +254,7 @@ const DataSourceFactory = (() => {
 
     if (!options.ephemeral) dataSources.set(name, newDs)
 
-    //debug &&
-    console.debug({ newDs })
+    debug && console.debug({ newDs })
     return newDs
   }
 

src/domain/thread-pool.js

@@ -349,8 +349,8 @@ export class ThreadPool extends EventEmitter {
   }
 
   async stopThreads (reason) {
-    for (const thread of this.threads)
-      console.warn(await this.stopThread(thread, reason))
+    for await (const thread of this.threads)
+      console.warn(this.stopThread(thread, reason))
     this.freeThreads.splice(0, this.freeThreads.length)
     return this
   }