transactionmanager.java

用Java写的面相对象的数据库管理系统

        } else {
            abortTransaction( ta, command );
        } 
        
        deleteTransaction();
    } 
    
    
    /**
     * Perform the specified command on behalf of the specified
     * thread/transaction. If the transaction has performed only one command
     * until now, this handles deadlocks by re-performing the command again
     * until it throws an exception or completes sucessfully. Otherwise an
     * exception is thrown
     * 
     * 
     * @return True if the command did not throw an exception
     * @throws Exception Any exception always signals an internal error.
     */
    protected boolean performCommand( Transaction ta, DbCommand command ) throws Exception {
        if (env.logWriter.hasTarget( LogWriter.DEBUG3 )) {
            env.logWriter.newEntry( this, "performCommand(): " + ta.toString() + ", " + command.toString(),
                    LogWriter.DEBUG3 );
        } 
        
        boolean result = true;
        
        try {
            // result goes in command.result
            result = ta.performCommand( command );
        
        // other than TransactionError exceptions are catched by
        // ta.performCommand()
        } catch (TransactionError e) {
            if (e.code() == TransactionError.DEADLOCK) {
                
                if (ta.commandCount > 1) {
                    env.logWriter.newEntry( this, ta.toString() + " deadlocked; throwing exception...",
                            LogWriter.WARN );
                    command.result = new DeadlockExc( "" );
                    result = false;
                } else {
                    Random rand = new Random();
                    boolean deadlocked = true;
                    while (deadlocked) {
                        try {
                            env.logWriter.newEntry( this, ta.toString() + " aborting... (DEADLOCK)", LogWriter.WARN );
                            abortTransaction( ta, command );
                            
                            long millis = (long)(rand.nextDouble() * 1000.0);
                            env.logWriter.newEntry( this, ta.toString() + " sleeping " + millis + " milliseconds...",
                                    LogWriter.WARN );
                            ta.sleep( millis );
                            
                            env.logWriter.newEntry( this, ta.toString() + " re-run...", LogWriter.WARN );
                            ta.reset();
                            result = ta.performCommand( command );
                            
                            deadlocked = false;
                        } catch (TransactionError ee) {
                            if (ee.code() == TransactionError.DEADLOCK) {
                            // still deadlocked
                            } else {
                                throw ee;
                            } 
                        } 
                    } 
                } 
            } else {
                // other than DEADLOCK TransactionErrors are internal errors
                throw e;
            } 
        } 
        return result;
    } 
    
    
    /**
     * Prpepare the specified transaction. Return true on success and false
     * if something failed. In this case the transaction is rolled back. This
     * method throws an exception only if an internal server error occured.
     */
    protected boolean prepareTransaction( Transaction ta, DbCommand command ) throws Exception {
        if (env.logWriter.hasTarget( LogWriter.DEBUG3 )) {
            env.logWriter.newEntry( this, "prepareTransaction()", LogWriter.DEBUG3 );
        } 
        
        // if the commit was requested by the client, the following conditions
        // are not internal error and so they must not throw an exception
        if (ta == null) {
            env.logWriter.newEntry( this, "prepareTransaction(): Thread is not joined to a transaction.",
                    LogWriter.WARN );
            command.result = new TransactionExc( "Thread is not joined to a transaction.", TransactionExc.STATE );
            return false;
        } else if (ta.status != Transaction.STATUS_STARTED) {
            env.logWriter.newEntry( this,
                    "prepareTransaction():" + ta.toString() + ": Transaction has inproper status.", LogWriter.WARN );
            command.result = new TransactionExc( "Transaction has inproper status.", TransactionExc.STATE );
            return false;
        } else if (ta.rollbackOnly) {
            env.logWriter.newEntry( this, "prepareTransaction():" + ta.toString() + ": rollback only.",
                    LogWriter.WARN );
            abortTransaction( ta, command );
            command.result = new TransactionExc( "Transaction is in rollback only status.", TransactionExc.ROLLBACK );
            return false;
        } else {
            try {
                // beginExclusion();
                ta.prepareCommit();
                return true;
            } catch (Throwable e) {
                env.logWriter.newEntry( this, "Prepare transaction failed: " + ta.toString() + "; aborting...", e,
                        LogWriter.WARN );
                abortTransaction( ta, command );
                
                command.result = new TransactionExc( e.toString(), TransactionExc.ROLLBACK );
                return false;
            } finally {
            // endExclusion();
            } 
        } 
    } 
    
    
    protected void commitTransaction( Transaction ta, DbCommand command ) throws Exception {
        if (env.logWriter.hasTarget( LogWriter.DEBUG3 )) {
            env.logWriter.newEntry( this, "commitTransaction()", LogWriter.DEBUG3 );
        } 
        
        // if the commit was requested by the client, the following conditions
        // are not internal error and so they must not throw an exception
        if (ta == null) {
            env.logWriter.newEntry( this, "commitTransaction(): Thread is not joined to a transaction.",
                    LogWriter.WARN );
            command.result = new TransactionExc( "Thread is not joined to a transaction.", TransactionExc.STATE );
        } else if (ta.status != Transaction.STATUS_PREPARED) {
            env.logWriter.newEntry( this,
                    "commitTransaction(): " + ta.toString() + ": Transaction has inproper status.", LogWriter.WARN );
            command.result = new TransactionExc( "Transaction has inproper status.", TransactionExc.STATE );
        } else {
            try {
                beginExclusion();
                ta.commit();
            } catch (Throwable e) {
                env.logWriter.newEntry( this, "Commit transaction failed: " + ta.toString(), e, LogWriter.WARN );
                command.result = e;
                if (e instanceof Error) {
                    throw (Error)e;
                } else {
                    throw (Exception)e;
                } 
            } finally {
                endExclusion();
                notifyWaitingTransactions();
            } 
        } 
    } 
    
    
    protected void abortTransaction( Transaction ta, DbCommand command ) throws Exception {
        if (env.logWriter.hasTarget( LogWriter.DEBUG3 )) {
            env.logWriter.newEntry( this, "abortTransaction()", LogWriter.DEBUG3 );
        } 
        
        // if the commit was requested by the client, the following conditions
        // are not internal error and so they must not throw an exception
        if (ta == null) {
            env.logWriter.newEntry( this, "abortTransaction(): Thread is not joined to a transaction.",
                    LogWriter.WARN );
            command.result = new TransactionExc( "Thread is not joined to a transaction.", TransactionExc.STATE );
        } else if (ta.status >= Transaction.STATUS_COMMITING) {
            env.logWriter.newEntry( this, "abortTransaction(): " + ta.toString() + ": Transaction has inproper status.",
                    LogWriter.WARN );
            command.result = new TransactionExc( "Transaction has inproper status.", TransactionExc.STATE );
        } else {
            try {
                beginExclusion();
                ta.abort( command );
            } catch (Throwable e) {
                env.logWriter.newEntry( this, "Aborting transaction failed: " + ta.toString(), e, LogWriter.WARN );
                command.result = e;
                if (e instanceof Error) {
                    throw (Error)e;
                } else {
                    throw (Exception)e;
                } 
            } finally {
                endExclusion();
                notifyWaitingTransactions();
            } 
        } 
    } 
    
    
    /**
     * Blocks execution until there is no thread scheduled for exclusive
     * execution or the exclusive thread is the current thread.
     */
    public void checkExclusion() {
        // evaluate the current thread only if there actually is an
        // exclusiveThread
        if (exclusiveThread != null && exclusiveThread != Thread.currentThread()) {
            synchronized (this) {
                while (exclusiveThread != null && exclusiveThread != Thread.currentThread()) {
                    try {
                        env.logWriter.newEntry( this, "checkExclusion(): waiting... (" + currentTA() + ")",
                                LogWriter.DEBUG );
                        wait();
                        env.logWriter.newEntry( this, "checkExclusion(): notified... (" + currentTA() + ")",
                                LogWriter.DEBUG );
                    } catch (InterruptedException e) {
                    } 
                } 
            } 
        } 
    } 
    
    
    protected synchronized void beginExclusion() {
        checkExclusion();
        
        Thread currentThread = Thread.currentThread();
        if (exclusiveThread != null && exclusiveThread != currentThread) {
            throw new RuntimeException( "Another thread already runs exclusively." );
        } 
        
        exclusiveThread = currentThread;
        
        if (env.logWriter.hasTarget( LogWriter.DEBUG3 )) {
            env.logWriter.newEntry( this, "beginExclusion(): ", LogWriter.DEBUG3 );
        } 
    } 
    
    
    protected synchronized void endExclusion() {
        if (env.logWriter.hasTarget( LogWriter.DEBUG3 )) {
            env.logWriter.newEntry( this, "endExclusion(): ", LogWriter.DEBUG3 );
        } 
        
        Thread currentThread = Thread.currentThread();
        if (exclusiveThread != currentThread) {
            throw new RuntimeException( "Current thread does not run exclusively." );
        } 
        
        exclusiveThread = null;
        notifyAll();
    } 
    
    
    /**
     * This method checks for deadlocks between all current transaction.
     */
    public synchronized void checkDeadlocks() throws Exception {
        DeadlockRecognition dr = env.deadlockRecognition();
        
        // count blocked transactions
        int blockedCount = 0;
        DxIterator it = taTable.iterator();
        for (Transaction ta; (ta = (Transaction)it.next()) != null;) {
            if (ta.blockedBy() != null) {
                blockedCount++;
            } 
        } 
        
        env.logWriter.newEntry( this,
                "*** checkDeadlocks: blockedCount=" + blockedCount + " exclusive=" + exclusiveThread,
                LogWriter.DEBUG3 );
        
        // we need to check things only if there are at least 2 blocked ta's
        if (blockedCount >= 2) {
            
            try {
                // ensure that nothing changes while we check for deadlocks
                beginExclusion();
                
                it = taTable.iterator();
                for (Transaction ta; (ta = (Transaction)it.next()) != null;) {
                    Transaction candidate = (Transaction)dr.detectDeadlock( ta );
                    if (candidate != null) {
                        env.logWriter.newEntry( this, "*** *** DEADLOCK DETECTED: ta=" + ta, LogWriter.WARN );
                        Thread.sleep( 5000 );
                        
                        // this will be checked by all transactions using
                        // isDeadlockTA() to determine if it has to abort
                        deadlockTA = candidate;
                        // env.store.containerForID (null, ta.blocker).notifyAllTAs (ta);
                        synchronized (deadlockTA) {
                            deadlockTA.notifyAll();
                        } 
                        
                        // DR runs on a higher priority than normal transactions, so
                        // we have to stop after there is one deadlock detected to let
                        // the notification wakeup released transactions; if there are
                        // more deadlocks we will catch them next time
                        return;
                    } 
                } 
            } finally {
                endExclusion();
            } 
        } 
    } 
    
    
    /**
     * Check if the given transaction should abort because of
     * a deadlock.
     */
    public synchronized boolean isDeadlockTA( Transaction ta ) {
        if (ta == deadlockTA) {
            deadlockTA = null;
            return true;
        } 
        return false;
    } 
    
}