[JAVA-263]: NullPointerException due to race condition during concurrent access to DBTCPTransport, comprehensive refactoring

Author: agirbal

src/main/com/mongodb/DBPort.java

@@ -30,7 +30,6 @@
  * represents a Port to the database, which is effectively a single connection to a server
  * Methods implemented at the port level should throw the raw exceptions like IOException,
  * so that the connector above can make appropriate decisions on how to handle.
- * @author antoine
  */
 public class DBPort {
 
@@ -298,7 +297,15 @@ void checkAuth( DB db ) throws IOException {
             throw new MongoException( "couldn't re-auth, username/password change?" );
         _authed.put( db , true );
     }
-    
+
+    /**
+     * Gets the pool that this port belongs to
+     * @return
+     */
+    public DBPortPool getPool() {
+        return _pool;
+    }
+
     final int _hashCode;
     final ServerAddress _sa;
     final InetSocketAddress _addr;

src/main/com/mongodb/DBPortPool.java

@@ -228,6 +228,10 @@ protected DBPort createNew(){
         return new DBPort( _addr , this , _options );
     }
 
+    public ServerAddress getServerAddress() {
+        return _addr;
+    }
+
     final MongoOptions _options;
     final private Semaphore _waitingSem;
     final ServerAddress _addr;

src/main/com/mongodb/DBTCPConnector.java

@@ -41,7 +41,7 @@ public DBTCPConnector( Mongo m , ServerAddress addr )
         if ( addr.isPaired() ){
             _allHosts = new ArrayList<ServerAddress>( addr.explode() );
             _rsStatus = new ReplicaSetStatus( m, _allHosts );
-            _createLogger.info( "switching to replica set mode : " + _allHosts + " -> " + _curMaster  );
+            _createLogger.info( "switching to replica set mode : " + _allHosts + " -> " + getAddress()  );
         }
         else {
             _set( addr );
@@ -64,7 +64,7 @@ public DBTCPConnector( Mongo m , List<ServerAddress> all )
         _allHosts = new ArrayList<ServerAddress>( all ); // make a copy so it can't be modified
         _rsStatus = new ReplicaSetStatus( m, _allHosts );
 
-        _createLogger.info( all  + " -> " + _curMaster );
+        _createLogger.info( all  + " -> " + getAddress() );
     }
 
     private static ServerAddress _checkAddress( ServerAddress addr ){
@@ -245,7 +245,8 @@ public Response call( DB db , DBCollection coll , OutMessage m , ServerAddress h
     }
 
     public ServerAddress getAddress(){
-        return _curMaster;
+        DBPortPool pool = _masterPortPool;
+        return pool != null ? pool.getServerAddress() : null;
     }
 
     /**
@@ -264,17 +265,21 @@ public List<ServerAddress> getAllAddress() {
     public List<ServerAddress> getServerAddressList() {
         if (_rsStatus != null) {
             return _rsStatus.getServerAddressList();
-        } else if (_curMaster != null) {
+        }
+        
+        ServerAddress master = getAddress();
+        if (master != null) {
             // single server
             List<ServerAddress> list = new ArrayList<ServerAddress>();
-            list.add(_curMaster);
+            list.add(master);
             return list;
         }
         return null;
     }
 
     public String getConnectPoint(){
-        return _curMaster.toString();
+        ServerAddress master = getAddress();
+        return master != null ? master.toString() : null;
     }
 
     boolean _error( Throwable t, boolean slaveOk )
@@ -292,89 +297,85 @@ DBPort get( boolean keep , boolean slaveOk , ServerAddress hostNeeded ){
 
             if ( hostNeeded != null ){
                 // asked for a specific host
-                _pool = _portHolder.get( hostNeeded );
-                return _pool.get();
+                return _portHolder.get( hostNeeded ).get();
             }
 
-            if ( _port != null ){
-                // we are within a request, should stick to same port
-                if ( _pool == _masterPortPool ) {
-                    return _port;
+            if ( _requestPort != null ){
+                // we are within a request, and have a port, should stick to it
+                if ( _requestPort.getPool() == _masterPortPool || !keep ) {
+                    // if keep is false, it's a read, so we use port even if master changed
+                    return _requestPort;
                 }
-                // master has changed
-                _pool.done( _port );
-                _port = null;
-                _pool = null;
+
+                // it's write and master has changed
+                // we fall back on new master and try to go on with request
+                // this may not be best behavior if spec of request is to stick with same server
+                _requestPort.getPool().done(_requestPort);
+                _requestPort = null;
             }
 
             if ( slaveOk && _rsStatus != null ){
                 // if slaveOk, try to use a secondary
                 ServerAddress slave = _rsStatus.getASecondary();
                 if ( slave != null ){
-                    _pool = _portHolder.get( slave );
-                    return _pool.get();
+                    return _portHolder.get( slave ).get();
                 }
             }
 
             // use master
-            _pool = _masterPortPool;
-            DBPort p = _pool.get();
+            DBPort p = _masterPortPool.get();
             if ( keep && _inRequest ) {
                 // if within request, remember port to stick to same server
-                _port = p;
+                _requestPort = p;
             }
 
             return p;
         }
 
         void done( DBPort p ){
-            if ( p != _port ){
-                _pool.done( p );
-                _pool = null;
-                _slave = null;
+            // keep request port
+            if ( p != _requestPort ){
+                p.getPool().done(p);
             }
-
         }
 
+        /**
+         * call this method when there is an IOException or other low level error on port.
+         * @param p
+         * @param e
+         */
         void error( DBPort p , Exception e ){
-            _pool.done( p );
+            p.getPool().done( p );
             p.close();
 
-            _port = null;
-            _pool = null;
-
+            _requestPort = null;
             _logger.log( Level.SEVERE , "MyPort.error called" , e );            
         }
 
         void requestEnsureConnection(){
             if ( ! _inRequest )
                 return;
 
-            if ( _port != null )
+            if ( _requestPort != null )
                 return;
 
-            if ( _pool == null )
-                _pool = _masterPortPool;
-
-            _port = _pool.get();
+            _requestPort = _masterPortPool.get();
         }
 
         void requestStart(){
             _inRequest = true;
         }
 
         void requestDone(){
-            if ( _port != null )
-                _pool.done( _port );
-            _port = null;
-            _pool = null;
+            if ( _requestPort != null )
+                _requestPort.getPool().done( _requestPort );
+            _requestPort = null;
             _inRequest = false;
         }
 
-        DBPort _port;
-        DBPortPool _pool;
+        DBPort _requestPort;
+//        DBPortPool _requestPool;
         boolean _inRequest;
-        ServerAddress _slave; // slave used for last read if any
     }
 
     void checkMaster( boolean force , boolean failIfNoMaster )
@@ -409,19 +410,18 @@ void testMaster()
     }
 
     private boolean _set( ServerAddress addr ){
-        if ( _curMaster == addr && _masterPortPool != null )
-            return false;
-        _curMaster = addr;
         _masterPortPool = _portHolder.get( addr );
         return true;
     }
 
     public String debugString(){
         StringBuilder buf = new StringBuilder( "DBTCPConnector: " );
-        if ( _allHosts != null )
+        if ( _rsStatus != null ) {
             buf.append( "replica set : " ).append( _allHosts );
-        else
-            buf.append( _curMaster ).append( " " ).append( _curMaster._addr );
+        } else {
+            ServerAddress master = getAddress();
+            buf.append( master ).append( " " ).append( master != null ? master._addr : null );
+        }
 
         return buf.toString();
     }
@@ -440,7 +440,7 @@ public boolean isOpen(){
     }
 
     private Mongo _mongo;
-    private ServerAddress _curMaster;
+//    private ServerAddress _curMaster;
     private DBPortPool _masterPortPool;
     private DBPortPool.Holder _portHolder;
     private final List<ServerAddress> _allHosts;

src/main/com/mongodb/util/SimplePool.java

@@ -59,7 +59,7 @@ public SimplePool( String name , int maxToKeep , int maxTotal , boolean trackLea
     /** 
      * callback to determine if an object is ok to be added back to the pool or used
      * will be called when something is put back into the queue and when it comes out
-     * @return true iff the object is ok to be added back to pool
+     * @return true if the object is ok to be added back to pool
      */
     public boolean ok( T t ){
         return true;