bzr commit into mysql-6.0-falcon-team branch (klewis:2949) Bug#41194

[Kevin Lewis]

#At file:///C:/Work/bzr/Merge/mysql-6.0-falcon-team/

 2949 Kevin Lewis	2008-12-19
      Bug#41194 - The Transaction knows about the serial log.  
      Let it make calls to SerialLogRecord functions directly.
      Separate the serial log flush of commit and rollback records
      from when we allow the gophers to start processing them.
      Once those records are flushed, the recovery will be able to 
      process them, but we do not want the gophers to do that until 
      the commit or rollback is fully finished.  
      There is no need for the prepare flush to also start a gopher.
modified:
  storage/falcon/Database.cpp
  storage/falcon/Database.h
  storage/falcon/Dbb.cpp
  storage/falcon/Dbb.h
  storage/falcon/SRLCommit.cpp
  storage/falcon/SRLPrepare.cpp
  storage/falcon/SRLRollback.cpp
  storage/falcon/SRLRollback.h
  storage/falcon/SerialLogRecord.cpp
  storage/falcon/SerialLogRecord.h
  storage/falcon/Transaction.cpp

per-file messages:
  storage/falcon/Database.cpp
    The Transaction knows about the serial log.  There is no
    reason to pass these through Database
  storage/falcon/Database.h
    The Transaction knows about the serial log.  There is no
    reason to pass these through Database
  storage/falcon/Dbb.cpp
    The Transaction knows about the serial log.  There is no
    reason to pass commit, prepare, and rollback through Dbb
  storage/falcon/Dbb.h
    The Transaction knows about the serial log.  There is no
    reason to pass commit, prepare, and rollback through Dbb
  storage/falcon/SRLCommit.cpp
    Do not mark the srlTransaction as committed and wake up 
    a gopher as soon as it is durable.  
    Do that later in Transaction::commit()
  storage/falcon/SRLPrepare.cpp
    No need to wake up a gopher for a prepare.  
    There is nothing for it to do.
  storage/falcon/SRLRollback.cpp
    Just pass in the transaction like in similar append() functions.
    Do not mark the srlTransaction as rolledBack and wake up 
    a gopher as soon as this transaction rollback is durable.  
    Do that later in Transaction::rollback()
  storage/falcon/SRLRollback.h
    Just pass in the transaction like in similar append() functions.
  storage/falcon/SerialLogRecord.cpp
    The Transaction knows about the serial log.  There is no
    reason to pass the wakeup through SerialLogRecord.
  storage/falcon/SerialLogRecord.h
    The Transaction knows about the serial log.  There is no
    reason to pass the wakeup through SerialLogRecord.
  storage/falcon/Transaction.cpp
    The Transaction knows about the serial log.  Let it make calls
    to SerialLogRecord functions directly.
    Separate the serial log flush of commit and rollback records
    from when we allow the gophers to start processing them.
    Once those records are flushed, the recovery will be able to 
    process them, but we do not want the gophers to do that until 
    the commit or rollback is fully finished.
=== modified file 'storage/falcon/Database.cpp'
--- a/storage/falcon/Database.cpp	2008-11-20 17:05:50 +0000
+++ b/storage/falcon/Database.cpp	2008-12-19 18:45:32 +0000
@@ -2251,16 +2251,6 @@ void Database::deleteRepositoryBlob(cons
 	repository->deleteBlob (volume, blobId, transaction);	
 }
 
-void Database::commit(Transaction *transaction)
-{
-	dbb->commit(transaction);
-}
-
-void Database::rollback(Transaction *transaction)
-{
-	dbb->rollback(transaction->transactionId, transaction->hasUpdates);
-}
-
 void Database::renameTable(Table* table, const char* newSchema, const char* newName)
 {
 	newSchema = getSymbol(newSchema);

=== modified file 'storage/falcon/Database.h'
--- a/storage/falcon/Database.h	2008-10-16 01:04:03 +0000
+++ b/storage/falcon/Database.h	2008-12-19 18:45:32 +0000
@@ -116,8 +116,6 @@ public:
 
 	void			shutdownNow();
 	void			dropDatabase();
-	void			rollback (Transaction *transaction);
-	void			commit (Transaction *transaction);
 	void			start();
 	void			deleteRepositoryBlob(const char *schema, const char *repositoryName, int volume,
int64 blobId, Transaction *transaction);
 	void			deleteRepository (Repository *repository);

=== modified file 'storage/falcon/Dbb.cpp'
--- a/storage/falcon/Dbb.cpp	2008-10-30 00:22:54 +0000
+++ b/storage/falcon/Dbb.cpp	2008-12-19 18:45:32 +0000
@@ -1169,34 +1169,6 @@ void Dbb::reportStatistics()
 	priorFlushWrites = flushWrites;
 }
 
-void Dbb::commit(Transaction *transaction)
-{
-	if (transaction->hasUpdates)
-		serialLog->logControl->commit.append(transaction);
-}
-
-void Dbb::prepareTransaction(TransId transId, int xidLength, const UCHAR *xid)
-{
-	serialLog->logControl->prepare.append(transId, xidLength, xid);
-}
-
-void Dbb::rollback(TransId transId, bool updateTransaction)
-{
-	if (updateTransaction)
-		{
-		if (serialLog)
-			serialLog->logControl->rollback.append(transId, updateTransaction);
-		//flush();
-		}
-}
-
-/***
-void Dbb::setRecovering(bool flag)
-{
-	recovering = flag;
-}
-***/
-
 void Dbb::enableSerialLog()
 {
 	Bdb *bdb = fetchPage(HEADER_PAGE, PAGE_header, Exclusive);

=== modified file 'storage/falcon/Dbb.h'
--- a/storage/falcon/Dbb.h	2008-10-30 00:22:54 +0000
+++ b/storage/falcon/Dbb.h	2008-12-19 18:45:32 +0000
@@ -113,10 +113,7 @@ public:
 	void	createSection(int32 sectionId, TransId transId);
 	void	dropDatabase();
 	void	enableSerialLog();
-	void	rollback (TransId transId, bool updateTransaction);
 	void	updateRecord(int32 sectionId, int32 recordId, Stream *stream, TransId transId, bool
earlyWrite);
-	void	prepareTransaction(TransId transId, int xidLength, const UCHAR *xid);
-	void	commit(Transaction *transaction);
 	void	reportStatistics();
 	bool	hasDirtyPages();
 	bool	deleteShadow (DatabaseCopy *shadow);

=== modified file 'storage/falcon/SRLCommit.cpp'
--- a/storage/falcon/SRLCommit.cpp	2007-10-04 20:45:34 +0000
+++ b/storage/falcon/SRLCommit.cpp	2008-12-19 18:45:32 +0000
@@ -46,18 +46,16 @@ void SRLCommit::append(Transaction *tran
 	transaction->addRef();
 	START_RECORD(srlCommit, "SRLCommit::append");
 	putInt(transaction->transactionId);
-	//uint64 commitBlockNumber = log->nextBlockNumber;
 	uint64 commitBlockNumber = log->getWriteBlockNumber();
 	SerialLogTransaction *srlTransaction =
log->getTransaction(transaction->transactionId);
-	
+	srlTransaction->setTransaction(transaction);
+
+	// Flush transactions with changes immediately for durability
+
 	if (transaction->hasUpdates)
 		log->flush(false, commitBlockNumber, &sync);
 	else
 		sync.unlock();
-
-	srlTransaction->setTransaction(transaction);
-	srlTransaction->setState(sltCommitted);
-	wakeup();
 }
 
 void SRLCommit::read()

=== modified file 'storage/falcon/SRLPrepare.cpp'
--- a/storage/falcon/SRLPrepare.cpp	2008-07-24 08:45:03 +0000
+++ b/storage/falcon/SRLPrepare.cpp	2008-12-19 18:45:32 +0000
@@ -47,15 +47,12 @@ void SRLPrepare::append(TransId transId,
 	putInt(transId);
 	putInt(xidLength);
 	putData(xidLength, xid);
-	SerialLogTransaction *transaction = log->getTransaction(transId);
+	SerialLogTransaction *srlTransaction = log->getTransaction(transId);
 
 	log->flush(false, log->nextBlockNumber, &sync);
 
-	if (transaction)
-		transaction->setState(sltPrepared);
-
-	if (transaction)
-		wakeup();
+	if (srlTransaction)
+		srlTransaction->setState(sltPrepared);
 }
 
 void SRLPrepare::read()

=== modified file 'storage/falcon/SRLRollback.cpp'
--- a/storage/falcon/SRLRollback.cpp	2008-07-24 08:45:03 +0000
+++ b/storage/falcon/SRLRollback.cpp	2008-12-19 18:45:32 +0000
@@ -38,23 +38,16 @@ SRLRollback::~SRLRollback()
 
 }
 
-void SRLRollback::append(TransId transId, bool updateTransaction)
+void SRLRollback::append(Transaction *transaction)
 {
 	START_RECORD(srlRollback, "SRLRollback::append");
-	putInt(transId);
+	putInt(transaction->transactionId);
 	uint64 commitBlockNumber = log->nextBlockNumber;
-	SerialLogTransaction *transaction = log->findTransaction(transId);
 
-	if (updateTransaction)
+	if (transaction->hasUpdates)
 		log->flush(false, commitBlockNumber, &sync);
 	else
 		sync.unlock();
-	
-	if (transaction)
-		{
-		transaction->setState(sltRolledBack);
-		wakeup();
-		}
 }
 
 void SRLRollback::read()

=== modified file 'storage/falcon/SRLRollback.h'
--- a/storage/falcon/SRLRollback.h	2007-09-20 15:44:25 +0000
+++ b/storage/falcon/SRLRollback.h	2008-12-19 18:45:32 +0000
@@ -33,7 +33,7 @@ public:
 	virtual void print();
 	virtual void pass1();
 	virtual void read();
-	void append(TransId transId, bool updateTransaction);
+	void append(Transaction *transaction);
 	SRLRollback();
 	virtual ~SRLRollback();
 

=== modified file 'storage/falcon/SerialLogRecord.cpp'
--- a/storage/falcon/SerialLogRecord.cpp	2008-11-20 17:05:50 +0000
+++ b/storage/falcon/SerialLogRecord.cpp	2008-12-19 18:45:32 +0000
@@ -226,11 +226,6 @@ void SerialLogRecord::startRecord()
 	log->startRecord();
 }
 
-void SerialLogRecord::wakeup()
-{
-	log->wakeup();
-}
-
 void SerialLogRecord::putStream(Stream *stream)
 {
 	putInt(stream->totalLength);

=== modified file 'storage/falcon/SerialLogRecord.h'
--- a/storage/falcon/SerialLogRecord.h	2008-11-14 02:30:11 +0000
+++ b/storage/falcon/SerialLogRecord.h	2008-12-19 18:45:32 +0000
@@ -98,7 +98,6 @@ public:
 	int				getInt(const UCHAR** ptr);
 	const UCHAR*	getData(int32 length);
 	void			putStream (Stream *stream);
-	void			wakeup();
 	void			startRecord();
 	void			putData(uint32 length, const UCHAR *data);
 	void			putInt(int32 number);

=== modified file 'storage/falcon/Transaction.cpp'
--- a/storage/falcon/Transaction.cpp	2008-12-16 20:40:38 +0000
+++ b/storage/falcon/Transaction.cpp	2008-12-19 18:45:32 +0000
@@ -37,6 +37,7 @@
 #include "TransactionManager.h"
 #include "SerialLog.h"
 #include "SerialLogControl.h"
+#include "SerialLogTransaction.h"
 #include "InfoTable.h"
 #include "Thread.h"
 #include "Format.h"
@@ -266,6 +267,11 @@ void Transaction::commit()
 
 	database->flushInversion(this);
 
+	// Write the commit message to the serial log for durability.
+	// If a crash happens after this, the recover will commit.
+
+	database->serialLog->logControl->commit.append(this);
+
 	// Transfer transaction from active list to committed list, set committed state
 
 	Sync syncActiveTransactions(&transactionManager->activeTransactions.syncObject,
"Transaction::commit(2)");
@@ -293,8 +299,12 @@ void Transaction::commit()
 	syncActiveTransactions.unlock();
 	
 	syncIsActive.unlock(); // signal waiting transactions
+	
+	// signal a gopher to start processing this transaction
 
-	database->commit(this);
+	SerialLogTransaction *srlTransaction =
database->serialLog->getTransaction(transactionId);
+	srlTransaction->setState(sltCommitted);
+	database->serialLog->wakeup();
 
 	delete [] xid;
 	xid = NULL;
@@ -411,12 +421,13 @@ void Transaction::rollback()
 
 	ASSERT(writePending);
 	writePending = false;
-	
+
 	if (hasUpdates)
+		{
 		database->serialLog->preCommit(this);
-		
-	database->rollback(this);
-	
+		database->serialLog->logControl->rollback.append(this);
+		}
+
 	if (xid)
 		{
 		delete [] xid;
@@ -433,6 +444,16 @@ void Transaction::rollback()
 	syncActiveTransactions.unlock();
 	state = RolledBack;
 	syncIsActive.unlock();
+
+	// Finish the SerialLogTransaction and signal a gopher
+
+	if (hasUpdates)
+		{
+		SerialLogTransaction *srlTransaction =
database->serialLog->getTransaction(transactionId);
+		srlTransaction->setState(sltRolledBack);
+		database->serialLog->wakeup();
+		}
+
 	release();
 }
 
@@ -455,7 +476,10 @@ void Transaction::prepare(int xidLen, co
 		
 	database->pageWriter->waitForWrites(this);
 	state = Limbo;
-	database->dbb->prepareTransaction(transactionId, xidLength, xid);
+
+	// Flush a prepare record to the serial log
+
+	database->serialLog->logControl->prepare.append(transactionId, xidLength, xid);
 
 	Sync sync(&syncDeferredIndexes, "Transaction::prepare");
 	sync.lock(Shared);
@@ -1413,12 +1437,15 @@ void Transaction::getInfo(InfoTable* inf
 		}
 }
 
+// Called by the gopher thread to complete this transaction
+
 void Transaction::fullyCommitted(void)
 {
 	ASSERT(inList);
+	ASSERT(!isActive());
 
 	if (useCount < 2)
-		Log::debug("Transaction::fullyCommitted: funny use count\n");
+		Log::debug("Transaction::fullyCommitted: Unusual use count=%d\n", useCount);
 
 	writeComplete();
 	releaseCommittedTransaction();