ISPN-1983 Improve efficiency of subsequent putForExternalRead calls

* Avoid creating unnecessary objects when PFER does not return previous
value.
* Make sure cache mgmt interceptor updates store count only when
operation is succesful, so when PFER or putIfAbsent do an actual put

core/src/main/java/org/infinispan/commands/write/PutKeyValueCommand.java

@@ -81,6 +81,11 @@ public Object acceptVisitor(InvocationContext ctx, Visitor visitor) throws Throw
    public Object perform(InvocationContext ctx) throws Throwable {
       Object o;
       MVCCEntry e = (MVCCEntry) ctx.lookupEntry(key);
+      if (e == null && ctx.hasFlag(Flag.PUT_FOR_EXTERNAL_READ)) {
+         successful = false;
+         return null;
+      }
+
       Object entryValue = e.getValue();
       if (entryValue != null && putIfAbsent && !e.isRemoved()) {
          successful = false;

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

@@ -35,6 +35,7 @@
 import org.infinispan.container.entries.ReadCommittedEntry;
 import org.infinispan.container.entries.RepeatableReadEntry;
 import org.infinispan.container.versioning.EntryVersion;
+import org.infinispan.context.Flag;
 import org.infinispan.context.InvocationContext;
 import org.infinispan.factories.annotations.Inject;
 import org.infinispan.factories.annotations.Start;
@@ -144,6 +145,10 @@ public final MVCCEntry wrapEntryForPut(InvocationContext ctx, Object key, Intern
          mvccEntry.undelete(undeleteIfNeeded);
       } else {
          InternalCacheEntry ice = (icEntry == null ? getFromContainer(key) : icEntry);
+         // A putForExternalRead is putIfAbsent, so if key present, do nothing
+         if (ice != null && ctx.hasFlag(Flag.PUT_FOR_EXTERNAL_READ))
+            return null;
+
          mvccEntry = ice != null ?
              wrapInternalCacheEntryForPut(ctx, key, ice) :
              newMvccEntryForPut(ctx, key);

core/src/main/java/org/infinispan/interceptors/CacheMgmtInterceptor.java

@@ -76,6 +76,7 @@ protected Log getLog() {
    }
 
    @Inject
+   @SuppressWarnings("unused")
    public void setDependencies(DataContainer dataContainer) {
       this.dataContainer = dataContainer;
    }
@@ -122,10 +123,12 @@ public Object visitPutMapCommand(InvocationContext ctx, PutMapCommand command) t
    public Object visitPutKeyValueCommand(InvocationContext ctx, PutKeyValueCommand command) throws Throwable {
       long t1 = System.nanoTime();
       Object retval = invokeNextInterceptor(ctx, command);
-      long t2 = System.nanoTime();
-      long intervalMilliseconds = nanosecondsIntervalToMilliseconds(t1, t2);
-      storeTimes.getAndAdd(intervalMilliseconds);
-      stores.incrementAndGet();
+      if (command.isSuccessful()) {
+         long t2 = System.nanoTime();
+         long intervalMilliseconds = nanosecondsIntervalToMilliseconds(t1, t2);
+         storeTimes.getAndAdd(intervalMilliseconds);
+         stores.incrementAndGet();
+      }
       return retval;
    }
 
@@ -178,6 +181,7 @@ public long getEvictions() {
 
    @ManagedAttribute(description = "Percentage hit/(hit+miss) ratio for the cache")
    @Metric(displayName = "Hit ratio", units = Units.PERCENTAGE, displayType = DisplayType.SUMMARY)
+   @SuppressWarnings("unused")
    public double getHitRatio() {
       long hitsL = hits.get();
       double total = hitsL + misses.get();
@@ -190,6 +194,7 @@ public double getHitRatio() {
 
    @ManagedAttribute(description = "read/writes ratio for the cache")
    @Metric(displayName = "Read/write ratio", units = Units.PERCENTAGE, displayType = DisplayType.SUMMARY)
+   @SuppressWarnings("unused")
    public double getReadWriteRatio() {
       if (stores.get() == 0)
          return 0;
@@ -198,6 +203,7 @@ public double getReadWriteRatio() {
 
    @ManagedAttribute(description = "Average number of milliseconds for a read operation on the cache")
    @Metric(displayName = "Average read time", units = Units.MILLISECONDS, displayType = DisplayType.SUMMARY)
+   @SuppressWarnings("unused")
    public long getAverageReadTime() {
       long total = hits.get() + misses.get();
       if (total == 0)
@@ -207,6 +213,7 @@ public long getAverageReadTime() {
 
    @ManagedAttribute(description = "Average number of milliseconds for a write operation in the cache")
    @Metric(displayName = "Average write time", units = Units.MILLISECONDS, displayType = DisplayType.SUMMARY)
+   @SuppressWarnings("unused")
    public long getAverageWriteTime() {
       if (stores.get() == 0)
          return 0;
@@ -227,6 +234,7 @@ public long getElapsedTime() {
 
    @ManagedAttribute(description = "Number of seconds since the cache statistics were last reset")
    @Metric(displayName = "Seconds since cache statistics were reset", units = Units.SECONDS, displayType = DisplayType.SUMMARY)
+   @SuppressWarnings("unused")
    public long getTimeSinceReset() {
       return TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - resetNanoseconds.get());
    }
@@ -248,8 +256,8 @@ public void resetStatistics() {
    }
 
    /**
-    * @param nanoStart
-    * @param nanoEnd
+    * @param nanoStart start time
+    * @param nanoEnd end time
     * @return the interval rounded in milliseconds
     */
    private long nanosecondsIntervalToMilliseconds(final long nanoStart, final long nanoEnd) {

core/src/test/java/org/infinispan/api/mvcc/PutForExternalReadTest.java

@@ -323,6 +323,13 @@ public boolean isSatisfied() throws Exception {
       });
    }
 
+   public void testMultipleIdenticalPutForExternalReadCalls() {
+      Cache cache1 = cache(0, "replSync");
+      cache1.putForExternalRead(key, value);
+      cache1.putForExternalRead(key, value2);
+      assertEquals(value, cache1.get(key));
+   }
+
    /**
     * Tests that setting a cacheModeLocal=true flag prevents propagation of the putForExternalRead().
     *

core/src/test/java/org/infinispan/jmx/CacheMgmtInterceptorMBeanTest.java

@@ -134,6 +134,22 @@ public void testStores() throws Exception {
       assertCurrentNumberOfEntries(4);
    }
 
+   public void testStoresPutForExternalRead() throws Exception {
+      assertStores(0);
+      cache.putForExternalRead("key", "value");
+      assertStores(1);
+      cache.putForExternalRead("key", "value");
+      assertStores(1);
+   }
+
+   public void testStoresPutIfAbsent() throws Exception {
+      assertStores(0);
+      cache.putIfAbsent("voooo", "doooo");
+      assertStores(1);
+      cache.putIfAbsent("voooo", "no-doooo");
+      assertStores(1);
+   }
+
    public void testRemoves() throws Exception {
       assertStores(0);
       assertRemoveHits(0);