Handle parallel pool connection requests in Thin mode for Issue #1591

Author: sharadraju

doc/src/release_notes.rst

@@ -23,6 +23,10 @@ Common Changes
 Thin Mode Changes
 +++++++++++++++++
 
+#)  Fixed bug in handling unexpected pool growth which is crossing pool max limit
+    due to improper handling of parallel connection request.
+    `Issue #1591 <https://github.com/oracle/node-oracledb/issues/1591>`__.
+
 #)  Fixed Bug to return proper error when invalid service name is
     configured in listener.
 

lib/thin/pool.js

@@ -36,6 +36,7 @@ const thinUtil = require('./util.js');
 const {getConnectionInfo} = require('./sqlnet/networkSession.js');
 const crypto = require('crypto');
 const EventEmitter = require('events');
+const Timers = require('timers');
 
 class ThinPoolImpl extends PoolImpl {
 
@@ -69,6 +70,9 @@ class ThinPoolImpl extends PoolImpl {
     this._privateKey = params.privateKey;
     this._obfuscatedPrivateKey = [];
     this._schedulerJob = null;
+    this._poolCloseWaiter = null;
+    this._pendingRequests = [];
+
     // password obfuscation
     if (this._password !== undefined) {
       const obj = protocolUtil.setObfuscatedValue(this._password);
@@ -121,8 +125,9 @@ class ThinPoolImpl extends PoolImpl {
       this._generateConnectionClass();
     }
 
-    // create minimum connections
-    await this._growPool(this._poolMin);
+    // create a background task. It will create minimum connections in the pool
+    // and expand the pool as required.
+    this.bgThreadFunc();
   }
 
   //---------------------------------------------------------------------------
@@ -151,15 +156,21 @@ class ThinPoolImpl extends PoolImpl {
   async _getConnAttrs() {
     let accessToken;
     const clonedAttrs = Object.assign({}, this._userConfig);
+    // deobfuscate password
     if (clonedAttrs.password === null) {
       clonedAttrs.password = protocolUtil.getDeobfuscatedValue(this._password,
         this._obfuscatedPassword);
     }
+
+    // deobfuscate wallet password
     if (clonedAttrs.walletPassword === null) {
       clonedAttrs.walletPassword =
         protocolUtil.getDeobfuscatedValue(this._walletPassword,
           this._obfuscatedWalletPassword);
     }
+
+    // deobfuscate token and private key
+    // check for token expiry
     if (clonedAttrs.token === null) {
       clonedAttrs.token =
         protocolUtil.getDeobfuscatedValue(this._token, this._obfuscatedToken);
@@ -230,6 +241,15 @@ class ThinPoolImpl extends PoolImpl {
   //---------------------------------------------------------------------------
   async close() {
 
+    // wait till background task for pool expansion is finished; if it is not
+    // currently running, wake it up!
+    await new Promise((resolve) => {
+      this._poolCloseWaiter = resolve;
+      if (this.bgWaiter) {
+        this.bgWaiter();
+      }
+    });
+
     // clear scheduled job
     if (this._schedulerJob) {
       clearTimeout(this._schedulerJob);
@@ -319,25 +339,6 @@ class ThinPoolImpl extends PoolImpl {
     return this._stmtCacheSize;
   }
 
-  //---------------------------------------------------------------------------
-  // _growPool()
-  //
-  // Grows the pool to include the specified number of connections.
-  //---------------------------------------------------------------------------
-  async _growPool(numConns) {
-    const clonedAttrs = await this._getConnAttrs();
-    while (numConns > 0) {
-      const conn = new ThinConnectionImpl();
-      conn._pool = this;
-      await conn.connect(clonedAttrs);
-      conn._newSession = true;
-      conn._dropSess = false;
-      conn._lastTimeUsed = Date.now();
-      this._freeConnectionList.push(conn);
-      numConns--;
-    }
-  }
-
   //---------------------------------------------------------------------------
   // _setScheduler()
   //
@@ -375,6 +376,94 @@ class ThinPoolImpl extends PoolImpl {
     this._setScheduler();
   }
 
+  //---------------------------------------------------------------------------
+  // _getNumConns()
+  //
+  // get number of connections need to be created
+  //---------------------------------------------------------------------------
+  _getNumConns() {
+    const usedConns = this._freeConnectionList.length + this._usedConnectionList.size;
+    // less connections in the pool than poolMin? restore to poolMin
+    if (usedConns < this._poolMin) {
+      return this._poolMin - usedConns;
+    // connections need to be created? create up to poolIncrement without exceeding poolMax
+    } else if (this._pendingRequests.length > 0) {
+      return Math.min(this._poolIncrement, this._poolMax - usedConns);
+    // no pending requests and we are already at poolMin so nothing to do!
+    } else {
+      return 0;
+    }
+  }
+
+  //---------------------------------------------------------------------------
+  // bgThreadFunc()
+  //
+  // method which runs in a background thread and is used to create connections.
+  // When first started, it creates poolMin connections. After that, it creates
+  // poolIncrement connections up to the value of poolMax when needed.
+  // The thread terminates automatically when the pool is closed.
+  //---------------------------------------------------------------------------
+  async bgThreadFunc() {
+
+    // continue until a close request is received
+    while (!this._poolCloseWaiter) {
+      // get count for connections to be created
+      const numConns = this._getNumConns();
+
+      // connection creation is going on serially and not concurrently
+      for (let i = 0; i < numConns; i++) {
+        try {
+          // get deobfuscated value
+          const config = await this._getConnAttrs();
+          const conn = new ThinConnectionImpl();
+          conn._pool = this;
+          await conn.connect(config);
+          conn._newSession = true;
+          conn._dropSess = false;
+          conn._lastTimeUsed = Date.now();
+          this._freeConnectionList.push(conn);
+        } catch (err) {
+          this._bgErr = err;
+        }
+
+        if (this._poolIncrement > 1 && (this._poolMax - this._usedConnectionList.size
+            - this._freeConnectionList.length) > 1) {
+          this._setScheduler();
+        }
+
+        // resolve pending request
+        if (this._pendingRequests.length > 0) {
+          const payload = this._pendingRequests.shift();
+          payload.resolve();
+        }
+
+        // give an opportunity for other "threads" to do their work.
+        await new Promise((resolve) => Timers.setImmediate(resolve));
+
+        // break loop when pool is closing
+        if (this._poolCloseWaiter) {
+          break;
+        }
+      }
+
+      // when pool is closing, break from while loop
+      if (this._poolCloseWaiter) {
+        break;
+      }
+
+      // if no pending requests, wait for pending requests to appear!
+      if (this._pendingRequests.length == 0 || this._bgErr) {
+        await new Promise((resolve) => {
+          this.bgWaiter = resolve;
+        });
+        this.bgWaiter = null;
+      }
+    }
+
+    // notify the closer that the close can actually take place
+    this._poolCloseWaiter();
+  }
+
   //---------------------------------------------------------------------------
   // acquire()
   //
@@ -422,17 +511,28 @@ class ThinPoolImpl extends PoolImpl {
 
     // no free connections exist at this point; if less than poolMin
     // connections exist, grow the pool to poolMin again; otherwise, increase
-    // the pool by poolIncrement up to poolMax
-    if (this._usedConnectionList.size < this._poolMin) {
-      await this._growPool(this._poolMin - this._usedConnectionList.size);
-    } else {
-      const sizeAvailable = this._poolMax - this._usedConnectionList.size;
-      await this._growPool(Math.min(this._poolIncrement, sizeAvailable));
-      if (this._poolIncrement > 1 && sizeAvailable > 1) {
-        this._setScheduler();
+    // the pool by poolIncrement up to poolMax. We are deferring this
+    // to the background thread function!
+    await new Promise((resolve) => {
+      this._pendingRequests.push({resolve: resolve});
+      if (this.bgWaiter) {
+        // this wakes up the function to do some more work
+        this.bgWaiter();
       }
-    }
+    });
 
+    if (this._bgErr) {
+      const err = this._bgErr;
+      this._bgErr = null;
+
+      // if an error has occurred in the background thread we clear it and then,
+      // if there are more pending requests we request the background thread
+      // function to try again.
+      if (this._pendingRequests.length > 0 && this.bgWaiter) {
+        this.bgWaiter();
+      }
+      throw err;
+    }
     // return a connection from the ones that were just built
     const conn = this._freeConnectionList.pop();
     this._usedConnectionList.add(conn);

test/poolExpansion.js

@@ -35,71 +35,107 @@
 const oracledb = require('oracledb');
 const assert = require('assert');
 const dbConfig = require('./dbconfig.js');
+const testUtil = require('../test/testsUtil.js');
 
 describe('278. Pool expansion', function() {
 
+  function sleep(ms) {
+    return new Promise((resolve) => {
+      setTimeout(resolve, ms);
+    });
+  }
+
   it('278.1 pool expansion when new connection created and within pool max limit', async function() {
-    if (dbConfig.test.drcp) this.skip();
+    if (!oracledb.thin || dbConfig.test.drcp) this.skip();
+    // thick driver is creating total connections exceeding poolMax
 
     const pool = await oracledb.createPool({
       ...dbConfig,
       poolMin           : 0,
-      poolMax           : 10,
+      poolMax           : 2,
       poolIncrement     : 4,
       homogeneous       : true
     });
     const conn = await pool.getConnection();
-    assert.deepStrictEqual(pool.connectionsOpen, 4);
-    assert.deepStrictEqual(pool.connectionsInUse, 1);
+    assert.strictEqual(pool.connectionsInUse, 1);
+
+    await testUtil.checkAndWait(100, 50, () => pool.connectionsOpen === 2);
+    // total connections created won't exceed pool max limit.
+    assert.strictEqual(pool.connectionsOpen, 2);
+    assert.strictEqual(pool.connectionsInUse, 1);
     await conn.close();
     await pool.close(0);
   });
 
-  (oracledb.thin ? it : it.skip)('278.2 pool expansion when new connection created and exceeding pool max limit', async function() {
+  it('278.2 pool expansion not done on creating minimum connection', async function() {
+    if (dbConfig.test.drcp) this.skip();
+
     const pool = await oracledb.createPool({
       ...dbConfig,
-      poolMin           : 0,
-      poolMax           : 4,
-      poolIncrement     : 7,
+      poolMin           : 2,
+      poolMax           : 10,
+      poolIncrement     : 4,
       homogeneous       : true
     });
-    const conn = await pool.getConnection();
-    assert.deepStrictEqual(pool.connectionsOpen, 4);
-    assert.deepStrictEqual(pool.connectionsInUse, 1);
-    await conn.close();
+
+    await testUtil.checkAndWait(100, 50, () => pool.connectionsOpen === 2);
+    assert.strictEqual(pool.connectionsOpen, 2);
+    assert.strictEqual(pool.connectionsInUse, 0);
+    await testUtil.checkAndWait(100, 50, () => pool.connectionsOpen === 2);
+    await sleep(1000);
+    assert.strictEqual(pool.connectionsOpen, 2);
+    assert.strictEqual(pool.connectionsInUse, 0);
     await pool.close(0);
   });
 
-  it('278.3 pool expansion not done on creating minimum connection', async function() {
+  it('278.3 no pool expansion while acquiring connection already present in pool', async function() {
     if (dbConfig.test.drcp) this.skip();
 
     const pool = await oracledb.createPool({
       ...dbConfig,
-      poolMin           : 5,
+      poolMin           : 2,
       poolMax           : 10,
-      poolIncrement     : 2,
+      poolIncrement     : 3,
       homogeneous       : true
     });
-    assert.deepStrictEqual(pool.connectionsOpen, 5);
-    assert.deepStrictEqual(pool.connectionsInUse, 0);
+    const conn = await pool.getConnection();
+    assert.strictEqual(pool.connectionsInUse, 1);
+    await testUtil.checkAndWait(100, 50, () => pool.connectionsOpen === 2);
+    await sleep(1000);
+    assert.strictEqual(pool.connectionsOpen, 2);
+    assert.strictEqual(pool.connectionsInUse, 1);
+    await conn.close();
     await pool.close(0);
   });
 
-  it('278.4 no pool expansion while acquiring connection already present in pool', async function() {
+  it('278.4 pool connection count not crossing pool max limit on parallel execution', async function() {
     if (dbConfig.test.drcp) this.skip();
 
     const pool = await oracledb.createPool({
       ...dbConfig,
-      poolMin           : 3,
+      poolMin           : 2,
       poolMax           : 10,
-      poolIncrement     : 3,
+      poolIncrement     : 2,
       homogeneous       : true
     });
-    const conn = await pool.getConnection();
-    assert.deepStrictEqual(pool.connectionsOpen, 3);
-    assert.deepStrictEqual(pool.connectionsInUse, 1);
-    await conn.close();
+    let conn1, conn2, conn3;
+    const routine1 = async function() {
+      conn1 =  await pool.getConnection();
+    };
+
+    const routine3 = async function() {
+      conn3 =  await pool.getConnection();
+    };
+
+    const routine2 = async function() {
+      conn2 =  await pool.getConnection();
+    };
+    await Promise.all([routine1(), routine2(), routine3()]);
+    assert.strictEqual(pool.connectionsInUse, 3);
+    await testUtil.checkAndWait(100, 50, () => pool.connectionsOpen === 4);
+    await conn1.close();
+    await conn3.close();
+    await conn2.close();
     await pool.close(0);
   });
-
 });