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,68 @@
+class Deque {
+  constructor() {
+    this._store = Object.create(null)
+    this._head = 0
+    this._tail = 0
+  }
+
+  push(item) {
+    this._store[this._tail++] = item
+  }
+
+  shift() {
+    if (this._head === this._tail) return undefined
+    const item = this._store[this._head]
+    this._store[this._head] = undefined
+    this._head++
+
+    if (this._head === this._tail) {
+      this._head = 0
+      this._tail = 0
+    }
+
+    return item
+  }
+
+  get length() {
+    return this._tail - this._head
+  }
+
+  clear() {
+    this._store = Object.create(null)
+    this._head = 0
+    this._tail = 0
+  }
+
+  remove(item) {
+    if (this._head === this._tail) return
+
+    const store = this._store
+    let write = this._head
+
+    for (let read = this._head; read < this._tail; read++) {
+      const current = store[read]
+      if (current !== item) {
+        store[write++] = current
+      }
+    }
+
+    for (let i = write; i < this._tail; i++) {
+      store[i] = undefined
+    }
+
+    this._tail = write
+
+    if (this._head === this._tail) {
+      this._head = 0
+      this._tail = 0
+    }
+  }
+
+  forEach(fn) {
+    for (let i = this._head; i < this._tail; i++) {
+      fn(this._store[i])
+    }
+  }
+}
+
+module.exports = Deque