refactor work queue

lib/wallet/cryptoApi.ts

@@ -28,11 +28,6 @@ import {CoinWithDenom} from "./wallet";
 import {PayCoinInfo} from "./types";
 import {RefreshSession} from "./types";
 
-interface RegistryEntry {
-  resolve: any;
-  reject: any;
-  workerIndex: number;
-}
 
 interface WorkerState {
   /**
@@ -41,19 +36,14 @@ interface WorkerState {
   w: Worker|null;
 
   /**
-   * Are we currently running a task on this worker?
+   * Work we're currently executing or null if not busy.
    */
-  busy: boolean;
+  currentWorkItem: WorkItem|null;
 
   /**
    * Timer to terminate the worker if it's not busy enough.
    */
   terminationTimerHandle: number|null;
-
-  /**
-   * Index of this worker in the list of workers.
-   */
-  workerIndex: number;
 }
 
 interface WorkItem {
@@ -61,6 +51,11 @@ interface WorkItem {
   args: any[];
   resolve: any;
   reject: any;
+
+  /**
+   * Serial id to identify a matching response.
+   */
+  rpcId: number;
 }
 
 
@@ -72,27 +67,21 @@ const NUM_PRIO = 5;
 
 export class CryptoApi {
   private nextRpcId: number = 1;
-  private rpcRegistry: {[n: number]: RegistryEntry} = {};
   private workers: WorkerState[];
   private workQueues: WorkItem[][];
   /**
    * Number of busy workers.
    */
   private numBusy: number = 0;
-  /**
-   * Number if pending work items.
-   */
-  private numWaiting: number = 0;
-
 
   /**
    * Start a worker (if not started) and set as busy.
    */
-  wake(ws: WorkerState) {
+  wake<T>(ws: WorkerState, work: WorkItem): void {
-    if (ws.busy) {
+    if (ws.currentWorkItem != null) {
       throw Error("assertion failed");
     }
-    ws.busy = true;
+    ws.currentWorkItem = work;
     this.numBusy++;
     if (!ws.w) {
       let w = new Worker("/lib/wallet/cryptoWorker.js");
@@ -100,27 +89,45 @@ export class CryptoApi {
       w.onerror = (e: ErrorEvent) => this.handleWorkerError(ws, e);
       ws.w = w;
     }
+
+    let msg: any = {
+      operation: work.operation, args: work.args,
+      id: work.rpcId
+    };
+    this.resetWorkerTimeout(ws);
+    ws.w!.postMessage(msg);
+  }
+
+  resetWorkerTimeout(ws: WorkerState) {
     if (ws.terminationTimerHandle != null) {
       clearTimeout(ws.terminationTimerHandle);
     }
     let destroy = () => {
-      if (ws.w && !ws.busy) {
+      // terminate worker if it's idle
+      if (ws.w && ws.currentWorkItem == null) {
         ws.w!.terminate();
         ws.w = null;
       }
-    }
+    };
     ws.terminationTimerHandle = setTimeout(destroy, 20 * 1000);
   }
 
   handleWorkerError(ws: WorkerState, e: ErrorEvent) {
+    if (ws.currentWorkItem) {
+      console.error(`error in worker during ${ws.currentWorkItem!.operation}`, e);
+    } else {
       console.error("error in worker", e);
+    }
+    console.error(e.message);
     try {
       ws.w!.terminate();
+      ws.w = null;
     } catch (e) {
       console.error(e);
     }
-    if (ws.busy) {
+    if (ws.currentWorkItem != null) {
-      ws.busy = false;
+      ws.currentWorkItem.reject(e);
+      ws.currentWorkItem = null;
       this.numBusy--;
     }
     this.findWork(ws);
@@ -132,14 +139,7 @@ export class CryptoApi {
       let q = this.workQueues[NUM_PRIO - i - 1];
       if (q.length != 0) {
         let work: WorkItem = q.shift()!;
-        let msg: any = {
+        this.wake(ws, work);
-          operation: work.operation,
-          args: work.args,
-          id: this.registerRpcId(work.resolve, work.reject, ws.workerIndex),
-        };
-
-        this.wake(ws);
-        ws.w!.postMessage(msg);
         return;
       }
     }
@@ -151,22 +151,19 @@ export class CryptoApi {
       console.error("rpc id must be number");
       return;
     }
-    if (!this.rpcRegistry[id]) {
+    let currentWorkItem = ws.currentWorkItem;
+    ws.currentWorkItem = null;
+    this.numBusy--;
+    this.findWork(ws);
+    if (!currentWorkItem) {
+      console.error("unsolicited response from worker");
+      return;
+    }
+    if (id != currentWorkItem.rpcId) {
       console.error(`RPC with id ${id} has no registry entry`);
       return;
     }
-    let {resolve, workerIndex} = this.rpcRegistry[id];
+    currentWorkItem.resolve(msg.data.result);
-    delete this.rpcRegistry[id];
-    if (workerIndex != ws.workerIndex) {
-      throw Error("assertion failed");
-    }
-    if (!ws.busy) {
-      throw Error("assertion failed");
-    }
-    ws.busy = false;
-    this.numBusy--;
-    resolve(msg.data.result);
-    this.findWork(ws);
   }
 
   constructor() {
@@ -175,9 +172,8 @@ export class CryptoApi {
     for (let i = 0; i < this.workers.length; i++) {
       this.workers[i] = {
         w: null,
-        busy: false,
         terminationTimerHandle: null,
-        workerIndex: i,
+        currentWorkItem: null,
       };
     }
     this.workQueues = [];
@@ -186,45 +182,34 @@ export class CryptoApi {
     }
   }
 
-
-  private registerRpcId(resolve: any, reject: any,
-                        workerIndex: number): number {
-    let id = this.nextRpcId++;
-    this.rpcRegistry[id] = {resolve, reject, workerIndex};
-    return id;
-  }
-
-
   private doRpc<T>(operation: string, priority: number,
                    ...args: any[]): Promise<T> {
+
+    return new Promise((resolve, reject) => {
+      let rpcId = this.nextRpcId++;
+      let workItem: WorkItem = {operation, args, resolve, reject, rpcId};
+
       if (this.numBusy == this.workers.length) {
         let q = this.workQueues[priority];
         if (!q) {
           throw Error("assertion failed");
         }
-      return new Promise<T>((resolve, reject) => {
+        this.workQueues[priority].push(workItem);
-        this.workQueues[priority].push({operation, args, resolve, reject});
+        return;
-      });
       }
 
       for (let i = 0; i < this.workers.length; i++) {
         let ws = this.workers[i];
-      if (ws.busy) {
+        if (ws.currentWorkItem != null) {
           continue;
         }
 
-      this.wake(ws);
+        this.wake<T>(ws, workItem);
-
+        return;
-      return new Promise<T>((resolve, reject) => {
-        let msg: any = {
-          operation, args,
-          id: this.registerRpcId(resolve, reject, i),
-        };
-        ws.w!.postMessage(msg);
-      });
       }
 
       throw Error("assertion failed");
+    });
   }
 
 

lib/wallet/cryptoLib.ts

@@ -79,10 +79,14 @@ namespace RpcFunctions {
     let coinPub = coinPriv.getPublicKey();
     let blindingFactor = native.RsaBlindingKeySecret.create();
     let pubHash: native.HashCode = coinPub.hash();
-    let ev: native.ByteArray = native.rsaBlind(pubHash,
+    let ev = native.rsaBlind(pubHash,
                              blindingFactor,
                              denomPub);
 
+    if (!ev) {
+      throw Error("couldn't blind (malicious exchange key?)");
+    }
+
     if (!denom.fee_withdraw) {
       throw Error("Field fee_withdraw missing");
     }
@@ -234,7 +238,7 @@ namespace RpcFunctions {
   }
 
 
-  function createWithdrawSession(kappa: number, meltCoin: Coin,
+  export function createWithdrawSession(kappa: number, meltCoin: Coin,
                                         newCoinDenoms: Denomination[],
                                         meltAmount: AmountJson,
                                         meltFee: AmountJson): RefreshSession {
@@ -268,9 +272,12 @@ namespace RpcFunctions {
         let blindingFactor = native.RsaBlindingKeySecret.create();
         let pubHash: native.HashCode = coinPub.hash();
         let denomPub = native.RsaPublicKey.fromCrock(newCoinDenoms[i].denom_pub);
-        let ev: native.ByteArray = native.rsaBlind(pubHash,
+        let ev = native.rsaBlind(pubHash,
                                  blindingFactor,
                                  denomPub);
+        if (!ev) {
+          throw Error("couldn't blind (malicious exchange key?)");
+        }
         let preCoin: RefreshPreCoin = {
           blindingKey: blindingFactor.toCrock(),
           coinEv: ev.toCrock(),

lib/wallet/cryptoWorker.ts

@@ -22,7 +22,7 @@
 "use strict";
 
 
-importScripts("../emscripten/libwrapper.js",
+importScripts("../emscripten/taler-emscripten-lib.js",
               "../vendor/system-csp-production.src.js");
 
 
@@ -46,7 +46,7 @@ if ("object" !== typeof Module) {
 
 {
   let mod = System.newModule({Module: Module});
-  let modName = System.normalizeSync("../emscripten/emsc");
+  let modName = System.normalizeSync("../emscripten/taler-emscripten-lib");
   console.log("registering", modName);
   System.set(modName, mod);
 }

lib/wallet/emscriptif.ts

@@ -14,8 +14,8 @@
  TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
  */
 
-import { AmountJson } from "./types";
+import {AmountJson} from "./types";
-import * as EmscWrapper from "../emscripten/emsc";
+import * as EmscWrapper from "../emscripten/taler-emscripten-lib";
 
 /**
  * High-level interface to emscripten-compiled modules used
@@ -137,7 +137,7 @@ var emscAlloc = {
                           ['number', 'number', 'number']),
   rsa_blind: getEmsc('GNUNET_CRYPTO_rsa_blind',
                      'number',
-    ['number', 'number', 'number', 'number']),
+                     ['number', 'number', 'number', 'number', 'number']),
   rsa_blinding_key_create: getEmsc('GNUNET_CRYPTO_rsa_blinding_key_create',
                                    'number',
                                    ['number']),
@@ -519,7 +519,7 @@ abstract class PackedArenaObject extends MallocArenaObject {
                                   this.size());
     buf.destroy();
     if (res < 1) {
-      throw { error: "wrong encoding" };
+      throw {error: "wrong encoding"};
     }
   }
 
@@ -618,7 +618,6 @@ export class EcdsaPrivateKey extends PackedArenaObject {
 mixinStatic(EcdsaPrivateKey, fromCrock);
 
 
-
 function fromCrock(s: string) {
   let x = new this();
   x.alloc();
@@ -1123,13 +1122,23 @@ mixin(RsaSignature, makeEncode(emscAlloc.rsa_signature_encode));
 export function rsaBlind(hashCode: HashCode,
                          blindingKey: RsaBlindingKeySecret,
                          pkey: RsaPublicKey,
-  arena?: Arena): ByteArray {
+                         arena?: Arena): ByteArray|null {
-  let ptr = emscAlloc.malloc(PTR_SIZE);
+  let buf_ptr_out = emscAlloc.malloc(PTR_SIZE);
-  let s = emscAlloc.rsa_blind(hashCode.nativePtr,
+  let buf_size_out = emscAlloc.malloc(PTR_SIZE);
+  let res = emscAlloc.rsa_blind(hashCode.nativePtr,
                                 blindingKey.nativePtr,
                                 pkey.nativePtr,
-    ptr);
+                                buf_ptr_out,
-  return new ByteArray(s, Module.getValue(ptr, '*'), arena);
+                                buf_size_out);
+  let buf_ptr = Module.getValue(buf_ptr_out, '*');
+  let buf_size = Module.getValue(buf_size_out, '*');
+  emsc.free(buf_ptr_out);
+  emsc.free(buf_size_out);
+  if (res != GNUNET_OK) {
+    // malicious key
+    return null;
+  }
+  return new ByteArray(buf_size, buf_ptr, arena);
 }
 
 
@@ -1208,16 +1217,19 @@ export interface FreshCoin {
   blindingKey: RsaBlindingKeySecret;
 }
 
-export function setupFreshCoin(secretSeed: TransferSecretP, coinIndex: number): FreshCoin {
+export function setupFreshCoin(secretSeed: TransferSecretP,
+                               coinIndex: number): FreshCoin {
   let priv = new EddsaPrivateKey();
   priv.isWeak = true;
   let blindingKey = new RsaBlindingKeySecret();
   blindingKey.isWeak = true;
 
-  let buf = kdf(priv.size() + blindingKey.size(), UInt32.fromNumber(coinIndex), ByteArray.fromString("taler-coin-derivation"));
+  let buf = kdf(priv.size() + blindingKey.size(),
+                UInt32.fromNumber(coinIndex),
+                ByteArray.fromString("taler-coin-derivation"));
 
   priv.nativePtr = buf.nativePtr;
   blindingKey.nativePtr = buf.nativePtr + priv.size();
 
-  return { priv, blindingKey };
+  return {priv, blindingKey};
 }

lib/wallet/wallet.ts

@@ -797,8 +797,6 @@ export class Wallet {
   }
 
   async storeCoin(coin: Coin): Promise<void> {
-    console.log("storing coin", new Date());
-
     let historyEntry: HistoryRecord = {
       type: "withdraw",
       timestamp: (new Date).getTime(),

lib/wallet/wxMessaging.ts

@@ -286,7 +286,6 @@ class ChromeNotifier implements Notifier {
   }
 
   notify() {
-    console.log("notifying all ports");
     for (let p of this.ports) {
       p.postMessage({ notify: true });
     }

pages/confirm-create-reserve.tsx

@@ -172,7 +172,7 @@ class ExchangeSelection extends ImplicitStateComponent<ExchangeSelectionProps> {
 
 
   renderAdvanced(): JSX.Element {
-    if (this.detailCollapsed() && this.url() !== null) {
+    if (this.detailCollapsed() && this.url() !== null && !this.statusString()) {
       return (
         <button className="linky"
                 onClick={() => this.detailCollapsed(false)}>
@@ -224,6 +224,13 @@ class ExchangeSelection extends ImplicitStateComponent<ExchangeSelectionProps> {
         <em>{shortName}</em>
       </p>;
     }
+    if (this.statusString()) {
+      return (
+        <p>
+          <strong style="color: red;">A problem occured, see below.</strong>
+        </p>
+      );
+    }
     return (
       <p>
         Information about fees will be available when an exchange provider is selected.