cloudstack-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dar...@apache.org
Subject [29/47] New Transaction API
Date Wed, 23 Oct 2013 19:43:46 GMT
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/f62e28c1/framework/db/src/com/cloud/utils/db/TransactionLegacy.java
----------------------------------------------------------------------
diff --git a/framework/db/src/com/cloud/utils/db/TransactionLegacy.java b/framework/db/src/com/cloud/utils/db/TransactionLegacy.java
new file mode 100755
index 0000000..b191491
--- /dev/null
+++ b/framework/db/src/com/cloud/utils/db/TransactionLegacy.java
@@ -0,0 +1,1174 @@
+// 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
+// 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 com.cloud.utils.db;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Savepoint;
+import java.sql.Statement;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.sql.DataSource;
+
+import org.apache.commons.dbcp.ConnectionFactory;
+import org.apache.commons.dbcp.DriverManagerConnectionFactory;
+import org.apache.commons.dbcp.PoolableConnectionFactory;
+import org.apache.commons.dbcp.PoolingDataSource;
+import org.apache.commons.pool.KeyedObjectPoolFactory;
+import org.apache.commons.pool.impl.GenericObjectPool;
+import org.apache.commons.pool.impl.StackKeyedObjectPoolFactory;
+import org.apache.log4j.Logger;
+import org.jasypt.encryption.pbe.StandardPBEStringEncryptor;
+import org.jasypt.properties.EncryptableProperties;
+
+import com.cloud.utils.Pair;
+import com.cloud.utils.PropertiesUtil;
+import com.cloud.utils.crypt.EncryptionSecretKeyChecker;
+import com.cloud.utils.exception.CloudRuntimeException;
+import com.cloud.utils.mgmt.JmxUtil;
+
+/**
+ * Transaction abstracts away the Connection object in JDBC.  It allows the
+ * following things that the Connection object does not.
+ * 
+ *   1. Transaction can be started at an entry point and whether the DB
+ *      actions should be auto-commit or not determined at that point.
+ *   2. DB Connection is allocated only when it is needed.
+ *   3. Code does not need to know if a transaction has been started or not.
+ *      It just starts/ends a transaction and we resolve it correctly with
+ *      the previous actions.
+ *
+ * Note that this class is not synchronous but it doesn't need to be because
+ * it is stored with TLS and is one per thread.  Use appropriately.
+ */
+public class TransactionLegacy {
+    private static final Logger s_logger = Logger.getLogger(Transaction.class.getName() + "." + "Transaction");
+    private static final Logger s_stmtLogger = Logger.getLogger(Transaction.class.getName() + "." + "Statement");
+    private static final Logger s_lockLogger = Logger.getLogger(Transaction.class.getName() + "." + "Lock");
+    private static final Logger s_connLogger = Logger.getLogger(Transaction.class.getName() + "." + "Connection");
+
+    private static final ThreadLocal<TransactionLegacy> tls = new ThreadLocal<TransactionLegacy>();
+    private static final String START_TXN = "start_txn";
+    private static final String CURRENT_TXN = "current_txn";
+    private static final String CREATE_TXN = "create_txn";
+    private static final String CREATE_CONN = "create_conn";
+    private static final String STATEMENT = "statement";
+    private static final String ATTACHMENT = "attachment";
+
+    public static final short CLOUD_DB = 0;
+    public static final short USAGE_DB = 1;
+    public static final short AWSAPI_DB = 2;
+    public static final short SIMULATOR_DB = 3;
+
+    public static final short CONNECTED_DB = -1;
+
+    private static AtomicLong s_id = new AtomicLong();
+    private static final TransactionMBeanImpl s_mbean = new TransactionMBeanImpl();
+    static {
+        try {
+            JmxUtil.registerMBean("Transaction", "Transaction", s_mbean);
+        } catch (Exception e) {
+            s_logger.error("Unable to register mbean for transaction", e);
+        }
+        
+        /* FIXME: We need a better solution for this
+         * Initialize encryption if we need it for db.properties
+         */ 
+        EncryptionSecretKeyChecker enc = new EncryptionSecretKeyChecker();
+        enc.check();  
+    }
+
+    private final LinkedList<StackElement> _stack;
+    private long _id;
+
+    private final LinkedList<Pair<String, Long>> _lockTimes = new LinkedList<Pair<String, Long>>();
+
+    private String _name;
+    private Connection _conn;
+    private boolean _txn;
+    private short _dbId;
+    private long _txnTime;
+    private Statement _stmt;
+    private String _creator;
+
+    private TransactionLegacy _prev = null;
+
+    public static TransactionLegacy currentTxn() {
+        TransactionLegacy txn = tls.get();
+        assert txn != null : "No Transaction on stack.  Did you mark the method with @DB?";
+
+        assert checkAnnotation(3, txn) : "Did you even read the guide to use Transaction...IOW...other people's code? Try method can't be private.  What about @DB? hmmm... could that be it? " + txn;
+        return txn;
+    }
+
+    public static TransactionLegacy open(final short databaseId) {
+        String name = buildName();
+        if (name == null) {
+            name = CURRENT_TXN;
+        }
+        return open(name, databaseId, true);
+    }
+
+    //
+    // Usage of this transaction setup should be limited, it will always open a new transaction context regardless of whether or not there is other
+    // transaction context in the stack. It is used in special use cases that we want to control DB connection explicitly and in the mean time utilize
+    // the existing DAO features
+    //
+    public void transitToUserManagedConnection(Connection conn) {
+        assert(_conn == null /*&& _stack.size() <= 1*/) : "Can't change to a user managed connection unless the stack is empty and the db connection is null, you may have forgotten to invoke transitToAutoManagedConnection to close out the DB connection: " + toString();
+        _conn = conn;
+        _dbId = CONNECTED_DB;
+    }
+
+    public void transitToAutoManagedConnection(short dbId) {
+        // assert(_stack.size() <= 1) : "Can't change to auto managed connection unless your stack is empty";
+        _dbId = dbId;
+        _conn = null;
+    }
+
+    public static TransactionLegacy open(final String name) {
+        return open(name, TransactionLegacy.CLOUD_DB, false);
+    }
+
+    public static TransactionLegacy open(final String name, final short databaseId, final boolean forceDbChange) {
+        TransactionLegacy txn = tls.get();
+        boolean isNew = false;
+        if (txn == null) {
+            if (s_logger.isTraceEnabled()) {
+                s_logger.trace("Creating the transaction: " + name);
+            }
+            txn = new TransactionLegacy(name, false, databaseId);
+            tls.set(txn);
+            isNew = true;
+        } else if (forceDbChange) {
+            final short currentDbId = txn.getDatabaseId();
+            if (currentDbId != databaseId) {
+                // we need to end the current transaction and switch databases
+                txn.close(txn.getName());
+
+                txn = new TransactionLegacy(name, false, databaseId);
+                tls.set(txn);
+                isNew = true;
+            }
+        }
+
+        txn.takeOver(name, false);
+        if (isNew) {
+            s_mbean.addTransaction(txn);
+        }
+        return txn;
+    }
+
+    protected StackElement peekInStack(Object obj) {
+        final Iterator<StackElement> it = _stack.iterator();
+        while (it.hasNext()) {
+            StackElement next = it.next();
+            if (next.type == obj) {
+                return next;
+            }
+        }
+        return null;
+    }
+
+    public void registerLock(String sql) {
+        if (_txn && s_lockLogger.isDebugEnabled()) {
+            Pair<String, Long> time = new Pair<String, Long>(sql, System.currentTimeMillis());
+            _lockTimes.add(time);
+        }
+    }
+
+    public boolean dbTxnStarted() {
+        return _txn;
+    }
+
+    public static Connection getStandaloneConnectionWithException() throws SQLException {
+        Connection conn = s_ds.getConnection();
+        if (s_connLogger.isTraceEnabled()) {
+            s_connLogger.trace("Retrieving a standalone connection: dbconn" + System.identityHashCode(conn));
+        }
+        return conn;
+    }
+
+    public static Connection getStandaloneConnection() {
+        try {
+            return getStandaloneConnectionWithException();
+        } catch (SQLException e) {
+            s_logger.error("Unexpected exception: ", e);
+            return null;
+        }
+    }
+
+    public static Connection getStandaloneUsageConnection() {
+        try {
+            Connection conn = s_usageDS.getConnection();
+            if (s_connLogger.isTraceEnabled()) {
+                s_connLogger.trace("Retrieving a standalone connection for usage: dbconn" + System.identityHashCode(conn));
+            }
+            return conn;
+        } catch (SQLException e) {
+            s_logger.warn("Unexpected exception: ", e);
+            return null;
+        }
+    }
+
+    public static Connection getStandaloneAwsapiConnection() {
+        try {
+            Connection conn = s_awsapiDS.getConnection();
+            if (s_connLogger.isTraceEnabled()) {
+                s_connLogger.trace("Retrieving a standalone connection for usage: dbconn" + System.identityHashCode(conn));
+            }
+            return conn;
+        } catch (SQLException e) {
+            s_logger.warn("Unexpected exception: ", e);
+            return null;
+        }
+    }
+    
+    public static Connection getStandaloneSimulatorConnection() {
+    	try {
+    		Connection conn = s_simulatorDS.getConnection();
+    		if (s_connLogger.isTraceEnabled()) {
+                s_connLogger.trace("Retrieving a standalone connection for simulator: dbconn" + System.identityHashCode(conn));
+            }
+            return conn;
+        } catch (SQLException e) {
+            s_logger.warn("Unexpected exception: ", e);
+            return null;
+        }
+    }
+
+    protected void attach(TransactionAttachment value) {
+        _stack.push(new StackElement(ATTACHMENT, value));
+    }
+
+    protected TransactionAttachment detach(String name) {
+        Iterator<StackElement> it = _stack.descendingIterator();
+        while (it.hasNext()) {
+            StackElement element = it.next();
+            if (element.type == ATTACHMENT) {
+                TransactionAttachment att = (TransactionAttachment)element.ref;
+                if (name.equals(att.getName())) {
+                    it.remove();
+                    return att;
+                }
+            }
+        }
+        assert false : "Are you sure you attached this: " + name;
+        return null;
+    }
+
+    public static void attachToTxn(TransactionAttachment value) {
+        TransactionLegacy txn = tls.get();
+        assert txn != null && txn.peekInStack(CURRENT_TXN) != null: "Come on....how can we attach something to the transaction if you haven't started it?";
+
+        txn.attach(value);
+    }
+
+    public static TransactionAttachment detachFromTxn(String name) {
+        TransactionLegacy txn = tls.get();
+        assert txn != null : "No Transaction in TLS";
+        return txn.detach(name);
+    }
+
+    protected static boolean checkAnnotation(int stack, TransactionLegacy txn) {
+        final StackTraceElement[] stacks = Thread.currentThread().getStackTrace();
+        StackElement se = txn.peekInStack(CURRENT_TXN);
+        if (se == null) {
+            return false;
+        }
+        
+        StringBuffer sb = new StringBuffer();
+        for (; stack < stacks.length; stack++) {
+            String methodName = stacks[stack].getMethodName();
+            sb.append(" ").append(methodName);
+            if (methodName.equals(se.ref)){
+                return true;
+            }
+        }
+        
+        // relax stack structure for several places that @DB required injection is not in place
+        s_logger.warn("Non-standard stack context that Transaction context is manaully placed into the calling chain. Stack chain: " + sb);
+        return true;
+    }
+
+    protected static String buildName() {
+        if (s_logger.isDebugEnabled()) {
+            final StackTraceElement[] stacks = Thread.currentThread().getStackTrace();
+            final StringBuilder str = new StringBuilder();
+            int i = 3, j = 3;
+            while (j < 15 && i < stacks.length) {
+                StackTraceElement element = stacks[i];
+                String filename = element.getFileName();
+                String method = element.getMethodName();
+                if ((filename != null && filename.equals("<generated>")) || (method != null && method.equals("invokeSuper"))) {
+                    i++;
+                    continue;
+                }
+
+                str.append("-").append(stacks[i].getClassName().substring(stacks[i].getClassName().lastIndexOf(".") + 1)).append(".").append(stacks[i].getMethodName()).append(":").append(stacks[i].getLineNumber());
+                j++;
+                i++;
+            }
+            return str.toString();
+        }
+
+        return "";
+    }
+
+    public TransactionLegacy(final String name, final boolean forLocking, final short databaseId) {
+        _name = name;
+        _conn = null;
+        _stack = new LinkedList<StackElement>();
+        _txn = false;
+        _dbId = databaseId;
+        _id = s_id.incrementAndGet();
+        _creator = Thread.currentThread().getName();
+    }
+
+    public String getCreator() {
+        return _creator;
+    }
+
+    public long getId() {
+        return _id;
+    }
+
+    public String getName() {
+        return _name;
+    }
+
+    public Short getDatabaseId() {
+        return _dbId;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder str = new StringBuilder((_name != null ? _name : ""));
+        str.append(" : ");
+        for (final StackElement se : _stack) {
+            if (se.type == CURRENT_TXN) {
+                str.append(se.ref).append(", ");
+            }
+        }
+
+        return str.toString();
+    }
+
+    protected void mark(final String name) {
+        _stack.push(new StackElement(CURRENT_TXN, name));
+    }
+
+    public boolean lock(final String name, final int timeoutSeconds) {
+        Merovingian2 lockMaster = Merovingian2.getLockMaster();
+        if (lockMaster == null) {
+            throw new CloudRuntimeException("There's no support for locking yet");
+        }
+        return lockMaster.acquire(name, timeoutSeconds);
+    }
+
+    public boolean release(final String name) {
+        Merovingian2 lockMaster = Merovingian2.getLockMaster();
+        if (lockMaster == null) {
+            throw new CloudRuntimeException("There's no support for locking yet");
+        }
+        return lockMaster.release(name);
+    }
+
+    public void start() {
+        if (s_logger.isTraceEnabled()) {
+            s_logger.trace("txn: start requested by: " + buildName());
+        }
+
+        _stack.push(new StackElement(START_TXN, null));
+
+        if (_txn) {
+            s_logger.trace("txn: has already been started.");
+            return;
+        }
+
+        _txn = true;
+
+        _txnTime = System.currentTimeMillis();
+        if (_conn != null) {
+            try {
+                s_logger.trace("txn: set auto commit to false");
+                _conn.setAutoCommit(false);
+            } catch (final SQLException e) {
+                s_logger.warn("Unable to set auto commit: ", e);
+                throw new CloudRuntimeException("Unable to set auto commit: ", e);
+            }
+        }
+    }
+
+    protected void closePreviousStatement() {
+        if (_stmt != null) {
+            try {
+                if (s_stmtLogger.isTraceEnabled()) {
+                    s_stmtLogger.trace("Closing: " + _stmt.toString());
+                }
+                try {
+                    ResultSet rs = _stmt.getResultSet();
+                    if (rs != null && _stmt.getResultSetHoldability() != ResultSet.HOLD_CURSORS_OVER_COMMIT) {
+                        rs.close();
+                    }
+                } catch(SQLException e) {
+                    s_stmtLogger.trace("Unable to close resultset");
+                }
+                _stmt.close();
+            } catch (final SQLException e) {
+                s_stmtLogger.trace("Unable to close statement: " + _stmt.toString());
+            } finally {
+                _stmt = null;
+            }
+        }
+    }
+
+    /**
+     * Prepares an auto close statement.  The statement is closed automatically if it is
+     * retrieved with this method.
+     * 
+     * @param sql sql String
+     * @return PreparedStatement
+     * @throws SQLException if problem with JDBC layer.
+     * 
+     * @see java.sql.Connection
+     */
+    public PreparedStatement prepareAutoCloseStatement(final String sql) throws SQLException {
+        PreparedStatement stmt = prepareStatement(sql);
+        closePreviousStatement();
+        _stmt = stmt;
+        return stmt;
+    }
+
+    public PreparedStatement prepareStatement(final String sql) throws SQLException {
+        final Connection conn = getConnection();
+        final PreparedStatement pstmt = conn.prepareStatement(sql);
+        if (s_stmtLogger.isTraceEnabled()) {
+            s_stmtLogger.trace("Preparing: " + sql);
+        }
+        return pstmt;
+    }
+
+    /**
+     * Prepares an auto close statement.  The statement is closed automatically if it is
+     * retrieved with this method.
+     * 
+     * @param sql sql String
+     * @param autoGeneratedKeys keys that are generated
+     * @return PreparedStatement
+     * @throws SQLException if problem with JDBC layer.
+     * 
+     * @see java.sql.Connection
+     */
+    public PreparedStatement prepareAutoCloseStatement(final String sql, final int autoGeneratedKeys) throws SQLException {
+        final Connection conn = getConnection();
+        final PreparedStatement pstmt = conn.prepareStatement(sql, autoGeneratedKeys);
+        if (s_stmtLogger.isTraceEnabled()) {
+            s_stmtLogger.trace("Preparing: " + sql);
+        }
+        closePreviousStatement();
+        _stmt = pstmt;
+        return pstmt;
+    }
+
+    /**
+     * Prepares an auto close statement.  The statement is closed automatically if it is
+     * retrieved with this method.
+     * 
+     * @param sql sql String
+     * @param columnNames names of the columns
+     * @return PreparedStatement
+     * @throws SQLException if problem with JDBC layer.
+     * 
+     * @see java.sql.Connection
+     */
+    public PreparedStatement prepareAutoCloseStatement(final String sql, final String[] columnNames) throws SQLException {
+        final Connection conn = getConnection();
+        final PreparedStatement pstmt = conn.prepareStatement(sql, columnNames);
+        if (s_stmtLogger.isTraceEnabled()) {
+            s_stmtLogger.trace("Preparing: " + sql);
+        }
+        closePreviousStatement();
+        _stmt = pstmt;
+        return pstmt;
+    }
+
+    /**
+     * Prepares an auto close statement.  The statement is closed automatically if it is
+     * retrieved with this method.
+     * 
+     * @param sql sql String
+     * @return PreparedStatement
+     * @throws SQLException if problem with JDBC layer.
+     * 
+     * @see java.sql.Connection
+     */
+    public PreparedStatement prepareAutoCloseStatement(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
+        final Connection conn = getConnection();
+        final PreparedStatement pstmt = conn.prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
+        if (s_stmtLogger.isTraceEnabled()) {
+            s_stmtLogger.trace("Preparing: " + sql);
+        }
+        closePreviousStatement();
+        _stmt = pstmt;
+        return pstmt;
+    }
+
+    /**
+     * Returns the db connection.
+     * 
+     * Note: that you can call getConnection() but beaware that
+     * all prepare statements from the Connection are not garbage
+     * collected!
+     * 
+     * @return DB Connection but make sure you understand that
+     *         you are responsible for closing the PreparedStatement.
+     * @throws SQLException
+     */
+    public Connection getConnection() throws SQLException {
+        if (_conn == null) {
+            switch (_dbId) {
+            case CLOUD_DB:
+                if(s_ds != null) {
+                    _conn = s_ds.getConnection();
+                } else {
+                    s_logger.warn("A static-initialized variable becomes null, process is dying?");
+                    throw new CloudRuntimeException("Database is not initialized, process is dying?");
+                }
+                break;
+            case USAGE_DB:
+                if(s_usageDS != null) {
+                    _conn = s_usageDS.getConnection();
+                } else {
+                    s_logger.warn("A static-initialized variable becomes null, process is dying?");
+                    throw new CloudRuntimeException("Database is not initialized, process is dying?");
+                }
+                break;
+            case AWSAPI_DB:
+        	if(s_awsapiDS != null) {
+        	    _conn = s_awsapiDS.getConnection();
+        	} else {
+        	    s_logger.warn("A static-initialized variable becomes null, process is dying?");
+                throw new CloudRuntimeException("Database is not initialized, process is dying?");
+        	}
+                break;
+
+            case SIMULATOR_DB:
+                if(s_simulatorDS != null) {
+                    _conn = s_simulatorDS.getConnection();
+                } else {
+                    s_logger.warn("A static-initialized variable becomes null, process is dying?");
+                    throw new CloudRuntimeException("Database is not initialized, process is dying?");
+                }
+                break;
+            default:
+
+        	throw new CloudRuntimeException("No database selected for the transaction");
+            }
+            _conn.setAutoCommit(!_txn);
+
+            //
+            // MySQL default transaction isolation level is REPEATABLE READ,
+            // to reduce chances of DB deadlock, we will use READ COMMITED isolation level instead
+            // see http://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html
+            //
+            _stack.push(new StackElement(CREATE_CONN, null));
+            if (s_connLogger.isTraceEnabled()) {
+                s_connLogger.trace("Creating a DB connection with " + (_txn ? " txn: " : " no txn: ") + " for " + _dbId + ": dbconn" + System.identityHashCode(_conn) + ". Stack: " + buildName());
+            }
+        } else {
+            s_logger.trace("conn: Using existing DB connection");
+        }
+
+        return _conn;
+    }
+
+    protected boolean takeOver(final String name, final boolean create) {
+        if (_stack.size() != 0) {
+            if (!create) {
+                // If it is not a create transaction, then let's just use the current one.
+                if (s_logger.isTraceEnabled()) {
+                    s_logger.trace("Using current transaction: " + toString());
+                }
+                mark(name);
+                return false;
+            }
+
+            final StackElement se = _stack.getFirst();
+            if (se.type == CREATE_TXN) {
+                // This create is called inside of another create.  Which is ok?
+                // We will let that create be responsible for cleaning up.
+                if (s_logger.isTraceEnabled()) {
+                    s_logger.trace("Create using current transaction: " + toString());
+                }
+                mark(name);
+                return false;
+            }
+
+            s_logger.warn("Encountered a transaction that has leaked.  Cleaning up. " + toString());
+            cleanup();
+        }
+
+        if (s_logger.isTraceEnabled()) {
+            s_logger.trace("Took over the transaction: " + name);
+        }
+        _stack.push(new StackElement(create ? CREATE_TXN : CURRENT_TXN, name));
+        _name = name;
+        return true;
+    }
+
+    public void cleanup() {
+        closePreviousStatement();
+
+        removeUpTo(null, null);
+        if (_txn) {
+            rollbackTransaction();
+        }
+        _txn = false;
+        _name = null;
+
+        closeConnection();
+
+        _stack.clear();
+        Merovingian2 lockMaster = Merovingian2.getLockMaster();
+        if (lockMaster != null) {
+            lockMaster.cleanupThread();
+        }
+    }
+
+    public void close() {
+        removeUpTo(CURRENT_TXN, null);
+
+        if (_stack.size() == 0) {
+            s_logger.trace("Transaction is done");
+            cleanup();
+        }
+    }
+
+    /**
+     * close() is used by endTxn to close the connection.  This method only
+     * closes the connection if the name is the same as what's stored.
+     * 
+     * @param name
+     * @return true if this close actually closes the connection.  false if not.
+     */
+    public boolean close(final String name) {
+        if (_name == null) {    // Already cleaned up.
+            if (s_logger.isTraceEnabled()) {
+                s_logger.trace("Already cleaned up." + buildName());
+            }
+            return true;
+        }
+
+        if (!_name.equals(name)) {
+            close();
+            return false;
+        }
+
+        if (s_logger.isDebugEnabled() && _stack.size() > 2) {
+            s_logger.debug("Transaction is not closed properly: " + toString() + ".  Called by " + buildName());
+        }
+
+        cleanup();
+
+        s_logger.trace("All done");
+        return true;
+    }
+
+    protected boolean hasTxnInStack() {
+        return peekInStack(START_TXN) != null;
+    }
+
+    protected void clearLockTimes() {
+        if (s_lockLogger.isDebugEnabled()) {
+            for (Pair<String, Long> time : _lockTimes) {
+                s_lockLogger.trace("SQL " + time.first() + " took " + (System.currentTimeMillis() - time.second()));
+            }
+            _lockTimes.clear();
+        }
+    }
+
+    public boolean commit() {
+        if (!_txn) {
+            s_logger.warn("txn: Commit called when it is not a transaction: " + buildName());
+            return false;
+        }
+
+        Iterator<StackElement> it = _stack.iterator();
+        while (it.hasNext()) {
+            StackElement st = it.next();
+            if (st.type == START_TXN) {
+                it.remove();
+                break;
+            }
+        }
+
+        if (hasTxnInStack()) {
+            if (s_logger.isTraceEnabled()) {
+                s_logger.trace("txn: Not committing because transaction started elsewhere: " + buildName() + " / " + toString());
+            }
+            return false;
+        }
+
+        _txn = false;
+        try {
+            if (_conn != null) {
+                _conn.commit();
+                s_logger.trace("txn: DB Changes committed. Time = " + (System.currentTimeMillis() - _txnTime));
+                clearLockTimes();
+                closeConnection();
+            }
+            return true;
+        } catch (final SQLException e) {
+            rollbackTransaction();
+            throw new CloudRuntimeException("Unable to commit or close the connection. ", e);
+        }
+    }
+
+    protected void closeConnection() {
+        closePreviousStatement();
+
+        if (_conn == null) {
+            return;
+        }
+
+        if (_txn) {
+            s_connLogger.trace("txn: Not closing DB connection because we're still in a transaction.");
+            return;
+        }
+
+        try {
+            // we should only close db connection when it is not user managed
+            if (this._dbId != CONNECTED_DB) {
+                if (s_connLogger.isTraceEnabled()) {
+                    s_connLogger.trace("Closing DB connection: dbconn" + System.identityHashCode(_conn));
+                }                                
+                _conn.close();
+                _conn = null;  
+            }
+
+        } catch (final SQLException e) {
+            s_logger.warn("Unable to close connection", e);
+        }
+    }
+
+    protected void removeUpTo(String type, Object ref) {
+        boolean rollback = false;
+        Iterator<StackElement> it = _stack.iterator();
+        while (it.hasNext()) {
+            StackElement item = it.next();
+
+            it.remove();
+
+            try {
+                if (item.type == type && (ref == null || item.ref == ref)) {
+                    break;
+                }
+
+                if (item.type == CURRENT_TXN) {
+                    if (s_logger.isTraceEnabled()) {
+                        s_logger.trace("Releasing the current txn: " + (item.ref != null ? item.ref : ""));
+                    }
+                } else if (item.type == CREATE_CONN) {
+                    closeConnection();
+                } else if (item.type == START_TXN) {
+                    if (item.ref == null) {
+                        rollback = true;
+                    } else {
+                        try {
+                            _conn.rollback((Savepoint)ref);
+                            rollback = false;
+                        } catch (final SQLException e) {
+                            s_logger.warn("Unable to rollback Txn.", e);
+                        }
+                    }
+                } else if (item.type == STATEMENT) {
+                    try {
+                        if (s_stmtLogger.isTraceEnabled()) {
+                            s_stmtLogger.trace("Closing: " + ref.toString());
+                        }
+                        Statement stmt = (Statement)ref;
+                        try {
+                            ResultSet rs = stmt.getResultSet();
+                            if (rs != null) {
+                                rs.close();
+                            }
+                        } catch(SQLException e) {
+                            s_stmtLogger.trace("Unable to close resultset");
+                        }
+                        stmt.close();
+                    } catch (final SQLException e) {
+                        s_stmtLogger.trace("Unable to close statement: " + item);
+                    }
+                } else if (item.type == ATTACHMENT) {
+                    TransactionAttachment att = (TransactionAttachment)item.ref;
+                    if (s_logger.isTraceEnabled()) {
+                        s_logger.trace("Cleaning up " + att.getName());
+                    }
+                    att.cleanup();
+                }
+            } catch(Exception e) {
+                s_logger.error("Unable to clean up " + item, e);
+            }
+        }
+
+        if (rollback) {
+            rollback();
+        }
+    }
+
+    protected void rollbackTransaction() {
+        closePreviousStatement();
+        if (!_txn) {
+            if (s_logger.isTraceEnabled()) {
+                s_logger.trace("Rollback called for " + _name + " when there's no transaction: " + buildName());
+            }
+            return;
+        }
+        assert (!hasTxnInStack()) : "Who's rolling back transaction when there's still txn in stack?";
+        _txn = false;
+        try {
+            if (_conn != null) {
+                if (s_logger.isDebugEnabled()) {
+                    s_logger.debug("Rolling back the transaction: Time = " + (System.currentTimeMillis() - _txnTime) + " Name =  " + _name + "; called by " + buildName());
+                }
+                _conn.rollback();
+            }
+            clearLockTimes();
+            closeConnection();
+        } catch(final SQLException e) {
+            s_logger.warn("Unable to rollback", e);
+        }
+    }
+
+    protected void rollbackSavepoint(Savepoint sp) {
+        try {
+            if (_conn != null) {
+                _conn.rollback(sp);
+            }
+        } catch (SQLException e) {
+            s_logger.warn("Unable to rollback to savepoint " + sp);
+        }
+
+        if (!hasTxnInStack()) {
+            _txn = false;
+            closeConnection();
+        }
+    }
+
+    public void rollback() {
+        Iterator<StackElement> it = _stack.iterator();
+        while (it.hasNext()) {
+            StackElement st = it.next();
+            if (st.type == START_TXN) {
+                if (st.ref == null) {
+                    it.remove();
+                } else  {
+                    rollback((Savepoint)st.ref);
+                    return;
+                }
+            }
+        }
+
+        rollbackTransaction();
+    }
+
+    public Savepoint setSavepoint() throws SQLException {
+        _txn = true;
+        StackElement st = new StackElement(START_TXN, null);
+        _stack.push(st);
+        final Connection conn = getConnection();
+        final Savepoint sp = conn.setSavepoint();
+        st.ref = sp;
+
+        return sp;
+    }
+
+    public Savepoint setSavepoint(final String name) throws SQLException {
+        _txn = true;
+        StackElement st = new StackElement(START_TXN, null);
+        _stack.push(st);
+        final Connection conn = getConnection();
+        final Savepoint sp = conn.setSavepoint(name);
+        st.ref = sp;
+
+        return sp;
+    }
+
+    public void releaseSavepoint(final Savepoint sp) throws SQLException {
+        removeTxn(sp);
+        if (_conn != null) {
+            _conn.releaseSavepoint(sp);
+        }
+
+        if (!hasTxnInStack()) {
+            _txn = false;
+            closeConnection();
+        }
+    }
+
+    protected boolean hasSavepointInStack(Savepoint sp) {
+        Iterator<StackElement> it = _stack.iterator();
+        while (it.hasNext()) {
+            StackElement se = it.next();
+            if (se.type == START_TXN && se.ref == sp) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    protected void removeTxn(Savepoint sp) {
+        assert hasSavepointInStack(sp) : "Removing a save point that's not in the stack";
+
+        if (!hasSavepointInStack(sp)) {
+            return;
+        }
+
+        Iterator<StackElement> it = _stack.iterator();
+        while (it.hasNext()) {
+            StackElement se = it.next();
+            if (se.type == START_TXN) {
+                it.remove();
+                if (se.ref == sp) {
+                    return;
+                }
+            }
+        }
+    }
+
+    public void rollback(final Savepoint sp) {
+        removeTxn(sp);
+
+        rollbackSavepoint(sp);
+    }
+
+    public Connection getCurrentConnection() {
+        return _conn;
+    }
+
+    public List<StackElement> getStack() {
+        return _stack;
+    }
+
+    protected TransactionLegacy() {
+        _name = null;
+        _conn = null;
+        _stack = null;
+        _txn = false;
+        _dbId = -1;
+    }
+
+    @Override
+    protected void finalize() throws Throwable {
+        if (!(_conn == null && (_stack == null || _stack.size() == 0))) {
+            assert (false) : "Oh Alex oh alex...something is wrong with how we're doing this";
+            s_logger.error("Something went wrong that a transaction is orphaned before db connection is closed");
+            cleanup();
+        }
+    }
+
+    protected class StackElement {
+        public String type;
+        public Object ref;
+
+        public StackElement (String type, Object ref) {
+            this.type = type;
+            this.ref = ref;
+        }
+
+        @Override
+        public String toString() {
+            return type + "-" + ref;
+        }
+    }
+
+    private static DataSource s_ds;
+    private static DataSource s_usageDS;
+    private static DataSource s_awsapiDS;
+    private static DataSource s_simulatorDS;
+
+    static {
+        // Initialize with assumed db.properties file
+        initDataSource("db.properties");
+    }
+
+    public static void initDataSource(String propsFileName) {
+        try {
+            File dbPropsFile = PropertiesUtil.findConfigFile(propsFileName);
+            final Properties dbProps;
+            if (EncryptionSecretKeyChecker.useEncryption()) {
+                StandardPBEStringEncryptor encryptor = EncryptionSecretKeyChecker.getEncryptor();
+                dbProps = new EncryptableProperties(encryptor);
+            } else {
+                dbProps = new Properties();
+            }
+            try {
+                dbProps.load(new FileInputStream(dbPropsFile));
+            } catch (IOException e) {
+                s_logger.fatal("Unable to load db properties file, pl. check the classpath and file path configuration", e);
+                return;
+            } catch (NullPointerException e) {
+                s_logger.fatal("Unable to locate db properties file within classpath or absolute path: " + propsFileName);
+                return;
+            }
+
+            // FIXME:  If params are missing...default them????
+            final int cloudMaxActive = Integer.parseInt(dbProps.getProperty("db.cloud.maxActive"));
+            final int cloudMaxIdle = Integer.parseInt(dbProps.getProperty("db.cloud.maxIdle"));
+            final long cloudMaxWait = Long.parseLong(dbProps.getProperty("db.cloud.maxWait"));
+            final String cloudUsername = dbProps.getProperty("db.cloud.username");
+            final String cloudPassword = dbProps.getProperty("db.cloud.password");
+            final String cloudHost = dbProps.getProperty("db.cloud.host");
+            final int cloudPort = Integer.parseInt(dbProps.getProperty("db.cloud.port"));
+            final String cloudDbName = dbProps.getProperty("db.cloud.name");
+            final boolean cloudAutoReconnect = Boolean.parseBoolean(dbProps.getProperty("db.cloud.autoReconnect"));
+            final String cloudValidationQuery = dbProps.getProperty("db.cloud.validationQuery");
+            final String cloudIsolationLevel = dbProps.getProperty("db.cloud.isolation.level");
+
+            int isolationLevel = Connection.TRANSACTION_READ_COMMITTED;
+            if (cloudIsolationLevel == null) {
+                isolationLevel = Connection.TRANSACTION_READ_COMMITTED;
+            } else if (cloudIsolationLevel.equalsIgnoreCase("readcommitted")) {
+                isolationLevel = Connection.TRANSACTION_READ_COMMITTED;
+            } else if (cloudIsolationLevel.equalsIgnoreCase("repeatableread")) {
+                isolationLevel = Connection.TRANSACTION_REPEATABLE_READ;
+            } else if (cloudIsolationLevel.equalsIgnoreCase("serializable")) {
+                isolationLevel = Connection.TRANSACTION_SERIALIZABLE;
+            } else if (cloudIsolationLevel.equalsIgnoreCase("readuncommitted")) {
+                isolationLevel = Connection.TRANSACTION_READ_UNCOMMITTED;
+            } else {
+                s_logger.warn("Unknown isolation level " + cloudIsolationLevel + ".  Using read uncommitted");
+            }
+
+            final boolean cloudTestOnBorrow = Boolean.parseBoolean(dbProps.getProperty("db.cloud.testOnBorrow"));
+            final boolean cloudTestWhileIdle = Boolean.parseBoolean(dbProps.getProperty("db.cloud.testWhileIdle"));
+            final long cloudTimeBtwEvictionRunsMillis = Long.parseLong(dbProps.getProperty("db.cloud.timeBetweenEvictionRunsMillis"));
+            final long cloudMinEvcitableIdleTimeMillis = Long.parseLong(dbProps.getProperty("db.cloud.minEvictableIdleTimeMillis"));
+            final boolean cloudPoolPreparedStatements = Boolean.parseBoolean(dbProps.getProperty("db.cloud.poolPreparedStatements"));
+            final String url = dbProps.getProperty("db.cloud.url.params");
+
+            final boolean useSSL = Boolean.parseBoolean(dbProps.getProperty("db.cloud.useSSL"));
+            if (useSSL) {
+                System.setProperty("javax.net.ssl.keyStore", dbProps.getProperty("db.cloud.keyStore"));
+                System.setProperty("javax.net.ssl.keyStorePassword", dbProps.getProperty("db.cloud.keyStorePassword"));
+                System.setProperty("javax.net.ssl.trustStore", dbProps.getProperty("db.cloud.trustStore"));
+                System.setProperty("javax.net.ssl.trustStorePassword", dbProps.getProperty("db.cloud.trustStorePassword"));
+            }
+
+            final GenericObjectPool cloudConnectionPool = new GenericObjectPool(null, cloudMaxActive, GenericObjectPool.DEFAULT_WHEN_EXHAUSTED_ACTION,
+                    cloudMaxWait, cloudMaxIdle, cloudTestOnBorrow, false, cloudTimeBtwEvictionRunsMillis, 1, cloudMinEvcitableIdleTimeMillis, cloudTestWhileIdle);
+
+            final ConnectionFactory cloudConnectionFactory = new DriverManagerConnectionFactory("jdbc:mysql://" + cloudHost + ":" + cloudPort + "/" + cloudDbName +
+                    "?autoReconnect=" + cloudAutoReconnect + (url != null ? "&" + url : "") + (useSSL ? "&useSSL=true" : ""), cloudUsername, cloudPassword);
+
+            final KeyedObjectPoolFactory poolableObjFactory = (cloudPoolPreparedStatements ? new StackKeyedObjectPoolFactory() : null);
+
+            final PoolableConnectionFactory cloudPoolableConnectionFactory = new PoolableConnectionFactory(cloudConnectionFactory, cloudConnectionPool, poolableObjFactory,
+                    cloudValidationQuery, false, false, isolationLevel);
+
+            // Default Data Source for CloudStack
+            s_ds = new PoolingDataSource(cloudPoolableConnectionFactory.getPool());
+
+            // Configure the usage db
+            final int usageMaxActive = Integer.parseInt(dbProps.getProperty("db.usage.maxActive"));
+            final int usageMaxIdle = Integer.parseInt(dbProps.getProperty("db.usage.maxIdle"));
+            final long usageMaxWait = Long.parseLong(dbProps.getProperty("db.usage.maxWait"));
+            final String usageUsername = dbProps.getProperty("db.usage.username");
+            final String usagePassword = dbProps.getProperty("db.usage.password");
+            final String usageHost = dbProps.getProperty("db.usage.host");
+            final int usagePort = Integer.parseInt(dbProps.getProperty("db.usage.port"));
+            final String usageDbName = dbProps.getProperty("db.usage.name");
+            final boolean usageAutoReconnect = Boolean.parseBoolean(dbProps.getProperty("db.usage.autoReconnect"));
+            final String usageUrl = dbProps.getProperty("db.usage.url.params");
+
+            final GenericObjectPool usageConnectionPool = new GenericObjectPool(null, usageMaxActive, GenericObjectPool.DEFAULT_WHEN_EXHAUSTED_ACTION,
+                    usageMaxWait, usageMaxIdle);
+
+            final ConnectionFactory usageConnectionFactory = new DriverManagerConnectionFactory("jdbc:mysql://" + usageHost + ":" + usagePort + "/" + usageDbName +
+                    "?autoReconnect=" + usageAutoReconnect + (usageUrl != null ? "&" + usageUrl : ""), usageUsername, usagePassword);
+
+            final PoolableConnectionFactory usagePoolableConnectionFactory = new PoolableConnectionFactory(usageConnectionFactory, usageConnectionPool,
+                    new StackKeyedObjectPoolFactory(), null, false, false);
+
+            // Data Source for usage server
+            s_usageDS = new PoolingDataSource(usagePoolableConnectionFactory.getPool());
+
+            // Configure awsapi db
+            final String awsapiDbName = dbProps.getProperty("db.awsapi.name");
+            final GenericObjectPool awsapiConnectionPool = new GenericObjectPool(null, usageMaxActive, GenericObjectPool.DEFAULT_WHEN_EXHAUSTED_ACTION,
+                    usageMaxWait, usageMaxIdle);
+            final ConnectionFactory awsapiConnectionFactory = new DriverManagerConnectionFactory("jdbc:mysql://" + cloudHost + ":" + cloudPort + "/" + awsapiDbName +
+                    "?autoReconnect=" + usageAutoReconnect, cloudUsername, cloudPassword);
+            final PoolableConnectionFactory awsapiPoolableConnectionFactory = new PoolableConnectionFactory(awsapiConnectionFactory, awsapiConnectionPool,
+                    new StackKeyedObjectPoolFactory(), null, false, false);
+
+            // Data Source for awsapi
+            s_awsapiDS = new PoolingDataSource(awsapiPoolableConnectionFactory.getPool());
+
+            try {
+                // Configure the simulator db
+                final int simulatorMaxActive = Integer.parseInt(dbProps.getProperty("db.simulator.maxActive"));
+                final int simulatorMaxIdle = Integer.parseInt(dbProps.getProperty("db.simulator.maxIdle"));
+                final long simulatorMaxWait = Long.parseLong(dbProps.getProperty("db.simulator.maxWait"));
+                final String simulatorUsername = dbProps.getProperty("db.simulator.username");
+                final String simulatorPassword = dbProps.getProperty("db.simulator.password");
+                final String simulatorHost = dbProps.getProperty("db.simulator.host");
+                final int simulatorPort = Integer.parseInt(dbProps.getProperty("db.simulator.port"));
+                final String simulatorDbName = dbProps.getProperty("db.simulator.name");
+                final boolean simulatorAutoReconnect = Boolean.parseBoolean(dbProps.getProperty("db.simulator.autoReconnect"));
+
+                final GenericObjectPool simulatorConnectionPool = new GenericObjectPool(null, simulatorMaxActive, GenericObjectPool.DEFAULT_WHEN_EXHAUSTED_ACTION,
+                        simulatorMaxWait, simulatorMaxIdle);
+
+                final ConnectionFactory simulatorConnectionFactory = new DriverManagerConnectionFactory("jdbc:mysql://" + simulatorHost + ":" + simulatorPort + "/" + simulatorDbName +
+                        "?autoReconnect=" + simulatorAutoReconnect, simulatorUsername, simulatorPassword);
+
+                final PoolableConnectionFactory simulatorPoolableConnectionFactory = new PoolableConnectionFactory(simulatorConnectionFactory, simulatorConnectionPool,
+                        new StackKeyedObjectPoolFactory(), null, false, false);
+                s_simulatorDS = new PoolingDataSource(simulatorPoolableConnectionFactory.getPool());
+            } catch (Exception e) {
+                s_logger.debug("Simulator DB properties are not available. Not initializing simulator DS");
+            }
+        } catch (final Exception e) {
+            s_ds = getDefaultDataSource("cloud");
+            s_usageDS = getDefaultDataSource("cloud_usage");
+            s_simulatorDS = getDefaultDataSource("cloud_simulator");
+            s_logger.warn("Unable to load db configuration, using defaults with 5 connections. Falling back on assumed datasource on localhost:3306 using username:password=cloud:cloud. Please check your configuration", e);
+        }
+    }
+
+    private static DataSource getDefaultDataSource(final String database) {
+        final GenericObjectPool connectionPool = new GenericObjectPool(null, 5);
+        final ConnectionFactory connectionFactory = new DriverManagerConnectionFactory(
+           "jdbc:mysql://localhost:3306/" + database, "cloud", "cloud");
+        final PoolableConnectionFactory poolableConnectionFactory = new PoolableConnectionFactory(
+           connectionFactory, connectionPool, null, null, false, true);
+        return new PoolingDataSource(
+           /* connectionPool */poolableConnectionFactory.getPool());
+    }
+    
+}

http://git-wip-us.apache.org/repos/asf/cloudstack/blob/f62e28c1/framework/db/src/com/cloud/utils/db/TransactionMBeanImpl.java
----------------------------------------------------------------------
diff --git a/framework/db/src/com/cloud/utils/db/TransactionMBeanImpl.java b/framework/db/src/com/cloud/utils/db/TransactionMBeanImpl.java
index d51a9bd..73511b1 100644
--- a/framework/db/src/com/cloud/utils/db/TransactionMBeanImpl.java
+++ b/framework/db/src/com/cloud/utils/db/TransactionMBeanImpl.java
@@ -25,21 +25,21 @@ import java.util.concurrent.ConcurrentHashMap;
 
 import javax.management.StandardMBean;
 
-import com.cloud.utils.db.Transaction.StackElement;
+import com.cloud.utils.db.TransactionLegacy.StackElement;
 
 public class TransactionMBeanImpl extends StandardMBean implements TransactionMBean {
     
-    Map<Long, Transaction> _txns = new ConcurrentHashMap<Long, Transaction>();
+    Map<Long, TransactionLegacy> _txns = new ConcurrentHashMap<Long, TransactionLegacy>();
     
     public TransactionMBeanImpl() {
         super(TransactionMBean.class, false);
     }
     
-    public void addTransaction(Transaction txn) {
+    public void addTransaction(TransactionLegacy txn) {
         _txns.put(txn.getId(), txn);
     }
     
-    public void removeTransaction(Transaction txn) {
+    public void removeTransaction(TransactionLegacy txn) {
         _txns.remove(txn.getId());
     }
     
@@ -53,7 +53,7 @@ public class TransactionMBeanImpl extends StandardMBean implements TransactionMB
         int[] count = new int[2];
         count[0] = 0;
         count[1] = 0;
-        for (Transaction txn : _txns.values()) {
+        for (TransactionLegacy txn : _txns.values()) {
             if (txn.getStack().size() > 0) {
                 count[0]++;
             }
@@ -67,7 +67,7 @@ public class TransactionMBeanImpl extends StandardMBean implements TransactionMB
     @Override
     public List<Map<String, String>> getTransactions() {
         ArrayList<Map<String, String>> txns = new ArrayList<Map<String, String>>();
-        for (Transaction info : _txns.values()) {
+        for (TransactionLegacy info : _txns.values()) {
             txns.add(toMap(info));
         }
         return txns;
@@ -76,7 +76,7 @@ public class TransactionMBeanImpl extends StandardMBean implements TransactionMB
     @Override
     public List<Map<String, String>> getActiveTransactions() {
         ArrayList<Map<String, String>> txns = new ArrayList<Map<String, String>>();
-        for (Transaction txn : _txns.values()) {
+        for (TransactionLegacy txn : _txns.values()) {
             if (txn.getStack().size() > 0 || txn.getCurrentConnection() != null) {
                 txns.add(toMap(txn));
             }
@@ -84,7 +84,7 @@ public class TransactionMBeanImpl extends StandardMBean implements TransactionMB
         return txns;
     }
     
-    protected Map<String, String> toMap(Transaction txn) {
+    protected Map<String, String> toMap(TransactionLegacy txn) {
         Map<String, String> map = new HashMap<String, String>();
         map.put("name", txn.getName());
         map.put("id", Long.toString(txn.getId()));
@@ -103,7 +103,7 @@ public class TransactionMBeanImpl extends StandardMBean implements TransactionMB
     @Override
     public List<Map<String, String>> getTransactionsWithDatabaseConnection() {
         ArrayList<Map<String, String>> txns = new ArrayList<Map<String, String>>();
-        for (Transaction txn : _txns.values()) {
+        for (TransactionLegacy txn : _txns.values()) {
             if (txn.getCurrentConnection() != null) {
                 txns.add(toMap(txn));
             }

http://git-wip-us.apache.org/repos/asf/cloudstack/blob/f62e28c1/framework/db/src/com/cloud/utils/db/TransactionStatus.java
----------------------------------------------------------------------
diff --git a/framework/db/src/com/cloud/utils/db/TransactionStatus.java b/framework/db/src/com/cloud/utils/db/TransactionStatus.java
new file mode 100644
index 0000000..a167797
--- /dev/null
+++ b/framework/db/src/com/cloud/utils/db/TransactionStatus.java
@@ -0,0 +1,7 @@
+package com.cloud.utils.db;
+
+/**
+ * Placeholder for possible future features
+ */
+public interface TransactionStatus {
+}

http://git-wip-us.apache.org/repos/asf/cloudstack/blob/f62e28c1/framework/db/src/com/cloud/utils/db/TransactionWrappedExeception.java
----------------------------------------------------------------------
diff --git a/framework/db/src/com/cloud/utils/db/TransactionWrappedExeception.java b/framework/db/src/com/cloud/utils/db/TransactionWrappedExeception.java
new file mode 100644
index 0000000..70d4c16
--- /dev/null
+++ b/framework/db/src/com/cloud/utils/db/TransactionWrappedExeception.java
@@ -0,0 +1,36 @@
+/*
+ * 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 com.cloud.utils.db;
+
+import com.cloud.utils.exception.CloudRuntimeException;
+
+public class TransactionWrappedExeception extends CloudRuntimeException {
+
+    private static final long serialVersionUID = -3254037624055143300L;
+
+    Exception e;
+
+    public TransactionWrappedExeception(Exception e) {
+        this.e = e;
+    }
+
+    public Exception getWrapped() {
+        return e;
+    }
+}

http://git-wip-us.apache.org/repos/asf/cloudstack/blob/f62e28c1/framework/db/test/com/cloud/utils/db/DbTestDao.java
----------------------------------------------------------------------
diff --git a/framework/db/test/com/cloud/utils/db/DbTestDao.java b/framework/db/test/com/cloud/utils/db/DbTestDao.java
index 9530b3b..7db5ba8 100644
--- a/framework/db/test/com/cloud/utils/db/DbTestDao.java
+++ b/framework/db/test/com/cloud/utils/db/DbTestDao.java
@@ -29,7 +29,7 @@ public class DbTestDao extends GenericDaoBase<DbTestVO, Long> implements Generic
 
     @DB
     public void create(int fldInt, long fldLong, String fldString) {
-        Transaction txn = Transaction.currentTxn();
+        TransactionLegacy txn = TransactionLegacy.currentTxn();
         PreparedStatement pstmt = null;
         try {
             txn.start();
@@ -48,7 +48,7 @@ public class DbTestDao extends GenericDaoBase<DbTestVO, Long> implements Generic
 
     @DB
     public void update(int fldInt, long fldLong, String fldString) {
-        Transaction txn = Transaction.currentTxn();
+        TransactionLegacy txn = TransactionLegacy.currentTxn();
         PreparedStatement pstmt = null;
         try {
             txn.start();

http://git-wip-us.apache.org/repos/asf/cloudstack/blob/f62e28c1/framework/db/test/com/cloud/utils/db/DbTestUtils.java
----------------------------------------------------------------------
diff --git a/framework/db/test/com/cloud/utils/db/DbTestUtils.java b/framework/db/test/com/cloud/utils/db/DbTestUtils.java
index 11ae1aa..2458b8c 100644
--- a/framework/db/test/com/cloud/utils/db/DbTestUtils.java
+++ b/framework/db/test/com/cloud/utils/db/DbTestUtils.java
@@ -33,7 +33,7 @@ public class DbTestUtils {
             throw new RuntimeException("Unable to clean the database because I can't find " + file);
         }
         
-        Connection conn = Transaction.getStandaloneConnection();
+        Connection conn = TransactionLegacy.getStandaloneConnection();
         
         ScriptRunner runner = new ScriptRunner(conn, autoCommit, stopOnError);
         FileReader reader;
@@ -63,7 +63,7 @@ public class DbTestUtils {
             throw new RuntimeException("Unable to clean the database because I can't find " + file);
         }
         
-        Connection conn = Transaction.getStandaloneUsageConnection();
+        Connection conn = TransactionLegacy.getStandaloneUsageConnection();
         
         ScriptRunner runner = new ScriptRunner(conn, autoCommit, stopOnError);
         FileReader reader;

http://git-wip-us.apache.org/repos/asf/cloudstack/blob/f62e28c1/framework/db/test/com/cloud/utils/db/TransactionTest.java
----------------------------------------------------------------------
diff --git a/framework/db/test/com/cloud/utils/db/TransactionTest.java b/framework/db/test/com/cloud/utils/db/TransactionTest.java
index 101a533..92b2f36 100644
--- a/framework/db/test/com/cloud/utils/db/TransactionTest.java
+++ b/framework/db/test/com/cloud/utils/db/TransactionTest.java
@@ -41,7 +41,7 @@ public class TransactionTest {
         Connection conn = null;
         PreparedStatement pstmt = null;
         try {
-            conn = Transaction.getStandaloneConnection();
+            conn = TransactionLegacy.getStandaloneConnection();
 
             pstmt = conn.prepareStatement("CREATE TABLE `cloud`.`test` ("
                     + "`id` bigint unsigned NOT NULL UNIQUE AUTO_INCREMENT," + "`fld_int` int unsigned,"
@@ -75,27 +75,27 @@ public class TransactionTest {
      */
     public void testUserManagedConnection() {
         DbTestDao testDao = ComponentContext.inject(DbTestDao.class);
-        Transaction txn = Transaction.open("SingleConnectionThread");
+        TransactionLegacy txn = TransactionLegacy.open("SingleConnectionThread");
         Connection conn = null;
         try {
-            conn = Transaction.getStandaloneConnectionWithException();
+            conn = TransactionLegacy.getStandaloneConnectionWithException();
             txn.transitToUserManagedConnection(conn);
             // try two SQLs to make sure that they are using the same connection
             // acquired above.
             testDao.create(1, 1, "Record 1");
-            Connection checkConn = Transaction.currentTxn().getConnection();
+            Connection checkConn = TransactionLegacy.currentTxn().getConnection();
             if (checkConn != conn) {
                 Assert.fail("A new db connection is acquired instead of using old one after create sql");
             }
             testDao.update(2, 2, "Record 1");
-            Connection checkConn2 = Transaction.currentTxn().getConnection();
+            Connection checkConn2 = TransactionLegacy.currentTxn().getConnection();
             if (checkConn2 != conn) {
                 Assert.fail("A new db connection is acquired instead of using old one after update sql");
             }
         } catch (SQLException e) {
             Assert.fail(e.getMessage());
         } finally {
-            txn.transitToAutoManagedConnection(Transaction.CLOUD_DB);
+            txn.transitToAutoManagedConnection(TransactionLegacy.CLOUD_DB);
             txn.close();
 
             if (conn != null) {
@@ -117,28 +117,28 @@ public class TransactionTest {
         // acquire a db connection and keep it
         Connection conn = null;
         try {
-            conn = Transaction.getStandaloneConnectionWithException();
+            conn = TransactionLegacy.getStandaloneConnectionWithException();
         } catch (SQLException ex) {
             throw new CloudRuntimeException("Problem with getting db connection", ex);
         }
 
         // start heartbeat loop, make sure that each loop still use the same
         // connection
-        Transaction txn = null;
+        TransactionLegacy txn = null;
         for (int i = 0; i < 3; i++) {
-            txn = Transaction.open("HeartbeatSimulator");
+            txn = TransactionLegacy.open("HeartbeatSimulator");
             try {
 
                 txn.transitToUserManagedConnection(conn);
                 testDao.create(i, i, "Record " + i);
-                Connection checkConn = Transaction.currentTxn().getConnection();
+                Connection checkConn = TransactionLegacy.currentTxn().getConnection();
                 if (checkConn != conn) {
                     Assert.fail("A new db connection is acquired instead of using old one in loop " + i);
                 }
             } catch (SQLException e) {
                 Assert.fail(e.getMessage());
             } finally {
-                txn.transitToAutoManagedConnection(Transaction.CLOUD_DB);
+                txn.transitToAutoManagedConnection(TransactionLegacy.CLOUD_DB);
                 txn.close();
             }
         }
@@ -161,7 +161,7 @@ public class TransactionTest {
         Connection conn = null;
         PreparedStatement pstmt = null;
         try {
-            conn = Transaction.getStandaloneConnection();
+            conn = TransactionLegacy.getStandaloneConnection();
 
             pstmt = conn.prepareStatement("truncate table `cloud`.`test`");
             pstmt.execute();
@@ -189,7 +189,7 @@ public class TransactionTest {
         Connection conn = null;
         PreparedStatement pstmt = null;
         try {
-            conn = Transaction.getStandaloneConnection();
+            conn = TransactionLegacy.getStandaloneConnection();
 
             pstmt = conn.prepareStatement("DROP TABLE IF EXISTS `cloud`.`test`");
             pstmt.execute();

http://git-wip-us.apache.org/repos/asf/cloudstack/blob/f62e28c1/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobDaoImpl.java
----------------------------------------------------------------------
diff --git a/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobDaoImpl.java b/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobDaoImpl.java
index fb3845c..ed161e7 100644
--- a/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobDaoImpl.java
+++ b/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobDaoImpl.java
@@ -32,7 +32,7 @@ import com.cloud.utils.db.GenericDaoBase;
 import com.cloud.utils.db.SearchBuilder;
 import com.cloud.utils.db.SearchCriteria;
 import com.cloud.utils.db.SearchCriteria.Op;
-import com.cloud.utils.db.Transaction;
+import com.cloud.utils.db.TransactionLegacy;
 
 public class AsyncJobDaoImpl extends GenericDaoBase<AsyncJobVO, Long> implements AsyncJobDao {
     private static final Logger s_logger = Logger.getLogger(AsyncJobDaoImpl.class.getName());
@@ -182,7 +182,7 @@ public class AsyncJobDaoImpl extends GenericDaoBase<AsyncJobVO, Long> implements
                 + ", job_result='" + jobResultMessage + "' where job_status=" + JobInfo.Status.IN_PROGRESS.ordinal()
                 + " AND (job_executing_msid=? OR (job_executing_msid IS NULL AND job_init_msid=?))";
 		
-        Transaction txn = Transaction.currentTxn();
+        TransactionLegacy txn = TransactionLegacy.currentTxn();
         PreparedStatement pstmt = null;
         try {
             pstmt = txn.prepareAutoCloseStatement(sql);

http://git-wip-us.apache.org/repos/asf/cloudstack/blob/f62e28c1/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDaoImpl.java
----------------------------------------------------------------------
diff --git a/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDaoImpl.java b/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDaoImpl.java
index 20d8ba6..d4ca0d7 100644
--- a/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDaoImpl.java
+++ b/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDaoImpl.java
@@ -34,7 +34,7 @@ import com.cloud.utils.db.GenericDaoBase;
 import com.cloud.utils.db.SearchBuilder;
 import com.cloud.utils.db.SearchCriteria;
 import com.cloud.utils.db.SearchCriteria.Op;
-import com.cloud.utils.db.Transaction;
+import com.cloud.utils.db.TransactionLegacy;
 import com.cloud.utils.db.UpdateBuilder;
 import com.cloud.utils.exception.CloudRuntimeException;
 
@@ -157,7 +157,7 @@ public class AsyncJobJoinMapDaoImpl extends GenericDaoBase<AsyncJobJoinMapVO, Lo
 //
 //		Date cutDate = DateUtil.currentGMTTime();
 //
-//		Transaction txn = Transaction.currentTxn();
+//		TransactionLegacy txn = TransactionLegacy.currentTxn();
 //        PreparedStatement pstmt = null;
 //        try {
 //			txn.start();
@@ -213,7 +213,7 @@ public class AsyncJobJoinMapDaoImpl extends GenericDaoBase<AsyncJobJoinMapVO, Lo
     public List<Long> findJobsToWake(long joinedJobId) {
         // TODO: We should fix this.  We shouldn't be crossing daos in a dao code.
         List<Long> standaloneList = new ArrayList<Long>();
-        Transaction txn = Transaction.currentTxn();
+        TransactionLegacy txn = TransactionLegacy.currentTxn();
         String sql = "SELECT job_id FROM async_job_join_map WHERE join_job_id = ? AND job_id NOT IN (SELECT content_id FROM sync_queue_item)";
         try {
             PreparedStatement pstmt = txn.prepareStatement(sql);
@@ -231,7 +231,7 @@ public class AsyncJobJoinMapDaoImpl extends GenericDaoBase<AsyncJobJoinMapVO, Lo
     @Override
     public List<Long> findJobsToWakeBetween(Date cutDate) {
         List<Long> standaloneList = new ArrayList<Long>();
-        Transaction txn = Transaction.currentTxn();
+        TransactionLegacy txn = TransactionLegacy.currentTxn();
         try {
             String sql = "SELECT job_id FROM async_job_join_map WHERE next_wakeup < ? AND expiration > ? AND job_id NOT IN (SELECT content_id FROM sync_queue_item)";
             PreparedStatement pstmt = txn.prepareStatement(sql);
@@ -260,7 +260,7 @@ public class AsyncJobJoinMapDaoImpl extends GenericDaoBase<AsyncJobJoinMapVO, Lo
 //    public List<Long> wakeupByJoinedJobCompletion(long joinedJobId) {
 //        List<Long> standaloneList = new ArrayList<Long>();
 //
-//        Transaction txn = Transaction.currentTxn();
+//        TransactionLegacy txn = TransactionLegacy.currentTxn();
 //        PreparedStatement pstmt = null;
 //        try {
 //            txn.start();

http://git-wip-us.apache.org/repos/asf/cloudstack/blob/f62e28c1/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/SyncQueueDaoImpl.java
----------------------------------------------------------------------
diff --git a/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/SyncQueueDaoImpl.java b/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/SyncQueueDaoImpl.java
index f7d9d72..01efc4e 100644
--- a/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/SyncQueueDaoImpl.java
+++ b/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/SyncQueueDaoImpl.java
@@ -30,7 +30,7 @@ import com.cloud.utils.DateUtil;
 import com.cloud.utils.db.GenericDaoBase;
 import com.cloud.utils.db.SearchBuilder;
 import com.cloud.utils.db.SearchCriteria;
-import com.cloud.utils.db.Transaction;
+import com.cloud.utils.db.TransactionLegacy;
 
 public class SyncQueueDaoImpl extends GenericDaoBase<SyncQueueVO, Long> implements SyncQueueDao {
     private static final Logger s_logger = Logger.getLogger(SyncQueueDaoImpl.class.getName());
@@ -51,7 +51,7 @@ public class SyncQueueDaoImpl extends GenericDaoBase<SyncQueueVO, Long> implemen
         String sql = "INSERT IGNORE INTO sync_queue(sync_objtype, sync_objid, created, last_updated)" +
                 " values(?, ?, ?, ?)";
 		
-        Transaction txn = Transaction.currentTxn();
+        TransactionLegacy txn = TransactionLegacy.currentTxn();
         PreparedStatement pstmt = null;
         try {
             pstmt = txn.prepareAutoCloseStatement(sql);

http://git-wip-us.apache.org/repos/asf/cloudstack/blob/f62e28c1/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/SyncQueueItemDaoImpl.java
----------------------------------------------------------------------
diff --git a/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/SyncQueueItemDaoImpl.java b/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/SyncQueueItemDaoImpl.java
index 0cd231f..2f04a7c 100644
--- a/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/SyncQueueItemDaoImpl.java
+++ b/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/SyncQueueItemDaoImpl.java
@@ -37,7 +37,7 @@ import com.cloud.utils.db.GenericSearchBuilder;
 import com.cloud.utils.db.SearchBuilder;
 import com.cloud.utils.db.SearchCriteria;
 import com.cloud.utils.db.SearchCriteria.Op;
-import com.cloud.utils.db.Transaction;
+import com.cloud.utils.db.TransactionLegacy;
 
 @DB
 public class SyncQueueItemDaoImpl extends GenericDaoBase<SyncQueueItemVO, Long> implements SyncQueueItemDao {
@@ -83,7 +83,7 @@ public class SyncQueueItemDaoImpl extends GenericDaoBase<SyncQueueItemVO, Long>
 					 " ORDER BY i.id " +
 					 " LIMIT 0, ?";
 
-        Transaction txn = Transaction.currentTxn();
+        TransactionLegacy txn = TransactionLegacy.currentTxn();
         PreparedStatement pstmt = null;
         try {
             pstmt = txn.prepareAutoCloseStatement(sql);

http://git-wip-us.apache.org/repos/asf/cloudstack/blob/f62e28c1/framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
----------------------------------------------------------------------
diff --git a/framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java b/framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
index 93d50c1..ffc7b3a 100644
--- a/framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
+++ b/framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
@@ -70,8 +70,11 @@ import com.cloud.utils.db.GenericSearchBuilder;
 import com.cloud.utils.db.GlobalLock;
 import com.cloud.utils.db.SearchBuilder;
 import com.cloud.utils.db.SearchCriteria;
+import com.cloud.utils.db.TransactionCallback;
+import com.cloud.utils.db.TransactionCallbackNoReturn;
 import com.cloud.utils.db.SearchCriteria.Op;
 import com.cloud.utils.db.Transaction;
+import com.cloud.utils.db.TransactionStatus;
 import com.cloud.utils.exception.CloudRuntimeException;
 import com.cloud.utils.exception.ExceptionUtil;
 import com.cloud.utils.mgmt.JmxUtil;
@@ -177,19 +180,22 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
     @SuppressWarnings("unchecked")
     @Override
     @DB
-    public long submitAsyncJob(AsyncJob job, String syncObjType, long syncObjId) {
-        Transaction txt = Transaction.currentTxn();
+    public long submitAsyncJob(final AsyncJob job, final String syncObjType, final long syncObjId) {
         try {
             @SuppressWarnings("rawtypes")
-            GenericDao dao = GenericDaoBase.getDao(job.getClass());
+            final GenericDao dao = GenericDaoBase.getDao(job.getClass());
 
-            txt.start();
-            job.setInitMsid(getMsid());
-            dao.persist(job);
+            return Transaction.execute(new TransactionCallback<Long>() {
+                @Override
+                public Long doInTransaction(TransactionStatus status) {
+                    job.setInitMsid(getMsid());
+                    dao.persist(job);
+
+                    syncAsyncJobExecution(job, syncObjType, syncObjId, 1);
 
-            syncAsyncJobExecution(job, syncObjType, syncObjId, 1);
-            txt.commit();
-            return job.getId();
+                    return job.getId();
+                }
+            });
         } catch (Exception e) {
             String errMsg = "Unable to schedule async job for command " + job.getCmd() + ", unexpected exception.";
             s_logger.warn(errMsg, e);
@@ -199,123 +205,110 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
 
     @Override
     @DB
-    public void completeAsyncJob(long jobId, Status jobStatus, int resultCode, String resultObject) {
+    public void completeAsyncJob(final long jobId, final Status jobStatus, final int resultCode, final String resultObject) {
         if (s_logger.isDebugEnabled()) {
             s_logger.debug("Complete async job-" + jobId + ", jobStatus: " + jobStatus + ", resultCode: " + resultCode + ", result: " + resultObject);
         }
 
-        Transaction txn = Transaction.currentTxn();
-        try {
-            txn.start();
-            AsyncJobVO job = _jobDao.findById(jobId);
-            if (job == null) {
-                if (s_logger.isDebugEnabled()) {
-                    s_logger.debug("job-" + jobId + " no longer exists, we just log completion info here. " + jobStatus + ", resultCode: " + resultCode + ", result: " +
-                                   resultObject);
-                }
-
-                txn.rollback();
-                return;
+        final AsyncJobVO job = _jobDao.findById(jobId);
+        if (job == null) {
+            if (s_logger.isDebugEnabled()) {
+                s_logger.debug("job-" + jobId + " no longer exists, we just log completion info here. " + jobStatus + ", resultCode: " + resultCode + ", result: " +
+                               resultObject);
             }
 
-            if (job.getStatus() != JobInfo.Status.IN_PROGRESS) {
-                if (s_logger.isDebugEnabled()) {
-                    s_logger.debug("job-" + jobId + " is already completed.");
-                }
+            return;
+        }
 
-                txn.rollback();
-                return;
+        if (job.getStatus() != JobInfo.Status.IN_PROGRESS) {
+            if (s_logger.isDebugEnabled()) {
+                s_logger.debug("job-" + jobId + " is already completed.");
             }
 
-            job.setCompleteMsid(getMsid());
-            job.setStatus(jobStatus);
-            job.setResultCode(resultCode);
+            return;
+        }
 
-            // reset attached object
-            job.setInstanceType(null);
-            job.setInstanceId(null);
+        List<Long> wakeupList = Transaction.execute(new TransactionCallback<List<Long>>() {
+            @Override
+            public List<Long> doInTransaction(TransactionStatus status) {
+                job.setCompleteMsid(getMsid());
+                job.setStatus(jobStatus);
+                job.setResultCode(resultCode);
 
-            if (resultObject != null) {
-                job.setResult(resultObject);
-            }
+                // reset attached object
+                job.setInstanceType(null);
+                job.setInstanceId(null);
 
-            job.setLastUpdated(DateUtil.currentGMTTime());
-            _jobDao.update(jobId, job);
+                if (resultObject != null) {
+                    job.setResult(resultObject);
+                }
 
-            List<Long> wakeupList = wakeupByJoinedJobCompletion(jobId);
-            _joinMapDao.disjoinAllJobs(jobId);
+                job.setLastUpdated(DateUtil.currentGMTTime());
+                _jobDao.update(jobId, job);
 
-            txn.commit();
+                List<Long> wakeupList = wakeupByJoinedJobCompletion(jobId);
+                _joinMapDao.disjoinAllJobs(jobId);
 
-            for (Long id : wakeupList) {
-                // TODO, we assume that all jobs in this category is API job only
-                AsyncJobVO jobToWakeup = _jobDao.findById(id);
-                if (jobToWakeup != null && (jobToWakeup.getPendingSignals() & AsyncJob.Constants.SIGNAL_MASK_WAKEUP) != 0)
-                    scheduleExecution(jobToWakeup, false);
+                return wakeupList;
             }
+        });
 
-            _messageBus.publish(null, AsyncJob.Topics.JOB_STATE, PublishScope.GLOBAL, jobId);
-        } catch (Exception e) {
-            s_logger.error("Unexpected exception while completing async job-" + jobId, e);
-            txn.rollback();
+        for (Long id : wakeupList) {
+            // TODO, we assume that all jobs in this category is API job only
+            AsyncJobVO jobToWakeup = _jobDao.findById(id);
+            if (jobToWakeup != null && (jobToWakeup.getPendingSignals() & AsyncJob.Constants.SIGNAL_MASK_WAKEUP) != 0)
+                scheduleExecution(jobToWakeup, false);
         }
+
+        _messageBus.publish(null, AsyncJob.Topics.JOB_STATE, PublishScope.GLOBAL, jobId);
     }
 
     @Override
     @DB
-    public void updateAsyncJobStatus(long jobId, int processStatus, String resultObject) {
+    public void updateAsyncJobStatus(final long jobId, final int processStatus, final String resultObject) {
         if (s_logger.isDebugEnabled()) {
             s_logger.debug("Update async-job progress, job-" + jobId + ", processStatus: " + processStatus + ", result: " + resultObject);
         }
 
-        Transaction txt = Transaction.currentTxn();
-        try {
-            txt.start();
-            AsyncJobVO job = _jobDao.findById(jobId);
-            if (job == null) {
-                if (s_logger.isDebugEnabled()) {
-                    s_logger.debug("job-" + jobId + " no longer exists, we just log progress info here. progress status: " + processStatus);
-                }
-
-                txt.rollback();
-                return;
+        final AsyncJobVO job = _jobDao.findById(jobId);
+        if (job == null) {
+            if (s_logger.isDebugEnabled()) {
+                s_logger.debug("job-" + jobId + " no longer exists, we just log progress info here. progress status: " + processStatus);
             }
 
-            job.setProcessStatus(processStatus);
-            if (resultObject != null) {
-                job.setResult(resultObject);
-            }
-            job.setLastUpdated(DateUtil.currentGMTTime());
-            _jobDao.update(jobId, job);
-            txt.commit();
-        } catch (Exception e) {
-            s_logger.error("Unexpected exception while updating async job-" + jobId + " status: ", e);
-            txt.rollback();
+            return;
         }
+
+        Transaction.execute(new TransactionCallbackNoReturn() {
+            @Override
+            public void doInTransactionWithoutResult(TransactionStatus status) {
+                job.setProcessStatus(processStatus);
+                if (resultObject != null) {
+                    job.setResult(resultObject);
+                }
+                job.setLastUpdated(DateUtil.currentGMTTime());
+                _jobDao.update(jobId, job);
+            }
+        });
     }
 
     @Override
     @DB
-    public void updateAsyncJobAttachment(long jobId, String instanceType, Long instanceId) {
+    public void updateAsyncJobAttachment(final long jobId, final String instanceType, final Long instanceId) {
         if (s_logger.isDebugEnabled()) {
             s_logger.debug("Update async-job attachment, job-" + jobId + ", instanceType: " + instanceType + ", instanceId: " + instanceId);
         }
 
-        Transaction txt = Transaction.currentTxn();
-        try {
-            txt.start();
-
-            AsyncJobVO job = _jobDao.createForUpdate();
-            job.setInstanceType(instanceType);
-            job.setInstanceId(instanceId);
-            job.setLastUpdated(DateUtil.currentGMTTime());
-            _jobDao.update(jobId, job);
-
-            txt.commit();
-        } catch (Exception e) {
-            s_logger.error("Unexpected exception while updating async job-" + jobId + " attachment: ", e);
-            txt.rollback();
-        }
+        Transaction.execute(new TransactionCallbackNoReturn() {
+            @Override
+            public void doInTransactionWithoutResult(TransactionStatus status) {
+                AsyncJobVO job = _jobDao.createForUpdate();
+                job.setInstanceType(instanceType);
+                job.setInstanceId(instanceId);
+                job.setLastUpdated(DateUtil.currentGMTTime());
+                _jobDao.update(jobId, job);
+            }
+        });
     }
 
     @Override
@@ -493,15 +486,12 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
         return new ManagedContextRunnable() {
             @Override
             protected void runInContext() {
-                Transaction txn = null;
                 long runNumber = getJobRunNumber();
 
                 try {
                     //
                     // setup execution environment
                     //
-                    txn = Transaction.open(Transaction.CLOUD_DB);
-
                     try {
                         JmxUtil.registerMBean("AsyncJobManager", "Active Job " + job.getId(), new AsyncJobMBeanImpl(job));
                     } catch (Exception e) {
@@ -564,9 +554,6 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
                                 s_logger.trace("Unable to unregister job " + job.getId() + " to JMX monitoring due to exception " + ExceptionUtil.toString(e));
                         }
 
-                        if (txn != null)
-                            txn.close();
-
                         //
                         // clean execution environment
                         //
@@ -690,7 +677,6 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
         return new ManagedContextRunnable() {
             @Override
             protected void runInContext() {
-                Transaction txn = Transaction.open("AsyncJobManagerImpl.getHeartbeatTask");
                 try {
                     List<SyncQueueItemVO> l = _queueMgr.dequeueFromAny(getMsid(), MAX_ONETIME_SCHEDULE_SIZE);
                     if (l != null && l.size() > 0) {
@@ -711,12 +697,6 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
                     }
                 } catch (Throwable e) {
                     s_logger.error("Unexpected exception when trying to execute queue item, ", e);
-                } finally {
-                    try {
-                        txn.close();
-                    } catch (Throwable e) {
-                        s_logger.error("Unexpected exception", e);
-                    }
                 }
             }
         };
@@ -785,13 +765,15 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
     }
 
     @DB
-    protected void expungeAsyncJob(AsyncJobVO job) {
-        Transaction txn = Transaction.currentTxn();
-        txn.start();
-        _jobDao.expunge(job.getId());
-        //purge corresponding sync queue item
-        _queueMgr.purgeAsyncJobQueueItemId(job.getId());
-        txn.commit();
+    protected void expungeAsyncJob(final AsyncJobVO job) {
+        Transaction.execute(new TransactionCallbackNoReturn() {
+            @Override
+            public void doInTransactionWithoutResult(TransactionStatus status) {
+                _jobDao.expunge(job.getId());
+                //purge corresponding sync queue item
+                _queueMgr.purgeAsyncJobQueueItemId(job.getId());
+            }
+        });
     }
 
     private long getMsid() {
@@ -825,58 +807,60 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
             Collections.sort(result);
             Long[] ids = result.toArray(new Long[result.size()]);
 
-            SearchCriteria<AsyncJobVO> jobsSC = JobIdsSearch.create("ids", ids);
-            SearchCriteria<SyncQueueItemVO> queueItemsSC = QueueJobIdsSearch.create("contentIds", ids);
+            final SearchCriteria<AsyncJobVO> jobsSC = JobIdsSearch.create("ids", ids);
+            final SearchCriteria<SyncQueueItemVO> queueItemsSC = QueueJobIdsSearch.create("contentIds", ids);
 
-            Transaction txn = Transaction.currentTxn();
-            txn.start();
-            AsyncJobVO job = _jobDao.createForUpdate();
-            job.setPendingSignals(AsyncJob.Constants.SIGNAL_MASK_WAKEUP);
-            _jobDao.update(job, jobsSC);
+            Transaction.execute(new TransactionCallbackNoReturn() {
+                @Override
+                public void doInTransactionWithoutResult(TransactionStatus status) {
+                    AsyncJobVO job = _jobDao.createForUpdate();
+                    job.setPendingSignals(AsyncJob.Constants.SIGNAL_MASK_WAKEUP);
+                    _jobDao.update(job, jobsSC);
 
-            SyncQueueItemVO item = _queueItemDao.createForUpdate();
-            item.setLastProcessNumber(null);
-            item.setLastProcessMsid(null);
-            _queueItemDao.update(item, queueItemsSC);
-            txn.commit();
+                    SyncQueueItemVO item = _queueItemDao.createForUpdate();
+                    item.setLastProcessNumber(null);
+                    item.setLastProcessMsid(null);
+                    _queueItemDao.update(item, queueItemsSC);
+                }
+            });
         }
         return _joinMapDao.findJobsToWake(joinedJobId);
     }
 
     @DB
     protected List<Long> wakeupScan() {
-        Date cutDate = DateUtil.currentGMTTime();
-        Transaction txn = Transaction.currentTxn();
+        final Date cutDate = DateUtil.currentGMTTime();
 
         SearchCriteria<Long> sc = JoinJobTimeSearch.create();
         sc.setParameters("beginTime", cutDate);
         sc.setParameters("endTime", cutDate);
 
-        List<Long> result = _joinMapDao.customSearch(sc, null);
-
-        txn.start();
-        if (result.size() > 0) {
-            Collections.sort(result);
-            Long[] ids = result.toArray(new Long[result.size()]);
+        final List<Long> result = _joinMapDao.customSearch(sc, null);
 
-            AsyncJobVO job = _jobDao.createForUpdate();
-            job.setPendingSignals(AsyncJob.Constants.SIGNAL_MASK_WAKEUP);
+        return Transaction.execute(new TransactionCallback<List<Long>>() {
+            @Override
+            public List<Long> doInTransaction(TransactionStatus status) {
+                if (result.size() > 0) {
+                    Collections.sort(result);
+                    Long[] ids = result.toArray(new Long[result.size()]);
 
-            SearchCriteria<AsyncJobVO> sc2 = JobIdsSearch.create("ids", ids);
-            SearchCriteria<SyncQueueItemVO> queueItemsSC = QueueJobIdsSearch.create("contentIds", ids);
+                    AsyncJobVO job = _jobDao.createForUpdate();
+                    job.setPendingSignals(AsyncJob.Constants.SIGNAL_MASK_WAKEUP);
 
-            _jobDao.update(job, sc2);
+                    SearchCriteria<AsyncJobVO> sc2 = JobIdsSearch.create("ids", ids);
+                    SearchCriteria<SyncQueueItemVO> queueItemsSC = QueueJobIdsSearch.create("contentIds", ids);
 
-            SyncQueueItemVO item = _queueItemDao.createForUpdate();
-            item.setLastProcessNumber(null);
-            item.setLastProcessMsid(null);
-            _queueItemDao.update(item, queueItemsSC);
-        }
+                    _jobDao.update(job, sc2);
 
-        List<Long> wakupIds = _joinMapDao.findJobsToWakeBetween(cutDate);
-        txn.commit();
+                    SyncQueueItemVO item = _queueItemDao.createForUpdate();
+                    item.setLastProcessNumber(null);
+                    item.setLastProcessMsid(null);
+                    _queueItemDao.update(item, queueItemsSC);
+                }
 
-        return wakupIds;
+                return _joinMapDao.findJobsToWakeBetween(cutDate);
+            }
+        });
     }
 
     @Override
@@ -933,18 +917,18 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
 
     @Override
     public void onManagementNodeLeft(List<? extends ManagementServerHost> nodeList, long selfNodeId) {
-        for (ManagementServerHost msHost : nodeList) {
-            Transaction txn = Transaction.open(Transaction.CLOUD_DB);
+        for (final ManagementServerHost msHost : nodeList) {
             try {
-                txn.start();
-                List<SyncQueueItemVO> items = _queueMgr.getActiveQueueItems(msHost.getId(), true);
-                cleanupPendingJobs(items);
-                _jobDao.resetJobProcess(msHost.getId(), ApiErrorCode.INTERNAL_ERROR.getHttpCode(), "job cancelled because of management server restart");
-                txn.commit();
+                Transaction.execute(new TransactionCallbackNoReturn() {
+                    @Override
+                    public void doInTransactionWithoutResult(TransactionStatus status) {
+                        List<SyncQueueItemVO> items = _queueMgr.getActiveQueueItems(msHost.getId(), true);
+                        cleanupPendingJobs(items);
+                        _jobDao.resetJobProcess(msHost.getId(), ApiErrorCode.INTERNAL_ERROR.getHttpCode(), "job cancelled because of management server restart");
+                    }
+                });
             } catch (Throwable e) {
                 s_logger.warn("Unexpected exception ", e);
-            } finally {
-                txn.close();
             }
         }
     }


Mime
View raw message