diff options
Diffstat (limited to 'jetty-nosql/src/main/java/org/eclipse/jetty/nosql/mongodb/MongoSessionIdManager.java')
| -rw-r--r-- | jetty-nosql/src/main/java/org/eclipse/jetty/nosql/mongodb/MongoSessionIdManager.java | 248 |
1 files changed, 163 insertions, 85 deletions
diff --git a/jetty-nosql/src/main/java/org/eclipse/jetty/nosql/mongodb/MongoSessionIdManager.java b/jetty-nosql/src/main/java/org/eclipse/jetty/nosql/mongodb/MongoSessionIdManager.java index 0038267046..d822ea7926 100644 --- a/jetty-nosql/src/main/java/org/eclipse/jetty/nosql/mongodb/MongoSessionIdManager.java +++ b/jetty-nosql/src/main/java/org/eclipse/jetty/nosql/mongodb/MongoSessionIdManager.java @@ -21,6 +21,7 @@ package org.eclipse.jetty.nosql.mongodb; import java.net.UnknownHostException; import java.util.HashSet; +import java.util.Iterator; import java.util.Random; import java.util.Set; import java.util.concurrent.TimeUnit; @@ -34,6 +35,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; @@ -49,13 +51,13 @@ import com.mongodb.MongoException; /** * Based partially on the JDBCSessionIdManager. - * + * <p> * Theory is that we really only need the session id manager for the local * instance so we have something to scavenge on, namely the list of known ids - * + * <p> * This class has a timer that runs a periodic scavenger thread to query * for all id's known to this node whose precalculated expiry time has passed. - * + * <p> * These found sessions are then run through the invalidateAll(id) method that * is a bit hinky but is supposed to notify all handlers this id is now DOA and * ought to be cleaned up. this ought to result in a save operation on the session @@ -109,10 +111,15 @@ 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; + + private int _scavengeBlockSize; /** @@ -182,6 +189,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,29 +207,60 @@ 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)); + /* + * 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 + * + * break scavenge query into blocks for faster mongo queries + */ + Set<String> block = new HashSet<String>(); + + Iterator<String> itor = _sessionsIds.iterator(); + while (itor.hasNext()) + { + block.add(itor.next()); + if ((_scavengeBlockSize > 0) && (block.size() == _scavengeBlockSize)) + { + //got a block + scavengeBlock (now, block); + //reset for next run + block.clear(); + } + } + + //non evenly divisble block size, or doing it all at once + if (!block.isEmpty()) + scavengeBlock(now, block); + } + + + /* ------------------------------------------------------------ */ + /** + * Check a block of session ids for expiry and thus scavenge. + * + * @param atTime purge at time + * @param ids set of session ids + */ + protected void scavengeBlock (long atTime, Set<String> ids) + { + if (ids == null) + return; + BasicDBObject query = new BasicDBObject(); + query.put(MongoSessionManager.__ID,new BasicDBObject("$in", ids )); + query.put(MongoSessionManager.__EXPIRY, new BasicDBObject("$gt", 0)); + query.put(MongoSessionManager.__EXPIRY, new BasicDBObject("$lt", atTime)); - DBCursor checkSessions = _sessions.find(query, new BasicDBObject(MongoSessionManager.__ID, 1)); + 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)); - } - } + for ( DBObject session : checkSessions ) + { + __log.debug("SessionIdManager:scavenge: expiring session {}", (String)session.get(MongoSessionManager.__ID)); + expireAll((String)session.get(MongoSessionManager.__ID)); + } } /* ------------------------------------------------------------ */ @@ -262,11 +306,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 +329,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); @@ -346,7 +400,7 @@ public class MongoSessionIdManager extends AbstractSessionIdManager /** * The period in seconds between scavenge checks. * - * @param scavengePeriod + * @param scavengePeriod the scavenge period in seconds */ public void setScavengePeriod(long scavengePeriod) { @@ -357,6 +411,42 @@ public class MongoSessionIdManager extends AbstractSessionIdManager } /* ------------------------------------------------------------ */ + /** When scavenging, the max number of session ids in the query. + * + * @param size the scavenge block size + */ + public void setScavengeBlockSize (int size) + { + _scavengeBlockSize = size; + } + + /* ------------------------------------------------------------ */ + public int getScavengeBlockSize () + { + return _scavengeBlockSize; + } + + + /* ------------------------------------------------------------ */ + /** + * The maximum number of items to return from a purge query. If <= 0 there is no limit. Defaults to 0 + * + * @param purgeLimit the purge limit + */ + public void setPurgeLimit(int purgeLimit) + { + _purgeLimit = purgeLimit; + } + + /* ------------------------------------------------------------ */ + public int getPurgeLimit() + { + return _purgeLimit; + } + + + + /* ------------------------------------------------------------ */ public void setPurgeDelay(long purgeDelay) { if ( isRunning() ) @@ -377,6 +467,7 @@ public class MongoSessionIdManager extends AbstractSessionIdManager /** * sets how old a session is to be persisted past the point it is * no longer valid + * @param purgeValidAge the purge valid age */ public void setPurgeInvalidAge(long purgeValidAge) { @@ -395,6 +486,7 @@ public class MongoSessionIdManager extends AbstractSessionIdManager * considered no longer viable and should be removed * * NOTE: set this value to 0 to disable purging of valid sessions + * @param purgeValidAge the purge valid age */ public void setPurgeValidAge(long purgeValidAge) { @@ -520,10 +612,7 @@ public class MongoSessionIdManager extends AbstractSessionIdManager __log.debug("MongoSessionIdManager:addSession {}", session.getId()); - synchronized (_sessionsIds) - { - _sessionsIds.add(session.getId()); - } + _sessionsIds.add(session.getId()); } @@ -536,10 +625,7 @@ public class MongoSessionIdManager extends AbstractSessionIdManager return; } - synchronized (_sessionsIds) - { - _sessionsIds.remove(session.getId()); - } + _sessionsIds.remove(session.getId()); } /* ------------------------------------------------------------ */ @@ -551,56 +637,51 @@ 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); - - //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); + if (sessionHandler != null) { - SessionHandler sessionHandler = ((ContextHandler)contexts[i]).getChildHandlerByClass(SessionHandler.class); - if (sessionHandler != null) - { - SessionManager manager = sessionHandler.getSessionManager(); + SessionManager manager = sessionHandler.getSessionManager(); - if (manager != null && manager instanceof MongoSessionManager) - { - ((MongoSessionManager)manager).invalidateSession(sessionId); - } + if (manager != null && manager instanceof MongoSessionManager) + { + ((MongoSessionManager)manager).invalidateSession(sessionId); } } - } - } + } + } + /* ------------------------------------------------------------ */ /** * Expire this session for all contexts that are sharing the session * id. - * @param sessionId + * @param sessionId the session id */ public void expireAll (String sessionId) { - synchronized (_sessionsIds) - { - _sessionsIds.remove(sessionId); + _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++) + //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); + if (sessionHandler != null) { - SessionHandler sessionHandler = ((ContextHandler)contexts[i]).getChildHandlerByClass(SessionHandler.class); - if (sessionHandler != null) - { - SessionManager manager = sessionHandler.getSessionManager(); + SessionManager manager = sessionHandler.getSessionManager(); - if (manager != null && manager instanceof MongoSessionManager) - { - ((MongoSessionManager)manager).expire(sessionId); - } + if (manager != null && manager instanceof MongoSessionManager) + { + ((MongoSessionManager)manager).expire(sessionId); } } } @@ -613,24 +694,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 + _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++) + //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); + if (sessionHandler != null) { - SessionHandler sessionHandler = ((ContextHandler)contexts[i]).getChildHandlerByClass(SessionHandler.class); - if (sessionHandler != null) - { - SessionManager manager = sessionHandler.getSessionManager(); + SessionManager manager = sessionHandler.getSessionManager(); - if (manager != null && manager instanceof MongoSessionManager) - { - ((MongoSessionManager)manager).renewSessionId(oldClusterId, oldNodeId, newClusterId, getNodeId(newClusterId, request)); - } + if (manager != null && manager instanceof MongoSessionManager) + { + ((MongoSessionManager)manager).renewSessionId(oldClusterId, oldNodeId, newClusterId, getNodeId(newClusterId, request)); } } } |