// statistics
int clrskipped = 0;
int logrecordseen = 0;
RePreparable lop = null;
// stream to read the log record - initial size 4096, scanLog needs
// to resize if the log record is larger than that.
ArrayInputStream rawInput = null;
try
{
StreamLogScan scanLog;
if (prepareStartAt == null)
{
// don't know where to start, scan from end of log
scanLog =
(StreamLogScan) logFactory.openBackwardsScan(prepareStopAt);
}
else
{
if (prepareStartAt.lessThan(prepareStopAt))
{
// nothing to prepare!
return;
}
scanLog = (StreamLogScan)
logFactory.openBackwardsScan(
((LogCounter) prepareStartAt).getValueAsLong(),
prepareStopAt);
}
if (SanityManager.DEBUG)
SanityManager.ASSERT(
scanLog != null, "cannot open log for prepare");
rawInput = new ArrayInputStream(new byte[4096]);
LogRecord record;
while ((record =
scanLog.getNextRecord(rawInput, prepareId, 0))
!= null)
{
if (SanityManager.DEBUG)
{
SanityManager.ASSERT(
record.getTransactionId().equals(prepareId),
"getNextRecord return unqualified log rec for prepare");
}
logrecordseen++;
if (record.isCLR())
{
clrskipped++;
// the loggable is still in the input stream, get rid of it
record.skipLoggable();
// read the prepareInstant
long prepareInstant = rawInput.readLong();
if (SanityManager.DEBUG)
{
if (SanityManager.DEBUG_ON(LogToFile.DBG_FLAG))
{
SanityManager.DEBUG(
LogToFile.DBG_FLAG,
"Skipping over CLRs, reset scan to " +
LogCounter.toDebugString(prepareInstant));
}
}
scanLog.resetPosition(new LogCounter(prepareInstant));
// scanLog now positioned at the beginning of the log
// record that was rolled back by this CLR.
// The scan is a backward one so getNextRecord will skip
// over the record that was rolled back and go to the one
// previous to it
continue;
}
if (record.requiresPrepareLocks())
{
lop = record.getRePreparable();
}
else
{
continue;
}
if (lop != null)
{
// Reget locks based on log record. reclaim all locks with
// a serializable locking policy, since we are only
// reclaiming write locks, isolation level does not matter
// much.
lop.reclaimPrepareLocks(
t,
t.newLockingPolicy(
LockingPolicy.MODE_RECORD,
TransactionController.ISOLATION_REPEATABLE_READ,
true));