Merge pull request neo4j#39 from neo4j/1.0-pool

Add session pooling

Author: pontusmelke

src/v1/driver.js

@@ -18,6 +18,7 @@
  */
 
 import Session from './session';
+import {Pool} from './internal/pool';
 import {connect} from "./internal/connector";
 
 /**
@@ -37,24 +38,74 @@ class Driver {
     this._openSessions = {};
     this._sessionIdGenerator = 0;
     this._token = token || {};
+    this._pool = new Pool(
+      this._createConnection.bind(this),
+      this._destroyConnection.bind(this),
+      this._validateConnection.bind(this)
+    );
   }
 
   /**
-   * Create and return new session
-   * @return {Session} new session.
+   * Create a new connection instance.
+   * @return {Connection} new connector-api session instance, a low level session API.
+   * @access private
    */
-  session() {
+  _createConnection( release ) {
     let sessionId = this._sessionIdGenerator++;
     let conn = connect(this._url);
     conn.initialize(this._userAgent, this._token);
-    let _driver = this;
-    let _session = new Session( conn, () => {
-      // On close of session, remove it from the list of open sessions
-      delete _driver._openSessions[sessionId];
-    });
+    conn._id = sessionId;
+    conn._release = () => release(conn);
+
+    this._openSessions[sessionId] = conn;
+    return conn;
+  }
+
+  /**
+   * Check that a connection is usable
+   * @return {boolean} true if the connection is open
+   * @access private
+   **/
+  _validateConnection( conn ) {
+    return conn.isOpen();
+  }
+
+  /**
+   * Dispose of a live session, closing any associated resources.
+   * @return {Session} new session.
+   * @access private
+   */
+  _destroyConnection( conn ) {
+    delete this._openSessions[conn._id];
+    conn.close();
+  }
+
+  /**
+   * Create and return new session
+   * @return {Session} new session.
+   */
+  session() {
+    let conn = this._pool.acquire();
+    return new Session( conn, (cb) => {
+      // This gets called on Session#close(), and is where we return
+      // the pooled 'connection' instance.
+
+      // We don't pool Session instances, to avoid users using the Session
+      // after they've called close. The `Session` object is just a thin
+      // wrapper around Connection anyway, so it makes little difference.
 
-    this._openSessions[sessionId] = _session;
-    return _session;
+      // Queue up a 'reset', to ensure the next user gets a clean
+      // session to work with. No need to flush, this will get sent
+      // along with whatever the next thing the user wants to do with
+      // this session ends up being, so we save the network round trip.
+      conn.reset();
+
+      // Return connection to the pool
+      conn._release();
+
+      // Call user callback
+      if(cb) { cb(); }
+    });
   }
 
   /**

src/v1/internal/ch-node.js

@@ -60,6 +60,12 @@ class NodeChannel {
         _self.onmessage( new NodeBuffer( buffer ) );
       }
     });
+
+    this._conn.on('error', function(err){
+      if( _self.onerror ) {
+        _self.onerror(err);
+      }
+    });
   }
 
   /**

src/v1/internal/ch-websocket.js

@@ -54,6 +54,12 @@ class WebSocketChannel {
         self.onmessage( b );
       } 
     };
+
+    this._ws.onerror = () => {
+      if( self.onerror ) {
+        self.onerror();
+      }
+    }
   }
 
   /**

src/v1/internal/connector.js

@@ -40,7 +40,8 @@ else {
 let
 // Signature bytes for each message type
 INIT = 0x01,            // 0000 0001 // INIT <user_agent>
-ACK_FAILURE = 0x0F,     // 0000 1111 // ACK_FAILURE
+ACK_FAILURE = 0x0D,     // 0000 1101 // ACK_FAILURE
+RESET = 0x0F,           // 0000 1111 // RESET
 RUN = 0x10,             // 0001 0000 // RUN <statement> <parameters>
 DISCARD_ALL = 0x2F,     // 0010 1111 // DISCARD *
 PULL_ALL = 0x3F,        // 0011 1111 // PULL *
@@ -177,6 +178,9 @@ class Connection {
     this._unpacker = new packstream.Unpacker();
     this._isHandlingFailure = false;
 
+    // Set to true on fatal errors, to get this out of session pool.
+    this._isBroken = false;
+
     // For deserialization, explain to the unpacker how to unpack nodes, rels, paths;
     this._unpacker.structMappers[NODE] = _mappers.node;
     this._unpacker.structMappers[RELATIONSHIP] = _mappers.rel;
@@ -198,14 +202,16 @@ class Connection {
         }
 
       } else {
-        // TODO: Report error
+        this._isBroken = true;
         console.log("FATAL, unknown protocol version:", proposed)
       }
     };
 
+    this._ch.onerror = () => self._isBroken = true;
+
     this._dechunker.onmessage = (buf) => {
       self._handleMessage( self._unpacker.unpack( buf ) );
-    }
+    };
 
     let handshake = alloc( 5 * 4 );
     //magic preamble
@@ -269,6 +275,7 @@ class Connection {
         }
         break;
       default:
+        this._isBroken = true;
         console.log("UNKNOWN MESSAGE: ", msg);
     }
   }
@@ -301,6 +308,13 @@ class Connection {
     this._chunker.messageBoundary();
   }
 
+  /** Queue a RESET-message to be sent to the database */
+  reset( observer ) {
+    this._queueObserver(observer);
+    this._packer.packStruct( RESET );
+    this._chunker.messageBoundary();
+  }
+
   /** Queue a ACK_FAILURE-message to be sent to the database */
   _ackFailure( observer ) {
     this._queueObserver(observer);
@@ -325,6 +339,11 @@ class Connection {
     this._chunker.flush();
   }
 
+  /** Check if this connection is in working condition */
+  isOpen() {
+    return !this._isBroken && this._ch._open;
+  }
+
   /**
    * Call close on the channel.
    * @param {function} cb - Function to call on close.

src/v1/internal/pool.js

@@ -0,0 +1,61 @@
+/**
+ * Copyright (c) 2002-2016 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+class Pool {
+  /**
+   * @param create  an allocation function that creates a new resource. It's given
+   *                a single argument, a function that will return the resource to
+   *                the pool if invoked, which is meant to be called on .dispose
+   *                or .close or whatever mechanism the resource uses to finalize.
+   * @param destroy called with the resource when it is evicted from this pool
+   * @param validate called at various times (like when an instance is acquired and
+   *                 when it is returned). If this returns false, the resource will
+   *                 be evicted
+   * @param maxIdle the max number of resources that are allowed idle in the pool at
+   *                any time. If this threshold is exceeded, resources will be evicted.
+   */
+  constructor(create, destroy=(()=>true), validate=(()=>true), maxIdle=50) {
+    this._create = create;
+    this._destroy = destroy;
+    this._validate = validate;
+    this._maxIdle = maxIdle;
+    this._pool = [];
+    this._release = this._release.bind(this);
+  }
+
+  acquire() {
+    if( this._pool.length > 0 ) {
+      return this._pool.pop();
+    } else {
+      return this._create( this._release );
+    }
+  }
+
+  _release(resource) {
+    if( this._pool.length >= this._maxIdle || !this._validate(resource) ) {
+      this._destroy(resource);
+    } else {
+      this._pool.push(resource);
+    }
+  }
+}
+
+export default {
+    Pool
+}

src/v1/session.js

@@ -47,14 +47,14 @@ class Session {
    * @param {Object} parameters - Map with parameters to use in statement
    * @return {Result} - New Result
    */
-  run(statement, parameters) {
+  run(statement, parameters = {}) {
     if(typeof statement === 'object' && statement.text) {
       parameters = statement.parameters || {};
       statement = statement.text;
     }
     let streamObserver = new StreamObserver();
     if (!this._hasTx) {
-      this._conn.run(statement, parameters || {}, streamObserver);
+      this._conn.run(statement, parameters, streamObserver);
       this._conn.pullAll(streamObserver);
       this._conn.sync();
     } else {
@@ -68,7 +68,7 @@ class Session {
    * Begin a new transaction in this session. A session can have at most one transaction running at a time, if you
    * want to run multiple concurrent transactions, you should use multiple concurrent sessions.
    *
-   * While a transaction is open the session cannot be used to run statements.
+   * While a transaction is open the session cannot be used to run statements outside the transaction.
    *
    * @returns {Transaction} - New Transaction
    */
@@ -82,13 +82,12 @@ class Session {
   }
 
   /**
-   * Close connection
-   * @param {function()} cb - Function to be called on connection close
+   * Close this session
+   * @param {function()} cb - Function to be called after the session has been closed
    * @return
    */
-  close(cb) {
-    this._onClose();
-    this._conn.close(cb);
+  close(cb=(()=>null)) {
+    this._onClose(cb);
   }
 }
 

test/internal/connector.test.js

@@ -21,7 +21,7 @@ var connect = require("../../lib/v1/internal/connector.js").connect;
 
 describe('connector', function() {
 
-  fit('should read/write basic messages', function(done) {
+  it('should read/write basic messages', function(done) {
     // Given
     var conn = connect("bolt://localhost")