You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@labs.apache.org by pp...@apache.org on 2008/01/09 00:46:41 UTC
svn commit: r610232 - in
/labs/fluid/slice/src/main/java/org/apache/openjpa/slice:
jdbc/DistributedStoreManager.java jdbc/DistributedTransactionManager.java
jdbc/XID.java transaction/ transaction/DistributedTransactionManager.java
transaction/XID.java
Author: ppoddar
Date: Tue Jan 8 15:46:33 2008
New Revision: 610232
URL: http://svn.apache.org/viewvc?rev=610232&view=rev
Log:
separate transaction management from store management
Added:
labs/fluid/slice/src/main/java/org/apache/openjpa/slice/transaction/
labs/fluid/slice/src/main/java/org/apache/openjpa/slice/transaction/DistributedTransactionManager.java
- copied, changed from r608351, labs/fluid/slice/src/main/java/org/apache/openjpa/slice/jdbc/DistributedTransactionManager.java
labs/fluid/slice/src/main/java/org/apache/openjpa/slice/transaction/XID.java
- copied, changed from r608351, labs/fluid/slice/src/main/java/org/apache/openjpa/slice/jdbc/XID.java
Removed:
labs/fluid/slice/src/main/java/org/apache/openjpa/slice/jdbc/DistributedTransactionManager.java
labs/fluid/slice/src/main/java/org/apache/openjpa/slice/jdbc/XID.java
Modified:
labs/fluid/slice/src/main/java/org/apache/openjpa/slice/jdbc/DistributedStoreManager.java
Modified: labs/fluid/slice/src/main/java/org/apache/openjpa/slice/jdbc/DistributedStoreManager.java
URL: http://svn.apache.org/viewvc/labs/fluid/slice/src/main/java/org/apache/openjpa/slice/jdbc/DistributedStoreManager.java?rev=610232&r1=610231&r2=610232&view=diff
==============================================================================
--- labs/fluid/slice/src/main/java/org/apache/openjpa/slice/jdbc/DistributedStoreManager.java (original)
+++ labs/fluid/slice/src/main/java/org/apache/openjpa/slice/jdbc/DistributedStoreManager.java Tue Jan 8 15:46:33 2008
@@ -18,26 +18,21 @@
*/
package org.apache.openjpa.slice.jdbc;
-import java.sql.Connection;
-import java.sql.SQLException;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collection;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
-import javax.sql.XAConnection;
-import javax.transaction.xa.XAException;
-import javax.transaction.xa.XAResource;
+import javax.transaction.Transaction;
+import javax.transaction.TransactionManager;
import org.apache.openjpa.conf.OpenJPAConfiguration;
import org.apache.openjpa.enhance.PersistenceCapable;
@@ -63,6 +58,9 @@
import org.apache.openjpa.meta.ClassMetaData;
import org.apache.openjpa.meta.FieldMetaData;
import org.apache.openjpa.slice.DistributionPolicy;
+import org.apache.openjpa.slice.transaction.DistributedNaiveTransaction;
+import org.apache.openjpa.slice.transaction.DistributedTransactionManager;
+import org.apache.openjpa.slice.transaction.NaiveTransactionManager;
import org.apache.openjpa.util.InternalException;
import org.apache.openjpa.util.StoreException;
import org.apache.openjpa.util.UserException;
@@ -81,7 +79,7 @@
private final List<SliceStoreManager> _slices;
private JDBCStoreManager _master;
private boolean isXA;
- private DistributedTransactionManager tm;
+ private TransactionManager _tm;
private final DistributedJDBCConfiguration _conf;
private static final Localizer _loc =
Localizer.forPackage(DistributedStoreManager.class);
@@ -193,24 +191,25 @@
}
public void begin() {
- if (isXA) {
- tm = new DistributedTransactionManager();
- for (SliceStoreManager slice : _slices) {
- try {
- tm.enlist(slice.getXAConnection().getXAResource());
- } catch (SQLException e) {
- throw new InternalException(e);
- }
- }
-
+ TransactionManager tm = getTransactionManager();
+ for (SliceStoreManager slice : _slices) {
try {
- tm.begin();
+ Transaction txn = tm.getTransaction();
+ if (isXA) {
+ txn.enlistResource(slice.getXAConnection().getXAResource());
+ } else { // This is the only place where casting to our
+ // internal implementation classes become necessary
+ ((DistributedNaiveTransaction) txn).enlistResource(slice);
+ }
} catch (Exception e) {
- throw new StoreException(e);
+ throw new InternalException(e);
}
- } else {
- for (SliceStoreManager child : _slices)
- child.begin();
+ }
+
+ try {
+ tm.begin();
+ } catch (Exception e) {
+ throw new StoreException(e);
}
}
@@ -237,16 +236,12 @@
}
public void commit() {
- if (isXA) {
- try {
- tm.commit();
- } catch (Exception e) {
- throw new StoreException(e);
- }
+ TransactionManager tm = getTransactionManager();
+ try {
+ tm.commit();
+ } catch (Exception e) {
+ throw new StoreException(e);
}
-
- for (SliceStoreManager slice : _slices)
- slice.commit();
}
public int compareVersion(OpenJPAStateManager sm, Object v1, Object v2) {
@@ -411,8 +406,12 @@
}
public void rollback() {
- for (SliceStoreManager slice : _slices)
- slice.rollback();
+ TransactionManager tm = getTransactionManager();
+ try {
+ tm.rollback();
+ } catch (Exception e) {
+ throw new StoreException(e);
+ }
}
public void rollbackOptimistic() {
@@ -444,6 +443,16 @@
return selectStore(sm, edata).syncVersion(sm, edata);
}
+ protected TransactionManager getTransactionManager() {
+ //TODO: This method requires to be smarter to get an appropriate
+ // transaction manager. Currently it uses either of the two native
+ // transaction manager based on XA-compliance of all the resources.
+ if (_tm == null)
+ _tm = (isXA) ? new DistributedTransactionManager()
+ : new NaiveTransactionManager();
+ return _tm;
+ }
+
private static class Flusher implements Callable<Collection> {
final SliceStoreManager store;
final Collection toFlush;
@@ -457,4 +466,5 @@
return store.flush(toFlush);
}
}
+
}
Copied: labs/fluid/slice/src/main/java/org/apache/openjpa/slice/transaction/DistributedTransactionManager.java (from r608351, labs/fluid/slice/src/main/java/org/apache/openjpa/slice/jdbc/DistributedTransactionManager.java)
URL: http://svn.apache.org/viewvc/labs/fluid/slice/src/main/java/org/apache/openjpa/slice/transaction/DistributedTransactionManager.java?p2=labs/fluid/slice/src/main/java/org/apache/openjpa/slice/transaction/DistributedTransactionManager.java&p1=labs/fluid/slice/src/main/java/org/apache/openjpa/slice/jdbc/DistributedTransactionManager.java&r1=608351&r2=610232&rev=610232&view=diff
==============================================================================
--- labs/fluid/slice/src/main/java/org/apache/openjpa/slice/jdbc/DistributedTransactionManager.java (original)
+++ labs/fluid/slice/src/main/java/org/apache/openjpa/slice/transaction/DistributedTransactionManager.java Tue Jan 8 15:46:33 2008
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.openjpa.slice.jdbc;
+package org.apache.openjpa.slice.transaction;
import static javax.transaction.xa.XAResource.TMJOIN;
import static javax.transaction.xa.XAResource.TMNOFLAGS;
@@ -36,57 +36,74 @@
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
+import org.apache.openjpa.lib.util.Localizer;
+
/**
- * A simple (naive?) implementation to support distributed transaction.
+ * A simple (naive?) implementation for distributed transaction across
+ * XA-complaint data sources.
* Assumes begin() and commit() being called on the same thread.
*
- * @author Pinaki Poddar
- *
+ * @author Pinaki Poddar
+ *
*/
public class DistributedTransactionManager implements TransactionManager {
- private Set<XAResource> _slices = new HashSet<XAResource>();
- private final ThreadLocal<XID> xids = new ThreadLocal<XID>();
-
- void enlist(XAResource slice) {
- _slices.add(slice);
- }
+ private final ThreadLocal<DistributedXATransaction> txns =
+ new ThreadLocal<DistributedXATransaction>();
+ private static final Localizer _loc =
+ Localizer.forPackage(DistributedTransactionManager.class);
public void begin() throws NotSupportedException, SystemException {
- XID xid = getXID(false);
+ DistributedXATransaction txn = getTransaction(false);
int i = 1;
- for (XAResource slice : _slices) {
+ Set<XAResource> resources = txn.getEnlistedResources();
+ for (XAResource resource : resources) {
try {
- XAResource existing = isSame(slice, _slices);
- XID branch = xid.branch("" + i++);
+ XAResource existing = isSame(resource, resources);
+ XID branch = txn.getXID().branch(i++);
int flag = (existing == null) ? TMNOFLAGS : TMJOIN;
- slice.start(branch, flag);
+ resource.start(branch, flag);
} catch (Exception e) {
throw new SystemException(e.toString());
}
}
- return;
}
public void commit() throws HeuristicMixedException,
HeuristicRollbackException, IllegalStateException,
RollbackException, SecurityException, SystemException {
- XID xid = getXID(true);
- int i = 1;
- for (XAResource slice : _slices) {
- XID branch = xid.branch("" + i++);
+ DistributedXATransaction txn = getTransaction(true);
+ Set<XAResource> resources = txn.getEnlistedResources();
+ int branchId = 1;
+ boolean nextPhase = true;
+ for (XAResource resource : resources) {
+ XID branch = txn.getXID().branch(branchId++);
try {
- slice.end(branch, TMSUCCESS);
- slice.prepare(branch);
+ resource.end(branch, TMSUCCESS);
+ resource.prepare(branch);
} catch (XAException e) {
- throw new SystemException(e.getMessage());
+ nextPhase = false;
+ }
+ }
+
+ branchId = 1; // reset
+ if (!nextPhase) {
+ for (XAResource resource : resources) {
+ try {
+ XID branch = txn.getXID().branch(branchId++);
+ resource.forget(branch);
+ } catch (XAException e) {
+ // ignore
+ }
+ throw new SystemException(_loc.get("prepare-failed")
+ .getMessage());
}
}
- i = 1;
- for (XAResource slice : _slices) {
- XID branch = xid.branch("" + i++);
+ branchId = 1; // reset
+ for (XAResource resource : resources) {
+ XID branch = txn.getXID().branch(branchId++);
try {
- slice.commit(branch, false);
+ resource.commit(branch, false);
} catch (XAException e) {
throw new SystemException(e.getMessage());
}
@@ -94,11 +111,11 @@
}
public int getStatus() throws SystemException {
- throw new UnsupportedOperationException();
+ return getTransaction().getStatus();
}
public Transaction getTransaction() throws SystemException {
- throw new UnsupportedOperationException();
+ return getTransaction(false);
}
public void resume(Transaction arg0) throws IllegalStateException,
@@ -108,11 +125,21 @@
public void rollback() throws IllegalStateException, SecurityException,
SystemException {
- throw new UnsupportedOperationException();
+ DistributedXATransaction txn = getTransaction(true);
+ Set<XAResource> slices = txn.getEnlistedResources();
+ int branchId = 1;
+ for (XAResource slice : slices) {
+ XID branch = txn.getXID().branch(branchId++);
+ try {
+ slice.end(branch, XAResource.TMFAIL);
+ slice.rollback(branch);
+ } catch (XAException e) {
+ }
+ }
}
public void setRollbackOnly() throws IllegalStateException, SystemException {
- throw new UnsupportedOperationException();
+ getTransaction().setRollbackOnly();
}
public void setTransactionTimeout(int arg0) throws SystemException {
@@ -139,18 +166,27 @@
+ Long.toHexString(System.identityHashCode(o));
}
- XID getXID(boolean mustExist) {
- XID xid = xids.get();
- if (xid == null) {
+ /**
+ * Gets the transaction associated with the current thread.
+ *
+ * @param mustExist if true, a transaction must be associated with the
+ * current thread a priori. If false, the current thread has no associated
+ * transaction, a new transaction is created with a global identifier
+ * and associated with the current thread.
+ */
+ DistributedXATransaction getTransaction(boolean mustExist) {
+ DistributedXATransaction txn = txns.get();
+ if (txn == null) {
if (mustExist)
- throw new RuntimeException("No XID with "
- + Thread.currentThread());
+ throw new IllegalStateException(_loc.get("no-txn-on-thread",
+ Thread.currentThread().getName()).getMessage());
byte[] global =
Long.toHexString(System.currentTimeMillis()).getBytes();
- xid = new XID(0, global, new byte[] { 0x1 });
- xids.set(xid);
+ XID xid = new XID(0, global, new byte[] { 0x1 });
+ txn = new DistributedXATransaction(xid, this);
+ txns.set(txn);
}
- return xid;
+ return txn;
}
}
Copied: labs/fluid/slice/src/main/java/org/apache/openjpa/slice/transaction/XID.java (from r608351, labs/fluid/slice/src/main/java/org/apache/openjpa/slice/jdbc/XID.java)
URL: http://svn.apache.org/viewvc/labs/fluid/slice/src/main/java/org/apache/openjpa/slice/transaction/XID.java?p2=labs/fluid/slice/src/main/java/org/apache/openjpa/slice/transaction/XID.java&p1=labs/fluid/slice/src/main/java/org/apache/openjpa/slice/jdbc/XID.java&r1=608351&r2=610232&rev=610232&view=diff
==============================================================================
--- labs/fluid/slice/src/main/java/org/apache/openjpa/slice/jdbc/XID.java (original)
+++ labs/fluid/slice/src/main/java/org/apache/openjpa/slice/transaction/XID.java Tue Jan 8 15:46:33 2008
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.openjpa.slice.jdbc;
+package org.apache.openjpa.slice.transaction;
import javax.transaction.xa.Xid;
@@ -51,6 +51,10 @@
return global;
}
+ XID branch(Number number) {
+ return branch((number == null) ? "null" : number.toString());
+ }
+
XID branch(String branch) {
return new XID(format, global, branch.getBytes());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@labs.apache.org
For additional commands, e-mail: commits-help@labs.apache.org