Merge pull request #838 from twiss/worker-buffer-transfer

Transfer buffers in workers

src/util.js

@@ -52,19 +52,19 @@ export default {
    * @param  {Object} obj           the options object to be passed to the web worker
    * @returns {Array<ArrayBuffer>}   an array of binary data to be passed
    */
-  getTransferables: function(obj) {
+  getTransferables: function(obj, zero_copy) {
     const transferables = [];
-    util.collectTransferables(obj, transferables);
+    util.collectTransferables(obj, transferables, zero_copy);
     return transferables.length ? transferables : undefined;
   },
 
-  collectTransferables: function(obj, collection) {
+  collectTransferables: function(obj, collection, zero_copy) {
     if (!obj) {
       return;
     }
 
-    if (util.isUint8Array(obj) && collection.indexOf(obj.buffer) === -1) {
+    if (util.isUint8Array(obj)) {
-      if (config.zero_copy) {
+      if (zero_copy && collection.indexOf(obj.buffer) === -1) {
         collection.push(obj.buffer);
       }
       return;
@@ -79,8 +79,12 @@ export default {
               const reader = stream.getReader(readable);
               const { port1, port2 } = new MessageChannel();
               port1.onmessage = async function({ data: { action } }) {
-                if (action === 'read') port1.postMessage(await reader.read());
+                if (action === 'read') {
-                else if (action === 'cancel') port1.postMessage(await transformed.cancel());
+                  const result = await reader.read();
+                  port1.postMessage(result, util.getTransferables(result, true));
+                } else if (action === 'cancel') {
+                  port1.postMessage(await transformed.cancel());
+                }
               };
               obj[key] = port2;
               collection.push(port2);
@@ -91,7 +95,7 @@ export default {
         if (Object.prototype.toString.call(value) === '[object MessagePort]') {
           throw new Error("Can't transfer the same stream twice.");
         }
-        util.collectTransferables(value, collection);
+        util.collectTransferables(value, collection, zero_copy);
       });
     }
   },

src/worker/async_proxy.js

@@ -22,12 +22,14 @@
  * @see module:openpgp.destroyWorker
  * @see module:worker/worker
  * @requires util
+ * @requires config
  * @requires crypto
  * @requires packet
  * @module worker/async_proxy
  */
 
 import util from '../util.js';
+import config from '../config';
 import crypto from '../crypto';
 import packet from '../packet';
 
@@ -112,7 +114,7 @@ AsyncProxy.prototype.getID = function() {
  */
 AsyncProxy.prototype.seedRandom = async function(workerId, size) {
   const buf = await crypto.random.getRandomBytes(size);
-  this.workers[workerId].postMessage({ event:'seed-random', buf }, util.getTransferables(buf));
+  this.workers[workerId].postMessage({ event:'seed-random', buf }, util.getTransferables(buf, true));
 };
 
 /**
@@ -143,9 +145,9 @@ AsyncProxy.prototype.delegate = function(method, options) {
     }
   }
 
-  return new Promise(async (resolve, reject) => {
+  return new Promise((resolve, reject) => {
     // clone packets (for web worker structured cloning algorithm)
-    this.workers[workerId].postMessage({ id:id, event:method, options:packet.clone.clonePackets(options) }, util.getTransferables(options));
+    this.workers[workerId].postMessage({ id:id, event:method, options:packet.clone.clonePackets(options) }, util.getTransferables(options, config.zero_copy));
     this.workers[workerId].requests++;
 
     // remember to handle parsing cloned packets from worker

src/worker/worker.js

@@ -133,5 +133,5 @@ function delegate(id, method, options) {
  * @param  {Object} event  Contains event type and data
  */
 function response(event) {
-  self.postMessage(event, openpgp.util.getTransferables(event.data));
+  self.postMessage(event, openpgp.util.getTransferables(event.data, true));
 }

test/general/openpgp.js

@@ -1741,7 +1741,7 @@ describe('[Sauce Labs Group 2] OpenPGP.js public api tests', function() {
                 if (i++ < 4) {
                   let randomBytes = await openpgp.crypto.random.getRandomBytes(10);
                   controller.enqueue(randomBytes);
-                  plaintext.push(randomBytes);
+                  plaintext.push(randomBytes.slice());
                 } else {
                   controller.close();
                 }

test/general/util.js

@@ -117,7 +117,6 @@ describe('Util unit tests', function() {
   });
 
   describe('getTransferables', function() {
-    let zero_copyVal;
     const buf1 = new Uint8Array(1);
     const buf2 = new Uint8Array(1);
     const obj = {
@@ -128,27 +127,18 @@ describe('Util unit tests', function() {
         }
       };
 
-    beforeEach(function() {
-      zero_copyVal = openpgp.config.zero_copy;
-      openpgp.config.zero_copy = true;
-    });
-
-    afterEach(function() {
-      openpgp.config.zero_copy = zero_copyVal;
-    });
-
     it('should return undefined when zero_copy is false', function() {
       openpgp.config.zero_copy = false;
-      expect(openpgp.util.getTransferables(obj)).to.be.undefined;
+      expect(openpgp.util.getTransferables(obj, false)).to.be.undefined;
     });
     it('should return undefined for no input', function() {
-      expect(openpgp.util.getTransferables()).to.be.undefined;
+      expect(openpgp.util.getTransferables(undefined, true)).to.be.undefined;
     });
     it('should return undefined for an empty oject', function() {
-      expect(openpgp.util.getTransferables({})).to.be.undefined;
+      expect(openpgp.util.getTransferables({}, true)).to.be.undefined;
     });
     it('should return two buffers', function() {
-      expect(openpgp.util.getTransferables(obj)).to.deep.equal([buf1.buffer, buf2.buffer]);
+      expect(openpgp.util.getTransferables(obj, true)).to.deep.equal([buf1.buffer, buf2.buffer]);
     });
   });