ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From agoncha...@apache.org
Subject incubator-ignite git commit: IGNITE-45 - Fixed getOrCreate.
Date Sun, 22 Mar 2015 19:44:31 GMT
Repository: incubator-ignite
Updated Branches:
  refs/heads/ignite-45 f4e9c473a -> 60d63a61a


IGNITE-45 - Fixed getOrCreate.


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

Branch: refs/heads/ignite-45
Commit: 60d63a61a33e7c1dbe4e1283793ba1dd11a64955
Parents: f4e9c47
Author: Alexey Goncharuk <agoncharuk@gridgain.com>
Authored: Sun Mar 22 12:44:27 2015 -0700
Committer: Alexey Goncharuk <agoncharuk@gridgain.com>
Committed: Sun Mar 22 12:44:27 2015 -0700

----------------------------------------------------------------------
 .../cache/DynamicCacheChangeRequest.java        |  17 ++
 .../GridCachePartitionExchangeManager.java      |   5 +-
 .../processors/cache/GridCacheProcessor.java    |  47 +++--
 .../GridDhtPartitionsExchangeFuture.java        |   3 +
 .../ignite/internal/util/IgniteUtils.java       |   3 -
 .../cache/IgniteDynamicCacheStartSelfTest.java  | 170 +++++++++++++++++++
 6 files changed, 224 insertions(+), 21 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60d63a61/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeRequest.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeRequest.java
index a539e1d..830078c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeRequest.java
@@ -54,6 +54,9 @@ public class DynamicCacheChangeRequest implements Serializable {
     /** Stop flag. */
     private boolean stop;
 
+    /** Fail if exists flag. */
+    private boolean failIfExists;
+
     /**
      * Constructor creates cache stop request.
      *
@@ -169,6 +172,20 @@ public class DynamicCacheChangeRequest implements Serializable {
         this.clientStartOnly = clientStartOnly;
     }
 
+    /**
+     * @return Fail if exists flag.
+     */
+    public boolean failIfExists() {
+        return failIfExists;
+    }
+
+    /**
+     * @param failIfExists Fail if exists flag.
+     */
+    public void failIfExists(boolean failIfExists) {
+        this.failIfExists = failIfExists;
+    }
+
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(DynamicCacheChangeRequest.class, this, "cacheName", cacheName());

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60d63a61/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index 201801f..e44d666 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -151,6 +151,8 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
                         for (DynamicCacheChangeRequest req : batch.requests()) {
                             if (cctx.cache().dynamicCacheRegistered(req))
                                 valid.add(req);
+                            else
+                                cctx.cache().completeStartFuture(req);
                         }
 
                         if (!F.isEmpty(valid)) {
@@ -672,9 +674,6 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
         if (fut.onAdded()) {
             exchWorker.addFuture(fut);
 
-            for (GridCacheContext cacheCtx : cctx.cacheContexts())
-                cacheCtx.preloader().onExchangeFutureAdded();
-
             return true;
         }
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60d63a61/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index 543e8c7..5bbba49 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -1248,11 +1248,17 @@ public class GridCacheProcessor extends GridProcessorAdapter {
     public boolean dynamicCacheRegistered(DynamicCacheChangeRequest req) {
         DynamicCacheDescriptor desc = registeredCaches.get(maskNull(req.cacheName()));
 
-        if (desc != null && desc.deploymentId().equals(req.deploymentId())) {
-            if (req.start())
-                return !desc.cancelled();
-            else
-                return desc.cancelled();
+        if (desc != null) {
+            if (desc.deploymentId().equals(req.deploymentId())) {
+                if (req.start())
+                    return !desc.cancelled();
+                else
+                    return desc.cancelled();
+            }
+
+            // If client requested cache start
+            if (req.initiatingNodeId() != null)
+                return true;
         }
 
         return false;
@@ -1418,19 +1424,26 @@ public class GridCacheProcessor extends GridProcessorAdapter {
                         registeredCaches.remove(masked, desc);
                 }
 
-                DynamicCacheStartFuture fut = (DynamicCacheStartFuture)pendingFuts.get(masked);
-
-                assert req.deploymentId() != null;
-                assert fut == null || fut.deploymentId != null;
-
-                if (fut != null && fut.deploymentId().equals(req.deploymentId())
&&
-                    F.eq(req.initiatingNodeId(), ctx.localNodeId()))
-                    fut.onDone();
+                completeStartFuture(req);
             }
         }
     }
 
     /**
+     * @param req Request to complete future for.
+     */
+    public void completeStartFuture(DynamicCacheChangeRequest req) {
+        DynamicCacheStartFuture fut = (DynamicCacheStartFuture)pendingFuts.get(maskNull(req.cacheName()));
+
+        assert req.deploymentId() != null;
+        assert fut == null || fut.deploymentId != null;
+
+        if (fut != null && fut.deploymentId().equals(req.deploymentId()) &&
+            F.eq(req.initiatingNodeId(), ctx.localNodeId()))
+            fut.onDone();
+    }
+
+    /**
      * Creates shared context.
      *
      * @param kernalCtx Kernal context.
@@ -1559,6 +1572,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
 
         DynamicCacheChangeRequest req = new DynamicCacheChangeRequest(cacheName, ctx.localNodeId());
 
+        req.failIfExists(failIfExists);
+
         if (ccfg != null) {
             if (desc != null && !desc.cancelled()) {
                 if (failIfExists)
@@ -1727,7 +1742,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
 
                 // Check if cache with the same name was concurrently started form different
node.
                 if (desc != null) {
-                    if (!req.clientStartOnly()) {
+                    if (!req.clientStartOnly() && req.failIfExists()) {
                         // If local node initiated start, fail the start future.
                         if (startFut != null && startFut.deploymentId().equals(req.deploymentId()))
{
                             startFut.onDone(new IgniteCacheExistsException("Failed to start
cache " +
@@ -1736,6 +1751,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
 
                         return;
                     }
+
+                    req.clientStartOnly(true);
                 }
                 else {
                     if (req.clientStartOnly()) {
@@ -1748,7 +1765,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
                     }
                 }
 
-                if (!req.clientStartOnly()) {
+                if (!req.clientStartOnly() && desc == null) {
                     DynamicCacheDescriptor startDesc = new DynamicCacheDescriptor(ccfg, req.deploymentId());
 
                     DynamicCacheDescriptor old = registeredCaches.put(maskNull(ccfg.getName()),
startDesc);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60d63a61/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
index b06f2de..8d21bc3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
@@ -434,6 +434,9 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
 
                 startCaches();
 
+                for (GridCacheContext cacheCtx : cctx.cacheContexts())
+                    cacheCtx.preloader().onExchangeFutureAdded();
+
                 assert discoEvt != null;
 
                 assert exchId.nodeId().equals(discoEvt.eventNode().id());

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60d63a61/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
index af327ce..1e5c3a8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
@@ -18,7 +18,6 @@
 package org.apache.ignite.internal.util;
 
 import org.apache.ignite.*;
-import org.apache.ignite.cache.*;
 import org.apache.ignite.cluster.*;
 import org.apache.ignite.compute.*;
 import org.apache.ignite.configuration.*;
@@ -7370,8 +7369,6 @@ public abstract class IgniteUtils {
 
         try {
             for (Class<?> c = cls != null ? cls : obj.getClass(); cls != Object.class;
cls = cls.getSuperclass()) {
-                Method[] mtds = c.getDeclaredMethods();
-
                 Method mtd = null;
 
                 for (Method declaredMtd : c.getDeclaredMethods()) {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60d63a61/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartSelfTest.java
index 0d30511..3caadb0 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartSelfTest.java
@@ -32,6 +32,7 @@ import org.apache.ignite.testframework.junits.common.*;
 import javax.cache.*;
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
 
 /**
  * Test for dynamic cache start.
@@ -745,4 +746,173 @@ public class IgniteDynamicCacheStartSelfTest extends GridCommonAbstractTest
{
             grid(0).destroyCache(DYNAMIC_CACHE_NAME);
         }
     }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testGetOrCreateMultiNode() throws Exception {
+        try {
+            final AtomicInteger cnt = new AtomicInteger();
+            final AtomicReference<Throwable> err = new AtomicReference<>();
+
+            GridTestUtils.runMultiThreaded(new Callable<Object>() {
+                @Override public Object call() throws Exception {
+                    int idx = cnt.getAndIncrement();
+
+                    try {
+                        CacheConfiguration cfg = new CacheConfiguration(DYNAMIC_CACHE_NAME);
+
+                        ignite(idx).getOrCreateCache(cfg);
+                    }
+                    catch (Exception e) {
+                        err.compareAndSet(null, e);
+                    }
+
+                    return null;
+                }
+            }, nodeCount(), "starter");
+
+            assertNull(err.get());
+
+            for (int i = 0; i < nodeCount(); i++) {
+                GridCacheContext<Object, Object> ctx = ((IgniteKernal) ignite(i)).internalCache(DYNAMIC_CACHE_NAME)
+                    .context();
+
+                assertTrue(ctx.affinityNode());
+                assertFalse(ctx.isNear());
+            }
+
+            lightCheckDynamicCache();
+        }
+        finally {
+            ignite(0).destroyCache(DYNAMIC_CACHE_NAME);
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testGetOrCreateNearOnlyMultiNode() throws Exception {
+        checkGetOrCreateNear(true);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testGetOrCreateNearMultiNode() throws Exception {
+        checkGetOrCreateNear(false);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void checkGetOrCreateNear(final boolean nearOnly) throws Exception {
+        try {
+            final AtomicInteger cnt = new AtomicInteger(nodeCount());
+            final AtomicReference<Throwable> err = new AtomicReference<>();
+
+            final int clientCnt = 2;
+
+            try {
+                testAttribute = false;
+
+                for (int i = 0; i < clientCnt; i++)
+                    startGrid(nodeCount() + i);
+
+                cnt.set(nodeCount());
+
+                final CacheConfiguration<Object, Object> cacheCfg = new CacheConfiguration<>(DYNAMIC_CACHE_NAME);
+                cacheCfg.setNodeFilter(NODE_FILTER);
+
+                if (nearOnly)
+                    ignite(0).createCache(cacheCfg);
+
+                GridTestUtils.runMultiThreaded(new Callable<Object>() {
+                    @Override public Object call() throws Exception {
+                        int idx = cnt.getAndIncrement();
+
+                        try {
+                            if (nearOnly)
+                                ignite(idx).getOrCreateCache(DYNAMIC_CACHE_NAME, new NearCacheConfiguration<>());
+                            else
+                                ignite(idx).getOrCreateCache(cacheCfg, new NearCacheConfiguration<>());
+                        }
+                        catch (Exception ex) {
+                            err.compareAndSet(null, ex);
+                        }
+
+                        return null;
+                    }
+                }, clientCnt, "starter");
+
+                assertNull(err.get());
+
+                for (int i = 0; i < nodeCount(); i++) {
+                    GridCacheContext<Object, Object> ctx = ((IgniteKernal) ignite(i)).internalCache(DYNAMIC_CACHE_NAME)
+                        .context();
+
+                    assertTrue(ctx.affinityNode());
+                    assertFalse(ctx.isNear());
+                }
+
+                for (int i = 0; i < clientCnt; i++) {
+                    GridCacheContext<Object, Object> ctx = ((IgniteKernal) ignite(nodeCount()
+ i))
+                        .internalCache(DYNAMIC_CACHE_NAME).context();
+
+                    assertFalse(ctx.affinityNode());
+                    assertTrue("Cache is not near for index: " + (nodeCount() + i), ctx.isNear());
+                }
+
+                lightCheckDynamicCache();
+            }
+            finally {
+                for (int i = 0; i < clientCnt; i++)
+                    stopGrid(nodeCount() + i);
+            }
+        }
+        finally {
+            ignite(0).destroyCache(DYNAMIC_CACHE_NAME);
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    private void lightCheckDynamicCache() throws Exception {
+        int nodes = F.size(G.allGrids());
+
+        for (int i = 0; i < nodes; i++) {
+            IgniteCache<Object, Object> jcache = ignite(i).jcache(DYNAMIC_CACHE_NAME);
+
+            for (int k = 0; k < 20; k++) {
+                int key = i + k * nodes;
+
+                jcache.put(key, key);
+            }
+        }
+
+        for (int i = 0; i < nodes; i++) {
+            IgniteCache<Object, Object> jcache = ignite(i).jcache(DYNAMIC_CACHE_NAME);
+
+            for (int k = 0; k < 20 * nodes; k++)
+                assertEquals(k, jcache.get(k));
+        }
+
+        for (int i = 0; i < nodes; i++) {
+            IgniteCache<Object, Object> jcache = ignite(i).jcache(DYNAMIC_CACHE_NAME);
+
+            for (int k = 0; k < 20; k++) {
+                int key = i + k * nodes;
+
+                assertEquals(key, jcache.getAndRemove(key));
+            }
+        }
+
+        for (int i = 0; i < nodes; i++) {
+            IgniteCache<Object, Object> jcache = ignite(i).jcache(DYNAMIC_CACHE_NAME);
+
+            for (int k = 0; k < 20 * nodes; k++)
+                assertNull(jcache.get(k));
+        }
+    }
 }


Mime
View raw message