Mailing-List: contact ode-commits-help@incubator.apache.org; run by ezmlm
Precedence: bulk
Reply-To: ode-dev@incubator.apache.org
Content-Type: text/plain; charset="utf-8"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
Subject: svn commit: r549162 -
 /incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelProcess.java
Date: Wed, 20 Jun 2007 17:34:37 -0000
To: ode-commits@incubator.apache.org
From: mszefler@apache.org
Message-Id: <20070620173437.883551A981A@eris.apache.org>

Author: mszefler
Date: Wed Jun 20 10:34:36 2007
New Revision: 549162

URL: http://svn.apache.org/viewvc?view=rev&rev=549162
Log:
Hydration latch.

Modified:
    incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelProcess.java

Modified: incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelProcess.java
URL: http://svn.apache.org/viewvc/incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelProcess.java?view=diff&rev=549162&r1=549161&r2=549162
==============================================================================
--- incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelProcess.java (original)
+++ incubator/ode/trunk/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelProcess.java Wed Jun 20 10:34:36 2007
@@ -18,9 +18,21 @@
  */
 package org.apache.ode.bpel.engine;
 
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+
+import javax.xml.namespace.QName;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.ode.bpel.common.FaultException;
+import org.apache.ode.bpel.dao.BpelDAOConnection;
 import org.apache.ode.bpel.dao.ProcessDAO;
 import org.apache.ode.bpel.dao.ProcessInstanceDAO;
 import org.apache.ode.bpel.evt.ProcessInstanceEvent;
@@ -54,15 +66,6 @@
 import org.w3c.dom.NodeList;
 import org.w3c.dom.Text;
 
-import javax.xml.namespace.QName;
-import java.io.InputStream;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
 /**
  * Entry point into the runtime of a BPEL process.
  * 
@@ -75,47 +78,56 @@
     private static final Messages __msgs = MessageBundle.getMessages(Messages.class);
 
     private volatile Map<OPartnerLink, PartnerLinkPartnerRoleImpl> _partnerRoles;
+
     private volatile Map<OPartnerLink, PartnerLinkMyRoleImpl> _myRoles;
+
     /** Mapping from {"Service Name" (QNAME) / port} to a myrole. */
     private volatile Map<Endpoint, PartnerLinkMyRoleImpl> _endpointToMyRoleMap;
 
     // Backup hashmaps to keep initial endpoints handy after dehydration
-    private Map<Endpoint,EndpointReference> _myEprs =
-            new HashMap<Endpoint, EndpointReference>();
-    private Map<Endpoint,EndpointReference> _partnerEprs =
-            new HashMap<Endpoint, EndpointReference>();
-    private Map<Endpoint,PartnerRoleChannel> _partnerChannels =
-            new HashMap<Endpoint, PartnerRoleChannel>();
+    private Map<Endpoint, EndpointReference> _myEprs = new HashMap<Endpoint, EndpointReference>();
+
+    private Map<Endpoint, EndpointReference> _partnerEprs = new HashMap<Endpoint, EndpointReference>();
+
+    private Map<Endpoint, PartnerRoleChannel> _partnerChannels = new HashMap<Endpoint, PartnerRoleChannel>();
 
     final QName _pid;
+
     private volatile OProcess _oprocess;
+
     // Has the process already been hydrated before?
     private boolean _hydratedOnce = false;
+
     /** Last time the process was used. */
     private volatile long _lastUsed;
 
     BpelEngineImpl _engine;
+
     DebuggerSupport _debugger;
+
     ExpressionLanguageRuntimeRegistry _expLangRuntimeRegistry;
+
     private ReplacementMap _replacementMap;
+
     final ProcessConf _pconf;
-    // Notifying the server when a process hydrates
-    private ProcessLifecycleCallback _lifeCallback;
+
     /** {@link MessageExchangeInterceptor}s registered for this process. */
     private final List<MessageExchangeInterceptor> _mexInterceptors = new ArrayList<MessageExchangeInterceptor>();
 
-    public BpelProcess(ProcessConf conf, BpelEventListener debugger, ProcessLifecycleCallback lifeCallback) {
+    /** Latch-like thing to control hydration/dehydration. */
+    private HydrationLatch _hydrationLatch;
+
+    public BpelProcess(ProcessConf conf, BpelEventListener debugger) {
         _pid = conf.getProcessId();
         _pconf = conf;
-        _lifeCallback = lifeCallback;
+        _hydrationLatch = new HydrationLatch();
     }
 
     public String toString() {
         return "BpelProcess[" + _pid + "]";
     }
 
-    public void recoverActivity(ProcessInstanceDAO instanceDAO, String channel, long activityId,
-                                String action, FaultData fault) {
+    public void recoverActivity(ProcessInstanceDAO instanceDAO, String channel, long activityId, String action, FaultData fault) {
         if (__log.isDebugEnabled())
             __log.debug("Recovering activity in process " + instanceDAO.getInstanceId() + " with action " + action);
         markused();
@@ -132,27 +144,33 @@
 
     /**
      * Entry point for message exchanges aimed at the my role.
+     * 
      * @param mex
      */
     void invokeProcess(MyRoleMessageExchangeImpl mex) {
-        PartnerLinkMyRoleImpl target = getMyRoleForService(mex.getServiceName());
-        if (target == null) {
-            String errmsg = __msgs.msgMyRoleRoutingFailure(mex.getMessageExchangeId());
-            __log.error(errmsg);
-            mex.setFailure(MessageExchange.FailureType.UNKNOWN_ENDPOINT, errmsg, null);
-            return;
-        }
+        _hydrationLatch.latch(1);
+        try {
+            PartnerLinkMyRoleImpl target = getMyRoleForService(mex.getServiceName());
+            if (target == null) {
+                String errmsg = __msgs.msgMyRoleRoutingFailure(mex.getMessageExchangeId());
+                __log.error(errmsg);
+                mex.setFailure(MessageExchange.FailureType.UNKNOWN_ENDPOINT, errmsg, null);
+                return;
+            }
 
-        mex.getDAO().setProcess(getProcessDAO());
-
-        if (!processInterceptors(mex, InterceptorInvoker.__onProcessInvoked)) {
-            __log.debug("Aborting processing of mex " + mex + " due to interceptors.");
-            return;
-        }
+            mex.getDAO().setProcess(getProcessDAO());
 
-        markused();
-        target.invokeMyRole(mex);
-        markused();
+            if (!processInterceptors(mex, InterceptorInvoker.__onProcessInvoked)) {
+                __log.debug("Aborting processing of mex " + mex + " due to interceptors.");
+                return;
+            }
+
+            markused();
+            target.invokeMyRole(mex);
+            markused();
+        } finally {
+            _hydrationLatch.release(1);
+        }
 
         // For a one way, once the engine is done, the mex can be safely released.
         if (mex.getPattern().equals(MessageExchange.MessageExchangePattern.REQUEST_ONLY)) {
@@ -178,16 +196,19 @@
         if (target != null) {
             mex.setPortOp(target._plinkDef.myRolePortType, target._plinkDef.getMyRoleOperation(mex.getOperationName()));
         } else {
-            __log.warn("Couldn't find endpoint from service " + mex.getServiceName()
-                    + " when initializing a myRole mex.");
+            __log.warn("Couldn't find endpoint from service " + mex.getServiceName() + " when initializing a myRole mex.");
         }
     }
 
     /**
      * Extract the value of a BPEL property from a BPEL messsage variable.
-     * @param msgData message variable data
-     * @param alias alias to apply
-     * @param target description of the data (for error logging only)
+     * 
+     * @param msgData
+     *            message variable data
+     * @param alias
+     *            alias to apply
+     * @param target
+     *            description of the data (for error logging only)
      * @return value of the property
      * @throws FaultException
      */
@@ -230,16 +251,15 @@
     }
 
     /**
-     * Get the element name for a given WSDL part. If the part is an
-     * <em>element</em> part, the name of that element is returned. If the
-     * part is an XML schema typed part, then the name of the part is returned
-     * in the null namespace.
-     * @param part WSDL {@link javax.wsdl.Part}
+     * Get the element name for a given WSDL part. If the part is an <em>element</em> part, the name of that element is returned.
+     * If the part is an XML schema typed part, then the name of the part is returned in the null namespace.
+     * 
+     * @param part
+     *            WSDL {@link javax.wsdl.Part}
      * @return name of element containing said part
      */
     static QName getElementNameForPart(OMessageVarType.Part part) {
-        return (part.type instanceof OElementVarType) ? ((OElementVarType) part.type).elementType
-                : new QName(null, part.name);
+        return (part.type instanceof OElementVarType) ? ((OElementVarType) part.type).elementType : new QName(null, part.name);
     }
 
     /**
@@ -247,12 +267,10 @@
      * 
      * @param mex
      *            message exchange
-     * @return <code>true</code> if execution should continue,
-     *         <code>false</code> otherwise
+     * @return <code>true</code> if execution should continue, <code>false</code> otherwise
      */
     boolean processInterceptors(MyRoleMessageExchangeImpl mex, InterceptorInvoker invoker) {
-        InterceptorContextImpl ictx = new InterceptorContextImpl(getEngine()._contexts.dao.getConnection(),
-                getProcessDAO(), _pconf);
+        InterceptorContextImpl ictx = new InterceptorContextImpl(_engine._contexts.dao.getConnection(), getProcessDAO(), _pconf);
 
         for (MessageExchangeInterceptor i : _mexInterceptors)
             if (!mex.processInterceptor(i, mex, ictx, invoker))
@@ -264,64 +282,68 @@
         return true;
     }
 
-
     /**
      * @see org.apache.ode.bpel.engine.BpelProcess#handleWorkEvent(java.util.Map<java.lang.String,java.lang.Object>)
      */
     public void handleWorkEvent(Map<String, Object> jobData) {
-        markused();
-
-        if (__log.isDebugEnabled()) {
-            __log.debug(ObjectPrinter.stringifyMethodEnter("handleWorkEvent", new Object[] { "jobData", jobData }));
-        }
-
-        WorkEvent we = new WorkEvent(jobData);
+        _hydrationLatch.latch(1);
+        try {
+            markused();
 
-        // Process level events
-        if (we.getType().equals(WorkEvent.Type.INVOKE_INTERNAL)) {
             if (__log.isDebugEnabled()) {
-                __log.debug("InvokeInternal event for mexid " + we.getMexId());
-            }
-            MyRoleMessageExchangeImpl mex = (MyRoleMessageExchangeImpl) getEngine().getMessageExchange(we.getMexId());
-            if (mex == null) throw new ContextException("Unable to find MEX " + we.getMexId());
-            invokeProcess(mex);
-        } else {
-            // Instance level events
-            ProcessInstanceDAO procInstance = getProcessDAO().getInstance(we.getIID());
-            if (procInstance == null) {
-                if (__log.isDebugEnabled()) {
-                    __log.debug("handleWorkEvent: no ProcessInstance found with iid " + we.getIID() + "; ignoring.");
-                }
-                return;
+                __log.debug(ObjectPrinter.stringifyMethodEnter("handleWorkEvent", new Object[] { "jobData", jobData }));
             }
 
-            BpelRuntimeContextImpl processInstance = createRuntimeContext(procInstance, null, null);
-            switch (we.getType()) {
-            case TIMER:
-                if (__log.isDebugEnabled()) {
-                    __log.debug("handleWorkEvent: TimerWork event for process instance " + processInstance);
-                }
-                processInstance.timerEvent(we.getChannel());
-                break;
-            case RESUME:
+
+            WorkEvent we = new WorkEvent(jobData);
+
+            // Process level events
+            if (we.getType().equals(WorkEvent.Type.INVOKE_INTERNAL)) {
                 if (__log.isDebugEnabled()) {
-                    __log.debug("handleWorkEvent: ResumeWork event for iid " + we.getIID());
+                    __log.debug("InvokeInternal event for mexid " + we.getMexId());
                 }
-                processInstance.execute();
-                break;
-            case INVOKE_RESPONSE:
-                if (__log.isDebugEnabled()) {
-                    __log.debug("InvokeResponse event for iid " + we.getIID());
+                MyRoleMessageExchangeImpl mex = (MyRoleMessageExchangeImpl) _engine.getMessageExchange(we.getMexId());
+                invokeProcess(mex);
+            } else {
+                // Instance level events
+                ProcessInstanceDAO procInstance = getProcessDAO().getInstance(we.getIID());
+                if (procInstance == null) {
+                    if (__log.isDebugEnabled()) {
+                        __log.debug("handleWorkEvent: no ProcessInstance found with iid " + we.getIID() + "; ignoring.");
+                    }
+                    return;
                 }
-                processInstance.invocationResponse(we.getMexId(), we.getChannel());
-                processInstance.execute();
-                break;
-            case MATCHER:
-                if (__log.isDebugEnabled()) {
-                    __log.debug("Matcher event for iid " + we.getIID());
+
+                BpelRuntimeContextImpl processInstance = createRuntimeContext(procInstance, null, null);
+                switch (we.getType()) {
+                case TIMER:
+                    if (__log.isDebugEnabled()) {
+                        __log.debug("handleWorkEvent: TimerWork event for process instance " + processInstance);
+                    }
+                    processInstance.timerEvent(we.getChannel());
+                    break;
+                case RESUME:
+                    if (__log.isDebugEnabled()) {
+                        __log.debug("handleWorkEvent: ResumeWork event for iid " + we.getIID());
+                    }
+                    processInstance.execute();
+                    break;
+                case INVOKE_RESPONSE:
+                    if (__log.isDebugEnabled()) {
+                        __log.debug("InvokeResponse event for iid " + we.getIID());
+                    }
+                    processInstance.invocationResponse(we.getMexId(), we.getChannel());
+                    processInstance.execute();
+                    break;
+                case MATCHER:
+                    if (__log.isDebugEnabled()) {
+                        __log.debug("Matcher event for iid " + we.getIID());
+                    }
+                    processInstance.matcherEvent(we.getCorrelatorId(), we.getCorrelationKey());
                 }
-                processInstance.matcherEvent(we.getCorrelatorId(), we.getCorrelationKey());
             }
+        } finally {
+            _hydrationLatch.release(1);
         }
     }
 
@@ -335,8 +357,8 @@
         for (Map.Entry<String, Endpoint> provide : _pconf.getProvideEndpoints().entrySet()) {
             OPartnerLink plink = oprocess.getPartnerLink(provide.getKey());
             if (plink == null) {
-                String errmsg = "Error in deployment descriptor for process " + _pid
-                        + "; reference to unknown partner link " + provide.getKey();
+                String errmsg = "Error in deployment descriptor for process " + _pid + "; reference to unknown partner link "
+                        + provide.getKey();
                 __log.error(errmsg);
                 throw new BpelEngineException(errmsg);
             }
@@ -347,13 +369,13 @@
         for (Map.Entry<String, Endpoint> invoke : _pconf.getInvokeEndpoints().entrySet()) {
             OPartnerLink plink = oprocess.getPartnerLink(invoke.getKey());
             if (plink == null) {
-                String errmsg = "Error in deployment descriptor for process " + _pid
-                        + "; reference to unknown partner link " + invoke.getKey();
+                String errmsg = "Error in deployment descriptor for process " + _pid + "; reference to unknown partner link "
+                        + invoke.getKey();
                 __log.error(errmsg);
                 throw new BpelEngineException(errmsg);
             }
-            __log.debug("Processing <invoke> element for process " + _pid + ": partnerlink " + invoke.getKey()
-                    + " --> " + invoke.getValue());
+            __log.debug("Processing <invoke> element for process " + _pid + ": partnerlink " + invoke.getKey() + " --> "
+                    + invoke.getValue());
         }
 
         for (OPartnerLink pl : oprocess.getAllPartnerLinks()) {
@@ -367,16 +389,16 @@
             }
 
             if (pl.hasPartnerRole()) {
-                PartnerLinkPartnerRoleImpl partnerRole = new PartnerLinkPartnerRoleImpl(
-                        this, pl, _pconf.getInvokeEndpoints().get(pl.getName()));
+                PartnerLinkPartnerRoleImpl partnerRole = new PartnerLinkPartnerRoleImpl(this, pl, _pconf.getInvokeEndpoints().get(
+                        pl.getName()));
                 _partnerRoles.put(pl, partnerRole);
             }
         }
     }
 
     ProcessDAO getProcessDAO() {
-        return _pconf.isTransient() ? getEngine()._contexts.inMemDao.getConnection().getProcess(_pid)
-                : getEngine()._contexts.dao.getConnection().getProcess(_pid);
+        return _pconf.isTransient() ? _engine._contexts.inMemDao.getConnection().getProcess(_pid) : getEngine()._contexts.dao
+                .getConnection().getProcess(_pid);
     }
 
     static String genCorrelatorId(OPartnerLink plink, String opName) {
@@ -385,7 +407,7 @@
 
     /**
      * De-serialize the compiled process representation from a stream.
-     *
+     * 
      * @param is
      *            input stream
      * @return process information from configuration database
@@ -417,8 +439,7 @@
         __log.debug("Activating " + _pid);
         // Activate all the my-role endpoints.
         for (Map.Entry<String, Endpoint> entry : _pconf.getProvideEndpoints().entrySet()) {
-            EndpointReference initialEPR = getEngine()._contexts
-                    .bindingContext.activateMyRoleEndpoint(_pid, entry.getValue());
+            EndpointReference initialEPR = _engine._contexts.bindingContext.activateMyRoleEndpoint(_pid, entry.getValue());
             __log.debug("Activated " + _pid + " myrole " + entry.getKey() + ": EPR is " + initialEPR);
             _myEprs.put(entry.getValue(), initialEPR);
         }
@@ -430,30 +451,45 @@
     void deactivate() {
         // Deactivate all the my-role endpoints.
         for (Endpoint endpoint : _myEprs.keySet())
-            getEngine()._contexts.bindingContext.deactivateMyRoleEndpoint(endpoint);
+            _engine._contexts.bindingContext.deactivateMyRoleEndpoint(endpoint);
 
-         // TODO Deactivate all the partner-role channels
+        // TODO Deactivate all the partner-role channels
     }
 
     EndpointReference getInitialPartnerRoleEPR(OPartnerLink link) {
-        PartnerLinkPartnerRoleImpl prole = getPartnerRoles().get(link);
-        if (prole == null)
-            throw new IllegalStateException("Unknown partner link " + link);
-        return prole.getInitialEPR();
+        _hydrationLatch.latch(1);
+        try {
+            PartnerLinkPartnerRoleImpl prole = _partnerRoles.get(link);
+            if (prole == null)
+                throw new IllegalStateException("Unknown partner link " + link);
+            return prole.getInitialEPR();
+        } finally {
+            _hydrationLatch.release(1);
+        }
     }
 
     Endpoint getInitialPartnerRoleEndpoint(OPartnerLink link) {
-        PartnerLinkPartnerRoleImpl prole = getPartnerRoles().get(link);
-        if (prole == null)
-            throw new IllegalStateException("Unknown partner link " + link);
-        return prole._initialPartner;
+        _hydrationLatch.latch(1);
+        try {
+            PartnerLinkPartnerRoleImpl prole = _partnerRoles.get(link);
+            if (prole == null)
+                throw new IllegalStateException("Unknown partner link " + link);
+            return prole._initialPartner;
+        } finally {
+            _hydrationLatch.release(1);
+        }
     }
 
     EndpointReference getInitialMyRoleEPR(OPartnerLink link) {
-        PartnerLinkMyRoleImpl myRole = getMyRoles().get(link);
-        if (myRole == null)
-            throw new IllegalStateException("Unknown partner link " + link);
-        return myRole.getInitialEPR();
+        _hydrationLatch.latch(1);
+        try {
+            PartnerLinkMyRoleImpl myRole = _myRoles.get(link);
+            if (myRole == null)
+                throw new IllegalStateException("Unknown partner link " + link);
+            return myRole.getInitialEPR();
+        } finally {
+            _hydrationLatch.release(1);
+        }
     }
 
     QName getPID() {
@@ -461,10 +497,15 @@
     }
 
     PartnerRoleChannel getPartnerRoleChannel(OPartnerLink partnerLink) {
-        PartnerLinkPartnerRoleImpl prole = getPartnerRoles().get(partnerLink);
-        if (prole == null)
-            throw new IllegalStateException("Unknown partner link " + partnerLink);
-        return prole._channel;
+        _hydrationLatch.latch(1);
+        try {
+            PartnerLinkPartnerRoleImpl prole = _partnerRoles.get(partnerLink);
+            if (prole == null)
+                throw new IllegalStateException("Unknown partner link " + partnerLink);
+            return prole._channel;
+        } finally {
+            _hydrationLatch.release(1);
+        }
     }
 
     public void saveEvent(ProcessInstanceEvent event, ProcessInstanceDAO instanceDao) {
@@ -476,8 +517,10 @@
 
         boolean enabled = _pconf.isEventEnabled(scopeNames, event.getType());
         if (enabled) {
-            if (instanceDao != null) saveInstanceEvent(event, instanceDao);
-            else __log.debug("Couldn't find instance to save event, no event generated!");
+            if (instanceDao != null)
+                saveInstanceEvent(event, instanceDao);
+            else
+                __log.debug("Couldn't find instance to save event, no event generated!");
         }
     }
 
@@ -485,98 +528,46 @@
         instanceDao.insertBpelEvent(event);
     }
 
+    /**
+     * Ask the process to dehydrate.
+     */
     void dehydrate() {
-        _oprocess = null;
-        _partnerRoles = null;
-        _myRoles = null;
-        _endpointToMyRoleMap = null;
-        _replacementMap = null;
-        _expLangRuntimeRegistry = null;
-    }
+        _hydrationLatch.latch(0);
 
-    void hydrate() {
-        markused();
-        __log.debug("Rehydrating process " + _pconf.getProcessId());
         try {
-            _oprocess = deserializeCompiledProcess(_pconf.getCBPInputStream());
-        } catch (Exception e) {
-            String errmsg = "Error reloading compiled process " + _pid + "; the file appears to be corrupted.";
-            __log.error(errmsg);
-            throw new BpelEngineException(errmsg, e);
-        }
-
-        _replacementMap = new ReplacementMapImpl(_oprocess);
-
-        // Create an expression language registry for this process
-        ExpressionLanguageRuntimeRegistry elangRegistry = new ExpressionLanguageRuntimeRegistry();
-        for (OExpressionLanguage elang : _oprocess.expressionLanguages) {
-            try {
-                elangRegistry.registerRuntime(elang);
-            } catch (ConfigurationException e) {
-                String msg = __msgs.msgExpLangRegistrationError(elang.expressionLanguageUri, elang.properties);
-                __log.error(msg, e);
-                throw new BpelEngineException(msg, e);
-            }
+            // We don't actually need to do anything, the latch will run the doDehydrate method
+            // when necessary..
+        } finally {
+            _hydrationLatch.release(0);
         }
-        _expLangRuntimeRegistry = elangRegistry;
-
-        setRoles(getOProcess());
 
-        if (!_hydratedOnce) {
-            for (PartnerLinkPartnerRoleImpl prole : getPartnerRoles().values()) {
-                PartnerRoleChannel channel = getEngine()._contexts.bindingContext.createPartnerRoleChannel(_pid,
-                        prole._plinkDef.partnerRolePortType, prole._initialPartner);
-                prole._channel = channel;
-                _partnerChannels.put(prole._initialPartner, prole._channel);
-                EndpointReference epr = channel.getInitialEndpointReference();
-                if (epr != null) {
-                    prole._initialEPR = epr;
-                    _partnerEprs.put(prole._initialPartner, epr);
-                }
-                __log.debug("Activated " + _pid + " partnerrole " + prole.getPartnerLinkName() + ": EPR is "
-                        + prole._initialEPR);
-            }
-            _hydratedOnce = true;
-        }
-
-        for (PartnerLinkMyRoleImpl myrole : getMyRoles().values()) {
-            myrole._initialEPR = _myEprs.get(myrole._endpoint);
-        }
-
-        for (PartnerLinkPartnerRoleImpl prole : getPartnerRoles().values()) {
-            prole._channel = _partnerChannels.get(prole._initialPartner);
-            if (_partnerEprs.get(prole._initialPartner) != null) {
-                prole._initialEPR = _partnerEprs.get(prole._initialPartner);
-            }
-        }
-
-        _lifeCallback.hydrated(this);
     }
 
     OProcess getOProcess() {
-        if (_oprocess == null) hydrate();
-        return _oprocess;
-    }
-
-    public Map<OPartnerLink, PartnerLinkMyRoleImpl> getMyRoles() {
-        if (_myRoles == null) hydrate();
-        return _myRoles;
-    }
-
-    public Map<OPartnerLink, PartnerLinkPartnerRoleImpl> getPartnerRoles() {
-        if (_partnerRoles == null) hydrate();
-        return _partnerRoles;
+        _hydrationLatch.latch(1);
+        try {
+            return _oprocess;
+        } finally {
+            _hydrationLatch.release(1);
+        }
     }
 
     private Map<Endpoint, PartnerLinkMyRoleImpl> getEndpointToMyRoleMap() {
-        if (_endpointToMyRoleMap == null) hydrate();
-        return _endpointToMyRoleMap;
+        _hydrationLatch.latch(1);
+        try {
+            return _endpointToMyRoleMap;
+        } finally {
+            _hydrationLatch.release(1);
+        }
     }
 
     public ReplacementMap getReplacementMap() {
-        if (_replacementMap == null) hydrate();
-        assert _replacementMap != null;
-        return _replacementMap;
+        _hydrationLatch.latch(1);
+        try {
+            return _replacementMap;
+        } finally {
+            _hydrationLatch.release(1);
+        }
     }
 
     BpelEngineImpl getEngine() {
@@ -591,6 +582,13 @@
         return _lastUsed;
     }
 
+    /**
+     * Get a hint as to whether this process is hydrated. Note this is only a hint, since things could change.
+     */
+    public boolean hintIsHydrated() {
+        return _oprocess != null;
+    }
+
     /** Keep track of the time the process was last used. */
     private final void markused() {
         _lastUsed = System.currentTimeMillis();
@@ -600,5 +598,155 @@
     BpelRuntimeContextImpl createRuntimeContext(ProcessInstanceDAO dao, PROCESS template,
             MyRoleMessageExchangeImpl instantiatingMessageExchange) {
         return new BpelRuntimeContextImpl(this, dao, template, instantiatingMessageExchange);
+
+    }
+
+    /**
+     * If necessary, create an object in the data store to represent the process. We'll re-use an existing object if it already
+     * exists and matches the GUID.
+     */
+    private void bounceProcessDAO(BpelDAOConnection conn, final QName pid, final long version, final OProcess oprocess) {
+        __log.debug("Creating process DAO for " + pid + " (guid=" + oprocess.guid + ")");
+        try {
+            boolean create = true;
+            ProcessDAO old = conn.getProcess(pid);
+            if (old != null) {
+                __log.debug("Found ProcessDAO for " + pid + " with GUID " + old.getGuid());
+                if (oprocess.guid == null) {
+                    // No guid, old version assume its good
+                    create = false;
+                } else {
+                    if (old.getGuid().equals(oprocess.guid)) {
+                        // Guids match, no need to create
+                        create = false;
+                    } else {
+                        // GUIDS dont match, delete and create new
+                        String errmsg = "ProcessDAO GUID " + old.getGuid() + " does not match " + oprocess.guid + "; replacing.";
+                        __log.debug(errmsg);
+                        old.delete();
+                    }
+                }
+            }
+
+            if (create) {
+                ProcessDAO newDao = conn.createProcess(pid, oprocess.getQName(), oprocess.guid, (int) version);
+                for (String correlator : oprocess.getCorrelators()) {
+                    newDao.addCorrelator(correlator);
+                }
+            }
+        } catch (BpelEngineException ex) {
+            throw ex;
+        } catch (Exception dce) {
+            __log.error("DbError", dce);
+            throw new BpelEngineException("DbError", dce);
+        }
+    }
+
+    private class HydrationLatch extends NStateLatch {
+
+        HydrationLatch() {
+            super(new Runnable[2]);
+            _transitions[0] = new Runnable() {
+                public void run() {
+                    doDehydrate();
+                }
+            };
+
+            _transitions[1] = new Runnable() {
+                public void run() {
+                    doHydrate();
+                }
+            };
+
+        }
+
+        private void doDehydrate() {
+            _oprocess = null;
+            _partnerRoles = null;
+            _myRoles = null;
+            _endpointToMyRoleMap = null;
+            _replacementMap = null;
+            _expLangRuntimeRegistry = null;
+        }
+
+        private void doHydrate() {
+            markused();
+            __log.debug("Rehydrating process " + _pconf.getProcessId());
+            try {
+                _oprocess = deserializeCompiledProcess(_pconf.getCBPInputStream());
+            } catch (Exception e) {
+                String errmsg = "Error reloading compiled process " + _pid + "; the file appears to be corrupted.";
+                __log.error(errmsg);
+                throw new BpelEngineException(errmsg, e);
+            }
+
+            _replacementMap = new ReplacementMapImpl(_oprocess);
+
+            // Create an expression language registry for this process
+            ExpressionLanguageRuntimeRegistry elangRegistry = new ExpressionLanguageRuntimeRegistry();
+            for (OExpressionLanguage elang : _oprocess.expressionLanguages) {
+                try {
+                    elangRegistry.registerRuntime(elang);
+                } catch (ConfigurationException e) {
+                    String msg = __msgs.msgExpLangRegistrationError(elang.expressionLanguageUri, elang.properties);
+                    __log.error(msg, e);
+                    throw new BpelEngineException(msg, e);
+                }
+            }
+            _expLangRuntimeRegistry = elangRegistry;
+
+            setRoles(_oprocess);
+
+            if (!_hydratedOnce) {
+                for (PartnerLinkPartnerRoleImpl prole : _partnerRoles.values()) {
+                    PartnerRoleChannel channel = _engine._contexts.bindingContext.createPartnerRoleChannel(_pid,
+                            prole._plinkDef.partnerRolePortType, prole._initialPartner);
+                    prole._channel = channel;
+                    _partnerChannels.put(prole._initialPartner, prole._channel);
+                    EndpointReference epr = channel.getInitialEndpointReference();
+                    if (epr != null) {
+                        prole._initialEPR = epr;
+                        _partnerEprs.put(prole._initialPartner, epr);
+                    }
+                    __log.debug("Activated " + _pid + " partnerrole " + prole.getPartnerLinkName() + ": EPR is "
+                            + prole._initialEPR);
+                }
+                _hydratedOnce = true;
+            }
+
+            for (PartnerLinkMyRoleImpl myrole : _myRoles.values()) {
+                myrole._initialEPR = _myEprs.get(myrole._endpoint);
+            }
+
+            for (PartnerLinkPartnerRoleImpl prole : _partnerRoles.values()) {
+                prole._channel = _partnerChannels.get(prole._initialPartner);
+                if (_partnerEprs.get(prole._initialPartner) != null) {
+                    prole._initialEPR = _partnerEprs.get(prole._initialPartner);
+                }
+            }
+
+
+            if (isInMemory()) {
+                bounceProcessDAO(_engine._contexts.inMemDao.getConnection(), _pid, _pconf.getVersion(), _oprocess);
+            } else if (_engine._contexts.scheduler.isTransacted()) {
+                // If we have a transaction, we do this in the current transaction.
+                bounceProcessDAO(_engine._contexts.dao.getConnection(), _pid, _pconf.getVersion(), _oprocess);
+            } else {
+                // If we do not have a transaction we need to create one. 
+                try {
+                    _engine._contexts.scheduler.execIsolatedTransaction(new Callable<Object>() {
+                        public Object call() throws Exception {
+                            bounceProcessDAO(_engine._contexts.dao.getConnection(), _pid, _pconf.getVersion(), _oprocess);
+                            return null;
+                        }
+                    });
+                } catch (Exception ex) {
+                    String errmsg = "DbError";
+                    __log.error(errmsg, ex);
+                    throw new BpelEngineException(errmsg, ex);
+                }
+            }
+        }
+
     }
 }