Fix pool.end race conditions

fixes #915

Author: dougwilson

Changes.md

@@ -4,6 +4,10 @@ This file is a manually maintained list of changes for each release. Feel free
 to add your changes here when sending pull requests. Also send corrections if
 you spot any mistakes.
 
+## HEAD
+
+* Fix `pool.end` race conditions #915
+
 ## v2.5.0 (2014-09-07)
 
 * Add code `POOL_ENQUEUELIMIT` to error reaching `queueLimit`

lib/Pool.js

@@ -27,6 +27,7 @@ Pool.prototype.getConnection = function (cb) {
   }
 
   var connection;
+  var pool = this;
 
   if (this._freeConnections.length > 0) {
     connection = this._freeConnections.shift();
@@ -39,17 +40,25 @@ Pool.prototype.getConnection = function (cb) {
 
     this._allConnections.push(connection);
 
+    connection._pool = null;
     return connection.connect({timeout: this.config.acquireTimeout}, function (err) {
-      if (this._closed) {
-        return cb(new Error('Pool is closed.'));
+      if (pool._closed) {
+        connection.destroy();
+        pool._removeConnection(connection);
+        cb(new Error('Pool is closed.'));
+        return;
       }
+
       if (err) {
-        return cb(err);
+        pool._removeConnection(connection);
+        cb(err);
+        return;
       }
 
-      this.emit('connection', connection);
-      return cb(null, connection);
-    }.bind(this));
+      connection._pool = pool;
+      pool.emit('connection', connection);
+      cb(null, connection);
+    });
   }
 
   if (!this.config.waitForConnections) {
@@ -70,13 +79,20 @@ Pool.prototype.acquireConnection = function acquireConnection(connection, cb) {
 
   connection._pool = null;
   connection.ping({timeout: this.config.acquireTimeout}, function(err) {
-    if (!err) {
+    if (!err && !pool._closed) {
       connection._pool = pool;
       cb(null, connection);
       return;
     }
 
     connection.destroy();
+
+    if (pool._closed) {
+      pool._removeConnection(connection);
+      cb(new Error('Pool is closed.'));
+      return;
+    }
+
     pool._connectionQueue.unshift(cb);
     pool._removeConnection(connection);
   });
@@ -136,7 +152,7 @@ Pool.prototype.end = function (cb) {
       return;
     }
 
-    if (err || ++closedConnections >= this._allConnections.length) {
+    if (err || this._allConnections.length === 0) {
       calledBack = true;
       return cb(err);
     }
@@ -148,10 +164,19 @@ Pool.prototype.end = function (cb) {
 
   while (this._allConnections.length) {
     connection = this._allConnections[0];
-    connection._pool = null;
-    connection._realEnd(endCB);
+
+    if (connection._pool === this) {
+      closedConnections++;
+      connection._pool = null;
+      connection._realEnd(endCB);
+    }
+
     this._removeConnection(connection);
   }
+
+  if (closedConnections === 0) {
+    return process.nextTick(endCB);
+  }
 };
 
 Pool.prototype.query = function (sql, values, cb) {

test/unit/pool-cluster/test-connection-order-fatal.js

@@ -0,0 +1,28 @@
+var assert  = require('assert');
+var common  = require('../../common');
+var cluster = common.createPoolCluster();
+var server  = common.createFakeServer();
+
+var poolConfig1 = common.getTestConfig({port: common.bogusPort});
+var poolConfig2 = common.getTestConfig({port: common.fakeServerPort});
+cluster.add('SLAVE1', poolConfig1);
+cluster.add('SLAVE2', poolConfig2);
+
+server.listen(common.fakeServerPort, function(err) {
+  assert.ifError(err);
+
+  var pool = cluster.of('SLAVE*', 'ORDER');
+  var wait = 2;
+
+  pool.getConnection(function (err, connection) {
+    assert.ifError(err);
+    assert.strictEqual(connection._clusterId, 'SLAVE2');
+    if (!--wait) server.destroy();
+  });
+
+  pool.getConnection(function (err, connection) {
+    assert.ifError(err);
+    assert.strictEqual(connection._clusterId, 'SLAVE2');
+    if (!--wait) server.destroy();
+  });
+});

test/unit/pool/test-connection-fatal-end.js

@@ -0,0 +1,16 @@
+var assert = require('assert');
+var common = require('../../common');
+var pool   = common.createPool({
+  connectionLimit : 1,
+  port            : common.bogusPort
+});
+
+pool.getConnection(function (err, conn) {
+  assert.ok(err);
+  assert.ok(err.fatal);
+  assert.equal(err.code, 'ECONNREFUSED');
+
+  pool.end(function (err) {
+    assert.ifError(err);
+  });
+});

test/unit/pool/test-end-ping.js

@@ -0,0 +1,42 @@
+var common = require('../../common');
+var assert = require('assert');
+var pool   = common.createPool({
+  connectionLimit    : 1,
+  port               : common.fakeServerPort,
+  queueLimit         : 5,
+  waitForConnections : true
+});
+
+var conn1Err  = null;
+var conn2Err  = null;
+var poolEnded = false;
+var server    = common.createFakeServer();
+
+server.listen(common.fakeServerPort, function (err) {
+  assert.ifError(err);
+
+  pool.getConnection(function (err, conn) {
+    assert.ifError(err);
+    conn.release();
+
+    pool.getConnection(function (err, conn) {
+      assert.ok(err);
+      assert.equal(err.message, 'Pool is closed.');
+    });
+
+    pool.end(function (err) {
+      assert.ifError(err);
+      server.destroy();
+    });
+  });
+});
+
+server.on('connection', function (conn) {
+  conn.handshake();
+  conn.on('ping', function () {
+    setTimeout(function () {
+      conn._sendPacket(new common.Packets.OkPacket());
+      conn._parser.resetPacketNumber();
+    }, 100);
+  });
+});