464839 Add limit to MongoSessionIdManager purge queries

jetty-nosql/src/main/java/org/eclipse/jetty/nosql/mongodb/MongoSessionIdManager.java

@@ -34,6 +34,7 @@ import org.eclipse.jetty.server.SessionManager;
 import org.eclipse.jetty.server.handler.ContextHandler;
 import org.eclipse.jetty.server.session.AbstractSessionIdManager;
 import org.eclipse.jetty.server.session.SessionHandler;
+import org.eclipse.jetty.util.ConcurrentHashSet;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
 import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler;
@@ -109,10 +110,13 @@ public class MongoSessionIdManager extends AbstractSessionIdManager
     
     /**
      * the collection of session ids known to this manager
-     * 
-     * TODO consider if this ought to be concurrent or not
      */
-    protected final Set<String> _sessionsIds = new HashSet<String>();
+    protected final Set<String> _sessionsIds = new ConcurrentHashSet<>();
+
+    /**
+     * The maximum number of items to return from a purge query.
+     */
+    private int _purgeLimit = 0;
     
     
     /**
@@ -182,6 +186,12 @@ public class MongoSessionIdManager extends AbstractSessionIdManager
                 BasicDBObjectBuilder.start().add("id",1).add("version",1).get(),
                 BasicDBObjectBuilder.start().add("unique",true).add("sparse",false).get());
 
+        // index our accessed and valid fields so that purges are faster, note that the "valid" field is first
+        // so that we can take advantage of index prefixes
+        // http://docs.mongodb.org/manual/core/index-compound/#compound-index-prefix
+        _sessions.ensureIndex(
+                BasicDBObjectBuilder.start().add(MongoSessionManager.__VALID, 1).add(MongoSessionManager.__ACCESSED, 1).get(),
+                BasicDBObjectBuilder.start().add("sparse", false).add("background", true).get());
     }
  
     /* ------------------------------------------------------------ */
@@ -194,28 +204,25 @@ public class MongoSessionIdManager extends AbstractSessionIdManager
     {
         long now = System.currentTimeMillis();
         __log.debug("SessionIdManager:scavenge:at {}", now);        
-        synchronized (_sessionsIds)
+        /*
+         * run a query returning results that:
+         *  - are in the known list of sessionIds
+         *  - the expiry time has passed
+         *  
+         *  we limit the query to return just the __ID so we are not sucking back full sessions
+         */
+        BasicDBObject query = new BasicDBObject();     
+        query.put(MongoSessionManager.__ID,new BasicDBObject("$in", _sessionsIds ));
+        query.put(MongoSessionManager.__EXPIRY, new BasicDBObject("$gt", 0));
+        query.put(MongoSessionManager.__EXPIRY, new BasicDBObject("$lt", now));
+        
+            
+        DBCursor checkSessions = _sessions.find(query, new BasicDBObject(MongoSessionManager.__ID, 1));
+                        
+        for ( DBObject session : checkSessions )
         {             
-            /*
+            __log.debug("SessionIdManager:scavenge: expiring session {}", (String)session.get(MongoSessionManager.__ID));
-             * run a query returning results that:
+            expireAll((String)session.get(MongoSessionManager.__ID));
-             *  - are in the known list of sessionIds
-             *  - the expiry time has passed
-             *  
-             *  we limit the query to return just the __ID so we are not sucking back full sessions
-             */
-            BasicDBObject query = new BasicDBObject();     
-            query.put(MongoSessionManager.__ID,new BasicDBObject("$in", _sessionsIds ));
-            query.put(MongoSessionManager.__EXPIRY, new BasicDBObject("$gt", 0));
-            query.put(MongoSessionManager.__EXPIRY, new BasicDBObject("$lt", now));
-        
-            
-            DBCursor checkSessions = _sessions.find(query, new BasicDBObject(MongoSessionManager.__ID, 1));
-                        
-            for ( DBObject session : checkSessions )
-            {             
-                __log.debug("SessionIdManager:scavenge: expiring session {}", (String)session.get(MongoSessionManager.__ID));
-                expireAll((String)session.get(MongoSessionManager.__ID));
-            }
         }      
     }
     
@@ -262,11 +269,16 @@ public class MongoSessionIdManager extends AbstractSessionIdManager
         __log.debug("PURGING");
         BasicDBObject invalidQuery = new BasicDBObject();
 
-        invalidQuery.put(MongoSessionManager.__ACCESSED, new BasicDBObject("$lt",System.currentTimeMillis() - _purgeInvalidAge));
         invalidQuery.put(MongoSessionManager.__VALID, false);
+        invalidQuery.put(MongoSessionManager.__ACCESSED, new BasicDBObject("$lt",System.currentTimeMillis() - _purgeInvalidAge));
         
         DBCursor oldSessions = _sessions.find(invalidQuery, new BasicDBObject(MongoSessionManager.__ID, 1));
 
+        if (_purgeLimit > 0)
+        {
+            oldSessions.limit(_purgeLimit);
+        }
+
         for (DBObject session : oldSessions)
         {
             String id = (String)session.get("id");
@@ -280,11 +292,16 @@ public class MongoSessionIdManager extends AbstractSessionIdManager
         {
             BasicDBObject validQuery = new BasicDBObject();
 
-            validQuery.put(MongoSessionManager.__ACCESSED,new BasicDBObject("$lt",System.currentTimeMillis() - _purgeValidAge));
             validQuery.put(MongoSessionManager.__VALID, true);
+            validQuery.put(MongoSessionManager.__ACCESSED,new BasicDBObject("$lt",System.currentTimeMillis() - _purgeValidAge));
 
             oldSessions = _sessions.find(validQuery,new BasicDBObject(MongoSessionManager.__ID,1));
 
+            if (_purgeLimit > 0)
+            {
+                oldSessions.limit(_purgeLimit);
+            }
+
             for (DBObject session : oldSessions)
             {
                 String id = (String)session.get(MongoSessionManager.__ID);
@@ -356,6 +373,24 @@ public class MongoSessionIdManager extends AbstractSessionIdManager
             _scavengePeriod = TimeUnit.SECONDS.toMillis(scavengePeriod);
     }
 
+
+    /* ------------------------------------------------------------ */
+    /**
+     * The maximum number of items to return from a purge query. If <= 0 there is no limit. Defaults to 0
+     */
+    public void setPurgeLimit(int purgeLimit)
+    {
+        _purgeLimit = purgeLimit;
+    }
+
+    /* ------------------------------------------------------------ */
+    public int getPurgeLimit()
+    {
+        return _purgeLimit;
+    }
+
+
+
     /* ------------------------------------------------------------ */
     public void setPurgeDelay(long purgeDelay)
     {
@@ -522,10 +557,7 @@ public class MongoSessionIdManager extends AbstractSessionIdManager
         
         __log.debug("MongoSessionIdManager:addSession {}", session.getId());
         
-        synchronized (_sessionsIds)
+        _sessionsIds.add(session.getId());
-        {
-            _sessionsIds.add(session.getId());
-        }
         
     }
 
@@ -538,10 +570,7 @@ public class MongoSessionIdManager extends AbstractSessionIdManager
             return;
         }
         
-        synchronized (_sessionsIds)
+        _sessionsIds.remove(session.getId());
-        {
-            _sessionsIds.remove(session.getId());
-        }
     }
 
     /* ------------------------------------------------------------ */
@@ -553,29 +582,27 @@ public class MongoSessionIdManager extends AbstractSessionIdManager
     @Override
     public void invalidateAll(String sessionId)
     {
-        synchronized (_sessionsIds)
+        _sessionsIds.remove(sessionId);
+            
+        //tell all contexts that may have a session object with this id to
+        //get rid of them
+        Handler[] contexts = _server.getChildHandlersByClass(ContextHandler.class);
+        for (int i=0; contexts!=null && i<contexts.length; i++)
         {
-            _sessionsIds.remove(sessionId);
+            SessionHandler sessionHandler = ((ContextHandler)contexts[i]).getChildHandlerByClass(SessionHandler.class);
-                
+            if (sessionHandler != null) 
-            //tell all contexts that may have a session object with this id to
-            //get rid of them
-            Handler[] contexts = _server.getChildHandlersByClass(ContextHandler.class);
-            for (int i=0; contexts!=null && i<contexts.length; i++)
             {
-                SessionHandler sessionHandler = ((ContextHandler)contexts[i]).getChildHandlerByClass(SessionHandler.class);
+                SessionManager manager = sessionHandler.getSessionManager();
-                if (sessionHandler != null) 
-                {
-                    SessionManager manager = sessionHandler.getSessionManager();
 
-                    if (manager != null && manager instanceof MongoSessionManager)
+                if (manager != null && manager instanceof MongoSessionManager)
-                    {
+                {
-                        ((MongoSessionManager)manager).invalidateSession(sessionId);
+                    ((MongoSessionManager)manager).invalidateSession(sessionId);
-                    }
                 }
             }
         }
     }      
  
+
     /* ------------------------------------------------------------ */
     /**
      * Expire this session for all contexts that are sharing the session 
@@ -584,25 +611,22 @@ public class MongoSessionIdManager extends AbstractSessionIdManager
      */
     public void expireAll (String sessionId)
     {
-        synchronized (_sessionsIds)
+        _sessionsIds.remove(sessionId);
+            
+            
+        //tell all contexts that may have a session object with this id to
+        //get rid of them
+        Handler[] contexts = _server.getChildHandlersByClass(ContextHandler.class);
+        for (int i=0; contexts!=null && i<contexts.length; i++)
         {
-            _sessionsIds.remove(sessionId);
+            SessionHandler sessionHandler = ((ContextHandler)contexts[i]).getChildHandlerByClass(SessionHandler.class);
-            
+            if (sessionHandler != null) 
-            
-            //tell all contexts that may have a session object with this id to
-            //get rid of them
-            Handler[] contexts = _server.getChildHandlersByClass(ContextHandler.class);
-            for (int i=0; contexts!=null && i<contexts.length; i++)
             {
-                SessionHandler sessionHandler = ((ContextHandler)contexts[i]).getChildHandlerByClass(SessionHandler.class);
+                SessionManager manager = sessionHandler.getSessionManager();
-                if (sessionHandler != null) 
-                {
-                    SessionManager manager = sessionHandler.getSessionManager();
 
-                    if (manager != null && manager instanceof MongoSessionManager)
+                if (manager != null && manager instanceof MongoSessionManager)
-                    {
+                {
-                        ((MongoSessionManager)manager).expire(sessionId);
+                    ((MongoSessionManager)manager).expire(sessionId);
-                    }
                 }
             }
         }      
@@ -615,24 +639,21 @@ public class MongoSessionIdManager extends AbstractSessionIdManager
         //generate a new id
         String newClusterId = newSessionId(request.hashCode());
 
-        synchronized (_sessionsIds)
+        _sessionsIds.remove(oldClusterId);//remove the old one from the list
+        _sessionsIds.add(newClusterId); //add in the new session id to the list
+
+        //tell all contexts to update the id 
+        Handler[] contexts = _server.getChildHandlersByClass(ContextHandler.class);
+        for (int i=0; contexts!=null && i<contexts.length; i++)
         {
-            _sessionsIds.remove(oldClusterId);//remove the old one from the list
+            SessionHandler sessionHandler = ((ContextHandler)contexts[i]).getChildHandlerByClass(SessionHandler.class);
-            _sessionsIds.add(newClusterId); //add in the new session id to the list
+            if (sessionHandler != null) 
-
-            //tell all contexts to update the id 
-            Handler[] contexts = _server.getChildHandlersByClass(ContextHandler.class);
-            for (int i=0; contexts!=null && i<contexts.length; i++)
             {
-                SessionHandler sessionHandler = ((ContextHandler)contexts[i]).getChildHandlerByClass(SessionHandler.class);
+                SessionManager manager = sessionHandler.getSessionManager();
-                if (sessionHandler != null) 
-                {
-                    SessionManager manager = sessionHandler.getSessionManager();
 
-                    if (manager != null && manager instanceof MongoSessionManager)
+                if (manager != null && manager instanceof MongoSessionManager)
-                    {
+                {
-                        ((MongoSessionManager)manager).renewSessionId(oldClusterId, oldNodeId, newClusterId, getNodeId(newClusterId, request));
+                    ((MongoSessionManager)manager).renewSessionId(oldClusterId, oldNodeId, newClusterId, getNodeId(newClusterId, request));
-                    }
                 }
             }
         }

tests/test-sessions/test-mongodb-sessions/src/test/java/org/eclipse/jetty/nosql/mongodb/PurgeInvalidSessionTest.java

@@ -126,6 +126,81 @@ public class PurgeInvalidSessionTest
     }
 
 
+    @Test
+    public void testPurgeInvalidSessionsWithLimit() throws Exception
+    {
+        String contextPath = "";
+        String servletMapping = "/server";
+        long purgeInvalidAge = 1000; //1 sec
+        int purgeLimit = 5; // only purge 5 sessions for each purge run
+
+        //ensure scavenging is turned off so the purger gets a chance to find the session
+        MongoTestServer server = createServer(0, 1, 0);
+        ServletContextHandler context = server.addContext(contextPath);
+        context.addServlet(TestServlet.class, servletMapping);
+
+        // disable purging so we can call it manually below
+        MongoSessionManager sessionManager = (MongoSessionManager)context.getSessionHandler().getSessionManager();
+        MongoSessionIdManager idManager = (MongoSessionIdManager)server.getServer().getSessionIdManager();
+        idManager.setPurge(false);
+        idManager.setPurgeLimit(purgeLimit);
+        idManager.setPurgeInvalidAge(purgeInvalidAge);
+        // don't purge valid sessions
+        idManager.setPurgeValidAge(0);
+
+
+        server.start();
+        int port=server.getPort();
+        try
+        {
+            // cleanup any previous sessions that are invalid so that we are starting fresh
+            idManager.purgeFully();
+            long sessionCountAtTestStart = sessionManager.getSessionStoreCount();
+
+            HttpClient client = new HttpClient();
+            client.start();
+            try
+            {
+                // create double the purge limit of sessions, and make them all invalid
+                for (int i = 0; i < purgeLimit * 2; i++)
+                {
+                    ContentResponse response = client.GET("http://localhost:" + port + contextPath + servletMapping + "?action=create");
+                    assertEquals(HttpServletResponse.SC_OK, response.getStatus());
+                    String sessionCookie = response.getHeaders().get("Set-Cookie");
+                    assertTrue(sessionCookie != null);
+                    // Mangle the cookie, replacing Path with $Path, etc.
+                    sessionCookie = sessionCookie.replaceFirst("(\\W)(P|p)ath=", "$1\\$Path=");
+
+
+                    Request request = client.newRequest("http://localhost:" + port + contextPath + servletMapping + "?action=invalidate");
+                    request.header("Cookie", sessionCookie);
+                    response = request.send();
+                    assertEquals(HttpServletResponse.SC_OK, response.getStatus());
+                }
+
+                // sleep for our invalid age period so that the purge below does something
+                Thread.sleep(purgeInvalidAge * 2);
+
+                // validate that we have the right number of sessions before we purge
+                assertEquals("Expected to find right number of sessions before purge", sessionCountAtTestStart + (purgeLimit * 2), sessionManager.getSessionStoreCount());
+
+                // run our purge we should still have items in the DB
+                idManager.purge();
+                assertEquals("Expected to find sessions remaining in db after purge run with limit set",
+                        sessionCountAtTestStart + purgeLimit, sessionManager.getSessionStoreCount());
+            }
+            finally
+            {
+                client.stop();
+            }
+        }
+        finally
+        {
+            server.stop();
+        }
+    }
+
+
     public static class TestServlet extends HttpServlet
     {
         DBCollection _sessions;
@@ -163,6 +238,8 @@ public class PurgeInvalidSessionTest
                 dbSession = _sessions.findOne(new BasicDBObject("id", id));       
                 assertNotNull(dbSession);
                 assertTrue(dbSession.containsField(MongoSessionManager.__INVALIDATED));
+                assertTrue(dbSession.containsField(MongoSessionManager.__VALID));
+                assertTrue(dbSession.get(MongoSessionManager.__VALID).equals(false));
             }
             else if ("test".equals(action))
             {

tests/test-sessions/test-mongodb-sessions/src/test/java/org/eclipse/jetty/nosql/mongodb/PurgeValidSessionTest.java

@@ -123,6 +123,75 @@ public class PurgeValidSessionTest
     }
 
 
+    @Test
+    public void testPurgeValidSessionWithPurgeLimitSet() throws Exception
+    {
+        String contextPath = "";
+        String servletMapping = "/server";
+        long purgeDelay = 1000; //1 sec
+        long purgeValidAge = 1000; // 1 sec
+        int purgeLimit = 5; // only purge 5 items in each purge run
+
+        //ensure scavenging is turned off so the purger gets a chance to find the session
+        MongoTestServer server = createServer(0, 1, 0);
+        ServletContextHandler context = server.addContext(contextPath);
+        context.addServlet(TestServlet.class, servletMapping);
+
+        MongoSessionManager sessionManager = (MongoSessionManager)context.getSessionHandler().getSessionManager();
+        MongoSessionIdManager idManager = (MongoSessionIdManager)server.getServer().getSessionIdManager();
+        // disable purging we will run it manually below
+        idManager.setPurge(false);
+        idManager.setPurgeLimit(purgeLimit);
+        idManager.setPurgeDelay(purgeDelay);
+        idManager.setPurgeValidAge(purgeValidAge); //purge valid sessions older than
+
+        server.start();
+        int port=server.getPort();
+
+        try
+        {
+            // start with no sessions
+            idManager.purgeFully();
+
+            HttpClient client = new HttpClient();
+            client.start();
+            try
+            {
+                // create double our purgeLmit number of sessions
+                for (int i = 0; i < purgeLimit * 2; i++)
+                {
+                    ContentResponse response = client.GET("http://localhost:" + port + contextPath + servletMapping + "?action=create");
+                    assertEquals(HttpServletResponse.SC_OK, response.getStatus());
+                    String sessionCookie = response.getHeaders().get("Set-Cookie");
+                    assertTrue(sessionCookie != null);
+                    // don't remember our cookies from call to call
+                    client.getCookieStore().removeAll();
+                }
+
+                // delay long enough for purgeValidAge to apply
+                Thread.sleep(2* purgeValidAge);
+
+                // validate that we have the right number of sessions before we purge
+                assertEquals("Expected to find right number of sessions before purge", purgeLimit * 2, sessionManager.getSessionStoreCount());
+
+                // run our purge
+                idManager.purge();
+
+                assertEquals("Expected to find sessions remaining in db after purge run with limit set",
+                        purgeLimit, sessionManager.getSessionStoreCount());
+            }
+            finally
+            {
+                client.stop();
+            }
+        }
+        finally
+        {
+            server.stop();
+        }
+
+    }
+
     public static class TestServlet extends HttpServlet
     {
         DBCollection _sessions;