antville/helma
antville/helma
1
1
Fork 0
Code Issues 5 Pull requests 15 Projects Releases 5 Activity Actions 1

* Merge -r 8778:8790 from branches/refactor_transactor.

Browse source
This commit is contained in:
hns 2008-02-08 15:58:14 +00:00
parent a44af35c59
commit 58b7de53c4
12 changed files with 393 additions and 145 deletions
Download patch file Download diff file Expand all files Collapse all files

5
src/helma/objectmodel/db/DbSource.java
View file

@ -69,9 +69,8 @@ public class DbSource {
public synchronized Connection getConnection()
throws ClassNotFoundException, SQLException {
Connection con;
Transactor tx = null;
if (Thread.currentThread() instanceof Transactor) {
tx = (Transactor) Thread.currentThread();
Transactor tx = Transactor.getInstance();
if (tx != null) {
con = tx.getConnection(this);
} else {
con = getThreadLocalConnection();

35
src/helma/objectmodel/db/Node.java
View file

@ -266,9 +266,9 @@ public final class Node implements INode, Serializable {
return; // no need to lock transient node
}
Transactor current = (Transactor) Thread.currentThread();
Transactor tx = Transactor.getInstanceOrFail();
if (!current.isActive()) {
if (!tx.isActive()) {
throw new helma.framework.TimeoutException();
}
@ -279,14 +279,14 @@ public final class Node implements INode, Serializable {
" was invalidated by another thread.");
}
if ((lock != null) && (lock != current) && lock.isAlive() && lock.isActive()) {
if ((lock != null) && (lock != tx) && lock.isAlive() && lock.isActive()) {
// nmgr.logEvent("Concurrency conflict for " + this + ", lock held by " + lock);
throw new ConcurrencyException("Tried to modify " + this +
" from two threads at the same time.");
}
current.visitDirtyNode(this);
lock = current;
tx.visitDirtyNode(this);
lock = tx;
}
/**
@ -306,9 +306,8 @@ public final class Node implements INode, Serializable {
state = s;
if (Thread.currentThread() instanceof Transactor) {
Transactor tx = (Transactor) Thread.currentThread();
Transactor tx = Transactor.getInstance();
if (tx != null) {
if (s == CLEAN) {
clearWriteLock();
tx.dropDirtyNode(this);
@ -332,9 +331,11 @@ public final class Node implements INode, Serializable {
// the process of being persistified - except if "manual" subnoderelation is set.
if ((state == TRANSIENT || state == NEW) && subnodeRelation == null) {
return;
} else if (Thread.currentThread() instanceof Transactor) {
Transactor tx = (Transactor) Thread.currentThread();
tx.visitParentNode(this);
} else {
Transactor tx = Transactor.getInstance();
if (tx != null) {
tx.visitParentNode(this);
}
}
}
@ -934,7 +935,7 @@ public final class Node implements INode, Serializable {
}
if (state != TRANSIENT) {
Transactor tx = (Transactor) Thread.currentThread();
Transactor tx = Transactor.getInstanceOrFail();
SyntheticKey key = new SyntheticKey(this.getKey(), prop);
tx.visitCleanNode(key, node);
nmgr.registerNode(node, key);
@ -1250,7 +1251,7 @@ public final class Node implements INode, Serializable {
// nodemanager. Otherwise, we just evict whatever was there before
if (create) {
// register group node with transactor
Transactor tx = (Transactor) Thread.currentThread();
Transactor tx = Transactor.getInstanceOrFail();
tx.visitCleanNode(node);
nmgr.registerNode(node);
} else {
@ -2387,7 +2388,7 @@ public final class Node implements INode, Serializable {
// this is done anyway when the node becomes persistent.
if (n.state != TRANSIENT) {
// check node in with transactor cache
Transactor tx = (Transactor) Thread.currentThread();
Transactor tx = Transactor.getInstanceOrFail();
// tx.visitCleanNode (new DbKey (dbm, nID), n);
// UPDATE: using n.getKey() instead of manually constructing key. HW 2002/09/13
@ -2539,9 +2540,9 @@ public final class Node implements INode, Serializable {
getHandle().becomePersistent();
// register node with the transactor
Transactor current = (Transactor) Thread.currentThread();
current.visitDirtyNode(this);
current.visitCleanNode(this);
Transactor tx = Transactor.getInstanceOrFail();
tx.visitDirtyNode(this);
tx.visitCleanNode(this);
// recursively make children persistable
makeChildrenPersistable();

30
src/helma/objectmodel/db/NodeManager.java
View file

@ -149,7 +149,7 @@ public final class NodeManager {
public void deleteNode(Node node) throws Exception {
if (node != null) {
synchronized (this) {
Transactor tx = (Transactor) Thread.currentThread();
Transactor tx = Transactor.getInstanceOrFail();
node.setState(Node.INVALID);
deleteNode(db, tx.txn, node);
@ -162,7 +162,7 @@ public final class NodeManager {
* a reference to another node via a NodeHandle/Key.
*/
public Node getNode(Key key) throws Exception {
Transactor tx = (Transactor) Thread.currentThread();
Transactor tx = Transactor.getInstanceOrFail();
// See if Transactor has already come across this node
Node node = tx.getCleanNode(key);
@ -212,7 +212,7 @@ public final class NodeManager {
return null;
}
Transactor tx = (Transactor) Thread.currentThread();
Transactor tx = Transactor.getInstanceOrFail();
Key key;
DbMapping otherDbm = rel == null ? null : rel.otherType;
@ -408,8 +408,9 @@ public final class NodeManager {
public void evictKey(Key key) {
cache.remove(key);
// also drop key from thread-local transactor cache
if (Thread.currentThread() instanceof Transactor) {
((Transactor) Thread.currentThread()).dropCleanNode(key);
Transactor tx = Transactor.getInstance();
if (tx != null) {
tx.dropCleanNode(key);
}
}
@ -1856,14 +1857,17 @@ public final class NodeManager {
if (id == null) {
return null;
} else if (Thread.currentThread() instanceof Transactor) {
// Check if the node is already registered with the transactor -
// it may be in the process of being DELETED, but do return the
// new node if the old one has been marked as INVALID.
DbKey key = new DbKey(dbmap, id);
Node dirtyNode = ((Transactor) Thread.currentThread()).getDirtyNode(key);
if (dirtyNode != null && dirtyNode.getState() != Node.INVALID) {
return dirtyNode;
} else {
Transactor tx = Transactor.getInstance();
if (tx != null) {
// Check if the node is already registered with the transactor -
// it may be in the process of being DELETED, but do return the
// new node if the old one has been marked as INVALID.
DbKey key = new DbKey(dbmap, id);
Node dirtyNode = tx.getDirtyNode(key);
if (dirtyNode != null && dirtyNode.getState() != Node.INVALID) {
return dirtyNode;
}
}
}

74
src/helma/objectmodel/db/Transactor.java
View file

@ -28,7 +28,7 @@ import java.util.*;
* A subclass of thread that keeps track of changed nodes and triggers
* changes in the database when a transaction is commited.
*/
public class Transactor extends Thread {
public class Transactor {
// The associated node manager
NodeManager nmgr;
@ -61,15 +61,18 @@ public class Transactor extends Thread {
// a name to log the transaction. For HTTP transactions this is the rerquest path
private String tname;
// the thread we're associated with
private Thread thread;
private static final ThreadLocal <Transactor> txtor = new ThreadLocal <Transactor> ();
/**
* Creates a new Transactor object.
*
* @param runnable ...
* @param group ...
* @param nmgr ...
* @param nmgr the NodeManager used to fetch and persist nodes.
*/
public Transactor(Runnable runnable, ThreadGroup group, NodeManager nmgr) {
super(group, runnable, group.getName());
private Transactor(NodeManager nmgr) {
this.thread = Thread.currentThread();
this.nmgr = nmgr;
dirtyNodes = new HashMap();
@ -82,6 +85,41 @@ public class Transactor extends Thread {
killed = false;
}
/**
* Get the transactor for the current thread or null if none exists.
* @return the transactor associated with the current thread
*/
public static Transactor getInstance() {
return txtor.get();
}
/**
* Get the transactor for the current thread or throw a IllegalStateException if none exists.
* @return the transactor associated with the current thread
* @throws IllegalStateException if no transactor is associated with the current thread
*/
public static Transactor getInstanceOrFail() throws IllegalStateException {
Transactor tx = txtor.get();
if (tx == null)
throw new IllegalStateException("Operation requires a Transactor, " +
"but current thread does not have one.");
return tx;
}
/**
* Get the transactor for the current thread, creating a new one if none exists.
* @param nmgr the NodeManager used to create the transactor
* @return the transactor associated with the current thread
*/
public static Transactor getInstance(NodeManager nmgr) {
Transactor t = txtor.get();
if (t == null) {
t = new Transactor(nmgr);
txtor.set(t);
}
return t;
}
/**
* Mark a Node as modified/created/deleted during this transaction
*
@ -185,6 +223,15 @@ public class Transactor extends Thread {
return active;
}
/**
* Check whether the thread associated with this transactor is alive.
* This is a proxy to Thread.isAlive().
* @return true if the thread running this transactor is currently alive.
*/
public boolean isAlive() {
return thread != null && thread.isAlive();
}
/**
* Register a db connection with this transactor thread.
* @param src the db source
@ -430,28 +477,29 @@ public class Transactor extends Thread {
* Kill this transaction thread. Used as last measure only.
*/
public synchronized void kill() {
killed = true;
interrupt();
thread.interrupt();
// Interrupt the thread if it has not noticed the flag (e.g. because it is busy
// reading from a network socket).
if (isAlive()) {
interrupt();
if (thread.isAlive()) {
thread.interrupt();
try {
join(1000);
thread.join(1000);
} catch (InterruptedException ir) {
// interrupted by other thread
}
}
if (isAlive() && "true".equals(nmgr.app.getProperty("requestTimeoutStop"))) {
if (thread.isAlive() && "true".equals(nmgr.app.getProperty("requestTimeoutStop"))) {
// still running - check if we ought to stop() it
try {
Thread.sleep(2000);
if (isAlive()) {
if (thread.isAlive()) {
// thread is still running, pull emergency break
nmgr.app.logEvent("Stopping Thread for Transactor " + this);
stop();
thread.stop();
}
} catch (InterruptedException ir) {
// interrupted by other thread

76
src/helma/objectmodel/db/WrappedNodeManager.java
View file

@ -18,7 +18,6 @@ package helma.objectmodel.db;
import helma.objectmodel.ObjectNotFoundException;
import java.util.List;
import java.util.Vector;
/**
@ -56,11 +55,17 @@ public final class WrappedNodeManager {
* @return
*/
public Node getNode(Key key) {
Transactor tx = checkLocalTransactor();
try {
return nmgr.getNode(key);
beginLocalTransaction(tx, "getNode");
Node node = nmgr.getNode(key);
commitLocalTransaction(tx);
return node;
} catch (ObjectNotFoundException x) {
abortLocalTransaction(tx);
return null;
} catch (Exception x) {
abortLocalTransaction(tx);
nmgr.app.logError("Error retrieving Node for " + key, x);
throw new RuntimeException("Error retrieving Node", x);
}
@ -75,11 +80,17 @@ public final class WrappedNodeManager {
* @return
*/
public Node getNode(Node home, String id, Relation rel) {
Transactor tx = checkLocalTransactor();
try {
return nmgr.getNode(home, id, rel);
beginLocalTransaction(tx, "getNode");
Node node = nmgr.getNode(home, id, rel);
commitLocalTransaction(tx);
return node;
} catch (ObjectNotFoundException x) {
abortLocalTransaction(tx);
return null;
} catch (Exception x) {
abortLocalTransaction(tx);
nmgr.app.logError("Error retrieving Node \"" + id + "\" from " + home, x);
throw new RuntimeException("Error retrieving Node", x);
}
@ -94,9 +105,14 @@ public final class WrappedNodeManager {
* @return
*/
public SubnodeList getNodes(Node home, Relation rel) {
Transactor tx = checkLocalTransactor();
try {
return nmgr.getNodes(home, rel);
beginLocalTransaction(tx, "getNodes");
SubnodeList list = nmgr.getNodes(home, rel);
commitLocalTransaction(tx);
return list;
} catch (Exception x) {
abortLocalTransaction(tx);
throw new RuntimeException("Error retrieving Nodes", x);
}
}
@ -150,9 +166,13 @@ public final class WrappedNodeManager {
* @param node
*/
public void deleteNode(Node node) {
Transactor tx = checkLocalTransactor();
try {
beginLocalTransaction(tx, "deleteNode");
nmgr.deleteNode(node);
commitLocalTransaction(tx);
} catch (Exception x) {
abortLocalTransaction(tx);
throw new RuntimeException("Error deleting Node", x);
}
}
@ -236,9 +256,14 @@ public final class WrappedNodeManager {
* Gets the application's root node.
*/
public Node getRootNode() {
Transactor tx = checkLocalTransactor();
try {
return nmgr.getRootNode();
beginLocalTransaction(tx, "getRootNode");
Node node = nmgr.getRootNode();
commitLocalTransaction(tx);
return node;
} catch (Exception x) {
abortLocalTransaction(tx);
throw new RuntimeException(x.toString(), x);
}
}
@ -275,4 +300,45 @@ public final class WrappedNodeManager {
public DbMapping getDbMapping(String name) {
return nmgr.app.getDbMapping(name);
}
// helper methods to wrap execution inside local transactions
private Transactor checkLocalTransactor() {
Transactor tx = Transactor.getInstance();
if (tx != null) {
// transactor already associated with current thread - return null
return null;
}
return Transactor.getInstance(nmgr);
}
private void beginLocalTransaction(Transactor tx, String name) {
if (tx != null) {
try {
tx.begin(name);
} catch (Exception x) {
nmgr.app.logError("Error in beginLocalTransaction", x);
}
}
}
private void commitLocalTransaction(Transactor tx) {
if (tx != null) {
try {
tx.commit();
} catch (Exception x) {
nmgr.app.logError("Error in commitLocalTransaction", x);
}
}
}
private void abortLocalTransaction(Transactor tx) {
if (tx != null) {
try {
tx.abort();
} catch (Exception x) {
nmgr.app.logError("Error in abortLocalTransaction", x);
}
}
}
}