fix: pool busy (nodejs#1151)

Author: ronag

lib/balanced-pool.js

@@ -9,7 +9,7 @@ const {
   kNeedDrain,
   kAddClient,
   kRemoveClient,
-  kDispatch
+  kGetDispatcher
 } = require('./pool-base')
 const Pool = require('./pool')
 const { kUrl } = require('./core/symbols')
@@ -65,7 +65,7 @@ class BalancedPool extends PoolBase {
       .map((p) => p[kUrl].origin)
   }
 
-  [kDispatch] () {
+  [kGetDispatcher] () {
     // We validate that pools is greater than 0,
     // otherwise we would have to wait until an upstream
     // is added, which might never happen.

lib/pool-base.js

@@ -20,7 +20,7 @@ const kOnConnect = Symbol('onConnect')
 const kOnDisconnect = Symbol('onDisconnect')
 const kOnConnectionError = Symbol('onConnectionError')
 const kQueued = Symbol('queued')
-const kDispatch = Symbol('dispatch')
+const kGetDispatcher = Symbol('get dispatcher')
 const kAddClient = Symbol('add client')
 const kRemoveClient = Symbol('remove client')
 
@@ -190,19 +190,15 @@ class PoolBase extends Dispatcher {
         throw new ClientClosedError()
       }
 
-      const dispatcher = this[kDispatch]()
+      const dispatcher = this[kGetDispatcher]()
 
       if (!dispatcher) {
         this[kNeedDrain] = true
         this[kQueue].push({ opts, handler })
         this[kQueued]++
       } else if (!dispatcher.dispatch(opts, handler)) {
         dispatcher[kNeedDrain] = true
-        this[kNeedDrain] = this[kClients].some(dispatcher => (
-          !dispatcher[kNeedDrain] &&
-          dispatcher.closed !== true &&
-          dispatcher.destroyed !== true
-        ))
+        this[kNeedDrain] = !this[kGetDispatcher]()
       }
     } catch (err) {
       if (typeof handler.onError !== 'function') {
@@ -257,5 +253,5 @@ module.exports = {
   kNeedDrain,
   kAddClient,
   kRemoveClient,
-  kDispatch
+  kGetDispatcher
 }

lib/pool.js

@@ -5,7 +5,7 @@ const {
   kClients,
   kNeedDrain,
   kAddClient,
-  kDispatch
+  kGetDispatcher
 } = require('./pool-base')
 const Client = require('./client')
 const {
@@ -64,7 +64,7 @@ class Pool extends PoolBase {
     this[kFactory] = factory
   }
 
-  [kDispatch] () {
+  [kGetDispatcher] () {
     let dispatcher = this[kClients].find(dispatcher => !dispatcher[kNeedDrain])
 
     if (dispatcher) {

test/pool.js

@@ -321,11 +321,11 @@ test('backpressure algorithm', (t) => {
 
   writeMore = true
 
-  d4.client.emit('drain', new URL('http://notahost'))
+  d4.client.emit('drain', new URL('http://notahost'), [])
 
   pool.dispatch({}, noopHandler) // d5 = c1
 
-  d3.client.emit('drain', new URL('http://notahost'))
+  d3.client.emit('drain', new URL('http://notahost'), [])
 
   pool.dispatch({}, noopHandler) // d6 = c0
 
@@ -337,7 +337,7 @@ test('backpressure algorithm', (t) => {
   t.equal(d5.id, d4.id)
   t.equal(d3.id, d6.id)
 
-  t.equal(total, 2)
+  t.equal(total, 3)
 
   t.end()
 })
@@ -380,7 +380,7 @@ test('busy', (t) => {
         })
       })
       t.equal(client[kPending], n)
-      t.equal(client[kBusy], n > 2)
+      t.equal(client[kBusy], n > 1)
       t.equal(client[kSize], n)
       t.equal(client[kRunning], 0)
     }