LogicalTxnLog Class Reference

LogicalTxnLog defines the log structure used by LogicalTxn to record transaction data. More...

#include <LogicalTxnLog.h>

Inheritance diagram for LogicalTxnLog:


Public Member Functions
virtual	~LogicalTxnLog ()
void	setNextTxnId (TxnId nextTxnIdInit)
	Sets the nextTxnId.
SharedLogicalTxn	newLogicalTxn (SharedCacheAccessor pCacheAccessor)
	Starts a new LogicalTxn.
void	checkpoint (LogicalTxnLogCheckpointMemento &memento, CheckpointType checkpointType=CHECKPOINT_FLUSH_ALL)
	Checkpoints all active transactions.
void	deallocateCheckpointedLog (LogicalTxnLogCheckpointMemento const &memento, CheckpointType checkpointType=CHECKPOINT_FLUSH_ALL)
	Releases log space used by transactions which committed before current checkpoint.
TxnId	getOldestActiveTxnId ()
	Returns the transaction id of the oldest, active transaction.
Static Public Member Functions
static SharedLogicalTxnLog	newLogicalTxnLog (SegmentAccessor const &logSegmentAccessor, PseudoUuid const &onlineUuid, SharedSegmentFactory pSegmentFactory)
	Creates a new LogicalTxnLog in the given segment.
Protected Attributes
StrictMutex	mutex
LocalCondition	condition
Private Types
typedef std::vector< SharedLogicalTxn >	TxnList
typedef std::vector< SharedLogicalTxn >::iterator	TxnListIter
typedef std::vector< SharedSegment >	SegList
typedef std::vector< SharedSegment >::iterator	SegListIter
Private Member Functions
	LogicalTxnLog (SegmentAccessor const &logSegmentAccessor, PseudoUuid const &onlineUuid, SharedSegmentFactory pSegmentFactory)
void	removeTxn (SharedLogicalTxn pTxn)
void	commitTxn (SharedLogicalTxn pTxn)
void	rollbackTxn (SharedLogicalTxn pTxn)
void	checkpointTxn (SharedLogicalTxn pTxn)
void	commitTxnWithGroup (StrictMutexGuard &mutexGuard)
Private Attributes
SharedSegmentFactory	pSegmentFactory
	SegmentFactory to use for creating long log segments.
TxnId	nextTxnId
	TxnId generator.
TxnList	uncommittedTxns
	Collection of active txns which have not yet committed.
SegList	committedLongLogSegments
	Log segments for long log transactions which committed after the last checkpoint.
SegmentAccessor	logSegmentAccessor
	Accessor for segment for the main transaction log.
SharedSegOutputStream	pOutputStream
	Output stream for writing to the main transaction log.
LogicalTxnLogCheckpointMemento	lastCheckpointMemento
	Memento recorded at last checkpoint.
uint	nCommittedBeforeLastCheckpoint
	Size of committedLongLogSegments at time lastCheckpointMemento was recorded.
uint	groupCommitInterval
	Group commit interval in milliseconds, or 0 for immediate commit.
Friends
class	LogicalTxn

Detailed Description

LogicalTxnLog defines the log structure used by LogicalTxn to record transaction data.

Definition at line 40 of file LogicalTxnLog.h.

Member Typedef Documentation

typedef std::vector<SharedLogicalTxn> LogicalTxnLog::TxnList [private]

Definition at line 47 of file LogicalTxnLog.h.

typedef std::vector<SharedLogicalTxn>::iterator LogicalTxnLog::TxnListIter [private]

Definition at line 48 of file LogicalTxnLog.h.

typedef std::vector<SharedSegment> LogicalTxnLog::SegList [private]

Definition at line 50 of file LogicalTxnLog.h.

typedef std::vector<SharedSegment>::iterator LogicalTxnLog::SegListIter [private]

Definition at line 51 of file LogicalTxnLog.h.

Constructor & Destructor Documentation

LogicalTxnLog::LogicalTxnLog	(	SegmentAccessor const &	logSegmentAccessor,
		PseudoUuid const &	onlineUuid,
		SharedSegmentFactory	pSegmentFactory
	)			`[explicit, private]`

Definition at line 42 of file LogicalTxnLog.cpp.

References groupCommitInterval, lastCheckpointMemento, LogicalTxnLogCheckpointMemento::logPosition, logSegmentAccessor, nCommittedBeforeLastCheckpoint, CrcSegOutputStream::newCrcSegOutputStream(), LogicalTxnLogCheckpointMemento::nUncommittedTxns, SegmentAccessor::pCacheAccessor, pOutputStream, pSegmentFactory, and WRITE_EAGER_SYNC.

Referenced by newLogicalTxnLog().

00046     : pSegmentFactory(pSegmentFactoryInit),
00047       logSegmentAccessor(logSegmentAccessorInit)
00048 {
00049     // Set up cache accessor so that all page locks will be taken out
00050     // with a reserved TxnId.  Just for sanity-checking, set up a quota to make
00051     // sure logging never locks more than two pages at a time.
00052     logSegmentAccessor.pCacheAccessor = SharedCacheAccessor(
00053         new QuotaCacheAccessor(
00054             SharedQuotaCacheAccessor(),
00055             logSegmentAccessor.pCacheAccessor,
00056             2));
00057 
00058     // TODO: Support an option to skip CRC's for optimized non-durable logging.
00059     // Also support a paranoid option for recording CRC's for long logs.
00060     pOutputStream = CrcSegOutputStream::newCrcSegOutputStream(
00061         logSegmentAccessor,onlineUuid);
00062 
00063     // NOTE:  We only write one page at a time to the main log, and we always
00064     // need to wait after each page.  So request synchronous writes.
00065     pOutputStream->setWriteLatency(WRITE_EAGER_SYNC);
00066 
00067     pOutputStream->getSegPos(lastCheckpointMemento.logPosition);
00068     lastCheckpointMemento.nUncommittedTxns = 0;
00069     nCommittedBeforeLastCheckpoint = 0;
00070 
00071     groupCommitInterval = pSegmentFactory->getConfigMap().getIntParam(
00072         "groupCommitInterval", 30);
00073 }

LogicalTxnLog::~LogicalTxnLog ( ) [virtual]

Definition at line 90 of file LogicalTxnLog.cpp.

References committedLongLogSegments, and uncommittedTxns.

00091 {
00092     assert(uncommittedTxns.empty());
00093     assert(committedLongLogSegments.empty());
00094 }

Member Function Documentation

void LogicalTxnLog::removeTxn ( SharedLogicalTxn pTxn ) [private]

Definition at line 116 of file LogicalTxnLog.cpp.

References uncommittedTxns.

Referenced by commitTxn(), and rollbackTxn().

00117 {
00118     TxnListIter pFound = std::find(
00119         uncommittedTxns.begin(),
00120         uncommittedTxns.end(),
00121         pTxn);
00122     assert(pFound != uncommittedTxns.end());
00123     uncommittedTxns.erase(pFound);
00124 }

void LogicalTxnLog::commitTxn ( SharedLogicalTxn pTxn ) [private]

Definition at line 126 of file LogicalTxnLog.cpp.

References LogicalTxnEventMemento::cbActionLast, SegStreamPosition::cbOffset, CHECKPOINT_FLUSH_AND_UNMAP, committedLongLogSegments, commitTxnWithGroup(), LogicalTxnEventMemento::event, LogicalTxnEventMemento::EVENT_COMMIT, LogicalTxnEventMemento::logPosition, LogicalTxnEventMemento::longLog, SynchMonitoredObject::mutex, LogicalTxnEventMemento::nParticipants, NULL_PAGE_ID, pOutputStream, removeTxn(), SegStreamPosition::segByteId, CompoundId::setByteOffset(), CompoundId::setPageId(), and LogicalTxnEventMemento::txnId.

00127 {
00128     LogicalTxnEventMemento memento;
00129     memento.event = LogicalTxnEventMemento::EVENT_COMMIT;
00130     memento.txnId = pTxn->txnId;
00131     memento.cbActionLast = pTxn->svpt.cbActionPrev;
00132     memento.nParticipants = pTxn->participants.size();
00133     SharedSegment pSegment = pTxn->pOutputStream->getSegment();
00134     if (pSegment) {
00135         assert(pTxn->pOutputStream.unique());
00136         pTxn->pOutputStream->hardPageBreak();
00137         pTxn->pOutputStream->getSegOutputStream()->getSegPos(
00138             memento.logPosition);
00139         pTxn->pOutputStream.reset();
00140         pSegment->checkpoint(CHECKPOINT_FLUSH_AND_UNMAP);
00141         StrictMutexGuard mutexGuard(mutex);
00142         committedLongLogSegments.push_back(pSegment);
00143     } else {
00144         if (!pTxn->svpt.cbLogged) {
00145             // NOTE jvs 27-Feb-2006: "empty commit" is an important
00146             // optimization for queries in autocommit mode, where JDBC
00147             // specifies a commit whenever a cursor is closed.
00148             StrictMutexGuard mutexGuard(mutex);
00149             removeTxn(pTxn);
00150             return;
00151         }
00152         CompoundId::setPageId(memento.logPosition.segByteId,NULL_PAGE_ID);
00153         CompoundId::setByteOffset(
00154             memento.logPosition.segByteId,
00155             pTxn->svpt.cbLogged);
00156         memento.logPosition.cbOffset = pTxn->svpt.cbLogged;
00157     }
00158     memento.longLog = pSegment ? true : false;
00159     StrictMutexGuard mutexGuard(mutex);
00160     pOutputStream->writeValue(memento);
00161     if (!pSegment) {
00162         SharedByteInputStream pInputStream =
00163             pTxn->pOutputStream->getInputStream();
00164         uint cbActual;
00165         PConstBuffer pBuffer = pInputStream->getReadPointer(1,&cbActual);
00166         pOutputStream->writeBytes(pBuffer,cbActual);
00167     }
00168 
00169     commitTxnWithGroup(mutexGuard);
00170     removeTxn(pTxn);
00171 }

void LogicalTxnLog::rollbackTxn ( SharedLogicalTxn pTxn ) [private]

Definition at line 210 of file LogicalTxnLog.cpp.

References LogicalTxnEventMemento::cbActionLast, SegStreamPosition::cbOffset, LogicalTxnEventMemento::event, LogicalTxnEventMemento::EVENT_ROLLBACK, LogicalTxnEventMemento::logPosition, LogicalTxnEventMemento::longLog, SynchMonitoredObject::mutex, LogicalTxnEventMemento::nParticipants, NULL_PAGE_ID, pOutputStream, removeTxn(), SegStreamPosition::segByteId, CompoundId::setByteOffset(), CompoundId::setPageId(), and LogicalTxnEventMemento::txnId.

00211 {
00212     if (!pTxn->checkpointed) {
00213         // we never stored a checkpoint record for this txn, so during recovery
00214         // it can be ignored entirely
00215         StrictMutexGuard mutexGuard(mutex);
00216         removeTxn(pTxn);
00217         return;
00218     }
00219     // otherwise, write an EVENT_ROLLBACK so that the txn's fate is known
00220     // during recovery (eliminating the need for multiple passes over the log)
00221     LogicalTxnEventMemento memento;
00222     memento.event = LogicalTxnEventMemento::EVENT_ROLLBACK;
00223     memento.txnId = pTxn->txnId;
00224     memento.cbActionLast = 0;
00225     memento.nParticipants = 0;
00226     CompoundId::setPageId(memento.logPosition.segByteId,NULL_PAGE_ID);
00227     CompoundId::setByteOffset(memento.logPosition.segByteId,0);
00228     memento.logPosition.cbOffset = 0;
00229     memento.longLog = true;
00230     StrictMutexGuard mutexGuard(mutex);
00231     pOutputStream->writeValue(memento);
00232     // no need for group commit since caller doesn't need to wait for
00233     // commit confirmation
00234     removeTxn(pTxn);
00235 }

void LogicalTxnLog::checkpointTxn ( SharedLogicalTxn pTxn ) [private]

Definition at line 292 of file LogicalTxnLog.cpp.

References LogicalTxnEventMemento::cbActionLast, LogicalTxnEventMemento::event, LogicalTxnEventMemento::EVENT_CHECKPOINT, LogicalTxnEventMemento::logPosition, LogicalTxnEventMemento::longLog, LogicalTxnEventMemento::nParticipants, pOutputStream, and LogicalTxnEventMemento::txnId.

Referenced by checkpoint().

00293 {
00294     // NOTE: hardPageBreak will automatically convert small txns into large
00295     // ones.  It would probably be better to record their incomplete state in
00296     // the main log instead.
00297     LogicalTxnEventMemento memento;
00298     pTxn->describeAllParticipants();
00299     pTxn->pOutputStream->hardPageBreak();
00300     pTxn->pOutputStream->getSegOutputStream()->getSegPos(
00301         memento.logPosition);
00302     // TODO:  see previous comment on pLogSegment->checkpoint()
00303     pTxn->pOutputStream->getSegment()->checkpoint();
00304     memento.event = LogicalTxnEventMemento::EVENT_CHECKPOINT;
00305     memento.txnId = pTxn->txnId;
00306     memento.cbActionLast = pTxn->svpt.cbActionPrev;
00307     memento.longLog = true;
00308     memento.nParticipants = pTxn->participants.size();
00309     pOutputStream->writeValue(memento);
00310     pTxn->checkpointed = true;
00311 }

void LogicalTxnLog::commitTxnWithGroup ( StrictMutexGuard & mutexGuard ) [private]

Definition at line 173 of file LogicalTxnLog.cpp.

References SynchMonitoredObject::condition, convertTimeout(), CompoundId::getPageId(), groupCommitInterval, pOutputStream, and SegStreamPosition::segByteId.

Referenced by commitTxn().

00174 {
00175     boost::xtime groupCommitExpiration;
00176     if (groupCommitInterval) {
00177         convertTimeout(groupCommitInterval,groupCommitExpiration);
00178     }
00179     SegStreamPosition logPos;
00180     pOutputStream->getSegPos(logPos);
00181     PageId startPageId = CompoundId::getPageId(logPos.segByteId);
00182     for (;;) {
00183         bool timeout = true;
00184         if (groupCommitInterval) {
00185             timeout = !condition.timed_wait(mutexGuard,groupCommitExpiration);
00186 
00187             pOutputStream->getSegPos(logPos);
00188             PageId lastPageId = CompoundId::getPageId(logPos.segByteId);
00189             if (lastPageId != startPageId) {
00190                 // someone else has flushed for us
00191                 break;
00192             }
00193         }
00194 
00195         if (timeout) {
00196             // timeout:  we're in charge of flushing
00197 
00198             // NOTE:  Since we're using synchronous writes, there's no need to
00199             // checkpoint (assuming the underlying device has been correctly
00200             // initialized to write through).
00201             pOutputStream->hardPageBreak();
00202             condition.notify_all();
00203             break;
00204         } else {
00205             // spurious wakeup:  go 'round again
00206         }
00207     }
00208 }

SharedLogicalTxnLog LogicalTxnLog::newLogicalTxnLog	(	SegmentAccessor const &	logSegmentAccessor,
		PseudoUuid const &	onlineUuid,
		SharedSegmentFactory	pSegmentFactory
	)			`[static]`

Creates a new LogicalTxnLog in the given segment.

Parameters:

	logSegmentAccessor	accessor for segment to contain log data
	onlineUuid	UUID associated with log instance
	pSegmentFactory	SegmentFactory to use for creating long log segments

Returns:: shared_ptr to new LogicalTxnLog

Definition at line 81 of file LogicalTxnLog.cpp.

References LogicalTxnLog(), and logSegmentAccessor.

Referenced by Database::createTxnLog(), and LogicalTxnTest::testTxn().

00085 {
00086     return SharedLogicalTxnLog(
00087         new LogicalTxnLog(logSegmentAccessor,onlineUuid,pSegmentFactory));
00088 }

void LogicalTxnLog::setNextTxnId ( TxnId nextTxnIdInit )

Sets the nextTxnId.

Parameters:

nextTxnIdInit

nextTxnId to be set

Definition at line 75 of file LogicalTxnLog.cpp.

References logSegmentAccessor, nextTxnId, and SegmentAccessor::pCacheAccessor.

00076 {
00077     nextTxnId = nextTxnIdInit;
00078     logSegmentAccessor.pCacheAccessor->setTxnId(nextTxnId);
00079 }

SharedLogicalTxn LogicalTxnLog::newLogicalTxn ( SharedCacheAccessor pCacheAccessor )

Starts a new LogicalTxn.

Parameters:

pCacheAccessor

CacheAccessor to use for cache access to log data

Returns:: shared_ptr to new LogicalTxn

Definition at line 96 of file LogicalTxnLog.cpp.

References LogicalTxn, SynchMonitoredObject::mutex, nextTxnId, and uncommittedTxns.

00098 {
00099     StrictMutexGuard mutexGuard(mutex);
00100     // Set up cache accessor so that all page locks will be taken out
00101     // with the new TxnId.  Just for sanity-checking, set up a quota to make
00102     // sure logging never locks more than two pages at a time.
00103     pCacheAccessor = SharedCacheAccessor(
00104         new QuotaCacheAccessor(
00105             SharedQuotaCacheAccessor(),
00106             pCacheAccessor,
00107             2));
00108     pCacheAccessor->setTxnId(nextTxnId);
00109     SharedLogicalTxn pTxn(
00110         new LogicalTxn(nextTxnId,shared_from_this(),pCacheAccessor));
00111     uncommittedTxns.push_back(pTxn);
00112     ++nextTxnId;
00113     return pTxn;
00114 }

void LogicalTxnLog::checkpoint	(	LogicalTxnLogCheckpointMemento &	memento,
		CheckpointType	checkpointType = `CHECKPOINT_FLUSH_ALL`
	)

Checkpoints all active transactions.

Assumes that all transactions have already been quiesced for the duration of the checkpoint. In most cases, deallocateCheckpointedLog must also be called after the checkpoint.

Parameters:

	memento	receives the checkpoint record
	checkpointType	type of checkpoint to execute

Definition at line 237 of file LogicalTxnLog.cpp.

References CHECKPOINT_DISCARD, CHECKPOINT_FLUSH_FUZZY, checkpointTxn(), committedLongLogSegments, lastCheckpointMemento, LogicalTxnLogCheckpointMemento::logPosition, logSegmentAccessor, SynchMonitoredObject::mutex, nextTxnId, LogicalTxnLogCheckpointMemento::nextTxnId, LogicalTxnLogCheckpointMemento::nUncommittedTxns, pOutputStream, SegmentAccessor::pSegment, and uncommittedTxns.

00240 {
00241     StrictMutexGuard mutexGuard(mutex);
00242     if (checkpointType == CHECKPOINT_DISCARD) {
00243         uncommittedTxns.clear();
00244         committedLongLogSegments.clear();
00245         return;
00246     }
00247     pOutputStream->getSegPos(memento.logPosition);
00248     memento.nUncommittedTxns = uncommittedTxns.size();
00249     memento.nextTxnId = nextTxnId;
00250     std::for_each(
00251         uncommittedTxns.begin(),
00252         uncommittedTxns.end(),
00253         boost::bind(&LogicalTxnLog::checkpointTxn,this,_1));
00254     pOutputStream->hardPageBreak();
00255     logSegmentAccessor.pSegment->checkpoint(checkpointType);
00256     if (checkpointType == CHECKPOINT_FLUSH_FUZZY) {
00257         // memento gets lastCheckpointMemento, and lastCheckpointMemento gets
00258         // new memento just created above
00259         std::swap(memento,lastCheckpointMemento);
00260     }
00261 }

void LogicalTxnLog::deallocateCheckpointedLog	(	LogicalTxnLogCheckpointMemento const &	memento,
		CheckpointType	checkpointType = `CHECKPOINT_FLUSH_ALL`
	)

Releases log space used by transactions which committed before current checkpoint.

Divorced from the checkpoint operation itself to make possible atomicity as part of compound checkpoint sequences.

Parameters:

	memento	memento returned by last checkpoint() call
	checkpointType	type of checkpoint passed to the last checkpoint() call

Definition at line 263 of file LogicalTxnLog.cpp.

References CHECKPOINT_FLUSH_FUZZY, committedLongLogSegments, CompoundId::decBlockNum(), FIRST_LINEAR_PAGE_ID, CompoundId::getPageId(), LogicalTxnLogCheckpointMemento::logPosition, logSegmentAccessor, nCommittedBeforeLastCheckpoint, NULL_PAGE_ID, SegmentAccessor::pSegment, and SegStreamPosition::segByteId.

00266 {
00267     PageId lastObsoletePageId =
00268         CompoundId::getPageId(memento.logPosition.segByteId);
00269     if (lastObsoletePageId != FIRST_LINEAR_PAGE_ID) {
00270         assert(lastObsoletePageId != NULL_PAGE_ID);
00271         // Segment::deallocatePageRange is inclusive, so decrement to
00272         // exclude the checkpoint page
00273         CompoundId::decBlockNum(lastObsoletePageId);
00274         if (logSegmentAccessor.pSegment->isPageIdAllocated(
00275                 lastObsoletePageId))
00276         {
00277             logSegmentAccessor.pSegment->deallocatePageRange(
00278                 NULL_PAGE_ID,lastObsoletePageId);
00279         }
00280     }
00281 
00282     if (checkpointType == CHECKPOINT_FLUSH_FUZZY) {
00283         committedLongLogSegments.erase(
00284             committedLongLogSegments.begin(),
00285             committedLongLogSegments.begin() + nCommittedBeforeLastCheckpoint);
00286     } else {
00287         committedLongLogSegments.clear();
00288     }
00289     nCommittedBeforeLastCheckpoint = committedLongLogSegments.size();
00290 }

TxnId LogicalTxnLog::getOldestActiveTxnId ( )

Returns the transaction id of the oldest, active transaction.

Returns:: txnId of the oldest, active txn; if there are no active transactions, returns the current transaction id

Definition at line 313 of file LogicalTxnLog.cpp.

References SynchMonitoredObject::mutex, nextTxnId, NULL_TXN_ID, and uncommittedTxns.

00314 {
00315     StrictMutexGuard mutexGuard(mutex);
00316     TxnId oldestTxnId = NULL_TXN_ID;
00317     for (TxnListIter ppTxn = uncommittedTxns.begin();
00318         ppTxn != uncommittedTxns.end();
00319         ++ppTxn)
00320     {
00321         SharedLogicalTxn pTxn = *ppTxn;
00322         if (oldestTxnId == NULL_TXN_ID || pTxn->getTxnId() < oldestTxnId) {
00323             oldestTxnId = pTxn->getTxnId();
00324         }
00325     }
00326 
00327     // If there are no active txns, return the txnId that will be assigned to
00328     // the next, new txn
00329     if (oldestTxnId == NULL_TXN_ID) {
00330         return nextTxnId;
00331     } else {
00332         return oldestTxnId;
00333     }
00334 }