Changeset 2593

Timestamp:

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

Author:

jmoore

Message:

ticket:1018 - Second attempt at improving OmeroSessions in blitz

This tries to keep up with an individual client via a uuid set
on the client side. This may also be sub-optimal. A better idea
may be using individual ServiceFactory instances which talk to
the same session (in which case only the SessionManager? need be
distributed)

Location:

trunk/components

Files:

• blitz/resources/omero/API.ice (modified) (1 diff)
• blitz/resources/omero/Constants.ice (modified) (1 diff)
• blitz/src/ome/services/blitz/impl/ServiceFactoryI.java (modified) (4 diffs)
• tools/OmeroPy/src/omero/__init__.py (modified) (5 diffs)

trunk/components/blitz/resources/omero/API.ice

@@ -596 +596 @@
 
         /*
-         * Closes the service factory and all related services.
-         */
-        void close();
+         * Deprecated misnomer.
+         */
+        ["deprecated:close() is deprecated. use closeOnDestroy() instead."] void close();
+
+        /*
+         * Marks the session for closure rather than detachment, which will
+         * be triggered by the destruction of the Glacier2 connection via
+         * router.destroySession()
+         *
+         * Closing the session rather the detaching is more secure, since all
+         * resources are removed from the server and can safely be set once
+         * it is clear that a client is finished with those resources.
+         */
+        void closeOnDestroy();
+
+        /*
+         * Marks the session for detachment rather than closure, which will
+         * be triggered by the destruction of the Glacier2 connection via
+         * router.destroySession()
+         *
+         * This is the default and allows a lost session to be reconnected,
+         * at a slight security cost since the session will persist longer
+         * and can be used by others if the UUID is intercepted.
+         */
+        void detachOnDestroy();
 
         // Session management
-        
+
         /*
          * Returns a list of string ids for currently active services. This will

trunk/components/blitz/resources/omero/Constants.ice

@@ -12 +12 @@
 module omero {
   module constants {
+
+    /*
+     * Key in the ImplicitContext which must be filled
+     * by all omero.client implementations.
+     */
+    const string CLIENTUUID = "omero.client.uuid";
 
     /*

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

@@ -145 +145 @@
     final Map<String, Ice.Object> interactiveSlots;
 
+    /**
+     * The CLIENT_UUIDs for which a close should be invoked on this session.
+     */
+    final Map<String, Boolean> closeUuids;
+
     final Principal principal;
 
@@ -172 +177 @@
         // Setting up in memory store.
         Ehcache cache = manager.inMemoryCache(p.getName());
-        activeServants = getServantMap(ACTIVE_SERVANTS, cache);
-        interactiveSlots = getServantMap(INTERACTIVE_SLOTS, cache);
+        // TODO the getMap() method could be put in our own Cache class
+        activeServants = getMap(ACTIVE_SERVANTS, cache);
+        interactiveSlots = getMap(INTERACTIVE_SLOTS, cache);
+        closeUuids = getMap(CLOSE_UUIDS, cache);
     }
 
     static String ACTIVE_SERVANTS = "activeServants";
     static String INTERACTIVE_SLOTS = "interactiveSlots";
-
-    private Map<String, Ice.Object> getServantMap(String key, Ehcache cache) {
-        Map<String, Ice.Object> ids;
+    static String CLOSE_UUIDS = "closeUuids";
+
+    @SuppressWarnings("unchecked")
+    private <T> Map<String, T> getMap(String key, Ehcache cache) {
+        Map<String, T> ids;
         if (!cache.isKeyInCache(key)) {
-            ids = new ConcurrentHashMap<String, Ice.Object>();
+            ids = new ConcurrentHashMap<String, T>();
             cache.put(new Element(key, ids));
         } else {
-            ids = (Map<String, Ice.Object>) cache.get(key).getObjectValue();
+            ids = (Map<String, T>) cache.get(key).getObjectValue();
         }
         return ids;
@@ -464 +473 @@
     }
 
+    public void detachOnDestroy(Ice.Current current) {
+        String key = current.ctx.get(omero.constants.CLIENTUUID.value);
+        closeUuids.remove(key);
+    }
+
+    @Deprecated
     public void close(Ice.Current current) {
-
-        if (log.isInfoEnabled()) {
-            log.info(String.format("Closing %s session", this));
-        }
-
-        // Cleaning up resources
-
-        // INTERACTIVE
-        Set<String> ipIds = interactiveSlots.keySet();
-        for (String id : ipIds) {
-            InteractiveProcessorI ip = (InteractiveProcessorI) interactiveSlots
-                    .get(id);
-            ip.stop();
-            current.adapter.remove(Ice.Util.stringToIdentity(id));
-            interactiveSlots.remove(id);
-        }
-
-        // SERVANTS
-        // Here we call the "close()" method on all methods which require that
-        // logic allowing the IceMethodInvoker to raise the
-        // UnregisterServantEvent, otherwise there is a recursive call back
-        // to close.
-        List<String> ids = activeServices(current);
-        for (String idString : ids) {
-            Ice.Identity id = Ice.Util.stringToIdentity(idString);
-            Ice.Object obj = current.adapter.find(id);
-            try {
-                if (obj instanceof StatefulServiceInterface) {
-                    Method m = obj.getClass().getMethod("close",
-                            Ice.Current.class);
-                    Ice.Current __curr = new Ice.Current();
-                    __curr.id = id;
-                    __curr.adapter = current.adapter;
-                    __curr.operation = "close";
-                    __curr.mode = current.mode; // FIXME due to bug
-                    // http://www.zeroc.com/forums/bug-reports/3348-ice-current-hashcode-can-throw-npe-null-enum.html
-                    m.invoke(obj, __curr);
-                } else {
-                    unregisterServant(id, current);
-                }
-            } catch (Exception e) {
-                log.error("Failure to close: " + idString + "=" + obj, e);
-            }
-        }
-
-        // All resources cleaned up to the best of our ability,
-        // now we can remove the current session. If an exception if thrown,
-        // there's not much we can do.
-        try {
-            current.adapter.remove(ServiceFactoryI.sessionId(principal
-                    .getName()));
-            sessionManager.close(this.principal.getName());
-        } catch (Throwable t) {
-            // FIXME
-            InternalException ie = new InternalException(t.getMessage());
-            ie.setStackTrace(t.getStackTrace());
-        }
+        closeOnDestroy(current);
+    }
+
+    public void closeOnDestroy(Ice.Current current) {
+        String key = current.ctx.get(omero.constants.CLIENTUUID.value);
+        closeUuids.put(key, Boolean.TRUE);
     }
 
@@ -531 +495 @@
      */
     public void destroy(Ice.Current current) {
-        sessionManager.detach(this.principal.getName());
+
+        // First detach and get the reference count.
+        int ref = sessionManager.detach(this.principal.getName());
+
+        // If we are supposed to close, do only so if the ref count
+        // is < 1.
+        String key = current.ctx.get(omero.constants.CLIENTUUID.value);
+        if (key != null && ref < 1) {
+
+            ref = sessionManager.close(this.principal.getName());
+
+            // If someone has attached since we called detach() above, do not
+            // destroy this session.
+            if (ref > 0) {
+                return;
+            }
+
+            if (log.isInfoEnabled()) {
+                log.info(String.format("Closing %s session", this));
+            }
+
+            // Cleaning up resources
+
+            // INTERACTIVE
+            Set<String> ipIds = interactiveSlots.keySet();
+            for (String id : ipIds) {
+                InteractiveProcessorI ip = (InteractiveProcessorI) interactiveSlots
+                        .get(id);
+                ip.stop();
+                current.adapter.remove(Ice.Util.stringToIdentity(id));
+                interactiveSlots.remove(id);
+            }
+
+            // SERVANTS
+            // Here we call the "close()" method on all methods which require
+            // that
+            // logic allowing the IceMethodInvoker to raise the
+            // UnregisterServantEvent, otherwise there is a recursive call back
+            // to close.
+            List<String> ids = activeServices(current);
+            for (String idString : ids) {
+                Ice.Identity id = Ice.Util.stringToIdentity(idString);
+                Ice.Object obj = current.adapter.find(id);
+                try {
+                    if (obj instanceof StatefulServiceInterface) {
+                        Method m = obj.getClass().getMethod("close",
+                                Ice.Current.class);
+                        Ice.Current __curr = new Ice.Current();
+                        __curr.id = id;
+                        __curr.adapter = current.adapter;
+                        __curr.operation = "close";
+                        __curr.mode = current.mode; // FIXME due to bug
+                        // http://www.zeroc.com/forums/bug-reports/3348-ice-current-hashcode-can-throw-npe-null-enum.html
+                        m.invoke(obj, __curr);
+                    } else {
+                        unregisterServant(id, current);
+                    }
+                } catch (Exception e) {
+                    log.error("Failure to close: " + idString + "=" + obj, e);
+                }
+            }
+
+            // All resources cleaned up to the best of our ability,
+            // now we can remove the current session. If an exception if thrown,
+            // there's not much we can do.
+            try {
+                current.adapter.remove(ServiceFactoryI.sessionId(principal
+                        .getName()));
+            } catch (Throwable t) {
+                // FIXME
+                InternalException ie = new InternalException(t.getMessage());
+                ie.setStackTrace(t.getStackTrace());
+            }
+        }
     }
 

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

@@ -18 +18 @@
 from omero_ext import pysys
 import omero_Constants_ice
+import uuid
 
 class client(object):
@@ -31 +32 @@
 
         self.sf = None
-        self.ic = None
-        ic = Ice.initialize(args,id)
-        if not ic:
+        self.ic = Ice.initialize(args,id)
+        if not self.ic:
             raise ClientError("Improper initialization")
+        # Register Object Factory
         self.of = ObjectFactory()
-        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
+        self.of.registerObjectFactory(self.ic)
+        # Define our unique identifier (used during close/detach)
+        self.ic.getImplicitContext().put(omero.constants.CLIENT_UUID, str(uuid.uuid4()))
 
     def __del__(self):
@@ -60 +45 @@
             self.destroyConnection()
         except exceptions.Exception, e:
-            print "Error in __del__:" + str(e.__class__)
+            print "Ignoring error in client.__del__:" + str(e.__class__)
 
     def getCommunicator(self):
@@ -216 +201 @@
             file.close()
 
-    def destroyConnection(self):
+    def closeSession(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 'sf' exists we remove it.
+        if not hasattr(self, 'sf'):
+            self.sf = None
+
         # 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()
@@ -250 +237 @@
         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'):
-            return
-        # But even if we do have 'sf', the connection may have been lost and 'close' will fail
-        try:
-            try:
-                self.sf.close()
-            except:
-                pass
-        finally:
-            self.sf = None
-
     def _env(self, method, *args):
         """ Helper method to access session environment"""