diagnostics channel prototype

bench.js

const diagnosticsChannel = require('./diagnostics-channel')

const iterations = 1e7

function time(fn) {
  fn(iterations / 10) // warmup
  const start = process.hrtime.bigint()
  fn(iterations)
  const end = process.hrtime.bigint()
  const diff = end - start
  const seconds = Number(diff) / 1e9
  const opsPerSecond = 1 / seconds * iterations
  return { start, end, diff, seconds, opsPerSecond }
}

function durationMessage({ seconds, opsPerSecond }) {
  return `${seconds}s or ${opsPerSecond} ops per second`
}

function timeTest(name, fn) {
  console.log(`${name} took ${durationMessage(time(fn))}`)
}

function timeDiffTest(name, baseline, fn) {
  const baselineResult = time(baseline)
  const fnResult = time(fn)
  const overheadSeconds = fnResult.seconds - baselineResult.seconds
  console.log(`${name} baseline took ${durationMessage(baselineResult)}, and test took ${durationMessage(fnResult)}, overhead time is ${overheadSeconds}`)
}

timeDiffTest('overhead without subscribers', iterations => {
  let count = 0
  for (let i = 0; i < iterations; i++) {
    count++
  }
  return count
}, iterations => {
  let count = 0
  const publisher = diagnosticsChannel.publisher('test')

  for (let i = 0; i < iterations; i++) {
    count++
    if (publisher.shouldPublish()) {
      publisher.publish({ foo: 'bar' })
    }
  }
  return count
})

function testByPublisherCount(count) {
  console.log(`=== ${count} publishers ===`)

  let subscriber = diagnosticsChannel.subscriber(/publisher\.\.+/)

  for (let i = 0; i < count; i++) {
    diagnosticsChannel.publisher(`publisher.${i}`)
  }

  timeTest('diagnosticsChannel.subscribe with string', iterations => {
    for (let i = 0; i < iterations; i++) {
      diagnosticsChannel.subscribe('publisher.0', () => {})
    }
  })

  diagnosticsChannel._channels = {}
  diagnosticsChannel._patterns = []

  subscriber = diagnosticsChannel.subscriber(/publisher\.\.+/)

  for (let i = 0; i < count; i++) {
    diagnosticsChannel.publisher(`publisher.${i}`)
  }

  timeTest('diagnosticsChannel.subscribe with regex', iterations => {
    for (let i = 0; i < iterations; i++) {
      diagnosticsChannel.subscribe(/publisher\.\.+/, () => {})
    }
  })

  diagnosticsChannel._channels = {}
  diagnosticsChannel._patterns = []

  subscriber = diagnosticsChannel.subscriber(/publisher\.\.+/)

  for (let i = 0; i < count; i++) {
    diagnosticsChannel.publisher(`publisher.${i}`)
  }

  timeTest('subscriber.subscribe', iterations => {
    for (let i = 0; i < iterations; i++) {
      subscriber.subscribe(() => {})
    }
  })

  diagnosticsChannel._channels = {}
  diagnosticsChannel._patterns = []
}

testByPublisherCount(0)
for (let count = 8; count <= 256; count *= 8) {
  testByPublisherCount(count)
}
function testBySubscriberCount(count) {
  console.log(`=== ${count} subscribers ===`)

  const subscriber = diagnosticsChannel.subscriber(/fs\.\w+/)
  const publisher = diagnosticsChannel.publisher('fs.readFile')

  for (let i = 0; i < count; i++) {
    subscriber.subscribe(() => { })
  }

  timeTest('diagnosticsChannel.shouldPublish + diagnosticsChannel.publish', iterations => {
    for (let i = 0; i < iterations; i++) {
      if (publisher.shouldPublish('fs.readFile')) {
        diagnosticsChannel.publish('fs.readFile', { foo: 'bar' })
      }
    }
  })

  timeTest('publisher.publishIfActive', iterations => {
    for (let i = 0; i < iterations; i++) {
      publisher.publishIfActive(() => ({ foo: 'bar' }))
    }
  })

  timeTest('publisher.shouldPublish + publisher.publish', iterations => {
    for (let i = 0; i < iterations; i++) {
      if (publisher.shouldPublish()) {
        publisher.publish({ foo: 'bar' })
      }
    }
  })
}

testBySubscriberCount(0)
for (let count = 1; count <= 16; count *= 2) {
  testBySubscriberCount(count)
}

diagnostics-channel.js

const assert = require('assert')

class Subscribable {
  _subscribers = []
  active = false

  subscribe(subscription) {
    this._subscribers.push(subscription)
    this.active = true
  }
}

class Publisher extends Subscribable {
  constructor(name) {
    super()

    Object.defineProperty(this, 'name', {
      value: name
    })
  }

  shouldPublish() {
    return this.active
  }

  publishIfActive(fn) {
    if (this.active) {
      return this.publish(fn(this.name))
    }
    return false
  }

  publish(data) {
    for (const subscription of this._subscribers) {
      if (subscription(this.name, data) === false) {
        return false
      }
    }
    return true
  }
}

class Subscriber extends Subscribable {
  _watches = []

  constructor(pattern) {
    super()
    this._pattern = pattern
  }

  _watch(channel) {
    this._watches.push(channel)
    for (const subscription of this._subscribers) {
      channel.subscribe(subscription)
    }
  }

  subscribe(subscription) {
    for (const channel of this._watches) {
      channel.subscribe(subscription)
    }
    return super.subscribe(subscription)
  }

  _test(name) {
    return this._pattern.test(name)
  }

  _equal(pattern) {
    const { source, flags } = this._pattern
    return source === pattern.source && flags === pattern.flags
  }
}

class DiagnosticsChannel {
  _channels = {}
  _patterns = []

  publisher(name) {
    assert.equal(typeof name, 'string', 'publisher name must be a string')
    let channel = this._channels[name]
    if (channel) return channel

    channel = new Publisher(name)

    // New channel should be watched by existing matching patterns
    for (const pattern of this._patterns) {
      if (pattern._test(name)) {
        pattern._watch(channel)
      }
    }

    this._channels[name] = channel
    return channel
  }

  subscriber(pattern) {
    if (typeof pattern === 'string') {
      return this.publisher(pattern)
    }

    assert.ok(pattern instanceof RegExp,
      'subscriber pattern must be a string or regular expression')

    for (const patternChannel of this._patterns) {
      if (patternChannel._equal(pattern)) {
        return patternChannel
      }
    }

    const subscriber = new Subscriber(pattern)

    // New pattern should watch existing matching channels
    for (const channel of Object.values(this._channels)) {
      if (subscriber._test(channel.name)) {
        subscriber._watch(channel)
      }
    }

    this._patterns.push(subscriber)
    return subscriber
  }

  subscribe(pattern, subscriber) {
    return this.subscriber(pattern).subscribe(subscriber)
  }

  publish(name, data) {
    return this.publisher(name).publish(data)
  }

  shouldPublish(name) {
    return this.publisher(name).shouldPublish()
  }
}

module.exports = new DiagnosticsChannel()

results.txt

overhead without subscribers baseline took 0.010120932s or 988051298.0425124 ops per second, and test took 0.021208512s or 471508797.9769633 ops per second, overhead time is 0.01108758
=== 0 publishers ===
diagnosticsChannel.subscribe with string took 1.692570354s or 5908173.90625288 ops per second
diagnosticsChannel.subscribe with regex took 1.854457491s or 5392412.632013251 ops per second
subscriber.subscribe took 2.240160106s or 4463966.648283844 ops per second
=== 8 publishers ===
diagnosticsChannel.subscribe with string took 1.400673042s or 7139424.905130714 ops per second
diagnosticsChannel.subscribe with regex took 2.549689325s or 3922046.4634451102 ops per second
subscriber.subscribe took 1.622495676s or 6163344.622682372 ops per second
=== 64 publishers ===
diagnosticsChannel.subscribe with string took 1.900400764s or 5262047.979265072 ops per second
diagnosticsChannel.subscribe with regex took 1.900812859s or 5260907.170662192 ops per second
subscriber.subscribe took 2.212121474s or 4520547.410046977 ops per second
=== 0 subscribers ===
diagnosticsChannel.shouldPublish + diagnosticsChannel.publish took 0.019346825s or 516880676.8035582 ops per second
publisher.publishIfActive took 0.031522368s or 317235050.3616987 ops per second
publisher.shouldPublish + publisher.publish took 0.019751389s or 506293506.75033534 ops per second
=== 1 subscribers ===
diagnosticsChannel.shouldPublish + diagnosticsChannel.publish took 0.145752561s or 68609429.09949967 ops per second
publisher.publishIfActive took 0.063335529s or 157889263.07065344 ops per second
publisher.shouldPublish + publisher.publish took 0.047820078s or 209117182.9539885 ops per second
=== 2 subscribers ===
diagnosticsChannel.shouldPublish + diagnosticsChannel.publish took 0.159481455s or 62703215.24217346 ops per second
publisher.publishIfActive took 0.097503228s or 102560707.01577184 ops per second
publisher.shouldPublish + publisher.publish took 0.074112814s or 134929433.3905605 ops per second
=== 4 subscribers ===
diagnosticsChannel.shouldPublish + diagnosticsChannel.publish took 0.222900968s or 44862972.51073401 ops per second
publisher.publishIfActive took 0.142058528s or 70393521.18304366 ops per second
publisher.shouldPublish + publisher.publish took 0.118130945s or 84651824.29548836 ops per second
=== 8 subscribers ===
diagnosticsChannel.shouldPublish + diagnosticsChannel.publish took 0.280333796s or 35671760.38953219 ops per second
publisher.publishIfActive took 0.242853519s or 41177085.023009285 ops per second
publisher.shouldPublish + publisher.publish took 0.202397955s or 49407613.82692824 ops per second
=== 16 subscribers ===
diagnosticsChannel.shouldPublish + diagnosticsChannel.publish took 0.512226882s or 19522598.972070348 ops per second
publisher.publishIfActive took 0.502907779s or 19884361.34331499 ops per second
publisher.shouldPublish + publisher.publish took 0.433354862s or 23075776.636838567 ops per second