You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2016/03/03 19:31:43 UTC
[1/2] ignite git commit: Reverted artificial lock.
Repository: ignite
Updated Branches:
refs/heads/ignite-atomic-good-lock-bench 08ede8bde -> 59f7afdde
Reverted artificial lock.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/e42cb6e1
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/e42cb6e1
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/e42cb6e1
Branch: refs/heads/ignite-atomic-good-lock-bench
Commit: e42cb6e1c906b02c736fac3515a489a6f6851b8f
Parents: 08ede8b
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Thu Mar 3 21:27:44 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Thu Mar 3 21:27:44 2016 +0300
----------------------------------------------------------------------
.../dht/atomic/GridDhtAtomicCache.java | 396 +++++++++----------
1 file changed, 190 insertions(+), 206 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/e42cb6e1/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index 6abba47..e908c05 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -31,7 +31,6 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import javax.cache.expiry.ExpiryPolicy;
import javax.cache.processor.EntryProcessor;
@@ -1322,8 +1321,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
}
- private final ReadWriteLock lock = new ReentrantReadWriteLock();
-
/**
* Executes local update after preloader fetched values.
*
@@ -1336,218 +1333,205 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
GridNearAtomicUpdateRequest req,
CI2<GridNearAtomicUpdateRequest, GridNearAtomicUpdateResponse> completionCb
) {
- lock.readLock().lock();
+ GridNearAtomicUpdateResponse res = new GridNearAtomicUpdateResponse(ctx.cacheId(), nodeId, req.futureVersion(),
+ ctx.deploymentEnabled());
+
+ List<KeyCacheObject> keys = req.keys();
+
+ assert !req.returnValue() || (req.operation() == TRANSFORM || keys.size() == 1);
+
+ GridDhtAtomicUpdateFuture dhtFut = null;
+
+ boolean remap = false;
+
+ String taskName = ctx.kernalContext().task().resolveTaskName(req.taskNameHash());
+
+ IgniteCacheExpiryPolicy expiry = null;
try {
- GridNearAtomicUpdateResponse res = new GridNearAtomicUpdateResponse(ctx.cacheId(), nodeId, req.futureVersion(),
- ctx.deploymentEnabled());
+ // If batch store update is enabled, we need to lock all entries.
+ // First, need to acquire locks on cache entries, then check filter.
+ List<GridDhtCacheEntry> locked = lockEntries(keys, req.topologyVersion());
+
+ Collection<IgniteBiTuple<GridDhtCacheEntry, GridCacheVersion>> deleted = null;
+
+ try {
+ GridDhtPartitionTopology top = topology();
+
+ top.readLock();
+
+ try {
+ if (top.stopping()) {
+ res.addFailedKeys(keys, new IgniteCheckedException("Failed to perform cache operation " +
+ "(cache is stopped): " + name()));
+
+ completionCb.apply(req, res);
+
+ return;
+ }
+
+ // Do not check topology version for CLOCK versioning since
+ // partition exchange will wait for near update future (if future is on server node).
+ // Also do not check topology version if topology was locked on near node by
+ // external transaction or explicit lock.
+ if ((req.fastMap() && !req.clientRequest()) || req.topologyLocked() ||
+ !needRemap(req.topologyVersion(), top.topologyVersion())) {
+ ClusterNode node = ctx.discovery().node(nodeId);
+
+ if (node == null) {
+ U.warn(log, "Node originated update request left grid: " + nodeId);
+
+ return;
+ }
+
+ boolean hasNear = ctx.discovery().cacheNearNode(node, name());
+
+ GridCacheVersion ver = req.updateVersion();
+
+ if (ver == null) {
+ // Assign next version for update inside entries lock.
+ ver = ctx.versions().next(top.topologyVersion());
+
+ if (hasNear)
+ res.nearVersion(ver);
+ }
+
+ assert ver != null : "Got null version for update request: " + req;
+
+ if (log.isDebugEnabled())
+ log.debug("Using cache version for update request on primary node [ver=" + ver +
+ ", req=" + req + ']');
+
+ boolean sndPrevVal = !top.rebalanceFinished(req.topologyVersion());
+
+ dhtFut = createDhtFuture(ver, req, res, completionCb, false);
+
+ expiry = expiryPolicy(req.expiry());
+
+ GridCacheReturn retVal = null;
+
+ if (keys.size() > 1 && // Several keys ...
+ writeThrough() && !req.skipStore() && // and store is enabled ...
+ !ctx.store().isLocal() && // and this is not local store ...
+ !ctx.dr().receiveEnabled() // and no DR.
+ ) {
+ // This method can only be used when there are no replicated entries in the batch.
+ UpdateBatchResult updRes = updateWithBatch(node,
+ hasNear,
+ req,
+ res,
+ locked,
+ ver,
+ dhtFut,
+ completionCb,
+ ctx.isDrEnabled(),
+ taskName,
+ expiry,
+ sndPrevVal);
+
+ deleted = updRes.deleted();
+ dhtFut = updRes.dhtFuture();
+
+ if (req.operation() == TRANSFORM)
+ retVal = updRes.invokeResults();
+ }
+ else {
+ UpdateSingleResult updRes = updateSingle(node,
+ hasNear,
+ req,
+ res,
+ locked,
+ ver,
+ dhtFut,
+ completionCb,
+ ctx.isDrEnabled(),
+ taskName,
+ expiry,
+ sndPrevVal);
+
+ retVal = updRes.returnValue();
+ deleted = updRes.deleted();
+ dhtFut = updRes.dhtFuture();
+ }
+
+ if (retVal == null)
+ retVal = new GridCacheReturn(ctx, node.isLocal(), true, null, true);
- GridCacheReturn retVal = new GridCacheReturn(ctx, false, true, null, true);
+ res.returnValue(retVal);
- res.returnValue(retVal);
+ if (req.writeSynchronizationMode() != FULL_ASYNC)
+ req.cleanup(!node.isLocal());
+
+ if (dhtFut != null)
+ ctx.mvcc().addAtomicFuture(dhtFut.version(), dhtFut);
+ }
+ else
+ // Should remap all keys.
+ remap = true;
+ }
+ finally {
+ top.readUnlock();
+ }
+ }
+ catch (GridCacheEntryRemovedException e) {
+ assert false : "Entry should not become obsolete while holding lock.";
+
+ e.printStackTrace();
+ }
+ finally {
+ if (locked != null)
+ unlockEntries(locked, req.topologyVersion());
+
+ // Enqueue if necessary after locks release.
+ if (deleted != null) {
+ assert !deleted.isEmpty();
+ assert ctx.deferredDelete() : this;
+
+ for (IgniteBiTuple<GridDhtCacheEntry, GridCacheVersion> e : deleted)
+ ctx.onDeferredDelete(e.get1(), e.get2());
+ }
+ }
+ }
+ catch (GridDhtInvalidPartitionException ignore) {
+ assert !req.fastMap() || req.clientRequest() : req;
+
+ if (log.isDebugEnabled())
+ log.debug("Caught invalid partition exception for cache entry (will remap update request): " + req);
+
+ remap = true;
+ }
+ catch (Throwable e) {
+ // At least RuntimeException can be thrown by the code above when GridCacheContext is cleaned and there is
+ // an attempt to use cleaned resources.
+ U.error(log, "Unexpected exception during cache update", e);
+
+ res.addFailedKeys(keys, e);
completionCb.apply(req, res);
+
+ if (e instanceof Error)
+ throw e;
+
+ return;
}
- finally {
- lock.readLock().unlock();
+
+ if (remap) {
+ assert dhtFut == null;
+
+ res.remapKeys(keys);
+
+ completionCb.apply(req, res);
+ }
+ else {
+ // If there are backups, map backup update future.
+ if (dhtFut != null)
+ dhtFut.map();
+ // Otherwise, complete the call.
+ else
+ completionCb.apply(req, res);
}
-// List<KeyCacheObject> keys = req.keys();
-//
-// assert !req.returnValue() || (req.operation() == TRANSFORM || keys.size() == 1);
-//
-// GridDhtAtomicUpdateFuture dhtFut = null;
-//
-// boolean remap = false;
-//
-// String taskName = ctx.kernalContext().task().resolveTaskName(req.taskNameHash());
-//
-// IgniteCacheExpiryPolicy expiry = null;
-//
-// try {
-// // If batch store update is enabled, we need to lock all entries.
-// // First, need to acquire locks on cache entries, then check filter.
-// List<GridDhtCacheEntry> locked = lockEntries(keys, req.topologyVersion());
-//
-// Collection<IgniteBiTuple<GridDhtCacheEntry, GridCacheVersion>> deleted = null;
-//
-// try {
-// GridDhtPartitionTopology top = topology();
-//
-// top.readLock();
-//
-// try {
-// if (top.stopping()) {
-// res.addFailedKeys(keys, new IgniteCheckedException("Failed to perform cache operation " +
-// "(cache is stopped): " + name()));
-//
-// completionCb.apply(req, res);
-//
-// return;
-// }
-//
-// // Do not check topology version for CLOCK versioning since
-// // partition exchange will wait for near update future (if future is on server node).
-// // Also do not check topology version if topology was locked on near node by
-// // external transaction or explicit lock.
-// if ((req.fastMap() && !req.clientRequest()) || req.topologyLocked() ||
-// !needRemap(req.topologyVersion(), top.topologyVersion())) {
-// ClusterNode node = ctx.discovery().node(nodeId);
-//
-// if (node == null) {
-// U.warn(log, "Node originated update request left grid: " + nodeId);
-//
-// return;
-// }
-//
-// boolean hasNear = ctx.discovery().cacheNearNode(node, name());
-//
-// GridCacheVersion ver = req.updateVersion();
-//
-// if (ver == null) {
-// // Assign next version for update inside entries lock.
-// ver = ctx.versions().next(top.topologyVersion());
-//
-// if (hasNear)
-// res.nearVersion(ver);
-// }
-//
-// assert ver != null : "Got null version for update request: " + req;
-//
-// if (log.isDebugEnabled())
-// log.debug("Using cache version for update request on primary node [ver=" + ver +
-// ", req=" + req + ']');
-//
-// boolean sndPrevVal = !top.rebalanceFinished(req.topologyVersion());
-//
-// dhtFut = createDhtFuture(ver, req, res, completionCb, false);
-//
-// expiry = expiryPolicy(req.expiry());
-//
-// GridCacheReturn retVal = null;
-//
-// if (keys.size() > 1 && // Several keys ...
-// writeThrough() && !req.skipStore() && // and store is enabled ...
-// !ctx.store().isLocal() && // and this is not local store ...
-// !ctx.dr().receiveEnabled() // and no DR.
-// ) {
-// // This method can only be used when there are no replicated entries in the batch.
-// UpdateBatchResult updRes = updateWithBatch(node,
-// hasNear,
-// req,
-// res,
-// locked,
-// ver,
-// dhtFut,
-// completionCb,
-// ctx.isDrEnabled(),
-// taskName,
-// expiry,
-// sndPrevVal);
-//
-// deleted = updRes.deleted();
-// dhtFut = updRes.dhtFuture();
-//
-// if (req.operation() == TRANSFORM)
-// retVal = updRes.invokeResults();
-// }
-// else {
-// UpdateSingleResult updRes = updateSingle(node,
-// hasNear,
-// req,
-// res,
-// locked,
-// ver,
-// dhtFut,
-// completionCb,
-// ctx.isDrEnabled(),
-// taskName,
-// expiry,
-// sndPrevVal);
-//
-// retVal = updRes.returnValue();
-// deleted = updRes.deleted();
-// dhtFut = updRes.dhtFuture();
-// }
-//
-// if (retVal == null)
-// retVal = new GridCacheReturn(ctx, node.isLocal(), true, null, true);
-//
-// res.returnValue(retVal);
-//
-// if (req.writeSynchronizationMode() != FULL_ASYNC)
-// req.cleanup(!node.isLocal());
-//
-// if (dhtFut != null)
-// ctx.mvcc().addAtomicFuture(dhtFut.version(), dhtFut);
-// }
-// else
-// // Should remap all keys.
-// remap = true;
-// }
-// finally {
-// top.readUnlock();
-// }
-// }
-// catch (GridCacheEntryRemovedException e) {
-// assert false : "Entry should not become obsolete while holding lock.";
-//
-// e.printStackTrace();
-// }
-// finally {
-// if (locked != null)
-// unlockEntries(locked, req.topologyVersion());
-//
-// // Enqueue if necessary after locks release.
-// if (deleted != null) {
-// assert !deleted.isEmpty();
-// assert ctx.deferredDelete() : this;
-//
-// for (IgniteBiTuple<GridDhtCacheEntry, GridCacheVersion> e : deleted)
-// ctx.onDeferredDelete(e.get1(), e.get2());
-// }
-// }
-// }
-// catch (GridDhtInvalidPartitionException ignore) {
-// assert !req.fastMap() || req.clientRequest() : req;
-//
-// if (log.isDebugEnabled())
-// log.debug("Caught invalid partition exception for cache entry (will remap update request): " + req);
-//
-// remap = true;
-// }
-// catch (Throwable e) {
-// // At least RuntimeException can be thrown by the code above when GridCacheContext is cleaned and there is
-// // an attempt to use cleaned resources.
-// U.error(log, "Unexpected exception during cache update", e);
-//
-// res.addFailedKeys(keys, e);
-//
-// completionCb.apply(req, res);
-//
-// if (e instanceof Error)
-// throw e;
-//
-// return;
-// }
-//
-// if (remap) {
-// assert dhtFut == null;
-//
-// res.remapKeys(keys);
-//
-// completionCb.apply(req, res);
-// }
-// else {
-// // If there are backups, map backup update future.
-// if (dhtFut != null)
-// dhtFut.map();
-// // Otherwise, complete the call.
-// else
-// completionCb.apply(req, res);
-// }
-//
-// sendTtlUpdateRequest(expiry);
+ sendTtlUpdateRequest(expiry);
}
/**
[2/2] ignite git commit: Commented-out eviction stuff.
Posted by vo...@apache.org.
Commented-out eviction stuff.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/59f7afdd
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/59f7afdd
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/59f7afdd
Branch: refs/heads/ignite-atomic-good-lock-bench
Commit: 59f7afdde3b2487492d7b81bf9743df1ebe76abc
Parents: e42cb6e
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Thu Mar 3 21:31:33 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Thu Mar 3 21:31:33 2016 +0300
----------------------------------------------------------------------
.../internal/processors/cache/GridCacheIoManager.java | 2 +-
.../distributed/dht/atomic/GridDhtAtomicCache.java | 12 ++++++------
2 files changed, 7 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/59f7afdd/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
index b297827..f6a775f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
@@ -619,7 +619,7 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
cctx.mvcc().contextReset();
// Unwind eviction notifications.
- CU.unwindEvicts(cctx);
+ //CU.unwindEvicts(cctx);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/59f7afdd/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index e908c05..2ca54c3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -2611,12 +2611,12 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
// Optimization.
return;
- // Must touch all entries since update may have deleted entries.
- // Eviction manager will remove empty entries.
- for (GridCacheMapEntry entry : locked) {
- if (entry != null && (skip == null || !skip.contains(entry.key())))
- ctx.evicts().touch(entry, topVer);
- }
+// // Must touch all entries since update may have deleted entries.
+// // Eviction manager will remove empty entries.
+// for (GridCacheMapEntry entry : locked) {
+// if (entry != null && (skip == null || !skip.contains(entry.key())))
+// ctx.evicts().touch(entry, topVer);
+// }
}
/**