Skip working on low priority tasks because of a new oncoming task.

site/playground.html

@@ -24,8 +24,8 @@
   height: window.innerHeight,
   data: [
     { fn: 'x^2', nSamples: window.innerWidth*5, sampler: 'asyncInterval' },
-    { fn: 'x^2+1', nSamples: window.innerWidth*5, sampler: 'asyncInterval' },
-    { fn: 'x^2+2', nSamples: window.innerWidth*5, sampler: 'asyncInterval' },
+    { fn: 'sin(x)', nSamples: window.innerWidth*5, sampler: 'asyncInterval' },
+    { fn: '1/x', nSamples: window.innerWidth*5, sampler: 'asyncInterval' },
   ]
 })
 </script>

src/chart.ts

@@ -94,6 +94,11 @@ export class Chart extends EventEmitter.EventEmitter {
   private linkedGraphs: Array<Chart>
   private line: Line<[number, number]>
 
+  /**
+   * The number of times a function was rendered.
+   */
+  private generation: number
+
   /**
    * `svg` element that holds the graph (canvas + title + axes)
    */
@@ -131,6 +136,7 @@ export class Chart extends EventEmitter.EventEmitter {
     Chart.cache[this.id] = this
     this.linkedGraphs = [this]
     this.meta = {}
+    this.generation = 0
     this.setUpEventListeners()
   }
 
@@ -555,10 +561,15 @@ export class Chart extends EventEmitter.EventEmitter {
       // additional options needed in the graph-types/helpers
       d.index = index
 
+      // (hidden property)
+      // @ts-ignore
+      d.generation = self.generation
+
       const selection = d3Select(this)
       selection.call(globals.graphTypes[d.graphType](self))
       selection.call(helpers(self))
     })
+    this.generation += 1
   }
 
   buildZoomHelper() {

src/perf/interval-pipeline.ts

@@ -45,7 +45,7 @@ async function createData(nSamples: number, async: boolean) {
 
 async function compileAndEval() {
   const bench = new Bench()
-  const nSamples = 1000
+  const nSamples = 1800 /* window.innerWidth */
   globals.workerPool = new IntervalWorkerPool(8)
   bench.add(`compile and eval ${nSamples}`, async function () {
     await createData(nSamples, false)

src/samplers/interval.ts

@@ -30,19 +30,17 @@ async function asyncInterval1d({
   //
   // See more useful math in the utils tests
   const step = (absHi - absLo) / (nSamples - 1)
-  const nGroups = 1
+  const nGroups = 4
   const groupSize = (nSamples - 1) / nGroups
   const promises: Array<Promise<ArrayBuffer>> = []
   const interval2dTypedArrayGroups = interval2dTypedArray(nSamples, nGroups)
   for (let i = 0; i < nGroups; i += 1) {
     const lo = absLo + step * groupSize * i
     const hi = absLo + step * groupSize * (i + 1)
-    // console.log('nSamples', nSamples)
-    // console.log(absLo, absHi)
-    // console.log(lo, hi)
-    // console.log('groupSize', groupSize)
     // Transfers the typed arrays to the worker threads.
-    promises.push(workerPoolInterval.queue({ d, lo, hi, n: groupSize + 1, interval2d: interval2dTypedArrayGroups[i] }))
+    promises.push(
+      workerPoolInterval.queue({ d, nGroup: i, lo, hi, n: groupSize + 1, interval2d: interval2dTypedArrayGroups[i] })
+    )
   }
 
   const allWorkersDone = await Promise.all(promises)
@@ -56,12 +54,15 @@ async function asyncInterval1d({
     for (let j = 0; j < group.length; j += 4) {
       const x = { lo: group[j + 0], hi: group[j + 1] }
       const y = { lo: group[j + 2], hi: group[j + 3] }
-      if (y.lo === -Infinity && y.hi === Infinity) {
+      if (y.lo === Infinity && y.hi === -Infinity) {
+        // interval is empty
+        continue
+      } else if (y.lo === -Infinity && y.hi === Infinity) {
         // skip whole interval
         samples.push(null)
-        continue
+      } else {
+        samples.push([x, y])
       }
-      samples.push([x, y])
     }
   }
 

src/samplers/interval.worker.mjs

@@ -24,8 +24,8 @@ self.onmessage = ({ data }) => {
   for (; outIdx < out.length; outIdx += 4) {
     out[outIdx + 0] = 0
     out[outIdx + 1] = 0
-    out[outIdx + 2] = -Infinity
-    out[outIdx + 3] = Infinity
+    out[outIdx + 2] = Infinity
+    out[outIdx + 3] = -Infinity
   }
   self.postMessage({ interval2d: out, nTask }, [out.buffer])
 }

src/samplers/interval_worker_pool.ts

@@ -6,23 +6,27 @@ interface IntervalTask {
   lo: number
   hi: number
   n: number
+  nGroup: number
   interval2d: Float32Array
 
   // internal
+  valid?: boolean
   nTask?: number
 }
 
 class IntervalWorkerPool {
   private tasks: Array<IntervalTask>
   private idleWorkers: Array<Worker>
   private resolves: Map<number, (value: any) => void>
+  private rejects: Map<number, (value: any) => void>
   private nTasks: number
 
   constructor(nThreads: number) {
     this.nTasks = 0
     this.idleWorkers = []
     this.tasks = []
     this.resolves = new Map()
+    this.rejects = new Map()
 
     for (let i = 0; i < nThreads; i += 1) {
       // NOTE: new URL(...) cannot be a variable!
@@ -50,9 +54,20 @@ class IntervalWorkerPool {
 
   queue(task: IntervalTask): Promise<ArrayBuffer> {
     task.nTask = this.nTasks
+    task.valid = true
+
+    for (let i = 0; i < this.tasks.length; i += 1) {
+      if (this.tasks[i].d.index === task.d.index && this.tasks[i].nGroup === task.nGroup) {
+        this.tasks[i].valid = false
+      }
+    }
+
+    // new task
     this.tasks.push(task)
-    const p: Promise<ArrayBuffer> = new Promise((resolve) => {
+
+    const p: Promise<ArrayBuffer> = new Promise((resolve, reject) => {
       this.resolves[task.nTask] = resolve
+      this.rejects[task.nTask] = reject
     })
     this.nTasks += 1
     this.drain()
@@ -62,8 +77,12 @@ class IntervalWorkerPool {
   drain() {
     while (this.hasWork()) {
       const task = this.tasks.shift()
+      if (!task.valid) {
+        return
+      }
       const idleWorker = this.idleWorkers.shift()
 
+      // console.log(`working on task ${task.nTask}`)
       const dStripped: any = {}
       dStripped.fn = task.d.fn
       dStripped.scope = task.d.scope