fix: high throughput streams

benchmarks/.gitignore

@@ -1,2 +1,5 @@
 # clinic.js
 **clinic**
+
+# generated test files
+fixtures

benchmarks/README.md

@@ -9,6 +9,9 @@ Benchmarks for both pull-mplex and libp2p-mplex
 
 ## Tests
 
+**Note**: If additional profiling is needed on the benchmarks, [clinicjs](https://clinicjs.org/documentation/) can
+be used in conjunction with the benchmark commands listed below.
+
 ### Echo with New Streams
 This test will send a ping message from the dialer to the listener, which is echo'd back to the dialer.
 A new stream will be created for each echo.
@@ -23,12 +26,29 @@ $ node bench.js --lib=pull-mplex
 benchPingPong*100: 8027.089ms
 benchPingPong*100: 7544.150ms
 benchPingPong*100: 7553.991ms
-benchPingPong*100: 8382.416ms
-benchPingPong*100: 8384.485ms
 $ node bench.js --lib=libp2p-mplex
 benchPingPong*100: 9571.866ms
 benchPingPong*100: 10309.212ms
 benchPingPong*100: 10629.806ms
-benchPingPong*100: 10558.181ms
-benchPingPong*100: 10839.580ms
+
+```
+
+### Write Large Files
+This test will send the contents of a large file. The recipient will echo the file back to the dialer.
+
+**Setup**: Before running the large file benchmark you will need to create the files. You can do this by running ```npm run setup```.
+
+All flags can be omitted aside from `--lib`. The defaults are shown here for reference.
+**pull-mplex**: `node large-file.js --lib=pull-mplex --repeat=1 --runs=3`
+**libp2p-mplex**: `node large-file.js --lib=libp2p-mplex --repeat=1 --runs=3`
+
+```sh
+$ node large-file.js --lib=pull-mplex --repeat=1 --runs=3
+sendFile*1: 379.209ms
+sendFile*1: 291.188ms
+sendFile*1: 314.664ms
+$ node large-file.js --lib=libp2p-mplex --repeat=1 --runs=3
+sendFile*1: 516.938ms
+sendFile*1: 450.578ms
+sendFile*1: 433.172ms
 ```

benchmarks/bench.js

@@ -88,7 +88,10 @@ function buildPingPong (callback) {
     pull(
       pull.values(['ping']),
       stream,
-      pull.onEnd(cb)
+      pull.onEnd((err) => {
+        if (err) console.error(err)
+        cb(err)
+      })
     )
   }
 }

benchmarks/large-file.js

@@ -0,0 +1,107 @@
+'use strict'
+
+const path = require('path')
+const minimist = require('minimist')
+const bench = require('fastbench')
+const net = require('net')
+
+const pull = require('pull-stream')
+const toPull = require('stream-to-pull-stream')
+const pullFile = require('pull-file')
+
+const { files } = require('./util')
+
+let repeat
+let runs
+
+const argv = minimist(process.argv.slice(2), {
+  boolean: 'child',
+  default: {
+    child: true,
+    port: 3000,
+    host: 'localhost',
+    lib: 'pull-mplex',
+    repeat: 1,
+    runs: 3
+  }
+})
+
+/**
+ * Starts the needed servers, creates the connections and then
+ * calls back with the benchmark function to execute
+ * @param {function(function)} callback
+ */
+function buildSendFile (callback) {
+  let dialer
+  const mplex = require(argv.lib || 'pull-mplex')
+  repeat = argv.repeat
+  runs = argv.runs
+
+  start(argv)
+
+  function startServer (addr, cb) {
+    const server = net.createServer(function (socket) {
+      const connection = toPull.duplex(socket)
+      const listener = mplex.listener(connection)
+
+      listener.on('stream', (stream) => {
+        pull(
+          stream,
+          stream
+        )
+      })
+    })
+    server.listen(addr.port, function (err) {
+      if (err) throw err
+      cb()
+    })
+  }
+
+  function start (addr) {
+    startServer(addr, function () {
+      // Create the dialer
+      dialer = net.connect(addr.port, addr.host)
+      dialer.on('connect', function () {
+        const connection = toPull.duplex(dialer)
+        dialer = mplex.dialer(connection)
+        callback(null, sendFile)
+      })
+    })
+  }
+
+  function sendFile (cb) {
+    const stream = dialer.newStream((err) => {
+      if (err) console.log(err)
+    })
+
+    const inputFile = path.join(__dirname, `/fixtures/${files[0].name}.txt`)
+
+    pull(
+      pullFile(inputFile, { bufferSize: 1 << 20 }),
+      stream,
+      pull.collect(cb)
+    )
+  }
+}
+
+function times (num, run, cb) {
+  if (--num < 0) return cb()
+  run(function (err) {
+    if (err) throw err
+
+    times(num, run, cb)
+  })
+}
+
+buildSendFile(function (err, sendFile) {
+  if (err) throw err
+
+  // run sendFile `repeat` many times
+  const run = bench([sendFile], repeat)
+
+  // Do it `runs` many times
+  times(runs, run, function () {
+    // close the sockets the bad way
+    process.exit(0)
+  })
+})

benchmarks/package.json

@@ -4,8 +4,11 @@
   "description": "",
   "main": "bench.js",
   "scripts": {
+    "setup": "node util/create-files",
     "pull-mplex": "node bench.js --lib=pull-mplex",
-    "libp2p-mplex": "node bench.js --lib=libp2p-mplex"
+    "libp2p-mplex": "node bench.js --lib=libp2p-mplex",
+    "lg-pull-mplex": "node large-file.js --lib=pull-mplex",
+    "lg-libp2p-mplex": "node large-file.js --lib=libp2p-mplex"
   },
   "keywords": [],
   "author": "",
@@ -15,6 +18,7 @@
     "fastparallel": "^2.3.0",
     "libp2p-mplex": "~0.8.4",
     "minimist": "^1.2.0",
+    "pull-file": "^1.1.0",
     "pull-mplex": "../",
     "pull-stream": "^3.6.9",
     "stream-to-pull-stream": "^1.7.2"

benchmarks/util/create-files.js

@@ -0,0 +1,23 @@
+'use strict'
+
+const { generateFiles, verifyTestFiles } = require('./')
+
+/**
+ * This utility will verify or create files needed for the tests.
+ *
+ * @async
+ * @function verifyAndCreateFiles
+ */
+const verifyAndCreateFiles = async () => {
+  const valid = await verifyTestFiles()
+  if (!valid) {
+    console.log('Some files missing.  Generating files')
+    await generateFiles()
+  } else {
+    console.log('Files Verified')
+  }
+}
+if (require.main === module) {
+  verifyAndCreateFiles()
+}
+module.exports = verifyAndCreateFiles

benchmarks/util/index.js

@@ -0,0 +1,115 @@
+'use strict'
+
+const path = require('path')
+const crypto = require('crypto')
+const util = require('util')
+const fs = require('fs')
+const fsWriteFile = util.promisify(fs.writeFile)
+const fsMakeDir = util.promisify(fs.mkdir)
+const fsExists = util.promisify(fs.access)
+const fsStat = util.promisify(fs.lstat)
+const fsReadDir = util.promisify(fs.readdir)
+const KB = 1024
+const MB = KB * 1024
+
+const files = [{
+  size: 128 * MB,
+  name: '128MBFile'
+}, {
+  size: 64 * MB,
+  name: '64MBFile'
+}]
+
+async function generateFiles () {
+  const testPath = path.join(__dirname, `../fixtures/`)
+  for (let file of files) {
+    if (file.count) {
+      try {
+        await fsExists(`${testPath}${file.name}`)
+      } catch (err) {
+        await fsMakeDir(`${testPath}${file.name}`)
+      }
+      for (let i = 0; i < file.count; i++) {
+        write(crypto.randomBytes(file.size), `${file.name}/${file.name}-${i}`)
+      }
+    } else {
+      write(crypto.randomBytes(file.size), file.name)
+    }
+  }
+}
+
+async function write (data, name) {
+  await fsWriteFile(path.join(__dirname, `../fixtures/${name}.txt`), data)
+  console.log(`File ${name} created.`)
+}
+
+async function file (name) {
+  const isDir = await isDirectory(name.toLowerCase())
+  if (!isDir) {
+    const file = files.find((file) => {
+      return file.name === name.toLowerCase()
+    })
+    if (typeof file !== 'undefined' && file) {
+      return path.join(__dirname, `../fixtures/${file.name}.txt`)
+    } else {
+      if (name.includes(`/`)) {
+        return path.join(__dirname, `../fixtures/${name.toLowerCase()}`)
+      } else {
+        return file
+      }
+    }
+  } else {
+    const arr = await fsReadDir(path.join(__dirname, `../fixtures/${name.toLowerCase()}`))
+    const fullPath = arr.map((fileName) => {
+      return path.join(__dirname, `../fixtures/${name.toLowerCase()}/${fileName.toLowerCase()}`)
+    })
+    return fullPath
+  }
+}
+
+async function isDirectory (name) {
+  try {
+    const dir = path.join(__dirname, `../fixtures/${name.toLowerCase()}`)
+    const stats = await fsStat(dir)
+    return stats.isDirectory()
+  } catch (e) {
+    return false
+  }
+}
+
+async function verifyTestFiles () {
+  const fixtures = path.join(__dirname, `../fixtures`)
+  try {
+    await fsExists(fixtures)
+  } catch (e) {
+    await fsMakeDir(fixtures)
+  }
+  for (let f of files) {
+    if (f.count) {
+      console.log(`Verifying Directory ${f.name}`)
+      const dir = await isDirectory(f.name)
+      if (dir) {
+        const fileArr = file(f.name)
+        if (fileArr.length < f.count) {
+          console.log(`Missing files in directory ${f.name}`)
+          return false
+        }
+      } else {
+        console.log(`Missing directory ${f.name}`)
+        return false
+      }
+    } else {
+      const filePath = await file(f.name)
+      try {
+        console.log(`Verifying File ${f.name}`)
+        await fsExists(filePath)
+      } catch (err) {
+        console.log(`Missing ${f.name}`)
+        return false
+      }
+    }
+  }
+  return true
+}
+
+module.exports = { generateFiles, verifyTestFiles, files }

src/channel.js

@@ -135,7 +135,7 @@ class Channel extends EE {
    * @param {Buffer|string} data Logged with the metadata. Must be `.toString` capable. Default: `''`
    */
   _log (name, data) {
-    if (!debug.enabled) return
+    if (!log.enabled) return
     log({
       op: name,
       name: this._name,