| /* |
| * Copyright 2009-2012 by The Regents of the University of California |
| * Licensed 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 from |
| * |
| * 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 edu.uci.ics.asterix.transaction.management.service.locking; |
| |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.concurrent.locks.ReadWriteLock; |
| import java.util.concurrent.locks.ReentrantReadWriteLock; |
| |
| import edu.uci.ics.asterix.transaction.management.exception.ACIDException; |
| import edu.uci.ics.asterix.transaction.management.service.logging.LogType; |
| import edu.uci.ics.asterix.transaction.management.service.logging.LogUtil; |
| import edu.uci.ics.asterix.transaction.management.service.logging.LogicalLogLocator; |
| import edu.uci.ics.asterix.transaction.management.service.transaction.DatasetId; |
| import edu.uci.ics.asterix.transaction.management.service.transaction.ITransactionManager.TransactionState; |
| import edu.uci.ics.asterix.transaction.management.service.transaction.JobId; |
| import edu.uci.ics.asterix.transaction.management.service.transaction.TransactionContext; |
| import edu.uci.ics.asterix.transaction.management.service.transaction.TransactionManagementConstants.LockManagerConstants.LockMode; |
| import edu.uci.ics.asterix.transaction.management.service.transaction.TransactionSubsystem; |
| |
| /** |
| * An implementation of the ILockManager interface for the |
| * specific case of locking protocol with two lock modes: (S) and (X), |
| * where S lock mode is shown by 0, and X lock mode is shown by 1. |
| * |
| * @author pouria, kisskys |
| */ |
| |
| public class LockManager implements ILockManager { |
| |
| public static final boolean IS_DEBUG_MODE = false;//true |
| |
| public static final boolean ALLOW_ESCALATE_FROM_ENTITY_TO_DATASET = true; |
| //Threshold must be greater than 1 and should be reasonably large enough not to escalate too soon. |
| public static final int ESCALATE_TRHESHOLD_ENTITY_TO_DATASET = 1000; |
| private static final int DO_ESCALATE = 0; |
| private static final int ESCALATED = 1; |
| private static final int DONOT_ESCALATE = 2; |
| |
| private TransactionSubsystem txnSubsystem; |
| |
| //all threads accessing to LockManager's tables such as jobHT and datasetResourceHT |
| //are serialized through LockTableLatch. All threads waiting the latch will be fairly served |
| //in FIFO manner when the latch is available. |
| private final ReadWriteLock lockTableLatch; |
| private final ReadWriteLock waiterLatch; |
| private HashMap<JobId, JobInfo> jobHT; |
| private HashMap<DatasetId, DatasetLockInfo> datasetResourceHT; |
| |
| private EntityLockInfoManager entityLockInfoManager; |
| private EntityInfoManager entityInfoManager; |
| private LockWaiterManager lockWaiterManager; |
| |
| private DeadlockDetector deadlockDetector; |
| private TimeOutDetector toutDetector; |
| private DatasetId tempDatasetIdObj; //temporary object to avoid object creation |
| |
| private int tryLockDatasetGranuleRevertOperation; |
| |
| private LockRequestTracker lockRequestTracker; //for debugging |
| private ConsecutiveWakeupContext consecutiveWakeupContext; |
| |
| //TODO |
| //This code should be taken care properly when there is a way to avoid doubling memory space for txnIds. |
| private LogicalLogLocator logicalLogLocator; |
| |
| public LockManager(TransactionSubsystem txnSubsystem) throws ACIDException { |
| this.txnSubsystem = txnSubsystem; |
| this.lockTableLatch = new ReentrantReadWriteLock(true); |
| this.waiterLatch = new ReentrantReadWriteLock(true); |
| this.jobHT = new HashMap<JobId, JobInfo>(); |
| this.datasetResourceHT = new HashMap<DatasetId, DatasetLockInfo>(); |
| this.entityInfoManager = new EntityInfoManager(); |
| this.lockWaiterManager = new LockWaiterManager(); |
| this.entityLockInfoManager = new EntityLockInfoManager(entityInfoManager, lockWaiterManager); |
| this.deadlockDetector = new DeadlockDetector(jobHT, datasetResourceHT, entityLockInfoManager, |
| entityInfoManager, lockWaiterManager); |
| this.toutDetector = new TimeOutDetector(this); |
| this.tempDatasetIdObj = new DatasetId(0); |
| this.consecutiveWakeupContext = new ConsecutiveWakeupContext(); |
| |
| this.logicalLogLocator = LogUtil.getDummyLogicalLogLocator(txnSubsystem.getLogManager()); |
| |
| if (IS_DEBUG_MODE) { |
| this.lockRequestTracker = new LockRequestTracker(); |
| } |
| } |
| |
| @Override |
| public void lock(DatasetId datasetId, int entityHashValue, byte lockMode, TransactionContext txnContext) |
| throws ACIDException { |
| internalLock(datasetId, entityHashValue, lockMode, txnContext, false); |
| } |
| |
| private void internalLock(DatasetId datasetId, int entityHashValue, byte lockMode, TransactionContext txnContext, |
| boolean isInstant) throws ACIDException { |
| |
| JobId jobId = txnContext.getJobId(); |
| int jId = jobId.getId(); //int-type jobId |
| int dId = datasetId.getId(); //int-type datasetId |
| int entityInfo; |
| int eLockInfo = -1; |
| DatasetLockInfo dLockInfo = null; |
| JobInfo jobInfo; |
| byte datasetLockMode = entityHashValue == -1 ? lockMode : lockMode == LockMode.S ? LockMode.IS : LockMode.IX; |
| boolean doEscalate = false; |
| |
| latchLockTable(); |
| validateJob(txnContext); |
| |
| if (IS_DEBUG_MODE) { |
| trackLockRequest("Requested", RequestType.LOCK, datasetId, entityHashValue, lockMode, txnContext, |
| dLockInfo, eLockInfo); |
| } |
| |
| dLockInfo = datasetResourceHT.get(datasetId); |
| jobInfo = jobHT.get(jobId); |
| |
| if (ALLOW_ESCALATE_FROM_ENTITY_TO_DATASET) { |
| if (!isInstant && datasetLockMode == LockMode.IS && jobInfo != null && dLockInfo != null) { |
| int escalateStatus = needEscalateFromEntityToDataset(jobInfo, dId, lockMode); |
| switch (escalateStatus) { |
| case DO_ESCALATE: |
| entityHashValue = -1; |
| doEscalate = true; |
| break; |
| |
| case ESCALATED: |
| unlatchLockTable(); |
| return; |
| |
| default: |
| break; |
| } |
| } |
| } |
| |
| //#. if the datasetLockInfo doesn't exist in datasetResourceHT |
| if (dLockInfo == null || dLockInfo.isNoHolder()) { |
| if (dLockInfo == null) { |
| dLockInfo = new DatasetLockInfo(entityLockInfoManager, entityInfoManager, lockWaiterManager); |
| datasetResourceHT.put(new DatasetId(dId), dLockInfo); //datsetId obj should be created |
| } |
| entityInfo = entityInfoManager.allocate(jId, dId, entityHashValue, lockMode); |
| |
| //if dataset-granule lock |
| if (entityHashValue == -1) { //-1 stands for dataset-granule |
| entityInfoManager.increaseDatasetLockCount(entityInfo); |
| dLockInfo.increaseLockCount(datasetLockMode); |
| dLockInfo.addHolder(entityInfo); |
| } else { |
| entityInfoManager.increaseDatasetLockCount(entityInfo); |
| dLockInfo.increaseLockCount(datasetLockMode); |
| //add entityLockInfo |
| eLockInfo = entityLockInfoManager.allocate(); |
| dLockInfo.getEntityResourceHT().put(entityHashValue, eLockInfo); |
| entityInfoManager.increaseEntityLockCount(entityInfo); |
| entityLockInfoManager.increaseLockCount(eLockInfo, lockMode); |
| entityLockInfoManager.addHolder(eLockInfo, entityInfo); |
| } |
| |
| if (jobInfo == null) { |
| jobInfo = new JobInfo(entityInfoManager, lockWaiterManager, txnContext); |
| jobHT.put(jobId, jobInfo); //jobId obj doesn't have to be created |
| } |
| jobInfo.addHoldingResource(entityInfo); |
| |
| if (ALLOW_ESCALATE_FROM_ENTITY_TO_DATASET) { |
| if (!isInstant && datasetLockMode == LockMode.IS) { |
| jobInfo.increaseDatasetISLockCount(dId); |
| if (doEscalate) { |
| throw new IllegalStateException("ESCALATE_TRHESHOLD_ENTITY_TO_DATASET should not be set to " |
| + ESCALATE_TRHESHOLD_ENTITY_TO_DATASET); |
| } |
| } |
| } |
| |
| if (IS_DEBUG_MODE) { |
| trackLockRequest("Granted", RequestType.LOCK, datasetId, entityHashValue, lockMode, txnContext, |
| dLockInfo, eLockInfo); |
| } |
| |
| unlatchLockTable(); |
| return; |
| } |
| |
| //#. the datasetLockInfo exists in datasetResourceHT. |
| //1. handle dataset-granule lock |
| entityInfo = lockDatasetGranule(datasetId, entityHashValue, lockMode, txnContext); |
| |
| //2. handle entity-granule lock |
| if (entityHashValue != -1) { |
| lockEntityGranule(datasetId, entityHashValue, lockMode, entityInfo, txnContext); |
| } |
| |
| if (ALLOW_ESCALATE_FROM_ENTITY_TO_DATASET) { |
| if (!isInstant) { |
| if (doEscalate) { |
| //jobInfo must not be null. |
| assert jobInfo != null; |
| jobInfo.increaseDatasetISLockCount(dId); |
| //release pre-acquired locks |
| releaseDatasetISLocks(jobInfo, jobId, datasetId, txnContext); |
| } else if (datasetLockMode == LockMode.IS) { |
| if (jobInfo == null) { |
| jobInfo = jobHT.get(jobId); |
| //jobInfo must not be null; |
| assert jobInfo != null; |
| } |
| jobInfo.increaseDatasetISLockCount(dId); |
| } |
| } |
| } |
| |
| if (IS_DEBUG_MODE) { |
| trackLockRequest("Granted", RequestType.LOCK, datasetId, entityHashValue, lockMode, txnContext, dLockInfo, |
| eLockInfo); |
| } |
| unlatchLockTable(); |
| return; |
| } |
| |
| private void releaseDatasetISLocks(JobInfo jobInfo, JobId jobId, DatasetId datasetId, TransactionContext txnContext) |
| throws ACIDException { |
| int entityInfo; |
| int prevEntityInfo; |
| int entityHashValue; |
| int did;//int-type dataset id |
| |
| //while traversing all holding resources, |
| //release IS locks on the escalated dataset and |
| //release S locks on the corresponding enttites |
| //by calling unlock() method. |
| entityInfo = jobInfo.getLastHoldingResource(); |
| while (entityInfo != -1) { |
| prevEntityInfo = entityInfoManager.getPrevJobResource(entityInfo); |
| |
| //release a lock only if the datset is the escalated dataset and |
| //the entityHashValue is not -1("not -1" means a non-dataset-level lock) |
| did = entityInfoManager.getDatasetId(entityInfo); |
| entityHashValue = entityInfoManager.getPKHashVal(entityInfo); |
| if (did == datasetId.getId() && entityHashValue != -1) { |
| this.unlock(datasetId, entityHashValue, txnContext); |
| } |
| |
| entityInfo = prevEntityInfo; |
| } |
| } |
| |
| private int needEscalateFromEntityToDataset(JobInfo jobInfo, int datasetId, byte lockMode) { |
| //we currently allow upgrade only if the lockMode is S. |
| if (lockMode != LockMode.S) { |
| return DONOT_ESCALATE; |
| } |
| |
| int count = jobInfo.getDatasetISLockCount(datasetId); |
| if (count == ESCALATE_TRHESHOLD_ENTITY_TO_DATASET) { |
| return DO_ESCALATE; |
| } else if (count > ESCALATE_TRHESHOLD_ENTITY_TO_DATASET) { |
| return ESCALATED; |
| } else { |
| return DONOT_ESCALATE; |
| } |
| } |
| |
| private void validateJob(TransactionContext txnContext) throws ACIDException { |
| if (txnContext.getTxnState() == TransactionState.ABORTED) { |
| unlatchLockTable(); |
| throw new ACIDException("" + txnContext.getJobId() + " is in ABORTED state."); |
| } else if (txnContext.getStatus() == TransactionContext.TIMED_OUT_STATUS) { |
| try { |
| requestAbort(txnContext); |
| } finally { |
| unlatchLockTable(); |
| } |
| } |
| } |
| |
| private int lockDatasetGranule(DatasetId datasetId, int entityHashValue, byte lockMode, |
| TransactionContext txnContext) throws ACIDException { |
| JobId jobId = txnContext.getJobId(); |
| int jId = jobId.getId(); //int-type jobId |
| int dId = datasetId.getId(); //int-type datasetId |
| int waiterObjId; |
| int entityInfo = -1; |
| DatasetLockInfo dLockInfo; |
| JobInfo jobInfo; |
| boolean isUpgrade = false; |
| int weakerModeLockCount; |
| int waiterCount = 0; |
| byte datasetLockMode = entityHashValue == -1 ? lockMode : lockMode == LockMode.S ? LockMode.IS : LockMode.IX; |
| |
| dLockInfo = datasetResourceHT.get(datasetId); |
| jobInfo = jobHT.get(jobId); |
| |
| //check duplicated call |
| |
| //1. lock request causing duplicated upgrading requests from different threads in a same job |
| waiterObjId = dLockInfo.findUpgraderFromUpgraderList(jId, entityHashValue); |
| if (waiterObjId != -1) { |
| //make the caller wait on the same LockWaiter object |
| entityInfo = lockWaiterManager.getLockWaiter(waiterObjId).getEntityInfoSlot(); |
| waiterCount = handleLockWaiter(dLockInfo, -1, entityInfo, true, true, txnContext, jobInfo, waiterObjId); |
| |
| //Only for the first-get-up thread, the waiterCount will be more than 0 and |
| //the thread updates lock count on behalf of the all other waiting threads. |
| //Therefore, all the next-get-up threads will not update any lock count. |
| if (waiterCount > 0) { |
| //add ((the number of waiting upgrader) - 1) to entityInfo's dataset lock count and datasetLockInfo's lock count |
| //where -1 is for not counting the first upgrader's request since the lock count for the first upgrader's request |
| //is already counted. |
| weakerModeLockCount = entityInfoManager.getDatasetLockCount(entityInfo); |
| entityInfoManager.setDatasetLockMode(entityInfo, lockMode); |
| entityInfoManager.increaseDatasetLockCount(entityInfo, waiterCount - 1); |
| |
| if (entityHashValue == -1) { //dataset-granule lock |
| dLockInfo.increaseLockCount(LockMode.X, weakerModeLockCount + waiterCount - 1);//new lock mode |
| dLockInfo.decreaseLockCount(LockMode.S, weakerModeLockCount);//current lock mode |
| } else { |
| dLockInfo.increaseLockCount(LockMode.IX, weakerModeLockCount + waiterCount - 1); |
| dLockInfo.decreaseLockCount(LockMode.IS, weakerModeLockCount); |
| } |
| } |
| |
| return entityInfo; |
| } |
| |
| //2. lock request causing duplicated waiting requests from different threads in a same job |
| waiterObjId = dLockInfo.findWaiterFromWaiterList(jId, entityHashValue); |
| if (waiterObjId != -1) { |
| //make the caller wait on the same LockWaiter object |
| entityInfo = lockWaiterManager.getLockWaiter(waiterObjId).getEntityInfoSlot(); |
| waiterCount = handleLockWaiter(dLockInfo, -1, entityInfo, false, true, txnContext, jobInfo, waiterObjId); |
| |
| if (waiterCount > 0) { |
| entityInfoManager.increaseDatasetLockCount(entityInfo, waiterCount); |
| if (entityHashValue == -1) { |
| dLockInfo.increaseLockCount(datasetLockMode, waiterCount); |
| dLockInfo.addHolder(entityInfo); |
| } else { |
| dLockInfo.increaseLockCount(datasetLockMode, waiterCount); |
| //IS and IX holders are implicitly handled. |
| } |
| //add entityInfo to JobInfo's holding-resource list |
| jobInfo.addHoldingResource(entityInfo); |
| } |
| |
| return entityInfo; |
| } |
| |
| //3. lock request causing duplicated holding requests from different threads or a single thread in a same job |
| entityInfo = dLockInfo.findEntityInfoFromHolderList(jId, entityHashValue); |
| if (entityInfo == -1) { |
| |
| entityInfo = entityInfoManager.allocate(jId, dId, entityHashValue, lockMode); |
| if (jobInfo == null) { |
| jobInfo = new JobInfo(entityInfoManager, lockWaiterManager, txnContext); |
| jobHT.put(jobId, jobInfo); |
| } |
| |
| //wait if any upgrader exists or upgrading lock mode is not compatible |
| if (dLockInfo.getFirstUpgrader() != -1 || dLockInfo.getFirstWaiter() != -1 |
| || !dLockInfo.isCompatible(datasetLockMode)) { |
| |
| ///////////////////////////////////////////////////////////////////////////////////////////// |
| //[Notice] Mimicking SIX mode |
| //When the lock escalation from IS to S in dataset-level is allowed, the following case occurs |
| //DatasetLockInfo's SCount = 1 and the same job who carried out the escalation tries to insert, |
| //then the job should be able to insert without being blocked by itself. |
| //Our approach is to introduce SIX mode, but we don't have currently, |
| //so I simply mimicking SIX by allowing S and IX coexist in the dataset level |
| //only if their job id is identical for the requests. |
| if (ALLOW_ESCALATE_FROM_ENTITY_TO_DATASET) { |
| if (datasetLockMode == LockMode.IX && dLockInfo.getSCount() == 1 |
| && jobInfo.isDatasetLockGranted(dId, LockMode.S)) { |
| entityInfoManager.increaseDatasetLockCount(entityInfo); |
| //IX holders are implicitly handled without adding holder |
| dLockInfo.increaseLockCount(datasetLockMode); |
| //add entityInfo to JobInfo's holding-resource list |
| jobInfo.addHoldingResource(entityInfo); |
| return entityInfo; |
| } |
| } |
| /////////////////////////////////////////////////////////////////////////////////////////////// |
| |
| ///////////////////////////////////////////////////////////////////////////////////////////// |
| //[Notice] |
| //There has been no same caller as (jId, dId, entityHashValue) triplet. |
| //But there could be the same caller as (jId, dId) pair. |
| //For example, two requests (J1, D1, E1) and (J1, D1, E2) are considered as duplicated call in dataset-granule perspective. |
| //Therefore, the above duplicated call case is covered in the following code. |
| //find the same dataset-granule lock request, that is, (J1, D1) pair in the above example. |
| //if (jobInfo.isDatasetLockGranted(dId, datasetLockMode)) { |
| if (jobInfo.isDatasetLockGranted(dId, LockMode.IS)) { |
| if (dLockInfo.isCompatible(datasetLockMode)) { |
| //this is duplicated call |
| entityInfoManager.increaseDatasetLockCount(entityInfo); |
| if (entityHashValue == -1) { |
| dLockInfo.increaseLockCount(datasetLockMode); |
| dLockInfo.addHolder(entityInfo); |
| } else { |
| dLockInfo.increaseLockCount(datasetLockMode); |
| //IS and IX holders are implicitly handled. |
| } |
| //add entityInfo to JobInfo's holding-resource list |
| jobInfo.addHoldingResource(entityInfo); |
| |
| return entityInfo; |
| } else { |
| //considered as upgrader |
| waiterCount = handleLockWaiter(dLockInfo, -1, entityInfo, true, true, txnContext, jobInfo, -1); |
| if (waiterCount > 0) { |
| entityInfoManager.increaseDatasetLockCount(entityInfo); |
| if (entityHashValue == -1) { |
| dLockInfo.increaseLockCount(datasetLockMode); |
| dLockInfo.addHolder(entityInfo); |
| } else { |
| dLockInfo.increaseLockCount(datasetLockMode); |
| //IS and IX holders are implicitly handled. |
| } |
| //add entityInfo to JobInfo's holding-resource list |
| jobInfo.addHoldingResource(entityInfo); |
| } |
| return entityInfo; |
| } |
| } |
| ///////////////////////////////////////////////////////////////////////////////////////////// |
| |
| waiterCount = handleLockWaiter(dLockInfo, -1, entityInfo, false, true, txnContext, jobInfo, -1); |
| } else { |
| waiterCount = 1; |
| } |
| |
| if (waiterCount > 0) { |
| entityInfoManager.increaseDatasetLockCount(entityInfo); |
| if (entityHashValue == -1) { |
| dLockInfo.increaseLockCount(datasetLockMode); |
| dLockInfo.addHolder(entityInfo); |
| } else { |
| dLockInfo.increaseLockCount(datasetLockMode); |
| //IS and IX holders are implicitly handled. |
| } |
| //add entityInfo to JobInfo's holding-resource list |
| jobInfo.addHoldingResource(entityInfo); |
| } |
| } else { |
| isUpgrade = isLockUpgrade(entityInfoManager.getDatasetLockMode(entityInfo), lockMode); |
| if (isUpgrade) { //upgrade call |
| //wait if any upgrader exists or upgrading lock mode is not compatible |
| if (dLockInfo.getFirstUpgrader() != -1 || !dLockInfo.isUpgradeCompatible(datasetLockMode, entityInfo)) { |
| waiterCount = handleLockWaiter(dLockInfo, -1, entityInfo, true, true, txnContext, jobInfo, -1); |
| } else { |
| waiterCount = 1; |
| } |
| |
| if (waiterCount > 0) { |
| //add ((the number of waiting upgrader) - 1) to entityInfo's dataset lock count and datasetLockInfo's lock count |
| //where -1 is for not counting the first upgrader's request since the lock count for the first upgrader's request |
| //is already counted. |
| weakerModeLockCount = entityInfoManager.getDatasetLockCount(entityInfo); |
| entityInfoManager.setDatasetLockMode(entityInfo, lockMode); |
| entityInfoManager.increaseDatasetLockCount(entityInfo, waiterCount - 1); |
| |
| if (entityHashValue == -1) { //dataset-granule lock |
| dLockInfo.increaseLockCount(LockMode.X, weakerModeLockCount + waiterCount - 1);//new lock mode |
| dLockInfo.decreaseLockCount(LockMode.S, weakerModeLockCount);//current lock mode |
| } else { |
| dLockInfo.increaseLockCount(LockMode.IX, weakerModeLockCount + waiterCount - 1); |
| dLockInfo.decreaseLockCount(LockMode.IS, weakerModeLockCount); |
| } |
| } |
| } else { //duplicated call |
| entityInfoManager.increaseDatasetLockCount(entityInfo); |
| datasetLockMode = entityInfoManager.getDatasetLockMode(entityInfo); |
| |
| if (entityHashValue == -1) { //dataset-granule |
| dLockInfo.increaseLockCount(datasetLockMode); |
| } else { //entity-granule |
| datasetLockMode = datasetLockMode == LockMode.S ? LockMode.IS : LockMode.IX; |
| dLockInfo.increaseLockCount(datasetLockMode); |
| } |
| } |
| } |
| |
| return entityInfo; |
| } |
| |
| private void lockEntityGranule(DatasetId datasetId, int entityHashValue, byte lockMode, |
| int entityInfoFromDLockInfo, TransactionContext txnContext) throws ACIDException { |
| JobId jobId = txnContext.getJobId(); |
| int jId = jobId.getId(); //int-type jobId |
| int waiterObjId; |
| int eLockInfo = -1; |
| int entityInfo; |
| DatasetLockInfo dLockInfo; |
| JobInfo jobInfo; |
| boolean isUpgrade = false; |
| int waiterCount = 0; |
| int weakerModeLockCount; |
| |
| dLockInfo = datasetResourceHT.get(datasetId); |
| jobInfo = jobHT.get(jobId); |
| eLockInfo = dLockInfo.getEntityResourceHT().get(entityHashValue); |
| |
| if (eLockInfo != -1) { |
| //check duplicated call |
| |
| //1. lock request causing duplicated upgrading requests from different threads in a same job |
| waiterObjId = entityLockInfoManager.findUpgraderFromUpgraderList(eLockInfo, jId, entityHashValue); |
| if (waiterObjId != -1) { |
| entityInfo = lockWaiterManager.getLockWaiter(waiterObjId).getEntityInfoSlot(); |
| waiterCount = handleLockWaiter(dLockInfo, eLockInfo, -1, true, false, txnContext, jobInfo, waiterObjId); |
| |
| if (waiterCount > 0) { |
| weakerModeLockCount = entityInfoManager.getEntityLockCount(entityInfo); |
| entityInfoManager.setEntityLockMode(entityInfo, LockMode.X); |
| entityInfoManager.increaseEntityLockCount(entityInfo, waiterCount - 1); |
| |
| entityLockInfoManager.increaseLockCount(eLockInfo, LockMode.X, (short) (weakerModeLockCount |
| + waiterCount - 1));//new lock mode |
| entityLockInfoManager.decreaseLockCount(eLockInfo, LockMode.S, (short) weakerModeLockCount);//old lock mode |
| } |
| return; |
| } |
| |
| //2. lock request causing duplicated waiting requests from different threads in a same job |
| waiterObjId = entityLockInfoManager.findWaiterFromWaiterList(eLockInfo, jId, entityHashValue); |
| if (waiterObjId != -1) { |
| entityInfo = lockWaiterManager.getLockWaiter(waiterObjId).getEntityInfoSlot(); |
| waiterCount = handleLockWaiter(dLockInfo, eLockInfo, -1, false, false, txnContext, jobInfo, waiterObjId); |
| |
| if (waiterCount > 0) { |
| entityInfoManager.increaseEntityLockCount(entityInfo, waiterCount); |
| entityLockInfoManager.increaseLockCount(eLockInfo, lockMode, (short) waiterCount); |
| entityLockInfoManager.addHolder(eLockInfo, entityInfo); |
| } |
| return; |
| } |
| |
| //3. lock request causing duplicated holding requests from different threads or a single thread in a same job |
| entityInfo = entityLockInfoManager.findEntityInfoFromHolderList(eLockInfo, jId, entityHashValue); |
| if (entityInfo != -1) {//duplicated call or upgrader |
| |
| isUpgrade = isLockUpgrade(entityInfoManager.getEntityLockMode(entityInfo), lockMode); |
| if (isUpgrade) {//upgrade call |
| //wait if any upgrader exists or upgrading lock mode is not compatible |
| if (entityLockInfoManager.getUpgrader(eLockInfo) != -1 |
| || !entityLockInfoManager.isUpgradeCompatible(eLockInfo, lockMode, entityInfo)) { |
| waiterCount = handleLockWaiter(dLockInfo, eLockInfo, entityInfo, true, false, txnContext, |
| jobInfo, -1); |
| } else { |
| waiterCount = 1; |
| } |
| |
| if (waiterCount > 0) { |
| weakerModeLockCount = entityInfoManager.getEntityLockCount(entityInfo); |
| entityInfoManager.setEntityLockMode(entityInfo, lockMode); |
| entityInfoManager.increaseEntityLockCount(entityInfo, waiterCount - 1); |
| |
| entityLockInfoManager.increaseLockCount(eLockInfo, LockMode.X, (short) (weakerModeLockCount |
| + waiterCount - 1));//new lock mode |
| entityLockInfoManager.decreaseLockCount(eLockInfo, LockMode.S, (short) weakerModeLockCount);//old lock mode |
| } |
| |
| } else {//duplicated call |
| entityInfoManager.increaseEntityLockCount(entityInfo); |
| entityLockInfoManager.increaseLockCount(eLockInfo, entityInfoManager.getEntityLockMode(entityInfo)); |
| } |
| } else {//new call from this job, but still eLockInfo exists since other threads hold it or wait on it |
| entityInfo = entityInfoFromDLockInfo; |
| if (entityLockInfoManager.getUpgrader(eLockInfo) != -1 |
| || entityLockInfoManager.getFirstWaiter(eLockInfo) != -1 |
| || !entityLockInfoManager.isCompatible(eLockInfo, lockMode)) { |
| waiterCount = handleLockWaiter(dLockInfo, eLockInfo, entityInfo, false, false, txnContext, jobInfo, |
| -1); |
| } else { |
| waiterCount = 1; |
| } |
| |
| if (waiterCount > 0) { |
| entityInfoManager.increaseEntityLockCount(entityInfo, waiterCount); |
| entityLockInfoManager.increaseLockCount(eLockInfo, lockMode, (short) waiterCount); |
| entityLockInfoManager.addHolder(eLockInfo, entityInfo); |
| } |
| } |
| } else {//eLockInfo doesn't exist, so this lock request is the first request and can be granted without waiting. |
| eLockInfo = entityLockInfoManager.allocate(); |
| dLockInfo.getEntityResourceHT().put(entityHashValue, eLockInfo); |
| entityInfoManager.increaseEntityLockCount(entityInfoFromDLockInfo); |
| entityLockInfoManager.increaseLockCount(eLockInfo, lockMode); |
| entityLockInfoManager.addHolder(eLockInfo, entityInfoFromDLockInfo); |
| } |
| } |
| |
| @Override |
| public void unlock(DatasetId datasetId, int entityHashValue, TransactionContext txnContext) throws ACIDException { |
| internalUnlock(datasetId, entityHashValue, txnContext, false, false); |
| } |
| |
| @Override |
| public void unlock(DatasetId datasetId, int entityHashValue, TransactionContext txnContext, boolean commitFlag) |
| throws ACIDException { |
| internalUnlock(datasetId, entityHashValue, txnContext, false, commitFlag); |
| } |
| |
| private void instantUnlock(DatasetId datasetId, int entityHashValue, TransactionContext txnContext) throws ACIDException { |
| internalUnlock(datasetId, entityHashValue, txnContext, true, false); |
| } |
| |
| private void internalUnlock(DatasetId datasetId, int entityHashValue, TransactionContext txnContext, |
| boolean isInstant, boolean commitFlag) throws ACIDException { |
| JobId jobId = txnContext.getJobId(); |
| int eLockInfo = -1; |
| DatasetLockInfo dLockInfo = null; |
| JobInfo jobInfo; |
| int entityInfo = -1; |
| byte datasetLockMode; |
| |
| if (IS_DEBUG_MODE) { |
| if (entityHashValue == -1) { |
| throw new UnsupportedOperationException( |
| "Unsupported unlock request: dataset-granule unlock is not supported"); |
| } |
| } |
| |
| latchLockTable(); |
| validateJob(txnContext); |
| |
| if (IS_DEBUG_MODE) { |
| trackLockRequest("Requested", RequestType.UNLOCK, datasetId, entityHashValue, (byte) 0, txnContext, |
| dLockInfo, eLockInfo); |
| } |
| |
| //find the resource to be unlocked |
| dLockInfo = datasetResourceHT.get(datasetId); |
| jobInfo = jobHT.get(jobId); |
| if (dLockInfo == null || jobInfo == null) { |
| unlatchLockTable(); |
| throw new IllegalStateException("Invalid unlock request: Corresponding lock info doesn't exist."); |
| } |
| |
| eLockInfo = dLockInfo.getEntityResourceHT().get(entityHashValue); |
| |
| if (eLockInfo == -1) { |
| unlatchLockTable(); |
| throw new IllegalStateException("Invalid unlock request: Corresponding lock info doesn't exist."); |
| } |
| |
| //find the corresponding entityInfo |
| entityInfo = entityLockInfoManager.findEntityInfoFromHolderList(eLockInfo, jobId.getId(), entityHashValue); |
| if (entityInfo == -1) { |
| unlatchLockTable(); |
| throw new IllegalStateException("Invalid unlock request[" + jobId.getId() + "," + datasetId.getId() + "," |
| + entityHashValue + "]: Corresponding lock info doesn't exist."); |
| } |
| |
| datasetLockMode = entityInfoManager.getDatasetLockMode(entityInfo) == LockMode.S ? LockMode.IS : LockMode.IX; |
| |
| //decrease the corresponding count of dLockInfo/eLockInfo/entityInfo |
| dLockInfo.decreaseLockCount(datasetLockMode); |
| entityLockInfoManager.decreaseLockCount(eLockInfo, entityInfoManager.getEntityLockMode(entityInfo)); |
| entityInfoManager.decreaseDatasetLockCount(entityInfo); |
| entityInfoManager.decreaseEntityLockCount(entityInfo); |
| |
| if (entityInfoManager.getEntityLockCount(entityInfo) == 0 |
| && entityInfoManager.getDatasetLockCount(entityInfo) == 0) { |
| int threadCount = 0; //number of threads(in the same job) waiting on the same resource |
| int waiterObjId = jobInfo.getFirstWaitingResource(); |
| int waitingEntityInfo; |
| LockWaiter waiterObj; |
| |
| //TODO |
| //This code should be taken care properly when there is a way to avoid doubling memory space for txnIds. |
| //This commit log is written here in order to avoid increasing the memory space for managing transactionIds |
| if (commitFlag) { |
| if (txnContext.getTransactionType().equals(TransactionContext.TransactionType.READ_WRITE)) { |
| try { |
| txnSubsystem.getLogManager().log(LogType.ENTITY_COMMIT, txnContext, datasetId.getId(), |
| entityHashValue, -1, (byte) 0, 0, null, null, logicalLogLocator); |
| } catch (ACIDException e) { |
| try { |
| requestAbort(txnContext); |
| } finally { |
| unlatchLockTable(); |
| } |
| } |
| } |
| |
| txnContext.updateLastLSNForIndexes(logicalLogLocator.getLsn()); |
| } |
| |
| //1) wake up waiters and remove holder |
| //wake up waiters of dataset-granule lock |
| wakeUpDatasetLockWaiters(dLockInfo); |
| //wake up waiters of entity-granule lock |
| wakeUpEntityLockWaiters(eLockInfo); |
| //remove the holder from eLockInfo's holder list and remove the holding resource from jobInfo's holding resource list |
| //this can be done in the following single function call. |
| entityLockInfoManager.removeHolder(eLockInfo, entityInfo, jobInfo); |
| |
| //2) if |
| // there is no waiting thread on the same resource (this can be checked through jobInfo) |
| // then |
| // a) delete the corresponding entityInfo |
| // b) write commit log for the unlocked resource(which is a committed txn). |
| while (waiterObjId != -1) { |
| waiterObj = lockWaiterManager.getLockWaiter(waiterObjId); |
| waitingEntityInfo = waiterObj.getEntityInfoSlot(); |
| if (entityInfoManager.getDatasetId(waitingEntityInfo) == datasetId.getId() |
| && entityInfoManager.getPKHashVal(waitingEntityInfo) == entityHashValue) { |
| threadCount++; |
| break; |
| } |
| waiterObjId = waiterObj.getNextWaiterObjId(); |
| } |
| if (threadCount == 0) { |
| if (entityInfoManager.getEntityLockMode(entityInfo) == LockMode.X) { |
| //TODO |
| //write a commit log for the unlocked resource |
| //need to figure out that instantLock() also needs to write a commit log. |
| } |
| entityInfoManager.deallocate(entityInfo); |
| } |
| } |
| |
| //deallocate entityLockInfo's slot if there is no txn referring to the entityLockInfo. |
| if (entityLockInfoManager.getFirstWaiter(eLockInfo) == -1 |
| && entityLockInfoManager.getLastHolder(eLockInfo) == -1 |
| && entityLockInfoManager.getUpgrader(eLockInfo) == -1) { |
| dLockInfo.getEntityResourceHT().remove(entityHashValue); |
| entityLockInfoManager.deallocate(eLockInfo); |
| } |
| |
| //we don't deallocate datasetLockInfo even if there is no txn referring to the datasetLockInfo |
| //since the datasetLockInfo is likely to be referred to again. |
| |
| if (ALLOW_ESCALATE_FROM_ENTITY_TO_DATASET) { |
| if (!isInstant && datasetLockMode == LockMode.IS) { |
| jobInfo.decreaseDatasetISLockCount(datasetId.getId()); |
| } |
| } |
| |
| if (IS_DEBUG_MODE) { |
| trackLockRequest("Granted", RequestType.UNLOCK, datasetId, entityHashValue, (byte) 0, txnContext, |
| dLockInfo, eLockInfo); |
| } |
| unlatchLockTable(); |
| } |
| |
| @Override |
| public void releaseLocks(TransactionContext txnContext) throws ACIDException { |
| LockWaiter waiterObj; |
| int entityInfo; |
| int prevEntityInfo; |
| int entityHashValue; |
| DatasetLockInfo dLockInfo = null; |
| int eLockInfo = -1; |
| int did;//int-type dataset id |
| int datasetLockCount; |
| int entityLockCount; |
| byte lockMode; |
| boolean existWaiter = false; |
| |
| JobId jobId = txnContext.getJobId(); |
| |
| latchLockTable(); |
| |
| if (IS_DEBUG_MODE) { |
| trackLockRequest("Requested", RequestType.RELEASE_LOCKS, new DatasetId(0), 0, (byte) 0, txnContext, |
| dLockInfo, eLockInfo); |
| } |
| |
| JobInfo jobInfo = jobHT.get(jobId); |
| if (jobInfo == null) { |
| unlatchLockTable(); |
| return; |
| } |
| |
| //remove waiterObj of JobInfo |
| //[Notice] |
| //waiterObjs may exist if aborted thread is the caller of this function. |
| //Even if there are the waiterObjs, there is no waiting thread on the objects. |
| //If the caller of this function is an aborted thread, it is guaranteed that there is no waiting threads |
| //on the waiterObjs since when the aborted caller thread is waken up, all other waiting threads are |
| //also waken up at the same time through 'notifyAll()' call. |
| //In contrast, if the caller of this function is not an aborted thread, then there is no waiting object. |
| int waiterObjId = jobInfo.getFirstWaitingResource(); |
| int nextWaiterObjId; |
| while (waiterObjId != -1) { |
| existWaiter = true; |
| waiterObj = lockWaiterManager.getLockWaiter(waiterObjId); |
| nextWaiterObjId = waiterObj.getNextWaitingResourceObjId(); |
| entityInfo = waiterObj.getEntityInfoSlot(); |
| if (IS_DEBUG_MODE) { |
| if (jobId.getId() != entityInfoManager.getJobId(entityInfo)) { |
| throw new IllegalStateException("JobInfo(" + jobId + ") has diffrent Job(JID:" |
| + entityInfoManager.getJobId(entityInfo) + "'s lock request!!!"); |
| } |
| } |
| |
| //1. remove from waiter(or upgrader)'s list of dLockInfo or eLockInfo. |
| did = entityInfoManager.getDatasetId(entityInfo); |
| tempDatasetIdObj.setId(did); |
| dLockInfo = datasetResourceHT.get(tempDatasetIdObj); |
| |
| if (waiterObj.isWaitingOnEntityLock()) { |
| entityHashValue = entityInfoManager.getPKHashVal(entityInfo); |
| eLockInfo = dLockInfo.getEntityResourceHT().get(entityHashValue); |
| if (waiterObj.isWaiter()) { |
| entityLockInfoManager.removeWaiter(eLockInfo, waiterObjId); |
| } else { |
| entityLockInfoManager.removeUpgrader(eLockInfo, waiterObjId); |
| } |
| } else { |
| if (waiterObj.isWaiter()) { |
| dLockInfo.removeWaiter(waiterObjId); |
| } else { |
| dLockInfo.removeUpgrader(waiterObjId); |
| } |
| } |
| |
| //2. wake-up waiters |
| latchWaitNotify(); |
| synchronized (waiterObj) { |
| unlatchWaitNotify(); |
| waiterObj.setWait(false); |
| if (IS_DEBUG_MODE) { |
| System.out.println("" + Thread.currentThread().getName() + "\twake-up(D): WID(" + waiterObjId |
| + "),EID(" + waiterObj.getEntityInfoSlot() + ")"); |
| } |
| waiterObj.notifyAll(); |
| } |
| |
| //3. deallocate waiterObj |
| lockWaiterManager.deallocate(waiterObjId); |
| |
| //4. deallocate entityInfo only if this waiter is not an upgrader |
| if (entityInfoManager.getDatasetLockCount(entityInfo) == 0 |
| && entityInfoManager.getEntityLockCount(entityInfo) == 0) { |
| entityInfoManager.deallocate(entityInfo); |
| } |
| waiterObjId = nextWaiterObjId; |
| } |
| |
| //release holding resources |
| entityInfo = jobInfo.getLastHoldingResource(); |
| while (entityInfo != -1) { |
| prevEntityInfo = entityInfoManager.getPrevJobResource(entityInfo); |
| |
| //decrease lock count of datasetLock and entityLock |
| did = entityInfoManager.getDatasetId(entityInfo); |
| tempDatasetIdObj.setId(did); |
| dLockInfo = datasetResourceHT.get(tempDatasetIdObj); |
| entityHashValue = entityInfoManager.getPKHashVal(entityInfo); |
| |
| if (entityHashValue == -1) { |
| //decrease datasetLockCount |
| lockMode = entityInfoManager.getDatasetLockMode(entityInfo); |
| datasetLockCount = entityInfoManager.getDatasetLockCount(entityInfo); |
| if (datasetLockCount != 0) { |
| dLockInfo.decreaseLockCount(lockMode, datasetLockCount); |
| |
| //wakeup waiters of datasetLock and remove holder from datasetLockInfo |
| wakeUpDatasetLockWaiters(dLockInfo); |
| |
| //remove the holder from datasetLockInfo only if the lock is dataset-granule lock. |
| //--> this also removes the holding resource from jobInfo |
| //(Because the IX and IS lock's holders are handled implicitly, |
| //those are not in the holder list of datasetLockInfo.) |
| dLockInfo.removeHolder(entityInfo, jobInfo); |
| } |
| } else { |
| //decrease datasetLockCount |
| lockMode = entityInfoManager.getDatasetLockMode(entityInfo); |
| lockMode = lockMode == LockMode.S ? LockMode.IS : LockMode.IX; |
| datasetLockCount = entityInfoManager.getDatasetLockCount(entityInfo); |
| |
| if (datasetLockCount != 0) { |
| dLockInfo.decreaseLockCount(lockMode, datasetLockCount); |
| } |
| |
| //decrease entityLockCount |
| lockMode = entityInfoManager.getEntityLockMode(entityInfo); |
| entityLockCount = entityInfoManager.getEntityLockCount(entityInfo); |
| eLockInfo = dLockInfo.getEntityResourceHT().get(entityHashValue); |
| if (IS_DEBUG_MODE) { |
| if (eLockInfo < 0) { |
| System.out.println("eLockInfo:" + eLockInfo); |
| } |
| } |
| |
| if (entityLockCount != 0) { |
| entityLockInfoManager.decreaseLockCount(eLockInfo, lockMode, (short) entityLockCount); |
| } |
| |
| if (datasetLockCount != 0) { |
| //wakeup waiters of datasetLock and don't remove holder from datasetLockInfo |
| wakeUpDatasetLockWaiters(dLockInfo); |
| } |
| |
| if (entityLockCount != 0) { |
| //wakeup waiters of entityLock |
| wakeUpEntityLockWaiters(eLockInfo); |
| |
| //remove the holder from entityLockInfo |
| //--> this also removes the holding resource from jobInfo |
| entityLockInfoManager.removeHolder(eLockInfo, entityInfo, jobInfo); |
| } |
| |
| //deallocate entityLockInfo if there is no holder and waiter. |
| if (entityLockInfoManager.getLastHolder(eLockInfo) == -1 |
| && entityLockInfoManager.getFirstWaiter(eLockInfo) == -1 |
| && entityLockInfoManager.getUpgrader(eLockInfo) == -1) { |
| dLockInfo.getEntityResourceHT().remove(entityHashValue); |
| entityLockInfoManager.deallocate(eLockInfo); |
| // if (IS_DEBUG_MODE) { |
| // System.out.println("removed PK["+entityHashValue+"]"); |
| // } |
| } |
| } |
| |
| //deallocate entityInfo |
| entityInfoManager.deallocate(entityInfo); |
| // if (IS_DEBUG_MODE) { |
| // System.out.println("dellocate EntityInfo["+entityInfo+"]"); |
| // } |
| |
| entityInfo = prevEntityInfo; |
| } |
| |
| //remove JobInfo |
| jobHT.remove(jobId); |
| |
| if (existWaiter) { |
| txnContext.setStatus(TransactionContext.TIMED_OUT_STATUS); |
| txnContext.setTxnState(TransactionState.ABORTED); |
| } |
| |
| if (IS_DEBUG_MODE) { |
| trackLockRequest("Granted", RequestType.RELEASE_LOCKS, new DatasetId(0), 0, (byte) 0, txnContext, |
| dLockInfo, eLockInfo); |
| } |
| unlatchLockTable(); |
| } |
| |
| @Override |
| public void instantLock(DatasetId datasetId, int entityHashValue, byte lockMode, TransactionContext txnContext) |
| throws ACIDException { |
| |
| // try { |
| // internalLock(datasetId, entityHashValue, lockMode, txnContext); |
| // return; |
| // } finally { |
| // unlock(datasetId, entityHashValue, txnContext); |
| // } |
| internalLock(datasetId, entityHashValue, lockMode, txnContext, true); |
| instantUnlock(datasetId, entityHashValue, txnContext); |
| } |
| |
| @Override |
| public boolean tryLock(DatasetId datasetId, int entityHashValue, byte lockMode, TransactionContext txnContext) |
| throws ACIDException { |
| return internalTryLock(datasetId, entityHashValue, lockMode, txnContext, false); |
| } |
| |
| @Override |
| public boolean instantTryLock(DatasetId datasetId, int entityHashValue, byte lockMode, TransactionContext txnContext) |
| throws ACIDException { |
| boolean isGranted = false; |
| // try { |
| // isGranted = internalTryLock(datasetId, entityHashValue, lockMode, txnContext); |
| // return isGranted; |
| // } finally { |
| // if (isGranted) { |
| // unlock(datasetId, entityHashValue, txnContext); |
| // } |
| // } |
| isGranted = internalTryLock(datasetId, entityHashValue, lockMode, txnContext, true); |
| if (isGranted) { |
| instantUnlock(datasetId, entityHashValue, txnContext); |
| } |
| return isGranted; |
| } |
| |
| private boolean internalTryLock(DatasetId datasetId, int entityHashValue, byte lockMode, |
| TransactionContext txnContext, boolean isInstant) throws ACIDException { |
| JobId jobId = txnContext.getJobId(); |
| int jId = jobId.getId(); //int-type jobId |
| int dId = datasetId.getId(); //int-type datasetId |
| int entityInfo; |
| int eLockInfo = -1; |
| DatasetLockInfo dLockInfo = null; |
| JobInfo jobInfo; |
| byte datasetLockMode = entityHashValue == -1 ? lockMode : lockMode == LockMode.S ? LockMode.IS : LockMode.IX; |
| boolean isSuccess = false; |
| boolean doEscalate = false; |
| |
| latchLockTable(); |
| validateJob(txnContext); |
| |
| if (IS_DEBUG_MODE) { |
| trackLockRequest("Requested", RequestType.TRY_LOCK, datasetId, entityHashValue, lockMode, txnContext, |
| dLockInfo, eLockInfo); |
| } |
| |
| dLockInfo = datasetResourceHT.get(datasetId); |
| jobInfo = jobHT.get(jobId); |
| |
| if (ALLOW_ESCALATE_FROM_ENTITY_TO_DATASET) { |
| if (!isInstant && datasetLockMode == LockMode.IS && jobInfo != null && dLockInfo != null) { |
| int upgradeStatus = needEscalateFromEntityToDataset(jobInfo, dId, lockMode); |
| switch (upgradeStatus) { |
| case DO_ESCALATE: |
| entityHashValue = -1; |
| doEscalate = true; |
| break; |
| |
| case ESCALATED: |
| unlatchLockTable(); |
| return true; |
| |
| default: |
| break; |
| } |
| } |
| } |
| |
| //#. if the datasetLockInfo doesn't exist in datasetResourceHT |
| if (dLockInfo == null || dLockInfo.isNoHolder()) { |
| if (dLockInfo == null) { |
| dLockInfo = new DatasetLockInfo(entityLockInfoManager, entityInfoManager, lockWaiterManager); |
| datasetResourceHT.put(new DatasetId(dId), dLockInfo); //datsetId obj should be created |
| } |
| entityInfo = entityInfoManager.allocate(jId, dId, entityHashValue, lockMode); |
| |
| //if dataset-granule lock |
| if (entityHashValue == -1) { //-1 stands for dataset-granule |
| entityInfoManager.increaseDatasetLockCount(entityInfo); |
| dLockInfo.increaseLockCount(datasetLockMode); |
| dLockInfo.addHolder(entityInfo); |
| } else { |
| entityInfoManager.increaseDatasetLockCount(entityInfo); |
| dLockInfo.increaseLockCount(datasetLockMode); |
| //add entityLockInfo |
| eLockInfo = entityLockInfoManager.allocate(); |
| dLockInfo.getEntityResourceHT().put(entityHashValue, eLockInfo); |
| entityInfoManager.increaseEntityLockCount(entityInfo); |
| entityLockInfoManager.increaseLockCount(eLockInfo, lockMode); |
| entityLockInfoManager.addHolder(eLockInfo, entityInfo); |
| } |
| |
| if (jobInfo == null) { |
| jobInfo = new JobInfo(entityInfoManager, lockWaiterManager, txnContext); |
| jobHT.put(jobId, jobInfo); //jobId obj doesn't have to be created |
| } |
| jobInfo.addHoldingResource(entityInfo); |
| |
| if (ALLOW_ESCALATE_FROM_ENTITY_TO_DATASET) { |
| if (!isInstant && datasetLockMode == LockMode.IS) { |
| jobInfo.increaseDatasetISLockCount(dId); |
| if (doEscalate) { |
| //This exception is thrown when the threshold value is set to 1. |
| //We don't want to allow the lock escalation when there is a first lock request on a dataset. |
| throw new IllegalStateException("ESCALATE_TRHESHOLD_ENTITY_TO_DATASET should not be set to " |
| + ESCALATE_TRHESHOLD_ENTITY_TO_DATASET); |
| } |
| } |
| } |
| |
| if (IS_DEBUG_MODE) { |
| trackLockRequest("Granted", RequestType.TRY_LOCK, datasetId, entityHashValue, lockMode, txnContext, |
| dLockInfo, eLockInfo); |
| } |
| |
| unlatchLockTable(); |
| return true; |
| } |
| |
| //#. the datasetLockInfo exists in datasetResourceHT. |
| //1. handle dataset-granule lock |
| tryLockDatasetGranuleRevertOperation = 0; |
| entityInfo = tryLockDatasetGranule(datasetId, entityHashValue, lockMode, txnContext); |
| if (entityInfo == -2) {//-2 represents fail |
| isSuccess = false; |
| } else { |
| //2. handle entity-granule lock |
| if (entityHashValue != -1) { |
| isSuccess = tryLockEntityGranule(datasetId, entityHashValue, lockMode, entityInfo, txnContext); |
| if (!isSuccess) { |
| revertTryLockDatasetGranuleOperation(datasetId, entityHashValue, lockMode, entityInfo, txnContext); |
| } |
| } |
| } |
| |
| if (ALLOW_ESCALATE_FROM_ENTITY_TO_DATASET) { |
| if (!isInstant) { |
| if (doEscalate) { |
| //jobInfo must not be null. |
| assert jobInfo != null; |
| jobInfo.increaseDatasetISLockCount(dId); |
| //release pre-acquired locks |
| releaseDatasetISLocks(jobInfo, jobId, datasetId, txnContext); |
| } else if (datasetLockMode == LockMode.IS) { |
| if (jobInfo == null) { |
| jobInfo = jobHT.get(jobId); |
| //jobInfo must not be null; |
| assert jobInfo != null; |
| } |
| jobInfo.increaseDatasetISLockCount(dId); |
| } |
| } |
| } |
| |
| if (IS_DEBUG_MODE) { |
| if (isSuccess) { |
| trackLockRequest("Granted", RequestType.TRY_LOCK, datasetId, entityHashValue, lockMode, txnContext, |
| dLockInfo, eLockInfo); |
| } else { |
| trackLockRequest("Failed", RequestType.TRY_LOCK, datasetId, entityHashValue, lockMode, txnContext, |
| dLockInfo, eLockInfo); |
| } |
| } |
| |
| unlatchLockTable(); |
| |
| return isSuccess; |
| } |
| |
| private void trackLockRequest(String msg, int requestType, DatasetId datasetIdObj, int entityHashValue, |
| byte lockMode, TransactionContext txnContext, DatasetLockInfo dLockInfo, int eLockInfo) { |
| StringBuilder s = new StringBuilder(); |
| LockRequest request = new LockRequest(Thread.currentThread().getName(), requestType, datasetIdObj, |
| entityHashValue, lockMode, txnContext); |
| s.append(Thread.currentThread().getId() + ":"); |
| s.append(msg); |
| if (msg.equals("Granted")) { |
| if (dLockInfo != null) { |
| s.append("\t|D| "); |
| s.append(dLockInfo.getIXCount()).append(","); |
| s.append(dLockInfo.getISCount()).append(","); |
| s.append(dLockInfo.getXCount()).append(","); |
| s.append(dLockInfo.getSCount()).append(","); |
| if (dLockInfo.getFirstUpgrader() != -1) { |
| s.append("+"); |
| } else { |
| s.append("-"); |
| } |
| s.append(","); |
| if (dLockInfo.getFirstWaiter() != -1) { |
| s.append("+"); |
| } else { |
| s.append("-"); |
| } |
| } |
| |
| if (eLockInfo != -1) { |
| s.append("\t|E| "); |
| s.append(entityLockInfoManager.getXCount(eLockInfo)).append(","); |
| s.append(entityLockInfoManager.getSCount(eLockInfo)).append(","); |
| if (entityLockInfoManager.getUpgrader(eLockInfo) != -1) { |
| s.append("+"); |
| } else { |
| s.append("-"); |
| } |
| s.append(","); |
| if (entityLockInfoManager.getFirstWaiter(eLockInfo) != -1) { |
| s.append("+"); |
| } else { |
| s.append("-"); |
| } |
| } |
| } |
| |
| lockRequestTracker.addEvent(s.toString(), request); |
| if (msg.equals("Requested")) { |
| lockRequestTracker.addRequest(request); |
| } |
| System.out.println(request.prettyPrint() + "--> " + s.toString()); |
| } |
| |
| public String getHistoryForAllJobs() { |
| if (IS_DEBUG_MODE) { |
| return lockRequestTracker.getHistoryForAllJobs(); |
| } |
| return null; |
| } |
| |
| public String getHistoryPerJob() { |
| if (IS_DEBUG_MODE) { |
| return lockRequestTracker.getHistoryPerJob(); |
| } |
| return null; |
| } |
| |
| public String getRequestHistoryForAllJobs() { |
| if (IS_DEBUG_MODE) { |
| return lockRequestTracker.getRequestHistoryForAllJobs(); |
| } |
| return null; |
| } |
| |
| private void revertTryLockDatasetGranuleOperation(DatasetId datasetId, int entityHashValue, byte lockMode, |
| int entityInfo, TransactionContext txnContext) { |
| JobId jobId = txnContext.getJobId(); |
| DatasetLockInfo dLockInfo; |
| JobInfo jobInfo; |
| int lockCount; |
| byte datasetLockMode = entityHashValue == -1 ? lockMode : lockMode == LockMode.S ? LockMode.IS : LockMode.IX; |
| |
| dLockInfo = datasetResourceHT.get(datasetId); |
| jobInfo = jobHT.get(jobId); |
| |
| //see tryLockDatasetGranule() function to know the revert operation |
| switch (tryLockDatasetGranuleRevertOperation) { |
| |
| case 1://[revertOperation1]: reverting 'adding a holder' |
| |
| if (entityHashValue == -1) { |
| dLockInfo.decreaseLockCount(datasetLockMode); |
| dLockInfo.removeHolder(entityInfo, jobInfo); //--> this call removes entityInfo from JobInfo's holding-resource-list as well. |
| } else { |
| dLockInfo.decreaseLockCount(datasetLockMode); |
| jobInfo.removeHoldingResource(entityInfo); |
| } |
| entityInfoManager.decreaseDatasetLockCount(entityInfo); |
| if (jobInfo.getLastHoldingResource() == -1 && jobInfo.getFirstWaitingResource() == -1) { |
| jobHT.remove(jobId); |
| } |
| entityInfoManager.deallocate(entityInfo); |
| break; |
| |
| case 2://[revertOperation2]: reverting 'adding an upgrader' |
| lockCount = entityInfoManager.getDatasetLockCount(entityInfo); |
| if (entityHashValue == -1) { //dataset-granule lock |
| dLockInfo.decreaseLockCount(LockMode.X, lockCount); |
| dLockInfo.increaseLockCount(LockMode.S, lockCount); |
| } else { |
| dLockInfo.decreaseLockCount(LockMode.IX, lockCount); |
| dLockInfo.increaseLockCount(LockMode.IS, lockCount); |
| } |
| entityInfoManager.setDatasetLockMode(entityInfo, LockMode.S); |
| break; |
| |
| case 3://[revertOperation3]: reverting 'adding a duplicated call' |
| entityInfoManager.decreaseDatasetLockCount(entityInfo); |
| datasetLockMode = entityInfoManager.getDatasetLockMode(entityInfo); |
| if (entityHashValue == -1) { //dataset-granule |
| dLockInfo.decreaseLockCount(datasetLockMode); |
| } else { //entity-granule |
| datasetLockMode = datasetLockMode == LockMode.S ? LockMode.IS : LockMode.IX; |
| dLockInfo.decreaseLockCount(datasetLockMode); |
| } |
| |
| break; |
| default: |
| //do nothing; |
| } |
| } |
| |
| private int tryLockDatasetGranule(DatasetId datasetId, int entityHashValue, byte lockMode, |
| TransactionContext txnContext) throws ACIDException { |
| JobId jobId = txnContext.getJobId(); |
| int jId = jobId.getId(); //int-type jobId |
| int dId = datasetId.getId(); //int-type datasetId |
| int waiterObjId; |
| int entityInfo = -1; |
| DatasetLockInfo dLockInfo; |
| JobInfo jobInfo; |
| boolean isUpgrade = false; |
| int weakerModeLockCount; |
| byte datasetLockMode = entityHashValue == -1 ? lockMode : lockMode == LockMode.S ? LockMode.IS : LockMode.IX; |
| |
| dLockInfo = datasetResourceHT.get(datasetId); |
| jobInfo = jobHT.get(jobId); |
| |
| //check duplicated call |
| |
| //1. lock request causing duplicated upgrading requests from different threads in a same job |
| waiterObjId = dLockInfo.findUpgraderFromUpgraderList(jId, entityHashValue); |
| if (waiterObjId != -1) { |
| return -2; |
| } |
| |
| //2. lock request causing duplicated waiting requests from different threads in a same job |
| waiterObjId = dLockInfo.findWaiterFromWaiterList(jId, entityHashValue); |
| if (waiterObjId != -1) { |
| return -2; |
| } |
| |
| //3. lock request causing duplicated holding requests from different threads or a single thread in a same job |
| entityInfo = dLockInfo.findEntityInfoFromHolderList(jId, entityHashValue); |
| if (entityInfo == -1) { //new call from this job -> doesn't mean that eLockInfo doesn't exist since another thread might have create the eLockInfo already. |
| |
| ////////////////////////////////////////////////////////////////////////////////////// |
| //[part of revertOperation1] |
| entityInfo = entityInfoManager.allocate(jId, dId, entityHashValue, lockMode); |
| if (jobInfo == null) { |
| jobInfo = new JobInfo(entityInfoManager, lockWaiterManager, txnContext); |
| jobHT.put(jobId, jobInfo); |
| } |
| ////////////////////////////////////////////////////////////////////////////////////// |
| |
| //return fail if any upgrader exists or upgrading lock mode is not compatible |
| if (dLockInfo.getFirstUpgrader() != -1 || dLockInfo.getFirstWaiter() != -1 |
| || !dLockInfo.isCompatible(datasetLockMode)) { |
| |
| //[Notice] |
| //There has been no same caller as (jId, dId, entityHashValue) triplet. |
| //But there could be the same caller as (jId, dId) pair. |
| //For example, two requests (J1, D1, E1) and (J1, D1, E2) are considered as duplicated call in dataset-granule perspective. |
| //Therefore, the above duplicated call case is covered in the following code. |
| //find the same dataset-granule lock request, that is, (J1, D1) pair in the above example. |
| if (jobInfo.isDatasetLockGranted(dId, LockMode.IS)) { |
| if (dLockInfo.isCompatible(datasetLockMode)) { |
| //this is duplicated call |
| entityInfoManager.increaseDatasetLockCount(entityInfo); |
| if (entityHashValue == -1) { |
| dLockInfo.increaseLockCount(datasetLockMode); |
| dLockInfo.addHolder(entityInfo); |
| } else { |
| dLockInfo.increaseLockCount(datasetLockMode); |
| //IS and IX holders are implicitly handled. |
| } |
| //add entityInfo to JobInfo's holding-resource list |
| jobInfo.addHoldingResource(entityInfo); |
| |
| tryLockDatasetGranuleRevertOperation = 1; |
| |
| return entityInfo; |
| } |
| } |
| |
| //revert [part of revertOperation1] before return |
| if (jobInfo.getLastHoldingResource() == -1 && jobInfo.getFirstWaitingResource() == -1) { |
| jobHT.remove(jobId); |
| } |
| entityInfoManager.deallocate(entityInfo); |
| |
| return -2; |
| } |
| |
| ////////////////////////////////////////////////////////////////////////////////////// |
| //revert the following operations if the caller thread has to wait during this call. |
| //[revertOperation1] |
| entityInfoManager.increaseDatasetLockCount(entityInfo); |
| if (entityHashValue == -1) { |
| dLockInfo.increaseLockCount(datasetLockMode); |
| dLockInfo.addHolder(entityInfo); |
| } else { |
| dLockInfo.increaseLockCount(datasetLockMode); |
| //IS and IX holders are implicitly handled. |
| } |
| //add entityInfo to JobInfo's holding-resource list |
| jobInfo.addHoldingResource(entityInfo); |
| |
| //set revert operation to be reverted when tryLock() fails |
| tryLockDatasetGranuleRevertOperation = 1; |
| ////////////////////////////////////////////////////////////////////////////////////// |
| |
| } else { |
| isUpgrade = isLockUpgrade(entityInfoManager.getDatasetLockMode(entityInfo), lockMode); |
| if (isUpgrade) { //upgrade call |
| //return fail if any upgrader exists or upgrading lock mode is not compatible |
| if (dLockInfo.getFirstUpgrader() != -1 || !dLockInfo.isUpgradeCompatible(datasetLockMode, entityInfo)) { |
| return -2; |
| } |
| |
| //update entityInfo's dataset lock count and datasetLockInfo's lock count |
| weakerModeLockCount = entityInfoManager.getDatasetLockCount(entityInfo); |
| |
| ////////////////////////////////////////////////////////////////////////////////////// |
| //revert the following operations if the caller thread has to wait during this call. |
| //[revertOperation2] |
| entityInfoManager.setDatasetLockMode(entityInfo, lockMode); |
| |
| if (entityHashValue == -1) { //dataset-granule lock |
| dLockInfo.increaseLockCount(LockMode.X, weakerModeLockCount);//new lock mode |
| dLockInfo.decreaseLockCount(LockMode.S, weakerModeLockCount);//current lock mode |
| } else { |
| dLockInfo.increaseLockCount(LockMode.IX, weakerModeLockCount); |
| dLockInfo.decreaseLockCount(LockMode.IS, weakerModeLockCount); |
| } |
| tryLockDatasetGranuleRevertOperation = 2; |
| ////////////////////////////////////////////////////////////////////////////////////// |
| |
| } else { //duplicated call |
| |
| ////////////////////////////////////////////////////////////////////////////////////// |
| //revert the following operations if the caller thread has to wait during this call. |
| //[revertOperation3] |
| entityInfoManager.increaseDatasetLockCount(entityInfo); |
| datasetLockMode = entityInfoManager.getDatasetLockMode(entityInfo); |
| |
| if (entityHashValue == -1) { //dataset-granule |
| dLockInfo.increaseLockCount(datasetLockMode); |
| } else { //entity-granule |
| datasetLockMode = datasetLockMode == LockMode.S ? LockMode.IS : LockMode.IX; |
| dLockInfo.increaseLockCount(datasetLockMode); |
| } |
| |
| tryLockDatasetGranuleRevertOperation = 3; |
| ////////////////////////////////////////////////////////////////////////////////////// |
| |
| } |
| } |
| |
| return entityInfo; |
| } |
| |
| private boolean tryLockEntityGranule(DatasetId datasetId, int entityHashValue, byte lockMode, |
| int entityInfoFromDLockInfo, TransactionContext txnContext) throws ACIDException { |
| JobId jobId = txnContext.getJobId(); |
| int jId = jobId.getId(); //int-type jobId |
| int waiterObjId; |
| int eLockInfo = -1; |
| int entityInfo; |
| DatasetLockInfo dLockInfo; |
| boolean isUpgrade = false; |
| int weakerModeLockCount; |
| |
| dLockInfo = datasetResourceHT.get(datasetId); |
| eLockInfo = dLockInfo.getEntityResourceHT().get(entityHashValue); |
| |
| if (eLockInfo != -1) { |
| //check duplicated call |
| |
| //1. lock request causing duplicated upgrading requests from different threads in a same job |
| waiterObjId = entityLockInfoManager.findUpgraderFromUpgraderList(eLockInfo, jId, entityHashValue); |
| if (waiterObjId != -1) { |
| return false; |
| } |
| |
| //2. lock request causing duplicated waiting requests from different threads in a same job |
| waiterObjId = entityLockInfoManager.findWaiterFromWaiterList(eLockInfo, jId, entityHashValue); |
| if (waiterObjId != -1) { |
| return false; |
| } |
| |
| //3. lock request causing duplicated holding requests from different threads or a single thread in a same job |
| entityInfo = entityLockInfoManager.findEntityInfoFromHolderList(eLockInfo, jId, entityHashValue); |
| if (entityInfo != -1) {//duplicated call or upgrader |
| |
| isUpgrade = isLockUpgrade(entityInfoManager.getEntityLockMode(entityInfo), lockMode); |
| if (isUpgrade) {//upgrade call |
| //wait if any upgrader exists or upgrading lock mode is not compatible |
| if (entityLockInfoManager.getUpgrader(eLockInfo) != -1 |
| || !entityLockInfoManager.isUpgradeCompatible(eLockInfo, lockMode, entityInfo)) { |
| return false; |
| } |
| |
| weakerModeLockCount = entityInfoManager.getEntityLockCount(entityInfo); |
| entityInfoManager.setEntityLockMode(entityInfo, lockMode); |
| |
| entityLockInfoManager.increaseLockCount(eLockInfo, LockMode.X, (short) weakerModeLockCount);//new lock mode |
| entityLockInfoManager.decreaseLockCount(eLockInfo, LockMode.S, (short) weakerModeLockCount);//old lock mode |
| |
| } else {//duplicated call |
| entityInfoManager.increaseEntityLockCount(entityInfo); |
| entityLockInfoManager.increaseLockCount(eLockInfo, entityInfoManager.getEntityLockMode(entityInfo)); |
| } |
| } else {//new call from this job, but still eLockInfo exists since other threads hold it or wait on it |
| entityInfo = entityInfoFromDLockInfo; |
| if (entityLockInfoManager.getUpgrader(eLockInfo) != -1 |
| || entityLockInfoManager.getFirstWaiter(eLockInfo) != -1 |
| || !entityLockInfoManager.isCompatible(eLockInfo, lockMode)) { |
| return false; |
| } |
| |
| entityInfoManager.increaseEntityLockCount(entityInfo); |
| entityLockInfoManager.increaseLockCount(eLockInfo, lockMode); |
| entityLockInfoManager.addHolder(eLockInfo, entityInfo); |
| } |
| } else {//eLockInfo doesn't exist, so this lock request is the first request and can be granted without waiting. |
| eLockInfo = entityLockInfoManager.allocate(); |
| dLockInfo.getEntityResourceHT().put(entityHashValue, eLockInfo); |
| entityInfoManager.increaseEntityLockCount(entityInfoFromDLockInfo); |
| entityLockInfoManager.increaseLockCount(eLockInfo, lockMode); |
| entityLockInfoManager.addHolder(eLockInfo, entityInfoFromDLockInfo); |
| } |
| |
| return true; |
| } |
| |
| private void latchLockTable() { |
| lockTableLatch.writeLock().lock(); |
| } |
| |
| private void unlatchLockTable() { |
| lockTableLatch.writeLock().unlock(); |
| } |
| |
| private void latchWaitNotify() { |
| waiterLatch.writeLock().lock(); |
| } |
| |
| private void unlatchWaitNotify() { |
| waiterLatch.writeLock().unlock(); |
| } |
| |
| private int handleLockWaiter(DatasetLockInfo dLockInfo, int eLockInfo, int entityInfo, boolean isUpgrade, |
| boolean isDatasetLockInfo, TransactionContext txnContext, JobInfo jobInfo, int duplicatedWaiterObjId) |
| throws ACIDException { |
| int waiterId = -1; |
| LockWaiter waiter; |
| int waiterCount = 0; |
| boolean isInterruptedExceptionOccurred = false; |
| |
| if (duplicatedWaiterObjId != -1 |
| || isDeadlockFree(dLockInfo, eLockInfo, entityInfo, isDatasetLockInfo, isUpgrade)) {//deadlock free -> wait |
| if (duplicatedWaiterObjId == -1) { |
| waiterId = lockWaiterManager.allocate(); //initial value of waiterObj: wait = true, victim = false |
| waiter = lockWaiterManager.getLockWaiter(waiterId); |
| waiter.setEntityInfoSlot(entityInfo); |
| jobInfo.addWaitingResource(waiterId); |
| waiter.setBeginWaitTime(System.currentTimeMillis()); |
| } else { |
| waiterId = duplicatedWaiterObjId; |
| waiter = lockWaiterManager.getLockWaiter(waiterId); |
| } |
| |
| if (duplicatedWaiterObjId == -1) { |
| //add actor properly |
| if (isDatasetLockInfo) { |
| waiter.setWaitingOnEntityLock(false); |
| if (isUpgrade) { |
| dLockInfo.addUpgrader(waiterId); |
| waiter.setWaiter(false); |
| } else { |
| dLockInfo.addWaiter(waiterId); |
| waiter.setWaiter(true); |
| } |
| } else { |
| waiter.setWaitingOnEntityLock(true); |
| if (isUpgrade) { |
| waiter.setWaiter(false); |
| entityLockInfoManager.addUpgrader(eLockInfo, waiterId); |
| } else { |
| waiter.setWaiter(true); |
| entityLockInfoManager.addWaiter(eLockInfo, waiterId); |
| } |
| } |
| } |
| waiter.increaseWaiterCount(); |
| waiter.setFirstGetUp(true); |
| |
| latchWaitNotify(); |
| unlatchLockTable(); |
| synchronized (waiter) { |
| unlatchWaitNotify(); |
| while (waiter.needWait()) { |
| try { |
| if (IS_DEBUG_MODE) { |
| System.out.println("" + Thread.currentThread().getName() + "\twaits(" |
| + waiter.getWaiterCount() + "): WID(" + waiterId + "),EID(" |
| + waiter.getEntityInfoSlot() + ")"); |
| } |
| waiter.wait(); |
| } catch (InterruptedException e) { |
| //TODO figure-out what is the appropriate way to handle this exception |
| e.printStackTrace(); |
| isInterruptedExceptionOccurred = true; |
| waiter.setWait(false); |
| } |
| } |
| } |
| |
| if (isInterruptedExceptionOccurred) { |
| throw new ACIDException("InterruptedException is caught"); |
| } |
| |
| //waiter woke up -> remove/deallocate waiter object and abort if timeout |
| latchLockTable(); |
| |
| if (txnContext.getStatus() == TransactionContext.TIMED_OUT_STATUS || waiter.isVictim()) { |
| try { |
| requestAbort(txnContext); |
| } finally { |
| unlatchLockTable(); |
| } |
| } |
| |
| if (waiter.isFirstGetUp()) { |
| waiter.setFirstGetUp(false); |
| waiterCount = waiter.getWaiterCount(); |
| } else { |
| waiterCount = 0; |
| } |
| |
| waiter.decreaseWaiterCount(); |
| if (IS_DEBUG_MODE) { |
| System.out.println("" + Thread.currentThread().getName() + "\tgot-up!(" + waiter.getWaiterCount() |
| + "): WID(" + waiterId + "),EID(" + waiter.getEntityInfoSlot() + ")"); |
| } |
| if (waiter.getWaiterCount() == 0) { |
| //remove actor properly |
| if (isDatasetLockInfo) { |
| if (isUpgrade) { |
| dLockInfo.removeUpgrader(waiterId); |
| } else { |
| dLockInfo.removeWaiter(waiterId); |
| } |
| } else { |
| if (isUpgrade) { |
| entityLockInfoManager.removeUpgrader(eLockInfo, waiterId); |
| } else { |
| entityLockInfoManager.removeWaiter(eLockInfo, waiterId); |
| } |
| } |
| |
| //if (!isUpgrade && isDatasetLockInfo) { |
| jobInfo.removeWaitingResource(waiterId); |
| //} |
| lockWaiterManager.deallocate(waiterId); |
| } |
| |
| } else { //deadlock -> abort |
| //[Notice] |
| //Before requesting abort, the entityInfo for waiting datasetLock request is deallocated. |
| if (!isUpgrade && isDatasetLockInfo) { |
| //deallocate the entityInfo |
| entityInfoManager.deallocate(entityInfo); |
| } |
| try { |
| requestAbort(txnContext); |
| } finally { |
| unlatchLockTable(); |
| } |
| } |
| |
| return waiterCount; |
| } |
| |
| private boolean isDeadlockFree(DatasetLockInfo dLockInfo, int eLockInfo, int entityInfo, boolean isDatasetLockInfo, |
| boolean isUpgrade) { |
| return deadlockDetector.isSafeToAdd(dLockInfo, eLockInfo, entityInfo, isDatasetLockInfo, isUpgrade); |
| } |
| |
| private void requestAbort(TransactionContext txnContext) throws ACIDException { |
| txnContext.setStatus(TransactionContext.TIMED_OUT_STATUS); |
| txnContext.setStartWaitTime(TransactionContext.INVALID_TIME); |
| throw new ACIDException("Transaction " + txnContext.getJobId() |
| + " should abort (requested by the Lock Manager)"); |
| } |
| |
| /** |
| * For now, upgrading lock granule from entity-granule to dataset-granule is not supported!! |
| * |
| * @param fromLockMode |
| * @param toLockMode |
| * @return |
| */ |
| private boolean isLockUpgrade(byte fromLockMode, byte toLockMode) { |
| return fromLockMode == LockMode.S && toLockMode == LockMode.X; |
| } |
| |
| /** |
| * wake up upgraders first, then waiters. |
| * Criteria to wake up upgraders: if the upgrading lock mode is compatible, then wake up the upgrader. |
| */ |
| private void wakeUpDatasetLockWaiters(DatasetLockInfo dLockInfo) { |
| int waiterObjId = dLockInfo.getFirstUpgrader(); |
| int entityInfo; |
| LockWaiter waiterObj; |
| byte datasetLockMode; |
| byte lockMode; |
| boolean areAllUpgradersAwaken = true; |
| |
| consecutiveWakeupContext.reset(); |
| while (waiterObjId != -1) { |
| //wake up upgraders |
| waiterObj = lockWaiterManager.getLockWaiter(waiterObjId); |
| entityInfo = waiterObj.getEntityInfoSlot(); |
| datasetLockMode = entityInfoManager.getPKHashVal(entityInfo) == -1 ? LockMode.X : LockMode.IX; |
| if (dLockInfo.isUpgradeCompatible(datasetLockMode, entityInfo) |
| && consecutiveWakeupContext.isCompatible(datasetLockMode)) { |
| consecutiveWakeupContext.setLockMode(datasetLockMode); |
| //compatible upgrader is waken up |
| latchWaitNotify(); |
| synchronized (waiterObj) { |
| unlatchWaitNotify(); |
| waiterObj.setWait(false); |
| if (IS_DEBUG_MODE) { |
| System.out.println("" + Thread.currentThread().getName() + "\twake-up(D): WID(" + waiterObjId |
| + "),EID(" + waiterObj.getEntityInfoSlot() + ")"); |
| } |
| waiterObj.notifyAll(); |
| } |
| waiterObjId = waiterObj.getNextWaiterObjId(); |
| } else { |
| areAllUpgradersAwaken = false; |
| break; |
| } |
| } |
| |
| if (areAllUpgradersAwaken) { |
| //wake up waiters |
| waiterObjId = dLockInfo.getFirstWaiter(); |
| while (waiterObjId != -1) { |
| waiterObj = lockWaiterManager.getLockWaiter(waiterObjId); |
| entityInfo = waiterObj.getEntityInfoSlot(); |
| lockMode = entityInfoManager.getDatasetLockMode(entityInfo); |
| datasetLockMode = entityInfoManager.getPKHashVal(entityInfo) == -1 ? lockMode |
| : lockMode == LockMode.S ? LockMode.IS : LockMode.IX; |
| if (dLockInfo.isCompatible(datasetLockMode) && consecutiveWakeupContext.isCompatible(datasetLockMode)) { |
| consecutiveWakeupContext.setLockMode(datasetLockMode); |
| //compatible waiter is waken up |
| latchWaitNotify(); |
| synchronized (waiterObj) { |
| unlatchWaitNotify(); |
| waiterObj.setWait(false); |
| if (IS_DEBUG_MODE) { |
| System.out.println("" + Thread.currentThread().getName() + "\twake-up(D): WID(" |
| + waiterObjId + "),EID(" + waiterObj.getEntityInfoSlot() + ")"); |
| } |
| waiterObj.notifyAll(); |
| } |
| waiterObjId = waiterObj.getNextWaiterObjId(); |
| } else { |
| break; |
| } |
| } |
| } |
| } |
| |
| private void wakeUpEntityLockWaiters(int eLockInfo) { |
| boolean areAllUpgradersAwaken = true; |
| int waiterObjId = entityLockInfoManager.getUpgrader(eLockInfo); |
| int entityInfo; |
| LockWaiter waiterObj; |
| byte entityLockMode; |
| |
| consecutiveWakeupContext.reset(); |
| while (waiterObjId != -1) { |
| //wake up upgraders |
| waiterObj = lockWaiterManager.getLockWaiter(waiterObjId); |
| entityInfo = waiterObj.getEntityInfoSlot(); |
| if (entityLockInfoManager.isUpgradeCompatible(eLockInfo, LockMode.X, entityInfo) |
| && consecutiveWakeupContext.isCompatible(LockMode.X)) { |
| consecutiveWakeupContext.setLockMode(LockMode.X); |
| latchWaitNotify(); |
| synchronized (waiterObj) { |
| unlatchWaitNotify(); |
| waiterObj.setWait(false); |
| if (IS_DEBUG_MODE) { |
| System.out.println("" + Thread.currentThread().getName() + "\twake-up(E): WID(" + waiterObjId |
| + "),EID(" + waiterObj.getEntityInfoSlot() + ")"); |
| } |
| waiterObj.notifyAll(); |
| } |
| waiterObjId = waiterObj.getNextWaiterObjId(); |
| } else { |
| areAllUpgradersAwaken = false; |
| break; |
| } |
| } |
| |
| if (areAllUpgradersAwaken) { |
| //wake up waiters |
| waiterObjId = entityLockInfoManager.getFirstWaiter(eLockInfo); |
| while (waiterObjId != -1) { |
| waiterObj = lockWaiterManager.getLockWaiter(waiterObjId); |
| entityInfo = waiterObj.getEntityInfoSlot(); |
| entityLockMode = entityInfoManager.getEntityLockMode(entityInfo); |
| if (entityLockInfoManager.isCompatible(eLockInfo, entityLockMode) |
| && consecutiveWakeupContext.isCompatible(entityLockMode)) { |
| consecutiveWakeupContext.setLockMode(entityLockMode); |
| //compatible waiter is waken up |
| latchWaitNotify(); |
| synchronized (waiterObj) { |
| unlatchWaitNotify(); |
| waiterObj.setWait(false); |
| if (IS_DEBUG_MODE) { |
| System.out.println("" + Thread.currentThread().getName() + "\twake-up(E): WID(" |
| + waiterObjId + "),EID(" + waiterObj.getEntityInfoSlot() + ")"); |
| } |
| waiterObj.notifyAll(); |
| } |
| } else { |
| break; |
| } |
| waiterObjId = waiterObj.getNextWaiterObjId(); |
| } |
| } |
| } |
| |
| @Override |
| public String prettyPrint() throws ACIDException { |
| StringBuilder s = new StringBuilder("\n########### LockManager Status #############\n"); |
| return s + "\n"; |
| } |
| |
| public void sweepForTimeout() throws ACIDException { |
| JobInfo jobInfo; |
| int waiterObjId; |
| LockWaiter waiterObj; |
| |
| latchLockTable(); |
| |
| Iterator<Entry<JobId, JobInfo>> iter = jobHT.entrySet().iterator(); |
| while (iter.hasNext()) { |
| Map.Entry<JobId, JobInfo> pair = (Map.Entry<JobId, JobInfo>) iter.next(); |
| jobInfo = pair.getValue(); |
| waiterObjId = jobInfo.getFirstWaitingResource(); |
| while (waiterObjId != -1) { |
| waiterObj = lockWaiterManager.getLockWaiter(waiterObjId); |
| toutDetector.checkAndSetVictim(waiterObj); |
| waiterObjId = waiterObj.getNextWaiterObjId(); |
| } |
| } |
| |
| unlatchLockTable(); |
| } |
| } |
| |
| class ConsecutiveWakeupContext { |
| private boolean IS; |
| private boolean IX; |
| private boolean S; |
| private boolean X; |
| |
| public void reset() { |
| IS = false; |
| IX = false; |
| S = false; |
| X = false; |
| } |
| |
| public boolean isCompatible(byte lockMode) { |
| switch (lockMode) { |
| case LockMode.IX: |
| return !S && !X; |
| |
| case LockMode.IS: |
| return !X; |
| |
| case LockMode.X: |
| return !IS && !IX && !S && !X; |
| |
| case LockMode.S: |
| return !IX && !X; |
| |
| default: |
| throw new IllegalStateException("Invalid upgrade lock mode"); |
| } |
| } |
| |
| public void setLockMode(byte lockMode) { |
| switch (lockMode) { |
| case LockMode.IX: |
| IX = true; |
| return; |
| |
| case LockMode.IS: |
| IS = true; |
| return; |
| |
| case LockMode.X: |
| X = true; |
| return; |
| |
| case LockMode.S: |
| S = true; |
| return; |
| |
| default: |
| throw new IllegalStateException("Invalid lock mode"); |
| } |
| |
| } |
| |
| } |
