Some asynchronous patterns

asynchronicity.js

/**
 * asynchronicity.js
 *
 * Steve Lloyd <steve@repeatingbeats.com>
 *
 * Exploratory script covering various issues that arise working with
 * asychronous javascript functions. This script is intended to be run from
 * the command line with node, i.e.:
 *     $ node asynchronicity.js
 *
 * For simplicity, this script doesn't deal with errors. In reality, every
 * callback would need to handle error responses, and the async helpers
 * would need to exit early by invoking error functions.
 */

// Let's simulate some getters that take a long time to respond.
var slow_network = {

  get_first_part: function (callback) {
    setTimeout(function() { callback('hello'); }, 3000);
  },

  get_second_part: function (callback) {
    setTimeout(function() { callback('async'); }, 1000);
  },

  get_third_part: function (callback) {
    setTimeout(function() { callback('world'); }, 500);
  },

};

// Helper to report results and timing info for async operations
var exhibit = function (title) {

  var start_time = Date.now();

  return {
    log: function (msg) {
      console.log(title + ': ' + msg);
    },
    report: function (result) {
      console.log('report for ' + title +
                  ':\n\tresult: ' + result +
                  '\n\telapsed: ' +
                  (Date.now() - start_time) + ' ms');
    }
  };
};

// Let's do this in the most straightforward manner possible.
var exhibit_a = exhibit('basic nested callbacks');
slow_network.get_first_part(function (first_part) {
  slow_network.get_second_part(function (second_part) {
    slow_network.get_third_part(function (third_part) {
      exhibit_a.report([first_part, second_part, third_part].join(' '));
    });
  });
});

// That was simple, but the functions are independent. There's no reason to sit
// around doing nothing while we're waiting for results from each successive
// call.
var exhibit_b = exhibit('parallel polling');
var first, second, third;
slow_network.get_first_part(function (first_part) {
  first = first_part;
});
slow_network.get_second_part(function (second_part) {
  second = second_part;
});
slow_network.get_third_part(function (third_part) {
  third = third_part;
});

// Now we need a way to know when we have all of the results. We can use a
// timer ...
var interval_id = setInterval(function () {
  if (first && second && third) {
    clearInterval(interval_id);
    exhibit_b.report([first, second, third].join(' '));
  }
}, 50);

// But, the timer approach requires continuous polling and result-checking. We
// need a way of knowing when all the functions have returned. Let's create a
// helper to call the methods in parallel.
var async_helper = {

  // async_helper.parallel(func_0, func_1, ..., func_n, callback)
  //
  // Call two or more asynchronous functions in parallel and
  // invoke a callback when all functions have returned. Each
  // input function should take a single argument, a callback
  // function. This callback should be invoked with an arbitrary
  // result object when the asynchronous function calls back.
  //
  // Result objects will be stored in an array and passed as the
  // sole argument of async_helper.parallel's callback.

  parallel: function () {

    var args = Array.prototype.slice.call(arguments),
        callback = args.pop(),
        results = [],
        in_progress = args.length;

    args.forEach(function (async_call, index) {
      async_call(function (result) {
        results[index] = result;
        if (--in_progress == 0) {
          callback(results);
        }
      });
    });
  },

  // async_helper.sequence(func_0, func_1, ..., func_n, callback)
  //
  // Call two or more asynchronous functions sequentialy and
  // invoke a final callback at the end of the sequence. Each
  // input function takes two arguments.
  //
  // The first argment of each input function is a callback that
  // should be invoked with an arbitrary result object when the
  // async function calls back.
  //
  // The second argument of each input function is an array of
  // result objects corresponding to the ordered input functions.
  // This allows intermediary results to be passed down the
  // sequence.
  //
  // Result objects will be stored in an array and passed as the
  // sole argument of async_helper.sequence's callback.

  sequence: function () {

    var args = Array.prototype.slice.call(arguments),
        callback = args.pop(),
        results = [];

    function next() {
      var func = args.shift();
      func(function (result) {
        results.push(result);
        args.length > 0 ? next() : callback(results);
      }, results);
    }

    next();
  }

};

// Now we can call async functions in parallel without the ugly polling.
var exhibit_c = exhibit('async.parallel');
async_helper.parallel(
  slow_network.get_first_part,
  slow_network.get_second_part,
  slow_network.get_third_part,
  function (results) {
    exhibit_c.report(results.join(' '));
  }
);

// Sometimes the results of each function are neede for the next one. If we
// pretend that to be the case for 'hello async world', we can call the
// async getters in sequence without nesting the callbacks.
var exhibit_d = exhibit('async_helper.sequence');
async_helper.sequence(
  slow_network.get_first_part,
  slow_network.get_second_part,
  slow_network.get_third_part,
  function (results) {
    exhibit_d.report(results.join(' '));
  }
);

// Now, let's make a mixed call chain of sequence and parallel async
// functions. Our completely contrived example will assume we have a
// disk that is _extremely_ slow to read. Fortunately, the slow disk
// allows us to do some reads in parallel.

// Our contrived slow disk has two kinds of reads. Direct reads are
// invoked with a getter and call back with a numeric value.
// Indirect reads are also invoked with a getter, but indirect reads
// call back with the name of a direct getter that must then be
// called to retrieve the value.

// Lets build our slow disk to have the following calls:
//
//    'get_a' -> 'get_e' (indirect)
//    'get_b' -> 'get_c' (indirect)
//    'get_c' ->    2    (direct)
//    'get_d' ->    5    (direct)
//    'get_e' ->    1    (direct)

// Build an object that implements my slow disk behavior. 
var slow_disk = (function () {

  var getter_factory = function(getters) {

    var widget = {},
        params = null,
        name = null;

    function build_getter(val, delay) {
      return function (callback, results) {
        setTimeout(function() { callback(val); }, delay);
      };
    }

    for (name in getters) {
      params = getters[name];
      widget[name] = build_getter(params.val, params.delay);
    }
/*
    // You first reaction might be "A build_getter method? Why not
    // just loop through the getters and make the functions directly?"
    // We can't do this directly because the closures would all
    // reference the same value of params after the for loop exits.
    for (var name in getters) {
      console.log('getters[' + name + '].val = ' + getters[name].val);
      var params = getters[name];
      widget['get_' + name] = function (callback) {
        setTimeout(function() { callback(params.val); }, params.delay);
      }
    }
*/
    return widget;
  }

  return getter_factory({
    get_a: { val: 'get_e', delay: 2000 },
    get_b: { val: 'get_c', delay: 3000 },
    get_c: { val: 2, delay: 1000 },
    get_d: { val: 5, delay: 1500 },
    get_e: { val: 1, delay: 5000 },
  });

}());

// Now, we want to figure out the answer to:
//    a + b + c + d + e

// We can do the following five actions in parallel:
//    (1) sequence:
//          - get a
//          - get the value of what a points to
//    (2) sequence:
//          - get b
//          - get the value of what b points to
//    (3) get c
//    (4) get d
//    (5) get e
//

// We can easily nest the sequences inside a parallel call structure
var exhibit_e = exhibit('parallel/sequence mix');
async_helper.parallel(
  // Get a, then get the value of what a points to
  function (callback) {
    async_helper.sequence(
      slow_disk.get_a,
      function (callback, results) {
        slow_disk[results[0]](function (val) { callback(val); });
      },
      function (results) {
        // We want the result of the second function
        callback(results[1]);
      }
    );
  },
  // Get b, then get the value of what b points to
  function (callback) {
    async_helper.sequence(
      slow_disk.get_b,
      function (callback, results) {
        slow_disk[results[0]](function (val) { callback(val); });
      },
      function (results) {
        // Again, we want the result of the second function
        callback(results[1]);
      }
    );
  },
  slow_disk.get_c,
  slow_disk.get_d,
  slow_disk.get_e,
  function (results) {
    var sum = results.reduce(function (sum, val) { return sum + val; });
    exhibit_e.report('sum = ' + sum);
  }
);

// We can also cache results. Assuming that the value don't change often
// (and since I contrived the example, I declare that to be true), we
// should cache results so that we don't have to spend forever looking
// up get_e multiple times. We should do more than just caching results,
// because we shouldn't bother making a second call to get_e if there
// is another caller who has already invoked that call and is waiting
// on a response.
var async_cache = function(obj) {

  var responses = {},
      waiting = {},
      caching_obj = {},
      func = null;

  caching_obj.prototype = obj;

  function wrap_method(method, func) {
    return function(callback) {
      if (responses[method]) {
        // We already have this result, invoke callback immediately.
        callback(responses[method]);
      } else if (waiting[method]) {
        // Someone else is waiting for this result. Just add
        // ourself to the waiting list.
        waiting[method].push(callback);
      } else {
        // No one has asked for this yet. Store the caller's
        // callback on the waiting list and handle the response
        // callback by looping through the waiting list and
        // invoking all waiting callbacks.
        waiting[method] = [callback];
        func(function (val) {
          responses[method] = val;
          waiting[method].forEach(function(callback) {
            callback(val);
          });
          delete waiting[method];
        });
      }
    }
  }

  for (var method in obj) {
    func = obj[method];
    // Don't try cache magic for non-functions
    if (typeof func === 'function') {
      caching_obj[method] = wrap_method(method, func);
    }
  }

  return caching_obj;
}

// Wrap the slow disk with caching logic.
var smarter_slow_disk = async_cache(slow_disk);

// Also, up above we had two very similar looking chunks of code
// for the indirect sequences. Here's a helper for the indirect calls.
function get_indirect(indirect_method) {
  var self = smarter_slow_disk;
  return function (callback) {
    async_helper.sequence(
      self[indirect_method],
      function (callback, results) {
        self[results[0]](function (val) { callback(val); });
      },
      function (results) {
        callback(results[1]);
      }
    );
  };
};

// Finally, do our mixed example with caching.
var exhibit_f = exhibit('sequence/parallel mix with caching');
async_helper.parallel(
  get_indirect('get_a'),
  get_indirect('get_b'),
  smarter_slow_disk.get_c,
  smarter_slow_disk.get_d,
  smarter_slow_disk.get_e,
  function (results) {
    var sum = results.reduce(function (sum, val) { return sum + val; });
    exhibit_f.report('sum = ' + sum);
  }
);