ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From vkuliche...@apache.org
Subject [24/50] incubator-ignite git commit: ignite-341 - fixed
Date Tue, 31 Mar 2015 06:41:11 GMT
ignite-341 - fixed


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/5066aead
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/5066aead
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/5066aead

Branch: refs/heads/master
Commit: 5066aeadbe942e66f59869370ba28143229a143a
Parents: d4236d0
Author: S.Vladykin <svladykin@gridgain.com>
Authored: Thu Mar 26 19:06:06 2015 +0300
Committer: S.Vladykin <svladykin@gridgain.com>
Committed: Thu Mar 26 19:06:06 2015 +0300

----------------------------------------------------------------------
 .../query/h2/opt/GridH2AbstractKeyValueRow.java | 97 +++++++++++---------
 .../query/h2/opt/GridH2KeyValueRowOffheap.java  | 57 +++++++-----
 .../processors/query/h2/opt/GridH2Table.java    |  4 +-
 3 files changed, 89 insertions(+), 69 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5066aead/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2AbstractKeyValueRow.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2AbstractKeyValueRow.java
b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2AbstractKeyValueRow.java
index 2770a10..4a0809a 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2AbstractKeyValueRow.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2AbstractKeyValueRow.java
@@ -92,6 +92,8 @@ public abstract class GridH2AbstractKeyValueRow extends GridH2Row {
      * @throws IgniteSpiException If failed.
      */
     public static Value wrap(Object obj, int type) throws IgniteSpiException {
+        assert obj != null;
+
         switch (type) {
             case Value.BOOLEAN:
                 return ValueBoolean.get((Boolean)obj);
@@ -169,61 +171,74 @@ public abstract class GridH2AbstractKeyValueRow extends GridH2Row {
     }
 
     /**
-     * @param val Value.
-     * @throws IgniteCheckedException
-     */
-    public synchronized void unswapBeforeRemove(Object val) throws IgniteCheckedException
{
-        assert val != null;
-
-        Value oldVal = super.getValue(VAL_COL);
-
-        if (oldVal == null || oldVal instanceof WeakValue)
-            onUnswap(val);
-        // Else we would assert that val.equals(oldVal.getObject()) but value is not necessarily
implements equals() correctly.
-    }
-
-    /**
      * Should be called when entry getting unswapped.
      *
      * @param val Value.
+     * @param beforeRmv If this is unswap before remove.
      * @throws IgniteCheckedException If failed.
      */
-    public synchronized void onUnswap(Object val) throws IgniteCheckedException {
+    public synchronized void onUnswap(Object val, boolean beforeRmv) throws IgniteCheckedException
{
         setValue(VAL_COL, wrap(val, desc.valueType()));
+
+        notifyAll();
     }
 
     /**
      * Atomically updates weak value.
      *
-     * @param exp Expected value.
      * @param upd New value.
-     * @return Expected value if update succeeded, unexpected value otherwise.
+     * @return {@code null} If update succeeded, unexpected value otherwise.
      */
-    protected synchronized Value updateWeakValue(Value exp, Value upd) {
-        Value res = super.getValue(VAL_COL);
+    protected synchronized Value updateWeakValue(Value upd) {
+        Value res = peekValue(VAL_COL);
 
-        if (res != exp && !(res instanceof WeakValue))
+        if (res != null && !(res instanceof WeakValue))
             return res;
 
         setValue(VAL_COL, new WeakValue(upd));
 
-        return exp;
+        notifyAll();
+
+        return null;
     }
 
     /**
+     * @param attempt Attempt.
      * @return Synchronized value.
      */
-    protected synchronized Value syncValue() {
-        return super.getValue(VAL_COL);
+    protected synchronized Value syncValue(int attempt) {
+        Value v = peekValue(VAL_COL);
+
+        if (v == null && attempt != 0) {
+            try {
+                wait(attempt);
+            }
+            catch (InterruptedException e) {
+                throw new IgniteInterruptedException(e);
+            }
+
+            v = peekValue(VAL_COL);
+        }
+
+        return v;
+    }
+
+    /**
+     * @param col Column index.
+     * @return Value if exists.
+     */
+    protected final Value peekValue(int col) {
+        return getValueList()[col];
     }
 
     /** {@inheritDoc} */
     @Override public Value getValue(int col) {
         if (col < DEFAULT_COLUMNS_COUNT) {
-            Value v = super.getValue(col);
+            Value v = peekValue(col);
 
             if (col == VAL_COL) {
-                int loops = 0;
+                long start = 0;
+                int attempt = 0;
 
                 while ((v = WeakValue.unwrap(v)) == null) {
                     v = getOffheapValue(VAL_COL);
@@ -231,7 +246,7 @@ public abstract class GridH2AbstractKeyValueRow extends GridH2Row {
                     if (v != null) {
                         setValue(VAL_COL, v);
 
-                        if (super.getValue(KEY_COL) == null)
+                        if (peekValue(KEY_COL) == null)
                             cache();
 
                         return v;
@@ -245,28 +260,26 @@ public abstract class GridH2AbstractKeyValueRow extends GridH2Row {
                         if (valObj != null) {
                             Value upd = wrap(valObj, desc.valueType());
 
-                            Value res = updateWeakValue(null, upd);
-
-                            if (res == null) {
-                                if (super.getValue(KEY_COL) == null)
-                                    cache();
-
-                                return upd;
-                            }
+                            v = updateWeakValue(upd);
 
-                            v = res;
+                            return v == null ? upd : v;
                         }
                         else {
                             // If nothing found in swap then we should be already unswapped.
-                            v = syncValue();
+                            v = syncValue(attempt);
                         }
-
-                        if (++loops == 1000_000)
-                            throw new IllegalStateException("Failed to get value for key:
" + k);
                     }
                     catch (IgniteCheckedException e) {
                         throw new IgniteException(e);
                     }
+
+                    attempt++;
+
+                    if (start == 0)
+                        start = U.currentTimeMillis();
+                    else if (U.currentTimeMillis() - start > 15_000) // Loop for at most
15 seconds.
+                        throw new IgniteException("Failed to get value for key: " + k +
+                            ". This can happen due to a long GC pause.");
                 }
             }
 
@@ -279,7 +292,7 @@ public abstract class GridH2AbstractKeyValueRow extends GridH2Row {
 
                 setValue(KEY_COL, v);
 
-                if (super.getValue(VAL_COL) == null)
+                if (peekValue(VAL_COL) == null)
                     cache();
             }
 
@@ -337,10 +350,10 @@ public abstract class GridH2AbstractKeyValueRow extends GridH2Row {
 
         addOffheapRowId(sb);
 
-        Value v = super.getValue(KEY_COL);
+        Value v = peekValue(KEY_COL);
         sb.a("[ key: ").a(v == null ? "nil" : v.getString());
 
-        v = WeakValue.unwrap(super.getValue(VAL_COL));
+        v = WeakValue.unwrap(peekValue(VAL_COL));
         sb.a(", val: ").a(v == null ? "nil" : v.getString());
 
         sb.a(" ][ ");

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5066aead/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2KeyValueRowOffheap.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2KeyValueRowOffheap.java
b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2KeyValueRowOffheap.java
index 9c1d90a..9c2c1b2 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2KeyValueRowOffheap.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2KeyValueRowOffheap.java
@@ -120,6 +120,7 @@ public class GridH2KeyValueRowOffheap extends GridH2AbstractKeyValueRow
{
      */
     @SuppressWarnings("LockAcquiredButNotSafelyReleased")
     private static Lock lock(long ptr) {
+        assert ptr > 0 : ptr;
         assert (ptr & 7) == 0 : ptr; // Unsafe allocated pointers aligned.
 
         Lock l = lock.getLock(ptr >>> 3);
@@ -198,13 +199,14 @@ public class GridH2KeyValueRowOffheap extends GridH2AbstractKeyValueRow
{
 
             final long valPtr = mem.readLongVolatile(p);
 
-            if (valPtr == 0)
-                return; // Nothing to swap.
+            if (valPtr <= 0)
+                throw new IllegalStateException("Already swapped: " + ptr);
+
+            if (!mem.casLong(p, valPtr, 0))
+                throw new IllegalStateException("Concurrent unswap: " + ptr);
 
             desc.guard().finalizeLater(new Runnable() {
                 @Override public void run() {
-                    mem.casLong(p, valPtr, 0); // If it was unswapped concurrently we will
not update.
-
                     mem.release(valPtr, mem.readInt(valPtr) + OFFSET_VALUE);
                 }
             });
@@ -215,37 +217,40 @@ public class GridH2KeyValueRowOffheap extends GridH2AbstractKeyValueRow
{
     }
 
     /** {@inheritDoc} */
-    @Override public synchronized void unswapBeforeRemove(Object val) throws IgniteCheckedException
{
-        assert val != null;
-
-        onUnswap(val);
-    }
-
-    /** {@inheritDoc} */
     @SuppressWarnings("NonSynchronizedMethodOverridesSynchronizedMethod")
-    @Override protected Value updateWeakValue(Value exp, Value upd) {
+    @Override protected synchronized Value updateWeakValue(Value upd) {
         setValue(VAL_COL, upd);
 
-        return exp;
+        notifyAll();
+
+        return null;
     }
 
     /** {@inheritDoc} */
-    @Override public synchronized void onUnswap(Object val) throws IgniteCheckedException
{
-        super.onUnswap(val);
+    @Override public synchronized void onUnswap(Object val, boolean beforeRmv) throws IgniteCheckedException
{
+        assert val != null;
 
-        long p = ptr;
-
-        assert p > 0 : p;
+        final long p = ptr;
 
         Lock l = lock(p);
 
         try {
             GridUnsafeMemory mem = desc.memory();
 
-            if (mem.readLongVolatile(p + OFFSET_VALUE_REF) != 0)
-                return; // The offheap value is in its place, nothing to do here.
+            if (mem.readLongVolatile(p + OFFSET_VALUE_REF) != 0) {
+                if (beforeRmv)
+                    return; // The offheap value is in its place, nothing to do here.
+                else
+                    throw new IllegalStateException("Unswap without swap: " + p);
+            }
 
-            Value v = getValue(VAL_COL); // We just set the value above, so it will be returned
right away.
+            Value v = peekValue(VAL_COL);
+
+            if (v == null) {
+                setValue(VAL_COL, wrap(val, desc.valueType()));
+
+                v = peekValue(VAL_COL);
+            }
 
             byte[] bytes = new byte[SIZE_CALCULATOR.getValueLen(v)];
 
@@ -263,11 +268,13 @@ public class GridH2KeyValueRowOffheap extends GridH2AbstractKeyValueRow
{
         finally {
             l.unlock();
         }
+
+        notifyAll();
     }
 
     /** {@inheritDoc} */
-    @Override protected synchronized Value syncValue() {
-        Value v = super.syncValue();
+    @Override protected Value syncValue(int attempt) {
+        Value v = super.syncValue(attempt);
 
         if (v != null)
             return v;
@@ -283,8 +290,8 @@ public class GridH2KeyValueRowOffheap extends GridH2AbstractKeyValueRow
{
         GridUnsafeMemory mem = desc.memory();
 
         if (p == 0) { // Serialize data to offheap memory.
-            Value key = getValue(KEY_COL);
-            Value val = getValue(VAL_COL);
+            Value key = peekValue(KEY_COL);
+            Value val = peekValue(VAL_COL);
 
             assert key != null;
             assert val != null;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5066aead/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Table.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Table.java
b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Table.java
index e686c5e..62b3b5e 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Table.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Table.java
@@ -157,7 +157,7 @@ public class GridH2Table extends TableBase {
             if (val == null)
                 row.onSwap();
             else
-                row.onUnswap(val);
+                row.onUnswap(val, false);
 
             return true;
         }
@@ -382,7 +382,7 @@ public class GridH2Table extends TableBase {
                     Value v = row.getValue(GridH2AbstractKeyValueRow.VAL_COL);
 
                     if (v != null)
-                        ((GridH2AbstractKeyValueRow)old).unswapBeforeRemove(v.getObject());
+                        ((GridH2AbstractKeyValueRow)old).onUnswap(v.getObject(), true);
                 }
 
                 if (old != null) {


Mime
View raw message