Expanded for NodeConf EU

code-surfer.js

@@ -1,14 +1,14 @@
 import React from 'react'
 import { CodeSurfer } from "mdx-deck-code-surfer"
-import vsDarkPlus from "prism-react-renderer/themes/vsDarkPlus"
+// import vsDarkPlus from "prism-react-renderer/themes/vsDarkPlus"
 
 const CustomCodeSurfer = ({ title, code, steps }) => (
   <CodeSurfer
     title={title}
     code={code}
     lang="javascript"
     showNumbers={false}
-    theme={vsDarkPlus}
+    // theme={vsDarkPlus}
     steps={steps}
   />
 )

assets/async-iterator-pipe.js → examples/async-iterator-pipe.js

@@ -9,11 +9,13 @@ function pipeDirect (source, target) {
   if (target.write && target.end) {
     pipeStream(source, target)
     return target
-  } else if (typeof target === 'function') {
+  }
+
+  if (typeof target === 'function') {
     return target(source)
-  } else {
-    throw new Error('Unrecognized target type')
   }
+
+  throw new Error('Unrecognized target type')
 }
 
 function pipe (source, ...transforms) {

examples/microtask-priority.js

@@ -0,0 +1,19 @@
+async function* doUntilTimeout (task, ms) {
+  let running = true
+  setTimeout(() => {
+    running = false
+  }, ms)
+
+  while (running) {
+    yield await task()
+  }
+}
+
+let n = 0
+const numbers = doUntilTimeout(() => n++, 1000)
+
+let sum = 0
+for await (let number of numbers) {
+  sum += number
+}
+console.log(sum)

assets/stream-line-split.js → examples/stream-line-split.js

@@ -1,31 +1,38 @@
 const pipe = require('async-iterator-pipe')
 
 async function* lineSplit (iterator) {
-  let buffer = ''
+  let buffer = Buffer.alloc(0)
+
   for await (let chunk of iterator) {
-    buffer += chunk.toString()
-    let position = buffer.indexOf('\n')
+    buffer = Buffer.concat([ buffer, chunk ])
+
+    let position = buffer.indexOf(0x0a)
     while (position >= 0) {
       yield buffer.slice(0, position)
+
       buffer = buffer.slice(position + 1)
-      position = buffer.indexOf('\n')
+      position = buffer.indexOf(0x0a)
     }
   }
 }
 
 async function* csv (iterator) {
   let keys
+
   for await (let line of iterator) {
-    const values = line.split(',')
+    const values = line.toString().split(',')
+
     if (!keys) {
       keys = values
-    } else {
-      const data = {}
-      for (let i = 0; i < values.length; i++) {
-        data[keys[i]] = values[i]
-      }
-      yield data
+      continue
+    }
+
+    const data = {}
+    for (let i = 0; i < values.length; i++) {
+      data[keys[i]] = values[i]
     }
+
+    yield data
   }
 }
 

intro/about.js

@@ -0,0 +1,5 @@
+module.exports = {
+  "name": "Stephen Belanger",
+  "github": "https://github.com/qard",
+  "twitter": "https://twitter.com/stephenbelanger"
+}

package.json

@@ -13,10 +13,17 @@
   "author": "Stephen Belanger <[email protected]> (https://github.com/qard)",
   "license": "MIT",
   "devDependencies": {
-    "mdx-deck": "^1.8.2",
+    "async-iterator-pipe": "^1.0.1",
+    "blessed": "^0.1.81",
+    "blessed-contrib": "^4.8.18",
+    "mdx-deck": "^1.10.2",
     "mdx-deck-code-surfer": "^0.5.5",
-    "now": "^13.1.3",
-    "prism-react-renderer": "^0.1.5",
-    "raw-loader": "^1.0.0"
-  }
+    "now": "^16.3.1",
+    "prism-react-renderer": "^1.0.1",
+    "raw-loader": "^3.1.0",
+    "relative-microtime": "^2.0.0",
+    "styled-components": "^4.4.0",
+    "terminal-in-react": "^4.3.1"
+  },
+  "dependencies": {}
 }

performance/capped-array.js

@@ -0,0 +1,18 @@
+class CappedArray extends Array {
+  constructor(limit) {
+    super()
+    this.limit = limit
+    for (let i = 0; i < limit; i++) {
+      this.push(0)
+    }
+  }
+
+  push (value) {
+    super.push(value)
+    while (this.length > this.limit) {
+      this.shift()
+    }
+  }
+}
+
+module.exports = CappedArray

performance/container.js

@@ -0,0 +1,7 @@
+const test = process.argv[2]
+const mode = process.argv[3]
+
+const producer = require(`./${test}/producer`)
+const run = require(`./${test}/${mode}`)
+
+run(producer(), process.stdout)

performance/csv-to-json/async-iterator.js

@@ -0,0 +1,41 @@
+const pipe = require('async-iterator-pipe')
+const { splitLine, zipKeyValueLists } = require('./helper')
+
+async function* csv (iterator) {
+  let buffer = Buffer.alloc(0)
+  let keys
+
+  for await (let chunk of iterator) {
+    buffer = Buffer.concat([ buffer, chunk ])
+
+    let position = buffer.indexOf(0x0a)
+    while (position >= 0 && buffer.length) {
+      const line = buffer.slice(0, position).toString()
+
+      const values = splitLine(line)
+      if (!keys) {
+        keys = values
+      } else {
+        yield zipKeyValueLists(keys, values)
+      }
+
+      buffer = buffer.slice(position + 1)
+      position = buffer.indexOf(0x0a)
+    }
+  }
+}
+
+async function* toJSON (iterator) {
+  for await (let item of iterator) {
+    yield `${JSON.stringify(item)}\n`
+  }
+}
+
+module.exports = (input, output) => {
+  pipe(
+    input,
+    csv,
+    toJSON,
+    output
+  )
+}

performance/csv-to-json/helper.js

@@ -0,0 +1,15 @@
+module.exports = {
+  splitLine (line) {
+    return line
+      .match(/\s*(?:([^,]+)|"([^,]+)")/g)
+      .map(value => value.trim().replace(/"/g, ''))
+  },
+
+  zipKeyValueLists (keys, values) {
+    const data = {}
+    for (let i = 0; i < values.length; i++) {
+      data[keys[i]] = values[i]
+    }
+    return data
+  }
+}

performance/csv-to-json/producer.js

@@ -0,0 +1,43 @@
+const { Readable } = require('stream')
+
+module.exports = (headerCount = 10, batchSize = 10) => {
+  const headers = []
+  for (let i = 0; i < headerCount; i++) {
+    headers.push(`header_${i}`)
+  }
+  const headerBuffer = Buffer.from(`${headers.join(',')}\n`)
+
+  function makeLine () {
+    const values = []
+    for (let i = 0; i < headerCount; i++) {
+      values.push(Math.random() * 1000)
+    }
+    return values.join(',')
+  }
+
+  function makeLines (n) {
+    const lines = []
+    for (let i = 0; i < n; i++) {
+      lines.push(makeLine())
+    }
+    return lines.join('\n')
+  }
+
+  let buffer = Buffer.from(headerBuffer)
+
+  return new Readable({
+    read (size) {
+      while (buffer.length < size) {
+        buffer = Buffer.concat([
+          buffer,
+          Buffer.from(makeLines(batchSize))
+        ])
+      }
+
+      setImmediate(() => {
+        this.push(buffer.slice(0, size))
+        buffer = buffer.slice(size + 1)
+      })
+    }
+  })
+}

performance/csv-to-json/stream.js

@@ -0,0 +1,49 @@
+const { Transform, pipeline } = require('stream')
+const { splitLine, zipKeyValueLists } = require('./helper')
+
+const csv = new Transform({
+  readableObjectMode: true,
+  transform (data, encoding, callback) {
+    if (!this.buffer) this.buffer = Buffer.alloc(0)
+    let buffer = Buffer.concat([ this.buffer, data ])
+
+    let position = buffer.indexOf(0x0a)
+    while (position >= 0 && buffer.length) {
+      const line = buffer.slice(0, position).toString()
+
+      const values = splitLine(line)
+      if (!this.headers) {
+        this.headers = values
+      } else {
+        this.push(zipKeyValueLists(this.headers, values))
+      }
+
+      buffer = buffer.slice(position + 1)
+      position = buffer.indexOf(0x0a)
+    }
+
+    if (buffer.length) {
+      this.buffer = buffer
+    }
+
+    setImmediate(callback)
+  }
+})
+
+const ldjson = new Transform({
+  writableObjectMode: true,
+  transform (data, encoding, callback) {
+    this.push(`${JSON.stringify(data)}\n`, encoding)
+    setImmediate(callback)
+  }
+})
+
+module.exports = (input, output) => {
+  pipeline(
+    input,
+    csv,
+    ldjson,
+    output,
+    err => console.error(err)
+  )
+}

performance/make-line.js

@@ -0,0 +1,25 @@
+const MeasureStream = require('./measure-stream')
+
+// const { join } = require('path')
+// const { spawn } = require('child_process')
+// 
+// module.exports = (test, mode) => {
+//   const gauge = new Measure(mode)
+// 
+//   const proc = spawn('node', ['container.js', test, mode], {
+//     cwd: __dirname
+//   })
+//   proc.stdout.pipe(gauge)
+// 
+//   return gauge.data
+// }
+
+module.exports = (test, mode) => {
+  const gauge = new MeasureStream(mode)
+
+  const producer = require(`./${test}/producer`)
+  const run = require(`./${test}/${mode}`)
+  run(producer(), gauge)
+
+  return gauge.data
+}

performance/measure-stream.js

@@ -0,0 +1,40 @@
+const { Writable } = require('stream')
+const microtime = require('relative-microtime')
+
+const CappedArray = require('./capped-array')
+
+const colors = ['red', 'yellow']
+let n = 0
+
+class MeasureStream extends Writable {
+  constructor(mode) {
+    super()
+    this.timer = microtime()
+
+    this.data = {
+      title: mode,
+      x: new CappedArray(50),
+      y: new CappedArray(50),
+      style: {
+        line: colors[n++]
+      }
+    }
+  }
+
+  _write (chunk, encoding, callback) {
+    const { x, y } = this.data
+    const now = this.timer() // ns
+    const bucket = Math.floor(now / 1000 / 1000)
+
+    if (!x.length || x[x.length - 1] !== bucket) {
+      x.push(bucket)
+      y.push(chunk.length)
+    } else {
+      y[y.length - 1] = chunk.length + y[y.length - 1]
+    }
+
+    setImmediate(callback)
+  }
+}
+
+module.exports = MeasureStream

performance/notes.md

@@ -0,0 +1,4 @@
+1. show upper test together to demonstrate async-iterators are equal or slightly faster
+2. show upper test separately to confirm absolute performance in isolation
+3. show csv-to-json separately first to show absolute performance
+4. show csv-to-json together to demonstrate issue of microtask queue starving Node.js idle queue