Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -195,6 +195,18 @@ public GridCacheMvccCandidate removeCandidate(IgniteTxKey key, @Nullable GridCac
if (deque.isEmpty())
cands.remove(cand.key());
}
else {
// Another lock of the thread holds the key, the future of this version could add a reentry only.
GridCacheMvccCandidate reentry = first.unenter(ver);

if (reentry != null) {
boolean rmvd = deque.removeIf(c -> c == reentry);

assert rmvd : reentry;

cand = reentry;
}
}
}

boolean empty = cands.isEmpty();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,9 @@ public class GridCacheMvccCandidate implements CacheLockCandidates {
/** Linked reentry. */
private GridCacheMvccCandidate reentry;

/** Version of the explicit lock future that created this reentry, see {@link #unenter(GridCacheVersion)}. */
private GridCacheVersion reentryVer;

/** Previous lock for the thread. */
@GridToStringExclude
private volatile GridCacheMvccCandidate prev;
Expand Down Expand Up @@ -198,6 +201,14 @@ public void topologyVersion(AffinityTopologyVersion topVer) {
* @return Reentry candidate.
*/
public GridCacheMvccCandidate reenter() {
return reenter(null);
}

/**
* @param reentryVer Version of the explicit lock future that creates the reentry, {@code null} otherwise.
* @return Reentry candidate.
*/
public GridCacheMvccCandidate reenter(@Nullable GridCacheVersion reentryVer) {
GridCacheMvccCandidate old = reentry;

GridCacheMvccCandidate reentry = new GridCacheMvccCandidate(
Expand All @@ -217,6 +228,7 @@ public GridCacheMvccCandidate reenter() {
read());

reentry.topVer = topVer;
reentry.reentryVer = reentryVer;

if (old != null)
reentry.reentry = old;
Expand All @@ -242,6 +254,27 @@ public GridCacheMvccCandidate reenter() {
return null;
}

/**
* Unlinks the reentry created by the explicit lock future of the given version, so that a failed future reverts
* its reentry of a key the thread already holds.
*
* @param reentryVer Version of the future.
* @return Unlinked reentry or {@code null} if the future has not created a reentry of this candidate.
*/
@Nullable public GridCacheMvccCandidate unenter(GridCacheVersion reentryVer) {
for (GridCacheMvccCandidate prev = this; prev.reentry != null; prev = prev.reentry) {
GridCacheMvccCandidate reentry = prev.reentry;

if (reentryVer.equals(reentry.reentryVer)) {
prev.reentry = reentry.reentry;

return reentry;
}
}

return null;
}

/**
* @param parent Sets locks parent entry.
*/
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -824,7 +824,8 @@ public void removeLocks(long threadId, GridCacheVersion ver, Collection<KeyCache

GridCacheMvccCandidate lock = ctx.mvcc().removeExplicitLock(threadId, txKey, ver);

if (lock != null) {
// A reverted reentry does not release the lock, the thread still holds it.
if (lock != null && !lock.reentry()) {
AffinityTopologyVersion topVer = lock.topologyVersion();

if (map == null) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -387,7 +387,7 @@ private boolean implicitTx() {
cand.topologyVersion(topVer);
}
else
cand = cand.reenter();
cand = cand.reenter(lockVer);

cctx.mvcc().addExplicitLock(threadId, cand, topVer);
}
Expand Down Expand Up @@ -601,6 +601,12 @@ private synchronized void onError(Throwable t) {
}
}
}
else if (timeoutObj != null) {
// Serialized with the registration of explicit lock candidates, see map().
synchronized (timeoutObj) {
return onComplete(false, true);
}
}

return onComplete(false, true);
}
Expand Down Expand Up @@ -933,10 +939,17 @@ private void mapOnTopology(final boolean remap, @Nullable final Runnable c) {
*/
private void map(Collection<KeyCacheObject> keys, boolean remap, boolean topLocked) {
try {
map0(
keys,
remap,
topLocked);
if (tx != null || timeoutObj == null)
map0(keys, remap, topLocked);
else {
// Candidates registered after the future has been completed by the timeout or cancellation are never
// released: undoLocks() has already run. The completion holds the same monitor, so the mapping is
// atomic with it. Transactions release their candidates on rollback instead.
synchronized (timeoutObj) {
if (!isDone())
map0(keys, remap, topLocked);
}
}
}
catch (IgniteCheckedException ex) {
onDone(false, ex);
Expand Down
Loading
Loading