[jboss-cvs] JBossCache/src/org/jboss/cache/statetransfer ...
Vladmir Blagojevic
vladimir.blagojevic at jboss.com
Tue Sep 12 17:12:45 EDT 2006
User: vblagojevic
Date: 06/09/12 17:12:45
Modified: src/org/jboss/cache/statetransfer
StateTransferFactory.java
Added: src/org/jboss/cache/statetransfer
DefaultStateTransferGenerator.java
DefaultStateTransferIntegrator.java
Removed: src/org/jboss/cache/statetransfer
AbstractStateTransferGenerator.java
AbstractStateTransferIntegrator.java
Log:
renamed AbstractStateTransfer(Generator|Integrator) to DefaultStateTransfer(Generator|Integrator)
Revision Changes Path
1.12 +3 -3 JBossCache/src/org/jboss/cache/statetransfer/StateTransferFactory.java
(In the diff below, changes in quantity of whitespace are not shown.)
Index: StateTransferFactory.java
===================================================================
RCS file: /cvsroot/jboss/JBossCache/src/org/jboss/cache/statetransfer/StateTransferFactory.java,v
retrieving revision 1.11
retrieving revision 1.12
diff -u -b -r1.11 -r1.12
--- StateTransferFactory.java 11 Sep 2006 21:53:19 -0000 1.11
+++ StateTransferFactory.java 12 Sep 2006 21:12:45 -0000 1.12
@@ -18,7 +18,7 @@
* {@link StateTransferIntegrator} instances.
*
* @author <a href="brian.stansberry at jboss.com">Brian Stansberry</a>
- * @version $Revision: 1.11 $
+ * @version $Revision: 1.12 $
*/
public abstract class StateTransferFactory
{
@@ -43,7 +43,7 @@
if (version < RV_200 && version > 0) // <= 0 is actually a version > 15.31.63
throw new IllegalStateException("State transfer with cache replication version < 2.0.0 not supported");
else
- return new AbstractStateTransferGenerator(cache); // current default
+ return new DefaultStateTransferGenerator(cache); // current default
}
public static StateTransferIntegrator getStateTransferIntegrator(ObjectInputStream in, Fqn fqn, TreeCache treeCache)
@@ -70,7 +70,7 @@
if (version < RV_200 && version > 0) // <= 0 is actually a version > 15.31.63
throw new IllegalStateException("State transfer with cache replication version < 2.0.0 not supported");
else
- return new AbstractStateTransferIntegrator(fqn, treeCache); // current default
+ return new DefaultStateTransferIntegrator(fqn, treeCache); // current default
}
}
1.1 date: 2006/09/12 21:12:45; author: vblagojevic; state: Exp;JBossCache/src/org/jboss/cache/statetransfer/DefaultStateTransferGenerator.java
Index: DefaultStateTransferGenerator.java
===================================================================
/*
* JBoss, the OpenSource J2EE webOS
*
* Distributable under LGPL license.
* See terms of license at gnu.org.
*/
package org.jboss.cache.statetransfer;
import java.io.ObjectOutputStream;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.cache.DataNode;
import org.jboss.cache.Fqn;
import org.jboss.cache.TreeCache;
import org.jboss.cache.Version;
import org.jboss.cache.loader.NodeData;
public class DefaultStateTransferGenerator implements StateTransferGenerator
{
public static final short STATE_TRANSFER_VERSION = Version.getVersionShort("2.0.0.GA");
private Log log = LogFactory.getLog(getClass().getName());
private TreeCache cache;
private Set internalFqns;
protected DefaultStateTransferGenerator(TreeCache cache)
{
this.cache = cache;
this.internalFqns = cache.getInternalFqns();
}
public void generateState(ObjectOutputStream out, DataNode rootNode, boolean generateTransient,
boolean generatePersistent, boolean suppressErrors) throws Throwable
{
Fqn fqn = rootNode.getFqn();
Throwable encouteredException = null;
try
{
//transient + marker
try
{
out.writeShort(STATE_TRANSFER_VERSION);
if (generateTransient)
{
if (log.isTraceEnabled())
log.trace("writing transient state for " + fqn);
marshallTransientState(rootNode, out);
if (log.isTraceEnabled())
log.trace("transient state succesfully written");
}
}
catch (Throwable t)
{
encouteredException=t;
log.error("failed getting the in-memory (transient) state", t);
out.writeObject(t);
}
finally
{
out.writeObject(StateTransferManager.STREAMING_DELIMETER_NODE);
}
//associated + marker
try
{
if (log.isTraceEnabled())
log.trace("writing associated state");
marshallAssociatedState(fqn, out);
if (log.isTraceEnabled())
log.trace("associated state succesfully written");
}
catch (Throwable t)
{
encouteredException=t;
log.error("failed writing associated state", t);
out.writeObject(t);
}
finally
{
out.writeObject(StateTransferManager.STREAMING_DELIMETER_NODE);
}
//persistent + marker
try
{
if (generatePersistent)
{
if (log.isTraceEnabled())
log.trace("writing persistent state for " +
fqn + ",using " + cache.getCacheLoader().getClass());
if (fqn.isRoot())
{
cache.getCacheLoader().loadEntireState(out);
}
else
{
cache.getCacheLoader().loadState(fqn, out);
}
if (log.isTraceEnabled())
log.trace("persistent state succesfully written");
}
}
catch (Throwable t)
{
encouteredException=t;
log.error("failed getting the persistent state", t);
out.writeObject(t);
}
finally
{
out.writeObject(StateTransferManager.STREAMING_DELIMETER_NODE);
}
}
finally
{
out.close();
if(encouteredException!=null && !suppressErrors)
{
throw encouteredException;
}
}
}
/**
* Do a preorder traversal: visit the node first, then the node's children
*
* @param out
* @throws Exception
*/
protected void marshallTransientState(DataNode node, ObjectOutputStream out) throws Exception
{
if (internalFqns.contains(node.getFqn()))
return;
Map attrs;
NodeData nd;
// first handle the current node
attrs = node.getData();
if (attrs == null || attrs.size() == 0)
nd = new NodeData(node.getFqn());
else
nd = new NodeData(node.getFqn(), attrs);
out.writeObject(nd);
// then visit the children
Map children = node.getChildren();
if (children == null)
return;
for (Iterator it = children.entrySet().iterator(); it.hasNext();)
{
Map.Entry entry = (Map.Entry) it.next();
marshallTransientState((DataNode) entry.getValue(), out);
}
}
/**
* Does nothing in this base class; can be overridden in a subclass.
*/
protected void marshallAssociatedState(Fqn fqn, ObjectOutputStream baos) throws Exception
{
// no-op in this base class
}
protected TreeCache getTreeCache()
{
return cache;
}
}
1.1 date: 2006/09/12 21:12:45; author: vblagojevic; state: Exp;JBossCache/src/org/jboss/cache/statetransfer/DefaultStateTransferIntegrator.java
Index: DefaultStateTransferIntegrator.java
===================================================================
/*
* JBoss, the OpenSource J2EE webOS
*
* Distributable under LGPL license.
* See terms of license at gnu.org.
*/
package org.jboss.cache.statetransfer;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.cache.DataNode;
import org.jboss.cache.Fqn;
import org.jboss.cache.TreeCache;
import org.jboss.cache.buddyreplication.BuddyManager;
import org.jboss.cache.factories.NodeFactory;
import org.jboss.cache.loader.CacheLoader;
import org.jboss.cache.loader.NodeData;
public class DefaultStateTransferIntegrator implements StateTransferIntegrator
{
protected Log log = LogFactory.getLog(getClass().getName());
private TreeCache cache;
private Fqn targetFqn;
private NodeFactory factory;
private byte nodeType;
private Set internalFqns;
public DefaultStateTransferIntegrator(Fqn targetFqn, TreeCache cache)
{
this.targetFqn = targetFqn;
this.cache = cache;
this.factory = NodeFactory.getInstance();
this.nodeType = cache.getConfiguration().isNodeLockingOptimistic()
? NodeFactory.NODE_TYPE_OPTIMISTIC_NODE
: NodeFactory.NODE_TYPE_TREENODE;
this.internalFqns = cache.getInternalFqns();
}
protected void integrateTransientState(ObjectInputStream in,DataNode target, ClassLoader cl) throws Exception
{
boolean transientSet = false;
ClassLoader oldCL = setClassLoader(cl);
try
{
if (log.isTraceEnabled())
log.trace("integrating transient state for " + target);
integrateTransientState(target, in);
transientSet = true;
if (log.isTraceEnabled())
log.trace("transient state successfully integrated for " + targetFqn);
notifyAllNodesCreated(target);
}
finally
{
if (!transientSet)
{
// Clear any existing state from the targetRoot
target.clear();
target.removeAllChildren();
}
resetClassLoader(oldCL);
}
}
/**
* Provided for subclasses that deal with associated state.
*
* @throws Exception
*/
protected void integrateAssociatedState(ObjectInputStream in) throws Exception
{
// no-op in this base class
// just read marker
in.readObject();
}
protected void integratePersistentState(ObjectInputStream in) throws Exception
{
CacheLoader loader = cache.getCacheLoader();
boolean persistentSet=false;
if (loader == null)
{
log.error("cache loader is null, cannot set persistent state");
}
else
{
if (log.isTraceEnabled())
log.trace("integrating persistent state using " + loader.getClass());
try
{
if (targetFqn.isRoot())
{
loader.storeEntireState(in);
}
else
{
loader.storeState(targetFqn, in);
}
persistentSet=true;
}
finally
{
if(!persistentSet)
{
if (log.isTraceEnabled())
log.trace("persistent state integration failed, removing all nodes from loader");
loader.remove(targetFqn);
}
else
{
if (log.isTraceEnabled())
log.trace("persistent state integrated successfully");
}
}
}
}
/**
* Generates NodeAdded notifications for all nodes of the tree. This is
* called whenever the tree is initially retrieved (state transfer)
*/
private void notifyAllNodesCreated(DataNode curr)
{
DataNode n;
Map children;
if (curr == null) return;
getCache().getNotifier().notifyNodeCreated(curr.getFqn(), true);
getCache().getNotifier().notifyNodeCreated(curr.getFqn(), false);
if ((children = curr.getChildren()) != null)
{
for (Iterator it = children.values().iterator(); it.hasNext();)
{
n = (DataNode) it.next();
notifyAllNodesCreated(n);
}
}
}
private ClassLoader setClassLoader(ClassLoader newLoader)
{
ClassLoader oldClassLoader = null;
if (newLoader != null)
{
oldClassLoader = Thread.currentThread().getContextClassLoader();
Thread.currentThread().setContextClassLoader(newLoader);
}
return oldClassLoader;
}
private void resetClassLoader(ClassLoader oldLoader)
{
if (oldLoader != null)
Thread.currentThread().setContextClassLoader(oldLoader);
}
private void integrateTransientState(DataNode target, ObjectInputStream in) throws IOException,
ClassNotFoundException
{
Set retainedNodes = retainInternalNodes(target);
target.removeAllChildren();
// Read the first NodeData and integrate into our target
NodeData nd = (NodeData) in.readObject();
//are there any transient nodes at all?
if (!nd.isMarker())
{
Map attrs = nd.getAttributes();
if (attrs != null)
target.put(attrs, true);
else
target.clear();
// Check whether this is an integration into the buddy backup
// subtree
Fqn tferFqn = nd.getFqn();
Fqn tgtFqn = target.getFqn();
boolean move = tgtFqn.isChildOrEquals(BuddyManager.BUDDY_BACKUP_SUBTREE_FQN)
&& !tferFqn.isChildOrEquals(tgtFqn);
// If it is an integration, calculate how many levels of offset
int offset = move ? tgtFqn.size() - tferFqn.size() : 0;
integrateStateTransferChildren(target, offset, in);
integrateRetainedNodes(target, retainedNodes);
}
}
private NodeData integrateStateTransferChildren(DataNode parent, int offset, ObjectInputStream in)
throws IOException, ClassNotFoundException
{
int parent_level = parent.getFqn().size();
int target_level = parent_level + 1;
Fqn fqn;
int size;
Object name;
NodeData nd = (NodeData) in.readObject();
while (nd != null && !nd.isMarker())
{
fqn = nd.getFqn();
// If we need to integrate into the buddy backup subtree,
// change the Fqn to fit under it
if (offset > 0)
fqn = new Fqn(parent.getFqn().getFqnChild(offset), fqn);
size = fqn.size();
if (size <= parent_level)
return nd;
else if (size > target_level)
throw new IllegalStateException("NodeData " + fqn + " is not a direct child of " + parent.getFqn());
name = fqn.get(size - 1);
// We handle this NodeData. Create a DataNode and
// integrate its data
DataNode target = factory.createDataNode(nodeType, name, fqn, parent, nd.getAttributes(), true, cache);
parent.addChild(name, target);
// Recursively call, which will walk down the tree
// and return the next NodeData that's a child of our parent
nd = integrateStateTransferChildren(target, offset, in);
}
return null;
}
private Set retainInternalNodes(DataNode target)
{
Set result = new HashSet();
Fqn targetFqn = target.getFqn();
for (Iterator it = internalFqns.iterator(); it.hasNext();)
{
Fqn internalFqn = (Fqn) it.next();
if (internalFqn.isChildOf(targetFqn))
{
DataNode internalNode = getInternalNode(target, internalFqn);
if (internalNode != null)
result.add(internalNode);
}
}
return result;
}
private DataNode getInternalNode(DataNode parent, Fqn internalFqn)
{
Object name = internalFqn.get(parent.getFqn().size());
DataNode result = (DataNode) parent.getChild(name);
if (result != null)
{
if (internalFqn.size() < result.getFqn().size())
{
// need to recursively walk down the tree
result = getInternalNode(result, internalFqn);
}
}
return result;
}
private void integrateRetainedNodes(DataNode root, Set retainedNodes)
{
Fqn rootFqn = root.getFqn();
for (Iterator it = retainedNodes.iterator(); it.hasNext();)
{
DataNode retained = (DataNode) it.next();
if (retained.getFqn().isChildOf(rootFqn))
{
integrateRetainedNode(root, retained);
}
}
}
private void integrateRetainedNode(DataNode ancestor, DataNode descendant)
{
Fqn descFqn = descendant.getFqn();
Fqn ancFqn = ancestor.getFqn();
Object name = descFqn.get(ancFqn.size());
DataNode child = (DataNode) ancestor.getChild(name);
if (ancFqn.size() == descFqn.size() + 1)
{
if (child == null)
{
ancestor.addChild(name, descendant);
}
else
{
log.warn("Received unexpected internal node " + descFqn + " in transferred state");
}
}
else
{
if (child == null)
{
// Missing level -- have to create empty node
// This shouldn't really happen -- internal fqns should
// be immediately under the root
child = factory.createDataNode(nodeType, name, new Fqn(ancFqn, name), ancestor, null, true, cache);
ancestor.addChild(name, child);
}
// Keep walking down the tree
integrateRetainedNode(child, descendant);
}
}
protected TreeCache getCache()
{
return cache;
}
protected NodeFactory getFactory()
{
return factory;
}
protected byte getNodeType()
{
return nodeType;
}
protected Fqn getTargetFqn()
{
return targetFqn;
}
public void integrateState(ObjectInputStream ois, DataNode target, ClassLoader cl) throws Exception
{
Throwable cause=null;
//first try integrating transient state
try
{
integrateTransientState(ois, target, cl);
}
catch(Throwable t)
{
cause = t;
log.error("Failed integrating transient state.",t);
}
//then try integrating associated state
try
{
integrateAssociatedState(ois);
}
catch(Throwable t)
{
cause = t;
log.error("Failed integrating associated state.",t);
}
//finally try integrating persistent
try
{
integratePersistentState(ois);
}
catch (ClassCastException cce)
{
cause = cce;
log.error("Failed integrating persistent state. There are two possible causes:\n " +
"- state provider cacheloader generated exception during cacherloader.loadState() \n " +
"- one of cacheloaders is not adhering to stream format[JBCACHE-738].\n Caused by " + cce.getLocalizedMessage());
}
catch(Throwable t)
{
cause = t;
log.error("Failed integrating persistent state.", t);
}
finally
{
ois.close();
if(cause!=null)
{
throw new Exception("State transfer failed ");
}
}
}
}
More information about the jboss-cvs-commits
mailing list