* This method will wait until the transaction's block time has passed.
* @return
* @throws InterruptedException
*/
public AbstractTransaction take() throws InterruptedException {
AbstractTransaction retval = null;
// Ok now here is the tricky part. We don't have a txn that is
// ready to run, so we need to block ourselves until we get one.
// This could be for two reasons:
// (1) The queue is empty.
// (2) The waiting period for the next txn hasn't passed yet.
//
// Note that we can't simply attach ourselves to our inner queue because
// we don't want to get back the txn right when it gets added.
// We want to wait until the time period has passed.
if (trace.val)
LOG.trace(String.format("Partition %d :: Attempting to acquire lock", this.partitionId));
this.lock.lockInterruptibly();
try {
if (debug.val && this.state != QueueState.UNBLOCKED)
LOG.debug(String.format("Partition %d :: take() -> " +
"Current state is %s. Blocking until ready", this.partitionId, this.state));
while (this.state != QueueState.UNBLOCKED) {
if (trace.val)
LOG.trace(String.format("Partition %d :: take() -> Calculating how long to block",
this.partitionId));
long waitTime = -1;
boolean isEmpty = (this.state == QueueState.BLOCKED_EMPTY);
boolean needsUpdateQueue = false;
// If the queue isn't empty, then we need to figure out
// how long we should sleep for
if (isEmpty == false) {
// If we're blocked because of an ordering issue (i.e., we have a new txn
// in the system that is less than our current head of the queue, but we
// haven't inserted it yet), then we will want to wait for the full timeout
// period. We won't actually have to wait this long because somebody will poke
// us after the new txn is added to the queue.
if (this.state == QueueState.BLOCKED_ORDERING) {
waitTime = this.maxWaitTime;
} else {
waitTime = this.blockTimestamp - System.currentTimeMillis();
}
}
try {
// If we're empty, then we need to block indefinitely until we're poked
if (isEmpty) {
if (debug.val)
LOG.debug(String.format("Partition %d :: take() -> " +
"Blocking because queue is empty", this.partitionId));
this.isReady.await();
}
// Otherwise, we'll sleep until our time out and then
// check the queue status for ourselves
else if (waitTime > 0) {
// We are going to wait for the specified time
// The Condition will return true if somebody poked us, which
// means that somebody changed the queue state to UNBLOCKED
// for us. If we are woken because of a timeout, then the
// return status will be false, which means that we need to
// queue state ourself.
if (debug.val)
LOG.debug(String.format("Partition %d :: take() -> " +
"Blocking for %d ms", this.partitionId, waitTime));
needsUpdateQueue = (this.isReady.await(waitTime, TimeUnit.MILLISECONDS) == false);
}
// Our txn is ready to run now, so we don't need to block
else {
if (debug.val)
LOG.debug(String.format("Partition %d :: take() -> " +
"Ready to retrieve next txn immediately [waitTime=%d, isEmpty=%s]",
this.partitionId, waitTime, isEmpty));
needsUpdateQueue = true;
}
} catch (InterruptedException ex) {
this.isReady.signal();
throw ex;
}
if (needsUpdateQueue) this.checkQueueState(false);
} // WHILE
// The next txn is ready to run now!
assert(this.state == QueueState.UNBLOCKED);
retval = super.poll();
// 2012-01-06
// This could be null because there is a race condition if all of the
// txns are removed by another thread right before we try to
// poll our queue.
if (retval != null) {
this.lastTxnPopped = retval.getTransactionId();
this.txnsPopped++;
// Call this again to prime the next txn
this.checkQueueState(true);
}