ignite-971 Fixed offheap to swap eviction, added failover tests with …

…swap/offheap, added retries for tx 'check backup' rollback.

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheMetricsImpl.java

@@ -843,7 +843,7 @@ public void onOffHeapEvict() {
         offHeapEvicts.incrementAndGet();
 
         if (delegate != null)
-            delegate.onOffHeapRemove();
+            delegate.onOffHeapEvict();
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java

@@ -4096,21 +4096,22 @@ public void awaitLastFut() {
 
                     return t;
                 }
-                catch (IgniteInterruptedCheckedException | IgniteTxHeuristicCheckedException |
-                    IgniteTxRollbackCheckedException e) {
+                catch (IgniteInterruptedCheckedException | IgniteTxHeuristicCheckedException e) {
                     throw e;
                 }
                 catch (IgniteCheckedException e) {
-                    try {
-                        tx.rollback();
+                    if (!(e instanceof IgniteTxRollbackCheckedException)) {
+                        try {
+                            tx.rollback();
 
-                        e = new IgniteTxRollbackCheckedException("Transaction has been rolled back: " +
-                            tx.xid(), e);
-                    }
-                    catch (IgniteCheckedException | AssertionError | RuntimeException e1) {
-                        U.error(log, "Failed to rollback transaction (cache may contain stale locks): " + tx, e1);
+                            e = new IgniteTxRollbackCheckedException("Transaction has been rolled back: " +
+                                tx.xid(), e);
+                        }
+                        catch (IgniteCheckedException | AssertionError | RuntimeException e1) {
+                            U.error(log, "Failed to rollback transaction (cache may contain stale locks): " + tx, e1);
 
-                        U.addLastCause(e, e1, log);
+                            U.addLastCause(e, e1, log);
+                        }
                     }
 
                     if (X.hasCause(e, ClusterTopologyCheckedException.class) && i != retries - 1) {

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java

@@ -866,6 +866,19 @@ public Collection<GridCacheMvccCandidate> localCandidates(@Nullable GridCacheVer
      */
     public void updateTtl(@Nullable GridCacheVersion ver, long ttl);
 
+    /**
+     * Tries to do offheap -> swap eviction.
+     *
+     * @param entry Serialized swap entry.
+     * @param evictVer Version when entry was selected for eviction.
+     * @param obsoleteVer Obsolete version.
+     * @throws IgniteCheckedException If failed.
+     * @throws GridCacheEntryRemovedException If entry was removed.
+     * @return {@code True} if entry was obsoleted and written to swap.
+     */
+    public boolean offheapSwapEvict(byte[] entry, GridCacheVersion evictVer, GridCacheVersion obsoleteVer)
+        throws IgniteCheckedException, GridCacheEntryRemovedException;
+
     /**
      * @return Value.
      * @throws IgniteCheckedException If failed to read from swap storage.

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java

@@ -958,7 +958,7 @@ public void batchEvict(Collection<?> keys, @Nullable GridCacheVersion obsoleteVe
 
         List<GridCacheEntryEx> locked = new ArrayList<>(keys.size());
 
-        Set<GridCacheEntryEx> notRemove = null;
+        Set<GridCacheEntryEx> notRmv = null;
 
         Collection<GridCacheBatchSwapEntry> swapped = new ArrayList<>(keys.size());
 
@@ -990,10 +990,10 @@ public void batchEvict(Collection<?> keys, @Nullable GridCacheVersion obsoleteVe
                 locked.add(entry);
 
                 if (entry.obsolete()) {
-                    if (notRemove == null)
-                        notRemove = new HashSet<>();
+                    if (notRmv == null)
+                        notRmv = new HashSet<>();
 
-                    notRemove.add(entry);
+                    notRmv.add(entry);
 
                     continue;
                 }
@@ -1004,11 +1004,19 @@ public void batchEvict(Collection<?> keys, @Nullable GridCacheVersion obsoleteVe
                 GridCacheBatchSwapEntry swapEntry = entry.evictInBatchInternal(obsoleteVer);
 
                 if (swapEntry != null) {
+                    assert entry.obsolete() : entry;
+
                     swapped.add(swapEntry);
 
                     if (log.isDebugEnabled())
                         log.debug("Entry was evicted [entry=" + entry + ", localNode=" + cctx.nodeId() + ']');
                 }
+                else if (!entry.obsolete()) {
+                    if (notRmv == null)
+                        notRmv = new HashSet<>();
+
+                    notRmv.add(entry);
+                }
             }
 
             // Batch write to swap.
@@ -1025,7 +1033,7 @@ public void batchEvict(Collection<?> keys, @Nullable GridCacheVersion obsoleteVe
 
             // Remove entries and fire events outside the locks.
             for (GridCacheEntryEx entry : locked) {
-                if (entry.obsolete() && (notRemove == null || !notRemove.contains(entry))) {
+                if (entry.obsolete() && (notRmv == null || !notRmv.contains(entry))) {
                     entry.onMarkedObsolete();
 
                     cache.removeEntry(entry);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java

@@ -432,6 +432,41 @@ public boolean isStartVersion() {
         return info;
     }
 
+    /** {@inheritDoc} */
+    @Override public boolean offheapSwapEvict(byte[] entry, GridCacheVersion evictVer, GridCacheVersion obsoleteVer)
+        throws IgniteCheckedException, GridCacheEntryRemovedException {
+        assert cctx.swap().swapEnabled() && cctx.swap().offHeapEnabled() : this;
+
+        boolean obsolete;
+
+        synchronized (this) {
+            checkObsolete();
+
+            if (hasReaders() || !isStartVersion())
+                return false;
+
+            GridCacheMvcc mvcc = mvccExtras();
+
+            if (mvcc != null && !mvcc.isEmpty(obsoleteVer))
+                return false;
+
+            if (cctx.swap().offheapSwapEvict(key, entry, partition(), evictVer)) {
+                assert !hasValueUnlocked() : this;
+
+                obsolete = markObsolete0(obsoleteVer, false);
+
+                assert obsolete : this;
+            }
+            else
+                obsolete = false;
+        }
+
+        if (obsolete)
+            onMarkedObsolete();
+
+        return obsolete;
+    }
+
     /** {@inheritDoc} */
     @Override public CacheObject unswap() throws IgniteCheckedException, GridCacheEntryRemovedException {
         return unswap(true);
@@ -536,7 +571,7 @@ private void swap() throws IgniteCheckedException {
                     log.debug("Value did not change, skip write swap entry: " + this);
 
                 if (cctx.swap().offheapEvictionEnabled())
-                    cctx.swap().enableOffheapEviction(key());
+                    cctx.swap().enableOffheapEviction(key(), partition());
 
                 return;
             }
@@ -2988,7 +3023,7 @@ protected boolean hasValueUnlocked() {
         synchronized (this) {
             checkObsolete();
 
-            if (isNew() || (!preload && deletedUnlocked())) {
+            if ((isNew() && !cctx.swap().containsKey(key, partition())) || (!preload && deletedUnlocked())) {
                 long expTime = expireTime < 0 ? CU.toExpireTime(ttl) : expireTime;
 
                 val = cctx.kernalContext().cacheObjects().prepareForCache(val, cctx);
@@ -3643,6 +3678,9 @@ private synchronized <K, V> CacheEntryImplEx<K, V> wrapVersionedWithValue() {
         try {
             if (F.isEmptyOrNulls(filter)) {
                 synchronized (this) {
+                    if (obsoleteVersionExtras() != null)
+                        return true;
+
                     CacheObject prev = saveValueForIndexUnlocked();
 
                     if (!hasReaders() && markObsolete0(obsoleteVer, false)) {
@@ -3684,6 +3722,9 @@ private synchronized <K, V> CacheEntryImplEx<K, V> wrapVersionedWithValue() {
                         return false;
 
                     synchronized (this) {
+                        if (obsoleteVersionExtras() != null)
+                            return true;
+
                         if (!v.equals(ver))
                             // Version has changed since entry passed the filter. Do it again.
                             continue;
@@ -3768,6 +3809,13 @@ private void evictFailed(@Nullable CacheObject prevVal) throws IgniteCheckedExce
         try {
             if (!hasReaders() && markObsolete0(obsoleteVer, false)) {
                 if (!isStartVersion() && hasValueUnlocked()) {
+                    if (cctx.offheapTiered() && hasOffHeapPointer()) {
+                        if (cctx.swap().offheapEvictionEnabled())
+                            cctx.swap().enableOffheapEviction(key(), partition());
+
+                        return null;
+                    }
+
                     IgniteUuid valClsLdrId = null;
                     IgniteUuid keyClsLdrId = null;
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSwapEntryImpl.java

@@ -126,9 +126,9 @@ public static long expireTime(byte[] bytes) {
      * @return Version.
      */
     public static GridCacheVersion version(byte[] bytes) {
-        int off = VERSION_OFFSET; // Skip ttl, expire time.
+        long off = BYTE_ARR_OFF + VERSION_OFFSET; // Skip ttl, expire time.
 
-        boolean verEx = bytes[off++] != 0;
+        boolean verEx = UNSAFE.getByte(bytes, off++) != 0;
 
         return U.readVersion(bytes, off, verEx);
     }
@@ -157,26 +157,6 @@ public static GridCacheVersion version(byte[] bytes) {
         return new IgniteBiTuple<>(valBytes, type);
     }
 
-    /**
-     * @param bytes Entry bytes.
-     * @return Value bytes offset.
-     */
-    public static int valueOffset(byte[] bytes) {
-        assert bytes.length > 40 : bytes.length;
-
-        int off = VERSION_OFFSET; // Skip ttl, expire time.
-
-        boolean verEx = bytes[off++] != 0;
-
-        off += verEx ? VERSION_EX_SIZE : VERSION_SIZE;
-
-        off += 5; // Byte array flag + array size.
-
-        assert bytes.length >= off;
-
-        return off;
-    }
-
     /** {@inheritDoc} */
     @Override public byte[] valueBytes() {
         if (valBytes != null) {