Verify that the lock was acquired.  This shouldn't be called unless it's part of a recipe
built on top of ColumnPrefixDistributedRowLock.

@param curTimeInMicros
@throws BusyLockException
979332
public void verifyLock(long curTimeInMicros) throws Exception, BusyLockException, StaleLockException {
        if (lockColumn == null) 
            throw new IllegalStateException("verifyLock() called without attempting to take the lock");
        
        // Read back all columns. There should be only 1 if we got the lock
        Map<String, Long> lockResult = readLockColumns(readDataColumns);

        // Cleanup and check that we really got the lock
        for (Entry<String, Long> entry : lockResult.entrySet()) {
            // This is a stale lock that was never cleaned up
            if (entry.getValue() != 0 && curTimeInMicros > entry.getValue()) {
                if (failOnStaleLock) {
                    throw new StaleLockException("Stale lock on row '" + key + "'.  Manual cleanup requried.");
                }
                locksToDelete.add(entry.getKey());
            }
            // Lock already taken, and not by us
            else if (!entry.getKey().equals(lockColumn)) {
                throw new BusyLockException("Lock already acquired for row '" + key + "' with lock column " + entry.getKey());
            }
        }
    }
653970
private IEntityLock[] primSelect(String sql) throws LockingException {
        Connection conn = null;
        Statement stmnt = null;
        ResultSet rs = null;
        List locks = new ArrayList();

        if (log.isDebugEnabled()) log.debug("RDBMEntityLockStore.primSelect(): " + sql);

        try {
            conn = RDBMServices.getConnection();
            stmnt = conn.createStatement();
            try {
                rs = stmnt.executeQuery(sql);
                try {
                    while (rs.next()) {
                        locks.add(instanceFromResultSet(rs));
                    }
                } finally {
                    rs.close();
                }
            } finally {
                stmnt.close();
            }
        } catch (SQLException sqle) {
            log.error(sqle, sqle);
            throw new LockingException("Problem retrieving EntityLocks", sqle);
        } finally {
            RDBMServices.releaseConnection(conn);
        }

        return ((IEntityLock[]) locks.toArray(new IEntityLock[locks.size()]));
    }
1092763
@Override
    public void add(IEntityLock lock) throws LockingException {
        Connection conn = null;
        try {
            conn = RDBMServices.getConnection();
            primDeleteExpired(new Date(), lock.getEntityType(), lock.getEntityKey(), conn);
            primAdd(lock, conn);
        } catch (SQLException sqle) {
            throw new LockingException("Problem creating " + lock, sqle);
        } finally {
            RDBMServices.releaseConnection(conn);
        }
    }
1076980
@WithBridgeMethods(value = MySQLQuery.class, castRequired = true)
    public C lockInShareMode() {
        return addFlag(Position.END, LOCK_IN_SHARE_MODE);
    }
1001787
private void checkParams(int maxActiveMessages,
                             long messageLockExpiry,
                             int maxBatchSize)

                    throws SIIncorrectCallException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(
                        tc,
                        "checkParams",
                        new Object[] {
                                      Integer.valueOf(maxActiveMessages),
                                      Long.valueOf(messageLockExpiry),
                                      Integer.valueOf(maxBatchSize) });

        if (maxActiveMessages < 0)
        {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "checkParams", "SIIncorrectCallException maxActiveMessages < 0");
            throw new SIIncorrectCallException(
                            nls_cwsir.getFormattedMessage(
                                                          "REG_ASYNCH_CONSUMER_ERROR_CWSIR0141",
                                                          new Object[] { Integer.valueOf(maxActiveMessages) },
                                                          null));
        }

        if (messageLockExpiry < 0)
        {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "checkParams", "SIIncorrectCallException messageLockExpiry < 0");
            throw new SIIncorrectCallException(
                            nls_cwsir.getFormattedMessage(
                                                          "REG_ASYNCH_CONSUMER_ERROR_CWSIR0142",
                                                          new Object[] { Long.valueOf(messageLockExpiry) },
                                                          null));
        }

        if (maxBatchSize <= 0)
        {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "checkParams", "SIIncorrectCallException maxBatchSize <= 0");

            throw new SIIncorrectCallException(
                            nls_cwsir.getFormattedMessage(
                                                          "REG_ASYNCH_CONSUMER_ERROR_CWSIR0143",
                                                          new Object[] { Integer.valueOf(maxBatchSize) },
                                                          null));
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "checkParams");
    }
309042
private static boolean tryUpgradeUpdateToWriteLockMillis(
            long address, long timeInMillis, boolean interruptible) throws InterruptedException {
        long lastTime = System.currentTimeMillis();
        registerWait(address);
        try {
            do {
                if (tryUpgradeUpdateToWriteLockAndDeregisterWait0(address))
                    return true;
                checkInterrupted(interruptible);
                ThreadHints.onSpinWait();
                long now = System.currentTimeMillis();
                if (now != lastTime) {
                    lastTime = now;
                    timeInMillis--;
                }
            } while (timeInMillis >= 0);
            deregisterWait(address);
            return false;
        } catch (Throwable t) {
            throw tryDeregisterWaitAndRethrow(address, t);
        }
    }
785290
@Override
    public void commitChanges(SynchronizationBarrier barrier) throws JMSException
    {
    	checkNotClosed();
    	checkTransactionLock();
    	
    	if (!committables.isEmpty())
        {
    		long start = System.currentTimeMillis();
    		
    		Iterator<Committable> allCommitables = committables.iterator();
        	while (allCommitables.hasNext())
        	{
        		Committable committable = allCommitables.next();
        		committable.commitChanges(barrier);
        	}
        	
        	long end = System.currentTimeMillis();
        	notifyCommitTime(end-start);
        	
        	pendingChanges = false;
        }
    }
929522
private static boolean tryWriteLockMillis(
            long address, long timeInMillis, boolean interruptible) throws InterruptedException {
        long lastTime = System.currentTimeMillis();
        registerWait(address);
        try {
            do {
                if (LOCK.tryWriteLockAndDeregisterWait(A, null, address + LOCK_OFFSET))
                    return true;
                checkInterrupted(interruptible);
                ThreadHints.onSpinWait();
                long now = System.currentTimeMillis();
                if (now != lastTime) {
                    lastTime = now;
                    timeInMillis--;
                }
            } while (timeInMillis >= 0);
            deregisterWait(address);
            return false;
        } catch (Throwable t) {
            throw tryDeregisterWaitAndRethrow(address, t);
        }
    }
371165
protected synchronized void lock(ManagedObject managedObject)
                    throws ObjectManagerException
    {
        throw new InvalidStateException(this, InternalTransaction.stateTerminated, InternalTransaction.stateNames[InternalTransaction.stateTerminated]);
    }
1301778
public final void lockForReadInterruptibly(L locker) throws InterruptedException {
        if (Thread.interrupted()) {
            throw new InterruptedException();
        }
        if (!tryLockForRead(locker)) {
            lockForReadQueuedInterruptibly(locker, addReadWaiter());
        }
    }
1122727
@Override
	public Object visitPrepareCommand(TxInvocationContext ctx, PrepareCommand command) throws Throwable {
		if (ctx.isOriginLocal()) {
			// We can't wait to commit phase to remove the entry locally (invalidations are processed in 1pc
			// on remote nodes, so only local case matters here). The problem is that while the entry is locked
			// reads still can take place and we can read outdated collection after reading updated entity
			// owning this collection from DB; when this happens, the version lock on entity cannot protect
			// us against concurrent modification of the collection. Therefore, we need to remove the entry
			// here (even without lock!) and let possible update happen in commit phase.
			for (WriteCommand wc : command.getModifications()) {
				for (Object key : wc.getAffectedKeys()) {
					dataContainer.remove(key);
				}
			}
		}
		else {
			for (WriteCommand wc : command.getModifications()) {
            Collection<?> keys = wc.getAffectedKeys();
				if (log.isTraceEnabled()) {
					log.tracef("Invalidating keys %s with lock owner %s", keys, ctx.getLockOwner());
				}
				for (Object key : keys ) {
					putFromLoadValidator.beginInvalidatingKey(ctx.getLockOwner(), key);
				}
			}
		}
		return invokeNext(ctx, command);
	}
907503
final boolean addStoredBlockInternal(Block block,
                               DatanodeDescriptor node,
                               DatanodeDescriptor delNodeHint,
                               boolean initialBlockReport,
                               InitialReportWorker worker) throws IOException {
    InjectionHandler.processEvent(InjectionEvent.FSNAMESYSTEM_ADDSTORED_BLOCK,
        node);
    // either this is a direct call protected by writeLock, or
    // this is parallel initial block report
    assert (hasWriteLock() || (worker != null));
    
    // the call is invoked by the initial block report worker thread
    final boolean parallelInitialBlockReport = (worker != null);
    
    BlockInfo storedBlock = blocksMap.getStoredBlock(block);
    
    // handle the case for standby when we receive a block
    // that we don't know about yet - this is for block reports,
    // where we do not explicitly check this beforehand
    if (storedBlock == null && nameNode.shouldRetryAbsentBlocks() // standby
        && nameNode.shouldRetryAbsentBlock(block, storedBlock)) {
      // block should be retried
      return false;
    }
    
    if (storedBlock == null) {
      // then we need to do some special processing.
      storedBlock = blocksMap.getStoredBlockWithoutMatchingGS(block);

      if (storedBlock == null) {
        rejectAddStoredBlock(
          new Block(block), node,
          "Block not in blockMap with any generation stamp",
          initialBlockReport, parallelInitialBlockReport);
        return false;
      }

      INodeFile inode = storedBlock.getINode();
      if (inode == null) {
        rejectAddStoredBlock(
          new Block(block), node,
          "Block does not correspond to any file",
          initialBlockReport, parallelInitialBlockReport);
        return false;
      }

      boolean reportedOldGS = block.getGenerationStamp() < storedBlock.getGenerationStamp();
      boolean reportedNewGS = block.getGenerationStamp() > storedBlock.getGenerationStamp();
      boolean underConstruction = inode.isUnderConstruction();
      boolean isLastBlock = inode.getLastBlock() != null &&
        inode.getLastBlock().getBlockId() == block.getBlockId();

      // Don't add blocks to the DN when they're part of the in-progress last block
      // and have an inconsistent generation stamp. Instead just add them to targets
      // for recovery purposes. They will get added to the node when
      // commitBlockSynchronization runs
      if (reportedOldGS || reportedNewGS) {  // mismatched generation stamp
        if (underConstruction && isLastBlock) {
          NameNode.stateChangeLog.info(
              "BLOCK* NameSystem.addStoredBlock: "
              + "Targets updated: block " + block + " on " + node.getName() +
              " is added as a target for block " + storedBlock + " with size " +
              block.getNumBytes());
          
          lockParallelBRLock(parallelInitialBlockReport);
          try {
            ((INodeFileUnderConstruction) inode).addTarget(node,
                block.getGenerationStamp());
          } finally {
            unlockParallelBRLock(parallelInitialBlockReport);
          }
        } else {
          rejectAddStoredBlock(
              new Block(block), node,
              "Reported block has mismatched generation stamp " +
              "but is not the last block of " +
              "an under-construction file. (current generation is " +
              storedBlock.getGenerationStamp() + ")",
              initialBlockReport, parallelInitialBlockReport);
        }
        return false;
      }
    }

    INodeFile fileINode = storedBlock.getINode();
    if (fileINode == null) {
      rejectAddStoredBlock(
          new Block(block), node,
          "Block does not correspond to any file",
          initialBlockReport, parallelInitialBlockReport);
      return false;
    }

    assert storedBlock != null : "Block must be stored by now";

    // add block to the data-node
    boolean added;
    if (!parallelInitialBlockReport) {
      // full insert
      added = node.addBlock(storedBlock);
    } else {
      // insert DN descriptor into stored block
      int index = node.addBlockWithoutInsertion(storedBlock);
      added = index >= 0;
      // inform the worker so it can insert it into its local list
      worker.setCurrentStoredBlock(storedBlock, index);
    }
    
    // Is the block being reported the last block of an underconstruction file?
    boolean blockUnderConstruction = false;
    
    if (fileINode.isUnderConstruction()) {
      Block last = fileINode.getLastBlock();
      if (last == null) {
        // This should never happen, but better to handle it properly than to throw
        // an NPE below.
        LOG.error("Null blocks for reported block=" + block + " stored=" + storedBlock +
          " inode=" + fileINode);
        return false;
      }
      blockUnderConstruction = last.equals(storedBlock);
    }

    // block == storedBlock when this addStoredBlock is the result of a block report
    if (block.getNumBytes() != storedBlock.getNumBytes()) {
      if (!checkBlockSize(block, storedBlock.getINode())) {
        try {
          // New replica has an invalid block size. Mark it as corrupted.
          LOG.warn("Mark new replica " + block + " from " + node.getName() +
            "as corrupt because its length " + block.getNumBytes() +
            " is not valid");
          markBlockAsCorrupt(block, node, parallelInitialBlockReport);
        } catch (IOException e) {
          LOG.warn("Error in deleting bad block " + block + e);
        }
      } else {
        long cursize = storedBlock.getNumBytes();
        if (cursize == 0) {
          storedBlock.setNumBytes(block.getNumBytes());
        } else if (cursize != block.getNumBytes()) {
          String logMsg = "Inconsistent size for block " + block + 
            " reported from " + node.getName() +
            " current size is " + cursize +
            " reported size is " + block.getNumBytes();
          // If the block is still under construction this isn't likely
          // to be a problem, so just log at INFO level.
          if (blockUnderConstruction) {
            if (cursize != 1) { // cursize == 1 implies block was fsynced
              LOG.info(logMsg);
            }
          } else {
            LOG.warn(logMsg);
          }

          try {
            if (cursize > block.getNumBytes() && !blockUnderConstruction) {
              // new replica is smaller in size than existing block.
              // Mark the new replica as corrupt.
              LOG.warn("Mark new replica " + block + " from " + node.getName() +
                "as corrupt because its length is shorter than existing ones");
              markBlockAsCorrupt(block, node, parallelInitialBlockReport);
            } else {
              // new replica is larger in size than existing block.
              if (!blockUnderConstruction) {
                // Mark pre-existing replicas as corrupt.
                int numNodes = blocksMap.numNodes(block);
                int count = 0;
                DatanodeDescriptor nodes[] = new DatanodeDescriptor[numNodes];
                Iterator<DatanodeDescriptor> it = blocksMap.nodeIterator(block);
                for (; it != null && it.hasNext(); ) {
                  DatanodeDescriptor dd = it.next();
                  if (!dd.equals(node)) {
                    nodes[count++] = dd;
                  }
                }
                for (int j = 0; j < count; j++) {
                  LOG.warn("Mark existing replica " + block + " from " + node.getName() +
                  " as corrupt because its length is shorter than the new one");
                  markBlockAsCorrupt(block, nodes[j], parallelInitialBlockReport);
                }
              }
              //
              // change the size of block in blocksMap
              //
              storedBlock.setNumBytes(block.getNumBytes());
            }
          } catch (IOException e) {
            LOG.warn("Error in deleting bad block " + block + e);
          }
        }
      }
      block = storedBlock;
    } else {
      block = storedBlock;
    }
    assert storedBlock == block : "Block must be stored by now";

    int curReplicaDelta = 0;

    if (added) {
      curReplicaDelta = 1;
    } else {
      NameNode.stateChangeLog.info("BLOCK* NameSystem.addStoredBlock: "
        + "Redundant addStoredBlock request received for "
        + block + " on " + node.getName()
        + " size " + block.getNumBytes());
    }

    // filter out containingNodes that are marked for decommission.
    NumberReplicas num = countNodes(storedBlock);
    int numCurrentReplica = 0;
    int numLiveReplicas = num.liveReplicas();

    boolean popReplQueuesBefore = isPopulatingReplQueuesInternal();

    if (!popReplQueuesBefore) {
      // if we haven't populated the replication queues
      // then use a cheaper method to count
      // we only need live and decommissioned replicas.
      numCurrentReplica = numLiveReplicas + num.decommissionedReplicas();
    } else {
      // count live & decommissioned replicas
      numCurrentReplica = numLiveReplicas + pendingReplications.getNumReplicas(block);
    }

    if (blockUnderConstruction) {
      lockParallelBRLock(parallelInitialBlockReport);
      try {
        added = ((INodeFileUnderConstruction) fileINode).addTarget(node,
            block.getGenerationStamp());
      } finally {
        unlockParallelBRLock(parallelInitialBlockReport);
      }
    }

    // check whether safe replication is reached for the block. Do not increment
    // safe block count if this block has already been reported by the same
    // datanode before. We used the check for 'added' to achieve this.
    if (added && isInSafeModeInternal()) {
      int numSafeReplicas = numLiveReplicas + num.decommissionedReplicas();
      if (blockUnderConstruction) {
        // If this is the last block under construction then all the replicas so
        // far have been added to the "targets" field of
        // INodeFileUnderConstruction and hence lookup the replication factor
        // from there.
        // In this code path, a "valid" replica is added to the block, and we
        // only count "valid" replica to be safe replica. It is possible that
        // earlier a replica with higher generation stamp has been added and a
        // false was returned. In that case, if we only count number of targets
        // we will miss to count in the case that we first accepted more than
        // minReplication's number of higher generation stamp blocks but later
        // blocks with "valid" generation stamps themselves reached minimum
        // replication again.
        DatanodeDescriptor[] validDNs = ((INodeFileUnderConstruction) fileINode)
            .getValidTargets();
        numSafeReplicas = (validDNs != null) ? validDNs.length : 0;
      }
      
      if (!parallelInitialBlockReport && added) {
        // regular add stored block
        incrementSafeBlockCount(numSafeReplicas, initialBlockReport);
      } else if (parallelInitialBlockReport && added) {
        // for parallel initial block report increment local worker variable
        worker.incrementSafeBlockCount(numSafeReplicas);
      }
    }

    if (!popReplQueuesBefore && isPopulatingReplQueuesInternal()) {
      // we have just initialized the repl queues
      // must recompute num
      num = countNodes(storedBlock);
      numLiveReplicas = num.liveReplicas();
      numCurrentReplica = numLiveReplicas +
        pendingReplications.getNumReplicas(block);
    }

    //
    // if file is being actively written to and it is the last block, 
    // then do not check replication-factor here.
    //
    if (blockUnderConstruction) {
      return true;
    }

    // do not handle mis-replicated blocks during start up
    if (!isPopulatingReplQueuesInternal()) {
      return true;
    }

    if (curReplicaDelta == 0) {
      return true;
    }
    
    // handle underReplication
    short blockReplication = fileINode.getBlockReplication(storedBlock);
    updateNeededReplicationQueue(storedBlock, curReplicaDelta, numCurrentReplica,
        num.decommissionedReplicas, node, blockReplication);
    // handle over-replication
    if (numCurrentReplica > blockReplication) {
      // Put block into a queue and handle excess block asyncly
      if (delNodeHint == null || node == delNodeHint) {
        overReplicatedBlocks.add(block);
      } else {
        overReplicatedBlocks.add(new OverReplicatedBlock(block, node, delNodeHint));
      }
    }
    // If the file replication has reached desired value
    // we can remove any corrupt replicas the block may have
    int corruptReplicasCount = corruptReplicas.numCorruptReplicas(block);
    int numCorruptNodes = num.corruptReplicas();
    if (numCorruptNodes != corruptReplicasCount) {
      LOG.warn("Inconsistent number of corrupt replicas for " +
        block + "blockMap has " + numCorruptNodes +
        " but corrupt replicas map has " + corruptReplicasCount);
    }
    if ((corruptReplicasCount > 0) && (numLiveReplicas >= blockReplication)) {
      invalidateCorruptReplicas(block);
    }
    return true;
  }
997889
@Override
    public void lock(Object entity, LockModeType lockMode)
    {
        ivEm.lock(entity, lockMode);
    }
228924
private boolean isLocked(Class entityType, String entityKey, Integer lockType)
            throws LockingException {
        IEntityLock[] locks = retrieveLocks(entityType, entityKey, lockType);
        return locks.length > 0;
    }
1362593
@Override
	public DistributedSchedulingLock updateNGetDistributedScheduleByType(LockType type,int jobsBlockSize, long schedulingRefreshInterval) {
		requireNotDisposed();
		requireArgument(type != null, "Lock type cannot be null.");

		EntityManager em = emf.get();
		while(true){
			try{
				DistributedSchedulingLock distributedSchedulingLock = DistributedSchedulingLock.updateNGetDistributedScheduleByType(em, type, jobsBlockSize,  schedulingRefreshInterval);
				return distributedSchedulingLock;
			}catch(OptimisticLockException ex){
				_logger.info("Optimistic lock exception " + ex.toString());  
			}catch(Throwable th){
				_logger.error("Optimistic lock exception " + th.toString());   
			}
		}

	}
530964
public LockResource lockInode(InodeView inode, LockMode mode) {
    return mInodeLocks.get(inode.getId(), mode);
  }
643180
public void removeLocked( MessageLock lockRef ) throws JMSException
    {
    	checkTransactionLock();
    	
    	MessageStore targetStore;
    	if (lockRef.getDeliveryMode() == DeliveryMode.NON_PERSISTENT)
    	    targetStore = volatileStore;
    	else
    	{
    	    targetStore = persistentStore;
    	    if (requiresTransactionalUpdate())
    	    	pendingChanges = true;
    	}
    	
    	synchronized (storeLock)
        {
    		targetStore.delete(lockRef.getHandle());
        }
    }