| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| * |
| */ |
| package org.apache.directory.server.core.shared.txn; |
| |
| |
| import java.io.ByteArrayInputStream; |
| import java.io.IOException; |
| import java.io.ObjectInputStream; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.concurrent.ConcurrentLinkedQueue; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.concurrent.atomic.AtomicReference; |
| import java.util.concurrent.locks.Condition; |
| import java.util.concurrent.locks.Lock; |
| import java.util.concurrent.locks.ReadWriteLock; |
| import java.util.concurrent.locks.ReentrantLock; |
| import java.util.concurrent.locks.ReentrantReadWriteLock; |
| |
| import org.apache.directory.server.core.api.log.Log; |
| import org.apache.directory.server.core.api.log.LogAnchor; |
| import org.apache.directory.server.core.api.log.LogScanner; |
| import org.apache.directory.server.core.api.log.UserLogRecord; |
| import org.apache.directory.server.core.api.partition.Partition; |
| import org.apache.directory.server.core.api.schema.SchemaPartition; |
| import org.apache.directory.server.core.api.txn.TxnConflictException; |
| import org.apache.directory.server.core.api.txn.TxnHandle; |
| import org.apache.directory.server.core.api.txn.TxnLogManager; |
| import org.apache.directory.server.core.api.txn.logedit.LogEdit; |
| import org.apache.directory.server.core.api.txn.logedit.LogEdit.EditType; |
| import org.apache.directory.server.core.shared.txn.logedit.DataChangeContainer; |
| import org.apache.directory.server.core.shared.txn.logedit.TxnStateChange; |
| import org.apache.directory.server.core.shared.txn.logedit.TxnStateChange.ChangeState; |
| import org.apache.directory.shared.ldap.model.exception.LdapException; |
| import org.apache.directory.shared.ldap.model.name.Dn; |
| |
| |
| /** |
| * |
| * @author <a href="mailto:dev@directory.apache.org">Apache Directory Project</a> |
| */ |
| /** Package protected */ |
| class DefaultTxnManager implements TxnManagerInternal |
| { |
| /** wal log manager */ |
| private TxnLogManager txnLogManager; |
| |
| /** Write ahead log */ |
| private Log wal; |
| |
| /** List of committed txns in commit LSN order */ |
| private ConcurrentLinkedQueue<ReadWriteTxn> committedQueue = new ConcurrentLinkedQueue<ReadWriteTxn>(); |
| |
| /** Verify lock under which txn verification is done */ |
| private Lock verifyLock = new ReentrantLock(); |
| |
| /** Used to assign start and commit version numbers to writeTxns */ |
| private Lock writeTxnsLock = new ReentrantLock(); |
| |
| /** Used to order txns in case of conflicts */ |
| private ReentrantReadWriteLock optimisticLock = new ReentrantReadWriteLock(); |
| |
| /** Latest committed txn on which read only txns can depend */ |
| private AtomicReference<ReadWriteTxn> latestCommittedTxn = new AtomicReference<ReadWriteTxn>(); |
| |
| /** Latest verified write txn */ |
| private AtomicReference<ReadWriteTxn> latestVerifiedTxn = new AtomicReference<ReadWriteTxn>(); |
| |
| /** Latest flushed txn's logical commit time */ |
| private AtomicLong latestFlushedTxnLSN = new AtomicLong( LogAnchor.UNKNOWN_LSN ); |
| |
| /** Default flush interval in ms */ |
| private final static int DEFAULT_FLUSH_INTERVAL = 100; |
| |
| /** Flush interval */ |
| private int flushInterval; |
| |
| /** Flush lock */ |
| private Lock flushLock = new ReentrantLock(); |
| |
| /** Number of flushed txns */ |
| private int numFlushedTxns; |
| |
| /** Number of flushed */ |
| private int numFlushes; |
| |
| /** Take a checkpoint every 1000 flushes ~100 secs */ |
| private final static int DEFAULT_FLUSH_ROUNDS = 1000; |
| |
| /** Flush Condition object */ |
| private Condition flushCondition = flushLock.newCondition(); |
| |
| /** Whether flushing is failed */ |
| private boolean flushFailed; |
| |
| /** partitions to be synced after applying changes */ |
| private HashSet<Partition> flushedToPartitions = new HashSet<Partition>(); |
| |
| /** Backgorund syncing thread */ |
| private LogSyncer syncer; |
| |
| /** Initial committed txn */ |
| private ReadWriteTxn dummyTxn = new ReadWriteTxn(); |
| |
| private AtomicInteger pending = new AtomicInteger(); |
| |
| /** Logical data version number */ |
| private long logicalDataVersion; |
| |
| /** Initial scan point into the logs */ |
| private LogAnchor initialScanPoint; |
| |
| /** Initial set of committed txns */ |
| private HashSet<Long> txnsToRecover = new HashSet<Long>(); |
| |
| /** last flushed log anchor */ |
| private LogAnchor lastFlushedLogAnchor; |
| |
| /** Whether to avoid flushing */ |
| private boolean doNotFlush = false; |
| |
| /** Per thread txn context */ |
| static final ThreadLocal<Transaction> txnVar = |
| new ThreadLocal<Transaction>() |
| { |
| @Override |
| protected Transaction initialValue() |
| { |
| return null; |
| } |
| }; |
| |
| |
| /** |
| * Inits the txn manager. A dummy txn is put to the committed queue for read |
| * only txns to bump the ref count so that we can always keep track of the |
| * minimum existing txn. |
| * |
| * @param txnLogManager |
| * @param idComparator |
| * @param idSerializer |
| */ |
| public void init( TxnLogManagerInternal txnLogManager ) |
| { |
| this.txnLogManager = txnLogManager; |
| wal = txnLogManager.getWAL(); |
| flushInterval = DEFAULT_FLUSH_INTERVAL; |
| |
| committedQueue.clear(); |
| latestFlushedTxnLSN.set( LogAnchor.UNKNOWN_LSN ); |
| txnsToRecover.clear(); |
| logicalDataVersion = 0; |
| lastFlushedLogAnchor = new LogAnchor(); |
| |
| initialScanPoint = wal.getCheckPoint(); |
| //System.out.println("checkpoint " + initialScanPoint); |
| |
| lastFlushedLogAnchor.resetLogAnchor( initialScanPoint ); |
| |
| dummyTxn.commitTxn( initialScanPoint.getLogLSN() ); |
| latestCommittedTxn.set( dummyTxn ); |
| latestVerifiedTxn.set( dummyTxn ); |
| committedQueue.offer( dummyTxn ); |
| |
| getTxnsToReover(); |
| |
| if ( syncer == null ) |
| { |
| syncer = new LogSyncer(); |
| syncer.setDaemon( true ); |
| syncer.start(); |
| } |
| } |
| |
| |
| public void shutdown() |
| { |
| //System.out.println("in shutdown"); |
| syncer.interrupt(); |
| |
| try |
| { |
| syncer.join(); |
| } |
| catch ( InterruptedException e ) |
| { |
| //Ignore |
| } |
| |
| // Do a best effort last flush |
| flushLock.lock(); |
| |
| try |
| { |
| if ( !doNotFlush ) |
| { |
| flushTxns(); |
| } |
| |
| advanceCheckPoint( lastFlushedLogAnchor ); |
| } |
| catch ( Exception e ) |
| { |
| e.printStackTrace(); |
| } |
| finally |
| { |
| flushLock.unlock(); |
| } |
| |
| syncer = null; |
| } |
| |
| |
| /** |
| * {@inheritDoc} |
| */ |
| public long getLogicalDataVersion() |
| { |
| return logicalDataVersion; |
| } |
| |
| |
| /** |
| * {@inheritDoc} |
| */ |
| public void bumpLogicalDataVersion() |
| { |
| logicalDataVersion++; |
| } |
| |
| |
| /** |
| * {@inheritDoc} |
| */ |
| public TxnHandle retryTransaction() throws Exception |
| { |
| Transaction curTxn = getCurTxn(); |
| |
| // Should have a rw txn |
| if ( ( curTxn == null ) || |
| !( curTxn instanceof ReadWriteTxn ) ) |
| { |
| // Cannot start a TXN when a RW txn is ongoing |
| throw new IllegalStateException( "Unexpected txn state when trying txn: " + |
| curTxn ); |
| } |
| |
| // abort current txn and start a new read write txn |
| |
| abortReadWriteTxn( ( ReadWriteTxn ) curTxn ); |
| curTxn.abortTxn(); |
| return beginReadWriteTxn( true ); |
| } |
| |
| |
| /** |
| * {@inheritDoc} |
| */ |
| public Transaction beginTransaction( boolean readOnly ) throws Exception |
| { |
| Transaction curTxn = getCurTxn(); |
| |
| // Deal with an existing TXN |
| if ( curTxn != null ) |
| { |
| // Cannot start a TXN when a RW txn is ongoing |
| throw new IllegalStateException( "Cannot begin a txn when txn is already running: " + |
| curTxn ); |
| } |
| |
| // Normal situation : starts a brand new txn |
| if ( readOnly ) |
| { |
| return beginReadOnlyTxn(); |
| } |
| else |
| { |
| return beginReadWriteTxn( false ); |
| } |
| } |
| |
| |
| /** |
| * {@inheritDoc} |
| */ |
| public void commitTransaction() throws Exception, TxnConflictException |
| { |
| Transaction txn = getCurTxn(); |
| |
| if ( txn == null ) |
| { |
| throw new IllegalStateException( " trying to commit non existent txn " ); |
| } |
| |
| try |
| { |
| if ( flushFailed ) |
| { |
| throw new IOException( "Flushing of txns failed" ); |
| } |
| |
| prepareForEndingTxn( txn ); |
| |
| if ( txn instanceof ReadOnlyTxn ) |
| { |
| txn.commitTxn( txn.getStartTime() ); |
| } |
| else |
| { |
| commitReadWriteTxn( ( ReadWriteTxn ) txn ); |
| } |
| } |
| finally |
| { |
| // Release optimistic lock if it is held |
| releaseOptimisticLock(); |
| } |
| |
| setCurTxn( null ); |
| |
| //System.out.println( "TRAN: Committed " + txn ); |
| } |
| |
| |
| /** |
| * {@inheritDoc} |
| */ |
| public void abortTransaction() throws Exception |
| { |
| Transaction txn = getCurTxn(); |
| |
| if ( txn == null ) |
| { |
| throw new IllegalStateException( "Trying to abort while there is not txn " ); |
| } |
| |
| try |
| { |
| prepareForEndingTxn( txn ); |
| |
| if ( txn instanceof ReadWriteTxn ) |
| { |
| abortReadWriteTxn( ( ReadWriteTxn ) txn ); |
| } |
| |
| txn.abortTxn(); |
| } |
| finally |
| { |
| // Release optimistic lock if it is held |
| releaseOptimisticLock(); |
| setCurTxn( null ); |
| } |
| |
| //System.out.println( "TRAN: Aborted " + txn ); |
| |
| } |
| |
| |
| /** |
| * {@inheritDoc} |
| */ |
| public TxnHandle suspendCurTxn() |
| { |
| Transaction curTxn = txnVar.get(); |
| |
| setCurTxn( null ); |
| |
| return curTxn; |
| } |
| |
| |
| /** |
| * {@inheritDoc} |
| */ |
| public void resumeTxn( TxnHandle txnHandle ) |
| { |
| if ( txnHandle == null ) |
| { |
| throw new IllegalArgumentException( "Cannot accept a null handle when resuming a txn " ); |
| } |
| |
| Transaction curTxn = txnVar.get(); |
| |
| if ( curTxn != null ) |
| { |
| throw new IllegalStateException( " Trying to resume txn" + txnHandle |
| + " while there is already a txn running:" + curTxn ); |
| } |
| |
| setCurTxn( ( Transaction ) txnHandle ); |
| } |
| |
| |
| /** |
| * {@inheritDoc} |
| */ |
| public Transaction getCurTxn() |
| { |
| return ( Transaction ) txnVar.get(); |
| } |
| |
| |
| /** |
| * {@inheritDoc} |
| */ |
| public Transaction setCurTxn( TxnHandle transaction ) |
| { |
| Transaction previousTransaction = ( Transaction ) txnVar.get(); |
| |
| txnVar.set( ( Transaction ) transaction ); |
| |
| return previousTransaction; |
| } |
| |
| |
| /** |
| * {@inheritDoc} |
| */ |
| public void applyPendingTxns() |
| { |
| flushLock.lock(); |
| |
| try |
| { |
| flushTxns(); |
| } |
| catch ( Exception e ) |
| { |
| e.printStackTrace(); |
| // Ignore |
| } |
| finally |
| { |
| flushLock.unlock(); |
| } |
| |
| } |
| |
| |
| /** |
| * {@inheritDoc} |
| */ |
| public void startLogicalDataChange() throws LdapException |
| { |
| Transaction curTxn = getCurTxn(); |
| |
| if ( curTxn == null ) |
| { |
| return; |
| } |
| // Should have a rw txn |
| if ( !( curTxn instanceof ReadWriteTxn ) ) |
| { |
| // Cannot start a TXN when a RW txn is ongoing |
| throw new IllegalStateException( "Unexpected txn state when starting logical data change txn: " + |
| curTxn ); |
| } |
| |
| // If txn is already exclusive then it can start logical data change immediately |
| |
| if ( !curTxn.isOptimisticLockHeld() ) |
| { |
| throw new IllegalStateException( "Unexpected txn state when starting logical data change txn: " + |
| " txn is not holding optimistic lock:" + |
| curTxn ); |
| } |
| |
| // If lock is already held exclusively by the txn, then return |
| if ( optimisticLock.isWriteLockedByCurrentThread() ) |
| { |
| return; |
| } |
| |
| long txnLogicalDataVersion = curTxn.getLogicalDataVersion(); |
| |
| // Get operations lock in exclusive mode |
| optimisticLock.readLock().unlock(); |
| optimisticLock.writeLock().lock(); |
| |
| // If somebody raced and changed logical data, then bail out |
| if ( getLogicalDataVersion() != txnLogicalDataVersion ) |
| { |
| |
| TxnConflictException e = new TxnConflictException(); |
| throw new LdapException( e ); |
| } |
| |
| // Finally bump of logical data version number |
| bumpLogicalDataVersion(); |
| } |
| |
| |
| /** |
| * {@inheritDoc} |
| */ |
| public void endLogicalDataRead() |
| { |
| // Should only be called for read only txns |
| Transaction curTxn = getCurTxn(); |
| |
| if ( curTxn == null || !( curTxn instanceof ReadOnlyTxn ) ) |
| { |
| throw new IllegalStateException( "Unexpected txn state when ending logical data read:" + curTxn ); |
| } |
| |
| if ( !curTxn.isOptimisticLockHeld() ) |
| { |
| throw new IllegalStateException( |
| "Unexpected txn state when ending logical data read, optimistic lock not held:" + |
| curTxn ); |
| } |
| |
| releaseOptimisticLock(); |
| } |
| |
| |
| /** |
| * {@inheritDoc} |
| */ |
| public boolean prepareForLogicalDataReinit() |
| { |
| Transaction curTxn = getCurTxn(); |
| |
| if ( curTxn == null || !( curTxn instanceof ReadWriteTxn ) ) |
| { |
| throw new IllegalStateException( "Unexpected txn state when preparing for logical data reinit:" + curTxn ); |
| } |
| |
| if ( optimisticLock.isWriteLockedByCurrentThread() ) |
| { |
| return true; |
| } |
| else |
| { |
| return false; |
| } |
| } |
| |
| |
| /** |
| * If the thread holds optimistic lock, release it |
| */ |
| private void releaseOptimisticLock() |
| { |
| Transaction curTxn = getCurTxn(); |
| |
| if ( curTxn.isOptimisticLockHeld() ) |
| { |
| if ( optimisticLock.isWriteLockedByCurrentThread() ) |
| { |
| optimisticLock.writeLock().unlock(); |
| } |
| else |
| { |
| optimisticLock.readLock().unlock(); |
| } |
| |
| curTxn.clearOptimisticLockHeld(); |
| } |
| } |
| |
| |
| /** |
| * Begins a read only txn. A read only txn does not put any log edits |
| * to the txn log.Its start time is the latest committed txn's commit time. |
| */ |
| private Transaction beginReadOnlyTxn() |
| { |
| ReadOnlyTxn txn = new ReadOnlyTxn(); |
| ReadWriteTxn lastTxnToCheck = null; |
| |
| optimisticLock.readLock().lock(); |
| txn.setOptimisticLockHeld(); |
| txn.setLogicalDataVersion( logicalDataVersion ); |
| |
| lastTxnToCheck = latestCommittedTxn.get(); |
| lastTxnToCheck.getRefCount().getAndIncrement(); |
| |
| // Determine start time |
| long startTime; |
| |
| startTime = lastTxnToCheck.getCommitTime(); |
| txn.startTxn( startTime, logicalDataVersion ); |
| |
| buildCheckList( txn, lastTxnToCheck ); |
| |
| setCurTxn( txn ); |
| |
| //System.out.println( "TRAN: Started " + txn ); |
| |
| return txn; |
| } |
| |
| |
| /** |
| * Begins a read write txn. A start txn marker is inserted |
| * into the txn log and the lsn of that log record is the |
| * start time. |
| */ |
| private Transaction beginReadWriteTxn( boolean retry ) throws Exception |
| { |
| |
| ReadWriteTxn txn = new ReadWriteTxn(); |
| UserLogRecord logRecord = txn.getUserLogRecord(); |
| |
| LogEdit logEdit = new TxnStateChange( LogAnchor.UNKNOWN_LSN, |
| TxnStateChange.ChangeState.TXN_BEGIN ); |
| |
| logEdit.injectData( logRecord, UserLogRecord.LogEditType.TXN ); |
| |
| if ( retry == false ) |
| { |
| optimisticLock.readLock().lock(); |
| } |
| else |
| { |
| optimisticLock.writeLock().lock(); |
| } |
| |
| txn.setOptimisticLockHeld(); |
| txn.setLogicalDataVersion( logicalDataVersion ); |
| |
| /* |
| * Get the start time and last txn to depend on |
| * when mergin data under te writeTxnLock. |
| */ |
| ReadWriteTxn lastTxnToCheck = null; |
| |
| writeTxnsLock.lock(); |
| |
| try |
| { |
| txnLogManager.log( logRecord, false ); |
| txn.startTxn( logRecord.getLogAnchor().getLogLSN(), logicalDataVersion ); |
| txn.setTxnId( logRecord.getLogAnchor().getLogLSN() ); |
| |
| do |
| { |
| if ( lastTxnToCheck != null ) |
| { |
| lastTxnToCheck.getRefCount().decrementAndGet(); |
| } |
| |
| lastTxnToCheck = latestVerifiedTxn.get(); |
| |
| lastTxnToCheck.getRefCount().incrementAndGet(); |
| } |
| while ( lastTxnToCheck != latestVerifiedTxn.get() ); |
| } |
| catch ( Exception e ) |
| { |
| // Release optimistic lock if held |
| setCurTxn( txn ); |
| releaseOptimisticLock(); |
| |
| } |
| finally |
| { |
| writeTxnsLock.unlock(); |
| } |
| |
| // Finally build the check list |
| buildCheckList( txn, lastTxnToCheck ); |
| |
| setCurTxn( txn ); |
| |
| //System.out.println( "TRAN: Started " + txn ); |
| |
| return txn; |
| } |
| |
| |
| /** |
| * Builds the list of txns which the given txn should check while mergin what it read from |
| * the partitions with the changes in the txn log. These are the txns that committed before |
| * the start of the give txn and for which the changes are not flushed to the partitions yet. |
| * Note that, for some of these txns, flush to partitions could go on in parallel. |
| * |
| * @param txn txn for which we will build the check list |
| * @param lastTxnToCheck latest txn to check |
| */ |
| private void buildCheckList( Transaction txn, ReadWriteTxn lastTxnToCheck ) |
| { |
| long lastLSN = lastTxnToCheck.getCommitTime(); |
| |
| List<ReadWriteTxn> toCheckList = txn.getTxnsToCheck(); |
| long flushedLSN = latestFlushedTxnLSN.get(); |
| |
| // Get all the txns that has been committed before the new txn |
| for ( ReadWriteTxn toAdd : committedQueue ) |
| { |
| long commitTime = toAdd.getCommitTime(); |
| |
| if ( commitTime > lastLSN ) |
| { |
| break; |
| } |
| |
| /* |
| * Get latest flushed lsn and eliminate already flushed txn from the check list. |
| */ |
| if ( ( commitTime <= flushedLSN ) && ( toAdd != lastTxnToCheck ) ) |
| { |
| continue; |
| } |
| |
| toCheckList.add( toAdd ); |
| } |
| |
| // A read write txn, always has to check its changes |
| if ( txn instanceof ReadWriteTxn ) |
| { |
| txn.getTxnsToCheck().add( ( ReadWriteTxn ) txn ); |
| } |
| } |
| |
| |
| /** |
| * Called before ending a txn. Txn for which this txn bumped |
| * up the ref count is gotten and its ref count is decreased. |
| * |
| * @param txn txn which is about to commit or abort |
| */ |
| private void prepareForEndingTxn( Transaction txn ) |
| { |
| List<ReadWriteTxn> toCheck = txn.getTxnsToCheck(); |
| |
| // A read write txn, always has to check its changes |
| if ( txn instanceof ReadWriteTxn ) |
| { |
| |
| if ( toCheck.size() <= 0 ) |
| { |
| throw new IllegalStateException( |
| " prepareForEndingTxn: a read write txn should at least depend on itself:" + txn ); |
| } |
| |
| txn.getTxnsToCheck().remove( ( ReadWriteTxn ) txn ); |
| } |
| |
| if ( toCheck.size() > 0 ) |
| { |
| ReadWriteTxn lastTxnToCheck = toCheck.get( toCheck.size() - 1 ); |
| |
| if ( lastTxnToCheck.commitTime > txn.getStartTime() ) |
| { |
| throw new IllegalStateException( " prepareForEndingTxn: txn has unpexptected start time " + |
| txn + " expected: " + lastTxnToCheck ); |
| } |
| |
| if ( lastTxnToCheck.getRefCount().get() <= 0 ) |
| { |
| throw new IllegalStateException( " prepareForEndingTxn: lastTxnToCheck has unexpected ref cnt " + |
| txn + " expected: " + lastTxnToCheck ); |
| } |
| |
| lastTxnToCheck.getRefCount().decrementAndGet(); |
| |
| // if ( txn instanceof ReadOnlyTxn ) |
| // { |
| // System.out.println(" txn end " + txn.getStartTime() + " " + |
| // lastTxnToCheck.getRefCount() |
| // + " pending " + pending.decrementAndGet() ); |
| // } |
| } |
| } |
| |
| |
| /** |
| * Tries to commit the given read write txn. Before a read write txn can commit, it is |
| * verified against the txns that committed after this txn started. If a conflicting change is |
| * found, a conflict exception is thrown. |
| * |
| * If a txn can commit, a commit record is inserted into the txn log. The lsn of the commit record |
| * is the commit time of the txn. |
| * |
| * Note that, a txn is not committed until its commit record is synced to the underlying media. Say we haveread write txns rw1 and |
| * rw2 and that rw1 and rw2 is verified and their commit record are in the log but not synced to underlying media yet. A new read |
| * write txn rw3 and a read only txn r1 comes along. Since rw1 and rw2 wont be acked until they commit, r1 should not depend on rw1 and rw2 and can have a view |
| * as of a commit time before rw1 and rw2's commit time. If r1 depended rw1 or rw2 and we crashed before sycning rw1 and rw2's to the underlying media, |
| * r1 would have depended on a change that actually doesnt exist in the database. However, rw3 either has to depend on rw1 and rw2) or has to verify |
| * its changeset against rw1 and rw2 when it tries to commit. Whether the first thing(depending on rw1, rw2 and merging its changeset) or the second |
| * thing ( verifiying its change set against rw1 and rw2) is determined by the order of the lsns of the commit record of rw1 and rw2 and start record of rw3. |
| * Lets say we have this order in the txn log: |
| * commit record rw1, start record rw3, commit record rw2 |
| * then rw3 will merge its changes with that of rw1 and will verify its changes against rw2. When rw3 is merging its changeset with that of rw1, rw1 might not have |
| * committed yet as its commit record might not have made it to the underlying media but this is OK as rw3 cannot commit before rw1 because of the log. |
| * |
| * @param txn txn to commit. |
| * @throws Exception |
| * @throws TxnConflictException |
| */ |
| private void commitReadWriteTxn( ReadWriteTxn txn ) throws Exception, TxnConflictException |
| { |
| UserLogRecord logRecord = txn.getUserLogRecord(); |
| |
| LogEdit logEdit = new TxnStateChange( txn.getStartTime(), |
| TxnStateChange.ChangeState.TXN_COMMIT ); |
| |
| logEdit.injectData( logRecord, UserLogRecord.LogEditType.TXN ); |
| |
| verifyLock.lock(); |
| |
| //Verify txn and throw conflict exception if necessary |
| long startTime = txn.getStartTime(); |
| |
| for ( ReadWriteTxn toCheckTxn : committedQueue ) |
| { |
| // Check txns that committed after we started |
| if ( toCheckTxn.getCommitTime() < startTime ) |
| { |
| continue; |
| } |
| |
| if ( txn.hasConflict( toCheckTxn ) ) |
| { |
| verifyLock.unlock(); |
| throw new TxnConflictException(); |
| } |
| } |
| |
| writeTxnsLock.lock(); |
| |
| try |
| { |
| // TODO sync of log can be done outside the locks. |
| txnLogManager.log( logRecord, true ); |
| txn.commitTxn( logRecord.getLogAnchor().getLogLSN() ); |
| |
| latestVerifiedTxn.set( txn ); |
| committedQueue.offer( txn ); |
| |
| // TODO when sync is done outside the locks, advance latest commit outside the locks |
| latestCommittedTxn.set( txn ); |
| } |
| finally |
| { |
| writeTxnsLock.unlock(); |
| verifyLock.unlock(); |
| } |
| } |
| |
| |
| /** |
| * Aborts a read write txn. An abort record is inserted into the txn log. |
| * |
| * @param txn txn to abort |
| * @throws IOException |
| */ |
| private void abortReadWriteTxn( ReadWriteTxn txn ) throws Exception |
| { |
| UserLogRecord logRecord = txn.getUserLogRecord(); |
| |
| LogEdit logEdit = new TxnStateChange( txn.getStartTime(), |
| TxnStateChange.ChangeState.TXN_ABORT ); |
| |
| logEdit.injectData( logRecord, UserLogRecord.LogEditType.TXN ); |
| |
| txnLogManager.log( logRecord, false ); |
| } |
| |
| |
| /** |
| * Flush the changes of the txns in the committed queue. A txn is flushed |
| * only if flushing it will not cause a pending txn to see changes beyond its |
| * start time. |
| * throws Exception thrown if anything goes wrong during flush. |
| * |
| * @param shutdown is TxnManager is shutting down |
| * |
| */ |
| private void flushTxns() throws Exception |
| { |
| UserLogRecord lastLogRecord = null; |
| |
| // If flushing failed already, dont do anything anymore |
| if ( flushFailed ) |
| { |
| return; |
| } |
| |
| /* |
| * First get the latest committed txn ref and then the iterator. |
| * Order is important. |
| */ |
| ReadWriteTxn latestCommitted = latestCommittedTxn.get(); |
| long latestFlushedLsn = latestFlushedTxnLSN.get(); |
| flushedToPartitions.clear(); |
| |
| Iterator<ReadWriteTxn> it = committedQueue.iterator(); |
| ReadWriteTxn txnToFlush = null; |
| |
| while ( it.hasNext() ) |
| { |
| txnToFlush = it.next(); |
| |
| if ( txnToFlush.getCommitTime() > latestFlushedLsn ) |
| { |
| // Apply changes |
| txnToFlush.flushLogEdits( flushedToPartitions ); |
| |
| latestFlushedTxnLSN.set( txnToFlush.getCommitTime() ); |
| |
| lastLogRecord = txnToFlush.getUserLogRecord(); |
| } |
| |
| if ( txnToFlush == latestCommitted ) |
| { |
| // leave latest committed txn in queue and dont go beyond it. |
| break; |
| } |
| |
| numFlushedTxns++; |
| |
| // if ( numFlushes % 100 == 0 ) |
| // { |
| // System.out.println( "lastFlushed lsn: " + latestFlushedTxnLSN + " " + committedQueue.size() ); |
| // |
| // System.out.println( " last commit txn: " + latestCommitted.getCommitTime() ); |
| // System.out.println( "txnToFlush: " + txnToFlush.getRefCount() + " " + txnToFlush.getCommitTime() ); |
| // } |
| |
| /* |
| * If the latest flushed txn has ref count > 0, then |
| * following txns wont be flushed yet. |
| */ |
| |
| if ( txnToFlush.getRefCount().get() > 0 ) |
| { |
| // System.out.println( "breaking out: " + txnToFlush.getCommitTime() + " " + committedQueue.size() ); |
| break; |
| } |
| |
| // Remove from the queue |
| it.remove(); |
| } |
| |
| // Sync each flushed to partition |
| Iterator<Partition> partitionIt = flushedToPartitions.iterator(); |
| |
| while ( partitionIt.hasNext() ) |
| { |
| partitionIt.next().sync(); |
| } |
| |
| numFlushes++; |
| |
| if ( lastLogRecord != null ) |
| { |
| lastFlushedLogAnchor.resetLogAnchor( lastLogRecord.getLogAnchor() ); |
| } |
| |
| if ( numFlushes % DEFAULT_FLUSH_ROUNDS == 0 ) |
| { |
| advanceCheckPoint( lastFlushedLogAnchor ); |
| } |
| |
| } |
| |
| |
| private void advanceCheckPoint( LogAnchor checkPoint ) |
| { |
| wal.advanceCheckPoint( checkPoint ); |
| } |
| |
| |
| private void getTxnsToReover() |
| { |
| LogScanner logScanner = wal.beginScan( initialScanPoint ); |
| UserLogRecord logRecord = new UserLogRecord(); |
| byte userRecord[]; |
| |
| //System.out.println(" Get txns to recover " + initialScanPoint.getLogLSN() ); |
| |
| try |
| { |
| while ( logScanner.getNextRecord( logRecord ) ) |
| { |
| userRecord = logRecord.getDataBuffer(); |
| ObjectInputStream in = buildStream( userRecord ); |
| |
| EditType editType = EditType.values()[in.read()]; |
| |
| if ( editType == EditType.TXN_MARKER ) |
| { |
| TxnStateChange stateChange = new TxnStateChange(); |
| stateChange.readExternal( in ); |
| |
| if ( stateChange.getTxnState() == ChangeState.TXN_COMMIT ) |
| { |
| //System.out.println("Adding txn " + stateChange.getTxnID() + " to the tobe recovered txns"); |
| txnsToRecover.add( new Long( stateChange.getTxnID() ) ); |
| } |
| } |
| } |
| } |
| catch ( Exception e ) |
| { |
| e.printStackTrace(); |
| // Ignore |
| } |
| } |
| |
| |
| // Walk over the txn log records from the latest checkpoint and apply the |
| // log records to the partition |
| public void recoverPartition( Partition partition ) |
| { |
| Dn partitionSuffix = partition.getSuffixDn(); |
| |
| //System.out.println("Recover partition " + partitionSuffix); |
| |
| LogScanner logScanner = wal.beginScan( initialScanPoint ); |
| UserLogRecord logRecord = new UserLogRecord(); |
| byte userRecord[]; |
| |
| boolean recoveredChanges = false; |
| |
| try |
| { |
| while ( logScanner.getNextRecord( logRecord ) ) |
| { |
| userRecord = logRecord.getDataBuffer(); |
| ObjectInputStream in = buildStream( userRecord ); |
| |
| EditType editType = EditType.values()[in.read()]; |
| |
| if ( editType == EditType.DATA_CHANGE ) |
| { |
| DataChangeContainer dataChangeContainer = new DataChangeContainer(); |
| dataChangeContainer.readExternal( in ); |
| |
| //System.out.println("Data change container for " + dataChangeContainer.getPartitionDn() + |
| // " txn id " + dataChangeContainer.getTxnID() ); |
| |
| // If this change is for the partition we are tyring to recover |
| // and belongs to a txn that committed, then |
| Long txnID = new Long( dataChangeContainer.getTxnID() ); |
| |
| if ( txnsToRecover.contains( txnID ) ) |
| { |
| if ( dataChangeContainer.getPartitionDn().equals( partitionSuffix ) ) |
| { |
| //System.out.println("Apply change to partition " + partitionSuffix); |
| dataChangeContainer.setPartition( partition ); |
| dataChangeContainer.apply( true ); |
| recoveredChanges = true; |
| } |
| } |
| } |
| } |
| |
| if ( recoveredChanges && partition instanceof SchemaPartition ) |
| { |
| ( ( SchemaPartition ) partition ).getSchemaManager().reloadAllEnabled(); |
| } |
| } |
| catch ( Exception e ) |
| { |
| e.printStackTrace(); |
| // Ignore for now |
| } |
| } |
| |
| |
| public void setDoNotFlush() |
| { |
| doNotFlush = true; |
| } |
| |
| |
| public void unsetDoNotFlush() |
| { |
| doNotFlush = false; |
| } |
| |
| |
| private ObjectInputStream buildStream( byte[] buffer ) throws IOException |
| { |
| ObjectInputStream oIn = null; |
| ByteArrayInputStream in = new ByteArrayInputStream( buffer ); |
| |
| try |
| { |
| oIn = new ObjectInputStream( in ); |
| |
| return oIn; |
| } |
| catch ( IOException ioe ) |
| { |
| throw ioe; |
| } |
| } |
| |
| class LogSyncer extends Thread |
| { |
| @Override |
| public void run() |
| { |
| flushLock.lock(); |
| |
| try |
| { |
| while ( true ) |
| { |
| flushCondition.await( flushInterval, TimeUnit.MILLISECONDS ); |
| |
| if ( !doNotFlush ) |
| { |
| flushTxns(); |
| } |
| } |
| } |
| catch ( InterruptedException e ) |
| { |
| // Bail out |
| } |
| catch ( Exception e ) |
| { |
| e.printStackTrace(); |
| flushFailed = true; |
| } |
| finally |
| { |
| flushLock.unlock(); |
| } |
| } |
| } |
| } |
