You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by ki...@apache.org on 2016/05/06 16:23:14 UTC
[5/6] incubator-asterixdb git commit: Deadlock-free locking protocol
is enabled
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/23be9068/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/UpsertOperationCallbackFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/UpsertOperationCallbackFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/UpsertOperationCallbackFactory.java
index 707f986..87cb8e7 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/UpsertOperationCallbackFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/UpsertOperationCallbackFactory.java
@@ -26,6 +26,7 @@ import org.apache.asterix.common.transactions.ITransactionContext;
import org.apache.asterix.common.transactions.ITransactionSubsystem;
import org.apache.asterix.common.transactions.JobId;
import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallback;
@@ -47,7 +48,8 @@ public class UpsertOperationCallbackFactory extends AbstractOperationCallbackFac
@Override
public IModificationOperationCallback createModificationOperationCallback(String resourceName, long resourceId,
- int resourcePartition, Object resource, IHyracksTaskContext ctx) throws HyracksDataException {
+ int resourcePartition, Object resource, IHyracksTaskContext ctx, IOperatorNodePushable operatorNodePushable)
+ throws HyracksDataException {
ITransactionSubsystem txnSubsystem = txnSubsystemProvider.getTransactionSubsystem(ctx);
IIndexLifecycleManager indexLifeCycleManager = txnSubsystem.getAsterixAppRuntimeContextProvider()
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/23be9068/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/ConcurrentLockManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/ConcurrentLockManager.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/ConcurrentLockManager.java
index e268134..8dcc1be 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/ConcurrentLockManager.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/ConcurrentLockManager.java
@@ -22,7 +22,6 @@ package org.apache.asterix.transaction.management.service.locking;
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
@@ -46,6 +45,10 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
static final Logger LOGGER = Logger.getLogger(ConcurrentLockManager.class.getName());
static final Level LVL = Level.FINER;
+ public static final boolean ENABLED_DEADLOCK_FREE_LOCKING_PROTOCOL = true;
+
+ public static final int NIL = -1;
+ public static final long NILL = -1L;
public static final boolean DEBUG_MODE = false;//true
public static final boolean CHECK_CONSISTENCY = false;
@@ -54,8 +57,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
private ResourceArenaManager resArenaMgr;
private RequestArenaManager reqArenaMgr;
private JobArenaManager jobArenaMgr;
- private ConcurrentHashMap<Integer, Long> jobIdSlotMap;
- private ThreadLocal<DatasetLockCache> dsLockCache;
+ private ConcurrentHashMap<Integer, Long> jobId2JobSlotMap;
private LockManagerStats stats = new LockManagerStats(10000);
enum LockAction {
@@ -94,13 +96,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
resArenaMgr = new ResourceArenaManager(noArenas, lockManagerShrinkTimer);
reqArenaMgr = new RequestArenaManager(noArenas, lockManagerShrinkTimer);
jobArenaMgr = new JobArenaManager(noArenas, lockManagerShrinkTimer);
- jobIdSlotMap = new ConcurrentHashMap<>();
- dsLockCache = new ThreadLocal<DatasetLockCache>() {
- @Override
- protected DatasetLockCache initialValue() {
- return new DatasetLockCache();
- }
- };
+ jobId2JobSlotMap = new ConcurrentHashMap<Integer, Long>();
}
@Override
@@ -111,17 +107,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
final int dsId = datasetId.getId();
final int jobId = txnContext.getJobId().getId();
-
- if (entityHashValue != -1) {
- lock(datasetId, -1, LockMode.intentionMode(lockMode), txnContext);
- } else {
- if (dsLockCache.get().contains(jobId, dsId, lockMode)) {
- return;
- }
- }
-
final long jobSlot = findOrAllocJobSlot(jobId);
-
final ResourceGroup group = table.get(dsId, entityHashValue);
group.getLatch();
try {
@@ -160,9 +146,6 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
throw new IllegalStateException();
}
}
- if (entityHashValue == -1) {
- dsLockCache.get().put(jobId, dsId, lockMode);
- }
} catch (InterruptedException e) {
throw new WaitInterruptedException(txnContext, "interrupted", e);
} finally {
@@ -282,7 +265,15 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
* @return true if a cycle would be introduced, false otherwise
*/
private boolean introducesDeadlock(final long resSlot, final long jobSlot, final DeadlockTracker tracker) {
- return introducesDeadlock(resSlot, jobSlot, tracker, 0);
+ if (ENABLED_DEADLOCK_FREE_LOCKING_PROTOCOL) {
+ /**
+ * Due to the deadlock-free locking protocol, deadlock is not possible.
+ * So, this method always returns false.
+ */
+ return false;
+ } else {
+ return introducesDeadlock(resSlot, jobSlot, tracker, 0);
+ }
}
private boolean introducesDeadlock(final long resSlot, final long jobSlot, final DeadlockTracker tracker,
@@ -336,15 +327,8 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
final int dsId = datasetId.getId();
final int jobId = txnContext.getJobId().getId();
-
- if (entityHashValue != -1) {
- lock(datasetId, -1, LockMode.intentionMode(lockMode), txnContext);
- } else {
- throw new UnsupportedOperationException("instant locks are not supported on datasets");
- }
-
final ResourceGroup group = table.get(dsId, entityHashValue);
- if (group.firstResourceIndex.get() == -1l) {
+ if (group.firstResourceIndex.get() == NILL) {
validateJob(txnContext);
// if we do not have a resource in the group, we know that the
// resource that we are looking for is not locked
@@ -352,7 +336,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
}
// we only allocate a request slot if we actually have to wait
- long reqSlot = -1;
+ long reqSlot = NILL;
group.getLatch();
try {
@@ -374,7 +358,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
return;
case WAIT:
case CONV:
- if (reqSlot == -1) {
+ if (reqSlot == NILL) {
reqSlot = allocRequestSlot(resSlot, jobSlot, lockMode);
}
enqueueWaiter(group, reqSlot, resSlot, jobSlot, act, txnContext);
@@ -387,7 +371,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
} catch (InterruptedException e) {
throw new WaitInterruptedException(txnContext, "interrupted", e);
} finally {
- if (reqSlot != -1) {
+ if (reqSlot != NILL) {
// deallocate request, if we allocated one earlier
if (DEBUG_MODE)
LOGGER.finer("del req slot " + TypeUtil.Global.toString(reqSlot));
@@ -405,19 +389,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
final int dsId = datasetId.getId();
final int jobId = txnContext.getJobId().getId();
-
- if (entityHashValue != -1) {
- if (!tryLock(datasetId, -1, LockMode.intentionMode(lockMode), txnContext)) {
- return false;
- }
- } else {
- if (dsLockCache.get().contains(jobId, dsId, lockMode)) {
- return true;
- }
- }
-
final long jobSlot = findOrAllocJobSlot(jobId);
-
final ResourceGroup group = table.get(dsId, entityHashValue);
group.getLatch();
@@ -434,9 +406,6 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
// no break
case GET:
addHolder(reqSlot, resSlot, jobSlot);
- if (entityHashValue == -1) {
- dsLockCache.get().put(jobId, dsId, lockMode);
- }
return true;
case WAIT:
case CONV:
@@ -447,9 +416,6 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
} finally {
group.releaseLatch();
}
-
- // if we did acquire the dataset lock, but not the entity lock, we keep
- // it anyway and clean it up at the end of the job
}
@Override
@@ -460,17 +426,8 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
final int dsId = datasetId.getId();
final int jobId = txnContext.getJobId().getId();
-
- if (entityHashValue != -1) {
- if (!tryLock(datasetId, -1, LockMode.intentionMode(lockMode), txnContext)) {
- return false;
- }
- } else {
- throw new UnsupportedOperationException("instant locks are not supported on datasets");
- }
-
final ResourceGroup group = table.get(dsId, entityHashValue);
- if (group.firstResourceIndex.get() == -1l) {
+ if (group.firstResourceIndex.get() == NILL) {
validateJob(txnContext);
// if we do not have a resource in the group, we know that the
// resource that we are looking for is not locked
@@ -511,7 +468,8 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
throws ACIDException {
log("unlock", datasetId.getId(), entityHashValue, lockMode, txnContext);
final int jobId = txnContext.getJobId().getId();
- final long jobSlot = jobIdSlotMap.get(jobId);
+ final long jobSlot = jobId2JobSlotMap.get(jobId);
+
final int dsId = datasetId.getId();
unlock(dsId, entityHashValue, lockMode, jobSlot);
}
@@ -561,19 +519,15 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
} finally {
group.releaseLatch();
}
-
- // dataset intention locks are
- // a) kept in dsLockCache and
- // b) cleaned up only in releaseLocks at the end of the job
}
@Override
public void releaseLocks(ITransactionContext txnContext) throws ACIDException {
- log("releaseLocks", -1, -1, LockMode.ANY, txnContext);
+ log("releaseLocks", NIL, NIL, LockMode.ANY, txnContext);
stats.releaseLocks();
int jobId = txnContext.getJobId().getId();
- Long jobSlot = jobIdSlotMap.get(jobId);
+ Long jobSlot = jobId2JobSlotMap.get(jobId);
if (jobSlot == null) {
// we don't know the job, so there are no locks for it - we're done
return;
@@ -588,7 +542,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
synchronized (jobArenaMgr) {
holder = jobArenaMgr.getLastHolder(jobSlot);
}
- while (holder != -1) {
+ while (holder != NILL) {
long resource = reqArenaMgr.getResourceId(holder);
int dsId = resArenaMgr.getDatasetId(resource);
int pkHashVal = resArenaMgr.getPkHashVal(resource);
@@ -600,20 +554,20 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
if (DEBUG_MODE)
LOGGER.finer("del job slot " + TypeUtil.Global.toString(jobSlot));
jobArenaMgr.deallocate(jobSlot);
- jobIdSlotMap.remove(jobId);
+ jobId2JobSlotMap.remove(jobId);
stats.logCounters(LOGGER, Level.INFO, true);
}
private long findOrAllocJobSlot(int jobId) {
- Long jobSlot = jobIdSlotMap.get(jobId);
+ Long jobSlot = jobId2JobSlotMap.get(jobId);
if (jobSlot == null) {
jobSlot = new Long(jobArenaMgr.allocate());
if (DEBUG_MODE)
LOGGER.finer("new job slot " + TypeUtil.Global.toString(jobSlot) + " (" + jobId + ")");
jobArenaMgr.setJobId(jobSlot, jobId);
- Long oldSlot = jobIdSlotMap.putIfAbsent(jobId, jobSlot);
+ Long oldSlot = jobId2JobSlotMap.putIfAbsent(jobId, jobSlot);
if (oldSlot != null) {
- // if another thread allocated a slot for this jobId between
+ // if another thread allocated a slot for this jobThreadId between
// get(..) and putIfAbsent(..), we'll use that slot and
// deallocate the one we allocated
if (DEBUG_MODE)
@@ -629,7 +583,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
private long findOrAllocResourceSlot(ResourceGroup group, int dsId, int entityHashValue) {
long resSlot = findResourceInGroup(group, dsId, entityHashValue);
- if (resSlot == -1) {
+ if (resSlot == NILL) {
// we don't know about this resource, let's alloc a slot
resSlot = resArenaMgr.allocate();
resArenaMgr.setDatasetId(resSlot, dsId);
@@ -690,11 +644,15 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
// carefully distinguishing the different lock modes
long holder = resArenaMgr.getLastHolder(resource);
LockAction res = LockAction.WAIT;
- while (holder != -1) {
+ while (holder != NILL) {
if (job == reqArenaMgr.getJobSlot(holder)) {
if (reqArenaMgr.getLockMode(holder) == lockMode) {
return LockAction.GET;
} else {
+ if (ENABLED_DEADLOCK_FREE_LOCKING_PROTOCOL) {
+ throw new IllegalStateException(
+ "Lock conversion is not supported when deadlock-free locking protocol is enabled!");
+ }
res = LockAction.CONV;
}
}
@@ -706,7 +664,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
private long findResourceInGroup(ResourceGroup group, int dsId, int entityHashValue) {
stats.logCounters(LOGGER, Level.INFO, false);
long resSlot = group.firstResourceIndex.get();
- while (resSlot != -1) {
+ while (resSlot != NILL) {
// either we already have a lock on this resource or we have a
// hash collision
if (resArenaMgr.getDatasetId(resSlot) == dsId && resArenaMgr.getPkHashVal(resSlot) == entityHashValue) {
@@ -715,7 +673,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
resSlot = resArenaMgr.getNext(resSlot);
}
}
- return -1;
+ return NILL;
}
private void addHolder(long request, long resource, long job) {
@@ -732,7 +690,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
private boolean hasOtherHolders(long resSlot, long jobSlot) {
long holder = resArenaMgr.getLastHolder(resSlot);
- while (holder != -1) {
+ while (holder != NILL) {
if (reqArenaMgr.getJobSlot(holder) != jobSlot) {
return true;
}
@@ -771,10 +729,10 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
private long removeRequestFromJob(long holder, long unmodified) {
long prevForJob = reqArenaMgr.getPrevJobRequest(holder);
long nextForJob = reqArenaMgr.getNextJobRequest(holder);
- if (nextForJob != -1) {
+ if (nextForJob != NILL) {
reqArenaMgr.setPrevJobRequest(nextForJob, prevForJob);
}
- if (prevForJob == -1) {
+ if (prevForJob == NILL) {
return nextForJob;
} else {
reqArenaMgr.setNextJobRequest(prevForJob, nextForJob);
@@ -792,8 +750,8 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
@Override
public void add(long request, long resource, long job) {
long waiter = resArenaMgr.getFirstWaiter(resource);
- reqArenaMgr.setNextRequest(request, -1);
- if (waiter == -1) {
+ reqArenaMgr.setNextRequest(request, NILL);
+ if (waiter == NILL) {
resArenaMgr.setFirstWaiter(resource, request);
} else {
appendToRequestQueue(waiter, request);
@@ -858,7 +816,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
private void insertIntoJobQueue(long newRequest, long oldRequest) {
reqArenaMgr.setNextJobRequest(newRequest, oldRequest);
- reqArenaMgr.setPrevJobRequest(newRequest, -1);
+ reqArenaMgr.setPrevJobRequest(newRequest, NILL);
if (oldRequest >= 0) {
reqArenaMgr.setPrevJobRequest(oldRequest, newRequest);
}
@@ -866,7 +824,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
private void appendToRequestQueue(long head, long appendee) {
long next = reqArenaMgr.getNextRequest(head);
- while (next != -1) {
+ while (next != NILL) {
head = next;
next = reqArenaMgr.getNextRequest(head);
}
@@ -876,9 +834,9 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
private long removeRequestFromQueueForSlot(long head, long reqSlot) {
long cur = head;
long prev = cur;
- while (prev != -1) {
+ while (prev != NILL) {
cur = reqArenaMgr.getNextRequest(prev);
- if (cur == -1) {
+ if (cur == NILL) {
throw new IllegalStateException("request " + reqSlot + " not in queue");
}
if (cur == reqSlot) {
@@ -907,9 +865,9 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
private long removeRequestFromQueueForJob(long head, long jobSlot, byte lockMode) {
long holder = head;
long prev = holder;
- while (prev != -1) {
+ while (prev != NILL) {
holder = reqArenaMgr.getNextRequest(prev);
- if (holder == -1) {
+ if (holder == NILL) {
throw new IllegalStateException("no entry for job " + jobSlot + " in queue");
}
if (requestMatches(holder, jobSlot, lockMode)) {
@@ -925,7 +883,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
private int determineNewMaxMode(long resource, int oldMaxMode) {
int newMaxMode = LockMode.NL;
long holder = resArenaMgr.getLastHolder(resource);
- while (holder != -1) {
+ while (holder != NILL) {
int curLockMode = reqArenaMgr.getLockMode(holder);
if (curLockMode == oldMaxMode) {
// we have another lock of the same mode - we're done
@@ -976,10 +934,10 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
}
StringBuilder sb = new StringBuilder();
sb.append("{ op : ").append(string);
- if (id != -1) {
+ if (id != NIL) {
sb.append(" , dataset : ").append(id);
}
- if (entityHashValue != -1) {
+ if (entityHashValue != NIL) {
sb.append(" , entity : ").append(entityHashValue);
}
if (lockMode != LockMode.NL) {
@@ -1000,11 +958,11 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
if (group.tryLatch(100, TimeUnit.MILLISECONDS)) {
try {
long resSlot = group.firstResourceIndex.get();
- while (resSlot != -1) {
+ while (resSlot != NILL) {
int dsId = resArenaMgr.getDatasetId(resSlot);
int entityHashValue = resArenaMgr.getPkHashVal(resSlot);
long reqSlot = resArenaMgr.getLastHolder(resSlot);
- while (reqSlot != -1) {
+ while (reqSlot != NILL) {
byte lockMode = (byte) reqArenaMgr.getLockMode(reqSlot);
long jobSlot = reqArenaMgr.getJobSlot(reqSlot);
int jobId = jobArenaMgr.getJobId(jobSlot);
@@ -1026,7 +984,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
}
private void assertLockCanBeFoundInJobQueue(int dsId, int entityHashValue, byte lockMode, int jobId) {
- if (findLockInJobQueue(dsId, entityHashValue, jobId, lockMode) == -1) {
+ if (findLockInJobQueue(dsId, entityHashValue, jobId, lockMode) == NILL) {
String msg = "request for " + LockMode.toString(lockMode) + " lock on dataset " + dsId + " entity "
+ entityHashValue + " not found for job " + jobId + " in thread "
+ Thread.currentThread().getName();
@@ -1046,19 +1004,19 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
* job id
* @param lockMode
* lock mode
- * @return the slot of the request, if the lock request is found, -1 otherwise
+ * @return the slot of the request, if the lock request is found, NILL otherwise
*/
private long findLockInJobQueue(final int dsId, final int entityHashValue, final int jobId, byte lockMode) {
- Long jobSlot = jobIdSlotMap.get(jobId);
+ Long jobSlot = jobId2JobSlotMap.get(jobId);
if (jobSlot == null) {
- return -1;
+ return NILL;
}
long holder;
synchronized (jobArenaMgr) {
holder = jobArenaMgr.getLastHolder(jobSlot);
}
- while (holder != -1) {
+ while (holder != NILL) {
long resource = reqArenaMgr.getResourceId(holder);
if (dsId == resArenaMgr.getDatasetId(resource) && entityHashValue == resArenaMgr.getPkHashVal(resource)
&& jobSlot == reqArenaMgr.getJobSlot(holder)
@@ -1069,7 +1027,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
holder = reqArenaMgr.getNextJobRequest(holder);
}
}
- return -1;
+ return NILL;
}
private TablePrinter getResourceTablePrinter() {
@@ -1077,7 +1035,7 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
}
private TablePrinter getDumpTablePrinter() {
- return new DumpTablePrinter(table, resArenaMgr, reqArenaMgr, jobArenaMgr, jobIdSlotMap);
+ return new DumpTablePrinter(table, resArenaMgr, reqArenaMgr, jobArenaMgr, jobId2JobSlotMap);
}
public String printByResource() {
@@ -1115,46 +1073,4 @@ public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent
dumpState(os);
}
}
-
- private static class DatasetLockCache {
- private long jobId = -1;
- private HashMap<Integer, Byte> lockCache = new HashMap<Integer, Byte>();
- // size 1 cache to avoid the boxing/unboxing that comes with the
- // access to the HashMap
- private int cDsId = -1;
- private byte cDsLockMode = -1;
-
- public boolean contains(final int jobId, final int dsId, byte dsLockMode) {
- if (this.jobId == jobId) {
- if (this.cDsId == dsId && this.cDsLockMode == dsLockMode) {
- return true;
- }
- final Byte cachedLockMode = this.lockCache.get(dsId);
- if (cachedLockMode != null && cachedLockMode == dsLockMode) {
- this.cDsId = dsId;
- this.cDsLockMode = dsLockMode;
- return true;
- }
- } else {
- this.jobId = -1;
- this.cDsId = -1;
- this.cDsLockMode = -1;
- this.lockCache.clear();
- }
- return false;
- }
-
- public void put(final int jobId, final int dsId, byte dsLockMode) {
- this.jobId = jobId;
- this.cDsId = dsId;
- this.cDsLockMode = dsLockMode;
- this.lockCache.put(dsId, dsLockMode);
- }
-
- @Override
- public String toString() {
- return "[ " + jobId + " : " + lockCache.toString() + "]";
- }
- }
-
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/23be9068/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DatasetLockInfo.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DatasetLockInfo.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DatasetLockInfo.java
deleted file mode 100644
index dfb5d1e..0000000
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DatasetLockInfo.java
+++ /dev/null
@@ -1,536 +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.asterix.transaction.management.service.locking;
-
-import org.apache.asterix.transaction.management.service.transaction.TransactionManagementConstants.LockManagerConstants.LockMode;
-
-public class DatasetLockInfo {
- private EntityLockInfoManager entityLockInfoManager;
- private EntityInfoManager entityInfoManager;
- private LockWaiterManager lockWaiterManager;
- private PrimitiveIntHashMap entityResourceHT;
- private int IXCount;
- private int ISCount;
- private int XCount;
- private int SCount;
- private int lastHolder;
- private int firstWaiter;
- private int firstUpgrader;
-
- public DatasetLockInfo(EntityLockInfoManager entityLockInfoManager, EntityInfoManager entityInfoManager,
- LockWaiterManager lockWaiterManager) {
- this.entityLockInfoManager = entityLockInfoManager;
- this.entityInfoManager = entityInfoManager;
- this.lockWaiterManager = lockWaiterManager;
- entityResourceHT = new PrimitiveIntHashMap();
- lastHolder = -1; //-1 stands for end of list
- firstWaiter = -1;
- firstUpgrader = -1;
- }
-
- public void increaseLockCount(byte lockMode) {
- switch (lockMode) {
- case LockMode.IX:
- IXCount++;
- break;
- case LockMode.IS:
- ISCount++;
- break;
- case LockMode.X:
- XCount++;
- break;
- case LockMode.S:
- SCount++;
- break;
- default:
- throw new IllegalStateException("Invalid dataset lock mode");
- }
- }
-
- public void decreaseLockCount(byte lockMode) {
- switch (lockMode) {
- case LockMode.IX:
- IXCount--;
- break;
- case LockMode.IS:
- ISCount--;
- break;
- case LockMode.X:
- XCount--;
- break;
- case LockMode.S:
- SCount--;
- break;
- default:
- throw new IllegalStateException("Invalid dataset lock mode");
- }
- }
-
- public void increaseLockCount(byte lockMode, int count) {
- switch (lockMode) {
- case LockMode.IX:
- IXCount += count;
- break;
- case LockMode.IS:
- ISCount += count;
- break;
- case LockMode.X:
- XCount += count;
- break;
- case LockMode.S:
- SCount += count;
- break;
- default:
- throw new IllegalStateException("Invalid dataset lock mode");
- }
- }
-
- public void decreaseLockCount(byte lockMode, int count) {
- switch (lockMode) {
- case LockMode.IX:
- IXCount -= count;
- break;
- case LockMode.IS:
- ISCount -= count;
- break;
- case LockMode.X:
- XCount -= count;
- break;
- case LockMode.S:
- SCount -= count;
- break;
- default:
- throw new IllegalStateException("Invalid dataset lock mode");
- }
- }
-
- public boolean isUpgradeCompatible(byte lockMode, int entityInfo) {
- switch (lockMode) {
- //upgrade from IS -> IX
- //XCount is guaranteed to be 0.
- //upgrade is allowed if SCount is 0.
- case LockMode.IX:
- return SCount == 0;
-
- //upgrade from S -> X
- //XCount and IXCount are guaranteed to be 0.
- //upgrade is allowed if ISCount is 0.
- case LockMode.X:
- return ISCount == 0;
-
- default:
- throw new IllegalStateException("Invalid upgrade lock mode");
- }
- }
-
- public boolean isCompatible(byte lockMode) {
- switch (lockMode) {
- case LockMode.IX:
- return SCount == 0 && XCount == 0;
-
- case LockMode.IS:
- return XCount == 0;
-
- case LockMode.X:
- return ISCount == 0 && IXCount == 0 && SCount == 0 && XCount == 0;
-
- case LockMode.S:
- return IXCount == 0 && XCount == 0;
-
- default:
- throw new IllegalStateException("Invalid upgrade lock mode");
- }
- }
-
- public int findEntityInfoFromHolderList(int jobId, int hashVal) {
- int entityInfo;
- int eLockInfo;
- int waiterObjId;
- if (hashVal == -1) {//dataset-granule lock
- entityInfo = lastHolder;
- while (entityInfo != -1) {
- if (jobId == entityInfoManager.getJobId(entityInfo)) {
- return entityInfo;
- }
- entityInfo = entityInfoManager.getPrevEntityActor(entityInfo);
- }
- return -1;
- } else { //entity-granule lock
- eLockInfo = entityResourceHT.get(hashVal);
- if (eLockInfo == -1) {
- return -1;
- }
- entityInfo = entityLockInfoManager.findEntityInfoFromHolderList(eLockInfo, jobId, hashVal);
- if (entityInfo == -1) {
- //find the entityInfo from the waiter list of entityLockInfo.
- //There is a case where dataset-granule lock is acquired, but entity-granule lock is not acquired yet.
- //In this case, the waiter of the entityLockInfo represents the holder of the datasetLockInfo.
- waiterObjId = entityLockInfoManager.findWaiterFromWaiterList(eLockInfo, jobId, hashVal);
- if (waiterObjId != -1) {
- entityInfo = lockWaiterManager.getLockWaiter(waiterObjId).getEntityInfoSlot();
- }
- }
- return entityInfo;
- }
- }
-
- public int findWaiterFromWaiterList(int jobId, int hashVal) {
- int waiterObjId;
- LockWaiter waiterObj;
- int entityInfo = 0;
-
- waiterObjId = firstWaiter;
- while (waiterObjId != -1) {
- waiterObj = lockWaiterManager.getLockWaiter(waiterObjId);
- entityInfo = waiterObj.getEntityInfoSlot();
- if (jobId == entityInfoManager.getJobId(entityInfo)
- && hashVal == entityInfoManager.getPKHashVal(entityInfo)) {
- return waiterObjId;
- }
- waiterObjId = waiterObj.getNextWaiterObjId();
- }
-
- return -1;
- }
-
- public int findUpgraderFromUpgraderList(int jobId, int hashVal) {
- int waiterObjId;
- LockWaiter waiterObj;
- int entityInfo = 0;
-
- waiterObjId = firstUpgrader;
- while (waiterObjId != -1) {
- waiterObj = lockWaiterManager.getLockWaiter(waiterObjId);
- entityInfo = waiterObj.getEntityInfoSlot();
- if (jobId == entityInfoManager.getJobId(entityInfo)
- && hashVal == entityInfoManager.getPKHashVal(entityInfo)) {
- return waiterObjId;
- }
- waiterObjId = waiterObj.getNextWaiterObjId();
- }
-
- return -1;
- }
-
- public boolean isNoHolder() {
- return ISCount == 0 && IXCount == 0 && SCount == 0 && XCount == 0;
- }
-
- public void addHolder(int holder) {
- entityInfoManager.setPrevEntityActor(holder, lastHolder);
- lastHolder = holder;
- }
-
- /**
- * Remove holder from linked list of Actor.
- * Also, remove the corresponding resource from linked list of resource
- * in order to minimize JobInfo's resource link traversal.
- *
- * @param holder
- * @param jobInfo
- */
- public void removeHolder(int holder, JobInfo jobInfo) {
- int prev = lastHolder;
- int current = -1;
- int next;
-
- //remove holder from linked list of Actor
- while (prev != holder) {
- if (LockManager.IS_DEBUG_MODE) {
- if (prev == -1) {
- //shouldn't occur: debugging purpose
- try {
- throw new Exception();
- } catch (Exception e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
- }
- }
-
- current = prev;
- prev = entityInfoManager.getPrevEntityActor(current);
- }
-
- if (current != -1) {
- //current->prev = prev->prev
- entityInfoManager.setPrevEntityActor(current, entityInfoManager.getPrevEntityActor(prev));
- } else {
- //lastHolder = prev->prev
- lastHolder = entityInfoManager.getPrevEntityActor(prev);
- }
-
- //Notice!!
- //remove the corresponding resource from linked list of resource.
- //it is guaranteed that there is no waiter or upgrader in the JobInfo when this function is called.
- prev = entityInfoManager.getPrevJobResource(holder);
- next = entityInfoManager.getNextJobResource(holder);
-
- if (prev != -1) {
- entityInfoManager.setNextJobResource(prev, next);
- }
-
- if (next != -1) {
- entityInfoManager.setPrevJobResource(next, prev);
- } else {
- //This entityInfo(i.e., holder) is the last resource held by this job.
- jobInfo.setlastHoldingResource(holder);
- }
-
- //jobInfo.decreaseDatasetLockCount(holder);
- }
-
- /**
- * append new waiter to the end of waiters
- *
- * @param waiterObjId
- */
- public void addWaiter(int waiterObjId) {
- int lastObjId;
- LockWaiter lastObj = null;
-
- if (firstWaiter != -1) {
- //find the lastWaiter
- lastObjId = firstWaiter;
- while (lastObjId != -1) {
- lastObj = lockWaiterManager.getLockWaiter(lastObjId);
- lastObjId = lastObj.getNextWaiterObjId();
- }
- //last->next = new_waiter
- lastObj.setNextWaiterObjId(waiterObjId);
- } else {
- firstWaiter = waiterObjId;
- }
- //new_waiter->next = -1
- lastObj = lockWaiterManager.getLockWaiter(waiterObjId);
- lastObj.setNextWaiterObjId(-1);
-
- // if (LockManager.IS_DEBUG_MODE) {
- // System.out.println(printWaiters());
- // }
- }
-
- public void removeWaiter(int waiterObjId) {
- int currentObjId = firstWaiter;
- LockWaiter currentObj;
- LockWaiter prevObj = null;
- int prevObjId = -1;
- int nextObjId;
-
- while (currentObjId != waiterObjId) {
-
- if (LockManager.IS_DEBUG_MODE) {
- if (currentObjId == -1) {
- //shouldn't occur: debugging purpose
- try {
- throw new Exception();
- } catch (Exception e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
- }
- }
-
- prevObj = lockWaiterManager.getLockWaiter(currentObjId);
- prevObjId = currentObjId;
- currentObjId = prevObj.getNextWaiterObjId();
- }
-
- //get current waiter object
- currentObj = lockWaiterManager.getLockWaiter(currentObjId);
-
- //get next waiterObjId
- nextObjId = currentObj.getNextWaiterObjId();
-
- if (prevObjId != -1) {
- //prev->next = next
- prevObj.setNextWaiterObjId(nextObjId);
- } else {
- //removed first waiter. firstWaiter = current->next
- firstWaiter = nextObjId;
- }
-
- // if (LockManager.IS_DEBUG_MODE) {
- // System.out.println(printWaiters());
- // }
- }
-
- public void addUpgrader(int waiterObjId) {
- int lastObjId;
- LockWaiter lastObj = null;
-
- if (firstUpgrader != -1) {
- //find the lastWaiter
- lastObjId = firstUpgrader;
- while (lastObjId != -1) {
- lastObj = lockWaiterManager.getLockWaiter(lastObjId);
- lastObjId = lastObj.getNextWaiterObjId();
- }
- //last->next = new_waiter
- lastObj.setNextWaiterObjId(waiterObjId);
- } else {
- firstUpgrader = waiterObjId;
- }
- //new_waiter->next = -1
- lastObj = lockWaiterManager.getLockWaiter(waiterObjId);
- lastObj.setNextWaiterObjId(-1);
- }
-
- public void removeUpgrader(int waiterObjId) {
- int currentObjId = firstUpgrader;
- LockWaiter currentObj;
- LockWaiter prevObj = null;
- int prevObjId = -1;
- int nextObjId;
-
- while (currentObjId != waiterObjId) {
-
- if (LockManager.IS_DEBUG_MODE) {
- if (currentObjId == -1) {
- //shouldn't occur: debugging purpose
- try {
- throw new Exception();
- } catch (Exception e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
- }
- }
-
- prevObj = lockWaiterManager.getLockWaiter(currentObjId);
- prevObjId = currentObjId;
- currentObjId = prevObj.getNextWaiterObjId();
- }
-
- //get current waiter object
- currentObj = lockWaiterManager.getLockWaiter(currentObjId);
-
- //get next waiterObjId
- nextObjId = currentObj.getNextWaiterObjId();
-
- if (prevObjId != -1) {
- //prev->next = next
- prevObj.setNextWaiterObjId(nextObjId);
- } else {
- //removed first waiter. firstWaiter = current->next
- firstUpgrader = nextObjId;
- }
- }
-
- //debugging method
- public String printWaiters() {
- StringBuilder s = new StringBuilder();
- int waiterObjId;
- LockWaiter waiterObj;
- int entityInfo;
-
- s.append("WID\tWCT\tEID\tJID\tDID\tPK\n");
-
- waiterObjId = firstWaiter;
- while (waiterObjId != -1) {
- waiterObj = lockWaiterManager.getLockWaiter(waiterObjId);
- entityInfo = waiterObj.getEntityInfoSlot();
- s.append(waiterObjId).append("\t").append(waiterObj.getWaiterCount()).append("\t").append(entityInfo)
- .append("\t").append(entityInfoManager.getJobId(entityInfo)).append("\t")
- .append(entityInfoManager.getDatasetId(entityInfo)).append("\t")
- .append(entityInfoManager.getPKHashVal(entityInfo)).append("\n");
- waiterObjId = waiterObj.getNextWaiterObjId();
- }
-
- return s.toString();
- }
-
- public String coreDump() {
- StringBuilder sb = new StringBuilder();
- sb.append("\n\t firstUpgrader: " + firstUpgrader);
- sb.append("\n\t firstWaiter: " + firstWaiter);
- sb.append("\n\t lastHolder: " + lastHolder);
- sb.append("\n\t ISCount: " + ISCount);
- sb.append("\n\t IXCount: " + IXCount);
- sb.append("\n\t SCount: " + SCount);
- sb.append("\n\t XCount: " + XCount);
- sb.append("\n\t entityResourceHT");
- sb.append(entityResourceHT.prettyPrint());
- return sb.toString();
- }
-
- /////////////////////////////////////////////////////////
- // set/get method for private variable
- /////////////////////////////////////////////////////////
- public void setIXCount(int count) {
- IXCount = count;
- }
-
- public int getIXCount() {
- return IXCount;
- }
-
- public void setISCount(int count) {
- ISCount = count;
- }
-
- public int getISCount() {
- return ISCount;
- }
-
- public void setXCount(int count) {
- XCount = count;
- }
-
- public int getXCount() {
- return XCount;
- }
-
- public void setSCount(int count) {
- SCount = count;
- }
-
- public int getSCount() {
- return SCount;
- }
-
- public void setLastHolder(int holder) {
- lastHolder = holder;
- }
-
- public int getLastHolder() {
- return lastHolder;
- }
-
- public void setFirstWaiter(int waiter) {
- firstWaiter = waiter;
- }
-
- public int getFirstWaiter() {
- return firstWaiter;
- }
-
- public void setFirstUpgrader(int upgrader) {
- firstUpgrader = upgrader;
- }
-
- public int getFirstUpgrader() {
- return firstUpgrader;
- }
-
- public PrimitiveIntHashMap getEntityResourceHT() {
- return entityResourceHT;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/23be9068/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DeadlockDetector.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DeadlockDetector.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DeadlockDetector.java
deleted file mode 100644
index f8a4f5e..0000000
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DeadlockDetector.java
+++ /dev/null
@@ -1,255 +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.asterix.transaction.management.service.locking;
-
-import java.util.HashMap;
-
-import org.apache.asterix.common.transactions.DatasetId;
-import org.apache.asterix.common.transactions.JobId;
-
-/**
- * @author pouria, kisskys Performing a BFS search, upon adding each waiter to a waiting
- * list to avoid deadlocks this class implements such a loop-detector in
- * the wait-for-graph
- */
-
-public class DeadlockDetector {
-
- public static final boolean IS_DEBUG_MODE = true;//false
- private HashMap<JobId, JobInfo> jobHT;
- private HashMap<DatasetId, DatasetLockInfo> datasetResourceHT;
- private EntityLockInfoManager entityLockInfoManager;
- private EntityInfoManager entityInfoManager;
- private LockWaiterManager lockWaiterManager;
-
- private PrimitiveIntHashMap holderList;
- private PrimitiveIntHashMap nextHolderList;
- private PrimitiveIntHashMap resourceList;
- private PrimitiveIntHashMap visitedHolderList;
- private JobId tempJobIdObj; //temporary object to avoid object creation
- private DatasetId tempDatasetIdObj; //temporary object to avoid object creation
-
- public DeadlockDetector(HashMap<JobId, JobInfo> jobHT, HashMap<DatasetId, DatasetLockInfo> datasetResourceHT,
- EntityLockInfoManager entityLockInfoManager, EntityInfoManager entityInfoManager,
- LockWaiterManager lockWaiterManager) {
- this.jobHT = jobHT;
- this.datasetResourceHT = datasetResourceHT;
- this.entityLockInfoManager = entityLockInfoManager;
- this.entityInfoManager = entityInfoManager;
- this.lockWaiterManager = lockWaiterManager;
- holderList = new PrimitiveIntHashMap(1 << 6, 1 << 3, 180000);
- nextHolderList = new PrimitiveIntHashMap(1 << 6, 1 << 3, 180000);
- resourceList = new PrimitiveIntHashMap(1, 1 << 4, 180000);
- visitedHolderList = new PrimitiveIntHashMap(1 << 6, 1 << 3, 180000);
- tempJobIdObj = new JobId(0);
- tempDatasetIdObj = new DatasetId(0);
- }
-
- public boolean isSafeToAdd(DatasetLockInfo dLockInfo, int eLockInfo, int entityInfo, boolean isDatasetLockInfo,
- boolean isUpgrade) {
- int holder;
- int visitedHolder;
- int callerId = entityInfoManager.getJobId(entityInfo);
- int datasetId = entityInfoManager.getDatasetId(entityInfo);
- int hashValue = entityInfoManager.getPKHashVal(entityInfo);
- int resource;
- PrimitiveIntHashMap tempHolderList;
-
- holderList.clear(true);
- visitedHolderList.clear(true);
-
- //holderlist contains jobId
- //resourceList contains entityInfo's slot numbers instead of resourceId in order to avoid object creation
- //since resourceId consists of datasetId and PKHashValue
-
- //get holder list(jobId list)
- if (isDatasetLockInfo) {
- getHolderList(datasetId, -1, holderList);
- } else {
- getHolderList(datasetId, hashValue, holderList);
- }
-
- //check whether this caller is upgrader or not
- //if it is upgrader, then handle it as special case in the following manner
- //if there is another upgrader or waiter of which lock mode is not compatible with the caller's lock mode,
- //then this caller's wait causes deadlock.
- if (holderList.get(callerId) != -1) {
- if (isUpgrade && dLockInfo.getFirstUpgrader() != -1) {
- return false;
- }
- //there is no case such that while a job is holding any mode of lock on a dataset and waits for the same dataset as an waiter.
- //But the job may wait for the same dataset as an upgrader.
- }
-
- //TODO
- //check whether when there are multiple resources, the waiter and upgrader should be distinguished or not.
- //The current logic doesn't distinguish these two types of waiter.
-
- //while holderList is not empty
- holderList.beginIterate();
- holder = holderList.getNextKey();
- while (holder != -1) {
-
- nextHolderList.clear(true);
-
- while (holder != -1) {
- resourceList.clear(true);
- getWaitingResourceList(holder, resourceList);
- resourceList.beginIterate();
- resource = resourceList.getNextKey();
-
- while (resource != -1) {
- //get dataset holder
- getHolderList(entityInfoManager.getDatasetId(resource), -1, nextHolderList);
- //get entity holder
- getHolderList(entityInfoManager.getDatasetId(resource), entityInfoManager.getPKHashVal(resource),
- nextHolderList);
- if (nextHolderList.get(callerId) != -1) {
- return false;
- }
- resource = resourceList.getNextKey();
- }
-
- visitedHolderList.put(holder, -1);
- holder = holderList.getNextKey();
- }
-
- //remove visitedHolder for nextHolderList;
- visitedHolderList.beginIterate();
- visitedHolder = visitedHolderList.getNextKey();
- while (visitedHolder != -1) {
- nextHolderList.remove(visitedHolder);
- visitedHolder = visitedHolderList.getNextKey();
- }
-
- //swap holder list
- //set holderList to nextHolderList and nextHolderList to holderList
- tempHolderList = holderList;
- holderList = nextHolderList;
- nextHolderList = tempHolderList;
- holderList.beginIterate();
- holder = holderList.getNextKey();
- }
-
- return true;
- }
-
- /**
- * Get holder list of dataset if hashValue == -1. Get holder list of entity otherwise.
- * Where, a holder is a jobId, not entityInfo's slotNum
- *
- * @param datasetId
- * @param hashValue
- * @param holderList
- */
- private void getHolderList(int datasetId, int hashValue, PrimitiveIntHashMap holderList) {
- PrimitiveIntHashMap entityHT;
- DatasetLockInfo dLockInfo;
- int entityLockInfo;
- int entityInfo;
- int waiterObjId;
- LockWaiter waiterObj;
-
- //get datasetLockInfo
- tempDatasetIdObj.setId(datasetId);
- dLockInfo = datasetResourceHT.get(tempDatasetIdObj);
- if (dLockInfo == null) {
- return;
- }
-
- if (hashValue == -1) {
- //get S/X-lock holders of dataset
- entityInfo = dLockInfo.getLastHolder();
- while (entityInfo != -1) {
- holderList.put(entityInfoManager.getJobId(entityInfo), 0);
- entityInfo = entityInfoManager.getPrevEntityActor(entityInfo);
- }
-
- //get IS/IX-lock holders of dataset
- entityHT = dLockInfo.getEntityResourceHT();
- entityHT.beginIterate();
- entityLockInfo = entityHT.getNextValue();
- while (entityLockInfo != -1) {
-
- //1. add holder of eLockInfo to holerList
- entityInfo = entityLockInfoManager.getLastHolder(entityLockInfo);
- while (entityInfo != -1) {
- holderList.put(entityInfoManager.getJobId(entityInfo), 0);
- entityInfo = entityInfoManager.getPrevEntityActor(entityInfo);
- }
-
- //2. add waiter of eLockInfo to holderList since waiter of entityLock is a holder of datasetLock
- //(Upgraders need not to be added since upgraders are also holders)
- waiterObjId = entityLockInfoManager.getFirstWaiter(entityLockInfo);
- while (waiterObjId != -1) {
- waiterObj = lockWaiterManager.getLockWaiter(waiterObjId);
- entityInfo = waiterObj.getEntityInfoSlot();
- holderList.put(entityInfoManager.getJobId(entityInfo), 0);
- waiterObjId = waiterObj.getNextWaiterObjId();
- }
-
- entityLockInfo = entityHT.getNextValue();
- }
- } else {
- //get S/X-lock holders of entity
- entityHT = dLockInfo.getEntityResourceHT();
- entityLockInfo = entityHT.get(hashValue);
- if (entityLockInfo != -1) {
- entityInfo = entityLockInfoManager.getLastHolder(entityLockInfo);
- while (entityInfo != -1) {
- holderList.put(entityInfoManager.getJobId(entityInfo), 0);
- entityInfo = entityInfoManager.getPrevEntityActor(entityInfo);
- }
- }
- }
- return;
- }
-
- /**
- * Get waiting resource list of jobId, where a resource is represented with entityInfo's slot number
- *
- * @param jobId
- * @param resourceList
- */
- private void getWaitingResourceList(int jobId, PrimitiveIntHashMap resourceList) {
- JobInfo jobInfo;
- int waiterId;
- LockWaiter waiterObj;
- int entityInfo;
-
- //get JobInfo
- tempJobIdObj.setId(jobId);
- jobInfo = jobHT.get(tempJobIdObj);
- if (IS_DEBUG_MODE) {
- if (jobInfo == null) {
- System.out.println(Thread.currentThread().getName() + "jobId:" + jobId);
- }
- }
-
- //get WaiterObj
- waiterId = jobInfo.getFirstWaitingResource();
- while (waiterId != -1) {
- waiterObj = lockWaiterManager.getLockWaiter(waiterId);
- entityInfo = waiterObj.getEntityInfoSlot();
- resourceList.put(entityInfo, -1);
- waiterId = waiterObj.getNextWaitingResourceObjId();
- }
- return;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/23be9068/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DummyLockManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DummyLockManager.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DummyLockManager.java
deleted file mode 100644
index 8ffa775..0000000
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DummyLockManager.java
+++ /dev/null
@@ -1,93 +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.asterix.transaction.management.service.locking;
-
-import java.io.IOException;
-import java.io.OutputStream;
-
-import org.apache.asterix.common.exceptions.ACIDException;
-import org.apache.asterix.common.transactions.DatasetId;
-import org.apache.asterix.common.transactions.ILockManager;
-import org.apache.asterix.common.transactions.ITransactionContext;
-import org.apache.asterix.transaction.management.service.transaction.TransactionSubsystem;
-import org.apache.hyracks.api.lifecycle.ILifeCycleComponent;
-
-
-/**
- * A dummy implementation of the ILockManager interface. It assumes that all
- * requests are successful. It can be used to for jobs that are known to be
- * conflict free, but it'll yield terrible results if there are conflicts.
- *
- * @author tillw
- *
- */
-public class DummyLockManager implements ILockManager, ILifeCycleComponent {
-
- public DummyLockManager(TransactionSubsystem transactionSubsystem) {
- }
-
- @Override
- public void start() {
- }
-
- @Override
- public void stop(boolean dumpState, OutputStream ouputStream) throws IOException {
- }
-
- @Override
- public void lock(DatasetId datasetId, int entityHashValue, byte lockMode, ITransactionContext txnContext)
- throws ACIDException {
- }
-
- @Override
- public void releaseLocks(ITransactionContext txnContext) throws ACIDException {
- }
-
- @Override
- public void unlock(DatasetId datasetId, int entityHashValue, byte lockMode, ITransactionContext txnContext)
- throws ACIDException {
- }
-
- @Override
- public void instantLock(DatasetId datasetId, int entityHashValue, byte lockMode, ITransactionContext context)
- throws ACIDException {
- }
-
- @Override
- public boolean tryLock(DatasetId datasetId, int entityHashValue, byte lockMode, ITransactionContext context)
- throws ACIDException {
- return true;
- }
-
- @Override
- public boolean instantTryLock(DatasetId datasetId, int entityHashValue, byte lockMode,
- ITransactionContext txnContext) throws ACIDException {
- return true;
- }
-
- @Override
- public String prettyPrint() throws ACIDException {
- return "DummyLockManager";
- }
-
- @Override
- public void dumpState(OutputStream os) throws IOException {
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/23be9068/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DumpTablePrinter.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DumpTablePrinter.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DumpTablePrinter.java
index ffdb151..a75f756 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DumpTablePrinter.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/locking/DumpTablePrinter.java
@@ -26,18 +26,15 @@ public class DumpTablePrinter implements TablePrinter {
private ResourceArenaManager resArenaMgr;
private RequestArenaManager reqArenaMgr;
private JobArenaManager jobArenaMgr;
- private ConcurrentHashMap<Integer, Long> jobIdSlotMap;
+ private ConcurrentHashMap<Integer, Long> jobId2JobSlotMap;
- DumpTablePrinter(ResourceGroupTable table,
- ResourceArenaManager resArenaMgr,
- RequestArenaManager reqArenaMgr,
- JobArenaManager jobArenaMgr,
- ConcurrentHashMap<Integer, Long> jobIdSlotMap) {
+ DumpTablePrinter(ResourceGroupTable table, ResourceArenaManager resArenaMgr, RequestArenaManager reqArenaMgr,
+ JobArenaManager jobArenaMgr, ConcurrentHashMap<Integer, Long> jobId2JobSlotMap) {
this.table = table;
this.resArenaMgr = resArenaMgr;
this.reqArenaMgr = reqArenaMgr;
this.jobArenaMgr = jobArenaMgr;
- this.jobIdSlotMap = jobIdSlotMap;
+ this.jobId2JobSlotMap = jobId2JobSlotMap;
}
public StringBuilder append(StringBuilder sb) {
@@ -56,9 +53,9 @@ public class DumpTablePrinter implements TablePrinter {
sb.append(">>dump_end\t>>----- [reqArenaMgr] -----\n");
sb.append(">>dump_begin\t>>----- [jobIdSlotMap] -----\n");
- for (Integer i : jobIdSlotMap.keySet()) {
+ for (Integer i : jobId2JobSlotMap.keySet()) {
sb.append(i).append(" : ");
- TypeUtil.Global.append(sb, jobIdSlotMap.get(i));
+ TypeUtil.Global.append(sb, jobId2JobSlotMap.get(i));
sb.append("\n");
}
sb.append(">>dump_end\t>>----- [jobIdSlotMap] -----\n");