http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/apache/ignite/spi/swapspace/file/GridFileSwapCompactionSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/spi/swapspace/file/GridFileSwapCompactionSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/swapspace/file/GridFileSwapCompactionSelfTest.java new file mode 100644 index 0000000..5191085 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/spi/swapspace/file/GridFileSwapCompactionSelfTest.java @@ -0,0 +1,119 @@ +/* @java.file.header */ + +/* _________ _____ __________________ _____ + * __ ____/___________(_)______ /__ ____/______ ____(_)_______ + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ + */ + +package org.apache.ignite.spi.swapspace.file; + +import org.gridgain.grid.util.typedef.*; +import org.gridgain.testframework.junits.common.*; + +import java.io.File; +import java.nio.ByteBuffer; +import java.util.*; + +/** + * Test for {@link FileSwapSpaceSpi}. + */ +public class GridFileSwapCompactionSelfTest extends GridCommonAbstractTest { + /** + * @throws Exception If failed. + */ + public void testCompact() throws Exception { + File file = new File(UUID.randomUUID().toString()); + + X.println("file: " + file.getPath()); + + FileSwapSpaceSpi.SwapFile f = new FileSwapSpaceSpi.SwapFile(file, 8); + + Random rnd = new Random(); + + ArrayList<FileSwapSpaceSpi.SwapValue> arr = new ArrayList<>(); + + int size = 0; + + for (int a = 0; a < 100; a++) { + FileSwapSpaceSpi.SwapValue[] vals = new FileSwapSpaceSpi.SwapValue[1 + rnd.nextInt(10)]; + + int size0 = 0; + + for (int i = 0; i < vals.length; i++) { + byte[] bytes = new byte[1 + rnd.nextInt(49)]; + + rnd.nextBytes(bytes); + + size0 += bytes.length; + + vals[i] = new FileSwapSpaceSpi.SwapValue(bytes); + + arr.add(vals[i]); + } + + f.write(new FileSwapSpaceSpi.SwapValues(vals, size0), 1); + + size += size0; + + assertEquals(f.length(), size); + assertEquals(file.length(), size); + } + + int i = 0; + + for (FileSwapSpaceSpi.SwapValue val : arr) + assertEquals(val.idx(), ++i); + + i = 0; + + for (int cnt = arr.size() / 2; i < cnt; i++) { + + FileSwapSpaceSpi.SwapValue v = arr.remove(rnd.nextInt(arr.size())); + + assertTrue(f.tryRemove(v.idx(), v)); + } + + int hash0 = 0; + + for (FileSwapSpaceSpi.SwapValue val : arr) + hash0 += Arrays.hashCode(val.readValue(f.readCh)); + + ArrayList<T2<ByteBuffer, ArrayDeque<FileSwapSpaceSpi.SwapValue>>> bufs = new ArrayList(); + + for (;;) { + ArrayDeque<FileSwapSpaceSpi.SwapValue> que = new ArrayDeque<>(); + + ByteBuffer buf = f.compact(que, 1024); + + if (buf == null) + break; + + bufs.add(new T2(buf, que)); + } + + f.delete(); + + int hash1 = 0; + + for (FileSwapSpaceSpi.SwapValue val : arr) + hash1 += Arrays.hashCode(val.value(null)); + + assertEquals(hash0, hash1); + + File file0 = new File(UUID.randomUUID().toString()); + + FileSwapSpaceSpi.SwapFile f0 = new FileSwapSpaceSpi.SwapFile(file0, 8); + + for (T2<ByteBuffer, ArrayDeque<FileSwapSpaceSpi.SwapValue>> t : bufs) + f0.write(t.get2(), t.get1(), 1); + + int hash2 = 0; + + for (FileSwapSpaceSpi.SwapValue val : arr) + hash2 += Arrays.hashCode(val.readValue(f0.readCh)); + + assertEquals(hash2, hash1); + } +}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/apache/ignite/spi/swapspace/file/GridFileSwapSpaceSpiSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/spi/swapspace/file/GridFileSwapSpaceSpiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/swapspace/file/GridFileSwapSpaceSpiSelfTest.java new file mode 100644 index 0000000..566759e --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/spi/swapspace/file/GridFileSwapSpaceSpiSelfTest.java @@ -0,0 +1,345 @@ +/* @java.file.header */ + +/* _________ _____ __________________ _____ + * __ ____/___________(_)______ /__ ____/______ ____(_)_______ + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ + */ + +package org.apache.ignite.spi.swapspace.file; + +import org.apache.ignite.lang.*; +import org.apache.ignite.spi.*; +import org.apache.ignite.spi.swapspace.*; +import org.gridgain.grid.util.typedef.*; +import org.gridgain.grid.util.typedef.internal.*; +import org.jdk8.backport.*; +import org.jetbrains.annotations.*; + +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; + +/** + * Test for {@link FileSwapSpaceSpi}. + */ +public class GridFileSwapSpaceSpiSelfTest extends GridSwapSpaceSpiAbstractSelfTest { + /** {@inheritDoc} */ + @Override protected SwapSpaceSpi spi() { + FileSwapSpaceSpi s = new FileSwapSpaceSpi(); + + s.setMaximumSparsity(0.05f); + s.setWriteBufferSize(8 * 1024); + + return s; + } + + /** + * Tests if SPI works correctly with multithreaded writes. + * + * @throws Exception If failed. + */ + public void testMultithreadedWrite() throws Exception { + final AtomicLong valCntr = new AtomicLong(); + + final SwapKey key = new SwapKey("key"); + + final CountDownLatch wLatch = new CountDownLatch(1); + + final AtomicBoolean done = new AtomicBoolean(); + + IgniteFuture<?> wFut = multithreadedAsync(new Callable<Object>() { + @Nullable @Override public Object call() throws Exception { + while (!done.get()) { + long val = valCntr.incrementAndGet(); + + spi.store(null, key, Long.toString(val).getBytes(), context()); + + if (val == 1) + wLatch.countDown(); + } + + return null; + } + }, 8); + + wLatch.await(); + + IgniteFuture<?> rFut = multithreadedAsync(new Callable<Object>() { + @Nullable @Override public Object call() throws Exception { + while (valCntr.get() < 1000) { + byte[] val = spi.read(null, key, context()); + + assertNotNull(val); + + long lval = Long.parseLong(new String(val)); + + assertTrue(lval <= valCntr.get()); + } + + return null; + } + }, 8); + + rFut.get(); + + done.set(true); + + wFut.get(); + } + + /** + * @param i Integer. + * @return Swap key. + */ + private SwapKey key(int i) { + return new SwapKey(i, i % 11, U.intToBytes(i)); + } + + /** + * @throws Exception If failed. + */ + public void testMultithreadedOperations() throws Exception { + final ConcurrentHashMap8<SwapKey, byte[]> map = new ConcurrentHashMap8<>(); + + Random rnd = new Random(); + + final int keys = 25000; + + int hash0 = 0; + + final int minValSize = 5; + final int maxValSize = 9000; // More than write buffer size. + + for (int i = 0; i < keys; i++) { + byte[] val = new byte[minValSize + rnd.nextInt(maxValSize - minValSize)]; + + rnd.nextBytes(val); + + hash0 += i * Arrays.hashCode(val); + + assertNull(map.put(key(i), val)); + } + + assertEquals(keys, map.size()); + + for (int i = 0; i < keys; i++) + assertTrue(map.containsKey(key(i))); + + final String space = "test_space"; + + final AtomicBoolean fin = new AtomicBoolean(); + + final IgniteFuture<?> fut = multithreadedAsync(new Callable<Object>() { + @Override public Object call() throws Exception { + Random rnd = new Random(); + + while (!fin.get()) { + final SwapKey key = key(rnd.nextInt(keys)); + + switch(rnd.nextInt(13)) { + case 0: // store + byte[] val = map.remove(key); + + if (val != null) + spi.store(space, key, val, context()); + + break; + + case 1: // remove + spi.remove(space, key, new CIX1<byte[]>() { + @Override public void applyx(byte[] bytes) { + if (bytes != null) + assertNull(map.putIfAbsent(key, bytes)); + } + }, context()); + + break; + + case 2: // read + for (;;) { + val = spi.read(space, key, context()); + + if (val != null) + break; + + val = map.get(key); + + if (val != null) + break; + } + + break; + + case 3: // storeAll + case 4: + case 9: + Map<SwapKey, byte[]> m = new HashMap<>(); + + int cnt = 1 + rnd.nextInt(25); + + for (int i = 0; i < cnt; i++) { + SwapKey k = key(rnd.nextInt(keys)); + + val = map.remove(k); + + if (val != null) + assertNull(m.put(k, val)); + } + + if (m.isEmpty()) + break; + + spi.storeAll(space, m, context()); + + break; + + case 5: // readAll + HashSet<SwapKey> s = new HashSet<>(); + + cnt = 1 + rnd.nextInt(25); + + for (int i = 0; i < cnt; i++) { + SwapKey k = key(rnd.nextInt(keys)); + + val = map.get(k); + + if (val == null) + s.add(k); + } + + while (!s.isEmpty()) { + m = spi.readAll(space, s, context()); + + s.removeAll(m.keySet()); + + Iterator<SwapKey> iter = s.iterator(); + + while (iter.hasNext()) { + SwapKey k = iter.next(); + + if (map.containsKey(k)) + iter.remove(); + } + } + + break; + + case 6: // iterateKeys + IgniteSpiCloseableIterator<Integer> kIt = spi.keyIterator(space, context()); + + if (kIt == null) + break; + + while (kIt.hasNext()) + assertNotNull(kIt.next()); + + kIt.close(); + + break; + + case 7: // iterate + IgniteSpiCloseableIterator<Map.Entry<byte[], byte[]>> iter = spi.rawIterator(space); + + if (iter == null) + break; + + while (iter.hasNext()) { + Map.Entry<byte[], byte[]> entry = iter.next(); + + assertEquals(4, entry.getKey().length); + + byte[] v = entry.getValue(); + + assertTrue(v.length >= minValSize && v.length < maxValSize); + } + + iter.close(); + + break; + + case 8: // iterate partitions + iter = spi.rawIterator(space, rnd.nextInt(11)); + + if (iter == null) + break; + + while ( iter.hasNext()) { + Map.Entry<byte[], byte[]> entry = iter.next(); + + assertEquals(4, entry.getKey().length); + + byte[] v = entry.getValue(); + + assertTrue(v.length >= minValSize && v.length < maxValSize); + } + + iter.close(); + + break; + + default: // removeAll + s = new HashSet<>(); + + cnt = 1 + rnd.nextInt(25); + + for (int i = 0; i < cnt; i++) { + SwapKey k = key(rnd.nextInt(keys)); + + val = map.get(k); + + if (val == null) + s.add(k); + } + + if (s.isEmpty()) + break; + + spi.removeAll(space, s, new IgniteBiInClosure<SwapKey, byte[]>() { + @Override public void apply(SwapKey k, byte[] bytes) { + if (bytes != null) + assertNull(map.putIfAbsent(k, bytes)); + } + }, context()); + + break; + } + } + + return null; + } + }, 39); + + Thread.sleep(60000); + + System.out.println("stopping"); + + fin.set(true); + + fut.get(); + + assertEquals(keys, map.size() + spi.count(space)); + + int hash1 = 0; + + int cnt = 0; + + IgniteSpiCloseableIterator<Map.Entry<byte[], byte[]>> iter = spi.rawIterator(space); + + while (iter.hasNext()) { + Map.Entry<byte[], byte[]> entry = iter.next(); + + hash1 += U.bytesToInt(entry.getKey(), 0) * Arrays.hashCode(entry.getValue()); + + cnt++; + } + + assertEquals(cnt, spi.count(space)); + + for (Map.Entry<SwapKey, byte[]> entry : map.entrySet()) + hash1 += (Integer)entry.getKey().key() * Arrays.hashCode(entry.getValue()); + + assertEquals(hash0, hash1); + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/apache/ignite/spi/swapspace/inmemory/GridTestSwapSpaceSpi.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/spi/swapspace/inmemory/GridTestSwapSpaceSpi.java b/modules/core/src/test/java/org/apache/ignite/spi/swapspace/inmemory/GridTestSwapSpaceSpi.java new file mode 100644 index 0000000..e6ea5b6 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/spi/swapspace/inmemory/GridTestSwapSpaceSpi.java @@ -0,0 +1,444 @@ +/* @java.file.header */ + +/* _________ _____ __________________ _____ + * __ ____/___________(_)______ /__ ____/______ ____(_)_______ + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ + */ + +package org.apache.ignite.spi.swapspace.inmemory; + +import org.apache.ignite.lang.*; +import org.apache.ignite.spi.*; +import org.apache.ignite.spi.swapspace.*; +import org.gridgain.grid.util.typedef.*; +import org.jdk8.backport.*; +import org.jetbrains.annotations.*; + +import java.util.*; +import java.util.concurrent.*; + +import static org.apache.ignite.events.IgniteEventType.*; + +/** + * Test swap space SPI that stores values in map. + */ +@IgniteSpiMultipleInstancesSupport(true) +public class GridTestSwapSpaceSpi extends IgniteSpiAdapter implements SwapSpaceSpi { + /** Listener. */ + private SwapSpaceSpiListener lsnr; + + /** Spaces map. */ + private ConcurrentMap<String, Space> spaces = new ConcurrentHashMap8<>(); + + /** {@inheritDoc} */ + @Override public void spiStart(@Nullable String gridName) throws IgniteSpiException { + // No-op. + } + + /** {@inheritDoc} */ + @Override public void spiStop() throws IgniteSpiException { + // No-op. + } + + /** {@inheritDoc} */ + @Override public void clear(@Nullable String spaceName) throws IgniteSpiException { + Space space = space(spaceName); + + if (space != null) + space.clear(); + } + + /** {@inheritDoc} */ + @Override public long size(@Nullable String spaceName) throws IgniteSpiException { + Space space = space(spaceName); + + return space != null ? space.size() : 0; + } + + /** {@inheritDoc} */ + @Override public long count(@Nullable String spaceName) throws IgniteSpiException { + Space space = space(spaceName); + + return space != null ? space.count() : 0; + } + + /** {@inheritDoc} */ + @Override public byte[] read(@Nullable String spaceName, SwapKey key, SwapContext ctx) + throws IgniteSpiException { + Space space = space(spaceName); + + return space != null ? space.read(key) : null; + } + + /** {@inheritDoc} */ + @Override public Map<SwapKey, byte[]> readAll(@Nullable String spaceName, Iterable<SwapKey> keys, + SwapContext ctx) throws IgniteSpiException { + Space space = space(spaceName); + + return space != null ? space.readAll(keys) : Collections.<SwapKey, byte[]>emptyMap(); + } + + /** {@inheritDoc} */ + @Override public void remove(@Nullable String spaceName, SwapKey key, @Nullable IgniteInClosure<byte[]> c, + SwapContext ctx) throws IgniteSpiException { + Space space = space(spaceName); + + if (space != null) + space.remove(key, c); + } + + /** {@inheritDoc} */ + @Override public void removeAll(@Nullable String spaceName, Collection<SwapKey> keys, + @Nullable IgniteBiInClosure<SwapKey, byte[]> c, SwapContext ctx) throws IgniteSpiException { + Space space = space(spaceName); + + if (space != null) + space.removeAll(keys, c); + } + + /** {@inheritDoc} */ + @Override public void store(@Nullable String spaceName, SwapKey key, @Nullable byte[] val, SwapContext ctx) + throws IgniteSpiException { + ensureSpace(spaceName).store(key, val); + } + + /** {@inheritDoc} */ + @Override public void storeAll(@Nullable String spaceName, Map<SwapKey, byte[]> pairs, SwapContext ctx) + throws IgniteSpiException { + ensureSpace(spaceName).storeAll(pairs); + } + + /** {@inheritDoc} */ + @Override public void setListener(@Nullable SwapSpaceSpiListener evictLsnr) { + lsnr = evictLsnr; + } + + /** {@inheritDoc} */ + @Override public Collection<Integer> partitions(@Nullable String spaceName) throws IgniteSpiException { + Space space = space(spaceName); + + return space != null ? space.partitions() : Collections.<Integer>emptyList(); + } + + /** {@inheritDoc} */ + @Override public <K> IgniteSpiCloseableIterator<K> keyIterator(@Nullable String spaceName, SwapContext ctx) + throws IgniteSpiException { + return ensureSpace(spaceName).keyIterator(); + } + + /** {@inheritDoc} */ + @Override public IgniteSpiCloseableIterator<Map.Entry<byte[], byte[]>> rawIterator(@Nullable String spaceName) + throws IgniteSpiException { + return ensureSpace(spaceName).rawIterator(); + } + + /** {@inheritDoc} */ + @Override public IgniteSpiCloseableIterator<Map.Entry<byte[], byte[]>> rawIterator(@Nullable String spaceName, int part) + throws IgniteSpiException { + return ensureSpace(spaceName).rawIterator(part); + } + + /** + * @param spaceName Space name. + * @return Space object. + */ + @Nullable private Space space(String spaceName) { + return spaces.get(spaceName); + } + + /** + * Gets space, creates if does not exist. + * + * @param spaceName Space name. + * @return Space. + */ + private Space ensureSpace(String spaceName) { + Space space = spaces.get(spaceName); + + if (space == null) + space = F.addIfAbsent(spaces, spaceName, new Space(spaceName)); + + return space; + } + + private void fireEvent(int evtType, String spaceName, @Nullable byte[] key) { + SwapSpaceSpiListener lsnr0 = lsnr; + + if (lsnr0 != null) + lsnr0.onSwapEvent(evtType, spaceName, key); + } + + private class Space { + /** Data storage. */ + private ConcurrentMap<SwapKey, byte[]> data = new ConcurrentHashMap8<>(); + + private final String name; + + /** + * @param name Space name. + */ + private Space(String name) { + this.name = name; + } + + /** + * Clears space. + */ + public void clear() { + data.clear(); + + fireEvent(EVT_SWAP_SPACE_CLEARED, name, null); + } + + /** + * @return Space size. + */ + public long size() { + return data.size(); + } + + /** + * @return Space size. + */ + public long count() { + return data.size(); + } + + /** + * @param key Key to read. + * @return Read bytes. + */ + public byte[] read(SwapKey key) { + byte[] bytes = data.get(key); + + fireEvent(EVT_SWAP_SPACE_DATA_READ, name, key.keyBytes()); + + return bytes; + } + + /** + * @param keys Keys to read. + * @return Read keys. + */ + public Map<SwapKey, byte[]> readAll(Iterable<SwapKey> keys) { + Map<SwapKey, byte[]> res = new HashMap<>(); + + for (SwapKey key : keys) { + byte[] val = data.get(key); + + if (val != null) { + res.put(key, val); + + fireEvent(EVT_SWAP_SPACE_DATA_READ, name, key.keyBytes()); + } + } + + return res; + } + + /** + * @param key Key to remove. + * @param c Closure. + */ + public void remove(SwapKey key, IgniteInClosure<byte[]> c) { + byte[] val = data.remove(key); + + if (val != null) { + c.apply(val); + + fireEvent(EVT_SWAP_SPACE_DATA_REMOVED, name, key.keyBytes()); + } + } + + /** + * @param keys Keys to remove. + * @param c Closure to apply for removed values. + */ + public void removeAll(Iterable<SwapKey> keys, IgniteBiInClosure<SwapKey, byte[]> c) { + for (SwapKey key : keys) { + byte[] val = data.remove(key); + + if (val != null) { + c.apply(key, val); + + fireEvent(EVT_SWAP_SPACE_DATA_REMOVED, name, key.keyBytes()); + } + } + } + + /** + * @param key Key to store. + * @param val Value to store. + */ + public void store(SwapKey key, byte[] val) { + if (val != null) { + data.put(key, val); + + fireEvent(EVT_SWAP_SPACE_DATA_STORED, name, key.keyBytes()); + } + else { + val = data.remove(key); + + if (val != null) + fireEvent(EVT_SWAP_SPACE_DATA_REMOVED, name, key.keyBytes()); + } + } + + /** + * @param pairs Values to store. + */ + public void storeAll(Map<SwapKey, byte[]> pairs) { + for (Map.Entry<SwapKey, byte[]> entry : pairs.entrySet()) { + SwapKey key = entry.getKey(); + byte[] val = entry.getValue(); + + store(key, val); + } + } + + /** + * @return Partitions in space. + */ + public Collection<Integer> partitions() { + Collection<Integer> parts = new HashSet<>(); + + for (SwapKey key : data.keySet()) + parts.add(key.partition()); + + return parts; + } + + public <K> IgniteSpiCloseableIterator<K> keyIterator() { + final Iterator<SwapKey> it = data.keySet().iterator(); + + return new IgniteSpiCloseableIterator<K>() { + @Override public void close() { + // No-op. + } + + @Override public boolean hasNext() { + return it.hasNext(); + } + + @Override public K next() { + SwapKey next = it.next(); + + return (K)next.key(); + } + + @Override public void remove() { + it.remove(); + } + }; + } + + public IgniteSpiCloseableIterator<Map.Entry<byte[], byte[]>> rawIterator() { + final Iterator<Map.Entry<SwapKey, byte[]>> it = data.entrySet().iterator(); + + return new IgniteSpiCloseableIterator<Map.Entry<byte[], byte[]>>() { + @Override public void close() { + // No-op. + } + + @Override public boolean hasNext() { + return it.hasNext(); + } + + @Override public Map.Entry<byte[], byte[]> next() { + final Map.Entry<SwapKey, byte[]> next = it.next(); + + return new Map.Entry<byte[], byte[]>() { + @Override public byte[] getKey() { + return next.getKey().keyBytes(); + } + + @Override public byte[] getValue() { + return next.getValue(); + } + + @Override public byte[] setValue(byte[] val) { + return data.put(next.getKey(), val); + } + }; + } + + @Override public void remove() { + it.remove(); + } + }; + } + + public IgniteSpiCloseableIterator<Map.Entry<byte[], byte[]>> rawIterator(final int part) { + final Iterator<Map.Entry<SwapKey, byte[]>> it = data.entrySet().iterator(); + + return new IgniteSpiCloseableIterator<Map.Entry<byte[], byte[]>>() { + /** Next entry in this iterator. */ + private Map.Entry<SwapKey, byte[]> next; + + private Map.Entry<SwapKey, byte[]> cur; + + { + advance(); + } + + @Override public void close() { + // No-op. + } + + @Override public boolean hasNext() { + return next != null; + } + + @Override public Map.Entry<byte[], byte[]> next() { + if (next == null) + throw new NoSuchElementException(); + + final Map.Entry<SwapKey, byte[]> ret = next; + + cur = ret; + + advance(); + + return new Map.Entry<byte[], byte[]>() { + @Override public byte[] getKey() { + return ret.getKey().keyBytes(); + } + + @Override public byte[] getValue() { + return ret.getValue(); + } + + @Override public byte[] setValue(byte[] val) { + return data.put(ret.getKey(), val); + } + }; + } + + @Override public void remove() { + if (cur == null) + throw new IllegalStateException(); + + data.remove(cur.getKey(), cur.getValue()); + } + + private void advance() { + while (it.hasNext()) { + Map.Entry<SwapKey, byte[]> entry = it.next(); + + if(entry.getKey().partition() == part) { + cur = next; + + next = entry; + + return; + } + } + + next = null; + } + }; + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/apache/ignite/spi/swapspace/noop/GridNoopSwapSpaceSpiSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/spi/swapspace/noop/GridNoopSwapSpaceSpiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/swapspace/noop/GridNoopSwapSpaceSpiSelfTest.java new file mode 100644 index 0000000..654b305 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/spi/swapspace/noop/GridNoopSwapSpaceSpiSelfTest.java @@ -0,0 +1,53 @@ +/* @java.file.header */ + +/* _________ _____ __________________ _____ + * __ ____/___________(_)______ /__ ____/______ ____(_)_______ + * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ + */ + +package org.apache.ignite.spi.swapspace.noop; + +import org.apache.ignite.*; +import org.apache.ignite.configuration.*; +import org.apache.ignite.spi.discovery.tcp.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*; +import org.apache.ignite.spi.swapspace.*; +import org.gridgain.testframework.junits.common.*; + +/** + * Tests for "noop" realization of {@link org.apache.ignite.spi.swapspace.SwapSpaceSpi}. + */ +public class GridNoopSwapSpaceSpiSelfTest extends GridCommonAbstractTest { + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(gridName); + + TcpDiscoverySpi disco = new TcpDiscoverySpi(); + + disco.setIpFinder(new TcpDiscoveryVmIpFinder(true)); + + cfg.setDiscoverySpi(disco); + + return cfg; + } + + /** + * @throws Exception If test failed. + */ + public void testWithoutCacheUseNoopSwapSapce() throws Exception { + try { + Ignite ignite = startGrid(1); + + SwapSpaceSpi spi = ignite.configuration().getSwapSpaceSpi(); + + assertNotNull(spi); + + assertTrue(spi instanceof NoopSwapSpaceSpi); + } + finally { + stopAllGrids(); + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/apache/ignite/spi/swapspace/package.html ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/spi/swapspace/package.html b/modules/core/src/test/java/org/apache/ignite/spi/swapspace/package.html new file mode 100644 index 0000000..5cad80a --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/spi/swapspace/package.html @@ -0,0 +1,15 @@ +<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd";> +<!-- + @html.file.header + _________ _____ __________________ _____ + __ ____/___________(_)______ /__ ____/______ ____(_)_______ + _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ + / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / + \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ +--> +<html> +<body> + <!-- Package description. --> + Contains internal tests or test related classes and interfaces. +</body> +</html> http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/kernal/managers/GridManagerStopSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/managers/GridManagerStopSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/managers/GridManagerStopSelfTest.java index aa89e2a..28863d5 100644 --- a/modules/core/src/test/java/org/gridgain/grid/kernal/managers/GridManagerStopSelfTest.java +++ b/modules/core/src/test/java/org/gridgain/grid/kernal/managers/GridManagerStopSelfTest.java @@ -36,8 +36,8 @@ import org.apache.ignite.spi.eventstorage.*; import org.apache.ignite.spi.eventstorage.memory.*; import org.apache.ignite.spi.failover.always.*; import org.apache.ignite.spi.loadbalancing.roundrobin.*; -import org.gridgain.grid.spi.swapspace.*; -import org.gridgain.grid.spi.swapspace.file.*; +import org.apache.ignite.spi.swapspace.*; +import org.apache.ignite.spi.swapspace.file.*; import org.gridgain.testframework.junits.*; import org.gridgain.testframework.junits.common.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/kernal/managers/swapspace/GridSwapSpaceManagerSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/managers/swapspace/GridSwapSpaceManagerSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/managers/swapspace/GridSwapSpaceManagerSelfTest.java index db45129..5d404ea 100644 --- a/modules/core/src/test/java/org/gridgain/grid/kernal/managers/swapspace/GridSwapSpaceManagerSelfTest.java +++ b/modules/core/src/test/java/org/gridgain/grid/kernal/managers/swapspace/GridSwapSpaceManagerSelfTest.java @@ -14,8 +14,8 @@ import org.apache.ignite.configuration.*; import org.apache.ignite.events.*; import org.apache.ignite.lang.*; import org.gridgain.grid.kernal.*; -import org.gridgain.grid.spi.swapspace.*; -import org.gridgain.grid.spi.swapspace.file.*; +import org.apache.ignite.spi.swapspace.*; +import org.apache.ignite.spi.swapspace.file.*; import org.gridgain.grid.util.typedef.*; import org.gridgain.grid.util.lang.*; import org.gridgain.testframework.junits.common.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractFullApiSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractFullApiSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractFullApiSelfTest.java index c246fbb..fe7436a 100644 --- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractFullApiSelfTest.java +++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractFullApiSelfTest.java @@ -16,7 +16,7 @@ import org.apache.ignite.events.*; import org.apache.ignite.lang.*; import org.gridgain.grid.*; import org.gridgain.grid.cache.*; -import org.gridgain.grid.spi.swapspace.inmemory.*; +import org.apache.ignite.spi.swapspace.inmemory.*; import org.gridgain.grid.util.lang.*; import org.gridgain.grid.util.typedef.*; import org.gridgain.grid.util.typedef.internal.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheMemoryModeSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheMemoryModeSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheMemoryModeSelfTest.java index 84b3132..11befd4 100644 --- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheMemoryModeSelfTest.java +++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheMemoryModeSelfTest.java @@ -18,7 +18,7 @@ import org.gridgain.grid.cache.eviction.lru.*; 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.spi.swapspace.file.*; +import org.apache.ignite.spi.swapspace.file.*; import org.gridgain.grid.util.typedef.*; import org.gridgain.testframework.junits.common.*; import org.junit.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheOffHeapSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheOffHeapSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheOffHeapSelfTest.java index f741cbd..68a1b0b 100644 --- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheOffHeapSelfTest.java +++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheOffHeapSelfTest.java @@ -19,7 +19,7 @@ import org.gridgain.grid.cache.query.*; 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.spi.swapspace.file.*; +import org.apache.ignite.spi.swapspace.file.*; import org.gridgain.grid.util.typedef.*; import org.gridgain.grid.util.typedef.internal.*; import org.gridgain.testframework.junits.common.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheOffHeapTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheOffHeapTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheOffHeapTest.java index 66e2adb..7b03c6e 100644 --- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheOffHeapTest.java +++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheOffHeapTest.java @@ -16,7 +16,7 @@ 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.spi.swapspace.noop.*; +import org.apache.ignite.spi.swapspace.noop.*; import org.gridgain.testframework.junits.common.*; import java.util.concurrent.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheP2PUndeploySelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheP2PUndeploySelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheP2PUndeploySelfTest.java index 6954e2a..1f99504 100644 --- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheP2PUndeploySelfTest.java +++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheP2PUndeploySelfTest.java @@ -19,7 +19,7 @@ import org.gridgain.grid.kernal.processors.cache.distributed.*; 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.spi.swapspace.file.*; +import org.apache.ignite.spi.swapspace.file.*; import org.gridgain.grid.util.typedef.*; import org.gridgain.grid.util.typedef.internal.*; import org.gridgain.testframework.junits.common.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheSwapReloadSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheSwapReloadSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheSwapReloadSelfTest.java index 18ddd46..063d3ac 100644 --- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheSwapReloadSelfTest.java +++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheSwapReloadSelfTest.java @@ -19,8 +19,8 @@ import org.gridgain.grid.kernal.*; 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.spi.swapspace.*; -import org.gridgain.grid.spi.swapspace.file.*; +import org.apache.ignite.spi.swapspace.*; +import org.apache.ignite.spi.swapspace.file.*; import org.gridgain.grid.util.typedef.*; import org.gridgain.grid.util.typedef.internal.*; import org.gridgain.testframework.junits.common.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/GridCacheAbstractDistributedByteArrayValuesSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/GridCacheAbstractDistributedByteArrayValuesSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/GridCacheAbstractDistributedByteArrayValuesSelfTest.java index 0852e09..7029753 100644 --- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/GridCacheAbstractDistributedByteArrayValuesSelfTest.java +++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/GridCacheAbstractDistributedByteArrayValuesSelfTest.java @@ -13,7 +13,7 @@ import org.apache.ignite.*; import org.apache.ignite.configuration.*; import org.gridgain.grid.cache.*; import org.gridgain.grid.kernal.processors.cache.*; -import org.gridgain.grid.spi.swapspace.file.*; +import org.apache.ignite.spi.swapspace.file.*; import org.jetbrains.annotations.*; import java.util.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridCacheAbstractPartitionedOnlyByteArrayValuesSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridCacheAbstractPartitionedOnlyByteArrayValuesSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridCacheAbstractPartitionedOnlyByteArrayValuesSelfTest.java index 66d1af2..ac23a7e 100644 --- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridCacheAbstractPartitionedOnlyByteArrayValuesSelfTest.java +++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridCacheAbstractPartitionedOnlyByteArrayValuesSelfTest.java @@ -12,7 +12,7 @@ package org.gridgain.grid.kernal.processors.cache.distributed.dht; import org.apache.ignite.configuration.*; import org.gridgain.grid.cache.*; import org.gridgain.grid.kernal.processors.cache.distributed.*; -import org.gridgain.grid.spi.swapspace.file.*; +import org.apache.ignite.spi.swapspace.file.*; import static org.gridgain.grid.cache.GridCacheAtomicWriteOrderMode.*; import static org.gridgain.grid.cache.GridCacheAtomicityMode.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridCacheColocatedOptimisticTransactionSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridCacheColocatedOptimisticTransactionSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridCacheColocatedOptimisticTransactionSelfTest.java index bbc76cd..7fefb15 100644 --- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridCacheColocatedOptimisticTransactionSelfTest.java +++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/GridCacheColocatedOptimisticTransactionSelfTest.java @@ -15,7 +15,7 @@ import org.gridgain.grid.cache.*; 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.spi.swapspace.file.*; +import org.apache.ignite.spi.swapspace.file.*; import org.gridgain.grid.util.typedef.*; import org.gridgain.testframework.junits.common.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/replicated/GridCacheReplicatedUnswapAdvancedSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/replicated/GridCacheReplicatedUnswapAdvancedSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/replicated/GridCacheReplicatedUnswapAdvancedSelfTest.java index a2c721c..9abeada 100644 --- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/replicated/GridCacheReplicatedUnswapAdvancedSelfTest.java +++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/distributed/replicated/GridCacheReplicatedUnswapAdvancedSelfTest.java @@ -17,7 +17,7 @@ import org.gridgain.grid.cache.*; 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.spi.swapspace.file.*; +import org.apache.ignite.spi.swapspace.file.*; import org.gridgain.grid.util.typedef.internal.*; import org.gridgain.testframework.*; import org.gridgain.testframework.junits.common.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/local/GridCacheLocalByteArrayValuesSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/local/GridCacheLocalByteArrayValuesSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/local/GridCacheLocalByteArrayValuesSelfTest.java index 0abba7b..f9eb36b 100644 --- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/local/GridCacheLocalByteArrayValuesSelfTest.java +++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/local/GridCacheLocalByteArrayValuesSelfTest.java @@ -13,7 +13,7 @@ import org.apache.ignite.*; import org.apache.ignite.configuration.*; import org.gridgain.grid.cache.*; import org.gridgain.grid.kernal.processors.cache.*; -import org.gridgain.grid.spi.swapspace.file.*; +import org.apache.ignite.spi.swapspace.file.*; import org.gridgain.grid.util.typedef.*; import org.jetbrains.annotations.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/query/GridCacheSwapScanQueryAbstractSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/query/GridCacheSwapScanQueryAbstractSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/query/GridCacheSwapScanQueryAbstractSelfTest.java index 37c3707..1f74051 100644 --- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/query/GridCacheSwapScanQueryAbstractSelfTest.java +++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/query/GridCacheSwapScanQueryAbstractSelfTest.java @@ -18,7 +18,7 @@ import org.gridgain.grid.cache.query.*; 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.spi.swapspace.file.*; +import org.apache.ignite.spi.swapspace.file.*; import org.gridgain.testframework.*; import org.gridgain.testframework.junits.common.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/loadtest/swapspace/GridFileSwapSpaceSpiMultithreadedLoadTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/loadtest/swapspace/GridFileSwapSpaceSpiMultithreadedLoadTest.java b/modules/core/src/test/java/org/gridgain/grid/loadtest/swapspace/GridFileSwapSpaceSpiMultithreadedLoadTest.java index 68d0d4d..4f57454 100644 --- a/modules/core/src/test/java/org/gridgain/grid/loadtest/swapspace/GridFileSwapSpaceSpiMultithreadedLoadTest.java +++ b/modules/core/src/test/java/org/gridgain/grid/loadtest/swapspace/GridFileSwapSpaceSpiMultithreadedLoadTest.java @@ -12,8 +12,8 @@ package org.gridgain.grid.loadtest.swapspace; import org.apache.ignite.lang.*; import org.apache.ignite.spi.*; import org.gridgain.grid.*; -import org.gridgain.grid.spi.swapspace.*; -import org.gridgain.grid.spi.swapspace.file.*; +import org.apache.ignite.spi.swapspace.*; +import org.apache.ignite.spi.swapspace.file.*; import org.gridgain.grid.util.typedef.*; import org.gridgain.loadtests.util.*; import org.gridgain.testframework.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/spi/swapspace/GridSwapSpaceSpiAbstractSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/spi/swapspace/GridSwapSpaceSpiAbstractSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/spi/swapspace/GridSwapSpaceSpiAbstractSelfTest.java deleted file mode 100644 index 0a5f28a..0000000 --- a/modules/core/src/test/java/org/gridgain/grid/spi/swapspace/GridSwapSpaceSpiAbstractSelfTest.java +++ /dev/null @@ -1,634 +0,0 @@ -/* @java.file.header */ - -/* _________ _____ __________________ _____ - * __ ____/___________(_)______ /__ ____/______ ____(_)_______ - * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ - * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / - * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ - */ - -package org.gridgain.grid.spi.swapspace; - -import org.apache.ignite.lang.*; -import org.apache.ignite.spi.*; -import org.gridgain.grid.util.typedef.*; -import org.gridgain.grid.util.typedef.internal.*; -import org.gridgain.testframework.junits.common.*; -import org.jetbrains.annotations.*; - -import java.io.*; -import java.util.*; -import java.util.concurrent.*; - -import static java.util.concurrent.TimeUnit.*; -import static org.apache.ignite.events.IgniteEventType.*; -import static org.junit.Assert.*; - -/** - * Test for various {@link SwapSpaceSpi} implementations. - */ -public abstract class GridSwapSpaceSpiAbstractSelfTest extends GridCommonAbstractTest { - /** Default swap space name. */ - private static final String DFLT_SPACE_NAME = "dflt-space"; - - /** */ - protected static final String SPACE1 = "space1"; - - /** */ - protected static final String SPACE2 = "space2"; - - /** SPI to test. */ - protected SwapSpaceSpi spi; - - /** - * @return New {@link SwapSpaceSpi} instance. - */ - protected abstract SwapSpaceSpi spi(); - - /** {@inheritDoc} */ - @Override protected void beforeTest() throws Exception { - U.setWorkDirectory(null, U.getGridGainHome()); - - spi = spi(); - - getTestResources().inject(spi); - - spi.spiStart(""); - - spi.clear(DFLT_SPACE_NAME); - } - - /** @throws Exception If failed. */ - @Override protected void afterTest() throws Exception { - spi.spiStop(); - } - - /** - * @return Swap context. - */ - protected SwapContext context() { - return context(null); - } - - /** - * @param clsLdr Class loader. - * @return Swap context. - */ - private SwapContext context(@Nullable ClassLoader clsLdr) { - SwapContext ctx = new SwapContext(); - - ctx.classLoader(clsLdr != null ? clsLdr : getClass().getClassLoader()); - - return ctx; - } - - /** - * @param s String. - * @return Byte array. - */ - protected byte[] str2ByteArray(String s) { - return s.getBytes(); - } - - /** - * Tests the Create-Read-Update-Delete operations with a simple key. - * - * @throws Exception If failed. - */ - public void testSimpleCrud() throws Exception { - assertEquals(0, spi.count(DFLT_SPACE_NAME)); - - long key1 = 1; - - byte[] val1 = Long.toString(key1).getBytes(); - - spi.store(DFLT_SPACE_NAME, new SwapKey(key1), val1, context()); - - assertEquals(1, spi.count(DFLT_SPACE_NAME)); - - assertArrayEquals(spi.read(DFLT_SPACE_NAME, new SwapKey(key1), context()), val1); - - final byte[] val2 = "newValue".getBytes(); - - spi.store(DFLT_SPACE_NAME, new SwapKey(key1), val2, context()); - - assertEquals(1, spi.count(DFLT_SPACE_NAME)); - - assertArrayEquals(spi.read(DFLT_SPACE_NAME, new SwapKey(key1), context()), val2); - - spi.remove(DFLT_SPACE_NAME, new SwapKey(key1), new IgniteInClosure<byte[]>() { - @Override public void apply(byte[] old) { - assertArrayEquals(val2, old); - } - }, context()); - - assertEquals(0, spi.count(DFLT_SPACE_NAME)); - } - - /** - * Tests the Create-Read-Update-Delete operations with a simple key - * and different spaces. - * - * @throws Exception If failed. - */ - public void testSimpleCrudDifferentSpaces() throws Exception { - String space1 = SPACE1; - - spi.clear(space1); - - String space2 = SPACE2; - - spi.clear(space2); - - assertEquals(0, spi.count(space1)); - - assertEquals(0, spi.count(space2)); - - long key1 = 1; - - final byte[] val1 = Long.toString(key1).getBytes(); - - spi.store(space1, new SwapKey(key1), val1, context()); - - assertEquals(1, spi.count(space1)); - - assertEquals(0, spi.count(space2)); - - spi.store(space2, new SwapKey(key1), val1, context()); - - assertEquals(1, spi.count(space1)); - - assertEquals(1, spi.count(space2)); - - assertArrayEquals(spi.read(space1, new SwapKey(key1), context()), val1); - - assertArrayEquals(spi.read(space2, new SwapKey(key1), context()), val1); - - long key2 = 2; - - byte[] val2 = Long.toString(key2).getBytes(); - - spi.store(space1, new SwapKey(key2), val2, context()); - - assertEquals(2, spi.count(space1)); - - assertEquals(1, spi.count(space2)); - - assertArrayEquals(spi.read(space1, new SwapKey(key2), context()), val2); - - assertNull(spi.read(space2, new SwapKey(key2), context())); - - final byte[] val12 = "newValue".getBytes(); - - spi.store(space1, new SwapKey(key1), val12, context()); - - assertEquals(2, spi.count(space1)); - - assertEquals(1, spi.count(space2)); - - assertArrayEquals(spi.read(space1, new SwapKey(key1), context()), val12); - - assertArrayEquals(spi.read(space2, new SwapKey(key1), context()), val1); - - spi.remove(space1, new SwapKey(key1), new IgniteInClosure<byte[]>() { - @Override public void apply(byte[] old) { - assertArrayEquals(val12, old); - } - }, context()); - - assertEquals(1, spi.count(space1)); - - assertEquals(1, spi.count(space2)); - - spi.remove(space2, new SwapKey(key1), new IgniteInClosure<byte[]>() { - @Override public void apply(byte[] old) { - assertArrayEquals(val1, old); - } - }, context()); - - assertEquals(1, spi.count(space1)); - - assertEquals(0, spi.count(space2)); - } - - /** - * Tests the Create-Update-Delete operations with a key batches. - * - * @throws Exception If failed. - */ - public void testBatchCrud() throws Exception { - assertEquals(0, spi.count(DFLT_SPACE_NAME)); - - final Map<SwapKey, byte[]> batch = new HashMap<>(); - - int batchSize = 10; - - // Generate initial values. - for (int i = 0; i < batchSize; i++) - batch.put(new SwapKey(i), Integer.toString(i).getBytes()); - - spi.storeAll(DFLT_SPACE_NAME, batch, context()); - - assertEquals(batchSize, spi.count(DFLT_SPACE_NAME)); - - Map<SwapKey, byte[]> read = spi.readAll(DFLT_SPACE_NAME, batch.keySet(), context()); - - // Check all entries are as expected. - assertTrue(F.forAll(read, new P1<Map.Entry<SwapKey, byte[]>>() { - @Override public boolean apply(Map.Entry<SwapKey, byte[]> e) { - return Arrays.equals(batch.get(e.getKey()), e.getValue()); - } - })); - - // Generate new values. - for (int i = 0; i < batchSize; i++) - batch.put(new SwapKey(i), Integer.toString(i + 1).getBytes()); - - spi.storeAll(DFLT_SPACE_NAME, batch, context()); - - assertEquals(batchSize, spi.count(DFLT_SPACE_NAME)); - - read = spi.readAll(DFLT_SPACE_NAME, batch.keySet(), context()); - - // Check all entries are as expected. - assertTrue(F.forAll(read, new P1<Map.Entry<SwapKey, byte[]>>() { - @Override public boolean apply(Map.Entry<SwapKey, byte[]> e) { - return Arrays.equals(batch.get(e.getKey()), e.getValue()); - } - })); - - spi.removeAll(DFLT_SPACE_NAME, batch.keySet(), null, context()); - - assertEquals(0, spi.count(DFLT_SPACE_NAME)); - } - - /** - * @throws Exception If failed. - */ - public void testDeleteIfNotPersist() throws Exception { - spi.store(SPACE1, new SwapKey("key1"), "value1".getBytes(), context()); - - assertArrayEquals("value1".getBytes(), spi.read(SPACE1, new SwapKey("key1"), context())); - } - - /** - * @throws Exception If failed. - */ - public void testStoreReadRemove() throws Exception { - int cnt = 5; - - final CountDownLatch storeLatch = new CountDownLatch(cnt); - final CountDownLatch readLatch = new CountDownLatch(cnt); - final CountDownLatch rmvLatch = new CountDownLatch(cnt); - - spi.setListener(new SwapSpaceSpiListener() { - @Override public void onSwapEvent(int evtType, @Nullable String spaceName, @Nullable byte[] keyBytes) { - info("Received event: " + evtType); - - if (evtType == EVT_SWAP_SPACE_DATA_STORED) - storeLatch.countDown(); - - else if (evtType == EVT_SWAP_SPACE_DATA_READ) - readLatch.countDown(); - - else if (evtType == EVT_SWAP_SPACE_DATA_REMOVED) - rmvLatch.countDown(); - - else - assert false : "Unexpected event type: " + evtType; - } - }); - - for (int i = 0; i < cnt; i++) - assertNull(spi.read(SPACE1, new SwapKey("key" + i), context())); - - for (int i = 0; i < cnt; i++) - spi.store(SPACE1, new SwapKey("key" + i), str2ByteArray("value" + i), context()); - - assert storeLatch.await(5000, MILLISECONDS); - - for (int i = 0; i < cnt; i++) - assertArrayEquals(str2ByteArray("value" + i), spi.read(SPACE1, new SwapKey("key" + i), context())); - - assert readLatch.await(5000, MILLISECONDS); - - for (int i = 0; i < cnt; i++) { - final int tmp = i; - - spi.remove(SPACE1, new SwapKey("key" + i), new CI1<byte[]>() { - @Override public void apply(byte[] arr) { - assertArrayEquals(str2ByteArray("value" + tmp), arr); - - info("Removed correct value for: key" + tmp); - } - }, context()); - } - - assert rmvLatch.await(10000, MILLISECONDS); - - for (int i = 0; i < cnt; i++) - assertNull(spi.read(SPACE1, new SwapKey("key" + i), context())); - } - - - /** - * @throws Exception If failed. - */ - public void testStoreReadRemoveNulls() throws Exception { - int cnt = 5; - - final CountDownLatch storeLatch = new CountDownLatch(cnt); - final CountDownLatch readLatch = new CountDownLatch(cnt); - final CountDownLatch rmvLatch = new CountDownLatch(cnt); - - spi.setListener(new SwapSpaceSpiListener() { - @Override public void onSwapEvent(int evtType, @Nullable String spaceName, @Nullable byte[] keyBytes) { - info("Received event: " + evtType); - - if (evtType == EVT_SWAP_SPACE_DATA_STORED) - storeLatch.countDown(); - - else if (evtType == EVT_SWAP_SPACE_DATA_READ) - readLatch.countDown(); - - else if (evtType == EVT_SWAP_SPACE_DATA_REMOVED) - rmvLatch.countDown(); - - else - assert false : "Unexpected event type: " + evtType; - } - }); - - for (int i = 0; i < cnt; i++) - assertNull(spi.read(SPACE1, new SwapKey("key" + i), context())); - - for (int i = 0; i < cnt; i++) - spi.store(SPACE1, new SwapKey("key" + i), null, context()); - - assert storeLatch.await(5000, MILLISECONDS); - - for (int i = 0; i < cnt; i++) - assertNull(spi.read(SPACE1, new SwapKey("key" + i), context())); - - assert readLatch.await(5000, MILLISECONDS); - - for (int i = 0; i < cnt; i++) { - final int tmp = i; - - spi.remove(SPACE1, new SwapKey("key" + i), new CI1<byte[]>() { - @Override public void apply(byte[] arr) { - assertNull(arr); - - info("Removed correct value for: key" + tmp); - } - }, context()); - } - - assert rmvLatch.await(10000, MILLISECONDS); - - for (int i = 0; i < cnt; i++) - assertNull(spi.read(SPACE1, new SwapKey("key" + i), context())); - } - - /** - * @throws Exception If failed. - */ - public void testCollisions() throws Exception { - int cnt = 5; - - final CountDownLatch storeLatch = new CountDownLatch(cnt); - final CountDownLatch readLatch = new CountDownLatch(cnt); - final CountDownLatch rmvLatch = new CountDownLatch(cnt); - - spi.setListener(new SwapSpaceSpiListener() { - @Override public void onSwapEvent(int evtType, @Nullable String spaceName, @Nullable byte[] keyBytes) { - info("Received event: " + evtType); - - if (evtType == EVT_SWAP_SPACE_DATA_STORED) - storeLatch.countDown(); - - else if (evtType == EVT_SWAP_SPACE_DATA_READ) - readLatch.countDown(); - - else if (evtType == EVT_SWAP_SPACE_DATA_REMOVED) - rmvLatch.countDown(); - - else - assert false : "Unexpected event type: " + evtType; - } - }); - - List<Integer> keys = new ArrayList<>(cnt); - - final Map<Integer, String> entries = new HashMap<>(); - - for (int i = 0; i < cnt; i++) { - String val = "value" + i; - - spi.store(SPACE1, new SwapKey(new Key(i)), str2ByteArray(val), context()); - - keys.add(i); - - entries.put(i, val); - } - - assert storeLatch.await(5000, MILLISECONDS) : "Count: " + storeLatch.getCount(); - - for (int i = 0; i < cnt; i++) - assertArrayEquals(entries.get(i).getBytes(), - spi.read(SPACE1, new SwapKey(new Key(i)), context())); - - assert readLatch.await(5000, MILLISECONDS) : "Count: " + readLatch.getCount(); - - Collections.shuffle(keys); - - for (final Integer key : keys) { - spi.remove(SPACE1, new SwapKey(new Key(key)), new CI1<byte[]>() { - @Override public void apply(byte[] arr) { - assertArrayEquals(entries.get(key).getBytes(), arr); - - info("Removed correct entry for key: " + key); - } - }, context()); - } - - assert rmvLatch.await(5000, MILLISECONDS) : "Count: " + rmvLatch.getCount(); - - for (final Integer key : keys) - assertNull(spi.read(SPACE1, new SwapKey(new Key(key)), context())); - } - - - /** - * @throws Exception If failed. - */ - public void testIteration() throws Exception { - spi.clear(SPACE1); - - int cnt = 10; - - for (int i = 0; i < cnt; i++) - spi.store(SPACE1, new SwapKey("key" + i, i), str2ByteArray("value" + i), context()); - - for (int i = 0; i < cnt; i++) - assertArrayEquals(str2ByteArray("value" + i), - spi.read(SPACE1, new SwapKey("key" + i, i), context())); - - try (IgniteSpiCloseableIterator<Map.Entry<byte[], byte[]>> iter = spi.rawIterator(SPACE1)) { - assertNotNull(iter); - - int i = 0; - - while (iter.hasNext()) { - Map.Entry<byte[], byte[]> next = iter.next(); - - String key = getTestResources().getMarshaller().unmarshal(next.getKey(), null); - - info("Got from iterator [key=" + key + ", val=" + new String(next.getValue())); - - i++; - - iter.remove(); - } - - assertEquals(10, i); - } - - assertEquals(0, spi.count(SPACE1)); - } - - /** - * @throws Exception If failed. - */ - public void testIterationOverPartition() throws Exception { - spi.store(SPACE1, new SwapKey("key", 0), str2ByteArray("value"), context()); - - spi.clear(SPACE1); - - int cnt = 10; - - for (int i = 0; i < cnt; i++) - spi.store(SPACE1, new SwapKey("key" + i, i), str2ByteArray("value" + i), context()); - - for (int i = 0; i < cnt; i++) - assertArrayEquals(str2ByteArray("value" + i), - spi.read(SPACE1, new SwapKey("key" + i, i), context())); - - try (IgniteSpiCloseableIterator<Map.Entry<byte[], byte[]>> iter = spi.rawIterator(SPACE1, 5)) { - assertNotNull(iter); - - int i = 0; - - while (iter.hasNext()) { - Map.Entry<byte[], byte[]> next = iter.next(); - - String key = getTestResources().getMarshaller().unmarshal(next.getKey(), null); - - info("Got from iterator [key=" + key + ", val=" + new String(next.getValue())); - - assert "key5".equals(key); - - iter.remove(); - - assertNull(spi.read(SPACE1, new SwapKey(key, 5), context())); - - i++; - } - - assertEquals(1, i); - } - } - - /** - * @throws Exception If failed. - */ - public void testSwapIterator() throws Exception { - spi.store(SPACE1, new SwapKey("key", 0), str2ByteArray("value"), context()); - - spi.clear(SPACE1); - - int cnt = 10; - - for (int i = 0; i < cnt; i++) - spi.store(SPACE1, new SwapKey("key" + i, i), str2ByteArray("value" + i), context()); - - IgniteSpiCloseableIterator<Map.Entry<byte[], byte[]>> iter = spi.rawIterator(SPACE1); - - assertNotNull(iter); - - iter.close(); - - try { - iter.next(); - - assert false; - } - catch (NoSuchElementException e) { - info("Caught expected exception (illegal state): " + e); - } - } - - /** - * - */ - @SuppressWarnings({"PublicInnerClass"}) - public static class TestValue implements Serializable { - /** */ - private String val = "test-" + System.currentTimeMillis(); - - /** - * @return Value - */ - public String getValue() { - return val; - } - - /** {@inheritDoc} */ - @Override public boolean equals(Object obj) { - return obj instanceof TestValue && val.equals(((TestValue)obj).val); - } - - /** {@inheritDoc} */ - @Override public String toString() { - return S.toString(TestValue.class, this); - } - } - - /** - * Key. - */ - private static class Key { - /** Index. */ - private final int i; - - /** - * @param i Index. - */ - Key(int i) { - this.i = i; - } - - /** {@inheritDoc} */ - @Override public boolean equals(Object o) { - if (this == o) - return true; - - if (!(o instanceof Key)) - return false; - - Key key = (Key)o; - - return i == key.i; - } - - /** {@inheritDoc} */ - @Override public int hashCode() { - return 1; // 100% collision. - } - - /** {@inheritDoc} */ - @Override public String toString() { - return "Key: " + i; - } - } -} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/spi/swapspace/file/GridFileSwapCompactionSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/spi/swapspace/file/GridFileSwapCompactionSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/spi/swapspace/file/GridFileSwapCompactionSelfTest.java deleted file mode 100644 index 73b2491..0000000 --- a/modules/core/src/test/java/org/gridgain/grid/spi/swapspace/file/GridFileSwapCompactionSelfTest.java +++ /dev/null @@ -1,119 +0,0 @@ -/* @java.file.header */ - -/* _________ _____ __________________ _____ - * __ ____/___________(_)______ /__ ____/______ ____(_)_______ - * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ - * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / - * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ - */ - -package org.gridgain.grid.spi.swapspace.file; - -import org.gridgain.grid.util.typedef.*; -import org.gridgain.testframework.junits.common.*; - -import java.io.File; -import java.nio.ByteBuffer; -import java.util.*; - -/** - * Test for {@link FileSwapSpaceSpi}. - */ -public class GridFileSwapCompactionSelfTest extends GridCommonAbstractTest { - /** - * @throws Exception If failed. - */ - public void testCompact() throws Exception { - File file = new File(UUID.randomUUID().toString()); - - X.println("file: " + file.getPath()); - - FileSwapSpaceSpi.SwapFile f = new FileSwapSpaceSpi.SwapFile(file, 8); - - Random rnd = new Random(); - - ArrayList<FileSwapSpaceSpi.SwapValue> arr = new ArrayList<>(); - - int size = 0; - - for (int a = 0; a < 100; a++) { - FileSwapSpaceSpi.SwapValue[] vals = new FileSwapSpaceSpi.SwapValue[1 + rnd.nextInt(10)]; - - int size0 = 0; - - for (int i = 0; i < vals.length; i++) { - byte[] bytes = new byte[1 + rnd.nextInt(49)]; - - rnd.nextBytes(bytes); - - size0 += bytes.length; - - vals[i] = new FileSwapSpaceSpi.SwapValue(bytes); - - arr.add(vals[i]); - } - - f.write(new FileSwapSpaceSpi.SwapValues(vals, size0), 1); - - size += size0; - - assertEquals(f.length(), size); - assertEquals(file.length(), size); - } - - int i = 0; - - for (FileSwapSpaceSpi.SwapValue val : arr) - assertEquals(val.idx(), ++i); - - i = 0; - - for (int cnt = arr.size() / 2; i < cnt; i++) { - - FileSwapSpaceSpi.SwapValue v = arr.remove(rnd.nextInt(arr.size())); - - assertTrue(f.tryRemove(v.idx(), v)); - } - - int hash0 = 0; - - for (FileSwapSpaceSpi.SwapValue val : arr) - hash0 += Arrays.hashCode(val.readValue(f.readCh)); - - ArrayList<T2<ByteBuffer, ArrayDeque<FileSwapSpaceSpi.SwapValue>>> bufs = new ArrayList(); - - for (;;) { - ArrayDeque<FileSwapSpaceSpi.SwapValue> que = new ArrayDeque<>(); - - ByteBuffer buf = f.compact(que, 1024); - - if (buf == null) - break; - - bufs.add(new T2(buf, que)); - } - - f.delete(); - - int hash1 = 0; - - for (FileSwapSpaceSpi.SwapValue val : arr) - hash1 += Arrays.hashCode(val.value(null)); - - assertEquals(hash0, hash1); - - File file0 = new File(UUID.randomUUID().toString()); - - FileSwapSpaceSpi.SwapFile f0 = new FileSwapSpaceSpi.SwapFile(file0, 8); - - for (T2<ByteBuffer, ArrayDeque<FileSwapSpaceSpi.SwapValue>> t : bufs) - f0.write(t.get2(), t.get1(), 1); - - int hash2 = 0; - - for (FileSwapSpaceSpi.SwapValue val : arr) - hash2 += Arrays.hashCode(val.readValue(f0.readCh)); - - assertEquals(hash2, hash1); - } -} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ef258ece/modules/core/src/test/java/org/gridgain/grid/spi/swapspace/file/GridFileSwapSpaceSpiSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/gridgain/grid/spi/swapspace/file/GridFileSwapSpaceSpiSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/spi/swapspace/file/GridFileSwapSpaceSpiSelfTest.java deleted file mode 100644 index a7b7f85..0000000 --- a/modules/core/src/test/java/org/gridgain/grid/spi/swapspace/file/GridFileSwapSpaceSpiSelfTest.java +++ /dev/null @@ -1,345 +0,0 @@ -/* @java.file.header */ - -/* _________ _____ __________________ _____ - * __ ____/___________(_)______ /__ ____/______ ____(_)_______ - * _ / __ __ ___/__ / _ __ / _ / __ _ __ `/__ / __ __ \ - * / /_/ / _ / _ / / /_/ / / /_/ / / /_/ / _ / _ / / / - * \____/ /_/ /_/ \_,__/ \____/ \__,_/ /_/ /_/ /_/ - */ - -package org.gridgain.grid.spi.swapspace.file; - -import org.apache.ignite.lang.*; -import org.apache.ignite.spi.*; -import org.gridgain.grid.spi.swapspace.*; -import org.gridgain.grid.util.typedef.*; -import org.gridgain.grid.util.typedef.internal.*; -import org.jdk8.backport.*; -import org.jetbrains.annotations.*; - -import java.util.*; -import java.util.concurrent.*; -import java.util.concurrent.atomic.*; - -/** - * Test for {@link FileSwapSpaceSpi}. - */ -public class GridFileSwapSpaceSpiSelfTest extends GridSwapSpaceSpiAbstractSelfTest { - /** {@inheritDoc} */ - @Override protected SwapSpaceSpi spi() { - FileSwapSpaceSpi s = new FileSwapSpaceSpi(); - - s.setMaximumSparsity(0.05f); - s.setWriteBufferSize(8 * 1024); - - return s; - } - - /** - * Tests if SPI works correctly with multithreaded writes. - * - * @throws Exception If failed. - */ - public void testMultithreadedWrite() throws Exception { - final AtomicLong valCntr = new AtomicLong(); - - final SwapKey key = new SwapKey("key"); - - final CountDownLatch wLatch = new CountDownLatch(1); - - final AtomicBoolean done = new AtomicBoolean(); - - IgniteFuture<?> wFut = multithreadedAsync(new Callable<Object>() { - @Nullable @Override public Object call() throws Exception { - while (!done.get()) { - long val = valCntr.incrementAndGet(); - - spi.store(null, key, Long.toString(val).getBytes(), context()); - - if (val == 1) - wLatch.countDown(); - } - - return null; - } - }, 8); - - wLatch.await(); - - IgniteFuture<?> rFut = multithreadedAsync(new Callable<Object>() { - @Nullable @Override public Object call() throws Exception { - while (valCntr.get() < 1000) { - byte[] val = spi.read(null, key, context()); - - assertNotNull(val); - - long lval = Long.parseLong(new String(val)); - - assertTrue(lval <= valCntr.get()); - } - - return null; - } - }, 8); - - rFut.get(); - - done.set(true); - - wFut.get(); - } - - /** - * @param i Integer. - * @return Swap key. - */ - private SwapKey key(int i) { - return new SwapKey(i, i % 11, U.intToBytes(i)); - } - - /** - * @throws Exception If failed. - */ - public void testMultithreadedOperations() throws Exception { - final ConcurrentHashMap8<SwapKey, byte[]> map = new ConcurrentHashMap8<>(); - - Random rnd = new Random(); - - final int keys = 25000; - - int hash0 = 0; - - final int minValSize = 5; - final int maxValSize = 9000; // More than write buffer size. - - for (int i = 0; i < keys; i++) { - byte[] val = new byte[minValSize + rnd.nextInt(maxValSize - minValSize)]; - - rnd.nextBytes(val); - - hash0 += i * Arrays.hashCode(val); - - assertNull(map.put(key(i), val)); - } - - assertEquals(keys, map.size()); - - for (int i = 0; i < keys; i++) - assertTrue(map.containsKey(key(i))); - - final String space = "test_space"; - - final AtomicBoolean fin = new AtomicBoolean(); - - final IgniteFuture<?> fut = multithreadedAsync(new Callable<Object>() { - @Override public Object call() throws Exception { - Random rnd = new Random(); - - while (!fin.get()) { - final SwapKey key = key(rnd.nextInt(keys)); - - switch(rnd.nextInt(13)) { - case 0: // store - byte[] val = map.remove(key); - - if (val != null) - spi.store(space, key, val, context()); - - break; - - case 1: // remove - spi.remove(space, key, new CIX1<byte[]>() { - @Override public void applyx(byte[] bytes) { - if (bytes != null) - assertNull(map.putIfAbsent(key, bytes)); - } - }, context()); - - break; - - case 2: // read - for (;;) { - val = spi.read(space, key, context()); - - if (val != null) - break; - - val = map.get(key); - - if (val != null) - break; - } - - break; - - case 3: // storeAll - case 4: - case 9: - Map<SwapKey, byte[]> m = new HashMap<>(); - - int cnt = 1 + rnd.nextInt(25); - - for (int i = 0; i < cnt; i++) { - SwapKey k = key(rnd.nextInt(keys)); - - val = map.remove(k); - - if (val != null) - assertNull(m.put(k, val)); - } - - if (m.isEmpty()) - break; - - spi.storeAll(space, m, context()); - - break; - - case 5: // readAll - HashSet<SwapKey> s = new HashSet<>(); - - cnt = 1 + rnd.nextInt(25); - - for (int i = 0; i < cnt; i++) { - SwapKey k = key(rnd.nextInt(keys)); - - val = map.get(k); - - if (val == null) - s.add(k); - } - - while (!s.isEmpty()) { - m = spi.readAll(space, s, context()); - - s.removeAll(m.keySet()); - - Iterator<SwapKey> iter = s.iterator(); - - while (iter.hasNext()) { - SwapKey k = iter.next(); - - if (map.containsKey(k)) - iter.remove(); - } - } - - break; - - case 6: // iterateKeys - IgniteSpiCloseableIterator<Integer> kIt = spi.keyIterator(space, context()); - - if (kIt == null) - break; - - while (kIt.hasNext()) - assertNotNull(kIt.next()); - - kIt.close(); - - break; - - case 7: // iterate - IgniteSpiCloseableIterator<Map.Entry<byte[], byte[]>> iter = spi.rawIterator(space); - - if (iter == null) - break; - - while (iter.hasNext()) { - Map.Entry<byte[], byte[]> entry = iter.next(); - - assertEquals(4, entry.getKey().length); - - byte[] v = entry.getValue(); - - assertTrue(v.length >= minValSize && v.length < maxValSize); - } - - iter.close(); - - break; - - case 8: // iterate partitions - iter = spi.rawIterator(space, rnd.nextInt(11)); - - if (iter == null) - break; - - while ( iter.hasNext()) { - Map.Entry<byte[], byte[]> entry = iter.next(); - - assertEquals(4, entry.getKey().length); - - byte[] v = entry.getValue(); - - assertTrue(v.length >= minValSize && v.length < maxValSize); - } - - iter.close(); - - break; - - default: // removeAll - s = new HashSet<>(); - - cnt = 1 + rnd.nextInt(25); - - for (int i = 0; i < cnt; i++) { - SwapKey k = key(rnd.nextInt(keys)); - - val = map.get(k); - - if (val == null) - s.add(k); - } - - if (s.isEmpty()) - break; - - spi.removeAll(space, s, new IgniteBiInClosure<SwapKey, byte[]>() { - @Override public void apply(SwapKey k, byte[] bytes) { - if (bytes != null) - assertNull(map.putIfAbsent(k, bytes)); - } - }, context()); - - break; - } - } - - return null; - } - }, 39); - - Thread.sleep(60000); - - System.out.println("stopping"); - - fin.set(true); - - fut.get(); - - assertEquals(keys, map.size() + spi.count(space)); - - int hash1 = 0; - - int cnt = 0; - - IgniteSpiCloseableIterator<Map.Entry<byte[], byte[]>> iter = spi.rawIterator(space); - - while (iter.hasNext()) { - Map.Entry<byte[], byte[]> entry = iter.next(); - - hash1 += U.bytesToInt(entry.getKey(), 0) * Arrays.hashCode(entry.getValue()); - - cnt++; - } - - assertEquals(cnt, spi.count(space)); - - for (Map.Entry<SwapKey, byte[]> entry : map.entrySet()) - hash1 += (Integer)entry.getKey().key() * Arrays.hashCode(entry.getValue()); - - assertEquals(hash0, hash1); - } -}
