Use msgpack to encode & decode data for channel transfer (#204)

* Use msgpack to encode data for channel transfer

* Treat ArrayBuffer as an Array in toWebRData

Improves performance when working with `ArrayBuffer`s by avoiding
the incorrect creation of a very large array of names containing
stringified indexes.

src/package.json

@@ -72,6 +72,7 @@
     "node": ">=14.0.0"
   },
   "dependencies": {
+    "@msgpack/msgpack": "^2.8.0",
     "@types/emscripten": "^1.39.6",
     "jquery": "^3.6.0",
     "jstree": "^3.3.12",

src/webR/chan/channel-service.ts

@@ -5,9 +5,8 @@ import {
   Response,
   Request,
   newResponse,
-  encodeData,
-  decodeData,
 } from './message';
+import { encode, decode } from '@msgpack/msgpack';
 import { Endpoint } from './task-common';
 import { ChannelMain, ChannelWorker } from './channel';
 import { ChannelType } from './channel-common';
@@ -249,8 +248,8 @@ export class ServiceWorkerChannelWorker implements ChannelWorker {
           clientId: this.#mainThreadId,
           uuid: request.data.uuid,
         };
-        xhr.send(encodeData(fetchReqBody));
-        return decodeData(new Uint8Array(xhr.response as ArrayBuffer)) as Response;
+        xhr.send(encode(fetchReqBody));
+        return decode(xhr.response as ArrayBuffer) as Response;
       } catch (e: any) {
         if (e instanceof DOMException && retryCount++ < 1000) {
           console.log('Service worker request failed - resending request');

src/webR/chan/message.ts

@@ -93,27 +93,3 @@ export function newSyncRequest(msg: Message, data: SyncRequestData): SyncRequest
     data: { msg, reqData: data },
   };
 }
-
-const encoder = new TextEncoder();
-const decoder = new TextDecoder('utf-8');
-
-/**
- * Encode data for transfering from worker thread to main thread.
- * @param {any} data The message data to be serialised and encoded.
- * @return {Uint8Array} The encoded data.
- * @internal
- * */
-export function encodeData(data: any): Uint8Array {
-  // TODO: Pass a `replacer` function
-  return encoder.encode(JSON.stringify(data));
-}
-
-/**
- * Decode data that has been transferred from worker thread to main thread.
- * @param {any} data The message data to be decoded.
- * @return {unknown} The data after decoding.
- * @internal
- * */
-export function decodeData(data: Uint8Array): unknown {
-  return JSON.parse(decoder.decode(data)) as unknown;
-}

src/webR/chan/serviceworker.ts

@@ -1,5 +1,5 @@
 import { promiseHandles } from '../utils';
-import { decodeData, encodeData } from './message';
+import { encode, decode } from '@msgpack/msgpack';
 import { ServiceWorkerHandlers } from './channel';
 
 declare let self: ServiceWorkerGlobalScope;
@@ -35,7 +35,7 @@ async function sendRequest(clientId: string, uuid: string): Promise<Response> {
 
   const response = await requests[uuid].promise;
   const headers = { 'Cross-Origin-Embedder-Policy': 'require-corp' };
-  return new Response(encodeData(response), { headers });
+  return new Response(encode(response), { headers });
 }
 
 export function handleFetch(event: FetchEvent) {
@@ -46,7 +46,7 @@ export function handleFetch(event: FetchEvent) {
   }
   const requestBody = event.request.arrayBuffer();
   const requestReponse = requestBody.then(async (body) => {
-    const data = decodeData(new Uint8Array(body)) as { clientId: string; uuid: string };
+    const data = decode(body) as { clientId: string; uuid: string };
     return await sendRequest(data.clientId, data.uuid);
   });
   event.waitUntil(requestReponse);

src/webR/chan/task-main.ts

@@ -3,7 +3,8 @@
 import { Endpoint, SZ_BUF_FITS_IDX, SZ_BUF_SIZE_IDX, generateUUID } from './task-common';
 
 import { sleep } from '../utils';
-import { SyncRequestData, encodeData } from './message';
+import { SyncRequestData } from './message';
+import { encode } from '@msgpack/msgpack';
 
 import { IN_NODE } from '../compat';
 import type { Worker as NodeWorker } from 'worker_threads';
@@ -28,7 +29,7 @@ export async function syncResponse(endpoint: Endpoint, data: SyncRequestData, re
     let { taskId, sizeBuffer, dataBuffer, signalBuffer } = data;
     // console.warn(msg);
 
-    const bytes = encodeData(response);
+    const bytes = encode(response);
     const fits = bytes.length <= dataBuffer.length;
 
     Atomics.store(sizeBuffer, SZ_BUF_SIZE_IDX, bytes.length);

src/webR/chan/task-worker.ts

@@ -8,7 +8,8 @@ import {
   UUID_LENGTH,
 } from './task-common';
 
-import { newSyncRequest, Message, decodeData } from './message';
+import { newSyncRequest, Message } from './message';
+import { decode } from '@msgpack/msgpack';
 
 const decoder = new TextDecoder('utf-8');
 
@@ -99,7 +100,7 @@ export class SyncTask {
 
     const size = Atomics.load(sizeBuffer, SZ_BUF_SIZE_IDX);
     // console.log("===completing", taskId);
-    return decodeData(dataBuffer.slice(0, size));
+    return decode(dataBuffer.slice(0, size));
   }
 
   get result() {

src/webR/robj-worker.ts

@@ -1105,7 +1105,7 @@ function toWebRData(jsObj: WebRData): WebRData;
 function toWebRData(jsObj: WebRData): WebRData {
   if (isWebRDataJs(jsObj)) {
     return jsObj;
-  } else if (Array.isArray(jsObj)) {
+  } else if (Array.isArray(jsObj) || ArrayBuffer.isView(jsObj)) {
     return { names: null, values: jsObj };
   } else if (jsObj && typeof jsObj === 'object' && !isComplex(jsObj)) {
     return {