ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From yzhda...@apache.org
Subject [18/47] incubator-ignite git commit: # ignite-743
Date Fri, 17 Apr 2015 16:06:40 GMT
# ignite-743


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

Branch: refs/heads/ignite-485
Commit: 23a41dfd2b7904c72a9fdebe35ef00b02ad62b3a
Parents: d6434bb
Author: sboikov <sboikov@gridgain.com>
Authored: Thu Apr 16 10:51:10 2015 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Thu Apr 16 11:03:10 2015 +0300

----------------------------------------------------------------------
 .../processors/cache/GridCacheGateway.java      |  67 +++++-----
 .../processors/cache/IgniteCacheProxy.java      | 129 +++++++------------
 .../datastreamer/DataStreamerUpdateJob.java     |  12 --
 3 files changed, 81 insertions(+), 127 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/23a41dfd/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheGateway.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheGateway.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheGateway.java
index 97fada9..aa73414 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheGateway.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheGateway.java
@@ -66,10 +66,10 @@ public class GridCacheGateway<K, V> {
     /**
      * Enter a cache call.
      *
-     * @return {@code true} if enter successful, {@code false} if the cache or the node was
stopped.
+     * @return {@code True} if enter successful, {@code false} if the cache or the node was
stopped.
      */
     public boolean enterIfNotClosed() {
-        enterIfNotClosedNoLock();
+        onEnter();
 
         // Must unlock in case of unexpected errors to avoid
         // deadlocks during kernal stop.
@@ -87,17 +87,16 @@ public class GridCacheGateway<K, V> {
     /**
      * Enter a cache call without lock.
      *
-     * @return {@code true} if enter successful, {@code false} if the cache or the node was
stopped.
+     * @return {@code True} if enter successful, {@code false} if the cache or the node was
stopped.
      */
     public boolean enterIfNotClosedNoLock() {
-        if (ctx.deploymentEnabled())
-            ctx.deploy().onEnter();
+        onEnter();
 
         return !stopped;
     }
 
     /**
-     * Leave a cache call entered by {@link #enter()} method.
+     * Leave a cache call entered by {@link #enterNoLock} method.
      */
     public void leaveNoLock() {
         ctx.tm().resetContext();
@@ -125,6 +124,22 @@ public class GridCacheGateway<K, V> {
      * @return Previous projection set on this thread.
      */
     @Nullable public GridCacheProjectionImpl<K, V> enter(@Nullable GridCacheProjectionImpl<K,
V> prj) {
+        try {
+            GridCacheAdapter<K, V> cache = ctx.cache();
+
+            GridCachePreloader<K, V> preldr = cache != null ? cache.preloader() : null;
+
+            if (preldr == null)
+                throw new IllegalStateException("Grid is in invalid state to perform this
operation. " +
+                    "It either not started yet or has already being or have stopped [gridName="
+ ctx.gridName() + ']');
+
+            preldr.startFuture().get();
+        }
+        catch (IgniteCheckedException e) {
+            throw new IgniteException("Failed to wait for cache preloader start [cacheName="
+
+                ctx.name() + "]", e);
+        }
+
         onEnter();
 
         rwLock.readLock();
@@ -132,7 +147,7 @@ public class GridCacheGateway<K, V> {
         if (stopped) {
             rwLock.readUnlock();
 
-            throw new IllegalStateException("Dynamic cache has been stopped: " + ctx.name());
+            throw new IllegalStateException("Cache has been stopped: " + ctx.name());
         }
 
         // Must unlock in case of unexpected errors to avoid
@@ -155,38 +170,12 @@ public class GridCacheGateway<K, V> {
         onEnter();
 
         if (stopped)
-            throw new IllegalStateException("Dynamic cache has been stopped: " + ctx.name());
+            throw new IllegalStateException("Cache has been stopped: " + ctx.name());
 
         return setProjectionPerCall(prj);
     }
 
     /**
-     * On enter.
-     */
-    private void onEnter() {
-        try {
-            ctx.itHolder().checkWeakQueue();
-
-            GridCacheAdapter<K, V> cache = ctx.cache();
-
-            GridCachePreloader<K, V> preldr = cache != null ? cache.preloader() : null;
-
-            if (preldr == null)
-                throw new IllegalStateException("Grid is in invalid state to perform this
operation. " +
-                    "It either not started yet or has already being or have stopped [gridName="
+ ctx.gridName() + ']');
-
-            preldr.startFuture().get();
-        }
-        catch (IgniteCheckedException e) {
-            throw new IgniteException("Failed to wait for cache preloader start [cacheName="
+
-                ctx.name() + "]", e);
-        }
-
-        if (ctx.deploymentEnabled())
-            ctx.deploy().onEnter();
-    }
-
-    /**
      * Set thread local projection per call.
      *
      * @param prj Projection to guard.
@@ -230,6 +219,16 @@ public class GridCacheGateway<K, V> {
     /**
      *
      */
+    private void onEnter() {
+        ctx.itHolder().checkWeakQueue();
+
+        if (ctx.deploymentEnabled())
+            ctx.deploy().onEnter();
+    }
+
+    /**
+     *
+     */
     public void block() {
         stopped = true;
     }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/23a41dfd/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
index d7ef8ba..c1a2d6a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
@@ -81,7 +81,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K,
V
     @GridToStringExclude
     private CacheManager cacheMgr;
 
-    /** */
+    /** If {@code false} does not acquire read lock on gateway enter. */
     @GridToStringExclude
     private boolean lock;
 
@@ -104,7 +104,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K,
V
         @Nullable GridCacheProjectionImpl<K, V> prj,
         boolean async
     ) {
-        this(ctx, delegate, prj, async, false);
+        this(ctx, delegate, prj, async, true);
     }
 
     /**
@@ -112,12 +112,14 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K,
V
      * @param delegate Delegate.
      * @param prj Projection.
      * @param async Async support flag.
+     * @param lock If {@code false} does not acquire read lock on gateway enter.
      */
-    public IgniteCacheProxy(
+    private IgniteCacheProxy(
         GridCacheContext<K, V> ctx,
         GridCacheProjectionEx<K, V> delegate,
         @Nullable GridCacheProjectionImpl<K, V> prj,
-        boolean async, boolean lock
+        boolean async,
+        boolean lock
     ) {
         super(async);
 
@@ -136,10 +138,13 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K,
V
     }
 
     /**
+     * Gets cache proxy which does not acquire read lock on gateway enter, should be
+     * used only if grid read lock is externally acquired.
+     *
      * @return Ignite cache proxy with simple gate.
      */
     public IgniteCacheProxy<K, V> cacheNoGate() {
-        return new IgniteCacheProxy<>(ctx, delegate, prj, isAsync(), true);
+        return new IgniteCacheProxy<>(ctx, delegate, prj, isAsync(), false);
     }
 
     /**
@@ -234,7 +239,11 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K,
V
         try {
             GridCacheProjectionEx<K, V> prj0 = prj != null ? prj.withExpiryPolicy(plc)
: delegate.withExpiryPolicy(plc);
 
-            return new IgniteCacheProxy<>(ctx, prj0, (GridCacheProjectionImpl<K,
V>)prj0, isAsync());
+            return new IgniteCacheProxy<>(ctx,
+                prj0,
+                (GridCacheProjectionImpl<K, V>)prj0,
+                isAsync(),
+                lock);
         }
         finally {
             onLeave(prev);
@@ -779,8 +788,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K,
V
 
             if (completionLsnr != null) {
                 fut.listen(new CI1<IgniteInternalFuture<?>>() {
-                    @Override
-                    public void apply(IgniteInternalFuture<?> fut) {
+                    @Override public void apply(IgniteInternalFuture<?> fut) {
                         try {
                             fut.get();
 
@@ -1343,7 +1351,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K,
V
     }
 
     /**
-     *
+     * @return Proxy delegate.
      */
     public GridCacheProjectionEx delegate() {
         return delegate;
@@ -1406,7 +1414,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K,
V
 
     /** {@inheritDoc} */
     @Override protected IgniteCache<K, V> createAsyncInstance() {
-        return new IgniteCacheProxy<>(ctx, delegate, prj, true);
+        return new IgniteCacheProxy<>(ctx, delegate, prj, true, lock);
     }
 
     /**
@@ -1447,7 +1455,8 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K,
V
             return new IgniteCacheProxy<>((GridCacheContext<K1, V1>)ctx,
                 prj0,
                 prj0,
-                isAsync());
+                isAsync(),
+                lock);
         }
         finally {
             onLeave(prev);
@@ -1477,7 +1486,8 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K,
V
             return new IgniteCacheProxy<>(ctx,
                 prj0,
                 prj0,
-                isAsync());
+                isAsync(),
+                lock);
         }
         finally {
             onLeave(prev);
@@ -1507,38 +1517,6 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K,
V
         return legacyProxy;
     }
 
-    /** {@inheritDoc} */
-    @Override public void writeExternal(ObjectOutput out) throws IOException {
-        out.writeObject(ctx);
-
-        out.writeObject(delegate);
-
-        out.writeObject(prj);
-
-        out.writeBoolean(lock);
-    }
-
-    /** {@inheritDoc} */
-    @SuppressWarnings({"unchecked"})
-    @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
-        ctx = (GridCacheContext<K, V>)in.readObject();
-
-        delegate = (GridCacheProjectionEx<K, V>)in.readObject();
-
-        prj = (GridCacheProjectionImpl<K, V>)in.readObject();
-
-        gate = ctx.gate();
-
-        lock = in.readBoolean();
-    }
-
-    /** {@inheritDoc} */
-    @Override public IgniteFuture<?> rebalance() {
-        ctx.preloader().forcePreload();
-
-        return new IgniteFutureImpl<>(ctx.preloader().syncFuture());
-    }
-
     /**
      * @param prj Projection to guard.
      * @return Previous projection set on this thread.
@@ -1552,6 +1530,8 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K,
V
 
     /**
      * On enter.
+     *
+     * @return {@code True} if enter successful.
      */
     private boolean onEnterIfNoClose() {
         if (lock)
@@ -1581,52 +1561,39 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K,
V
     }
 
     /** {@inheritDoc} */
-    @Override public String toString() {
-        return S.toString(IgniteCacheProxy.class, this);
-    }
+    @Override public void writeExternal(ObjectOutput out) throws IOException {
+        out.writeObject(ctx);
 
-    /**
-     * Closeable iterator.
-     */
-    private abstract static class ClIter<X, Y> extends GridCloseableIteratorAdapter<Y>
{
-        /** */
-        private X cur;
+        out.writeObject(delegate);
 
-        /** */
-        private CacheQueryFuture<X> fut;
+        out.writeObject(prj);
 
-        /**
-         * @param fut Future.
-         */
-        protected ClIter(CacheQueryFuture<X> fut) {
-            this.fut = fut;
-        }
+        out.writeBoolean(lock);
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings({"unchecked"})
+    @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
+        ctx = (GridCacheContext<K, V>)in.readObject();
 
-        /** {@inheritDoc} */
-        @Override protected Y onNext() throws IgniteCheckedException {
-            if (!onHasNext())
-                throw new NoSuchElementException();
+        delegate = (GridCacheProjectionEx<K, V>)in.readObject();
 
-            X e = cur;
+        prj = (GridCacheProjectionImpl<K, V>)in.readObject();
 
-            cur = null;
+        gate = ctx.gate();
 
-            return convert(e);
-        }
+        lock = in.readBoolean();
+    }
 
-        /**
-         * @param x X.
-         */
-        protected abstract Y convert(X x);
+    /** {@inheritDoc} */
+    @Override public IgniteFuture<?> rebalance() {
+        ctx.preloader().forcePreload();
 
-        /** {@inheritDoc} */
-        @Override protected boolean onHasNext() throws IgniteCheckedException {
-            return cur != null || (cur = fut.next()) != null;
-        }
+        return new IgniteFutureImpl<>(ctx.preloader().syncFuture());
+    }
 
-        /** {@inheritDoc} */
-        @Override protected void onClose() throws IgniteCheckedException {
-            fut.cancel();
-        }
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(IgniteCacheProxy.class, this);
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/23a41dfd/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerUpdateJob.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerUpdateJob.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerUpdateJob.java
index 52471cd..21ba3ac 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerUpdateJob.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerUpdateJob.java
@@ -88,18 +88,6 @@ class DataStreamerUpdateJob implements GridPlainCallable<Object>
{
         if (log.isDebugEnabled())
             log.debug("Running put job [nodeId=" + ctx.localNodeId() + ", size=" + col.size()
+ ']');
 
-//        TODO IGNITE-77: restore adapter usage.
-//        TODO use cacheContext.awaitStarted() instead of preloader().startFuture().get()
-//        GridCacheAdapter<Object, Object> cache = ctx.cache().internalCache(cacheName);
-//
-//        IgniteFuture<?> f = cache.context().preloader().startFuture();
-//
-//        if (!f.isDone())
-//            f.get();
-//
-//        if (ignoreDepOwnership)
-//            cache.context().deploy().ignoreOwnership(true);
-
         IgniteCacheProxy cache = ctx.cache().jcache(cacheName).cacheNoGate();
 
         cache.context().awaitStarted();


Mime
View raw message