[JAVA-237]: Should only have 1 DBCleanerThread per Mongo instance

Author: agirbal

src/main/com/mongodb/DB.java

@@ -656,6 +656,9 @@ public int getOptions(){
         return _options.get();
     }
 
+    public abstract void cleanCursors( boolean force ) throws MongoException;
+
+
     final Mongo _mongo;
     final String _name;
 

src/main/com/mongodb/DBApiLayer.java

@@ -37,7 +37,6 @@ public class DBApiLayer extends DB {
     /** The maximum number of cursors allowed */
     static final int NUM_CURSORS_BEFORE_KILL = 100;
     static final int NUM_CURSORS_PER_BATCH = 20000;
-    static final int CLEANER_INTERVAL = 1000;
 
     //  --- show
 
@@ -68,8 +67,6 @@ protected DBApiLayer( Mongo mongo, String name , DBConnector connector ){
         _rootPlusDot = _root + ".";
 
         _connector = connector;
-        _cleaner = new DBCleanerThread(CLEANER_INTERVAL);
-        _cleaner.start();
     }
 
     public void requestStart(){
@@ -107,7 +104,7 @@ String _removeRoot( String ns ){
         return ns.substring( _root.length() + 1 );
     }
 
-    void _cleanCursors( boolean force )
+    public void cleanCursors( boolean force )
         throws MongoException {
 
         int sz = _deadCursorIds.size();
@@ -481,41 +478,10 @@ static class DeadCursor {
         final long id;
         final ServerAddress host;
     }
-
-    class DBCleanerThread implements Runnable {
-
-        Thread _thread;
-        int interval;
-
-        DBCleanerThread(int interval) {
-            this.interval = interval;
-        }
-
-        void start() {
-            // start thread as daemon, it's only a cleaner
-            _thread = new Thread(this);
-            _thread.setName(getName() + " - cleaner");
-            _thread.setDaemon(true);
-            _thread.start();
-        }
-
-        public void run() {
-            while (_connector.isOpen()) {
-                try {
-                    Thread.sleep(interval);
-                    _cleanCursors(true);
-                } catch (Throwable t) {
-                    // thread must never die, print full stack
-                    TRACE_LOGGER.log(Level.WARNING, t.getMessage(), t);
-                }
-            }
-        }
-    }
 
     final String _root;
     final String _rootPlusDot;
     final DBConnector _connector;
-    final DBCleanerThread _cleaner;
     final Map<String,MyCollection> _collections = Collections.synchronizedMap( new HashMap<String,MyCollection>() );
 
     ConcurrentLinkedQueue<DeadCursor> _deadCursorIds = new ConcurrentLinkedQueue<DeadCursor>();

src/main/com/mongodb/Mongo.java

@@ -105,6 +105,11 @@ public class Mongo {
      */
     public static final int MINOR_VERSION = 4;
 
+    static int cleanerIntervalMS;
+    static {
+        cleanerIntervalMS = Integer.parseInt(System.getProperty("com.mongodb.cleanerIntervalMS", "1000"));
+    }
+
     /**
      * returns a database object
      * @param addr the database address
@@ -185,6 +190,8 @@ public Mongo( ServerAddress addr , MongoOptions options )
         _options = options;
         _applyMongoOptions();
         _connector = new DBTCPConnector( this , _addr );
+        _cleaner = new DBCleanerThread();
+        _cleaner.start();
     }
 
     /**
@@ -220,6 +227,9 @@ public Mongo( ServerAddress left , ServerAddress right , MongoOptions options )
         _options = options;
         _applyMongoOptions();
         _connector = new DBTCPConnector( this , _addrs );
+
+        _cleaner = new DBCleanerThread();
+        _cleaner.start();
     }
 
     /**
@@ -252,6 +262,9 @@ public Mongo( List<ServerAddress> replicaSetSeeds , MongoOptions options )
         _options = options;
         _applyMongoOptions();
         _connector = new DBTCPConnector( this , _addrs );
+
+        _cleaner = new DBCleanerThread();
+        _cleaner.start();
     }
 
     /**
@@ -288,6 +301,8 @@ public Mongo( MongoURI uri )
             _connector = new DBTCPConnector( this , replicaSetSeeds );
         }
 
+        _cleaner = new DBCleanerThread();
+        _cleaner.start();
     }
 
     /**
@@ -494,6 +509,7 @@ DBTCPConnector getConnector() {
     final ConcurrentMap<String,DB> _dbs = new ConcurrentHashMap<String,DB>();
     private WriteConcern _concern = WriteConcern.NORMAL;
     final Bytes.OptionHolder _netOptions = new Bytes.OptionHolder( null );
+    final DBCleanerThread _cleaner;
 
     org.bson.util.SimplePool<PoolOutputBuffer> _bufferPool = 
         new org.bson.util.SimplePool<PoolOutputBuffer>( 1000 ){
@@ -559,4 +575,24 @@ String _toKey( MongoURI uri ){
 
     }
 
+    class DBCleanerThread extends Thread {
+
+        DBCleanerThread() {
+            setDaemon(true);
+            setName("MongoCleaner" + hashCode());
+        }
+
+        public void run() {
+            while (_connector.isOpen()) {
+                try {
+                    Thread.sleep(cleanerIntervalMS);
+                    for (DB db : _dbs.values()) {
+                        db.cleanCursors(true);
+                    }
+                } catch (Throwable t) {
+                    // thread must never die
+                }
+            }
+        }
+    }
 }