defer generator resumption until start of frame (#319)

* defer generator resumption until start of frame

* handle generator error

* assign promise if generator.next throws

* don’t expose value until it resolves

* remove needless try-catch

* compute downstream on error

* minimize diff

* clean and comment

* reuse promise

Author: mbostock

src/runtime.js

@@ -16,6 +16,7 @@ export default function Runtime(builtins = new Library, global = window_global)
   Object.defineProperties(this, {
     _dirty: {value: new Set},
     _updates: {value: new Set},
+    _precomputes: {value: [], writable: true},
     _computing: {value: null, writable: true},
     _init: {value: null, writable: true},
     _modules: {value: new Map},
@@ -34,6 +35,7 @@ Object.defineProperties(Runtime, {
 });
 
 Object.defineProperties(Runtime.prototype, {
+  _precompute: {value: runtime_precompute, writable: true, configurable: true},
   _compute: {value: runtime_compute, writable: true, configurable: true},
   _computeSoon: {value: runtime_computeSoon, writable: true, configurable: true},
   _computeNow: {value: runtime_computeNow, writable: true, configurable: true},
@@ -72,24 +74,32 @@ function runtime_module(define, observer = noop) {
   return module;
 }
 
+function runtime_precompute(callback) {
+  this._precomputes.push(callback);
+  this._compute();
+}
+
 function runtime_compute() {
   return this._computing || (this._computing = this._computeSoon());
 }
 
 function runtime_computeSoon() {
-  var runtime = this;
-  return new Promise(function(resolve) {
-    frame(function() {
-      resolve();
-      runtime._disposed || runtime._computeNow();
-    });
-  });
+  return new Promise(frame).then(() => this._disposed ? undefined : this._computeNow());
 }
 
-function runtime_computeNow() {
+async function runtime_computeNow() {
   var queue = [],
       variables,
-      variable;
+      variable,
+      precomputes = this._precomputes;
+
+  // If there are any paused generators, resume them before computing so they
+  // can update (if synchronous) before computing downstream variables.
+  if (precomputes.length) {
+    this._precomputes = [];
+    for (const callback of precomputes) callback();
+    await runtime_defer(3);
+  }
 
   // Compute the reachability of the transitive closure of dirty variables.
   // Any newly-reachable variable must also be recomputed.
@@ -159,6 +169,16 @@ function runtime_computeNow() {
   }
 }
 
+// We want to give generators, if they’re defined synchronously, a chance to
+// update before computing downstream variables. This creates a synchronous
+// promise chain of the given depth that we’ll await before recomputing
+// downstream variables.
+function runtime_defer(depth = 0) {
+  let p = Promise.resolve();
+  for (let i = 0; i < depth; ++i) p = p.then(() => {});
+  return p;
+}
+
 function variable_circular(variable) {
   const inputs = new Set(variable._inputs);
   for (const i of inputs) {
@@ -206,10 +226,21 @@ function variable_compute(variable) {
   variable._invalidate();
   variable._invalidate = noop;
   variable._pending();
-  var value0 = variable._value,
-      version = ++variable._version,
-      invalidation = null,
-      promise = variable._promise = Promise.all(variable._inputs.map(variable_value)).then(function(inputs) {
+
+  const value0 = variable._value;
+  const version = ++variable._version;
+
+  // Lazily-constructed invalidation variable; only constructed if referenced as an input.
+  let invalidation = null;
+
+  // If the variable doesn’t have any inputs, we can optimize slightly.
+  const promise = variable._promise = (variable._inputs.length
+      ? Promise.all(variable._inputs.map(variable_value)).then(define)
+      : new Promise(resolve => resolve(variable._definition.call(value0))))
+    .then(generate);
+
+  // Compute the initial value of the variable.
+  function define(inputs) {
     if (variable._version !== version) return;
 
     // Replace any reference to invalidation with the promise, lazily.
@@ -227,64 +258,77 @@ function variable_compute(variable) {
       }
     }
 
-    // Compute the initial value of the variable.
     return variable._definition.apply(value0, inputs);
-  }).then(function(value) {
-    // If the value is a generator, then retrieve its first value,
-    // and dispose of the generator if the variable is invalidated.
-    // Note that the cell may already have been invalidated here,
-    // in which case we need to terminate the generator immediately!
+  }
+
+  // If the value is a generator, then retrieve its first value, and dispose of
+  // the generator if the variable is invalidated. Note that the cell may
+  // already have been invalidated here, in which case we need to terminate the
+  // generator immediately!
+  function generate(value) {
     if (generatorish(value)) {
       if (variable._version !== version) return void value.return();
       (invalidation || variable_invalidator(variable)).then(variable_return(value));
-      return variable_precompute(variable, version, promise, value);
+      return variable_generate(variable, version, value);
     }
     return value;
-  });
-  promise.then(function(value) {
+  }
+
+  promise.then((value) => {
     if (variable._version !== version) return;
     variable._value = value;
     variable._fulfilled(value);
-  }, function(error) {
+  }, (error) => {
     if (variable._version !== version) return;
     variable._value = undefined;
     variable._rejected(error);
   });
 }
 
-function variable_precompute(variable, version, promise, generator) {
+function variable_generate(variable, version, generator) {
+  const runtime = variable._module._runtime;
+
+  // Retrieve the next value from the generator; if successful, invoke the
+  // specified callback. The returned promise resolves to the yielded value, or
+  // to undefined if the generator is done.
+  function compute(onfulfilled) {
+    return new Promise(resolve => resolve(generator.next())).then(({done, value}) => {
+      return done ? undefined : (value = Promise.resolve(value), value.then(onfulfilled), value);
+    });
+  }
+
+  // Retrieve the next value from the generator; if successful, fulfill the
+  // variable, compute downstream variables, and schedule the next value to be
+  // pulled from the generator at the start of the next animation frame. If not
+  // successful, reject the variable, compute downstream variables, and return.
   function recompute() {
-    var promise = new Promise(function(resolve) {
-      resolve(generator.next());
-    }).then(function(next) {
-      return next.done ? undefined : Promise.resolve(next.value).then(function(value) {
-        if (variable._version !== version) return;
-        variable_postrecompute(variable, value, promise).then(recompute);
-        variable._fulfilled(value);
-        return value;
-      });
+    const promise = compute((value) => {
+      if (variable._version !== version) return;
+      postcompute(value, promise).then(() => runtime._precompute(recompute));
+      variable._fulfilled(value);
     });
-    promise.catch(function(error) {
+    promise.catch((error) => {
       if (variable._version !== version) return;
-      variable_postrecompute(variable, undefined, promise);
+      postcompute(undefined, promise);
       variable._rejected(error);
     });
   }
-  return new Promise(function(resolve) {
-    resolve(generator.next());
-  }).then(function(next) {
-    if (next.done) return;
-    promise.then(recompute);
-    return next.value;
-  });
-}
 
-function variable_postrecompute(variable, value, promise) {
-  var runtime = variable._module._runtime;
-  variable._value = value;
-  variable._promise = promise;
-  variable._outputs.forEach(runtime._updates.add, runtime._updates); // TODO Cleaner?
-  return runtime._compute();
+  // After the generator fulfills or rejects, set its current value, promise,
+  // and schedule any downstream variables for update.
+  function postcompute(value, promise) {
+    variable._value = value;
+    variable._promise = promise;
+    variable._outputs.forEach(runtime._updates.add, runtime._updates);
+    return runtime._compute();
+  }
+
+  // When retrieving the first value from the generator, the promise graph is
+  // already established, so we only need to queue the next pull.
+  return compute(() => {
+    if (variable._version !== version) return;
+    runtime._precompute(recompute);
+  });
 }
 
 function variable_error(variable, error) {

test/module/value-test.js

@@ -37,6 +37,38 @@ tape("module.value(name) supports generators", async test => {
   test.deepEqual(await module.value("foo"), 3);
 });
 
+tape("module.value(name) supports generators that throw", async test => {
+  const runtime = new Runtime();
+  const module = runtime.module();
+  module.define("foo", [], function*() { yield 1; throw new Error("fooed"); });
+  module.define("bar", ["foo"], foo => foo);
+  const [foo1, bar1] = await Promise.all([module.value("foo"), module.value("bar")]);
+  test.deepEqual(foo1, 1);
+  test.deepEqual(bar1, 1);
+  try {
+    await module.value("foo");
+    test.fail();
+  } catch (error) {
+    test.deepEqual(error.message, "fooed");
+  }
+  try {
+    await module.value("bar");
+    test.fail();
+  } catch (error) {
+    test.deepEqual(error.message, "fooed");
+  }
+});
+
+tape("module.value(name) supports async generators", async test => {
+  const runtime = new Runtime();
+  const module = runtime.module();
+  module.define("foo", [], async function*() { yield 1; yield 2; yield 3; });
+  test.deepEqual(await module.value("foo"), 1);
+  test.deepEqual(await module.value("foo"), 2);
+  test.deepEqual(await module.value("foo"), 3);
+  test.deepEqual(await module.value("foo"), 3);
+});
+
 tape("module.value(name) supports promises", async test => {
   const runtime = new Runtime();
   const module = runtime.module();

test/variable/define-test.js

@@ -388,3 +388,101 @@ tape("variable.define correctly handles globals that throw", async test => {
   const foo = module.variable(true).define(["oops"], oops => oops);
   test.deepEqual(await valueof(foo), {error: "RuntimeError: oops"});
 });
+
+tape("variable.define allows other variables to begin computation before a generator may resume", async test => {
+  const runtime = new Runtime();
+  const module = runtime.module();
+  const main = runtime.module();
+  let i = 0;
+  let genIteration = 0;
+  let valIteration = 0;
+  const onGenFulfilled = value => {
+    if (genIteration === 0) {
+      test.equals(valIteration, 0);
+      test.equals(value, 1);
+      test.equals(i, 1);
+    } else if (genIteration === 1) {
+      test.equals(valIteration, 1);
+      test.equals(value, 2);
+      test.equals(i, 2);
+    } else if (genIteration === 2) {
+      test.equals(valIteration, 2);
+      test.equals(value, 3);
+      test.equals(i, 3);
+    } else {
+      test.fail();
+    }
+    genIteration++;
+  };
+  const onValFulfilled = value => {
+    if (valIteration === 0) {
+      test.equals(genIteration, 1);
+      test.equals(value, 1);
+      test.equals(i, 1);
+    } else if (valIteration === 1) {
+      test.equals(genIteration, 2);
+      test.equals(value, 2);
+      test.equals(i, 2);
+    } else if (valIteration === 2) {
+      test.equals(genIteration, 3);
+      test.equals(value, 3);
+      test.equals(i, 3);
+    } else {
+      test.fail();
+    }
+    valIteration++;
+  };
+  const gen = module.variable({fulfilled: onGenFulfilled}).define("gen", [], function*() {
+    i++;
+    yield i;
+    i++;
+    yield i;
+    i++;
+    yield i;
+  });
+  main.variable().import("gen", module);
+  const val = main.variable({fulfilled: onValFulfilled}).define("val", ["gen"], i => i);
+  test.equals(await gen._promise, undefined, "gen cell undefined");
+  test.equals(await val._promise, undefined, "val cell undefined");
+  await runtime._compute();
+  test.equals(await gen._promise, 1, "gen cell 1");
+  test.equals(await val._promise, 1, "val cell 1");
+  await runtime._compute();
+  test.equals(await gen._promise, 2, "gen cell 2");
+  test.equals(await val._promise, 2, "val cell 2");
+  await runtime._compute();
+  test.equals(await gen._promise, 3, "gen cell 3");
+  test.equals(await val._promise, 3, "val cell 3");
+});
+
+tape("variable.define allows other variables to begin computation before a generator may resume", async test => {
+  const runtime = new Runtime();
+  const main = runtime.module();
+  let i = 0;
+  let j = 0;
+  const gen = main.variable().define("gen", [], function*() {
+    i++;
+    yield i;
+    i++;
+    yield i;
+    i++;
+    yield i;
+  });
+  const val = main.variable(true).define("val", ["gen"], gen => {
+    j++;
+    test.equals(gen, j, "gen = j");
+    test.equals(gen, i, "gen = i");
+    return gen;
+  });
+  test.equals(await gen._promise, undefined, "gen = undefined");
+  test.equals(await val._promise, undefined, "val = undefined");
+  await runtime._compute();
+  test.equals(await gen._promise, 1, "gen cell 1");
+  test.equals(await val._promise, 1, "val cell 1");
+  await runtime._compute();
+  test.equals(await gen._promise, 2, "gen cell 2");
+  test.equals(await val._promise, 2, "val cell 2");
+  await runtime._compute();
+  test.equals(await gen._promise, 3, "gen cell 3");
+  test.equals(await val._promise, 3, "val cell 3");
+});