ISPN-2384 Avoid losing entries with concurrent passivation/activation

core/src/main/java/org/infinispan/container/DefaultDataContainer.java

@@ -23,13 +23,18 @@
 package org.infinispan.container;
 
 import net.jcip.annotations.ThreadSafe;
+import org.infinispan.CacheException;
 import org.infinispan.container.entries.InternalCacheEntry;
 import org.infinispan.container.versioning.EntryVersion;
+import org.infinispan.eviction.ActivationManager;
 import org.infinispan.eviction.EvictionManager;
 import org.infinispan.eviction.EvictionStrategy;
 import org.infinispan.eviction.EvictionThreadPolicy;
 import org.infinispan.eviction.PassivationManager;
 import org.infinispan.factories.annotations.Inject;
+import org.infinispan.loaders.CacheLoaderException;
+import org.infinispan.loaders.CacheLoaderManager;
+import org.infinispan.loaders.CacheStore;
 import org.infinispan.util.Immutables;
 import org.infinispan.util.concurrent.BoundedConcurrentHashMap;
 import org.infinispan.util.concurrent.BoundedConcurrentHashMap.Eviction;
@@ -64,6 +69,9 @@ public class DefaultDataContainer implements DataContainer {
    final protected DefaultEvictionListener evictionListener;
    private EvictionManager evictionManager;
    private PassivationManager passivator;
+   private ActivationManager activator;
+   private CacheLoaderManager clm;
+
 
    public DefaultDataContainer(int concurrencyLevel) {
       entries = ConcurrentMapFactory.makeConcurrentMap(128, concurrencyLevel);
@@ -99,10 +107,12 @@ protected DefaultDataContainer(int concurrencyLevel, int maxEntries, EvictionStr
 
    @Inject
    public void initialize(EvictionManager evictionManager, PassivationManager passivator,
-         InternalEntryFactory entryFactory) {
+         InternalEntryFactory entryFactory, ActivationManager activator, CacheLoaderManager clm) {
       this.evictionManager = evictionManager;
       this.passivator = passivator;
       this.entryFactory = entryFactory;
+      this.activator = activator;
+      this.clm = clm;
    }
 
    public static DataContainer boundedDataContainer(int concurrencyLevel, int maxEntries,
@@ -222,6 +232,21 @@ public void onEntryChosenForEviction(InternalCacheEntry entry) {
          passivator.passivate(entry);
       }
 
+      @Override
+      public void onEntryActivated(Object key) {
+         activator.activate(key);
+      }
+
+      @Override
+      public void onEntryRemoved(Object key) {
+         try {
+            CacheStore cacheStore = clm.getCacheStore();
+            if (cacheStore != null)
+               cacheStore.remove(key);
+         } catch (CacheLoaderException e) {
+            throw new CacheException(e);
+         }
+      }
    }
 
    private static class ImmutableEntryIterator extends EntryIterator {

core/src/main/java/org/infinispan/eviction/ActivationManager.java

@@ -0,0 +1,53 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2012 Red Hat Inc. and/or its affiliates and other
+ * contributors as indicated by the @author tags. All rights reserved.
+ * See the copyright.txt in the distribution for a full listing of
+ * individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.infinispan.eviction;
+
+import org.infinispan.factories.scopes.Scope;
+import org.infinispan.factories.scopes.Scopes;
+
+/**
+ * Controls activation of cache entries that have been passivated.
+ *
+ * @author Galder Zamarreño
+ * @since 5.2
+ */
+@Scope(Scopes.NAMED_CACHE)
+public interface ActivationManager {
+
+   /**
+    * Remove key and associated value from cache store
+    * and update the activation counter.
+    *
+    * @param key Key to remove
+    */
+   void activate(Object key);
+
+   /**
+    * Get number of activations executed.
+    *
+    * @return A long representing the number of activations
+    */
+   long getActivationCount();
+
+}

core/src/main/java/org/infinispan/eviction/ActivationManagerImpl.java

@@ -0,0 +1,138 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2012 Red Hat Inc. and/or its affiliates and other
+ * contributors as indicated by the @author tags. All rights reserved.
+ * See the copyright.txt in the distribution for a full listing of
+ * individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.infinispan.eviction;
+
+import org.infinispan.config.ConfigurationException;
+import org.infinispan.configuration.cache.Configuration;
+import org.infinispan.factories.annotations.Inject;
+import org.infinispan.factories.annotations.Start;
+import org.infinispan.jmx.annotations.MBean;
+import org.infinispan.jmx.annotations.ManagedAttribute;
+import org.infinispan.jmx.annotations.ManagedOperation;
+import org.infinispan.loaders.CacheLoaderException;
+import org.infinispan.loaders.CacheLoaderManager;
+import org.infinispan.loaders.CacheStore;
+import org.infinispan.util.logging.Log;
+import org.infinispan.util.logging.LogFactory;
+import org.rhq.helpers.pluginAnnotations.agent.MeasurementType;
+import org.rhq.helpers.pluginAnnotations.agent.Metric;
+import org.rhq.helpers.pluginAnnotations.agent.Operation;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Concrete implementation of activation logic manager.
+ *
+ * @author Galder Zamarreño
+ * @since 5.2
+ */
+@MBean(objectName = "Activation",
+      description = "Component that handles activating entries that have been passivated to a CacheStore by loading them into memory.")
+public class ActivationManagerImpl implements ActivationManager {
+
+   private static final Log log = LogFactory.getLog(ActivationManagerImpl.class);
+   private static final boolean trace = log.isTraceEnabled();
+
+   private final AtomicLong activations = new AtomicLong(0);
+   private CacheLoaderManager clm;
+   private CacheStore store;
+   private Configuration cfg;
+   private boolean enabled;
+
+   @ManagedAttribute(description = "Enables or disables the gathering of statistics by this component", writable = true)
+   private boolean statisticsEnabled = false;
+
+   @Inject
+   public void inject(CacheLoaderManager clm, Configuration cfg) {
+      this.clm = clm;
+      this.cfg = cfg;
+   }
+
+   @Start(priority = 10) // Just before the passivation manager
+   public void start() {
+      enabled = clm.isUsingPassivation() && !clm.isShared();
+      if (enabled) {
+         store = clm.getCacheStore();
+         if (store == null)
+            throw new ConfigurationException(
+                  "Passivation can only be used with a CacheLoader that implements CacheStore!");
+
+         statisticsEnabled = cfg.jmxStatistics().enabled();
+      }
+   }
+
+   @Override
+   public void activate(Object key) {
+      if (enabled) {
+         try {
+            if (store.remove(key) && statisticsEnabled) {
+               activations.incrementAndGet();
+            }
+         } catch (CacheLoaderException e) {
+            log.unableToRemoveEntryAfterActivation(key, e);
+         }
+      } else {
+         if (trace)
+            log.trace("Don't remove entry from shared cache store after activation.");
+      }
+   }
+
+   @Override
+   public long getActivationCount() {
+      return activations.get();
+   }
+
+   @ManagedAttribute(description = "Number of activation events")
+   @Metric(displayName = "Number of cache entries activated",
+         measurementType = MeasurementType.TRENDSUP)
+   @SuppressWarnings("unused")
+   public String getActivations() {
+      if (!statisticsEnabled)
+         return "N/A";
+
+      return String.valueOf(getActivationCount());
+   }
+
+   @ManagedOperation(description = "Resets statistics gathered by this component")
+   @Operation(displayName = "Reset statistics")
+   public void resetStatistics() {
+      activations.set(0);
+   }
+
+   /**
+    * Disables a cache loader of a given type, where type is the fully qualified class name of a {@link org.infinispan.loaders.CacheLoader} implementation.
+    *
+    * If the given type cannot be found, this is a no-op.  If more than one cache loader of the same type is configured,
+    * all cache loaders of the given type are disabled.
+    *
+    * @param loaderType fully qualified class name of the cache loader type to disable
+    */
+   @ManagedOperation(description = "Disable all cache loaders of a given type, where type is a fully qualified class name of the cache loader to disable")
+   @Operation(displayName = "Disable all cache loaders of a given type, where type is a fully qualified class name of the cache loader to disable")
+   @SuppressWarnings("unused")
+   public void disableCacheLoader(String loaderType) {
+      if (enabled) clm.disableCacheStore(loaderType);
+   }
+
+}

core/src/main/java/org/infinispan/factories/EmptyConstructorNamedCacheFactory.java

@@ -33,6 +33,8 @@
 import org.infinispan.context.TransactionalInvocationContextContainer;
 import org.infinispan.distribution.L1Manager;
 import org.infinispan.distribution.L1ManagerImpl;
+import org.infinispan.eviction.ActivationManager;
+import org.infinispan.eviction.ActivationManagerImpl;
 import org.infinispan.eviction.EvictionManager;
 import org.infinispan.eviction.EvictionManagerImpl;
 import org.infinispan.eviction.PassivationManager;
@@ -65,7 +67,8 @@
  * @since 4.0
  */
 @DefaultFactoryFor(classes = {CacheNotifier.class, CommandsFactory.class,
-                              CacheLoaderManager.class, InvocationContextContainer.class, PassivationManager.class,
+                              CacheLoaderManager.class, InvocationContextContainer.class,
+                              PassivationManager.class, ActivationManager.class,
                               BatchContainer.class, EvictionManager.class,
                               TransactionCoordinator.class, RecoveryAdminOperations.class, StateTransferLock.class,
                               ClusteringDependentLogic.class, LockContainer.class,
@@ -97,6 +100,8 @@ public <T> T construct(Class<T> componentType) {
             return (T) new CacheLoaderManagerImpl();
          } else if (componentType.equals(PassivationManager.class)) {
             return (T) new PassivationManagerImpl();
+         } else if (componentType.equals(ActivationManager.class)) {
+            return (T) new ActivationManagerImpl();
          } else if (componentType.equals(BatchContainer.class)) {
             return (T) new BatchContainer();
          } else if (componentType.equals(TransactionCoordinator.class)) {