Merge branch 'schedule'

Actually have a proper algorithmic scheduler in examples/jarvis.js now, allowing for low-latency responses and complex animation, etc.

2 files changed, 105 insertions, 2 deletions

diff --git a/examples/jarvis.js b/examples/jarvis.js
index 2dee842..0f0e1f1 100644
--- a/examples/jarvis.js
+++ b/examples/jarvis.js
@@ -7,7 +7,7 @@ const ms = require('../utils/measurespace');
 const vec = require('../utils/vec');
 const comb = require('../utils/comb');
 const Search = require('../tools/search');
-const Queue = require('../tools/queue');
+const sched = require('../tools/schedule');
 const ri = require('../utils/rectintersect');
 
 //// Basic handling on open and close of the socket
@@ -79,7 +79,7 @@ function tileHandler(send, source, tiles){
 }
 
 //// "userspace" functions for responses, like to tileUpdates
-var writes = new Queue(1000, 200, (elems) => main.write(elems));
+var writes = new sched.Queue(1000, 200, (elems) => main.write(elems));
 
 var command = 'jarvis'
 var sig     = 'feynmansfedora'
diff --git a/tools/schedule.js b/tools/schedule.js
new file mode 100644
index 0000000..033fddb
--- /dev/null
+++ b/tools/schedule.js
@@ -0,0 +1,103 @@
+// A mixed paradigm scheduler
+
+const bs = require('binary-search');
+
+// The core managed data type: a queue will be pushed a number of these and asked to manage it.
+exports.Job = function(data, prio=0, wt=1){
+  // Data is an array;
+  // Ideally, Both prio and wt are functions which take one argument: how many write cycles they've been waiting. This helps manage time-sensitive jobs.
+  // But this is too computationally difficult, so they are integer constants.
+
+  // Immutable Properties
+  this.data = data; //A set of work to be done (like character writes)
+  this.prio = prio; // If a.prio > b.prio, all of `a.data` will be sent before any of `b.data`
+  this.wt   = wt; // After numerous calls, (amount called of a/amount called of b) = a.wt/b.wt if a.prio = b.prio
+  this.maxr = () => this.data.length/this.wt; // A utility calculation: If a job has a lower maxr, it will run out of data earlier.
+
+  // Mutable Properties
+  this.wacc = 0; // mutable property: Queue will change this to keep track between dequeues of how much "left over" push real estate this should have.
+}
+
+exports.Queue = function(delayms, maxExport, call){
+  // Every delay ms, Queue executes `call(maxExport # of objs in queue)`
+  let jobs = {}; // Links priorities to unordered arrays of jobs
+  let prios = []; // Array of priorities (keys of jobs), sorted.
+  let open = true; // Is dequeue() allowed to be called (i.e., has the timeout expired?)
+  let disab = true; // Is the queue disabled?
+
+  this.enqueue = function(job){
+    if (! (job instanceof exports.Job)) job = new exports.Job(job);
+    let prio = job.prio;
+    if (!jobs[prio]) jobs[prio] = [];
+    jobs[prio].splice(0, Math.abs(bs(jobs[prio], job, (el, ne) => el.maxr() - ne.maxr())), job);
+    prios.splice(0, Math.abs(bs(prios, prio, (el, ne) => el-ne)), prio); // prios is meant to be sorted least to most, and each job layer is too (by "maximum number of rounds").
+    // These were sorted like this so that getNumOrAll could use [0] or [.length-1] or .pop instead of having to re-sort lists repetitively.
+    if (open) dequeue();
+  }
+  
+  this.enable = function(){
+    disab = false;
+    if (open) dequeue();
+  }
+
+  this.disable = function(){
+    disab = true;
+  }
+ 
+  function dequeue(){
+    // Wraps getNumOrAll, managing open/disab, and concatenating possibly multiple layers
+    if (disab){
+      open = true;
+      return;
+    }
+    if (prios.length == 0){
+      open = true;
+      return;
+    }
+    open = false;
+    let data = [];
+    while (prios.length > 0 && data.length < maxExport){
+      data.push(...getNumOrAll(prios[prios.length-1], maxExport-data.length));
+    }
+    call(data);
+    setTimeout(dequeue, delayms);
+  }
+
+  function getNumOrAll(prio, num){
+    // Step 1: (Pre-)sort by job.data.length/job.weight.
+    let jobq = jobs[prio];
+    let dequeued = [];
+
+    // Step 2: Start at lowest, and pop all until job.data.length>job.normweight*num (decreasing num as popping and recalc job.normweight). Delete the job. 
+    let weightsum = jobq.map(job => job.wt).reduce((acc, cur)=>acc+cur);
+    while (jobq[0] && jobq[0].data.length<(jobq[0].wacc+jobq[0].wt*num/weightsum)){
+      weightsum -= job.wt;
+      dequeued.push(...jobq.shift().data);
+    }
+
+    // Step 3: Then, pop job.normweight*num//1 elems from remaining, without num decrease or normweight recalc. But keep job.wacc = job.normweight*num%1
+    for (job of jobq){
+      job.wacc += job.wt*num/weightsum;
+      let data = job.data;
+      let topop = job.wacc-job.wacc%1;
+      job.wacc -= topop
+      dequeued.push(...data.splice(-topop));
+    }
+
+    // Step 4: Shallow copy job array, and sort by job.wacc.
+    for (job of jobq.splice().sort((el, ne) => el.wacc-ne.wacc)){
+      // Step 5: Iterate through array (high->low), and subtract 1 until the length of output is num.
+      if (dequeued.length == num) break;
+      job.wacc--;
+      dequeue.push(job.pop());
+    }
+
+    // Step 6: If empty, remove prio && jobs[prio]; return.
+    if (jobq.length == 0){
+      delete jobs[prio];
+      prios.splice(bs(prios, prio, (el, ne)=>el-ne), 1);
+    }
+
+    return dequeued;
+  }
+}