Processes with supporting trampolining

processes.js

/**
 * The simplest processes (cooperative tasks) with Scheduler
 * (sort of Erlang's processes but without messages yet)
 * See: http://www.dabeaz.com/coroutines/index.html
 *
 * Deps: JS 1.7 with generators (yield)
 *
 * This version uses "trampolining" technique to request
 * data from other processes in synchronous manner.
 * See also: https://gist.github.com/1127536 (without "trampolining")
 *
 * by Dmitry Soshnikov <dmitry.soshnikov@gmail.com>
 * MIT Style License
 */

var Scheduler = (function initScheduler() {

  var

  /**
   * run queue; processes
   * ready to be executed
   */
  runQueue = [],

  /**
   * processes map -- all
   * currently alive processes
   */
  processes = {},

  /**
   * to track count of
   * alive processes
   */
  processesCount = 0,

  /**
   * Autoinc process id, which is
   * assigned to newborn process
   */
  pid = 1;

  var _toString = Object.prototype.toString;
  function isGenerator(object) {
    return _toString.call(object) == "[object Generator]";
  }

  /**
   * Processes constructor
   * @param {Generator: function|object} loop
   */
  function Process(loop) {
    this.loop = typeof loop == "function" ? loop() : loop;
    this.pid = pid++;
    this.value = undefined;
    this.stack = [];
    processes[this.pid] = this;
    processesCount++;
  }

  /**
   * Executes the process
   */
  Process.prototype.run = function () {
    try {
      while (true) {
        var result = this.loop.send(this.value);
        // if the generator is returned it means
        // we request some value from it; pass
        // the control to it using "trampolining", i.e.:
        if (isGenerator(result)) {
          // we suspend our execution saving our loop onto the stack
          this.stack.push(this.loop);
          // and pass the control to the requested process:
          // it will be called on the next while iteration,
          // so the technique is just to replace our loop
          this.loop = result;
          this.value = undefined;
        } else {
          //if there is nothing in the stack, it means
          // we reach the final requested result
          if (!this.stack.length) {
            return result;
          }
          // else: we returned from trampoline, so
          // restore our loop from the stack
          // and continue our execution; the result which
          // is returned from the generator is requested
          // previosly value -- we pass it to our process
          this.loop = this.stack.pop();
          this.value = result;
        }
      }
    } catch (e if (e instanceof StopIteration)) {
      throw e;
    }
  };

  return {

    /**
     * Currently running process,
     * used in self() function
     */
    runningProcess: null,

    /**
     * Sheduling is just putting a
     * process to the run queue; it will be
     * executed on next iteration of the handle loop
     */
    schedule: function (process) {
      runQueue.unshift(process);
    },

    /**
     * Creates new process and
     * automatically schedules it
     */
    spawn: function (processLoop) {
      var process = new Process(processLoop);
      console.log("* Spawning new process with pid #" + process.pid);
      this.schedule(process);
      return process.pid;
    },

    /**
     * Terminates a process by pid
     * or the process itself parameter
     */
    terminate: function (process) {
      var pid = typeof process == "number" ? process : process.pid;
      console.log("* Process #" + pid + " is terminated");
      delete processes[pid];
      processesCount--;
      return pid;
    },

    /**
     * Main scheduling loop: simply retrive a current process
     * from the run queue, execute it, and schedule back
     */
    handleLoop: function () {
      console.log("* Scheduler loop started");
      while (processesCount) {
        try {
          // get the next process in the run queue
          var process = runQueue.pop();
          // if there is something
          if (process) {
            // then run it
            Scheduler.runningProcess = process;
            process.run();  // execute the next "reductions" part (upto the next "yield" statement inside)
            // and schedule back for next iteration
            this.schedule(process);
          }
        } catch (e if (e instanceof StopIteration)) {
          // if the process finishes,
          // just remove it from the prcesses map
          this.terminate(process);
        }
      }
      console.log("* Scheduler loop stopped");
    }

  };

})();

// creates new process
function spawn(process) {
  return Scheduler.spawn(process);
}

// helper function to get
// the pid of currently running process
function self() {
  return Scheduler.runningProcess.pid;
}

// two helper functions to
// test trampolining technique
function calculate(value) {
  yield (yield nested(value)) + 10
}

function nested(data) {
  yield data + 10;
}

// spawn 3 processes
for (var k = 3; k--;) {
  spawn(function () {
    console.log("Process #" + self() + " first entry");
    yield;
    var x = yield calculate(self());
    console.log("Process #" + self() + " second entry: " + x);
  });
}

// start main scheduling loop
Scheduler.handleLoop();

// Results:

// * Spawning new process with pid #1
// * Spawning new process with pid #2
// * Spawning new process with pid #3
// * Scheduler loop started

// Process #1 first entry
// Process #2 first entry
// Process #3 first entry

// Process #1 second entry: 21

// * Process #1 is terminated

// Process #2 second entry: 22

// * Process #2 is terminated

// Process #3 second entry: 23

// * Process #3 is terminated
// * Scheduler loop stopped