ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From akuznet...@apache.org
Subject [07/50] [abbrv] incubator-ignite git commit: #ignite-743: Implement IgniteCacheProxyLockFree.
Date Fri, 17 Apr 2015 14:36:11 GMT
#ignite-743:  Implement IgniteCacheProxyLockFree.


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

Branch: refs/heads/ignite-755
Commit: 56ef2695a429709388ffb13288524993f4f28d26
Parents: 196337a
Author: ivasilinets <ivasilinets@gridgain.com>
Authored: Tue Apr 14 16:17:24 2015 +0300
Committer: ivasilinets <ivasilinets@gridgain.com>
Committed: Tue Apr 14 16:17:24 2015 +0300

----------------------------------------------------------------------
 .../processors/cache/IgniteCacheProxy.java      |  995 +++-------------
 .../cache/IgniteCacheProxyLockFree.java         | 1127 +++++++++++++++++-
 .../datastreamer/DataStreamerUpdateJob.java     |    6 +-
 3 files changed, 1304 insertions(+), 824 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/56ef2695/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 44a5dbe..dcdf0be 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
@@ -22,18 +22,11 @@ import org.apache.ignite.cache.CacheManager;
 import org.apache.ignite.cache.*;
 import org.apache.ignite.cache.query.*;
 import org.apache.ignite.cluster.*;
-import org.apache.ignite.configuration.*;
 import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.processors.cache.query.*;
-import org.apache.ignite.internal.processors.query.*;
-import org.apache.ignite.internal.util.*;
-import org.apache.ignite.internal.util.future.*;
 import org.apache.ignite.internal.util.tostring.*;
-import org.apache.ignite.internal.util.typedef.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 import org.apache.ignite.lang.*;
 import org.apache.ignite.mxbean.*;
-import org.apache.ignite.spi.discovery.tcp.internal.*;
 import org.jetbrains.annotations.*;
 
 import javax.cache.*;
@@ -53,33 +46,15 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
     /** */
     private static final long serialVersionUID = 0L;
 
-    /** */
-    private static final IgniteBiPredicate ACCEPT_ALL = new IgniteBiPredicate() {
-        @Override public boolean apply(Object k, Object v) {
-            return true;
-        }
-    };
-
-    /** Context. */
-    private GridCacheContext<K, V> ctx;
-
-    /** Gateway. */
-    private GridCacheGateway<K, V> gate;
-
     /** Delegate. */
     @GridToStringInclude
-    private GridCacheProjectionEx<K, V> delegate;
+    private IgniteCacheProxyLockFree<K, V> delegate;
 
     /** Projection. */
     private GridCacheProjectionImpl<K, V> prj;
 
-    /** */
-    @GridToStringExclude
-    private GridCacheProxyImpl<K, V> legacyProxy;
-
-    /** */
-    @GridToStringExclude
-    private CacheManager cacheMgr;
+    /** Gateway. */
+    private GridCacheGateway<K, V> gate;
 
     /**
      * Empty constructor required for {@link Externalizable}.
@@ -103,22 +78,26 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         super(async);
 
         assert ctx != null;
-        assert delegate != null;
-
-        this.ctx = ctx;
-        this.delegate = delegate;
-        this.prj = prj;
 
         gate = ctx.gate();
 
-        legacyProxy = new GridCacheProxyImpl<>(ctx, delegate, prj);
+        this.prj = prj;
+
+        this.delegate = new IgniteCacheProxyLockFree<>(ctx, delegate, prj, async);
     }
 
     /**
      * @return Context.
      */
     public GridCacheContext<K, V> context() {
-        return ctx;
+        return delegate.context();
+    }
+
+    /**
+     * @return Lock free instance.
+     */
+    public IgniteCacheProxyLockFree<K,V> lockFree() {
+        return delegate;
     }
 
     /**
@@ -133,7 +112,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            return ctx.cache().metrics();
+            return delegate.metrics();
         }
         finally {
             gate.leave(prev);
@@ -145,20 +124,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            List<CacheMetrics> metrics = new ArrayList<>(grp.nodes().size());
-
-            for (ClusterNode node : grp.nodes()) {
-                Map<Integer, CacheMetrics> nodeCacheMetrics = ((TcpDiscoveryNode)node).cacheMetrics();
-
-                if (nodeCacheMetrics != null) {
-                    CacheMetrics e = nodeCacheMetrics.get(context().cacheId());
-
-                    if (e != null)
-                        metrics.add(e);
-                }
-            }
-
-            return new CacheMetricsSnapshot(ctx.cache().metrics(), metrics);
+            return delegate.metrics(grp);
         }
         finally {
             gate.leave(prev);
@@ -170,7 +136,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            return ctx.cache().mxBean();
+            return delegate.mxBean();
         }
         finally {
             gate.leave(prev);
@@ -179,12 +145,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
 
     /** {@inheritDoc} */
     @Override public <C extends Configuration<K, V>> C getConfiguration(Class<C> clazz) {
-        CacheConfiguration cfg = ctx.config();
-
-        if (!clazz.isAssignableFrom(cfg.getClass()))
-            throw new IllegalArgumentException();
-
-        return clazz.cast(cfg);
+        return delegate.getConfiguration(clazz);
     }
 
     /** {@inheritDoc} */
@@ -192,7 +153,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            return ctx.cache().randomEntry();
+            return delegate.randomEntry();
         }
         finally {
             gate.leave(prev);
@@ -204,9 +165,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            GridCacheProjectionEx<K, V> prj0 = prj != null ? prj.withExpiryPolicy(plc) : delegate.withExpiryPolicy(plc);
-
-            return new IgniteCacheProxy<>(ctx, prj0, (GridCacheProjectionImpl<K, V>)prj0, isAsync());
+            return delegate.withExpiryPolicy(plc);
         }
         finally {
             gate.leave(prev);
@@ -220,292 +179,68 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
 
     /** {@inheritDoc} */
     @Override public void loadCache(@Nullable IgniteBiPredicate<K, V> p, @Nullable Object... args) {
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
+
         try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync())
-                    setFuture(ctx.cache().globalLoadCacheAsync(p, args));
-                else
-                    ctx.cache().globalLoadCache(p, args);
-            }
-            finally {
-                gate.leave(prev);
-            }
+            delegate.loadCache(p, args);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Override public void localLoadCache(@Nullable IgniteBiPredicate<K, V> p, @Nullable Object... args) {
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
+
         try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync())
-                    setFuture(delegate.localLoadCacheAsync(p, args));
-                else
-                    delegate.localLoadCache(p, args);
-            }
-            finally {
-                gate.leave(prev);
-            }
+            delegate.localLoadCache(p, args);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Nullable @Override public V getAndPutIfAbsent(K key, V val) throws CacheException {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    setFuture(delegate.getAndPutIfAbsentAsync(key, val));
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return null;
-                }
-                else
-                    return delegate.getAndPutIfAbsent(key, val);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.getAndPutIfAbsent(key, val);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Override public Lock lock(K key) throws CacheException {
-        return lockAll(Collections.singleton(key));
+        return delegate.lock(key);
     }
 
     /** {@inheritDoc} */
     @Override public Lock lockAll(final Collection<? extends K> keys) {
-        return new CacheLockImpl<>(gate, delegate, prj, keys);
+        return delegate.lockAll(keys);
     }
 
     /** {@inheritDoc} */
     @Override public boolean isLocalLocked(K key, boolean byCurrThread) {
-        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-        try {
-            return byCurrThread ? delegate.isLockedByThread(key) : delegate.isLocked(key);
-        }
-        finally {
-            gate.leave(prev);
-        }
-    }
-
-    /**
-     * @param filter Filter.
-     * @param grp Optional cluster group.
-     * @return Cursor.
-     */
-    @SuppressWarnings("unchecked")
-    private QueryCursor<Entry<K,V>> query(Query filter, @Nullable ClusterGroup grp) {
-        final CacheQuery<Map.Entry<K,V>> qry;
-        final CacheQueryFuture<Map.Entry<K,V>> fut;
-
-        if (filter instanceof ScanQuery) {
-            IgniteBiPredicate<K, V> p = ((ScanQuery)filter).getFilter();
-
-            qry = delegate.queries().createScanQuery(p != null ? p : ACCEPT_ALL);
-
-            if (grp != null)
-                qry.projection(grp);
-
-            fut = qry.execute();
-        }
-        else if (filter instanceof TextQuery) {
-            TextQuery p = (TextQuery)filter;
-
-            qry = delegate.queries().createFullTextQuery(p.getType(), p.getText());
-
-            if (grp != null)
-                qry.projection(grp);
-
-            fut = qry.execute();
-        }
-        else if (filter instanceof SpiQuery) {
-            qry = ((GridCacheQueriesEx)delegate.queries()).createSpiQuery();
-
-            if (grp != null)
-                qry.projection(grp);
-
-            fut = qry.execute(((SpiQuery)filter).getArgs());
-        }
-        else {
-            if (filter instanceof SqlFieldsQuery)
-                throw new CacheException("Use methods 'queryFields' and 'localQueryFields' for " +
-                    SqlFieldsQuery.class.getSimpleName() + ".");
-
-            throw new CacheException("Unsupported query type: " + filter);
-        }
-
-        return new QueryCursorImpl<>(new GridCloseableIteratorAdapter<Entry<K,V>>() {
-            /** */
-            private Map.Entry<K,V> cur;
-
-            @Override protected Entry<K,V> onNext() throws IgniteCheckedException {
-                if (!onHasNext())
-                    throw new NoSuchElementException();
-
-                Map.Entry<K,V> e = cur;
-
-                cur = null;
-
-                return new CacheEntryImpl<>(e.getKey(), e.getValue());
-            }
-
-            @Override protected boolean onHasNext() throws IgniteCheckedException {
-                return cur != null || (cur = fut.next()) != null;
-            }
-
-            @Override protected void onClose() throws IgniteCheckedException {
-                fut.cancel();
-            }
-        });
-    }
-
-    /**
-     * @param local Enforce local.
-     * @return Local node cluster group.
-     */
-    private ClusterGroup projection(boolean local) {
-        if (local || ctx.isLocal() || isReplicatedDataNode())
-            return ctx.kernalContext().grid().cluster().forLocal();
-
-        if (ctx.isReplicated())
-            return ctx.kernalContext().grid().cluster().forDataNodes(ctx.name()).forRandom();
-
-        return null;
-    }
-
-    /**
-     * Executes continuous query.
-     *
-     * @param qry Query.
-     * @param loc Local flag.
-     * @return Initial iteration cursor.
-     */
-    @SuppressWarnings("unchecked")
-    private QueryCursor<Entry<K, V>> queryContinuous(ContinuousQuery qry, boolean loc) {
-        if (qry.getInitialQuery() instanceof ContinuousQuery)
-            throw new IgniteException("Initial predicate for continuous query can't be an instance of another " +
-                "continuous query. Use SCAN or SQL query for initial iteration.");
-
-        if (qry.getLocalListener() == null)
-            throw new IgniteException("Mandatory local listener is not set for the query: " + qry);
-
-        try {
-            final UUID routineId = ctx.continuousQueries().executeQuery(
-                qry.getLocalListener(),
-                qry.getRemoteFilter(),
-                qry.getPageSize(),
-                qry.getTimeInterval(),
-                qry.isAutoUnsubscribe(),
-                loc ? ctx.grid().cluster().forLocal() : null);
-
-            final QueryCursor<Cache.Entry<K, V>> cur =
-                qry.getInitialQuery() != null ? query(qry.getInitialQuery()) : null;
-
-            return new QueryCursor<Cache.Entry<K, V>>() {
-                @Override public Iterator<Cache.Entry<K, V>> iterator() {
-                    return cur != null ? cur.iterator() : new GridEmptyIterator<Cache.Entry<K, V>>();
-                }
-
-                @Override public List<Cache.Entry<K, V>> getAll() {
-                    return cur != null ? cur.getAll() : Collections.<Cache.Entry<K, V>>emptyList();
-                }
-
-                @Override public void close() {
-                    if (cur != null)
-                        cur.close();
-
-                    try {
-                        ctx.kernalContext().continuous().stopRoutine(routineId).get();
-                    }
-                    catch (IgniteCheckedException e) {
-                        throw U.convertException(e);
-                    }
-                }
-            };
-        }
-        catch (IgniteCheckedException e) {
-            throw U.convertException(e);
-        }
+        return delegate.isLocalLocked(key, byCurrThread);
     }
 
     /** {@inheritDoc} */
     @SuppressWarnings("unchecked")
     @Override public <R> QueryCursor<R> query(Query<R> qry) {
-        A.notNull(qry, "qry");
-
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            validate(qry);
-
-            if (qry instanceof ContinuousQuery)
-                return (QueryCursor<R>)queryContinuous((ContinuousQuery<K, V>)qry, qry.isLocal());
-
-            if (qry instanceof SqlQuery) {
-                SqlQuery p = (SqlQuery)qry;
-
-                if (isReplicatedDataNode() || ctx.isLocal() || qry.isLocal())
-                    return (QueryCursor<R>)new QueryCursorImpl<>(ctx.kernalContext().query().<K, V>queryLocal(ctx, p));
-
-                return (QueryCursor<R>)ctx.kernalContext().query().queryTwoStep(ctx, p);
-            }
-
-            if (qry instanceof SqlFieldsQuery) {
-                SqlFieldsQuery p = (SqlFieldsQuery)qry;
-
-                if (isReplicatedDataNode() || ctx.isLocal() || qry.isLocal())
-                    return (QueryCursor<R>)ctx.kernalContext().query().queryLocalFields(ctx, p);
-
-                return (QueryCursor<R>)ctx.kernalContext().query().queryTwoStep(ctx, p);
-            }
-
-            return (QueryCursor<R>)query(qry, projection(qry.isLocal()));
-        }
-        catch (Exception e) {
-            if (e instanceof CacheException)
-                throw e;
-
-            throw new CacheException(e);
+            return delegate.query(qry);
         }
         finally {
             gate.leave(prev);
         }
     }
 
-    /**
-     * @return {@code true} If this is a replicated cache and we are on a data node.
-     */
-    private boolean isReplicatedDataNode() {
-        return ctx.isReplicated() && ctx.affinityNode();
-    }
-
-    /**
-     * Checks query.
-     *
-     * @param qry Query
-     * @throws CacheException If query indexing disabled for sql query.
-     */
-    private void validate(Query qry) {
-        if (!GridQueryProcessor.isEnabled(ctx.config()) && !(qry instanceof ScanQuery) &&
-            !(qry instanceof ContinuousQuery))
-            throw new CacheException("Indexing is disabled for cache: " + ctx.cache().name());
-    }
-
     /** {@inheritDoc} */
     @Override public Iterable<Entry<K, V>> localEntries(CachePeekMode... peekModes) throws CacheException {
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
@@ -513,9 +248,6 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         try {
             return delegate.localEntries(peekModes);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
-        }
         finally {
             gate.leave(prev);
         }
@@ -526,7 +258,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            return delegate.queries().metrics();
+            return delegate.queryMetrics();
         }
         finally {
             gate.leave(prev);
@@ -538,7 +270,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            delegate.evictAll(keys);
+            delegate.localEvict(keys);
         }
         finally {
             gate.leave(prev);
@@ -550,10 +282,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            return delegate.localPeek(key, peekModes, null);
-        }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+            return delegate.localPeek(key, peekModes);
         }
         finally {
             gate.leave(prev);
@@ -562,18 +291,13 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
 
     /** {@inheritDoc} */
     @Override public void localPromote(Set<? extends K> keys) throws CacheException {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-            try {
-                delegate.promoteAll(keys);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            delegate.localPromote(keys);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
@@ -582,16 +306,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            if (isAsync()) {
-                setFuture(delegate.sizeAsync(peekModes));
-
-                return 0;
-            }
-            else
-                return delegate.size(peekModes);
-        }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+            return delegate.size(peekModes);
         }
         finally {
             gate.leave(prev);
@@ -605,9 +320,6 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         try {
             return delegate.localSize(peekModes);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
-        }
         finally {
             gate.leave(prev);
         }
@@ -615,47 +327,25 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
 
     /** {@inheritDoc} */
     @Override public V get(K key) {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    setFuture(delegate.getAsync(key));
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return null;
-                }
-                else
-                    return delegate.get(key);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.get(key);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Override public Map<K, V> getAll(Set<? extends K> keys) {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    setFuture(delegate.getAllAsync(keys));
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return null;
-                }
-                else
-                    return delegate.getAll(keys);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.getAll(keys);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
@@ -664,24 +354,13 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
      * @return Values map.
      */
     public Map<K, V> getAll(Collection<? extends K> keys) {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    setFuture(delegate.getAllAsync(keys));
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return null;
-                }
-                else
-                    return delegate.getAll(keys);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.getAll(keys);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
@@ -707,13 +386,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            if (isAsync()) {
-                setFuture(delegate.containsKeyAsync(key));
-
-                return false;
-            }
-            else
-                return delegate.containsKey(key);
+            return delegate.containsKey(key);
         }
         finally {
             gate.leave(prev);
@@ -725,13 +398,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            if (isAsync()) {
-                setFuture(delegate.containsKeysAsync(keys));
-
-                return false;
-            }
-            else
-                return delegate.containsKeys(keys);
+            return delegate.containsKeys(keys);
         }
         finally {
             gate.leave(prev);
@@ -747,22 +414,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            IgniteInternalFuture<?> fut = ctx.cache().loadAll(keys, replaceExisting);
-
-            if (completionLsnr != null) {
-                fut.listen(new CI1<IgniteInternalFuture<?>>() {
-                    @Override public void apply(IgniteInternalFuture<?> fut) {
-                        try {
-                            fut.get();
-
-                            completionLsnr.onCompletion();
-                        }
-                        catch (IgniteCheckedException e) {
-                            completionLsnr.onException(cacheException(e));
-                        }
-                    }
-                });
-            }
+            delegate.loadAll(keys, replaceExisting, completionLsnr);
         }
         finally {
             gate.leave(prev);
@@ -771,245 +423,133 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
 
     /** {@inheritDoc} */
     @Override public void put(K key, V val) {
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
+
         try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync())
-                    setFuture(delegate.putAsync(key, val));
-                else
-                    delegate.put(key, val);
-            }
-            finally {
-                gate.leave(prev);
-            }
+            delegate.put(key, val);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Override public V getAndPut(K key, V val) {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    setFuture(delegate.getAndPutAsync(key, val));
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return null;
-                }
-                else
-                    return delegate.getAndPut(key, val);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.getAndPut(key, val);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Override public void putAll(Map<? extends K, ? extends V> map) {
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
+
         try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync())
-                    setFuture(delegate.putAllAsync(map));
-                else
-                    delegate.putAll(map);
-            }
-            finally {
-                gate.leave(prev);
-            }
+            delegate.putAll(map);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Override public boolean putIfAbsent(K key, V val) {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    setFuture(delegate.putIfAbsentAsync(key, val));
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return false;
-                }
-                else
-                    return delegate.putIfAbsent(key, val);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.putIfAbsent(key, val);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Override public boolean remove(K key) {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    setFuture(delegate.removeAsync(key));
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return false;
-                }
-                else
-                    return delegate.remove(key);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.remove(key);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Override public boolean remove(K key, V oldVal) {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    setFuture(delegate.removeAsync(key, oldVal));
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return false;
-                }
-                else
-                    return delegate.remove(key, oldVal);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.remove(key, oldVal);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Override public V getAndRemove(K key) {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    setFuture(delegate.getAndRemoveAsync(key));
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return null;
-                }
-                else
-                    return delegate.getAndRemove(key);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.getAndRemove(key);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Override public boolean replace(K key, V oldVal, V newVal) {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    setFuture(delegate.replaceAsync(key, oldVal, newVal));
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return false;
-                }
-                else
-                    return delegate.replace(key, oldVal, newVal);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.replace(key, oldVal, newVal);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Override public boolean replace(K key, V val) {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    setFuture(delegate.replaceAsync(key, val));
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return false;
-                }
-                else
-                    return delegate.replace(key, val);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.replace(key, val);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Override public V getAndReplace(K key, V val) {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    setFuture(delegate.getAndReplaceAsync(key, val));
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return null;
-                }
-                else
-                    return delegate.getAndReplace(key, val);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.getAndReplace(key, val);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Override public void removeAll(Set<? extends K> keys) {
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
+
         try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync())
-                    setFuture(delegate.removeAllAsync(keys));
-                else
-                    delegate.removeAll(keys);
-            }
-            finally {
-                gate.leave(prev);
-            }
+            delegate.removeAll(keys);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
@@ -1018,13 +558,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            if (isAsync())
-                setFuture(delegate.removeAllAsync());
-            else
-                delegate.removeAll();
-        }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+            delegate.removeAll();
         }
         finally {
             gate.leave(prev);
@@ -1036,13 +570,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            if (isAsync())
-                setFuture(delegate.clearAsync(key));
-            else
-                delegate.clear(key);
-        }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+            delegate.clear(key);
         }
         finally {
             gate.leave(prev);
@@ -1054,13 +582,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            if (isAsync())
-                setFuture(delegate.clearAsync(keys));
-            else
-                delegate.clearAll(keys);
-        }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+            delegate.clearAll(keys);
         }
         finally {
             gate.leave(prev);
@@ -1072,13 +594,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            if (isAsync())
-                setFuture(delegate.clearAsync());
-            else
-                delegate.clear();
-        }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+            delegate.clear();
         }
         finally {
             gate.leave(prev);
@@ -1090,7 +606,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            delegate.clearLocally(key);
+            delegate.localClear(key);
         }
         finally {
             gate.leave(prev);
@@ -1102,8 +618,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            for (K key : keys)
-                delegate.clearLocally(key);
+            delegate.localClearAll(keys);
         }
         finally {
             gate.leave(prev);
@@ -1113,76 +628,26 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
     /** {@inheritDoc} */
     @Override public <T> T invoke(K key, EntryProcessor<K, V, T> entryProcessor, Object... args)
         throws EntryProcessorException {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    IgniteInternalFuture<EntryProcessorResult<T>> fut = delegate.invokeAsync(key, entryProcessor, args);
-
-                    IgniteInternalFuture<T> fut0 = fut.chain(new CX1<IgniteInternalFuture<EntryProcessorResult<T>>, T>() {
-                        @Override public T applyx(IgniteInternalFuture<EntryProcessorResult<T>> fut)
-                            throws IgniteCheckedException {
-                            EntryProcessorResult<T> res = fut.get();
-
-                            return res != null ? res.get() : null;
-                        }
-                    });
-
-                    setFuture(fut0);
-
-                    return null;
-                }
-                else {
-                    EntryProcessorResult<T> res = delegate.invoke(key, entryProcessor, args);
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return res != null ? res.get() : null;
-                }
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.invoke(key, entryProcessor, args);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Override public <T> T invoke(K key, CacheEntryProcessor<K, V, T> entryProcessor, Object... args)
         throws EntryProcessorException {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    IgniteInternalFuture<EntryProcessorResult<T>> fut = delegate.invokeAsync(key, entryProcessor, args);
-
-                    IgniteInternalFuture<T> fut0 = fut.chain(new CX1<IgniteInternalFuture<EntryProcessorResult<T>>, T>() {
-                        @Override public T applyx(IgniteInternalFuture<EntryProcessorResult<T>> fut)
-                            throws IgniteCheckedException {
-                            EntryProcessorResult<T> res = fut.get();
-
-                            return res != null ? res.get() : null;
-                        }
-                    });
-
-                    setFuture(fut0);
-
-                    return null;
-                }
-                else {
-                    EntryProcessorResult<T> res = delegate.invoke(key, entryProcessor, args);
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return res != null ? res.get() : null;
-                }
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.invoke(key, entryProcessor, args);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
@@ -1190,24 +655,13 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
     @Override public <T> Map<K, EntryProcessorResult<T>> invokeAll(Set<? extends K> keys,
                                                                    EntryProcessor<K, V, T> entryProcessor,
                                                                    Object... args) {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    setFuture(delegate.invokeAllAsync(keys, entryProcessor, args));
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return null;
-                }
-                else
-                    return delegate.invokeAll(keys, entryProcessor, args);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.invokeAll(keys, entryProcessor, args);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
@@ -1215,24 +669,13 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
     @Override public <T> Map<K, EntryProcessorResult<T>> invokeAll(Set<? extends K> keys,
         CacheEntryProcessor<K, V, T> entryProcessor,
         Object... args) {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    setFuture(delegate.invokeAllAsync(keys, entryProcessor, args));
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return null;
-                }
-                else
-                    return delegate.invokeAll(keys, entryProcessor, args);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.invokeAll(keys, entryProcessor, args);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
@@ -1240,42 +683,31 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
     @Override public <T> Map<K, EntryProcessorResult<T>> invokeAll(
         Map<? extends K, ? extends EntryProcessor<K, V, T>> map,
         Object... args) {
-        try {
-            GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
-
-            try {
-                if (isAsync()) {
-                    setFuture(delegate.invokeAllAsync(map, args));
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
-                    return null;
-                }
-                else
-                    return delegate.invokeAll(map, args);
-            }
-            finally {
-                gate.leave(prev);
-            }
+        try {
+            return delegate.invokeAll(map, args);
         }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+        finally {
+            gate.leave(prev);
         }
     }
 
     /** {@inheritDoc} */
     @Override public String getName() {
-        return delegate.name();
+        return delegate.getName();
     }
 
     /** {@inheritDoc} */
     @Override public CacheManager getCacheManager() {
-        return cacheMgr;
+        return delegate.getCacheManager();
     }
 
     /**
      * @param cacheMgr Cache manager.
      */
     public void setCacheManager(CacheManager cacheMgr) {
-        this.cacheMgr = cacheMgr;
+        delegate.setCacheManager(cacheMgr);
     }
 
     /** {@inheritDoc} */
@@ -1286,7 +718,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         IgniteInternalFuture<?> fut;
 
         try {
-            fut = ctx.kernalContext().cache().dynamicStopCache(ctx.name());
+            fut = context().kernalContext().cache().dynamicStopCache(context().name());
         }
         finally {
             gate.leave();
@@ -1296,7 +728,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
             fut.get();
         }
         catch (IgniteCheckedException e) {
-            throw cacheException(e);
+            throw CU.convertToCacheException(e);
         }
     }
 
@@ -1306,7 +738,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
             return true;
 
         try {
-            return ctx.kernalContext().cache().context().closed(ctx);
+            return context().kernalContext().cache().context().closed(context());
         }
         finally {
             gate.leave();
@@ -1317,20 +749,13 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
      *
      */
     public GridCacheProjectionEx delegate() {
-        return delegate;
+        return delegate.delegate();
     }
 
     /** {@inheritDoc} */
     @SuppressWarnings("unchecked")
     @Override public <T> T unwrap(Class<T> clazz) {
-        if (clazz.isAssignableFrom(getClass()))
-            return (T)this;
-        else if (clazz.isAssignableFrom(IgniteEx.class))
-            return (T)ctx.grid();
-        else if (clazz.isAssignableFrom(legacyProxy.getClass()))
-            return (T)legacyProxy;
-
-        throw new IllegalArgumentException("Unwrapping to class is not supported: " + clazz);
+        return delegate.unwrap(clazz);
     }
 
     /** {@inheritDoc} */
@@ -1338,10 +763,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            ctx.continuousQueries().executeJCacheQuery(lsnrCfg, false);
-        }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+            delegate.registerCacheEntryListener(lsnrCfg);
         }
         finally {
             gate.leave(prev);
@@ -1353,10 +775,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            ctx.continuousQueries().cancelJCacheQuery(lsnrCfg);
-        }
-        catch (IgniteCheckedException e) {
-            throw cacheException(e);
+            delegate.deregisterCacheEntryListener(lsnrCfg);
         }
         finally {
             gate.leave(prev);
@@ -1368,7 +787,7 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            return ctx.cache().igniteIterator();
+            return delegate.iterator();
         }
         finally {
             gate.leave(prev);
@@ -1377,7 +796,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 delegate.createAsyncInstance();
     }
 
     /**
@@ -1409,13 +828,13 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
         try {
             GridCacheProjectionImpl<K1, V1> prj0 = new GridCacheProjectionImpl<>(
                 (CacheProjection<K1, V1>)(prj != null ? prj : delegate),
-                (GridCacheContext<K1, V1>)ctx,
-                prj != null ? prj.skipStore() : false,
+                (GridCacheContext<K1, V1>)context(),
+                prj != null && prj.skipStore(),
                 prj != null ? prj.subjectId() : null,
                 true,
                 prj != null ? prj.expiry() : null);
 
-            return new IgniteCacheProxy<>((GridCacheContext<K1, V1>)ctx,
+            return new IgniteCacheProxy<>((GridCacheContext<K1, V1>)context(),
                 prj0,
                 prj0,
                 isAsync());
@@ -1438,14 +857,14 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
                 return this;
 
             GridCacheProjectionImpl<K, V> prj0 = new GridCacheProjectionImpl<>(
-                (prj != null ? prj : delegate),
-                ctx,
+                (prj != null ? prj : delegate.delegate()),
+                context(),
                 true,
                 prj != null ? prj.subjectId() : null,
                 prj != null && prj.isKeepPortable(),
                 prj != null ? prj.expiry() : null);
 
-            return new IgniteCacheProxy<>(ctx,
+            return new IgniteCacheProxy<>(context(),
                 prj0,
                 prj0,
                 isAsync());
@@ -1456,32 +875,15 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
     }
 
     /**
-     * @param e Checked exception.
-     * @return Cache exception.
-     */
-    private RuntimeException cacheException(IgniteCheckedException e) {
-        return CU.convertToCacheException(e);
-    }
-
-    /**
-     * @param fut Future for async operation.
-     */
-    private <R> void setFuture(IgniteInternalFuture<R> fut) {
-        curFut.set(new IgniteFutureImpl<>(fut));
-    }
-
-    /**
      * @return Legacy proxy.
      */
     @NotNull
     public GridCacheProxyImpl<K, V> legacyProxy() {
-        return legacyProxy;
+        return delegate.legacyProxy();
     }
 
     /** {@inheritDoc} */
     @Override public void writeExternal(ObjectOutput out) throws IOException {
-        out.writeObject(ctx);
-
         out.writeObject(delegate);
 
         out.writeObject(prj);
@@ -1490,69 +892,20 @@ public class IgniteCacheProxy<K, V> extends AsyncSupportAdapter<IgniteCache<K, V
     /** {@inheritDoc} */
     @SuppressWarnings({"unchecked"})
     @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-        ctx = (GridCacheContext<K, V>)in.readObject();
-
-        delegate = (GridCacheProjectionEx<K, V>)in.readObject();
+        delegate = (IgniteCacheProxyLockFree<K, V>)in.readObject();
 
         prj = (GridCacheProjectionImpl<K, V>)in.readObject();
 
-        gate = ctx.gate();
+        gate = delegate.context().gate();
     }
 
     /** {@inheritDoc} */
     @Override public IgniteFuture<?> rebalance() {
-        ctx.preloader().forcePreload();
-
-        return new IgniteFutureImpl<>(ctx.preloader().syncFuture());
+        return delegate.rebalance();
     }
 
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(IgniteCacheProxy.class, this);
     }
-
-    /**
-     * Closeable iterator.
-     */
-    private abstract static class ClIter<X, Y> extends GridCloseableIteratorAdapter<Y> {
-        /** */
-        private X cur;
-
-        /** */
-        private CacheQueryFuture<X> fut;
-
-        /**
-         * @param fut Future.
-         */
-        protected ClIter(CacheQueryFuture<X> fut) {
-            this.fut = fut;
-        }
-
-        /** {@inheritDoc} */
-        @Override protected Y onNext() throws IgniteCheckedException {
-            if (!onHasNext())
-                throw new NoSuchElementException();
-
-            X e = cur;
-
-            cur = null;
-
-            return convert(e);
-        }
-
-        /**
-         * @param x X.
-         */
-        protected abstract Y convert(X x);
-
-        /** {@inheritDoc} */
-        @Override protected boolean onHasNext() throws IgniteCheckedException {
-            return cur != null || (cur = fut.next()) != null;
-        }
-
-        /** {@inheritDoc} */
-        @Override protected void onClose() throws IgniteCheckedException {
-            fut.cancel();
-        }
-    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/56ef2695/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxyLockFree.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxyLockFree.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxyLockFree.java
index 81e1f24..976eeac 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxyLockFree.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxyLockFree.java
@@ -17,8 +17,1133 @@
 
 package org.apache.ignite.internal.processors.cache;
 
+import org.apache.ignite.*;
+import org.apache.ignite.cache.*;
+import org.apache.ignite.cache.CacheManager;
+import org.apache.ignite.cache.query.*;
+import org.apache.ignite.cluster.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.*;
+import org.apache.ignite.internal.processors.cache.query.*;
+import org.apache.ignite.internal.processors.query.*;
+import org.apache.ignite.internal.util.*;
+import org.apache.ignite.internal.util.future.*;
+import org.apache.ignite.internal.util.tostring.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.mxbean.*;
+import org.apache.ignite.spi.discovery.tcp.internal.*;
+import org.jetbrains.annotations.*;
+
+import javax.cache.*;
+import javax.cache.configuration.*;
+import javax.cache.expiry.*;
+import javax.cache.integration.*;
+import javax.cache.processor.*;
+import java.io.*;
+import java.util.*;
+import java.util.concurrent.locks.*;
+
 /**
  * Cache proxy lock free.
  */
-public class IgniteCacheProxyLockFree {
+public class IgniteCacheProxyLockFree <K, V> extends AsyncSupportAdapter<IgniteCache<K, V>>
+    implements IgniteCache<K, V>, Externalizable {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** */
+    private static final IgniteBiPredicate ACCEPT_ALL = new IgniteBiPredicate() {
+            @Override public boolean apply(Object k, Object v) {
+                return true;
+            }
+        };
+
+    /** Context. */
+    private GridCacheContext<K, V> ctx;
+
+    /** Delegate. */
+    @GridToStringInclude
+    private GridCacheProjectionEx<K, V> delegate;
+
+    /** Projection. */
+    private GridCacheProjectionImpl<K, V> prj;
+
+    /** */
+    @GridToStringExclude
+    private GridCacheProxyImpl<K, V> legacyProxy;
+
+    /** */
+    @GridToStringExclude
+    private CacheManager cacheMgr;
+
+    /**
+     * Empty constructor required for {@link Externalizable}.
+     */
+    public IgniteCacheProxyLockFree() {
+        // No-op.
+    }
+
+    /**
+     * @param ctx Context.
+     * @param delegate Delegate.
+     * @param prj Projection.
+     * @param async Async support flag.
+     */
+    public IgniteCacheProxyLockFree(
+        GridCacheContext<K, V> ctx,
+        GridCacheProjectionEx<K, V> delegate,
+        @Nullable GridCacheProjectionImpl<K, V> prj,
+        boolean async
+    ) {
+        super(async);
+
+        assert ctx != null;
+        assert delegate != null;
+
+        this.ctx = ctx;
+        this.delegate = delegate;
+        this.prj = prj;
+
+        legacyProxy = new GridCacheProxyImpl<>(ctx, delegate, prj);
+    }
+
+    /**
+     * @return Context.
+     */
+    public GridCacheContext<K, V> context() {
+        return ctx;
+    }
+
+    /** {@inheritDoc} */
+    @Override public CacheMetrics metrics() {
+        return ctx.cache().metrics();
+    }
+
+    /** {@inheritDoc} */
+    @Override public CacheMetrics metrics(ClusterGroup grp) {
+        List<CacheMetrics> metrics = new ArrayList<>(grp.nodes().size());
+
+        for (ClusterNode node : grp.nodes()) {
+            Map<Integer, CacheMetrics> nodeCacheMetrics = ((TcpDiscoveryNode)node).cacheMetrics();
+
+            if (nodeCacheMetrics != null) {
+                CacheMetrics e = nodeCacheMetrics.get(context().cacheId());
+
+                if (e != null)
+                    metrics.add(e);
+            }
+        }
+
+        return new CacheMetricsSnapshot(ctx.cache().metrics(), metrics);
+    }
+
+    /** {@inheritDoc} */
+    @Override public CacheMetricsMXBean mxBean() {
+        return ctx.cache().mxBean();
+    }
+
+    /** {@inheritDoc} */
+    @Override public <C extends Configuration<K, V>> C getConfiguration(Class<C> clazz) {
+        CacheConfiguration cfg = ctx.config();
+
+        if (!clazz.isAssignableFrom(cfg.getClass()))
+            throw new IllegalArgumentException();
+
+        return clazz.cast(cfg);
+    }
+
+    /** {@inheritDoc} */
+    @Nullable @Override public Entry<K, V> randomEntry() {
+        return ctx.cache().randomEntry();
+    }
+
+    /** {@inheritDoc} */
+    @Override public IgniteCache<K, V> withExpiryPolicy(ExpiryPolicy plc) {
+        GridCacheProjectionEx<K, V> prj0 = prj != null ? prj.withExpiryPolicy(plc) : delegate.withExpiryPolicy(plc);
+
+        return new IgniteCacheProxy<>(ctx, prj0, (GridCacheProjectionImpl<K, V>)prj0, isAsync());
+    }
+
+    /** {@inheritDoc} */
+    @Override public IgniteCache<K, V> withSkipStore() {
+        return skipStore();
+    }
+
+    /** {@inheritDoc} */
+    @Override public void loadCache(@Nullable IgniteBiPredicate<K, V> p, @Nullable Object... args) {
+        try {
+            if (isAsync())
+                setFuture(ctx.cache().globalLoadCacheAsync(p, args));
+            else
+                ctx.cache().globalLoadCache(p, args);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void localLoadCache(@Nullable IgniteBiPredicate<K, V> p, @Nullable Object... args) {
+        try {
+            if (isAsync())
+                setFuture(delegate.localLoadCacheAsync(p, args));
+            else
+                delegate.localLoadCache(p, args);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Nullable @Override public V getAndPutIfAbsent(K key, V val) throws CacheException {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.getAndPutIfAbsentAsync(key, val));
+
+                return null;
+            }
+            else
+                return delegate.getAndPutIfAbsent(key, val);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public Lock lock(K key) throws CacheException {
+        return lockAll(Collections.singleton(key));
+    }
+
+    /** {@inheritDoc} */
+    @Override public Lock lockAll(final Collection<? extends K> keys) {
+        return new CacheLockImpl<>(ctx.gate(), delegate, prj, keys);
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean isLocalLocked(K key, boolean byCurrThread) {
+        return byCurrThread ? delegate.isLockedByThread(key) : delegate.isLocked(key);
+    }
+
+    /**
+     * @param filter Filter.
+     * @param grp Optional cluster group.
+     * @return Cursor.
+     */
+    @SuppressWarnings("unchecked")
+    private QueryCursor<Entry<K,V>> query(Query filter, @Nullable ClusterGroup grp) {
+        final CacheQuery<Map.Entry<K,V>> qry;
+        final CacheQueryFuture<Map.Entry<K,V>> fut;
+
+        if (filter instanceof ScanQuery) {
+            IgniteBiPredicate<K, V> p = ((ScanQuery)filter).getFilter();
+
+            qry = delegate.queries().createScanQuery(p != null ? p : ACCEPT_ALL);
+
+            if (grp != null)
+                qry.projection(grp);
+
+            fut = qry.execute();
+        }
+        else if (filter instanceof TextQuery) {
+            TextQuery p = (TextQuery)filter;
+
+            qry = delegate.queries().createFullTextQuery(p.getType(), p.getText());
+
+            if (grp != null)
+                qry.projection(grp);
+
+            fut = qry.execute();
+        }
+        else if (filter instanceof SpiQuery) {
+            qry = ((GridCacheQueriesEx)delegate.queries()).createSpiQuery();
+
+            if (grp != null)
+                qry.projection(grp);
+
+            fut = qry.execute(((SpiQuery)filter).getArgs());
+        }
+        else {
+            if (filter instanceof SqlFieldsQuery)
+                throw new CacheException("Use methods 'queryFields' and 'localQueryFields' for " +
+                    SqlFieldsQuery.class.getSimpleName() + ".");
+
+            throw new CacheException("Unsupported query type: " + filter);
+        }
+
+        return new QueryCursorImpl<>(new GridCloseableIteratorAdapter<Entry<K,V>>() {
+            /** */
+            private Map.Entry<K,V> cur;
+
+            @Override protected Entry<K,V> onNext() throws IgniteCheckedException {
+                if (!onHasNext())
+                    throw new NoSuchElementException();
+
+                Map.Entry<K,V> e = cur;
+
+                cur = null;
+
+                return new CacheEntryImpl<>(e.getKey(), e.getValue());
+            }
+
+            @Override protected boolean onHasNext() throws IgniteCheckedException {
+                return cur != null || (cur = fut.next()) != null;
+            }
+
+            @Override protected void onClose() throws IgniteCheckedException {
+                fut.cancel();
+            }
+        });
+    }
+
+    /**
+     * @param local Enforce local.
+     * @return Local node cluster group.
+     */
+    private ClusterGroup projection(boolean local) {
+        if (local || ctx.isLocal() || isReplicatedDataNode())
+            return ctx.kernalContext().grid().cluster().forLocal();
+
+        if (ctx.isReplicated())
+            return ctx.kernalContext().grid().cluster().forDataNodes(ctx.name()).forRandom();
+
+        return null;
+    }
+
+    /**
+     * Executes continuous query.
+     *
+     * @param qry Query.
+     * @param loc Local flag.
+     * @return Initial iteration cursor.
+     */
+    @SuppressWarnings("unchecked")
+    private QueryCursor<Entry<K, V>> queryContinuous(ContinuousQuery qry, boolean loc) {
+        if (qry.getInitialQuery() instanceof ContinuousQuery)
+            throw new IgniteException("Initial predicate for continuous query can't be an instance of another " +
+                "continuous query. Use SCAN or SQL query for initial iteration.");
+
+        if (qry.getLocalListener() == null)
+            throw new IgniteException("Mandatory local listener is not set for the query: " + qry);
+
+        try {
+            final UUID routineId = ctx.continuousQueries().executeQuery(
+                qry.getLocalListener(),
+                qry.getRemoteFilter(),
+                qry.getPageSize(),
+                qry.getTimeInterval(),
+                qry.isAutoUnsubscribe(),
+                loc ? ctx.grid().cluster().forLocal() : null);
+
+            final QueryCursor<Cache.Entry<K, V>> cur =
+                qry.getInitialQuery() != null ? query(qry.getInitialQuery()) : null;
+
+            return new QueryCursor<Cache.Entry<K, V>>() {
+                @Override public Iterator<Cache.Entry<K, V>> iterator() {
+                    return cur != null ? cur.iterator() : new GridEmptyIterator<Cache.Entry<K, V>>();
+                }
+
+                @Override public List<Cache.Entry<K, V>> getAll() {
+                    return cur != null ? cur.getAll() : Collections.<Cache.Entry<K, V>>emptyList();
+                }
+
+                @Override public void close() {
+                    if (cur != null)
+                        cur.close();
+
+                    try {
+                        ctx.kernalContext().continuous().stopRoutine(routineId).get();
+                    }
+                    catch (IgniteCheckedException e) {
+                        throw U.convertException(e);
+                    }
+                }
+            };
+        }
+        catch (IgniteCheckedException e) {
+            throw U.convertException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings("unchecked")
+    @Override public <R> QueryCursor<R> query(Query<R> qry) {
+        A.notNull(qry, "qry");
+
+        try {
+            validate(qry);
+
+                if (qry instanceof ContinuousQuery)
+                return (QueryCursor<R>)queryContinuous((ContinuousQuery<K, V>)qry, qry.isLocal());
+
+            if (qry instanceof SqlQuery) {
+                SqlQuery p = (SqlQuery)qry;
+
+                if (isReplicatedDataNode() || ctx.isLocal() || qry.isLocal())
+                    return (QueryCursor<R>)new QueryCursorImpl<>(ctx.kernalContext().query().<K, V>queryLocal(ctx, p));
+
+                return (QueryCursor<R>)ctx.kernalContext().query().queryTwoStep(ctx, p);
+            }
+
+            if (qry instanceof SqlFieldsQuery) {
+                SqlFieldsQuery p = (SqlFieldsQuery)qry;
+
+                if (isReplicatedDataNode() || ctx.isLocal() || qry.isLocal())
+                    return (QueryCursor<R>)ctx.kernalContext().query().queryLocalFields(ctx, p);
+
+                return (QueryCursor<R>)ctx.kernalContext().query().queryTwoStep(ctx, p);
+            }
+
+            return (QueryCursor<R>)query(qry, projection(qry.isLocal()));
+        }
+        catch (Exception e) {
+            if (e instanceof CacheException)
+                throw e;
+
+            throw new CacheException(e);
+        }
+    }
+
+    /**
+     * @return {@code true} If this is a replicated cache and we are on a data node.
+     */
+    private boolean isReplicatedDataNode() {
+        return ctx.isReplicated() && ctx.affinityNode();
+    }
+
+    /**
+     * Checks query.
+     *
+     * @param qry Query
+     * @throws CacheException If query indexing disabled for sql query.
+     */
+    private void validate(Query qry) {
+        if (!GridQueryProcessor.isEnabled(ctx.config()) && !(qry instanceof ScanQuery) &&
+            !(qry instanceof ContinuousQuery))
+            throw new CacheException("Indexing is disabled for cache: " + ctx.cache().name());
+    }
+
+    /** {@inheritDoc} */
+    @Override public Iterable<Entry<K, V>> localEntries(CachePeekMode... peekModes) throws CacheException {
+        try {
+            return delegate.localEntries(peekModes);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public QueryMetrics queryMetrics() {
+        return delegate.queries().metrics();
+    }
+
+    /** {@inheritDoc} */
+    @Override public void localEvict(Collection<? extends K> keys) {
+        delegate.evictAll(keys);
+    }
+
+    /** {@inheritDoc} */
+    @Nullable @Override public V localPeek(K key, CachePeekMode... peekModes) {
+        try {
+            return delegate.localPeek(key, peekModes, null);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void localPromote(Set<? extends K> keys) throws CacheException {
+        try {
+            delegate.promoteAll(keys);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public int size(CachePeekMode... peekModes) throws CacheException {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.sizeAsync(peekModes));
+
+                return 0;
+            }
+            else
+                return delegate.size(peekModes);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public int localSize(CachePeekMode... peekModes) {
+        try {
+            return delegate.localSize(peekModes);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public V get(K key) {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.getAsync(key));
+
+                return null;
+            }
+            else
+                return delegate.get(key);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public Map<K, V> getAll(Set<? extends K> keys) {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.getAllAsync(keys));
+
+                return null;
+            }
+            else
+                return delegate.getAll(keys);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /**
+     * @param keys Keys.
+     * @return Values map.
+     */
+    public Map<K, V> getAll(Collection<? extends K> keys) {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.getAllAsync(keys));
+
+                return null;
+            }
+            else
+                return delegate.getAll(keys);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /**
+     * Gets entry set containing internal entries.
+     *
+     * @param filter Filter.
+     * @return Entry set.
+     */
+    public Set<Entry<K, V>> entrySetx(CacheEntryPredicate... filter) {
+        return delegate.entrySetx(filter);
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean containsKey(K key) {
+        if (isAsync()) {
+            setFuture(delegate.containsKeyAsync(key));
+
+            return false;
+        }
+        else
+            return delegate.containsKey(key);
+
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean containsKeys(Set<? extends K> keys) {
+        if (isAsync()) {
+            setFuture(delegate.containsKeysAsync(keys));
+
+            return false;
+        }
+        else
+            return delegate.containsKeys(keys);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void loadAll(
+        Set<? extends K> keys,
+        boolean replaceExisting,
+        @Nullable final CompletionListener completionLsnr
+    ) {
+        IgniteInternalFuture<?> fut = ctx.cache().loadAll(keys, replaceExisting);
+
+        if (completionLsnr != null) {
+            fut.listen(new CI1<IgniteInternalFuture<?>>() {
+                @Override public void apply(IgniteInternalFuture<?> fut) {
+                    try {
+                        fut.get();
+
+                        completionLsnr.onCompletion();
+                    }
+                    catch (IgniteCheckedException e) {
+                        completionLsnr.onException(cacheException(e));
+                    }
+                }
+            });
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void put(K key, V val) {
+        try {
+            if (isAsync())
+                setFuture(delegate.putAsync(key, val));
+            else
+                delegate.put(key, val);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public V getAndPut(K key, V val) {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.getAndPutAsync(key, val));
+
+                return null;
+            }
+            else
+                return delegate.getAndPut(key, val);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void putAll(Map<? extends K, ? extends V> map) {
+        try {
+            if (isAsync())
+                setFuture(delegate.putAllAsync(map));
+            else
+                delegate.putAll(map);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean putIfAbsent(K key, V val) {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.putIfAbsentAsync(key, val));
+
+                return false;
+            }
+            else
+                return delegate.putIfAbsent(key, val);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean remove(K key) {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.removeAsync(key));
+
+                return false;
+            }
+            else
+                return delegate.remove(key);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean remove(K key, V oldVal) {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.removeAsync(key, oldVal));
+
+                return false;
+            }
+            else
+                return delegate.remove(key, oldVal);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public V getAndRemove(K key) {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.getAndRemoveAsync(key));
+
+                return null;
+            }
+            else
+                return delegate.getAndRemove(key);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean replace(K key, V oldVal, V newVal) {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.replaceAsync(key, oldVal, newVal));
+
+                return false;
+            }
+            else
+                return delegate.replace(key, oldVal, newVal);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean replace(K key, V val) {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.replaceAsync(key, val));
+
+                return false;
+            }
+            else
+                return delegate.replace(key, val);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public V getAndReplace(K key, V val) {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.getAndReplaceAsync(key, val));
+
+                return null;
+            }
+            else
+                return delegate.getAndReplace(key, val);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void removeAll(Set<? extends K> keys) {
+        try {
+            if (isAsync())
+                setFuture(delegate.removeAllAsync(keys));
+            else
+                delegate.removeAll(keys);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void removeAll() {
+        try {
+            if (isAsync())
+                setFuture(delegate.removeAllAsync());
+            else
+                delegate.removeAll();
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void clear(K key) {
+        try {
+            if (isAsync())
+                setFuture(delegate.clearAsync(key));
+            else
+                delegate.clear(key);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void clearAll(Set<? extends K> keys) {
+        try {
+            if (isAsync())
+                setFuture(delegate.clearAsync(keys));
+            else
+                delegate.clearAll(keys);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void clear() {
+        try {
+            if (isAsync())
+                setFuture(delegate.clearAsync());
+            else
+                delegate.clear();
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void localClear(K key) {
+        delegate.clearLocally(key);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void localClearAll(Set<? extends K> keys) {
+        for (K key : keys)
+            delegate.clearLocally(key);
+    }
+
+    /** {@inheritDoc} */
+    @Override public <T> T invoke(K key, EntryProcessor<K, V, T> entryProcessor, Object... args)
+        throws EntryProcessorException {
+        try {
+            if (isAsync()) {
+                IgniteInternalFuture<EntryProcessorResult<T>> fut = delegate.invokeAsync(key, entryProcessor, args);
+
+                IgniteInternalFuture<T> fut0 = fut.chain(new CX1<IgniteInternalFuture<EntryProcessorResult<T>>, T>() {
+                    @Override public T applyx(IgniteInternalFuture<EntryProcessorResult<T>> fut)
+                        throws IgniteCheckedException {
+                        EntryProcessorResult<T> res = fut.get();
+
+                        return res != null ? res.get() : null;
+                    }
+                });
+
+                setFuture(fut0);
+
+                return null;
+            }
+            else {
+                EntryProcessorResult<T> res = delegate.invoke(key, entryProcessor, args);
+
+                return res != null ? res.get() : null;
+            }
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public <T> T invoke(K key, CacheEntryProcessor<K, V, T> entryProcessor, Object... args)
+        throws EntryProcessorException {
+        try {
+            if (isAsync()) {
+                IgniteInternalFuture<EntryProcessorResult<T>> fut = delegate.invokeAsync(key, entryProcessor, args);
+
+                IgniteInternalFuture<T> fut0 = fut.chain(new CX1<IgniteInternalFuture<EntryProcessorResult<T>>, T>() {
+                    @Override public T applyx(IgniteInternalFuture<EntryProcessorResult<T>> fut)
+                        throws IgniteCheckedException {
+                        EntryProcessorResult<T> res = fut.get();
+
+                        return res != null ? res.get() : null;
+                    }
+                });
+
+                setFuture(fut0);
+
+                return null;
+            }
+            else {
+                EntryProcessorResult<T> res = delegate.invoke(key, entryProcessor, args);
+
+                return res != null ? res.get() : null;
+            }
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public <T> Map<K, EntryProcessorResult<T>> invokeAll(Set<? extends K> keys,
+                                                                   EntryProcessor<K, V, T> entryProcessor,
+                                                                   Object... args) {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.invokeAllAsync(keys, entryProcessor, args));
+
+                return null;
+            }
+            else
+                return delegate.invokeAll(keys, entryProcessor, args);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public <T> Map<K, EntryProcessorResult<T>> invokeAll(Set<? extends K> keys,
+                                                                   CacheEntryProcessor<K, V, T> entryProcessor,
+                                                                   Object... args) {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.invokeAllAsync(keys, entryProcessor, args));
+
+                return null;
+            }
+            else
+                return delegate.invokeAll(keys, entryProcessor, args);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public <T> Map<K, EntryProcessorResult<T>> invokeAll(
+        Map<? extends K, ? extends EntryProcessor<K, V, T>> map,
+        Object... args) {
+        try {
+            if (isAsync()) {
+                setFuture(delegate.invokeAllAsync(map, args));
+
+                return null;
+            }
+            else
+                return delegate.invokeAll(map, args);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public String getName() {
+        return delegate.name();
+    }
+
+    /** {@inheritDoc} */
+    @Override public CacheManager getCacheManager() {
+        return cacheMgr;
+    }
+
+    /**
+     * @param cacheMgr Cache manager.
+     */
+    public void setCacheManager(CacheManager cacheMgr) {
+        this.cacheMgr = cacheMgr;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void close() {
+        IgniteInternalFuture<?> fut;
+
+        fut = ctx.kernalContext().cache().dynamicStopCache(ctx.name());
+
+        try {
+            fut.get();
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean isClosed() {
+        return ctx.kernalContext().cache().context().closed(ctx);
+    }
+
+    /**
+     *
+     */
+    public GridCacheProjectionEx delegate() {
+        return delegate;
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings("unchecked")
+    @Override public <T> T unwrap(Class<T> clazz) {
+        if (clazz.isAssignableFrom(getClass()))
+            return (T)this;
+        else if (clazz.isAssignableFrom(IgniteEx.class))
+            return (T)ctx.grid();
+        else if (clazz.isAssignableFrom(legacyProxy.getClass()))
+            return (T)legacyProxy;
+
+        throw new IllegalArgumentException("Unwrapping to class is not supported: " + clazz);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void registerCacheEntryListener(CacheEntryListenerConfiguration<K, V> lsnrCfg) {
+        try {
+            ctx.continuousQueries().executeJCacheQuery(lsnrCfg, false);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void deregisterCacheEntryListener(CacheEntryListenerConfiguration<K, V> lsnrCfg) {
+        try {
+            ctx.continuousQueries().cancelJCacheQuery(lsnrCfg);
+        }
+        catch (IgniteCheckedException e) {
+            throw cacheException(e);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public Iterator<Cache.Entry<K, V>> iterator() {
+        return ctx.cache().igniteIterator();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected IgniteCache<K, V> createAsyncInstance() {
+        return new IgniteCacheProxyLockFree<>(ctx, delegate, prj, true);
+    }
+
+    /**
+     * Creates projection that will operate with portable objects. <p> Projection returned by this method will force
+     * cache not to deserialize portable objects, so keys and values will be returned from cache API methods without
+     * changes. Therefore, signature of the projection can contain only following types: <ul> <li>{@code PortableObject}
+     * for portable classes</li> <li>All primitives (byte, int, ...) and there boxed versions (Byte, Integer, ...)</li>
+     * <li>Arrays of primitives (byte[], int[], ...)</li> <li>{@link String} and array of {@link String}s</li>
+     * <li>{@link UUID} and array of {@link UUID}s</li> <li>{@link Date} and array of {@link Date}s</li> <li>{@link
+     * java.sql.Timestamp} and array of {@link java.sql.Timestamp}s</li> <li>Enums and array of enums</li> <li> Maps,
+     * collections and array of objects (but objects inside them will still be converted if they are portable) </li>
+     * </ul> <p> For example, if you use {@link Integer} as a key and {@code Value} class as a value (which will be
+     * stored in portable format), you should acquire following projection to avoid deserialization:
+     * <pre>
+     * CacheProjection<Integer, GridPortableObject> prj = cache.keepPortable();
+     *
+     * // Value is not deserialized and returned in portable format.
+     * GridPortableObject po = prj.get(1);
+     * </pre>
+     * <p> Note that this method makes sense only if cache is working in portable mode ({@code
+     * CacheConfiguration#isPortableEnabled()} returns {@code true}. If not, this method is no-op and will return
+     * current projection.
+     *
+     * @return Projection for portable objects.
+     */
+    public <K1, V1> IgniteCache<K1, V1> keepPortable() {
+        GridCacheProjectionImpl<K1, V1> prj0 = new GridCacheProjectionImpl<>(
+            (CacheProjection<K1, V1>)(prj != null ? prj : delegate),
+            (GridCacheContext<K1, V1>)ctx,
+            prj != null && prj.skipStore(),
+            prj != null ? prj.subjectId() : null,
+            true,
+            prj != null ? prj.expiry() : null);
+
+        return new IgniteCacheProxyLockFree<>((GridCacheContext<K1, V1>)ctx,
+            prj0,
+            prj0,
+            isAsync());
+    }
+
+    /**
+     * @return Cache with skip store enabled.
+     */
+    public IgniteCache<K, V> skipStore() {
+        boolean skip = prj != null && prj.skipStore();
+
+        if (skip)
+            return this;
+
+        GridCacheProjectionImpl<K, V> prj0 = new GridCacheProjectionImpl<>(
+            (prj != null ? prj : delegate),
+            ctx,
+            true,
+            prj != null ? prj.subjectId() : null,
+            prj != null && prj.isKeepPortable(),
+            prj != null ? prj.expiry() : null);
+
+        return new IgniteCacheProxyLockFree<>(ctx,
+            prj0,
+            prj0,
+            isAsync());
+    }
+
+    /**
+     * @param e Checked exception.
+     * @return Cache exception.
+     */
+    private RuntimeException cacheException(IgniteCheckedException e) {
+        return CU.convertToCacheException(e);
+    }
+
+    /**
+     * @param fut Future for async operation.
+     */
+    private <R> void setFuture(IgniteInternalFuture<R> fut) {
+        curFut.set(new IgniteFutureImpl<>(fut));
+    }
+
+    /**
+     * @return Legacy proxy.
+     */
+    @NotNull
+    public GridCacheProxyImpl<K, V> legacyProxy() {
+        return legacyProxy;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void writeExternal(ObjectOutput out) throws IOException {
+        out.writeObject(ctx);
+
+        out.writeObject(delegate);
+
+        out.writeObject(prj);
+    }
+
+    /** {@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();
+    }
+
+    /** {@inheritDoc} */
+    @Override public IgniteFuture<?> rebalance() {
+        ctx.preloader().forcePreload();
+
+        return new IgniteFutureImpl<>(ctx.preloader().syncFuture());
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(IgniteCacheProxyLockFree.class, this);
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/56ef2695/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 e06946f..323c8ea 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
@@ -100,10 +100,12 @@ class DataStreamerUpdateJob implements GridPlainCallable<Object> {
 //        if (ignoreDepOwnership)
 //            cache.context().deploy().ignoreOwnership(true);
 
-        IgniteCacheProxy cache = ctx.cache().jcache(cacheName);
+        IgniteCacheProxyLockFree cache = ctx.cache().jcache(cacheName).lockFree();
+
+        cache.context().awaitStarted();
 
         if (skipStore)
-            cache = (IgniteCacheProxy<?, ?>)cache.withSkipStore();
+            cache = (IgniteCacheProxyLockFree<?, ?>)cache.withSkipStore();
 
         if (ignoreDepOwnership)
             cache.context().deploy().ignoreOwnership(true);


Mime
View raw message