OF-573: Improve session concurrency and consistency across local and remote...

OF-573: Improve session concurrency and consistency across local and remote (clustered) client sessions

git-svn-id: http://svn.igniterealtime.org/svn/repos/openfire/trunk@13312 b35dd754-fafc-0310-a699-88a17e54d16e

src/java/org/jivesoftware/openfire/http/HttpSession.java

@@ -53,6 +53,7 @@ import org.jivesoftware.openfire.net.VirtualConnection;
 import org.jivesoftware.openfire.session.LocalClientSession;
 import org.jivesoftware.util.JiveConstants;
 import org.jivesoftware.util.JiveGlobals;
+import org.jivesoftware.util.TaskEngine;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.xmlpull.v1.XmlPullParserException;
@@ -1002,61 +1003,60 @@ public class HttpSession extends LocalClientSession {
 	        isClosed = true;
     	}
 
-        if (pendingElements.size() > 0) {
-            failDelivery();
-        }
-
-        for (SessionListener listener : listeners) {
-            listener.sessionClosed(this);
+        // close connection(s) and deliver pending elements (if any)
+        synchronized (connectionQueue) {
+	        for (HttpConnection toClose : connectionQueue) {
+	            try {
+	            	if (!toClose.isClosed()) {
+	            		if (!pendingElements.isEmpty() && toClose.getRequestId() == lastRequestID + 1) {
+	            			synchronized(pendingElements) {
+		            			deliver(toClose, pendingElements);
+				                lastRequestID = toClose.getRequestId();
+				                pendingElements.clear();
+	            			}
+            			} else {
+            				toClose.deliverBody(null);
+            			}
+	            	}
+	            } catch (HttpConnectionClosedException e) {
+	            	/* ignore ... already closed */
+	            }
+	        }
         }
-        this.listeners.clear();
-    }
 
-    private void failDelivery() {
     	synchronized (pendingElements) {
 	        for (Deliverable deliverable : pendingElements) {
-	            Collection<Packet> packet = deliverable.getPackets();
-	            if (packet != null) {
-	                failDelivery(packet);
-	            }
+	            failDelivery(deliverable.getPackets());
 	        }
 	        pendingElements.clear();
-    	}
+        }
 
-        synchronized (connectionQueue) {
-	        for (HttpConnection toClose : connectionQueue) {
-	            if (!toClose.isDelivered()) {
-	                Delivered delivered = retrieveDeliverable(toClose.getRequestId());
-	                if (delivered != null) {
-	                    failDelivery(delivered.getPackets());
-	                }
-	                else {
-	                    Log.warn("Packets could not be found for session " + getStreamID() + " cannot " +
-	                            "be delivered to client");
-	                }
-	            }
-	            toClose.close();
-	            fireConnectionClosed(toClose);
-	        }
+        for (SessionListener listener : listeners) {
+            listener.sessionClosed(this);
         }
+        this.listeners.clear();
     }
 
-    private void failDelivery(Collection<Packet> packets) {
+    private void failDelivery(final Collection<Packet> packets) {
         if (packets == null) {
             // Do nothing if someone asked to deliver nothing :)
             return;
         }
-        for (Packet packet : packets) {
-            try {
-                backupDeliverer.deliver(packet);
-            }
-            catch (UnauthorizedException e) {
-                Log.error("Unable to deliver message to backup deliverer", e);
-            }
-        }
+        // use a separate thread to schedule backup delivery
+   		TaskEngine.getInstance().submit(new Runnable() {
+			public void run() {
+		        for (Packet packet : packets) {
+    	            try {
+        				backupDeliverer.deliver(packet);
+    	            }
+    	            catch (UnauthorizedException e) {
+    	                Log.error("Unable to deliver message to backup deliverer", e);
+    	            }
+		        }
+			}
+   		});
     }
 
-
     private String createEmptyBody() {
         Element body = DocumentHelper.createElement("body");
         body.addNamespace("", "http://jabber.org/protocol/httpbind");

src/java/org/jivesoftware/openfire/spi/RoutingTableImpl.java

@@ -65,6 +65,7 @@ public class RoutingTableImpl extends BasicModule implements RoutingTable, Clust
     public static final String ANONYMOUS_C2S_CACHE_NAME = "Routing AnonymousUsers Cache";
     public static final String S2S_CACHE_NAME = "Routing Servers Cache";
     public static final String COMPONENT_CACHE_NAME = "Routing Components Cache";
+    public static final String C2S_SESSION_NAME = "Routing User Sessions";
 
     /**
      * Cache (unlimited, never expire) that holds outgoing sessions to remote servers from this server.
@@ -108,7 +109,7 @@ public class RoutingTableImpl extends BasicModule implements RoutingTable, Clust
         componentsCache = CacheFactory.createCache(COMPONENT_CACHE_NAME);
         usersCache = CacheFactory.createCache(C2S_CACHE_NAME);
         anonymousUsersCache = CacheFactory.createCache(ANONYMOUS_C2S_CACHE_NAME);
-        usersSessions = CacheFactory.createCache("Routing User Sessions");
+        usersSessions = CacheFactory.createCache(C2S_SESSION_NAME);
         localRoutingTable = new LocalRoutingTable();
     }
 
@@ -313,6 +314,9 @@ public class RoutingTableImpl extends BasicModule implements RoutingTable, Clust
 		                if (remotePacketRouter != null) {
 		                    routed = remotePacketRouter
 		                            .routePacket(clientRoute.getNodeID().toByteArray(), jid, packet);
+		                    if (!routed) {
+		                    	removeClientRoute(jid); // drop invalid client route
+		                    }
 		                }
 		            }
 		        }
@@ -706,19 +710,26 @@ public class RoutingTableImpl extends BasicModule implements RoutingTable, Clust
             }
             else {
                 // Address is a bare JID so return all AVAILABLE resources of user
-                Collection<String> sessions = usersSessions.get(route.toBareJID());
-                if (sessions != null) {
-                    // Select only available sessions
-                    for (String jid : sessions) {
-                        ClientRoute clientRoute = usersCache.get(jid);
-                        if (clientRoute == null) {
-                            clientRoute = anonymousUsersCache.get(jid);
-                        }
-                        if (clientRoute != null && (clientRoute.isAvailable() ||
-                                presenceUpdateHandler.hasDirectPresence(new JID(jid), requester))) {
-                            jids.add(new JID(jid));
-                        }
-                    }
+                Lock lock = CacheFactory.getLock(route.toBareJID(), usersSessions);
+                try {
+                    lock.lock(); // temporarily block new sessions for this JID
+	                Collection<String> sessions = usersSessions.get(route.toBareJID());
+	                if (sessions != null) {
+	                    // Select only available sessions
+	                    for (String jid : sessions) {
+	                        ClientRoute clientRoute = usersCache.get(jid);
+	                        if (clientRoute == null) {
+	                            clientRoute = anonymousUsersCache.get(jid);
+	                        }
+	                        if (clientRoute != null && (clientRoute.isAvailable() ||
+	                                presenceUpdateHandler.hasDirectPresence(new JID(jid), requester))) {
+	                            jids.add(new JID(jid));
+	                        }
+	                    }
+	                }
+                }
+                finally {
+                	lock.unlock();
                 }
             }
         }

src/plugins/hazelcast/src/java/com/jivesoftware/openfire/session/ClientSessionTask.java

@@ -19,22 +19,23 @@
 
 package com.jivesoftware.openfire.session;
 
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+
 import org.jivesoftware.openfire.XMPPServer;
 import org.jivesoftware.openfire.cluster.NodeID;
 import org.jivesoftware.openfire.session.ClientSession;
 import org.jivesoftware.openfire.session.Session;
 import org.jivesoftware.openfire.spi.ClientRoute;
 import org.jivesoftware.openfire.spi.RoutingTableImpl;
-import org.jivesoftware.util.Log;
 import org.jivesoftware.util.cache.Cache;
 import org.jivesoftware.util.cache.CacheFactory;
 import org.jivesoftware.util.cache.ExternalizableUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.xmpp.packet.JID;
 
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-
 /**
  * Class that defines possible remote operations that could be performed
  * on remote client sessions.
@@ -42,7 +43,11 @@ import java.io.ObjectOutput;
  * @author Gaston Dombiak
  */
 public class ClientSessionTask extends RemoteSessionTask {
+
+    private static Logger logger = LoggerFactory.getLogger(ClientSessionTask.class);
+
     private JID address;
+    private transient Session session;
 
     public ClientSessionTask() {
         super();
@@ -54,10 +59,17 @@ public class ClientSessionTask extends RemoteSessionTask {
     }
 
     Session getSession() {
-        return XMPPServer.getInstance().getRoutingTable().getClientRoute(address);
+    	if (session == null) {
+    		session = XMPPServer.getInstance().getRoutingTable().getClientRoute(address);
+    	}
+    	return session;
     }
 
     public void run() {
+    	if (getSession() == null || getSession().isClosed()) {
+    		logger.error("Session not found for JID: " + address);
+    		return;
+    	}
         super.run();
 
         ClientSession session = (ClientSession) getSession();
@@ -67,7 +79,7 @@ public class ClientSessionTask extends RemoteSessionTask {
             ClientRoute route = usersCache.get(address.toString());
             NodeID nodeID = route.getNodeID();
 
-            Log.warn("Found remote session instead of local session. JID: " + address + " found in Node: " +
+            logger.warn("Found remote session instead of local session. JID: " + address + " found in Node: " +
                     nodeID.toByteArray() + " and local node is: " + XMPPServer.getInstance().getNodeID().toByteArray());
         }
         if (operation == Operation.isInitialized) {

src/plugins/hazelcast/src/java/com/jivesoftware/openfire/session/RemoteSession.java

@@ -19,6 +19,10 @@
 
 package com.jivesoftware.openfire.session;
 
+import java.net.UnknownHostException;
+import java.util.Date;
+
+import org.jivesoftware.openfire.SessionManager;
 import org.jivesoftware.openfire.StreamID;
 import org.jivesoftware.openfire.session.Session;
 import org.jivesoftware.util.cache.CacheFactory;
@@ -26,9 +30,6 @@ import org.jivesoftware.util.cache.ClusterTask;
 import org.xmpp.packet.JID;
 import org.xmpp.packet.Packet;
 
-import java.net.UnknownHostException;
-import java.util.Date;
-
 /**
  * Base class for sessions being hosted in other cluster nodes. Almost all
  * messages will be forwarded to the actual session in some remote cluster node.
@@ -168,7 +169,16 @@ public abstract class RemoteSession implements Session {
      * @throws IllegalStateException if requested node was not found or not running in a cluster.
      */
     protected Object doSynchronousClusterTask(ClusterTask task) {
-        return CacheFactory.doSynchronousClusterTask(task, nodeID);
+        try {
+        	return CacheFactory.doSynchronousClusterTask(task, nodeID);
+        } catch (IllegalStateException ise) {
+        	if (task instanceof RemoteSessionTask) {
+	        	// clean up invalid session
+	        	SessionManager.getInstance().removeSession(null, 
+	        			((RemoteSessionTask)task).getSession().getAddress(), false, false);
+        	}
+        	throw ise;
+        }
     }
 
     /**
@@ -178,7 +188,16 @@ public abstract class RemoteSession implements Session {
      * @throws IllegalStateException if requested node was not found or not running in a cluster. 
      */
     protected void doClusterTask(ClusterTask task) {
-        CacheFactory.doClusterTask(task, nodeID);
+        try {
+        	CacheFactory.doClusterTask(task, nodeID);
+        } catch (IllegalStateException ise) {
+        	if (task instanceof RemoteSessionTask) {
+	        	// clean up invalid session
+	        	SessionManager.getInstance().removeSession(null, 
+	        			((RemoteSessionTask)task).getSession().getAddress(), false, false);
+        	}
+        	throw ise;
+        }
     }
 
     /**

src/plugins/hazelcast/src/java/com/jivesoftware/util/cluster/ClusterPacketRouter.java

@@ -46,7 +46,7 @@ public class ClusterPacketRouter implements RemotePacketRouter {
             CacheFactory.doClusterTask(new RemotePacketExecution(receipient, packet), nodeID);
             return true;
         } catch (IllegalStateException  e) {
-            logger.warn("Error while routing packet to remote node", e);
+            logger.warn("Error while routing packet to remote node: " + e);
             return false;
         }
     }