WIP: Sending commands from the worker back to the master

Author: FranklinWaller

examples/simple-url-call/src/flux/mod.rs

@@ -9,11 +9,11 @@ pub fn get_call_result_size() -> u32 {
 }
 
 pub fn call() {
-    let mut result_data: Vec<u8> = Vec::with_capacity(get_call_result_size() as usize);
-
     let call_data = "This is some call data";
-
+    
     unsafe {
         raw::call(call_data.as_ptr(), call_data.len() as i32);
     }
+
+    let mut result_data: Vec<u8> = Vec::with_capacity(get_call_result_size() as usize);
 }

src/Process.worker.ts

@@ -1,6 +1,6 @@
 import { VmContext } from "./models/Context";
 import { MessageType } from "./models/WorkerMessage";
-import { executeWasm } from "./services/WasmService";
+import { executeWasm } from "./services/VirtualMachineService";
 import { extractMessageFromEvent, workerAddEventListener, selfPostMessage } from "./services/WorkerService";
 
 workerAddEventListener('message', async (event: MessageEvent) => {

src/ProcessMethods.ts

@@ -1,10 +1,42 @@
+import fetch from 'node-fetch';
 import { MethodCallMessage } from "./models/WorkerMessage";
+import { sendBufferLengthToWorker, storeAndNotify } from "./services/WorkerService";
 
 class ProcessMethods {
+    private result?: Buffer;
+
     constructor(private notifierBuffer: SharedArrayBuffer) {}
 
+    private async call(args: any[]) {
+        try {
+            const response = await fetch('https://pokeapi.co/api/v2/pokemon/ditto');
+            const text = await response.text();
+
+            const result = JSON.stringify({
+                type: 'Result',
+                status: response.status,
+                result: text,
+            });
+
+            this.result = Buffer.from(result);
+            sendBufferLengthToWorker(this.notifierBuffer, this.result);
+        } catch(error) {
+            console.error('[call]', error);
+        }
+    }
+
+    private async writeResultToBuffer(args: any[]) {
+        const u8WriteBuffer = new Uint8Array(args[0] as SharedArrayBuffer);
+        u8WriteBuffer.set(this.result ?? Buffer.from([]));
+        storeAndNotify(this.notifierBuffer, 0, 1);
+    }
+
     executeMethodCallMessage(message: MethodCallMessage) {
-        
+        if (message.value.methodName === 'call') {
+            this.call(message.value.args);
+        } else if (message.value.methodName === 'writeResultToBuffer') {
+            this.writeResultToBuffer(message.value.args);
+        }
     }
 }
 

src/models/Context.ts

@@ -21,13 +21,15 @@ export class VmContext {
     gasLimit: string;
     args: string[];
     env: ProcessEnvOptions['env'];
+    notifierBuffer: SharedArrayBuffer;
 
-    constructor(context: Context) {
+    constructor(context: WorkerContext) {
         this.gasUsed = new Big(context.startingGas ?? '0');
         this.binary = context.binary;
         this.randomSeed = context.randomSeed;
         this.gasLimit = context.gasLimit;
         this.args = context.args;
+        this.notifierBuffer = context.notifierBuffer;
         this.env = {
             ...context.env,
             GAS_LIMIT: this.gasLimit,

src/services/HexService.ts

@@ -0,0 +1,22 @@
+/**
+ * Converts numbers to a hex format
+ * Makes sure that small numbers are in 2 byte pairs
+ *
+ * @export
+ * @param {number} num
+ * @returns
+ */
+export function numberToHex(num: number, padBytes: number = 2): string {
+    let result = num.toString(16);
+
+    if ((result.length % 2) > 0) {
+        result = '0' + result;
+    }
+
+    if (result.length < padBytes) {
+        const missingAmountOfBytes = new Array(padBytes - result.length).fill('0');
+        result = missingAmountOfBytes.join('') + result;
+    }
+
+    return result;
+}

src/services/WasmService.ts renamed to src/services/VirtualMachineService.ts

@@ -6,7 +6,6 @@ import { VmContext } from "../models/Context";
 import { ExecuteResult } from '../models/ExecuteResult';
 import { generateFixedBuffer } from './CryptoService';
 import WasmImports from './WasmImports';
-import createWasmImports from './WasmImports';
 
 const metering = require('wasm-metering');
 
@@ -23,10 +22,6 @@ export async function executeWasm(context: VmContext): Promise<ExecuteResult> {
 
     try {
         const random = alea(context.randomSeed);
-        const memory = new WebAssembly.Memory({
-            initial: 5000,
-        });
-
         const wasi = new WASI({
             args: context.args,
             env: context.env,

src/services/WasmImports.ts

@@ -1,19 +1,24 @@
 import { VmContext } from '../models/Context';
+import { callMethodOnMainThread } from './WorkerService';
 
 export default class WasmImports {
     private memory?: WebAssembly.Memory;
     private lastCallResult: {} = {};
 
     constructor(
         private vmContext: VmContext, 
-        private wasiImports: Record<string, Record<string, Function>>
+        private wasiImports: Record<string, Record<string, Function>>,
     ) {}
 
     private call(callDataOffset: number, callDataLength: number) {
         if (!this.memory) throw new ReferenceError('ERR_NO_MEMORY');
         const callDataBytes = new Uint8Array(this.memory.buffer, callDataOffset, callDataLength);
-        const callDataString = new TextDecoder().decode(callDataBytes);;
-        
+        const callDataString = new TextDecoder().decode(callDataBytes);
+        const resultBuffer = callMethodOnMainThread(this.vmContext.notifierBuffer, 'call', [callDataString]);   
+
+        if (resultBuffer) {
+            console.log('[] resultBuffer -> ', resultBuffer.toString());
+        }
     }
 
     public setMemory(memory: WebAssembly.Memory) {

src/services/WorkerService.ts

@@ -1,5 +1,6 @@
 import { MessageType, RequestMessage } from "../models/WorkerMessage";
 import { isNodeJs } from "./EnvironmentService";
+import { numberToHex } from "./HexService";
 
 export function createWorker(url: URL, options?: any): Worker {
     let WorkerConstructor: any = null;
@@ -90,7 +91,8 @@ export function callMethodOnMainThread(notifierBuffer: SharedArrayBuffer, method
 
     // The main thread only writes the length of the result inside the notifierbuffer
     // We create a big enough shared buffer so the main thread can write the full result
-    const bufferLength = parseInt(Buffer.from(notifierBuffer).toString('hex'), 16);
+    const writtenToBuffer = Buffer.from(notifierBuffer).slice(0); // slice 1 of since that is our notify index
+    const bufferLength = parseInt(writtenToBuffer.toString('hex'), 16);
 
     if (bufferLength === 0) {
         return null;
@@ -110,4 +112,12 @@ export function callMethodOnMainThread(notifierBuffer: SharedArrayBuffer, method
     resetSharedBuffer(notifierBuffer, 0);
 
     return Buffer.from(valueBuffer);
+}
+
+export function sendBufferLengthToWorker(notifierBuffer: SharedArrayBuffer, value?: Buffer) {
+    const lengthInHex = numberToHex(value?.length ?? 0, (notifierBuffer.byteLength * 2));
+    const u8NotifierBuffer = new Uint8Array(notifierBuffer);
+    u8NotifierBuffer.set(Buffer.from(lengthInHex, 'hex'), 0);
+
+    storeAndNotify(notifierBuffer, 0, 1);
 }