OF-39 Initial work on removing in memory caching of all node items. ...

OF-39 Initial work on removing in memory caching of all node items.  Everything is done through the database with some buffering to keep reasonable performance.

git-svn-id: http://svn.igniterealtime.org/svn/repos/openfire/branches/pubsub_clustering@12915 b35dd754-fafc-0310-a699-88a17e54d16e

src/java/org/jivesoftware/openfire/pep/PEPService.java

@@ -23,12 +23,9 @@ package org.jivesoftware.openfire.pep;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.Map;
-import java.util.Queue;
 import java.util.Set;
 import java.util.TimeZone;
-import java.util.Timer;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingQueue;
 
 import org.dom4j.DocumentHelper;
 import org.dom4j.Element;
@@ -49,7 +46,6 @@ import org.jivesoftware.openfire.pubsub.PubSubEngine;
 import org.jivesoftware.openfire.pubsub.PubSubPersistenceManager;
 import org.jivesoftware.openfire.pubsub.PubSubService;
 import org.jivesoftware.openfire.pubsub.PublishedItem;
-import org.jivesoftware.openfire.pubsub.PublishedItemTask;
 import org.jivesoftware.openfire.pubsub.models.AccessModel;
 import org.jivesoftware.openfire.pubsub.models.PublisherModel;
 import org.jivesoftware.openfire.roster.Roster;
@@ -77,12 +73,6 @@ import org.xmpp.packet.PacketExtension;
  */
 public class PEPService implements PubSubService, Cacheable {
 	
-    /**
-     * Timer to save published items to the database or remove deleted or old
-     * items.
-     */
-    private static final Timer timer = new Timer("PEP service maintenance");
-
     /**
      * Date format to use for time stamps in delayed event notifications.
      */
@@ -134,16 +124,6 @@ public class PEPService implements PubSubService, Cacheable {
      */
     private Map<String, Map<String, String>> barePresences = new ConcurrentHashMap<String, Map<String, String>>();
 
-    /**
-     * Queue that holds the items that need to be added to the database.
-     */
-    private Queue<PublishedItem> itemsToAdd = new LinkedBlockingQueue<PublishedItem>(10000);
-
-    /**
-     * Queue that holds the items that need to be deleted from the database.
-     */
-    private Queue<PublishedItem> itemsToDelete = new LinkedBlockingQueue<PublishedItem>(10000);
-
     /**
      * Manager that keeps the list of ad-hoc commands and processing command
      * requests.
@@ -155,17 +135,6 @@ public class PEPService implements PubSubService, Cacheable {
      */
     private EntityCapabilitiesManager entityCapsManager = EntityCapabilitiesManager.getInstance();
 
-    /**
-     * The time to elapse between each execution of the maintenance process.
-     * Default is 2 minutes.
-     */
-    private int items_task_timeout = 2 * 60 * 1000;
-
-    /**
-     * Task that saves or deletes published items from the database.
-     */
-    private PublishedItemTask publishedItemTask;
-
     static {
         fastDateFormat = FastDateFormat.getInstance(JiveConstants.XMPP_DATETIME_FORMAT, TimeZone.getTimeZone("UTC"));
     }
@@ -184,13 +153,6 @@ public class PEPService implements PubSubService, Cacheable {
         adHocCommandManager = new AdHocCommandManager();
         adHocCommandManager.addCommand(new PendingSubscriptionsCommand(this));
 
-        // Save or delete published items from the database every 2 minutes
-        // starting in 2 minutes (default values)
-        publishedItemTask = new PublishedItemTask(this) {
-        	
-        };
-        timer.schedule(publishedItemTask, items_task_timeout, items_task_timeout);
-
         // Load default configuration for leaf nodes
         leafDefaultConfiguration = PubSubPersistenceManager.loadDefaultConfiguration(this, true);
         if (leafDefaultConfiguration == null) {
@@ -551,50 +513,14 @@ public class PEPService implements PubSubService, Cacheable {
         }
     }
 
-    public void queueItemToAdd(PublishedItem newItem) {
-        PubSubEngine.queueItemToAdd(this, newItem);
-    }
-
-    public void queueItemToRemove(PublishedItem removedItem) {
-        PubSubEngine.queueItemToRemove(this, removedItem);
-    }
-
     public Map<String, Map<String, String>> getBarePresences() {
         return barePresences;
     }
 
-    public Queue<PublishedItem> getItemsToAdd() {
-        return itemsToAdd;
-    }
-
-    public Queue<PublishedItem> getItemsToDelete() {
-        return itemsToDelete;
-    }
-
     public AdHocCommandManager getManager() {
         return adHocCommandManager;
     }
 
-    public PublishedItemTask getPublishedItemTask() {
-        return publishedItemTask;
-    }
-
-    public void setPublishedItemTask(PublishedItemTask task) {
-        publishedItemTask = task;
-    }
-
-    public Timer getTimer() {
-        return timer;
-    }
-
-    public int getItemsTaskTimeout() {
-        return items_task_timeout;
-    }
-
-    public void setItemsTaskTimeout(int timeout) {
-        items_task_timeout = timeout;
-    }
-
 	public int getCachedSize() {
 		// Rather arbitrary. Don't use this for size-based eviction policies!
 		return 600;

src/java/org/jivesoftware/openfire/pep/PublishedPEPServiceTask.java

-/**
- * $RCSfile: $
- * $Revision: $
- * $Date: $
- *
- * Copyright (C) 2005-2008 Jive Software. All rights reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.jivesoftware.openfire.pep;
-
-import org.jivesoftware.openfire.pubsub.PublishedItemTask;
-
-/**
- * TimerTask that unloads services from memory, after they have been expired
- * from cache.
- * 
- * @author Guus der Kinderen, guus.der.kinderen@gmail.com
- */
-public class PublishedPEPServiceTask extends PublishedItemTask {
-
-	private final PEPServiceManager manager;
-
-	public PublishedPEPServiceTask(PEPService service, PEPServiceManager manager) {
-		super(service);
-		this.manager = manager;
-	}
-
-	@Override
-	public void run() {
-		// Somewhat of a hack to unload the PEPService after it has been removed
-		// from the cache. New scheduled packets will re-instate the service.
-		PEPService service = (PEPService) this.getService();
-		if (manager.hasCachedService(service.getAddress())) {
-			super.run();
-		} else {
-			manager.unload(service);
-		}
-	}
-}

src/java/org/jivesoftware/openfire/pubsub/Node.java

@@ -348,7 +348,7 @@ public abstract class Node {
 
         if (savedToDB) {
             // Add or update the affiliate in the database
-            PubSubPersistenceManager.saveAffiliation(service, this, affiliate, created);
+            PubSubPersistenceManager.saveAffiliation(this, affiliate, created);
         }
         return affiliate;
     }
@@ -367,7 +367,7 @@ public abstract class Node {
         affiliates.remove(affiliate);
         if (savedToDB) {
             // Remove the affiliate from the database
-            PubSubPersistenceManager.removeAffiliation(service, this, affiliate);
+            PubSubPersistenceManager.removeAffiliation(this, affiliate);
         }
     }
 
@@ -1133,6 +1133,15 @@ public abstract class Node {
         return false;
     }
 
+    /**
+     * Returns the {@link PubSubService} to which this node belongs.
+     *
+     * @return the pubsub service.
+     */
+    public PubSubService getService() {
+        return service;
+    }
+
     /**
      * Returns the unique identifier for a node within the context of a pubsub service.
      *
@@ -1698,16 +1707,16 @@ public abstract class Node {
     public void saveToDB() {
         // Make the room persistent
         if (!savedToDB) {
-            PubSubPersistenceManager.createNode(service, this);
+            PubSubPersistenceManager.createNode(this);
             // Set that the node is now in the DB
             setSavedToDB(true);
             // Save the existing node affiliates to the DB
             for (NodeAffiliate affialiate : affiliates) {
-                PubSubPersistenceManager.saveAffiliation(service, this, affialiate, true);
+                PubSubPersistenceManager.saveAffiliation(this, affialiate, true);
             }
             // Add new subscriptions to the database
             for (NodeSubscription subscription : subscriptionsByID.values()) {
-                PubSubPersistenceManager.saveSubscription(service, this, subscription, true);
+                PubSubPersistenceManager.saveSubscription(this, subscription, true);
             }
             // Add the new node to the list of available nodes
             service.addNode(this);
@@ -1717,7 +1726,7 @@ public abstract class Node {
             }
         }
         else {
-            PubSubPersistenceManager.updateNode(service, this);
+            PubSubPersistenceManager.updateNode(this);
         }
     }
 
@@ -1776,7 +1785,7 @@ public abstract class Node {
      */
     public boolean delete() {
         // Delete node from the database
-        if (PubSubPersistenceManager.removeNode(service, this)) {
+        if (PubSubPersistenceManager.removeNode(this)) {
             // Remove this node from the parent node (if any)
             if (parent != null) {
                 parent.removeChildNode(this);
@@ -1837,7 +1846,7 @@ public abstract class Node {
             parent.addChildNode(this);
         }
         if (savedToDB) {
-            PubSubPersistenceManager.updateNode(service, this);
+            PubSubPersistenceManager.updateNode(this);
         }
     }
 
@@ -2018,7 +2027,7 @@ public abstract class Node {
 
         if (savedToDB) {
             // Add the new subscription to the database
-            PubSubPersistenceManager.saveSubscription(service, this, subscription, true);
+            PubSubPersistenceManager.saveSubscription(this, subscription, true);
         }
 
         if (originalIQ != null) {
@@ -2072,7 +2081,7 @@ public abstract class Node {
         }
         if (savedToDB) {
             // Remove the subscription from the database
-            PubSubPersistenceManager.removeSubscription(service, this, subscription);
+            PubSubPersistenceManager.removeSubscription(subscription);
         }
         // Check if we need to unsubscribe from the presence of the owner
         if (isPresenceBasedDelivery() && getSubscriptions(subscription.getOwner()).isEmpty()) {

src/java/org/jivesoftware/openfire/pubsub/NodeSubscription.java

@@ -514,7 +514,7 @@ public class NodeSubscription {
         }
         if (savedToDB) {
             // Update the subscription in the backend store
-            PubSubPersistenceManager.saveSubscription(service, node, this, false);
+            PubSubPersistenceManager.saveSubscription(node, this, false);
         }
         // Check if the service needs to subscribe or unsubscribe from the owner presence
         if (!node.isPresenceBasedDelivery() && wasUsingPresence != !presenceStates.isEmpty()) {
@@ -878,7 +878,7 @@ public class NodeSubscription {
 
         if (savedToDB) {
             // Update the subscription in the backend store
-            PubSubPersistenceManager.saveSubscription(service, node, this, false);
+            PubSubPersistenceManager.saveSubscription(node, this, false);
         }
 
         // Send last published item (if node is leaf node and subscription status is ok)

src/java/org/jivesoftware/openfire/pubsub/PubSubEngine.java

@@ -1732,23 +1732,7 @@ public class PubSubEngine {
     }
 
     public void shutdown(PubSubService service) {
-        // Stop the maintenance processes
-    	service.getPublishedItemTask().cancel();
-        // Delete from the database items contained in the itemsToDelete queue
-        PublishedItem entry;
-        while (!service.getItemsToDelete().isEmpty()) {
-            entry = service.getItemsToDelete().poll();
-            if (entry != null) {
-                PubSubPersistenceManager.removePublishedItem(service, entry);
-            }
-        }
-        // Save to the database items contained in the itemsToAdd queue
-        while (!service.getItemsToAdd().isEmpty()) {
-            entry = service.getItemsToAdd().poll();
-            if (entry != null) {
-                PubSubPersistenceManager.createPublishedItem(service, entry);
-            }
-        }
+    	PubSubPersistenceManager.shutdown();
         // Stop executing ad-hoc commands
         service.getManager().stop();
         
@@ -1843,77 +1827,6 @@ public class PubSubEngine {
         }
     }
 
-    /*******************************************************************************
-     * Methods related to PubSub maintenance tasks. Such as
-     * saving or deleting published items.
-     ******************************************************************************/
-
-    /**
-     * Schedules the maintenance task for repeated <i>fixed-delay execution</i>,
-     * beginning after the specified delay.  Subsequent executions take place
-     * at approximately regular intervals separated by the specified period.
-     *
-     * @param service the PubSub service this action is to be performed for.
-     * @param timeout the new frequency of the maintenance task.
-     */
-    void setPublishedItemTaskTimeout(PubSubService service, int timeout) {
-        int items_task_timeout = service.getItemsTaskTimeout();
-        if (items_task_timeout == timeout) {
-            return;
-        }
-        // Cancel the existing task because the timeout has changed
-        PublishedItemTask publishedItemTask = service.getPublishedItemTask();
-        if (publishedItemTask != null) {
-            publishedItemTask.cancel();
-        }
-        service.setItemsTaskTimeout(timeout);
-        // Create a new task and schedule it with the new timeout
-        service.setPublishedItemTask(new PublishedItemTask(service));
-        service.getTimer().schedule(publishedItemTask, items_task_timeout, items_task_timeout);
-    }
-
-    /**
-     * Adds the item to the queue of items to remove from the database. The queue is going
-     * to be processed by another thread.
-     *
-     * @param service the PubSub service this action is to be performed for.
-     * @param removedItem the item to remove from the database.
-     */
-    public static void queueItemToRemove(PubSubService service, PublishedItem removedItem) {
-        // Remove the removed item from the queue of items to add to the database
-        if (!service.getItemsToAdd().remove(removedItem)) {
-            // The item is already present in the database so add the removed item
-            // to the queue of items to delete from the database
-            service.getItemsToDelete().add(removedItem);
-        }
-    }
-
-    /**
-     * Adds the item to the queue of items to add to the database. The queue is going
-     * to be processed by another thread.
-     *
-     * @param service the PubSub service this action is to be performed for.
-     * @param newItem the item to add to the database.
-     */
-    public static void queueItemToAdd(PubSubService service, PublishedItem newItem) {
-        service.getItemsToAdd().add(newItem);
-    }
-
-    /**
-     * Cancels any queued operation for the specified list of items. This operation is
-     * usually required when a node was deleted so any pending operation of the node items
-     * should be cancelled.
-     *
-     * @param service the PubSub service this action is to be performed for.
-     * @param items the list of items to remove the from queues.
-     */
-    void cancelQueuedItems(PubSubService service, Collection<PublishedItem> items) {
-        for (PublishedItem item : items) {
-            service.getItemsToAdd().remove(item);
-            service.getItemsToDelete().remove(item);
-        }
-    }
-
     /**
 	 * Checks to see if the jid given is a component by looking at the routing
 	 * table. Similar to {@link InternalComponentManager#hasComponent(JID)}.

src/java/org/jivesoftware/openfire/pubsub/PubSubModule.java

@@ -100,36 +100,11 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
     private Map<String, Map<String, String>> barePresences =
             new ConcurrentHashMap<String, Map<String, String>>();
     
-    /**
-     * Queue that holds the items that need to be added to the database.
-     */
-    private Queue<PublishedItem> itemsToAdd = new LinkedBlockingQueue<PublishedItem>(10000);
-    /**
-     * Queue that holds the items that need to be deleted from the database.
-     */
-    private Queue<PublishedItem> itemsToDelete = new LinkedBlockingQueue<PublishedItem>(10000);
-    
     /**
      * Manager that keeps the list of ad-hoc commands and processing command requests.
      */
     private AdHocCommandManager manager;
     
-    /**
-     * The time to elapse between each execution of the maintenance process. Default
-     * is 2 minutes.
-     */
-    private int items_task_timeout = 2 * 60 * 1000;
-
-    /**
-     * Task that saves or deletes published items from the database.
-     */
-    private PublishedItemTask publishedItemTask;
-
-    /**
-     * Timer to save published items to the database or remove deleted or old items.
-     */
-    private Timer timer = new Timer("PubSub maintenance");
-
     /**
      * Returns the permission policy for creating nodes. A true value means that not anyone can
      * create a node, only the JIDs listed in <code>allowedToCreate</code> are allowed to create
@@ -188,11 +163,6 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
         // Initialize the ad-hoc commands manager to use for this pubsub service
         manager = new AdHocCommandManager();
         manager.addCommand(new PendingSubscriptionsCommand(this));
-        
-        // Save or delete published items from the database every 2 minutes starting in
-        // 2 minutes (default values)
-        publishedItemTask = new PublishedItemTask(this);
-        timer.schedule(publishedItemTask, items_task_timeout, items_task_timeout);
     }
 
     public void process(Packet packet) {
@@ -303,14 +273,6 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
         PubSubEngine.presenceSubscriptionRequired(this, node, user);
     }
 
-    public void queueItemToAdd(PublishedItem newItem) {
-        PubSubEngine.queueItemToAdd(this, newItem);
-    }
-
-    public void queueItemToRemove(PublishedItem removedItem) {
-        PubSubEngine.queueItemToRemove(this, removedItem);
-    }
-
     public String getServiceName() {
         return serviceName;
     }
@@ -828,38 +790,10 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
         return barePresences;
     }
 
-    public Queue<PublishedItem> getItemsToAdd() {
-        return itemsToAdd;
-    }
-
-    public Queue<PublishedItem> getItemsToDelete() {
-        return itemsToDelete;
-    }
-
     public AdHocCommandManager getManager() {
         return manager;
     }
 
-    public PublishedItemTask getPublishedItemTask() {
-        return publishedItemTask;
-    }
-
-    public void setPublishedItemTask(PublishedItemTask task) {
-        publishedItemTask = task;
-    }
-
-    public Timer getTimer() {
-        return timer;
-    }
-    
-    public int getItemsTaskTimeout() {
-        return items_task_timeout;
-    }
-
-    public void setItemsTaskTimeout(int timeout) {
-        items_task_timeout = timeout;
-    }
-
     public void propertySet(String property, Map<String, Object> params) {
         if (property.equals("xmpp.pubsub.enabled")) {
             boolean enabled = Boolean.parseBoolean((String)params.get("value"));

src/java/org/jivesoftware/openfire/pubsub/PubSubService.java

@@ -239,76 +239,10 @@ public interface PubSubService {
      */
     boolean isMultipleSubscriptionsEnabled();
 
-    /**
-     * Adds the item to the queue of items to add to the database. The queue is going
-     * to be processed by another thread.
-     *
-     * @param newItem the item to add to the database.
-     */
-    void queueItemToAdd(PublishedItem newItem);
-
-    /**
-     * Gets the queue that holds the items that need to be added to the database.
-     * 
-     * @return the queue that holds the items that need to be added to the database.
-     */
-    Queue<PublishedItem> getItemsToAdd();
-
-    /**
-     * Gets the queue that holds the items that need to be deleted from the database.
-     * 
-     * @return the queue that holds the items that need to be deleted from the database.
-     */
-    Queue<PublishedItem> getItemsToDelete();
-
     /**
      * Returns the ad-hoc commands manager used for this service.
      * 
      * @return the ad-hoc commands manager used for this service.
      */
     AdHocCommandManager getManager();
-
-    /**
-     * Returns the published item task used for this service.
-     * 
-     * @return the published item task used for this service.
-     */
-    PublishedItemTask getPublishedItemTask();
-
-    /**
-     * Sets the published item task used for this service.
-     * 
-     * @param task the PublishedItemTask to set for this service.
-     */
-    void setPublishedItemTask(PublishedItemTask task);
-
-    /**
-     * Adds the item to the queue of items to remove from the database. The queue is going
-     * to be processed by another thread.
-     *
-     * @param removedItem the item to remove from the database.
-     */
-    void queueItemToRemove(PublishedItem removedItem);
-
-    /**
-     * Returns the timer used for the maintenance process of this service.
-     * 
-     * @return the timer used for the maintenance process of this service.
-     */
-    Timer getTimer();
-
-    /**
-     * Returns the timeout value for the published items maintenance task.
-     * 
-     * @return the timeout value for the published items maintenance task.
-     */
-    int getItemsTaskTimeout();
-
-    /**
-     * Sets the timeout value for the published items maintenance task.
-     *
-     * @param timeout the timeout value for the published items maintenance task.
-     */
-    void setItemsTaskTimeout(int timeout);
-
 }

src/java/org/jivesoftware/openfire/pubsub/PublishedItemTask.java

-/**
- * $RCSfile: $
- * $Revision: $
- * $Date: $
- *
- * Copyright (C) 2005-2008 Jive Software. All rights reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.jivesoftware.openfire.pubsub;
-
-import java.util.Queue;
-import java.util.TimerTask;
-
-import org.jivesoftware.openfire.pep.PEPService;
-import org.jivesoftware.util.LocaleUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * A timed maintenance task that updates the database by adding and/or
- * removing <code>PublishedItem</code>s in regular intervals.
- * 
- * @author Matt Tucker
- */
-public class PublishedItemTask extends TimerTask {
-	
-	private static final Logger Log = LoggerFactory.getLogger(PublishedItemTask.class);
-
-    /**
-     * Queue that holds the items that need to be added to the database.
-     */
-    private Queue<PublishedItem> itemsToAdd = null;
-
-    /**
-     * Queue that holds the items that need to be deleted from the database.
-     */
-    private Queue<PublishedItem> itemsToDelete = null;
-
-    /**
-     * The service to perform the published item tasks on.
-     */
-    private PubSubService service = null;
-
-    /**
-     * The number of items to save on each run of the maintenance process.
-     */
-    private int items_batch_size = 50;
-
-    public PublishedItemTask(PubSubService service) {
-        this.service = service;
-        this.itemsToAdd = service.getItemsToAdd();
-        this.itemsToDelete = service.getItemsToDelete();
-    }
-
-    @Override
-	public void run() {
-        try {
-            PublishedItem entry;
-            boolean success;
-            // Delete from the database items contained in the itemsToDelete queue
-            for (int index = 0; index <= items_batch_size && !itemsToDelete.isEmpty(); index++) {
-                entry = itemsToDelete.poll();
-                if (entry != null) {
-                    success = PubSubPersistenceManager.removePublishedItem(service, entry);
-                    if (!success) {
-                        itemsToDelete.add(entry);
-                    }
-                }
-            }
-            // Save to the database items contained in the itemsToAdd queue
-            for (int index = 0; index <= items_batch_size && !itemsToAdd.isEmpty(); index++) {
-                entry = itemsToAdd.poll();
-                if (entry != null) {
-                    success = PubSubPersistenceManager.createPublishedItem(service, entry);
-                    if (!success) {
-                        itemsToAdd.add(entry);
-                    }
-                }
-            }
-        } catch (Throwable e) {
-            Log.error(LocaleUtils.getLocalizedString("admin.error"), e);
-        }
-    }
-    
-	protected PubSubService getService() {
-		return service;
-	}
-}