Changeset 2592

Timestamp:

07/05/08 22:16:04 (3 months ago)

Author:

jmoore

Message:

ticket:1018 - Initial version of OmeroSessions reference counting

Implementation is incomplete in the blitz case since there are two
possible "closers" - the user and the glacier session, which must
be taken into account.

Location:

trunk/components

Files:

• blitz/src/ome/services/blitz/impl/ServiceFactoryI.java (modified) (1 diff)
• server/src/ome/services/sessions/SessionContext.java (modified) (1 diff)
• server/src/ome/services/sessions/SessionContextImpl.java (modified) (2 diffs)
• server/src/ome/services/sessions/SessionManager.java (modified) (1 diff)
• server/src/ome/services/sessions/SessionManagerImpl.java (modified) (8 diffs)
• server/test/ome/server/utests/sessions/SessMgrUnitTest.java (modified) (11 diffs)
• tools/OmeroPy/src/omero/__init__.py (modified) (5 diffs)
• tools/OmeroPy/test/integration/isession.py (modified) (1 diff)

trunk/components/blitz/src/ome/services/blitz/impl/ServiceFactoryI.java

@@ -524 +524 @@
     }
 
+    /**
+     * Destruction simply decrements the reference count for a session to allow
+     * reconnecting to it. This means that the Glacier timeout property is
+     * fairly unimportant. If a Glacier connection times out or is otherwise
+     * destroyed, a client can attempt to reconnect
+     */
     public void destroy(Ice.Current current) {
-        if (log.isInfoEnabled()) {
-            log.info(String.format("Destroying %s session", current.id.name));
-        }
-        close(current);
+        sessionManager.detach(this.principal.getName());
     }
 

trunk/components/server/src/ome/services/sessions/SessionContext.java

@@ -24 +24 @@
 
     Session getSession();
+
     List<String> getUserRoles();
 
+    // Reference counting
+
+    /**
+     * Return the current number of references which this session is aware of.
+     */
+    int refCount();
+
+    /**
+     * Increment the current {@link #refCount() reference count} and return the
+     * new value atomically.
+     */
+    int increment();
+
+    /**
+     * Decrement the current {@link #refCount() reference count} and return the
+     * new value atomically.
+     */
+    int decrement();
 }

trunk/components/server/src/ome/services/sessions/SessionContextImpl.java

@@ -11 +11 @@
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import ome.model.meta.Session;
 
 public class SessionContextImpl implements SessionContext {
+
+    private final AtomicInteger refCount = new AtomicInteger(0);
     private final Session session;
     private final List<Long> leaderOfGroups;
@@ -28 +31 @@
                 mGroups));
         this.roles = Collections.unmodifiableList(new ArrayList(roles));
+    }
+
+    public int refCount() {
+        return refCount.get();
+    }
+
+    public int increment() {
+        return refCount.incrementAndGet();
+    }
+
+    public int decrement() {
+        return refCount.decrementAndGet();
     }
 

trunk/components/server/src/ome/services/sessions/SessionManager.java

@@ -51 +51 @@
     Session create(Principal principal);
 
+    /**
+     * 
+     * @param session
+     * @return
+     */
     Session update(Session session);
+
+    /**
+     * Allows decrementing the reference count for a session without calling the
+     * actual {@link #close(String)} logic. This is useful when it is assumed
+     * that another user will re-attach to the same session. A timeout can still
+     * cause the session to be removed.
+     * 
+     * @param uuid
+     * @return
+     */
+    int detach(String uuid);
 
     /**

trunk/components/server/src/ome/services/sessions/SessionManagerImpl.java

@@ -73 +73 @@
 
     // Injected
-    OmeroContext context;
-    Roles roles;
-    SessionCache cache;
-    Executor executor;
-    long defaultTimeToIdle;
-    long defaultTimeToLive;
+    protected OmeroContext context;
+    protected Roles roles;
+    protected SessionCache cache;
+    protected Executor executor;
+    protected long defaultTimeToIdle;
+    protected long defaultTimeToLive;
 
     /**
@@ -84 +84 @@
      * {@link Executor}
      */
-    Principal asroot;
-    SessionContext sc;
+    protected Principal asroot;
+    protected SessionContext sc;
 
     // ~ Injectors
@@ -155 +155 @@
             SessionContext context = cache.getSessionContext(credentials);
             if (context != null) {
+                context.increment();
                 return context.getSession(); // EARLY EXIT!
             }
@@ -161 +162 @@
         }
 
-        boolean ok = executeCheckPassword(_principal, credentials);
+        // Though trusted values, if we receive a null principal, not ok;
+        boolean ok = _principal == null ? false : executeCheckPassword(
+                _principal, credentials);
 
         if (!ok) {
@@ -179 +182 @@
                     .getName());
             if (context != null) {
+                context.increment();
                 return context.getSession(); // EARLY EXIT!
             }
@@ -205 +209 @@
             throw re;
         }
+
+        // All successful, increment and return.
+        ctx.increment();
         return session;
     }
@@ -294 +301 @@
     }
 
+    public int getReferenceCount(String uuid) {
+        SessionContext ctx = cache.getSessionContext(uuid);
+        return ctx.refCount();
+    }
+
+    public int detach(String uuid) {
+        SessionContext ctx = cache.getSessionContext(uuid);
+        return ctx.decrement();
+    }
+
     /*
      */
@@ -302 +319 @@
             ctx = cache.getSessionContext(uuid);
         } catch (SessionException se) {
-            ctx = null;
-        }
-
-        if (ctx == null) {
-            return;
-        }
-
-        Session s = ctx.getSession();
-        s.setClosed(new Timestamp(System.currentTimeMillis()));
-        update(s);
-
-        try {
-            context.publishEvent(new DestroySessionMessage(this, s.getUuid()));
-        } catch (RuntimeException re) {
-            log.warn("Session destruction cancelled by event listener", re);
-            throw re;
-        }
-        // This the publishEvent returns successfully, then we update our cache
-        // since ehcache is not tx-friendly.
-        cache.removeSession(uuid);
+            return; // EARLY EXIT!
+        }
+
+        int refCount = ctx.decrement();
+        if (refCount < 1) {
+
+            Session s = ctx.getSession();
+            s.setClosed(new Timestamp(System.currentTimeMillis()));
+            update(s);
+
+            try {
+                context.publishEvent(new DestroySessionMessage(this, s
+                        .getUuid()));
+            } catch (RuntimeException re) {
+                log.warn("Session destruction cancelled by event listener", re);
+                throw re;
+            }
+            // This the publishEvent returns successfully, then we update our
+            // cache
+            // since ehcache is not tx-friendly.
+            cache.removeSession(uuid);
+        }
     }
 

trunk/components/server/test/ome/server/utests/sessions/SessMgrUnitTest.java

@@ -10 +10 @@
 import java.util.List;
 
-import ome.conditions.ApiUsageException;
+import net.sf.ehcache.CacheManager;
+import ome.api.local.LocalAdmin;
+import ome.api.local.LocalQuery;
+import ome.api.local.LocalUpdate;
+import ome.conditions.AuthenticationException;
 import ome.conditions.SecurityViolation;
 import ome.conditions.SessionException;
@@ -20 +24 @@
 import ome.model.meta.ExperimenterGroup;
 import ome.model.meta.Session;
+import ome.services.sessions.SessionContext;
 import ome.services.sessions.SessionManagerImpl;
+import ome.services.sessions.state.SessionCache;
+import ome.services.util.Executor;
+import ome.system.OmeroContext;
 import ome.system.Principal;
 import ome.system.Roles;
+import ome.testing.MockServiceFactory;
 
 import org.jmock.Mock;
 import org.jmock.MockObjectTestCase;
 import org.jmock.core.Constraint;
+import org.jmock.core.Invocation;
+import org.jmock.core.Stub;
 import org.testng.annotations.BeforeTest;
 import org.testng.annotations.Test;
@@ -36 +47 @@
 public class SessMgrUnitTest extends MockObjectTestCase {
 
-    private static class TestSessionManager extends SessionManagerImpl {
-
-    }
-
-    private Mock adminMock, updateMock, queryMock;
-    private TestSessionManager mgr;
-    private Session session;
-
-    @BeforeTest
-    public void config() {
-        mgr = new TestSessionManager();
-        mgr.setRoles(new Roles());
-
-        session = new Session();
-        session.setUuid("uuid");
-        session.setId(1L);
-    }
-
+    private final class DoWorkStub implements Stub {
+        public Object invoke(Invocation i) throws Throwable {
+            Executor.Work work = (Executor.Work) i.parameterValues.get(1);
+            return work.doWork(null, null, sf);
+        }
+
+        public StringBuffer describeTo(StringBuffer sb) {
+            sb.append("calls doWork on work");
+            return sb;
+        }
+    }
+
+    private final class TestManager extends SessionManagerImpl {
+        Session doDefine() {
+            return define("uuid", "message", System.currentTimeMillis(),
+                    defaultTimeToIdle, defaultTimeToLive, "Test", "rw----");
+        }
+    }
+
+    private final OmeroContext oc = new OmeroContext(
+            "classpath:ome/testing/empty.xml");
+    private final MockServiceFactory sf = new MockServiceFactory();
+    private Mock exMock;
+    private TestManager mgr;
+    private SessionCache cache;
+
+    // State
+    Session session = new Session();
     Principal principal = new Principal("u", "g", "Test");
     String credentials = "password";
@@ -62 +83 @@
     List<String> userRoles = Collections.singletonList("single");
 
+    @BeforeTest
+    public void config() {
+
+        exMock = mock(Executor.class);
+        sf.mockAdmin = mock(LocalAdmin.class);
+        sf.mockUpdate = mock(LocalUpdate.class);
+        sf.mockQuery = mock(LocalQuery.class);
+
+        cache = new SessionCache();
+        cache.setCacheManager(CacheManager.getInstance());
+
+        mgr = new TestManager();
+        mgr.setRoles(new Roles());
+        mgr.setSessionCache(cache);
+        mgr.setExecutor((Executor) exMock.proxy());
+        mgr.setApplicationContext(oc);
+        mgr.setDefaultTimeToIdle(100 * 1000L);
+        mgr.setDefaultTimeToLive(300 * 1000L);
+
+        session = mgr.doDefine();
+        session.setId(1L);
+
+        user.setOmeName(principal.getName());
+    }
+
     void prepareSessionCreation() {
-        adminMock.expects(once()).method("checkPassword").will(
+
+        exMock.expects(atLeastOnce()).method("execute").will(new DoWorkStub());
+
+        sf.mockAdmin.expects(once()).method("checkPassword").will(
                 returnValue(true));
-        adminMock.expects(once()).method("userProxy").will(returnValue(user));
-        adminMock.expects(once()).method("groupProxy").will(returnValue(group));
-        adminMock.expects(once()).method("getMemberOfGroupIds").will(
+        sf.mockAdmin.expects(atLeastOnce()).method("userProxy").will(
+                returnValue(user));
+        sf.mockAdmin.expects(atLeastOnce()).method("groupProxy").will(
+                returnValue(group));
+        sf.mockAdmin.expects(atLeastOnce()).method("getMemberOfGroupIds").will(
                 returnValue(m_ids));
-        adminMock.expects(once()).method("getLeaderOfGroupIds").will(
+        sf.mockAdmin.expects(atLeastOnce()).method("getLeaderOfGroupIds").will(
                 returnValue(l_ids));
-        adminMock.expects(once()).method("getUserRoles").will(
+        sf.mockAdmin.expects(atLeastOnce()).method("getUserRoles").will(
                 returnValue(userRoles));
-        adminMock.expects(once()).method("checkPassword").will(
+        sf.mockAdmin.expects(once()).method("checkPassword").will(
                 returnValue(true));
-        queryMock.expects(once()).method("findAllByQuery")
+        sf.mockQuery.expects(once()).method("findAllByQuery")
                 .will(
                         returnValue(Collections
                                 .singletonList(new ExperimenterGroup())));
-        updateMock.expects(once()).method("saveObject");
+        sf.mockUpdate.expects(atLeastOnce()).method("saveAndReturnObject")
+                .will(returnValue(session));
+
+    }
+
+    void prepareSessionUpdate() {
+        prepareSessionCreation();
+        // exMock.expects(atLeastOnce()).method("execute").will(new
+        // DoWorkStub());
+        //
+        // sf.mockAdmin.expects(once()).method("userProxy")
+        // .will(returnValue(user));
+        // sf.mockAdmin.expects(atLeastOnce()).method("groupProxy").will(
+        // returnValue(group));
+        // sf.mockUpdate.expects(once()).method("saveAndReturnObject").will(
+        // new SetIdStub(1L));
+        //
     }
 
@@ -91 +158 @@
 
         prepareSessionCreation();
-        session = mgr.create(principal, credentials);
+        assert session == mgr.create(principal, credentials);
         assertNotNull(session);
         assertNotNull(session.getUuid());
@@ -141 +208 @@
         testCreateNewSession();
         session.setDefaultEventType("somethingnew");
-        updateMock.expects(once()).method("saveAndReturnObject").will(
+        sf.mockUpdate.expects(once()).method("saveAndReturnObject").will(
                 returnValue(rv));
+
+        prepareSessionUpdate();
         Session test = mgr.update(session);
-        assertTrue(test == rv); // insists that a copy is performed
+        assertFalse(test == rv); // insists that a copy is performed
     }
 
@@ -204 +273 @@
 
         };
-        updateMock.expects(once()).method("saveObject").with(closedSession);
+        sf.mockUpdate.expects(once()).method("saveObject").with(closedSession);
         mgr.close(session.getUuid());
         assertNull(mgr.find(session.getUuid()));
@@ -224 +293 @@
 
         };
-        updateMock.expects(once()).method("saveObject").with(closedSession);
+        sf.mockUpdate.expects(once()).method("saveObject").with(closedSession);
         mgr.close(session.getUuid());
         assertNull(mgr.find(session.getUuid()));
@@ -260 +329 @@
 
     void prepareForCreateSession() {
-        adminMock.expects(once()).method("userProxy").will(returnValue(user));
-        adminMock.expects(once()).method("groupProxy").will(returnValue(group));
-        adminMock.expects(once()).method("getMemberOfGroupIds").will(
+        sf.mockAdmin.expects(once()).method("userProxy")
+                .will(returnValue(user));
+        sf.mockAdmin.expects(once()).method("groupProxy").will(
+                returnValue(group));
+        sf.mockAdmin.expects(atLeastOnce()).method("getMemberOfGroupIds").will(
                 returnValue(m_ids));
-        adminMock.expects(once()).method("getLeaderOfGroupIds").will(
+        sf.mockAdmin.expects(atLeastOnce()).method("getLeaderOfGroupIds").will(
                 returnValue(l_ids));
-        adminMock.expects(once()).method("getUserRoles").will(
+        sf.mockAdmin.expects(once()).method("getUserRoles").will(
                 returnValue(userRoles));
-        adminMock.expects(once()).method("checkPassword").will(
+        sf.mockAdmin.expects(once()).method("checkPassword").will(
                 returnValue(true));
     }
 
-    @Test(expectedExceptions = ApiUsageException.class)
+    @Test(expectedExceptions = AuthenticationException.class)
     public void testCreateSessionFailsAUEOnNullPrincipal() throws Exception {
-        adminMock.expects(once()).method("checkPassword").will(
+        sf.mockAdmin.expects(once()).method("checkPassword").will(
                 returnValue(false));
         mgr.create(null, "password");
     }
 
-    @Test(expectedExceptions = ApiUsageException.class)
+    @Test(expectedExceptions = AuthenticationException.class)
     public void testCreateSessionFailsAUEOnNullOmeName() throws Exception {
-        adminMock.expects(once()).method("checkPassword").will(
+        sf.mockAdmin.expects(once()).method("checkPassword").will(
                 returnValue(false));
         mgr.create(new Principal(null, null, null), "password");
     }
 
-    @Test(expectedExceptions = SecurityViolation.class)
+    @Test(expectedExceptions = AuthenticationException.class)
     public void testCreateSessionFailsSV() throws Exception {
-        adminMock.expects(once()).method("checkPassword").will(
+        sf.mockAdmin.expects(once()).method("checkPassword").will(
                 returnValue(false));
         mgr.create(principal, "password");
@@ -296 +367 @@
     public void testChecksForDefaultGroupsOnCreation() throws Exception {
         prepareForCreateSession();
-        queryMock.expects(once()).method("findAllByQuery").will(
+        sf.mockAdmin.expects(once()).method("getDefaultGroup").will(
+                returnValue(group));
+        sf.mockQuery.expects(once()).method("findAllByQuery").will(
                 returnValue(Collections.EMPTY_LIST));
         mgr.create(new Principal("user", "user", "User"), "user");
@@ -321 +394 @@
         fail("NYI");
     }
+
+    @Test
+    public void testReferenceCounting() throws Exception {
+        testCreateNewSession();
+        String uuid = session.getUuid();
+        SessionContext ctx = cache.getSessionContext(uuid);
+
+        assertEquals(1, ctx.refCount());
+        assertNull(ctx.getSession().getClosed());
+
+        mgr.create(new Principal(uuid));
+        assertEquals(2, ctx.refCount());
+        assertNull(ctx.getSession().getClosed());
+
+        mgr.close(uuid);
+        assertEquals(1, ctx.refCount());
+        assertNull(ctx.getSession().getClosed());
+
+        prepareSessionUpdate();
+        mgr.close(uuid);
+        assertEquals(0, ctx.refCount());
+        assertNotNull(ctx.getSession().getClosed());
+    }
 }

trunk/components/tools/OmeroPy/src/omero/__init__.py

@@ -11 +11 @@
 """
 
-import exceptions
+import exceptions, traceback
 import Ice, Glacier2
 import api
@@ -21 +21 @@
 class client(object):
     """
-    Central blitz entry point
+    Central blitz entry point. Currently useful for a single session, after closing the
+    connection, create another instance.
 
     Typical usage includes:
@@ -37 +38 @@
         self.of.registerObjectFactory(ic)
         self.ic = ic
+        self.close_on_destroy = True
+
+    def closeOnDestroy(self):
+        """
+        Closes the blitz session when destroyConnection() is called on exit.
+        This prevents other clients from attaching to the same blitz session,
+        and is more secure.
+        """
+        self.close_on_destroy = True
+
+    def detachOnDestroy(self):
+        """
+        Prevents the blitz session from being closed when destroyConnection()
+        is called. This will allow other clients to attach to the session for
+        distribtued work. This is less secure, but in many cases valuable.
+        """
         self.close_on_destroy = False
 
-    def closeOnDestroy(self):
-        self.close_on_destroy = True
-
     def __del__(self):
-        if self.close_on_destroy:
-            self.closeSession()
-        if self.ic:
-            try:
-                self.ic.destroy()
-            except (), msg:
-                pysys.stderr.write("Ice exception while destroying communicator:")
-                pysys.stderr.write(msg)
-            self.ic = None
+        try:
+            self.destroyConnection()
+        except exceptions.Exception, e:
+            print "Error in __del__:" + str(e.__class__)
 
     def getCommunicator(self):
@@ -207 +216 @@
             file.close()
 
+    def destroyConnection(self):
+        """
+        Closes the Router connection created by createSession(). Due to a bug in Ice,
+        only one connection is allowed per communicator, so we also destroy the communicator.
+        """
+        # If 'ic' does not exist we don't have anything to do
+        if not hasattr(self, 'ic') or not self.ic:
+            return
+
+        if self.close_on_destroy:
+            self.closeSession()
+
+        prx = self.ic.getDefaultRouter()
+        router = Glacier2.RouterPrx.checkedCast(prx)
+
+        # Now destroy the actual session if possible,
+        # which will always trigger an exception,
+        # regardless of actually being connected or not
+        if router:
+            try:
+                router.destroySession()
+            except exceptions.Exception, e:
+                pass
+                # SNEE happens if we call sf.close() before
+                # calling destroySession(). CLE happens since
+                # we are disconnecting
+
+        try:
+            self.ic.destroy()
+        except (), msg:
+            pysys.stderr.write("Ice exception while destroying communicator:")
+            pysys.stderr.write(msg)
+        self.ic = None
+
     def closeSession(self):
+        """
+        Call close on the session (ServiceFactoryPrx) which first decrements
+        the reference count. If its reference count becomes 0, then the session
+        will be removed from the server.
+        """
         # If 'sf' does not exist we don't have a session at all
         if not hasattr(self, 'sf'):