transactionmanager.java

Java的面向对象数据库系统的源代码

            }
        } catch (Exception e) {
            e.printStackTrace();
            if (env.logWriter.hasTarget(LogWriter.WARN)) {
                env.logWriter.newEntry(this,"completeTransaction("+command+"): We've caught an exception (following) and that's why we'll abort now.",e,LogWriter.WARN);
            }
            throw e;
        } catch (Error e) {
            e.printStackTrace();
            if (env.logWriter.hasTarget(LogWriter.WARN)) {
                env.logWriter.newEntry(this,"completeTransaction("+command+"): We've caught an error (following) and that's why we'll abort now.",e,LogWriter.WARN);
            }
            throw e;
        } finally {
            if (!alright) {
                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(): start: " + 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 DeadlockException( "" );
                    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 );

                            ta.setDeadlocked(false);

                            long millis = (long)(rand.nextDouble() * ta.increaseDeadlockWaitTimeMaximum());
                            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;
            }
        } finally {
            if (env.logWriter.hasTarget( LogWriter.DEBUG3 )) {
                env.logWriter.newEntry( this, "performCommand(): end: " + ta.toString() + ", " + command.toString(),LogWriter.DEBUG3);
            }
        }
        return result;
    }


    /**
     * Prepare 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 TransactionException( "Thread is not joined to a transaction.", TransactionException.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 TransactionException( "Transaction has inproper status.", TransactionException.STATE );
            return false;
        } else if (ta.rollbackOnly) {
            env.logWriter.newEntry( this, "prepareTransaction():" + ta.toString() + ": rollback only.",
                    LogWriter.WARN );
            abortTransaction( ta, command );
            command.result = new TransactionException( "Transaction is in rollback only status.", TransactionException.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 TransactionException( e.toString(), TransactionException.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 TransactionException( "Thread is not joined to a transaction.", TransactionException.STATE );
        } else  if (ta.status != Transaction.STATUS_PREPARED) {
            env.logWriter.newEntry( this, "commitTransaction(): " + ta.toString() + ": Transaction has inproper status.", LogWriter.WARN );
            command.result = new TransactionException( "Transaction has inproper status.", TransactionException.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 TransactionException( "Thread is not joined to a transaction.", TransactionException.STATE );
        } else if (ta.status >= Transaction.STATUS_COMMITING) {
            env.logWriter.newEntry( this, "abortTransaction(): " + ta.toString() + ": Transaction has inproper status.",
                    LogWriter.WARN );
            command.result = new TransactionException( "Transaction has inproper status.", TransactionException.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 {
                        if (false) {
                            env.logWriter.newEntry(this,"checkExclusion(): waiting... ("+currentTA()+")",LogWriter.DEBUG2);
                        }
                        wait();
                        if (false) {
                            env.logWriter.newEntry(this,"checkExclusion(): notified... ("+currentTA()+")",LogWriter.DEBUG2);
                        }
                    } 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() {
        DeadlockRecognition dr = env.deadlockRecognition();

        // count blocked transactions
        int blockedCount = 0;
        DxIterator it = taTable.iterator();
        for (Transaction ta; (ta = (Transaction)it.next()) != null;) {
            if (ta.isBlocked()) {
                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 );


                        if (false) { // Why should we sleep here?
//                            Thread.sleep(5000);
                        } else {
                            /*
                                We should just synchronized so that we commit the now-set "deadlocked" flags of
                                the deadlocked transactions into main memory where they can read them out.
                            */
                            /* Even that commit should happen sooner or later. */
                            /*
                            synchronized (candidate) {
                            }
                            Thread.sleep(1);
                            */
                        }

                        // this will be checked by all transactions using
                        // isDeadlockTA() to determine if it has to abort
                        deadlockTA = candidate;
                        // env.storeManager.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;
    }

    /**
    	Starts the GarbageColection pre-phase. This is the time where all transactions have to complete (either commit or rollback)
    	until there are no transactions left which were created before this call.
    */
    public void startGarbageCollectionWaitForCurrentTransactionsToCompletePhase(GarbageCollector garbageCollector) {
    	synchronized (this) {
    		DxIterator i = taTable.elementSet().iterator();

    		while (i.next()!=null) {
    			garbageCollector.addTransactionRequiredToComplete(i.object());
    		}
	   		garbageCollector.checkForEndOfWaitForCurrentTransactionsToCompletePhase();
    	}
    }
}

// :indentSize=4:tabSize=4:noTabs=true: