perf(client): improve query queue performance by introducing Deque

packages/pg/lib/client.js

@@ -11,6 +11,7 @@ const Query = require('./query')
 const defaults = require('./defaults')
 const Connection = require('./connection')
 const crypto = require('./crypto/utils')
+const Deque = require('./deque')
 
 const activeQueryDeprecationNotice = nodeUtils.deprecate(
   () => {},
@@ -79,7 +80,7 @@ class Client extends EventEmitter {
         keepAliveInitialDelayMillis: c.keepAliveInitialDelayMillis || 0,
         encoding: this.connectionParameters.client_encoding || 'utf8',
       })
-    this._queryQueue = []
+    this._queryQueue = new Deque()
     this.binary = c.binary || defaults.binary
     this.processID = null
     this.secretKey = null
@@ -124,7 +125,7 @@ class Client extends EventEmitter {
     }
 
     this._queryQueue.forEach(enqueueError)
-    this._queryQueue.length = 0
+    this._queryQueue.clear()
   }
 
   _connect(callback) {
@@ -608,10 +609,7 @@ class Client extends EventEmitter {
         query.callback = () => {}
 
         // Remove from queue
-        const index = this._queryQueue.indexOf(query)
-        if (index > -1) {
-          this._queryQueue.splice(index, 1)
-        }
+        this._queryQueue.remove(query)
 
         this._pulseQueryQueue()
       }, readTimeout)

packages/pg/lib/deque.js

@@ -0,0 +1,83 @@
+class Deque {
+  constructor() {
+    this._head = null
+    this._tail = null
+    this._size = 0
+    this._index = new WeakMap()
+  }
+
+  push(item) {
+    const node = { value: item, prev: this._tail, next: null }
+
+    if (this._tail) {
+      node.prev = this._tail
+      this._tail.next = node
+      this._tail = node
+    } else {
+      this._head = this._tail = node
+    }
+
+    this._index.set(item, node)
+    this._size++
+  }
+
+  shift() {
+    if (!this._head) return undefined
+
+    const node = this._head
+    const value = node.value
+
+    this._head = node.next
+    if (this._head) {
+      this._head.prev = null
+    } else {
+      this._tail = null
+    }
+
+    this._index.delete(value)
+    this._size--
+
+    node.prev = node.next = null
+
+    return value
+  }
+
+  get length() {
+    return this._size
+  }
+
+  clear() {
+    this._head = null
+    this._tail = null
+    this._size = 0
+    this._index = new WeakMap()
+  }
+
+  remove(item) {
+    const node = this._index.get(item)
+    if (!node) return false
+
+    if (node.prev) node.prev.next = node.next
+    else this._head = node.next
+
+    if (node.next) node.next.prev = node.prev
+    else this._tail = node.prev
+
+    this._index.delete(item)
+    this._size--
+
+    node.prev = node.next = null
+
+    return true
+  }
+
+  forEach(fn) {
+    let curr = this._head
+    while (curr) {
+      fn(curr.value)
+      curr = curr.next
+    }
+  }
+}
+
+module.exports = Deque