ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [55/55] [abbrv] incubator-ignite git commit: # ignite-42
Date Thu, 22 Jan 2015 13:18:25 GMT
# ignite-42


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

Branch: refs/heads/sprint-1
Commit: b9288ffec904eefc70ca49b8b04f1b5ffcd0a3a0
Parents: 55b5a6d
Author: sboikov <sboikov@gridgain.com>
Authored: Thu Jan 22 14:22:21 2015 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Thu Jan 22 15:59:02 2015 +0300

----------------------------------------------------------------------
 .../processors/cache/GridCacheAdapter.java      |  16 +-
 .../processors/cache/GridCacheStoreManager.java | 261 +++++++++++++++++--
 .../IgniteCacheLoaderWriterAbstractTest.java    |  22 +-
 .../GridDataLoaderProcessorSelfTest.java        |  25 +-
 4 files changed, 277 insertions(+), 47 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9288ffe/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheAdapter.java
b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheAdapter.java
index 7992cd3..99d4cde 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheAdapter.java
@@ -3519,7 +3519,7 @@ public abstract class GridCacheAdapter<K, V> extends GridMetadataAwareAdapter
im
      */
     private void localLoadAndUpdate(final Collection<? extends K> keys) throws IgniteCheckedException
{
         try (final IgniteDataLoader<K, V> ldr = ctx.kernalContext().<K, V>dataLoad().dataLoader(ctx.namex(),
false)) {
-            ldr.updater(new SkipStoreUpdater<K, V>());
+            ldr.skipStore(true);
 
             final Collection<Map.Entry<K, V>> col = new ArrayList<>(ldr.perNodeBufferSize());
 
@@ -5338,18 +5338,4 @@ public abstract class GridCacheAdapter<K, V> extends GridMetadataAwareAdapter
im
                 ldr.addData(col);
         }
     }
-
-    /**
-     *
-     */
-    static class SkipStoreUpdater<K, V> implements IgniteDataLoadCacheUpdater<K,
V> {
-        /** {@inheritDoc} */
-        @Override public void update(IgniteCache<K, V> cache, Collection<Map.Entry<K,
V>> entries)
-            throws IgniteCheckedException {
-            cache = cache.flagsOn(SKIP_STORE);
-
-            for (Map.Entry<K, V> e : entries)
-                cache.put(e.getKey(), e.getValue());
-        }
-    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9288ffe/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheStoreManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheStoreManager.java
b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheStoreManager.java
index 60e2425..5bf2404 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheStoreManager.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheStoreManager.java
@@ -579,26 +579,7 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K,
V> {
         }
         else {
             if (store != null) {
-                Collection<Cache.Entry<? extends K, ?>> entries;
-
-                if (convertPortable) {
-                    entries = new ArrayList<>(map.size());
-
-                    for (Map.Entry<K, IgniteBiTuple<V, GridCacheVersion>> e :
map.entrySet()) {
-                        K k = e.getKey();
-                        Object v = locStore ? e.getValue() : e.getValue().get1();
-
-                        entries.add(new CacheEntryImpl<>(
-                            (K)cctx.unwrapPortableIfNeeded(k, false),
-                            (V)cctx.unwrapPortableIfNeeded(v, false)));
-                    }
-                }
-                else {
-                    entries = new ArrayList<>(map.size());
-
-                    for (Map.Entry<K, IgniteBiTuple<V, GridCacheVersion>> e :
map.entrySet())
-                        entries.add(new CacheEntryImpl<>(e.getKey(), locStore ? e.getValue()
: e.getValue().get1()));
-                }
+                EntriesView entries = new EntriesView(map);
 
                 if (log.isDebugEnabled())
                     log.debug("Storing values in cache store [entries=" + entries + ']');
@@ -894,4 +875,244 @@ public class GridCacheStoreManager<K, V> extends GridCacheManagerAdapter<K,
V> {
             return ses0 != null ? ses0.cacheName() : null;
         }
     }
+
+    /**
+     *
+     */
+    @SuppressWarnings("unchecked")
+    private class EntriesView extends AbstractCollection<Cache.Entry<? extends K, ?>>
{
+        /** */
+        private final Map<K, IgniteBiTuple<V, GridCacheVersion>> map;
+
+        /** */
+        private Set<K> rmvd;
+
+        /** */
+        private boolean cleared;
+
+        /**
+         * @param map Map.
+         */
+        private EntriesView(Map<K, IgniteBiTuple<V, GridCacheVersion>> map) {
+            assert map != null;
+
+            this.map = map;
+        }
+
+        /** {@inheritDoc} */
+        @Override public int size() {
+            return cleared ? 0 : (map.size() - (rmvd != null ? rmvd.size() : 0));
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean isEmpty() {
+            return cleared || !iterator().hasNext();
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean contains(Object o) {
+            if (cleared || !(o instanceof Cache.Entry))
+                return false;
+
+            Cache.Entry<? extends K, ?> e = (Cache.Entry<? extends K, ?>)o;
+
+            return map.containsKey(e.getKey());
+        }
+
+        /** {@inheritDoc} */
+        @NotNull @Override public Iterator<Cache.Entry<? extends K, ?>> iterator()
{
+            if (cleared)
+                return F.emptyIterator();
+
+            final Iterator<Map.Entry<K, IgniteBiTuple<V, GridCacheVersion>>>
it0 = map.entrySet().iterator();
+
+            return new Iterator<Cache.Entry<? extends K, ?>>() {
+                /** */
+                private Cache.Entry<? extends K, ?> cur;
+
+                /** */
+                private Cache.Entry<? extends K, ?> next;
+
+                /**
+                 *
+                 */
+                {
+                    checkNext();
+                }
+
+                /**
+                 *
+                 */
+                private void checkNext() {
+                    while (it0.hasNext()) {
+                        Map.Entry<K, IgniteBiTuple<V, GridCacheVersion>> e =
it0.next();
+
+                        K k = e.getKey();
+
+                        if (rmvd != null && rmvd.contains(k))
+                            continue;
+
+                        Object v = locStore ? e.getValue() : e.getValue().get1();
+
+                        if (convertPortable) {
+                            k = (K)cctx.unwrapPortableIfNeeded(k, false);
+                            v = cctx.unwrapPortableIfNeeded(v, false);
+                        }
+
+                        next = new CacheEntryImpl<>(k, v);
+
+                        break;
+                    }
+                }
+
+                @Override public boolean hasNext() {
+                    return next != null;
+                }
+
+                @Override public Cache.Entry<? extends K, ?> next() {
+                    if (next == null)
+                        throw new NoSuchElementException();
+
+                    cur = next;
+
+                    next = null;
+
+                    checkNext();
+
+                    return cur;
+                }
+
+                @Override public void remove() {
+                    if (cur == null)
+                        throw new IllegalStateException();
+
+                    addRemoved(cur);
+
+                    cur = null;
+                }
+            };
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean add(Cache.Entry<? extends K, ?> entry) {
+            throw new UnsupportedOperationException();
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean addAll(Collection<? extends Cache.Entry<? extends
K, ?>> col) {
+            throw new UnsupportedOperationException();
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean remove(Object o) {
+            if (cleared || !(o instanceof Cache.Entry))
+                return false;
+
+            Cache.Entry<? extends K, ?> e = (Cache.Entry<? extends K, ?>)o;
+
+            if (rmvd != null && rmvd.contains(e.getKey()))
+                return false;
+
+            if (mapContains(e)) {
+                addRemoved(e);
+
+                return true;
+            }
+
+            return false;
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean containsAll(Collection<?> col) {
+            if (cleared)
+                return false;
+
+            for (Object o : col) {
+                if (contains(o))
+                    return false;
+            }
+
+            return true;
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean removeAll(Collection<?> col) {
+            if (cleared)
+                return false;
+
+            boolean modified = false;
+
+            for (Object o : col) {
+                 if (remove(o))
+                     modified = true;
+            }
+
+            return modified;
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean retainAll(Collection<?> col) {
+            if (cleared)
+                return false;
+
+            boolean modified = false;
+
+            for (Cache.Entry<? extends K, ?> e : this) {
+                if (!col.contains(e)) {
+                    addRemoved(e);
+
+                    modified = true;
+                }
+            }
+
+            return modified;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void clear() {
+            cleared = true;
+        }
+
+        /**
+         * @param e Entry.
+         */
+        private void addRemoved(Cache.Entry<? extends K, ?> e) {
+            if (rmvd == null)
+                rmvd = new HashSet<>();
+
+            rmvd.add(e.getKey());
+        }
+
+        /**
+         * @param e Entry.
+         * @return {@code True} if original map contains entry.
+         */
+        private boolean mapContains(Cache.Entry<? extends K, ?> e) {
+            K key = (K)(convertPortable ? cctx.marshalToPortable(e.getKey()) : e.getKey());
+
+            return map.containsKey(key);
+
+        }
+
+        /** {@inheritDoc} */
+        public String toString() {
+            Iterator<Cache.Entry<? extends K, ?>> it = iterator();
+
+            if (!it.hasNext())
+                return "[]";
+
+            SB sb = new SB("[");
+
+            while (true) {
+                Cache.Entry<? extends K, ?> e = it.next();
+
+                sb.a(e.toString());
+
+                if (!it.hasNext())
+                    return sb.a(']').toString();
+
+                sb.a(", ");
+            }
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9288ffe/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheLoaderWriterAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheLoaderWriterAbstractTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheLoaderWriterAbstractTest.java
index 17282f6..64e10c7 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheLoaderWriterAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheLoaderWriterAbstractTest.java
@@ -325,8 +325,19 @@ public abstract class IgniteCacheLoaderWriterAbstractTest extends IgniteCacheAbs
 
             writerCallCnt.incrementAndGet();
 
-            for (Cache.Entry<? extends Integer, ? extends Integer> e : entries)
+            Iterator<Cache.Entry<? extends Integer, ? extends Integer>> it =
entries.iterator();
+
+            while (it.hasNext()) {
+                Cache.Entry<? extends Integer, ? extends Integer> e = it.next();
+
                 storeMap.put(e.getKey(), e.getValue());
+
+                it.remove();
+            }
+
+            assertTrue(entries.isEmpty());
+
+            assertEquals(0, entries.size());
         }
 
         /** {@inheritDoc} */
@@ -352,8 +363,15 @@ public abstract class IgniteCacheLoaderWriterAbstractTest extends IgniteCacheAbs
 
             writerCallCnt.incrementAndGet();
 
-            for (Object key : keys)
+            Iterator<?> it = keys.iterator();
+
+            while (it.hasNext()) {
+                Object key = it.next();
+
                 storeMap.remove(key);
+
+                it.remove();
+            }
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b9288ffe/modules/core/src/test/java/org/gridgain/grid/kernal/processors/dataload/GridDataLoaderProcessorSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/dataload/GridDataLoaderProcessorSelfTest.java
b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/dataload/GridDataLoaderProcessorSelfTest.java
index aeb86e6..9d415c0 100644
--- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/dataload/GridDataLoaderProcessorSelfTest.java
+++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/dataload/GridDataLoaderProcessorSelfTest.java
@@ -19,21 +19,22 @@ package org.gridgain.grid.kernal.processors.dataload;
 
 import org.apache.ignite.*;
 import org.apache.ignite.cache.*;
+import org.apache.ignite.cache.store.*;
 import org.apache.ignite.configuration.*;
 import org.apache.ignite.events.*;
 import org.apache.ignite.lang.*;
 import org.apache.ignite.marshaller.optimized.*;
-import org.apache.ignite.transactions.*;
 import org.gridgain.grid.cache.*;
 import org.gridgain.grid.cache.eviction.fifo.*;
 import org.apache.ignite.spi.discovery.tcp.*;
 import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
 import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
-import org.gridgain.grid.cache.store.*;
 import org.gridgain.grid.util.typedef.internal.*;
 import org.gridgain.testframework.junits.common.*;
 import org.jetbrains.annotations.*;
 
+import javax.cache.*;
+import javax.cache.configuration.*;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.*;
@@ -78,7 +79,7 @@ public class GridDataLoaderProcessorSelfTest extends GridCommonAbstractTest
{
     }
 
     /** {@inheritDoc} */
-    @SuppressWarnings({"IfMayBeConditional"})
+    @SuppressWarnings({"IfMayBeConditional", "unchecked"})
     @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception
{
         IgniteConfiguration cfg = super.getConfiguration(gridName);
 
@@ -105,7 +106,11 @@ public class GridDataLoaderProcessorSelfTest extends GridCommonAbstractTest
{
             cc.setEvictSynchronized(false);
             cc.setEvictNearSynchronized(false);
 
-            cc.setStore(store);
+            if (store != null) {
+                cc.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(store));
+                cc.setReadThrough(true);
+                cc.setWriteThrough(true);
+            }
 
             cfg.setCacheConfiguration(cc);
         }
@@ -780,7 +785,7 @@ public class GridDataLoaderProcessorSelfTest extends GridCommonAbstractTest
{
                 storeMap.put(i, i);
 
             try (IgniteDataLoader<Object, Object> ldr = ignite.dataLoader(null)) {
-                ldr.skipStore(false);
+                assertFalse(ldr.skipStore());
 
                 for (int i = 0; i < 1000; i++)
                     ldr.removeData(i);
@@ -860,19 +865,19 @@ public class GridDataLoaderProcessorSelfTest extends GridCommonAbstractTest
{
     /**
      *
      */
-    private class TestStore extends GridCacheStoreAdapter<Object, Object> {
+    private class TestStore extends CacheStoreAdapter<Object, Object> {
         /** {@inheritDoc} */
-        @Nullable @Override public Object load(@Nullable IgniteTx tx, Object key) throws
IgniteCheckedException {
+        @Nullable @Override public Object load(Object key) {
             return storeMap.get(key);
         }
 
         /** {@inheritDoc} */
-        @Override public void put(@Nullable IgniteTx tx, Object key, Object val) throws IgniteCheckedException
{
-            storeMap.put(key, val);
+        @Override public void write(Cache.Entry<?, ?> entry) {
+            storeMap.put(entry.getKey(), entry.getValue());
         }
 
         /** {@inheritDoc} */
-        @Override public void remove(@Nullable IgniteTx tx, Object key) throws IgniteCheckedException
{
+        @Override public void delete(Object key) {
             storeMap.remove(key);
         }
     }


Mime
View raw message