You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2015/04/03 12:47:40 UTC
[34/50] [abbrv] incubator-ignite git commit: # GG-9973: Fixed.
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6629012/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheWriteBehindStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheWriteBehindStore.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheWriteBehindStore.java
new file mode 100644
index 0000000..5ce42f9
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheWriteBehindStore.java
@@ -0,0 +1,1015 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.store;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.store.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.*;
+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.internal.util.worker.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.lifecycle.*;
+import org.apache.ignite.thread.*;
+import org.jetbrains.annotations.*;
+import org.jsr166.*;
+
+import javax.cache.integration.*;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+import java.util.concurrent.locks.*;
+
+import static javax.cache.Cache.*;
+
+/**
+ * Internal wrapper for a {@link CacheStore} that enables write-behind logic.
+ * <p/>
+ * The general purpose of this approach is to reduce cache store load under high
+ * store update rate. The idea is to cache all write and remove operations in a pending
+ * map and delegate these changes to the underlying store either after timeout or
+ * if size of a pending map exceeded some pre-configured value. Another performance gain
+ * is achieved due to combining a group of similar operations to a single batch update.
+ * <p/>
+ * The essential flush size for the write-behind cache should be at least the estimated
+ * count of simultaneously written keys. In case of significantly smaller value there would
+ * be triggered a lot of flush events that will result in a high cache store load.
+ * <p/>
+ * Since write operations to the cache store are deferred, transaction support is lost; no
+ * transaction objects are passed to the underlying store.
+ */
+public class GridCacheWriteBehindStore<K, V> implements CacheStore<K, V>, LifecycleAware {
+ /** Default write cache initial capacity. */
+ public static final int DFLT_INITIAL_CAPACITY = 1024;
+
+ /** Overflow ratio for critical cache size calculation. */
+ public static final float CACHE_OVERFLOW_RATIO = 1.5f;
+
+ /** Default concurrency level of write cache. */
+ public static final int DFLT_CONCUR_LVL = 64;
+
+ /** Write cache initial capacity. */
+ private int initCap = DFLT_INITIAL_CAPACITY;
+
+ /** Concurrency level for write cache access. */
+ private int concurLvl = DFLT_CONCUR_LVL;
+
+ /** When cache size exceeds this value eldest entry will be stored to the underlying store. */
+ private int cacheMaxSize = CacheConfiguration.DFLT_WRITE_BEHIND_FLUSH_SIZE;
+
+ /** Critical cache size. If cache size exceeds this value, data flush performed synchronously. */
+ private int cacheCriticalSize;
+
+ /** Count of worker threads performing underlying store updates. */
+ private int flushThreadCnt = CacheConfiguration.DFLT_WRITE_FROM_BEHIND_FLUSH_THREAD_CNT;
+
+ /** Cache flush frequency. All pending operations will be performed in not less then this value ms. */
+ private long cacheFlushFreq = CacheConfiguration.DFLT_WRITE_BEHIND_FLUSH_FREQUENCY;
+
+ /** Maximum batch size for put and remove operations */
+ private int batchSize = CacheConfiguration.DFLT_WRITE_BEHIND_BATCH_SIZE;
+
+ /** Grid name. */
+ private String gridName;
+
+ /** Cache name. */
+ private String cacheName;
+
+ /** Underlying store. */
+ private CacheStore<K, V> store;
+
+ /** Write cache. */
+ private ConcurrentLinkedHashMap<K, StatefulValue<K, V>> writeCache;
+
+ /** Flusher threads. */
+ private GridWorker[] flushThreads;
+
+ /** Atomic flag indicating store shutdown. */
+ private AtomicBoolean stopping = new AtomicBoolean(true);
+
+ /** Flush lock. */
+ private Lock flushLock = new ReentrantLock();
+
+ /** Condition to determine records available for flush. */
+ private Condition canFlush = flushLock.newCondition();
+
+ /** Variable for counting total cache overflows. */
+ private AtomicInteger cacheTotalOverflowCntr = new AtomicInteger();
+
+ /** Variable contains current number of overflow events. */
+ private AtomicInteger cacheOverflowCntr = new AtomicInteger();
+
+ /** Variable for counting key-value pairs that are in {@link ValueStatus#RETRY} state. */
+ private AtomicInteger retryEntriesCnt = new AtomicInteger();
+
+ /** Log. */
+ private IgniteLogger log;
+
+ /** Store manager. */
+ private CacheStoreManager storeMgr;
+
+ /**
+ * Creates a write-behind cache store for the given store.
+ *
+ * @param storeMgr Store manager.
+ * @param gridName Grid name.
+ * @param cacheName Cache name.
+ * @param log Grid logger.
+ * @param store {@code GridCacheStore} that need to be wrapped.
+ */
+ public GridCacheWriteBehindStore(
+ CacheStoreManager storeMgr,
+ String gridName,
+ String cacheName,
+ IgniteLogger log,
+ CacheStore<K, V> store) {
+ this.storeMgr = storeMgr;
+ this.gridName = gridName;
+ this.cacheName = cacheName;
+ this.log = log;
+ this.store = store;
+ }
+
+ /**
+ * Sets initial capacity for the write cache.
+ *
+ * @param initCap Initial capacity.
+ */
+ public void setInitialCapacity(int initCap) {
+ this.initCap = initCap;
+ }
+
+ /**
+ * Sets concurrency level for the write cache. Concurrency level is expected count of concurrent threads
+ * attempting to update cache.
+ *
+ * @param concurLvl Concurrency level.
+ */
+ public void setConcurrencyLevel(int concurLvl) {
+ this.concurLvl = concurLvl;
+ }
+
+ /**
+ * Sets the maximum size of the write cache. When the count of unique keys in write cache exceeds this value,
+ * the eldest entry in the cache is immediately scheduled for write to the underlying store.
+ *
+ * @param cacheMaxSize Max cache size.
+ */
+ public void setFlushSize(int cacheMaxSize) {
+ this.cacheMaxSize = cacheMaxSize;
+ }
+
+ /**
+ * Gets the maximum size of the write-behind buffer. When the count of unique keys
+ * in write buffer exceeds this value, the buffer is scheduled for write to the underlying store.
+ * <p/>
+ * If this value is {@code 0}, then flush is performed only on time-elapsing basis. However,
+ * when this value is {@code 0}, the cache critical size is set to
+ * {@link CacheConfiguration#DFLT_WRITE_BEHIND_CRITICAL_SIZE}
+ *
+ * @return Buffer size that triggers flush procedure.
+ */
+ public int getWriteBehindFlushSize() {
+ return cacheMaxSize;
+ }
+
+ /**
+ * Sets the number of threads that will perform store update operations.
+ *
+ * @param flushThreadCnt Count of worker threads.
+ */
+ public void setFlushThreadCount(int flushThreadCnt) {
+ this.flushThreadCnt = flushThreadCnt;
+ }
+
+ /**
+ * Gets the number of flush threads that will perform store update operations.
+ *
+ * @return Count of worker threads.
+ */
+ public int getWriteBehindFlushThreadCount() {
+ return flushThreadCnt;
+ }
+
+ /**
+ * Sets the cache flush frequency. All pending operations on the underlying store will be performed
+ * within time interval not less then this value.
+ *
+ * @param cacheFlushFreq Time interval value in milliseconds.
+ */
+ public void setFlushFrequency(long cacheFlushFreq) {
+ this.cacheFlushFreq = cacheFlushFreq;
+ }
+
+ /**
+ * Gets the cache flush frequency. All pending operations on the underlying store will be performed
+ * within time interval not less then this value.
+ * <p/>
+ * If this value is {@code 0}, then flush is performed only when buffer size exceeds flush size.
+ *
+ * @return Flush frequency in milliseconds.
+ */
+ public long getWriteBehindFlushFrequency() {
+ return cacheFlushFreq;
+ }
+
+ /**
+ * Sets the maximum count of similar operations that can be grouped to a single batch.
+ *
+ * @param batchSize Maximum count of batch.
+ */
+ public void setBatchSize(int batchSize) {
+ this.batchSize = batchSize;
+ }
+
+ /**
+ * Gets the maximum count of similar (put or remove) operations that can be grouped to a single batch.
+ *
+ * @return Maximum size of batch.
+ */
+ public int getWriteBehindStoreBatchSize() {
+ return batchSize;
+ }
+
+ /**
+ * Gets count of entries that were processed by the write-behind store and have not been
+ * flushed to the underlying store yet.
+ *
+ * @return Total count of entries in cache store internal buffer.
+ */
+ public int getWriteBehindBufferSize() {
+ return writeCache.sizex();
+ }
+
+ /**
+ * @return Underlying store.
+ */
+ public CacheStore<K, V> store() {
+ return store;
+ }
+
+ /**
+ * Performs all the initialization logic for write-behind cache store.
+ * This class must not be used until this method returns.
+ */
+ @Override public void start() {
+ assert cacheFlushFreq != 0 || cacheMaxSize != 0;
+
+ if (stopping.compareAndSet(true, false)) {
+ if (log.isDebugEnabled())
+ log.debug("Starting write-behind store for cache '" + cacheName + '\'');
+
+ cacheCriticalSize = (int)(cacheMaxSize * CACHE_OVERFLOW_RATIO);
+
+ if (cacheCriticalSize == 0)
+ cacheCriticalSize = CacheConfiguration.DFLT_WRITE_BEHIND_CRITICAL_SIZE;
+
+ flushThreads = new GridWorker[flushThreadCnt];
+
+ writeCache = new ConcurrentLinkedHashMap<>(initCap, 0.75f, concurLvl);
+
+ for (int i = 0; i < flushThreads.length; i++) {
+ flushThreads[i] = new Flusher(gridName, "flusher-" + i, log);
+
+ new IgniteThread(flushThreads[i]).start();
+ }
+ }
+ }
+
+ /**
+ * Gets count of write buffer overflow events since initialization. Each overflow event causes
+ * the ongoing flush operation to be performed synchronously.
+ *
+ * @return Count of cache overflow events since start.
+ */
+ public int getWriteBehindTotalCriticalOverflowCount() {
+ return cacheTotalOverflowCntr.get();
+ }
+
+ /**
+ * Gets count of write buffer overflow events in progress at the moment. Each overflow event causes
+ * the ongoing flush operation to be performed synchronously.
+ *
+ * @return Count of cache overflow events since start.
+ */
+ public int getWriteBehindCriticalOverflowCount() {
+ return cacheOverflowCntr.get();
+ }
+
+ /**
+ * Gets count of cache entries that are in a store-retry state. An entry is assigned a store-retry state
+ * when underlying store failed due some reason and cache has enough space to retain this entry till
+ * the next try.
+ *
+ * @return Count of entries in store-retry state.
+ */
+ public int getWriteBehindErrorRetryCount() {
+ return retryEntriesCnt.get();
+ }
+
+ /**
+ * Performs shutdown logic for store. No put, get and remove requests will be processed after
+ * this method is called.
+ */
+ @Override public void stop() {
+ if (stopping.compareAndSet(false, true)) {
+ if (log.isDebugEnabled())
+ log.debug("Stopping write-behind store for cache '" + cacheName + '\'');
+
+ wakeUp();
+
+ boolean graceful = true;
+
+ for (GridWorker worker : flushThreads)
+ graceful &= U.join(worker, log);
+
+ if (!graceful)
+ log.warning("Shutdown was aborted");
+ }
+ }
+
+ /**
+ * Forces all entries collected to be flushed to the underlying store.
+ * @throws IgniteCheckedException If failed.
+ */
+ public void forceFlush() throws IgniteCheckedException {
+ wakeUp();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void loadCache(IgniteBiInClosure<K, V> clo, @Nullable Object... args) {
+ store.loadCache(clo, args);
+ }
+
+ /** {@inheritDoc} */
+ @Override public Map<K, V> loadAll(Iterable<? extends K> keys) {
+ if (log.isDebugEnabled())
+ log.debug("Store load all [keys=" + keys + ']');
+
+ Map<K, V> loaded = new HashMap<>();
+
+ Collection<K> remaining = new LinkedList<>();
+
+ for (K key : keys) {
+ StatefulValue<K, V> val = writeCache.get(key);
+
+ if (val != null) {
+ val.readLock().lock();
+
+ try {
+ if (val.operation() == StoreOperation.PUT)
+ loaded.put(key, val.entry().getValue());
+ else
+ assert val.operation() == StoreOperation.RMV : val.operation();
+ }
+ finally {
+ val.readLock().unlock();
+ }
+ }
+ else
+ remaining.add(key);
+ }
+
+ // For items that were not found in queue.
+ if (!remaining.isEmpty()) {
+ Map<K, V> loaded0 = store.loadAll(remaining);
+
+ if (loaded0 != null)
+ loaded.putAll(loaded0);
+ }
+
+ return loaded;
+ }
+
+ /** {@inheritDoc} */
+ @Override public V load(K key) {
+ if (log.isDebugEnabled())
+ log.debug("Store load [key=" + key + ']');
+
+ StatefulValue<K, V> val = writeCache.get(key);
+
+ if (val != null) {
+ val.readLock().lock();
+
+ try {
+ switch (val.operation()) {
+ case PUT:
+ return val.entry().getValue();
+
+ case RMV:
+ return null;
+
+ default:
+ assert false : "Unexpected operation: " + val.status();
+ }
+ }
+ finally {
+ val.readLock().unlock();
+ }
+ }
+
+ return store.load(key);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeAll(Collection<Entry<? extends K, ? extends V>> entries) {
+ for (Entry<? extends K, ? extends V> e : entries)
+ write(e);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void write(Entry<? extends K, ? extends V> entry) {
+ try {
+ if (log.isDebugEnabled())
+ log.debug("Store put [key=" + entry.getKey() + ", val=" + entry.getValue() + ']');
+
+ updateCache(entry.getKey(), entry, StoreOperation.PUT);
+ }
+ catch (IgniteInterruptedCheckedException e) {
+ throw new CacheWriterException(U.convertExceptionNoWrap(e));
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void deleteAll(Collection<?> keys) {
+ for (Object key : keys)
+ delete(key);
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public void delete(Object key) {
+ try {
+ if (log.isDebugEnabled())
+ log.debug("Store remove [key=" + key + ']');
+
+ updateCache((K)key, null, StoreOperation.RMV);
+ }
+ catch (IgniteInterruptedCheckedException e) {
+ throw new CacheWriterException(U.convertExceptionNoWrap(e));
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void sessionEnd(boolean commit) {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(GridCacheWriteBehindStore.class, this);
+ }
+
+ /**
+ * Performs flush-consistent cache update for the given key.
+ *
+ * @param key Key for which update is performed.
+ * @param val New value, may be null for remove operation.
+ * @param operation Updated value status
+ * @throws IgniteInterruptedCheckedException If interrupted while waiting for value to be flushed.
+ */
+ private void updateCache(K key,
+ @Nullable Entry<? extends K, ? extends V> val,
+ StoreOperation operation)
+ throws IgniteInterruptedCheckedException {
+ StatefulValue<K, V> newVal = new StatefulValue<>(val, operation);
+
+ StatefulValue<K, V> prev;
+
+ while ((prev = writeCache.putIfAbsent(key, newVal)) != null) {
+ prev.writeLock().lock();
+
+ try {
+ if (prev.status() == ValueStatus.PENDING) {
+ // Flush process in progress, try again.
+ prev.waitForFlush();
+
+ continue;
+ }
+ else if (prev.status() == ValueStatus.FLUSHED)
+ // This entry was deleted from map before we acquired the lock.
+ continue;
+ else if (prev.status() == ValueStatus.RETRY)
+ // New value has come, old value is no longer in RETRY state,
+ retryEntriesCnt.decrementAndGet();
+
+ assert prev.status() == ValueStatus.NEW || prev.status() == ValueStatus.RETRY;
+
+ prev.update(val, operation, ValueStatus.NEW);
+
+ break;
+ }
+ finally {
+ prev.writeLock().unlock();
+ }
+ }
+
+ // Now check the map size
+ if (writeCache.sizex() > cacheCriticalSize)
+ // Perform single store update in the same thread.
+ flushSingleValue();
+ else if (cacheMaxSize > 0 && writeCache.sizex() > cacheMaxSize)
+ wakeUp();
+ }
+
+ /**
+ * Flushes one upcoming value to the underlying store. Called from
+ * {@link #updateCache(Object, Entry, StoreOperation)} method in case when current map size exceeds
+ * critical size.
+ */
+ private void flushSingleValue() {
+ cacheOverflowCntr.incrementAndGet();
+
+ try {
+ Map<K, StatefulValue<K, V>> batch = null;
+
+ for (Map.Entry<K, StatefulValue<K, V>> e : writeCache.entrySet()) {
+ StatefulValue<K, V> val = e.getValue();
+
+ val.writeLock().lock();
+
+ try {
+ ValueStatus status = val.status();
+
+ if (acquired(status))
+ // Another thread is helping us, continue to the next entry.
+ continue;
+
+ if (val.status() == ValueStatus.RETRY)
+ retryEntriesCnt.decrementAndGet();
+
+ assert retryEntriesCnt.get() >= 0;
+
+ val.status(ValueStatus.PENDING);
+
+ batch = Collections.singletonMap(e.getKey(), val);
+ }
+ finally {
+ val.writeLock().unlock();
+ }
+
+ if (!batch.isEmpty()) {
+ applyBatch(batch, false);
+
+ cacheTotalOverflowCntr.incrementAndGet();
+
+ return;
+ }
+ }
+ }
+ finally {
+ cacheOverflowCntr.decrementAndGet();
+ }
+ }
+
+ /**
+ * Performs batch operation on underlying store.
+ *
+ * @param valMap Batch map.
+ * @param initSes {@code True} if need to initialize session.
+ */
+ private void applyBatch(Map<K, StatefulValue<K, V>> valMap, boolean initSes) {
+ assert valMap.size() <= batchSize;
+
+ StoreOperation operation = null;
+
+ // Construct a map for underlying store
+ Map<K, Entry<? extends K, ? extends V>> batch = U.newLinkedHashMap(valMap.size());
+
+ for (Map.Entry<K, StatefulValue<K, V>> e : valMap.entrySet()) {
+ if (operation == null)
+ operation = e.getValue().operation();
+
+ assert operation == e.getValue().operation();
+
+ assert e.getValue().status() == ValueStatus.PENDING;
+
+ batch.put(e.getKey(), e.getValue().entry());
+ }
+
+ if (updateStore(operation, batch, initSes)) {
+ for (Map.Entry<K, StatefulValue<K, V>> e : valMap.entrySet()) {
+ StatefulValue<K, V> val = e.getValue();
+
+ val.writeLock().lock();
+
+ try {
+ val.status(ValueStatus.FLUSHED);
+
+ StatefulValue<K, V> prev = writeCache.remove(e.getKey());
+
+ // Additional check to ensure consistency.
+ assert prev == val : "Map value for key " + e.getKey() + " was updated during flush";
+
+ val.signalFlushed();
+ }
+ finally {
+ val.writeLock().unlock();
+ }
+ }
+ }
+ else {
+ // Exception occurred, we must set RETRY status
+ for (StatefulValue<K, V> val : valMap.values()) {
+ val.writeLock().lock();
+
+ try {
+ val.status(ValueStatus.RETRY);
+
+ retryEntriesCnt.incrementAndGet();
+
+ val.signalFlushed();
+ }
+ finally {
+ val.writeLock().unlock();
+ }
+ }
+ }
+ }
+
+ /**
+ * Tries to update store with the given values and returns {@code true} in case of success.
+ *
+ * <p/> If any exception in underlying store is occurred, this method checks the map size.
+ * If map size exceeds some critical value, then it returns {@code true} and this value will
+ * be lost. If map size does not exceed critical value, it will return false and value will
+ * be retained in write cache.
+ *
+ * @param operation Status indicating operation that should be performed.
+ * @param vals Key-Value map.
+ * @param initSes {@code True} if need to initialize session.
+ * @return {@code true} if value may be deleted from the write cache,
+ * {@code false} otherwise
+ */
+ private boolean updateStore(StoreOperation operation,
+ Map<K, Entry<? extends K, ? extends V>> vals,
+ boolean initSes) {
+
+ if (initSes && storeMgr != null)
+ storeMgr.writeBehindSessionInit();
+
+ try {
+ boolean threwEx = true;
+
+ try {
+ switch (operation) {
+ case PUT:
+ store.writeAll(vals.values());
+
+ break;
+
+ case RMV:
+ store.deleteAll(vals.keySet());
+
+ break;
+
+ default:
+ assert false : "Unexpected operation: " + operation;
+ }
+
+ threwEx = false;
+
+ return true;
+ }
+ finally {
+ if (initSes && storeMgr != null)
+ storeMgr.writeBehindSessionEnd(threwEx);
+ }
+ }
+ catch (Exception e) {
+ LT.warn(log, e, "Unable to update underlying store: " + store);
+
+ if (writeCache.sizex() > cacheCriticalSize || stopping.get()) {
+ for (Map.Entry<K, Entry<? extends K, ? extends V>> entry : vals.entrySet()) {
+ Object val = entry.getValue() != null ? entry.getValue().getValue() : null;
+
+ log.warning("Failed to update store (value will be lost as current buffer size is greater " +
+ "than 'cacheCriticalSize' or node has been stopped before store was repaired) [key=" +
+ entry.getKey() + ", val=" + val + ", op=" + operation + "]");
+ }
+
+ return true;
+ }
+
+ return false;
+ }
+ }
+
+ /**
+ * Wakes up flushing threads if map size exceeded maximum value or in case of shutdown.
+ */
+ private void wakeUp() {
+ flushLock.lock();
+
+ try {
+ canFlush.signalAll();
+ }
+ finally {
+ flushLock.unlock();
+ }
+ }
+
+ /**
+ * Thread that performs time-based flushing of written values to the underlying storage.
+ */
+ private class Flusher extends GridWorker {
+ /** {@inheritDoc */
+ protected Flusher(String gridName, String name, IgniteLogger log) {
+ super(gridName, name, log);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void body() throws InterruptedException, IgniteInterruptedCheckedException {
+ while (!stopping.get() || writeCache.sizex() > 0) {
+ awaitOperationsAvailable();
+
+ flushCache(writeCache.entrySet().iterator());
+ }
+ }
+
+ /**
+ * This method awaits until enough elements in map are available or given timeout is over.
+ *
+ * @throws InterruptedException If awaiting was interrupted.
+ */
+ private void awaitOperationsAvailable() throws InterruptedException {
+ flushLock.lock();
+
+ try {
+ do {
+ if (writeCache.sizex() <= cacheMaxSize || cacheMaxSize == 0) {
+ if (cacheFlushFreq > 0)
+ canFlush.await(cacheFlushFreq, TimeUnit.MILLISECONDS);
+ else
+ canFlush.await();
+ }
+ }
+ while (writeCache.sizex() == 0 && !stopping.get());
+ }
+ finally {
+ flushLock.unlock();
+ }
+ }
+
+ /**
+ * Removes values from the write cache and performs corresponding operation
+ * on the underlying store.
+ *
+ * @param it Iterator for write cache.
+ */
+ private void flushCache(Iterator<Map.Entry<K,StatefulValue<K, V>>> it) {
+ StoreOperation operation = null;
+
+ Map<K, StatefulValue<K, V>> batch = null;
+ Map<K, StatefulValue<K, V>> pending = U.newLinkedHashMap(batchSize);
+
+ while (it.hasNext()) {
+ Map.Entry<K, StatefulValue<K, V>> e = it.next();
+
+ StatefulValue<K, V> val = e.getValue();
+
+ val.writeLock().lock();
+
+ try {
+ ValueStatus status = val.status();
+
+ if (acquired(status))
+ // Another thread is helping us, continue to the next entry.
+ continue;
+
+ if (status == ValueStatus.RETRY)
+ retryEntriesCnt.decrementAndGet();
+
+ assert retryEntriesCnt.get() >= 0;
+
+ val.status(ValueStatus.PENDING);
+
+ // We scan for the next operation and apply batch on operation change. Null means new batch.
+ if (operation == null)
+ operation = val.operation();
+
+ if (operation != val.operation()) {
+ // Operation is changed, so we need to perform a batch.
+ batch = pending;
+ pending = U.newLinkedHashMap(batchSize);
+
+ operation = val.operation();
+
+ pending.put(e.getKey(), val);
+ }
+ else
+ pending.put(e.getKey(), val);
+
+ if (pending.size() == batchSize) {
+ batch = pending;
+ pending = U.newLinkedHashMap(batchSize);
+
+ operation = null;
+ }
+ }
+ finally {
+ val.writeLock().unlock();
+ }
+
+ if (batch != null && !batch.isEmpty()) {
+ applyBatch(batch, true);
+ batch = null;
+ }
+ }
+
+ // Process the remainder.
+ if (!pending.isEmpty())
+ applyBatch(pending, true);
+ }
+ }
+
+ /**
+ * For test purposes only.
+ *
+ * @return Write cache for the underlying store operations.
+ */
+ Map<K, StatefulValue<K, V>> writeCache() {
+ return writeCache;
+ }
+
+ /**
+ * Enumeration that represents possible operations on the underlying store.
+ */
+ private enum StoreOperation {
+ /** Put key-value pair to the underlying store. */
+ PUT,
+
+ /** Remove key from the underlying store. */
+ RMV
+ }
+
+ /**
+ * Enumeration that represents possible states of value in the map.
+ */
+ private enum ValueStatus {
+ /** Value is scheduled for write or delete from the underlying cache but has not been captured by flusher. */
+ NEW,
+
+ /** Value is captured by flusher and store operation is performed at the moment. */
+ PENDING,
+
+ /** Store operation has failed and it will be re-tried at the next flush. */
+ RETRY,
+
+ /** Store operation succeeded and this value will be removed by flusher. */
+ FLUSHED,
+ }
+
+ /**
+ * Checks if given status indicates pending or complete flush operation.
+ *
+ * @param status Status to check.
+ * @return {@code true} if status indicates any pending or complete store update operation.
+ */
+ private boolean acquired(ValueStatus status) {
+ return status == ValueStatus.PENDING || status == ValueStatus.FLUSHED;
+ }
+
+ /**
+ * A state-value-operation trio.
+ *
+ * @param <V> Value type.
+ */
+ private static class StatefulValue<K, V> extends ReentrantReadWriteLock {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Value. */
+ @GridToStringInclude
+ private Entry<? extends K, ? extends V> val;
+
+ /** Store operation. */
+ private StoreOperation storeOperation;
+
+ /** Value status. */
+ private ValueStatus valStatus;
+
+ /** Condition to wait for flush event */
+ private Condition flushCond = writeLock().newCondition();
+
+ /**
+ * Creates a state-value pair with {@link ValueStatus#NEW} status.
+ *
+ * @param val Value.
+ * @param storeOperation Store operation.
+ */
+ private StatefulValue(Entry<? extends K, ? extends V> val, StoreOperation storeOperation) {
+ assert storeOperation == StoreOperation.PUT || storeOperation == StoreOperation.RMV;
+
+ this.val = val;
+ this.storeOperation = storeOperation;
+ valStatus = ValueStatus.NEW;
+ }
+
+ /**
+ * @return Stored value.
+ */
+ private Entry<? extends K, ? extends V> entry() {
+ return val;
+ }
+
+ /**
+ * @return Store operation.
+ */
+ private StoreOperation operation() {
+ return storeOperation;
+ }
+
+ /**
+ * @return Value status
+ */
+ private ValueStatus status() {
+ return valStatus;
+ }
+
+ /**
+ * Updates value status.
+ *
+ * @param valStatus Value status.
+ */
+ private void status(ValueStatus valStatus) {
+ this.valStatus = valStatus;
+ }
+
+ /**
+ * Updates both value and value status.
+ *
+ * @param val Value.
+ * @param storeOperation Store operation.
+ * @param valStatus Value status.
+ */
+ private void update(@Nullable Entry<? extends K, ? extends V> val,
+ StoreOperation storeOperation,
+ ValueStatus valStatus) {
+ this.val = val;
+ this.storeOperation = storeOperation;
+ this.valStatus = valStatus;
+ }
+
+ /**
+ * Awaits a signal on flush condition
+ *
+ * @throws IgniteInterruptedCheckedException If thread was interrupted.
+ */
+ private void waitForFlush() throws IgniteInterruptedCheckedException {
+ U.await(flushCond);
+ }
+
+ /**
+ * Signals flush condition.
+ */
+ @SuppressWarnings({"SignalWithoutCorrespondingAwait"})
+ private void signalFlushed() {
+ flushCond.signalAll();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (!(o instanceof StatefulValue))
+ return false;
+
+ StatefulValue other = (StatefulValue)o;
+
+ return F.eq(val, other.val) && F.eq(valStatus, other.valStatus);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ int res = val != null ? val.hashCode() : 0;
+
+ res = 31 * res + valStatus.hashCode();
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(StatefulValue.class, this);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6629012/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
index acd3202..1b66b4a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
@@ -24,6 +24,7 @@ import org.apache.ignite.internal.managers.communication.*;
import org.apache.ignite.internal.processors.affinity.*;
import org.apache.ignite.internal.processors.cache.*;
import org.apache.ignite.internal.processors.cache.distributed.near.*;
+import org.apache.ignite.internal.processors.cache.store.*;
import org.apache.ignite.internal.processors.cache.version.*;
import org.apache.ignite.internal.transactions.*;
import org.apache.ignite.internal.util.*;
@@ -444,11 +445,11 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter
*
* @return Store manager.
*/
- protected GridCacheStoreManager store() {
+ protected CacheStoreManager store() {
if (!activeCacheIds().isEmpty()) {
int cacheId = F.first(activeCacheIds());
- GridCacheStoreManager store = cctx.cacheContext(cacheId).store();
+ CacheStoreManager store = cctx.cacheContext(cacheId).store();
return store.configured() ? store : null;
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6629012/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
index f2407ce..10146a4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@ -25,6 +25,7 @@ import org.apache.ignite.internal.managers.communication.*;
import org.apache.ignite.internal.processors.cache.*;
import org.apache.ignite.internal.processors.cache.distributed.near.*;
import org.apache.ignite.internal.processors.cache.dr.*;
+import org.apache.ignite.internal.processors.cache.store.*;
import org.apache.ignite.internal.processors.cache.version.*;
import org.apache.ignite.internal.processors.dr.*;
import org.apache.ignite.internal.transactions.*;
@@ -370,7 +371,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
}
return new GridFinishedFuture<>(
- cacheCtx.store().loadAllFromStore(this, keys, c));
+ cacheCtx.store().loadAll(this, keys, c));
}
catch (IgniteCheckedException e) {
return new GridFinishedFuture<>(e);
@@ -387,7 +388,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
return false;
}
- return cacheCtx.store().loadAllFromStore(IgniteTxLocalAdapter.this, keys, c);
+ return cacheCtx.store().loadAll(IgniteTxLocalAdapter.this, keys, c);
}
},
true);
@@ -492,17 +493,17 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
*/
@SuppressWarnings({"CatchGenericClass"})
protected void batchStoreCommit(Iterable<IgniteTxEntry> writeEntries) throws IgniteCheckedException {
- GridCacheStoreManager store = store();
+ CacheStoreManager store = store();
- if (store != null && store.writeThrough() && storeEnabled() &&
- (!internal() || groupLock()) && (near() || store.writeToStoreFromDht())) {
+ if (store != null && store.isWriteThrough() && storeEnabled() &&
+ (!internal() || groupLock()) && (near() || store.isWriteToStoreFromDht())) {
try {
if (writeEntries != null) {
Map<Object, IgniteBiTuple<Object, GridCacheVersion>> putMap = null;
List<Object> rmvCol = null;
- GridCacheStoreManager writeStore = null;
+ CacheStoreManager writeStore = null;
- boolean skipNear = near() && store.writeToStoreFromDht();
+ boolean skipNear = near() && store.isWriteToStoreFromDht();
for (IgniteTxEntry e : writeEntries) {
if (skipNear && e.cached().isNear())
@@ -527,7 +528,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
if (rmvCol != null && !rmvCol.isEmpty()) {
assert writeStore != null;
- writeStore.removeAllFromStore(this, rmvCol);
+ writeStore.removeAll(this, rmvCol);
// Reset.
rmvCol.clear();
@@ -537,7 +538,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
// Batch-process puts if cache ID has changed.
if (writeStore != null && writeStore != cacheCtx.store() && putMap != null && !putMap.isEmpty()) {
- writeStore.putAllToStore(this, putMap);
+ writeStore.putAll(this, putMap);
// Reset.
putMap.clear();
@@ -568,7 +569,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
if (putMap != null && !putMap.isEmpty()) {
assert writeStore != null;
- writeStore.putAllToStore(this, putMap);
+ writeStore.putAll(this, putMap);
// Reset.
putMap.clear();
@@ -577,7 +578,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
}
if (writeStore != null && writeStore != cacheCtx.store() && rmvCol != null && !rmvCol.isEmpty()) {
- writeStore.removeAllFromStore(this, rmvCol);
+ writeStore.removeAll(this, rmvCol);
// Reset.
rmvCol.clear();
@@ -609,7 +610,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
assert writeStore != null;
// Batch put at the end of transaction.
- writeStore.putAllToStore(this, putMap);
+ writeStore.putAll(this, putMap);
}
if (rmvCol != null && !rmvCol.isEmpty()) {
@@ -617,12 +618,12 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
assert writeStore != null;
// Batch remove at the end of transaction.
- writeStore.removeAllFromStore(this, rmvCol);
+ writeStore.removeAll(this, rmvCol);
}
}
// Commit while locks are held.
- store.txEnd(this, true);
+ store.sessionEnd(this, true);
}
catch (IgniteCheckedException ex) {
commitError(ex);
@@ -981,11 +982,11 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
}
}
else {
- GridCacheStoreManager store = store();
+ CacheStoreManager store = store();
if (store != null && (!internal() || groupLock())) {
try {
- store.txEnd(this, true);
+ store.sessionEnd(this, true);
}
catch (IgniteCheckedException e) {
commitError(e);
@@ -1086,11 +1087,11 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter
cctx.tm().rollbackTx(this);
- GridCacheStoreManager store = store();
+ CacheStoreManager store = store();
- if (store != null && (near() || store.writeToStoreFromDht())) {
+ if (store != null && (near() || store.isWriteToStoreFromDht())) {
if (!internal() || groupLock())
- store.txEnd(this, false);
+ store.sessionEnd(this, false);
}
}
catch (Error | IgniteCheckedException | RuntimeException e) {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6629012/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessor.java
index 42586d2..6a8117f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessor.java
@@ -156,10 +156,4 @@ public interface IgniteCacheObjectProcessor extends GridProcessor {
* with {@link IgniteImmutable} annotation.
*/
public boolean immutable(Object obj);
-
- /**
- * @param cacheName Cache name.
- * @return {@code True} if portable format should be preserved when passing values to cache store.
- */
- public boolean keepPortableInStore(@Nullable String cacheName);
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6629012/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessorImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessorImpl.java
index f65b7bd..6e46757 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessorImpl.java
@@ -234,11 +234,6 @@ public class IgniteCacheObjectProcessorImpl extends GridProcessorAdapter impleme
}
/** {@inheritDoc} */
- @Override public boolean keepPortableInStore(@Nullable String cacheName) {
- return false;
- }
-
- /** {@inheritDoc} */
@Override public void onCacheProcessorStarted() {
// No-op.
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6629012/modules/core/src/main/java/org/apache/ignite/internal/processors/plugin/CachePluginManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/plugin/CachePluginManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/plugin/CachePluginManager.java
index 69ca1ae..580ff49 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/plugin/CachePluginManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/plugin/CachePluginManager.java
@@ -22,6 +22,7 @@ import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.*;
import org.apache.ignite.internal.processors.cache.*;
import org.apache.ignite.internal.processors.cache.dr.*;
+import org.apache.ignite.internal.processors.cache.store.*;
import org.apache.ignite.plugin.*;
import java.util.*;
@@ -76,10 +77,13 @@ public class CachePluginManager extends GridCacheManagerAdapter {
/**
* Creates optional component.
*
+ * @param ctx Kernal context.
+ * @param cfg Cache configuration.
* @param cls Component class.
* @return Created component.
*/
- public <T> T createComponent(Class<T> cls) {
+ @SuppressWarnings("unchecked")
+ public <T> T createComponent(GridKernalContext ctx, CacheConfiguration cfg, Class<T> cls) {
for (CachePluginProvider provider : providers) {
T res = (T)provider.createComponent(cls);
@@ -91,6 +95,8 @@ public class CachePluginManager extends GridCacheManagerAdapter {
return (T)new GridOsCacheDrManager();
else if (cls.equals(CacheConflictResolutionManager.class))
return (T)new CacheOsConflictResolutionManager<>();
+ else if (cls.equals(CacheStoreManager.class))
+ return (T)new CacheOsStoreManager(ctx, cfg);
throw new IgniteException("Unsupported component type: " + cls);
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6629012/modules/core/src/main/resources/META-INF/classnames.properties
----------------------------------------------------------------------
diff --git a/modules/core/src/main/resources/META-INF/classnames.properties b/modules/core/src/main/resources/META-INF/classnames.properties
index 55c1f9d..d7213a4 100644
--- a/modules/core/src/main/resources/META-INF/classnames.properties
+++ b/modules/core/src/main/resources/META-INF/classnames.properties
@@ -389,10 +389,10 @@ org.apache.ignite.internal.processors.cache.GridCacheProcessor$LocalAffinityFunc
org.apache.ignite.internal.processors.cache.GridCacheProjectionImpl
org.apache.ignite.internal.processors.cache.GridCacheProxyImpl
org.apache.ignite.internal.processors.cache.GridCacheReturn
-org.apache.ignite.internal.processors.cache.GridCacheStoreManager$1
-org.apache.ignite.internal.processors.cache.GridCacheStoreManager$2
-org.apache.ignite.internal.processors.cache.GridCacheStoreManager$3
-org.apache.ignite.internal.processors.cache.GridCacheStoreManager$4
+org.apache.ignite.internal.processors.cache.store.GridCacheStoreManagerAdapter$1
+org.apache.ignite.internal.processors.cache.store.GridCacheStoreManagerAdapter$2
+org.apache.ignite.internal.processors.cache.store.GridCacheStoreManagerAdapter$3
+org.apache.ignite.internal.processors.cache.store.GridCacheStoreManagerAdapter$4
org.apache.ignite.internal.processors.cache.GridCacheSwapManager$10
org.apache.ignite.internal.processors.cache.GridCacheSwapManager$12
org.apache.ignite.internal.processors.cache.GridCacheSwapManager$14
@@ -433,9 +433,9 @@ org.apache.ignite.internal.processors.cache.GridCacheUtils$8
org.apache.ignite.internal.processors.cache.GridCacheUtils$9
org.apache.ignite.internal.processors.cache.GridCacheValueCollection
org.apache.ignite.internal.processors.cache.GridCacheValueCollection$1
-org.apache.ignite.internal.processors.cache.GridCacheWriteBehindStore$StatefulValue
-org.apache.ignite.internal.processors.cache.GridCacheWriteBehindStore$StoreOperation
-org.apache.ignite.internal.processors.cache.GridCacheWriteBehindStore$ValueStatus
+org.apache.ignite.internal.processors.cache.store.GridCacheWriteBehindStore$StatefulValue
+org.apache.ignite.internal.processors.cache.store.GridCacheWriteBehindStore$StoreOperation
+org.apache.ignite.internal.processors.cache.store.GridCacheWriteBehindStore$ValueStatus
org.apache.ignite.internal.processors.cache.IgniteCacheProxy
org.apache.ignite.internal.processors.cache.IgniteCacheProxy$1
org.apache.ignite.internal.processors.cache.IgniteCacheProxy$2
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6629012/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedWritesTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedWritesTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedWritesTest.java
index b9f9602..c4ba385 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedWritesTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedWritesTest.java
@@ -26,14 +26,13 @@ import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
import org.apache.ignite.testframework.junits.common.*;
import org.apache.ignite.transactions.*;
-import javax.cache.configuration.*;
import java.util.concurrent.atomic.*;
import static org.apache.ignite.cache.CacheAtomicityMode.*;
/**
* Test that in {@link CacheMode#PARTITIONED} mode cache writes values only to the near cache store. <p/> This check
- * is needed because in current implementation if {@link GridCacheWriteBehindStore} assumes that and user store is
+ * is needed because in current implementation if {@link org.apache.ignite.internal.processors.cache.store.GridCacheWriteBehindStore} assumes that and user store is
* wrapped only in near cache (see {@link GridCacheProcessor} init logic).
*/
@SuppressWarnings({"unchecked"})
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6629012/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreAbstractSelfTest.java
deleted file mode 100644
index 883a216..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreAbstractSelfTest.java
+++ /dev/null
@@ -1,189 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.cache;
-
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.testframework.junits.common.*;
-
-import java.util.*;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.*;
-
-/**
- * Harness for {@link GridCacheWriteBehindStore} tests.
- */
-public abstract class GridCacheWriteBehindStoreAbstractSelfTest extends GridCommonAbstractTest {
- /** Write cache size. */
- public static final int CACHE_SIZE = 1024;
-
- /** Value dump interval. */
- public static final int FLUSH_FREQUENCY = 1000;
-
- /** Underlying store. */
- protected GridCacheTestStore delegate = new GridCacheTestStore();
-
- /** Tested store. */
- protected GridCacheWriteBehindStore<Integer, String> store;
-
- /** {@inheritDoc} */
- @Override protected void afterTestsStopped() throws Exception {
- delegate = null;
- store = null;
-
- super.afterTestsStopped();
- }
-
- /**
- * Initializes store.
- *
- * @param flushThreadCnt Count of flush threads
- * @throws Exception If failed.
- */
- protected void initStore(int flushThreadCnt) throws Exception {
- store = new GridCacheWriteBehindStore<>(null, "", "", log, delegate);
-
- store.setFlushFrequency(FLUSH_FREQUENCY);
-
- store.setFlushSize(CACHE_SIZE);
-
- store.setFlushThreadCount(flushThreadCnt);
-
- delegate.reset();
-
- store.start();
- }
-
- /**
- * Shutdowns store.
- *
- * @throws Exception If failed.
- */
- protected void shutdownStore() throws Exception {
- store.stop();
-
- assertTrue("Store cache must be empty after shutdown", store.writeCache().isEmpty());
- }
-
- /**
- * Performs multiple put, get and remove operations in several threads on a store. After
- * all threads finished their operations, returns the total set of keys that should be
- * in underlying store.
- *
- * @param threadCnt Count of threads that should update keys.
- * @param keysPerThread Count of unique keys assigned to a thread.
- * @return Set of keys that was totally put in store.
- * @throws Exception If failed.
- */
- protected Set<Integer> runPutGetRemoveMultithreaded(int threadCnt, final int keysPerThread) throws Exception {
- final ConcurrentMap<String, Set<Integer>> perThread = new ConcurrentHashMap<>();
-
- final AtomicBoolean running = new AtomicBoolean(true);
-
- final AtomicInteger cntr = new AtomicInteger();
-
- final AtomicInteger operations = new AtomicInteger();
-
- IgniteInternalFuture<?> fut = multithreadedAsync(new Runnable() {
- @SuppressWarnings({"NullableProblems"})
- @Override public void run() {
- // Initialize key set for this thread.
- Set<Integer> set = new HashSet<>();
-
- Set<Integer> old = perThread.putIfAbsent(Thread.currentThread().getName(), set);
-
- if (old != null)
- set = old;
-
- List<Integer> original = new ArrayList<>();
-
- Random rnd = new Random();
-
- for (int i = 0; i < keysPerThread; i++)
- original.add(cntr.getAndIncrement());
-
- try {
- while (running.get()) {
- int op = rnd.nextInt(3);
- int idx = rnd.nextInt(keysPerThread);
-
- int key = original.get(idx);
-
- switch (op) {
- case 0:
- store.write(new CacheEntryImpl<>(key, "val" + key));
- set.add(key);
-
- operations.incrementAndGet();
-
- break;
-
- case 1:
- store.delete(key);
- set.remove(key);
-
- operations.incrementAndGet();
-
- break;
-
- case 2:
- default:
- store.write(new CacheEntryImpl<>(key, "broken"));
-
- String val = store.load(key);
-
- assertEquals("Invalid intermediate value: " + val, "broken", val);
-
- store.write(new CacheEntryImpl<>(key, "val" + key));
-
- set.add(key);
-
- // 2 put operations performed here.
- operations.incrementAndGet();
- operations.incrementAndGet();
- operations.incrementAndGet();
-
- break;
- }
- }
- }
- catch (Exception e) {
- error("Unexpected exception in put thread", e);
-
- assert false;
- }
- }
- }, threadCnt, "put");
-
- U.sleep(10000);
-
- running.set(false);
-
- fut.get();
-
- log().info(">>> " + operations + " operations performed totally");
-
- Set<Integer> total = new HashSet<>();
-
- for (Set<Integer> threadVals : perThread.values()) {
- total.addAll(threadVals);
- }
-
- return total;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6629012/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreAbstractTest.java
deleted file mode 100644
index 305b8bb..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreAbstractTest.java
+++ /dev/null
@@ -1,349 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.cache;
-
-import org.apache.ignite.*;
-import org.apache.ignite.cache.*;
-import org.apache.ignite.configuration.*;
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.spi.discovery.tcp.*;
-import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
-import org.apache.ignite.testframework.junits.common.*;
-import org.apache.ignite.transactions.*;
-import org.jetbrains.annotations.*;
-
-import javax.cache.configuration.*;
-import java.util.*;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.*;
-
-import static org.apache.ignite.cache.CacheAtomicityMode.*;
-import static org.apache.ignite.transactions.TransactionConcurrency.*;
-import static org.apache.ignite.transactions.TransactionIsolation.*;
-
-/**
- * Basic store test.
- */
-public abstract class GridCacheWriteBehindStoreAbstractTest extends GridCommonAbstractTest {
- /** Flush frequency. */
- private static final int WRITE_FROM_BEHIND_FLUSH_FREQUENCY = 1000;
-
- /** Cache store. */
- private static final GridCacheTestStore store = new GridCacheTestStore();
-
- /**
- *
- */
- protected GridCacheWriteBehindStoreAbstractTest() {
- super(true /*start grid. */);
- }
-
- /** {@inheritDoc} */
- @Override protected void beforeTest() throws Exception {
- store.resetTimestamp();
- }
-
- /** {@inheritDoc} */
- @Override protected void afterTest() throws Exception {
- IgniteCache<?, ?> cache = jcache();
-
- if (cache != null)
- cache.clear();
-
- store.reset();
- }
-
- /** @return Caching mode. */
- protected abstract CacheMode cacheMode();
-
- /** {@inheritDoc} */
- @SuppressWarnings("unchecked")
- @Override protected final IgniteConfiguration getConfiguration() throws Exception {
- IgniteConfiguration c = super.getConfiguration();
-
- TcpDiscoverySpi disco = new TcpDiscoverySpi();
-
- disco.setIpFinder(new TcpDiscoveryVmIpFinder(true));
-
- c.setDiscoverySpi(disco);
-
- CacheConfiguration cc = defaultCacheConfiguration();
-
- cc.setCacheMode(cacheMode());
- cc.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
- cc.setSwapEnabled(false);
- cc.setAtomicityMode(TRANSACTIONAL);
-
- cc.setCacheStoreFactory(singletonFactory(store));
- cc.setReadThrough(true);
- cc.setWriteThrough(true);
- cc.setLoadPreviousValue(true);
-
- cc.setWriteBehindEnabled(true);
- cc.setWriteBehindFlushFrequency(WRITE_FROM_BEHIND_FLUSH_FREQUENCY);
-
- c.setCacheConfiguration(cc);
-
- return c;
- }
-
- /** @throws Exception If test fails. */
- public void testWriteThrough() throws Exception {
- IgniteCache<Integer, String> cache = jcache();
-
- Map<Integer, String> map = store.getMap();
-
- assert map.isEmpty();
-
- Transaction tx = grid().transactions().txStart(OPTIMISTIC, REPEATABLE_READ);
-
- try {
- for (int i = 1; i <= 10; i++) {
- cache.put(i, Integer.toString(i));
-
- checkLastMethod(null);
- }
-
- tx.commit();
- }
- finally {
- tx.close();
- }
-
- // Need to wait WFB flush timeout.
- U.sleep(WRITE_FROM_BEHIND_FLUSH_FREQUENCY + 100);
-
- checkLastMethod("putAll");
-
- assert cache.size() == 10;
-
- for (int i = 1; i <= 10; i++) {
- String val = map.get(i);
-
- assert val != null;
- assert val.equals(Integer.toString(i));
- }
-
- store.resetLastMethod();
-
- tx = grid().transactions().txStart();
-
- try {
- for (int i = 1; i <= 10; i++) {
- String val = cache.getAndRemove(i);
-
- checkLastMethod(null);
-
- assert val != null;
- assert val.equals(Integer.toString(i));
- }
-
- tx.commit();
- }
- finally {
- tx.close();
- }
-
- // Need to wait WFB flush timeout.
- U.sleep(WRITE_FROM_BEHIND_FLUSH_FREQUENCY + 100);
-
- checkLastMethod("removeAll");
-
- assert map.isEmpty();
- }
-
- /** @throws Exception If test failed. */
- public void testReadThrough() throws Exception {
- IgniteCache<Integer, String> cache = jcache();
-
- Map<Integer, String> map = store.getMap();
-
- assert map.isEmpty();
-
- try (Transaction tx = grid().transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
- for (int i = 1; i <= 10; i++)
- cache.put(i, Integer.toString(i));
-
- checkLastMethod(null);
-
- tx.commit();
- }
-
- // Need to wait WFB flush timeout.
- U.sleep(WRITE_FROM_BEHIND_FLUSH_FREQUENCY + 100);
-
- checkLastMethod("putAll");
-
- for (int i = 1; i <= 10; i++) {
- String val = map.get(i);
-
- assert val != null;
- assert val.equals(Integer.toString(i));
- }
-
- cache.clear();
-
- assert cache.localSize() == 0;
- assert cache.localSize() == 0;
-
- // Need to wait WFB flush timeout.
- U.sleep(WRITE_FROM_BEHIND_FLUSH_FREQUENCY + 100);
-
- assert map.size() == 10;
-
- for (int i = 1; i <= 10; i++) {
- // Read through.
- String val = cache.get(i);
-
- checkLastMethod("load");
-
- assert val != null;
- assert val.equals(Integer.toString(i));
- }
-
- assert cache.size() == 10;
-
- cache.clear();
-
- assert cache.localSize() == 0;
- assert cache.localSize() == 0;
-
- assert map.size() == 10;
-
- Set<Integer> keys = new HashSet<>();
-
- for (int i = 1; i <= 10; i++)
- keys.add(i);
-
- // Read through.
- Map<Integer, String> vals = cache.getAll(keys);
-
- checkLastMethod("loadAll");
-
- assert vals != null;
- assert vals.size() == 10;
-
- for (int i = 1; i <= 10; i++) {
- String val = vals.get(i);
-
- assert val != null;
- assert val.equals(Integer.toString(i));
- }
-
- // Write through.
- cache.removeAll(keys);
-
- // Need to wait WFB flush timeout.
- U.sleep(WRITE_FROM_BEHIND_FLUSH_FREQUENCY + 100);
-
- checkLastMethod("removeAll");
-
- assert cache.localSize() == 0;
- assert cache.localSize() == 0;
-
- assert map.isEmpty();
- }
-
- /** @throws Exception If failed. */
- public void testMultithreaded() throws Exception {
- final ConcurrentMap<String, Set<Integer>> perThread = new ConcurrentHashMap<>();
-
- final AtomicBoolean running = new AtomicBoolean(true);
-
- final IgniteCache<Integer, String> cache = jcache();
-
- IgniteInternalFuture<?> fut = multithreadedAsync(new Runnable() {
- @SuppressWarnings({"NullableProblems"})
- @Override public void run() {
- // Initialize key set for this thread.
- Set<Integer> set = new HashSet<>();
-
- Set<Integer> old = perThread.putIfAbsent(Thread.currentThread().getName(), set);
-
- if (old != null)
- set = old;
-
- Random rnd = new Random();
-
- int keyCnt = 20000;
-
- while (running.get()) {
- int op = rnd.nextInt(2);
- int key = rnd.nextInt(keyCnt);
-
- switch (op) {
- case 0:
- cache.put(key, "val" + key);
- set.add(key);
-
- break;
-
- case 1:
- default:
- cache.remove(key);
- set.remove(key);
-
- break;
- }
- }
- }
- }, 10, "put");
-
- U.sleep(10000);
-
- running.set(false);
-
- fut.get();
-
- U.sleep(5 * WRITE_FROM_BEHIND_FLUSH_FREQUENCY);
-
- Map<Integer, String> stored = store.getMap();
-
- for (Map.Entry<Integer, String> entry : stored.entrySet()) {
- int key = entry.getKey();
-
- assertEquals("Invalid value for key " + key, "val" + key, entry.getValue());
-
- boolean found = false;
-
- for (Set<Integer> threadPuts : perThread.values()) {
- if (threadPuts.contains(key)) {
- found = true;
-
- break;
- }
- }
-
- assert found : "No threads found that put key " + key;
- }
- }
-
- /** @param mtd Expected last method value. */
- private void checkLastMethod(@Nullable String mtd) {
- String lastMtd = store.getLastMethod();
-
- if (mtd == null)
- assert lastMtd == null : "Last method must be null: " + lastMtd;
- else {
- assert lastMtd != null : "Last method must be not null";
- assert lastMtd.equals(mtd) : "Last method does not match [expected=" + mtd + ", lastMtd=" + lastMtd + ']';
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6629012/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreLocalTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreLocalTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreLocalTest.java
deleted file mode 100644
index 6c050ca..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreLocalTest.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.cache;
-
-import org.apache.ignite.cache.*;
-
-/**
- * Tests {@link GridCacheWriteBehindStore} in grid configuration.
- */
-public class GridCacheWriteBehindStoreLocalTest extends GridCacheWriteBehindStoreAbstractTest {
- /** {@inheritDoc} */
- @Override protected CacheMode cacheMode() {
- return CacheMode.LOCAL;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6629012/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreMultithreadedSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreMultithreadedSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreMultithreadedSelfTest.java
deleted file mode 100644
index 9607784..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreMultithreadedSelfTest.java
+++ /dev/null
@@ -1,163 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.cache;
-
-import org.apache.ignite.internal.util.typedef.internal.*;
-
-import java.util.*;
-
-/**
- * Multithreaded tests for {@link GridCacheWriteBehindStore}.
- */
-public class GridCacheWriteBehindStoreMultithreadedSelfTest extends GridCacheWriteBehindStoreAbstractSelfTest {
- /**
- * This test performs complex set of operations on store from multiple threads.
- *
- * @throws Exception If failed.
- */
- public void testPutGetRemove() throws Exception {
- initStore(2);
-
- Set<Integer> exp;
-
- try {
- exp = runPutGetRemoveMultithreaded(10, 10);
- }
- finally {
- shutdownStore();
- }
-
- Map<Integer, String> map = delegate.getMap();
-
- Collection<Integer> extra = new HashSet<>(map.keySet());
-
- extra.removeAll(exp);
-
- assertTrue("The underlying store contains extra keys: " + extra, extra.isEmpty());
-
- Collection<Integer> missing = new HashSet<>(exp);
-
- missing.removeAll(map.keySet());
-
- assertTrue("Missing keys in the underlying store: " + missing, missing.isEmpty());
-
- for (Integer key : exp)
- assertEquals("Invalid value for key " + key, "val" + key, map.get(key));
- }
-
- /**
- * Tests that cache would keep values if underlying store fails.
- *
- * @throws Exception If failed.
- */
- public void testStoreFailure() throws Exception {
- delegate.setShouldFail(true);
-
- initStore(2);
-
- Set<Integer> exp;
-
- try {
- exp = runPutGetRemoveMultithreaded(10, 10);
-
- U.sleep(FLUSH_FREQUENCY);
-
- info(">>> There are " + store.getWriteBehindErrorRetryCount() + " entries in RETRY state");
-
- delegate.setShouldFail(false);
-
- // Despite that we set shouldFail flag to false, flush thread may just have caught an exception.
- // If we move store to the stopping state right away, this value will be lost. That's why this sleep
- // is inserted here to let all exception handlers in write-behind store exit.
- U.sleep(1000);
- }
- finally {
- shutdownStore();
- }
-
- Map<Integer, String> map = delegate.getMap();
-
- Collection<Integer> extra = new HashSet<>(map.keySet());
-
- extra.removeAll(exp);
-
- assertTrue("The underlying store contains extra keys: " + extra, extra.isEmpty());
-
- Collection<Integer> missing = new HashSet<>(exp);
-
- missing.removeAll(map.keySet());
-
- assertTrue("Missing keys in the underlying store: " + missing, missing.isEmpty());
-
- for (Integer key : exp)
- assertEquals("Invalid value for key " + key, "val" + key, map.get(key));
- }
-
- /**
- * Tests store consistency in case of high put rate, when flush is performed from the same thread
- * as put or remove operation.
- *
- * @throws Exception If failed.
- */
- public void testFlushFromTheSameThread() throws Exception {
- // 50 milliseconds should be enough.
- delegate.setOperationDelay(50);
-
- initStore(2);
-
- Set<Integer> exp;
-
- int start = store.getWriteBehindTotalCriticalOverflowCount();
-
- try {
- //We will have in total 5 * CACHE_SIZE keys that should be enough to grow map size to critical value.
- exp = runPutGetRemoveMultithreaded(5, CACHE_SIZE);
- }
- finally {
- log.info(">>> Done inserting, shutting down the store");
-
- shutdownStore();
- }
-
- // Restore delay.
- delegate.setOperationDelay(0);
-
- Map<Integer, String> map = delegate.getMap();
-
- int end = store.getWriteBehindTotalCriticalOverflowCount();
-
- log.info(">>> There are " + exp.size() + " keys in store, " + (end - start) + " overflows detected");
-
- assertTrue("No cache overflows detected (a bug or too few keys or too few delay?)", end > start);
-
- Collection<Integer> extra = new HashSet<>(map.keySet());
-
- extra.removeAll(exp);
-
- assertTrue("The underlying store contains extra keys: " + extra, extra.isEmpty());
-
- Collection<Integer> missing = new HashSet<>(exp);
-
- missing.removeAll(map.keySet());
-
- assertTrue("Missing keys in the underlying store: " + missing, missing.isEmpty());
-
- for (Integer key : exp)
- assertEquals("Invalid value for key " + key, "val" + key, map.get(key));
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6629012/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStorePartitionedMultiNodeSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStorePartitionedMultiNodeSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStorePartitionedMultiNodeSelfTest.java
deleted file mode 100644
index 8fb4f68..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStorePartitionedMultiNodeSelfTest.java
+++ /dev/null
@@ -1,215 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.cache;
-
-import org.apache.ignite.*;
-import org.apache.ignite.cache.*;
-import org.apache.ignite.cache.store.*;
-import org.apache.ignite.configuration.*;
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-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.apache.ignite.testframework.junits.common.*;
-import org.apache.ignite.transactions.*;
-
-import java.util.*;
-
-import static org.apache.ignite.cache.CacheAtomicityMode.*;
-import static org.apache.ignite.transactions.TransactionConcurrency.*;
-import static org.apache.ignite.transactions.TransactionIsolation.*;
-
-/**
- * Tests write-behind store with near and dht commit option.
- */
-public class GridCacheWriteBehindStorePartitionedMultiNodeSelfTest extends GridCommonAbstractTest {
- /** Grids to start. */
- private static final int GRID_CNT = 5;
-
- /** Ip finder. */
- private static final TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
-
- /** Flush frequency. */
- public static final int WRITE_BEHIND_FLUSH_FREQ = 1000;
-
- /** Stores per grid. */
- private GridCacheTestStore[] stores = new GridCacheTestStore[GRID_CNT];
-
- /** Start grid counter. */
- private int idx;
-
- /** {@inheritDoc} */
- @SuppressWarnings("unchecked")
- @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
- IgniteConfiguration c = super.getConfiguration(gridName);
-
- TcpDiscoverySpi disco = new TcpDiscoverySpi();
-
- disco.setIpFinder(ipFinder);
-
- c.setDiscoverySpi(disco);
-
- CacheConfiguration cc = defaultCacheConfiguration();
-
- cc.setCacheMode(CacheMode.PARTITIONED);
- cc.setWriteBehindEnabled(true);
- cc.setWriteBehindFlushFrequency(WRITE_BEHIND_FLUSH_FREQ);
- cc.setAtomicityMode(TRANSACTIONAL);
- cc.setNearConfiguration(new NearCacheConfiguration());
-
- CacheStore store = stores[idx] = new GridCacheTestStore();
-
- cc.setCacheStoreFactory(singletonFactory(store));
- cc.setReadThrough(true);
- cc.setWriteThrough(true);
- cc.setLoadPreviousValue(true);
-
- c.setCacheConfiguration(cc);
-
- idx++;
-
- return c;
- }
-
- /** {@inheritDoc} */
- @Override protected void afterTestsStopped() throws Exception {
- stores = null;
-
- super.afterTestsStopped();
- }
-
- /**
- * @throws Exception If failed.
- */
- private void prepare() throws Exception {
- idx = 0;
-
- startGrids(GRID_CNT);
- }
-
- /** {@inheritDoc} */
- @Override protected void afterTest() throws Exception {
- super.afterTest();
-
- stopAllGrids();
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testSingleWritesOnDhtNode() throws Exception {
- checkSingleWrites();
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testBatchWritesOnDhtNode() throws Exception {
- checkBatchWrites();
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testTxWritesOnDhtNode() throws Exception {
- checkTxWrites();
- }
-
- /**
- * @throws Exception If failed.
- */
- private void checkSingleWrites() throws Exception {
- prepare();
-
- IgniteCache<Integer, String> cache = grid(0).cache(null);
-
- for (int i = 0; i < 100; i++)
- cache.put(i, String.valueOf(i));
-
- checkWrites();
- }
-
- /**
- * @throws Exception If failed.
- */
- private void checkBatchWrites() throws Exception {
- prepare();
-
- Map<Integer, String> map = new HashMap<>();
-
- for (int i = 0; i < 100; i++)
- map.put(i, String.valueOf(i));
-
- grid(0).cache(null).putAll(map);
-
- checkWrites();
- }
-
- /**
- * @throws Exception If failed.
- */
- private void checkTxWrites() throws Exception {
- prepare();
-
- IgniteCache<Object, Object> cache = grid(0).cache(null);
-
- try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
- for (int i = 0; i < 100; i++)
- cache.put(i, String.valueOf(i));
-
- tx.commit();
- }
-
- checkWrites();
- }
-
- /**
- * @throws IgniteInterruptedCheckedException If sleep was interrupted.
- */
- private void checkWrites() throws IgniteInterruptedCheckedException {
- U.sleep(WRITE_BEHIND_FLUSH_FREQ * 2);
-
- Collection<Integer> allKeys = new ArrayList<>(100);
-
- for (int i = 0; i < GRID_CNT; i++) {
- Map<Integer,String> map = stores[i].getMap();
-
- assertFalse("Missing writes for node: " + i, map.isEmpty());
-
- allKeys.addAll(map.keySet());
-
- // Check there is no intersection.
- for (int j = 0; j < GRID_CNT; j++) {
- if (i == j)
- continue;
-
- Collection<Integer> intersection = new HashSet<>(stores[j].getMap().keySet());
-
- intersection.retainAll(map.keySet());
-
- assertTrue(intersection.isEmpty());
- }
- }
-
- assertEquals(100, allKeys.size());
-
- for (int i = 0; i < 100; i++)
- assertTrue(allKeys.contains(i));
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6629012/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStorePartitionedTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStorePartitionedTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStorePartitionedTest.java
deleted file mode 100644
index f9e454f..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStorePartitionedTest.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.cache;
-
-import org.apache.ignite.cache.*;
-
-/**
- * Tests {@link GridCacheWriteBehindStore} in partitioned configuration.
- */
-public class GridCacheWriteBehindStorePartitionedTest extends GridCacheWriteBehindStoreAbstractTest {
- /** {@inheritDoc} */
- @Override protected CacheMode cacheMode() {
- return CacheMode.PARTITIONED;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c6629012/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreReplicatedTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreReplicatedTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreReplicatedTest.java
deleted file mode 100644
index c809f90..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheWriteBehindStoreReplicatedTest.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.cache;
-
-import org.apache.ignite.cache.*;
-
-/**
- * Tests {@link GridCacheWriteBehindStore} in grid configuration.
- */
-public class GridCacheWriteBehindStoreReplicatedTest extends GridCacheWriteBehindStoreAbstractTest {
- /** {@inheritDoc} */
- @Override protected CacheMode cacheMode() {
- return CacheMode.REPLICATED;
- }
-}