Add rate limiting and throttling capabilities. Fixes #1.

Author: coreybutler

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@ngnjs/queue",
-  "version": "1.0.0-alpha.10",
+  "version": "1.0.0-alpha.11",
   "description": "A lightweight NGN queue (taskrunner).",
   "type": "module",
   "main": "src/index.js",
@@ -35,6 +35,12 @@
     "test:deno": "dev test -rt deno tests/*.js",
     "test:browser": "dev test -rt browser tests/*.js",
     "manually": "dev test -rt manual tests/*.js",
+    "test:browser:sanity": "dev test -rt browser tests/*-sanity.js",
+    "test:deno:sanity": "dev test -rt deno tests/*-sanity.js",
+    "test:node:sanity": "dev test -rt node tests/*-sanity.js",
+    "test:browser:throttle": "dev test -rt browser tests/*-throttling.js",
+    "test:deno:throttle": "dev test -rt deno tests/*-throttling.js",
+    "test:node:throttle": "dev test -rt node tests/*-throttling.js",
     "report:syntax": "dev report syntax --pretty",
     "report:size": "dev report size ./.dist/**/*.js ./.dist/**/*.js.map",
     "report:compat": "dev report compatibility ./src/**/*.js",

src/queue.js

@@ -48,9 +48,9 @@ export default class Queue extends NGN.EventEmitter {
   }
 
   /**
-   * Remove an queue item.
+   * Remove a queue item.
    * @param {Item[]} Item
-   * Any number of queue items (or queue item indexes) can be added.
+   * Any number of queue items (or queue item indexes) can be removed.
    */
   remove () {
     for (const item of arguments) {

src/ratelimiter.js

@@ -0,0 +1,122 @@
+import Reference from '@ngnjs/plugin'
+import Throttle from './throttle.js'
+
+const NGN = new Reference()
+NGN.requires('EventEmitter', 'Middleware')
+const { EventEmitter, Middleware } = NGN
+
+const PERFORMANCE = async () => {
+  return globalThis.process !== undefined
+    ? (await import('perf_hooks')).performance
+    : globalThis.performance
+}
+
+export default class RateLimiter extends EventEmitter {
+  #max
+  #duration
+  #queue
+  #maxconcurrent
+
+  #runBatch = async (batch, sequential) => {
+    return new Promise(resolve => {
+      if (!sequential) {
+        if (!isNaN(this.#maxconcurrent) && this.#maxconcurrent > 0) {
+          const limiter = new Throttle(this.#maxconcurrent, batch)
+          limiter.on('end', resolve)
+          limiter.run()
+          return
+        }
+
+        this.afterOnce('task.done', batch.length, resolve)
+
+        // Parallel processing
+        for (const task of batch) {
+          task.once('done', () => this.emit('task.done', task))
+          task.run()
+        }
+      } else {
+        // Sequential processing
+        const process = new Middleware()
+        for (const task of batch) {
+          process.use(next => {
+            this.emit('activetask', task)
+            task.once('done', () => {
+              this.emit('task.done', task)
+              next()
+            })
+            task.run()
+          })
+        }
+
+        process.run(resolve)
+      }
+    })
+  }
+
+  constructor (max, duration, tasks) {
+    super()
+    this.#max = max
+    this.#duration = duration
+    this.#queue = tasks
+  }
+
+  set maxConcurrent (value) {
+    this.#maxconcurrent = value
+  }
+
+  get maxConcurrent () {
+    return this.#maxconcurrent
+  }
+
+  async run (sequential = false) {
+    if (this.#queue.length === 0) {
+      return this.emit('done')
+    }
+
+    const performance = await PERFORMANCE()
+
+    // Batch the queue tasks
+    const batch = []
+    while (this.#queue.length > 0) {
+      batch.push(this.#queue.splice(0, this.#max))
+    }
+
+    // Begin processing
+    const process = new Middleware()
+    while (batch.length > 0) {
+      const tasks = batch.shift()
+
+      process.use(async next => {
+        const start = performance.now()
+
+        this.emit('batch.start', {
+          id: batch.length + 1,
+          start,
+          tasks: tasks.length
+        })
+
+        await this.#runBatch(tasks, sequential)
+
+        let mark = performance.now()
+        const data = { endProcessing: mark }
+
+        while ((mark - start) < this.#duration) {
+          mark = performance.now()
+        }
+
+        this.emit('batch.end', Object.assign(data, {
+          id: batch.length + 1,
+          start,
+          end: mark,
+          duration: Math.ceil(mark - start),
+          processingTime: data.endProcessing - start,
+          tasks: tasks.length
+        }))
+
+        next()
+      })
+    }
+
+    process.run(() => this.emit('done'))
+  }
+}

src/runner.js

@@ -1,6 +1,8 @@
 import Reference from '@ngnjs/plugin'
 import Processor from './queue.js'
 import Item from './item.js'
+import RateLimiter from './ratelimiter.js'
+import Throttle from './throttle.js'
 
 const NGN = new Reference().requires('EventEmitter', 'Middleware', 'WARN', 'INFO', 'ERROR')
 const { WARN, ERROR, EventEmitter, Middleware } = NGN
@@ -12,10 +14,21 @@ export default class Queue extends EventEmitter {
   #status = 'pending'
   #timer
   #timeout = 0
+  #ratelimit = null
+  #maxconcurrent
+  #progress = { completed: 0, total: 0 }
 
   constructor (cfg = {}) {
     super(...arguments)
 
+    if (cfg.rateLimit) {
+      this.rate = cfg.rateLimit
+    }
+
+    if (cfg.maxConcurrent) {
+      this.maxConcurrent = cfg.maxConcurrent
+    }
+
     // Create a task queue
     this.#queue = new Processor({
       name: this.name,
@@ -98,6 +111,116 @@ export default class Queue extends EventEmitter {
     return this.#status
   }
 
+  /**
+   * Rate limiting restricts the maximum number of tasks
+   * which run within the specified duration. For example,
+   * a max of 100 tasks per 60 seconds (60K milliseconds)
+   * will will take 10 minutes to process 1000 tasks
+   * (1000 tasks/100 tasks/minute = 10 minutes).
+   * @param {numeric} max
+   * @param {numeric} duration
+   */
+  rateLimit (max, duration) {
+    if (isNaN(max) || max <= 0) {
+      this.#ratelimit = null
+      return
+    }
+
+    if (isNaN(max) || isNaN(duration)) {
+      throw new Error('rate limiting requires a count and duration (milliseconds)')
+    }
+
+    this.#ratelimit = [max, duration]
+  }
+
+  /**
+   * Remove rate limiting (if set).
+   */
+  removeRateLimit () {
+    this.#ratelimit = null
+  }
+
+  /**
+   * @param {Array} rate
+   * A shortcut attribute for setting the rate limit.
+   * This must be a 2 element array (ex: `[100, 60000]`)
+   * or `null` (to remove rate).
+   */
+  set rate (value) {
+    if (!Array.isArray(value) || value.length !== 2) {
+      throw new Error('rate attribute only accepts an array with 2 numeric elements')
+    }
+
+    this.rateLimit(...value)
+  }
+
+  get rate () {
+    return this.#ratelimit
+  }
+
+  /**
+   * @param {numeric} maxConcurrent
+   * The maximum concurrent number of tasks allowed to
+   * operate at the same time. Set this to `null` or any
+   * value `<= 0` to remove concurrency limits.
+   */
+  set maxConcurrent (value) {
+    if (value === null) {
+      this.#maxconcurrent = null
+      return
+    }
+
+    if (isNaN(value)) {
+      throw new Error('maxConcurrent must be a number or null.')
+    }
+
+    if (value < 0) {
+      this.#maxconcurrent = null
+      return
+    }
+
+    this.#maxconcurrent = value
+  }
+
+  get maxConcurrent () {
+    return this.#maxconcurrent
+  }
+
+  /**
+   * @typedef plan
+   * The expected operation plan.
+   * @param {numeric} tasksRemaining
+   * The number of tasks which still need to be processed
+   * @param {numeric} tasksCompleted
+   * The number of tasks which have been completed
+   * @param {numeric} minimumDuration
+   * The minimum duration (in milliseconds) the queue will take to complete.
+   * This is calculated using the rate limit.
+   */
+
+  /**
+   * @param {object} plan
+   * Returns a plan object.
+   */
+  get plan () {
+    if (this.#progress.total === 0 && this.#queue.items.length > 0) {
+      this.#progress.total = this.#queue.items.length
+    }
+
+    const plan = {
+      tasksRemaining: this.#progress.total - this.#progress.completed,
+      tasksCompleted: this.#progress.completed
+    }
+
+    if (Array.isArray(this.#ratelimit)) {
+      plan.minimumDuration = (this.#progress.total / this.#ratelimit[0]) * this.#ratelimit[1]
+    } else {
+      plan.minimumDuration = null
+    }
+
+    return Object.freeze(plan)
+  }
+
   reset () {
     this.#queue.afterOnce('reset', this.#queue.size, () => {
       this.#cancelled = false
@@ -252,6 +375,8 @@ export default class Queue extends EventEmitter {
       return
     }
 
+    this.#progress = { completed: 0, total: 0 }
+
     // Immediately "complete" when the queue is empty.
     if (this.#queue.size === 0) {
       this._status = 'pending'
@@ -261,6 +386,8 @@ export default class Queue extends EventEmitter {
 
     // Update the status
     this.#processing = true
+    this.#progress.total = this.#queue.items.length
+    this.#progress.completed = 0
     this._status = 'running'
 
     // Add a timer
@@ -269,12 +396,44 @@ export default class Queue extends EventEmitter {
       this.#timer = setTimeout(() => this.abort(true, activeItem), this.#timeout)
     }
 
+    const isConcurrencyLimited = !isNaN(this.#maxconcurrent) && this.#maxconcurrent > 0
+    const isRateLimited = this.#ratelimit !== null
+    const isLimited = isConcurrencyLimited || isRateLimited
+
+    if (isLimited) {
+      if (isRateLimited) {
+        const limiter = new RateLimiter(...this.#ratelimit, this.#queue.items)
+
+        if (isConcurrencyLimited) {
+          limiter.maxConcurrent = this.#maxconcurrent
+        }
+
+        limiter.on('activetask', item => { activeItem = item })
+        limiter.on('task.done', task => { this.#progress.completed += 1 })
+        limiter.relay('batch.*', this, 'limited')
+        limiter.on('done', () => this.emit('end'))
+
+        ;(async () => {
+          await limiter.run(sequential)
+        })()
+
+        return
+      } else if (!sequential) {
+        const limiter = new Throttle(this.#maxconcurrent, this.#queue.items)
+        limiter.on('task.done', task => { this.#progress.completed += 1 })
+        limiter.on('done', () => this.emit('end'))
+        limiter.run()
+        return
+      }
+    }
+
     if (!sequential) {
       this.afterOnce('task.done', this.size, 'end')
 
       // Run in parallel
       // const TOKEN = Symbol('queue runner')
       for (const task of this.#queue.items) {
+        task.once('done', () => { this.#progress.completed += 1 })
         task.run()
       }
     } else {
@@ -283,7 +442,10 @@ export default class Queue extends EventEmitter {
       for (const task of this.#queue.items) {
         process.use(next => {
           activeItem = task
-          task.once('done', next)
+          task.once('done', () => {
+            this.#progress.completed += 1
+            next()
+          })
           task.run()
         })
       }