| /* |
| * 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 org.apache.ode.bpel.engine; |
| |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.apache.ode.bpel.clapi.ClusterLock; |
| import org.apache.ode.bpel.dao.MessageExchangeDAO; |
| import org.apache.ode.bpel.dao.ProcessDAO; |
| import org.apache.ode.bpel.dao.ProcessInstanceDAO; |
| import org.apache.ode.bpel.evt.BpelEvent; |
| import org.apache.ode.bpel.iapi.BpelEngine; |
| import org.apache.ode.bpel.iapi.BpelEngineException; |
| import org.apache.ode.bpel.iapi.ContextException; |
| import org.apache.ode.bpel.iapi.Endpoint; |
| import org.apache.ode.bpel.iapi.Message; |
| import org.apache.ode.bpel.iapi.MessageExchange; |
| import org.apache.ode.bpel.iapi.OdeGlobalConfig; |
| import org.apache.ode.bpel.iapi.MyRoleMessageExchange; |
| import org.apache.ode.bpel.iapi.PartnerRoleMessageExchange; |
| import org.apache.ode.bpel.iapi.ProcessState; |
| import org.apache.ode.bpel.iapi.Scheduler; |
| import org.apache.ode.bpel.iapi.MessageExchange.FailureType; |
| import org.apache.ode.bpel.iapi.MessageExchange.MessageExchangePattern; |
| import org.apache.ode.bpel.iapi.MessageExchange.Status; |
| import org.apache.ode.bpel.iapi.MyRoleMessageExchange.CorrelationStatus; |
| import org.apache.ode.bpel.iapi.Scheduler.JobDetails; |
| import org.apache.ode.bpel.iapi.Scheduler.JobType; |
| import org.apache.ode.bpel.intercept.InterceptorInvoker; |
| import org.apache.ode.bpel.intercept.MessageExchangeInterceptor; |
| import org.apache.ode.bpel.intercept.ProcessCountThrottler; |
| import org.apache.ode.bpel.intercept.ProcessSizeThrottler; |
| import org.apache.ode.bpel.obj.OConstants; |
| import org.apache.ode.bpel.obj.OPartnerLink; |
| import org.apache.ode.bpel.obj.OProcess; |
| import org.apache.ode.bpel.runtime.InvalidProcessException; |
| import org.apache.ode.utils.DOMUtils; |
| import org.apache.ode.utils.Namespaces; |
| import org.apache.ode.utils.msg.MessageBundle; |
| import org.w3c.dom.Document; |
| import org.w3c.dom.Element; |
| |
| import javax.wsdl.Operation; |
| import javax.wsdl.PortType; |
| import javax.xml.namespace.QName; |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Random; |
| import java.util.concurrent.TimeUnit; |
| |
| /** |
| * Implementation of the {@link BpelEngine} interface: provides the server methods that should be invoked in the context of a |
| * transaction. |
| * |
| * @author mszefler |
| * @author Matthieu Riou <mriou at apache dot org> |
| */ |
| public class BpelEngineImpl implements BpelEngine { |
| private static final Logger __log = LoggerFactory.getLogger(BpelEngineImpl.class); |
| |
| /** RNG, for delays */ |
| private Random _random = new Random(System.currentTimeMillis()); |
| |
| private static double _delayMean = 0; |
| |
| static { |
| try { |
| String delay = System.getenv("ODE_DEBUG_TX_DELAY"); |
| if (delay != null && delay.length() > 0) { |
| _delayMean = Double.valueOf(delay); |
| __log.info("Stochastic debugging delay activated. Delay (Mean)=" + _delayMean + "ms."); |
| } |
| } catch (Throwable t) { |
| if (__log.isDebugEnabled()) { |
| __log.debug("Could not read ODE_DEBUG_TX_DELAY environment variable; assuming 0 (mean) delay", t); |
| } else { |
| __log.info("Could not read ODE_DEBUG_TX_DELAY environment variable; assuming 0 (mean) delay"); |
| } |
| } |
| } |
| |
| private static final Messages __msgs = MessageBundle.getMessages(Messages.class); |
| |
| private static final double PROCESS_OVERHEAD_MEMORY_FACTOR = 1.2; |
| |
| /** Active processes, keyed by process id. */ |
| public final HashMap<QName, BpelProcess> _activeProcesses = new HashMap<QName, BpelProcess>(); |
| |
| /** Mapping from myrole service name to active process. */ |
| private final HashMap<QName, List<BpelProcess>> _serviceMap = new HashMap<QName, List<BpelProcess>>(); |
| |
| /** Mapping from a potentially shared endpoint to its EPR */ |
| private SharedEndpoints _sharedEps; |
| |
| /** Manage instance-level locks. */ |
| private final ClusterLock<Long> _instanceLockManager; |
| |
| final Contexts _contexts; |
| |
| private final Map<QName, Long> _hydratedSizes = new HashMap<QName, Long>(); |
| private final Map<QName, Long> _unhydratedSizes = new HashMap<QName, Long>(); |
| |
| public BpelEngineImpl(Contexts contexts) { |
| _contexts = contexts; |
| if(_contexts.clusterManager != null) { |
| _instanceLockManager = _contexts.clusterManager.getInstanceLock(); |
| } |
| else _instanceLockManager = new InstanceLockManager(); |
| _sharedEps = new SharedEndpoints(); |
| _sharedEps.init(); |
| |
| } |
| |
| public SharedEndpoints getSharedEndpoints() { |
| return _sharedEps; |
| } |
| |
| public MyRoleMessageExchange createMessageExchange(String clientKey, QName targetService, |
| String operation, String pipedMexId) |
| throws BpelEngineException { |
| |
| List<BpelProcess> targets = route(targetService, null); |
| List<BpelProcess> activeTargets = new ArrayList<BpelProcess>(); |
| for (BpelProcess target : targets) { |
| if (target.getConf().getState() == ProcessState.ACTIVE) { |
| activeTargets.add(target); |
| } |
| } |
| |
| if (targets == null || targets.size() == 0) |
| throw new BpelEngineException("NoSuchService: " + targetService); |
| |
| if (targets.size() == 1 || activeTargets.size() == 1) { |
| // If the number of targets is one, create and return a simple MEX |
| BpelProcess target; |
| if (activeTargets.size() == 1) { |
| target = activeTargets.get(0); |
| } else { |
| target = targets.get(0); |
| } |
| return createNewMyRoleMex(target, clientKey, targetService, operation, pipedMexId); |
| } else { |
| // If the number of targets is greater than one, create and return |
| // a brokered MEX that embeds the simple MEXs for each of the targets |
| BpelProcess template = activeTargets.get(0); |
| ArrayList<MyRoleMessageExchange> meps = new ArrayList<MyRoleMessageExchange>(); |
| for (BpelProcess target : activeTargets) { |
| meps.add(createNewMyRoleMex(target, clientKey, targetService, operation, pipedMexId)); |
| } |
| return createNewMyRoleMex(template, meps); |
| } |
| } |
| |
| private MyRoleMessageExchange createNewMyRoleMex(BpelProcess target, String clientKey, QName targetService, String operation, String pipedMexId) { |
| MessageExchangeDAO dao; |
| if (target == null || target.isInMemory()) { |
| dao = _contexts.inMemDao.getConnection().createMessageExchange(MessageExchangeDAO.DIR_PARTNER_INVOKES_MYROLE); |
| } else { |
| dao = _contexts.dao.getConnection().createMessageExchange(MessageExchangeDAO.DIR_PARTNER_INVOKES_MYROLE); |
| } |
| dao.setCorrelationId(clientKey); |
| dao.setCorrelationStatus(CorrelationStatus.UKNOWN_ENDPOINT.toString()); |
| dao.setPattern(MessageExchangePattern.UNKNOWN.toString()); |
| dao.setCallee(targetService); |
| dao.setStatus(Status.NEW.toString()); |
| dao.setOperation(operation); |
| dao.setPipedMessageExchangeId(pipedMexId); |
| MyRoleMessageExchangeImpl mex = new MyRoleMessageExchangeImpl(target, this, dao); |
| |
| if (target != null) { |
| target.initMyRoleMex(mex); |
| } |
| return mex; |
| } |
| |
| /** |
| * Return a brokered MEX that delegates invocations to each of the embedded |
| * MEXs contained in the <code>meps</code> list, using the appropriate style. |
| * |
| * @param target |
| * @param meps |
| * @return |
| * @throws BpelEngineException |
| */ |
| private MyRoleMessageExchange createNewMyRoleMex(BpelProcess target, List<MyRoleMessageExchange> meps) |
| throws BpelEngineException { |
| MyRoleMessageExchangeImpl templateMex = (MyRoleMessageExchangeImpl) meps.get(0); |
| MessageExchangeDAO templateMexDao = templateMex.getDAO(); |
| return new BrokeredMyRoleMessageExchangeImpl(target, this, meps, templateMexDao, templateMex); |
| } |
| |
| |
| public MyRoleMessageExchange createMessageExchange(String clientKey, QName targetService, String operation) { |
| return createMessageExchange(clientKey, targetService, operation, null); |
| } |
| |
| private void setMessageExchangeProcess(String mexId, ProcessDAO processDao) { |
| MessageExchangeDAO mexdao = _contexts.inMemDao.getConnection().getMessageExchange(mexId); |
| if (mexdao == null) mexdao = _contexts.dao.getConnection().getMessageExchange(mexId); |
| if (mexdao != null) |
| mexdao.setProcess(processDao); |
| } |
| |
| public MessageExchange getMessageExchange(String mexId) { |
| MessageExchangeDAO mexdao = _contexts.inMemDao.getConnection().getMessageExchange(mexId); |
| if (mexdao == null) mexdao = _contexts.dao.getConnection().getMessageExchange(mexId); |
| if (mexdao == null) |
| return null; |
| |
| ProcessDAO pdao = mexdao.getProcess(); |
| BpelProcess process = pdao == null ? null : _activeProcesses.get(pdao.getProcessId()); |
| |
| MessageExchangeImpl mex; |
| switch (mexdao.getDirection()) { |
| case MessageExchangeDAO.DIR_BPEL_INVOKES_PARTNERROLE: |
| if (process == null) { |
| String errmsg = __msgs.msgProcessNotActive(pdao.getProcessId()); |
| __log.error(errmsg); |
| // TODO: Perhaps we should define a checked exception for this |
| // condition. |
| throw new BpelEngineException(errmsg); |
| } |
| { |
| OPartnerLink plink = (OPartnerLink) process.getOProcess().getChild(mexdao.getPartnerLinkModelId()); |
| PortType ptype = plink.getPartnerRolePortType(); |
| Operation op = plink.getPartnerRoleOperation(mexdao.getOperation()); |
| // TODO: recover Partner's EPR |
| mex = createPartnerRoleMessageExchangeImpl(mexdao, ptype, op, plink, process); |
| } |
| break; |
| case MessageExchangeDAO.DIR_PARTNER_INVOKES_MYROLE: |
| mex = new MyRoleMessageExchangeImpl(process, this, mexdao); |
| if (process != null) { |
| Object child = process.getOProcess().getChild(mexdao.getPartnerLinkModelId()); |
| if (child instanceof OPartnerLink) { |
| OPartnerLink plink = (OPartnerLink) child; |
| // the partner link might not be hydrated |
| if (plink != null) { |
| PortType ptype = plink.getMyRolePortType(); |
| Operation op = plink.getMyRoleOperation(mexdao.getOperation()); |
| mex.setPortOp(ptype, op); |
| } |
| } |
| } |
| break; |
| default: |
| String errmsg = "BpelEngineImpl: internal error, invalid MexDAO direction: " + mexId; |
| __log.error(errmsg); |
| throw new BpelEngineException(errmsg); |
| } |
| |
| return mex; |
| } |
| |
| // enable extensibility |
| protected PartnerRoleMessageExchangeImpl createPartnerRoleMessageExchangeImpl( |
| MessageExchangeDAO mexdao, PortType ptype, Operation op, OPartnerLink plink, BpelProcess process) { |
| return new PartnerRoleMessageExchangeImpl(this, mexdao, ptype, op, null, plink.hasMyRole() ? process |
| .getInitialMyRoleEPR(plink) : null, process.getPartnerRoleChannel(plink)); |
| } |
| |
| BpelProcess unregisterProcess(QName process) { |
| BpelProcess p = _activeProcesses.remove(process); |
| if (__log.isDebugEnabled()) { |
| __log.debug("Unregister process: serviceId=" + process + ", process=" + p); |
| } |
| if (p != null) { |
| if (__log.isDebugEnabled()) |
| __log.debug("Deactivating process " + p.getPID()); |
| |
| boolean deactivate = true; |
| Iterator<Endpoint> endpointItr = p.getServiceNames().iterator(); |
| |
| ProcessState state = null; |
| |
| while(endpointItr.hasNext()){ |
| Endpoint endPoint = endpointItr.next(); |
| List<BpelProcess> processList = _serviceMap.get(endPoint.serviceName); |
| |
| Iterator<BpelProcess> processListItr = processList.iterator(); |
| |
| while(processListItr.hasNext()){ |
| BpelProcess entryProcess = processListItr.next(); |
| state = entryProcess.getConf().getState(); |
| // Don't deactivate process services if there is another process in Active/Retired state and |
| // associated with the same service name |
| if( (ProcessState.ACTIVE.equals(state) || ProcessState.RETIRED.equals(state)) && |
| !(entryProcess.getPID().equals(p.getPID()))) { |
| |
| deactivate = false; |
| } else { |
| if(entryProcess.getPID().equals(process)) { |
| processListItr.remove(); |
| } |
| } |
| } |
| } |
| |
| if(deactivate){ |
| // unregister the services provided by the process |
| p.deactivate(); |
| } |
| |
| // release the resources held by this process |
| p.dehydrate(); |
| // update the process footprints list |
| _hydratedSizes.remove(p.getPID()); |
| } |
| return p; |
| } |
| |
| boolean isProcessRegistered(QName pid) { |
| return _activeProcesses.containsKey(pid); |
| } |
| |
| public BpelProcess getProcess(QName pid) { |
| return _activeProcesses.get(pid); |
| } |
| |
| /** |
| * Register a process with the engine. |
| * @param process the process to register |
| */ |
| void registerProcess(BpelProcess process) { |
| _activeProcesses.put(process.getPID(), process); |
| for (Endpoint e : process.getServiceNames()) { |
| if (__log.isDebugEnabled()) { |
| __log.debug("Register process: serviceId=" + e + ", process=" + process); |
| } |
| List<BpelProcess> processes = _serviceMap.get(e.serviceName); |
| if (processes == null) { |
| processes = new ArrayList<BpelProcess>(); |
| _serviceMap.put(e.serviceName, processes); |
| } |
| // Remove any older version of the process from the list |
| Iterator<BpelProcess> processesIter = processes.iterator(); |
| while (processesIter.hasNext()) { |
| BpelProcess cachedVersion = processesIter.next(); |
| if (__log.isDebugEnabled()) { |
| __log.debug("cached version " + cachedVersion.getPID() + " vs registering version " + process.getPID()); |
| } |
| if (cachedVersion.getProcessType().equals(process.getProcessType())) { |
| if (cachedVersion.getConf().getState() == ProcessState.ACTIVE |
| && process.getConf().getState() == ProcessState.ACTIVE |
| && !OdeGlobalConfig.autoRetireProcess()) { |
| throw new ContextException("Can't activate two processes of the same name: " + process.getConf().getPackage() + ", " + cachedVersion.getConf().getPackage() + ", name: " + process.getProcessType()); |
| } |
| //Check for versions to retain newer one |
| if (cachedVersion.getVersion() > process.getVersion()) { |
| __log.debug("removing current version"); |
| process.activate(this); |
| process.deactivate(); |
| return; |
| } else { |
| __log.debug("removing cached older version"); |
| processesIter.remove(); |
| cachedVersion.deactivate(); |
| } |
| |
| } |
| } |
| processes.add(process); |
| } |
| process.activate(this); |
| } |
| |
| /** |
| * Route to a process using the service id. Note, that we do not need the endpoint name here, we are assuming that two processes |
| * would not be registered under the same service qname but different endpoint. |
| * |
| * @param service |
| * target service id |
| * @param request |
| * request message |
| * @return process corresponding to the targetted service, or <code>null</code> if service identifier is not recognized. |
| */ |
| List<BpelProcess> route(QName service, Message request) { |
| // TODO: use the message to route to the correct service if more than |
| // one service is listening on the same endpoint. |
| List<BpelProcess> routed = _serviceMap.get(service); |
| if (__log.isDebugEnabled()) |
| __log.debug("Routed: svcQname " + service + " --> " + routed); |
| return routed; |
| |
| } |
| |
| OProcess getOProcess(QName processId) { |
| BpelProcess process = _activeProcesses.get(processId); |
| |
| if (process == null) return null; |
| return process.getOProcess(); |
| } |
| |
| private List<BpelProcess> getAllProcesses(QName processId) { |
| String qName = processId.toString(); |
| if(qName.lastIndexOf("-") > 0) { |
| qName = qName.substring(0, qName.lastIndexOf("-")); |
| } |
| List<BpelProcess> ret = new ArrayList<BpelProcess>(); |
| Iterator<Map.Entry<QName, BpelProcess>> it = _activeProcesses.entrySet().iterator(); |
| while (it.hasNext()) { |
| Map.Entry<QName, BpelProcess> pairs = it.next(); |
| if(pairs.getKey().toString().startsWith(qName)) { |
| ret.add(pairs.getValue()); |
| } |
| } |
| return ret; |
| } |
| |
| public void acquireInstanceLock(final Long iid) { |
| // We lock the instance to prevent concurrent transactions and prevent unnecessary rollbacks, |
| // Note that we don't want to wait too long here to get our lock, since we are likely holding |
| // on to scheduler's locks of various sorts. |
| try { |
| _instanceLockManager. |
| lock(iid, 1, TimeUnit.MICROSECONDS); |
| _contexts.scheduler.registerSynchronizer(new Scheduler.Synchronizer() { |
| public void afterCompletion(boolean success) { |
| _instanceLockManager.unlock(iid); |
| } |
| public void beforeCompletion() { } |
| }); |
| } catch (InterruptedException e) { |
| // Retry later. |
| __log.debug("Thread interrupted, job will be rescheduled"); |
| throw new Scheduler.JobProcessorException(true); |
| } catch (org.apache.ode.bpel.engine.InstanceLockManager.TimeoutException e) { |
| __log.debug("Instance " + iid + " is busy, rescheduling job."); |
| throw new Scheduler.JobProcessorException(true); |
| } |
| } |
| |
| public void onScheduledJob(Scheduler.JobInfo jobInfo) throws Scheduler.JobProcessorException { |
| final JobDetails we = jobInfo.jobDetail; |
| |
| if( __log.isTraceEnabled() ) __log.trace("[JOB] onScheduledJob " + jobInfo + "" + we.getInstanceId()); |
| |
| acquireInstanceLock(we.getInstanceId()); |
| |
| // DONT PUT CODE HERE-need this method real tight in a try/catch block, we need to handle |
| // all types of failure here, the scheduler is not going to know how to handle our errors, |
| // ALSO we have to release the lock obtained above (IMPORTANT), lest the whole system come |
| // to a grinding halt. |
| BpelProcess process = null; |
| try { |
| if (we.getProcessId() != null) { |
| process = _activeProcesses.get(we.getProcessId()); |
| } else { |
| ProcessInstanceDAO instance; |
| if (we.getInMem()) instance = _contexts.inMemDao.getConnection().getInstance(we.getInstanceId()); |
| else instance = _contexts.dao.getConnection().getInstance(we.getInstanceId()); |
| |
| if (instance == null) { |
| __log.debug(__msgs.msgScheduledJobReferencesUnknownInstance(we.getInstanceId())); |
| // nothing we can do, this instance is not in the database, it will always fail, not |
| // exactly an error since can occur in normal course of events. |
| return; |
| } |
| ProcessDAO processDao = instance.getProcess(); |
| process = _activeProcesses.get(processDao.getProcessId()); |
| } |
| |
| if (process == null) { |
| // The process is not active, there's nothing we can do with this job |
| if (__log.isDebugEnabled()) { |
| __log.debug("Process " + we.getProcessId() + " can't be found, job abandoned."); |
| } |
| return; |
| } |
| |
| ClassLoader cl = Thread.currentThread().getContextClassLoader(); |
| try { |
| Thread.currentThread().setContextClassLoader(process._classLoader); |
| if (we.getType().equals(JobType.INVOKE_CHECK)) { |
| if (__log.isDebugEnabled()) __log.debug("handleJobDetails: InvokeCheck event for mexid " + we.getMexId()); |
| |
| sendPartnerRoleFailure(we, MessageExchange.FailureType.COMMUNICATION_ERROR); |
| return; |
| } else if (we.getType().equals(JobType.INVOKE_INTERNAL)) { |
| if (__log.isDebugEnabled()) __log.debug("handleJobDetails: InvokeInternal event for mexid " + we.getMexId()); |
| |
| setMessageExchangeProcess(we.getMexId(), process.getProcessDAO()); |
| MyRoleMessageExchangeImpl mex = (MyRoleMessageExchangeImpl) getMessageExchange(we.getMexId()); |
| if (!process.processInterceptors(mex, InterceptorInvoker.__onJobScheduled)) { |
| boolean isTwoWay = Boolean.valueOf(mex.getProperty("isTwoWay")); |
| if (isTwoWay) { |
| String causeCodeValue = mex.getProperty("causeCode"); |
| mex.getDAO().setProcess(process.getProcessDAO()); |
| sendMyRoleFault(process, we, causeCodeValue != null ? |
| Integer.valueOf(causeCodeValue) : InvalidProcessException.DEFAULT_CAUSE_CODE); |
| return; |
| } else { |
| throw new Scheduler.JobProcessorException(checkRetry(we)); |
| } |
| } |
| } |
| if (we.getType() == JobType.INVOKE_INTERNAL || we.getType() == JobType.MEX_MATCHER) { |
| List<BpelProcess> processes = getAllProcesses(we.getProcessId()); |
| boolean routed = false; |
| jobInfo.jobDetail.detailsExt.put("enqueue", false); |
| for(BpelProcess proc : processes) { |
| routed = routed || proc.handleJobDetails(jobInfo.jobDetail); |
| } |
| if(!routed && we.getType() == JobType.INVOKE_INTERNAL) { |
| jobInfo.jobDetail.detailsExt.put("enqueue", true); |
| process.handleJobDetails(jobInfo.jobDetail); |
| } |
| } |
| else { |
| process.handleJobDetails(jobInfo.jobDetail); |
| } |
| debuggingDelay(); |
| } finally { |
| Thread.currentThread().setContextClassLoader(cl); |
| } |
| } catch (Scheduler.JobProcessorException e) { |
| throw e; |
| } catch (BpelEngineException bee) { |
| __log.error(__msgs.msgScheduledJobFailed(we), bee); |
| throw new Scheduler.JobProcessorException(bee, checkRetry(we)); |
| } catch (ContextException ce) { |
| __log.error(__msgs.msgScheduledJobFailed(we), ce); |
| throw new Scheduler.JobProcessorException(ce, checkRetry(we)); |
| } catch (InvalidProcessException ipe) { |
| __log.error(__msgs.msgScheduledJobFailed(we), ipe); |
| sendMyRoleFault(process, we, ipe.getCauseCode()); |
| } catch (RuntimeException rte) { |
| __log.error(__msgs.msgScheduledJobFailed(we), rte); |
| throw new Scheduler.JobProcessorException(rte, checkRetry(we)); |
| } catch (Throwable t) { |
| __log.error(__msgs.msgScheduledJobFailed(we), t); |
| throw new Scheduler.JobProcessorException(t, checkRetry(we)); |
| } |
| } |
| |
| private boolean checkRetry(JobDetails we) { |
| // Only retry if the job is NOT in memory. Not that this does not guaranty that a retry will be scheduled. |
| // Actually events are not retried if not persisted and the scheduler might choose to discard the event if it has been retried too many times. |
| return !we.getInMem(); |
| } |
| |
| /** |
| * Block the thread for random amount of time. Used for testing for races and the like. The delay generated is exponentially |
| * distributed with the mean obtained from the <code>ODE_DEBUG_TX_DELAY</code> environment variable. |
| */ |
| private void debuggingDelay() { |
| // Do a delay for debugging purposes. |
| if (_delayMean != 0) |
| try { |
| long delay = randomExp(_delayMean); |
| // distribution |
| // with mean |
| // _delayMean |
| __log.warn("Debugging delay has been activated; delaying transaction for " + delay + "ms."); |
| Thread.sleep(delay); |
| } catch (InterruptedException e) { |
| ; // ignore |
| } |
| } |
| |
| private long randomExp(double mean) { |
| double u = _random.nextDouble(); // Uniform |
| long delay = (long) (-Math.log(u) * mean); // Exponential |
| return delay; |
| } |
| |
| void fireEvent(BpelEvent event) { |
| // Note that the eventListeners list is a copy-on-write array, so need |
| // to mess with synchronization. |
| for (org.apache.ode.bpel.iapi.BpelEventListener l : _contexts.eventListeners) { |
| l.onEvent(event); |
| } |
| } |
| |
| /** |
| * Get the list of globally-registered message-exchange interceptors. |
| * |
| * @return list |
| */ |
| List<MessageExchangeInterceptor> getGlobalInterceptors() { |
| return _contexts.globalInterceptors; |
| } |
| |
| |
| public void registerMessageExchangeInterceptor(MessageExchangeInterceptor interceptor) { |
| _contexts.globalInterceptors.add(interceptor); |
| } |
| |
| public void unregisterMessageExchangeInterceptor(MessageExchangeInterceptor interceptor) { |
| _contexts.globalInterceptors.remove(interceptor); |
| } |
| |
| public void unregisterMessageExchangeInterceptor(Class interceptorClass) { |
| MessageExchangeInterceptor candidate = null; |
| for (MessageExchangeInterceptor interceptor : _contexts.globalInterceptors) { |
| if (interceptor.getClass().isAssignableFrom(interceptorClass)) { |
| candidate = interceptor; |
| break; |
| } |
| } |
| if (candidate != null) { |
| _contexts.globalInterceptors.remove(candidate); |
| } |
| } |
| |
| public long getTotalBpelFootprint() { |
| long bpelFootprint = 0; |
| for (BpelProcess process : _activeProcesses.values()) { |
| Long size = _hydratedSizes.get(process.getPID()); |
| if (size == null) { |
| size = _unhydratedSizes.get(process.getPID()); |
| } |
| if (size != null && size.longValue() > 0) { |
| bpelFootprint += size; |
| } |
| } |
| return bpelFootprint; |
| } |
| |
| public long getHydratedFootprint() { |
| long hydratedFootprint = 0; |
| for (BpelProcess process : _activeProcesses.values()) { |
| if (!process.hintIsHydrated()) { |
| continue; |
| } |
| Long size = _hydratedSizes.get(process.getPID()); |
| if (size == null) { |
| size = _unhydratedSizes.get(process.getPID()); |
| } |
| if (size != null && size.longValue() > 0) { |
| hydratedFootprint += size; |
| } |
| } |
| return hydratedFootprint; |
| } |
| |
| public long getHydratedProcessSize(QName processName) { |
| return getHydratedProcessSize(_activeProcesses.get(processName)); |
| } |
| |
| private long getHydratedProcessSize(BpelProcess process) { |
| long potentialGrowth = 0; |
| if (!process.hintIsHydrated()) { |
| Long mySize = _hydratedSizes.get(process.getPID()); |
| if (mySize == null) { |
| mySize = _unhydratedSizes.get(process.getPID()); |
| } |
| if (mySize != null && mySize.longValue() > 0) { |
| potentialGrowth = mySize.longValue(); |
| } |
| } |
| return getHydratedProcessSize(potentialGrowth); |
| } |
| |
| private long getHydratedProcessSize(long potentialGrowth) { |
| long processMemory = (long) |
| ((getHydratedFootprint() + potentialGrowth) * |
| PROCESS_OVERHEAD_MEMORY_FACTOR); |
| return processMemory; |
| } |
| |
| public int getHydratedProcessCount(QName processName) { |
| int processCount = 0; |
| for (BpelProcess process : _activeProcesses.values()) { |
| if (process.hintIsHydrated() || process.getPID().equals(processName)) { |
| processCount++; |
| } |
| } |
| return processCount; |
| } |
| |
| private long _processThrottledMaximumSize = Long.MAX_VALUE; |
| private int _processThrottledMaximumCount = Integer.MAX_VALUE; |
| private int _instanceThrottledMaximumCount = Integer.MAX_VALUE; |
| private boolean _hydrationThrottled = false; |
| |
| public void setInstanceThrottledMaximumCount( |
| int instanceThrottledMaximumCount) { |
| this._instanceThrottledMaximumCount = instanceThrottledMaximumCount; |
| } |
| |
| public int getInstanceThrottledMaximumCount() { |
| return _instanceThrottledMaximumCount; |
| } |
| |
| public void setProcessThrottledMaximumCount( |
| int hydrationThrottledMaximumCount) { |
| this._processThrottledMaximumCount = hydrationThrottledMaximumCount; |
| if (hydrationThrottledMaximumCount < Integer.MAX_VALUE) { |
| registerMessageExchangeInterceptor(new ProcessCountThrottler()); |
| } else { |
| unregisterMessageExchangeInterceptor(ProcessCountThrottler.class); |
| } |
| } |
| |
| public int getProcessThrottledMaximumCount() { |
| return _processThrottledMaximumCount; |
| } |
| |
| public void setProcessThrottledMaximumSize( |
| long hydrationThrottledMaximumSize) { |
| this._processThrottledMaximumSize = hydrationThrottledMaximumSize; |
| if (hydrationThrottledMaximumSize < Long.MAX_VALUE) { |
| registerMessageExchangeInterceptor(new ProcessSizeThrottler()); |
| } else { |
| unregisterMessageExchangeInterceptor(ProcessSizeThrottler.class); |
| } |
| } |
| |
| public long getProcessThrottledMaximumSize() { |
| return _processThrottledMaximumSize; |
| } |
| |
| public void setProcessSize(QName processId, boolean hydratedOnce) { |
| BpelProcess process = _activeProcesses.get(processId); |
| long processSize = process.sizeOf(); |
| if (hydratedOnce) { |
| _hydratedSizes.put(process.getPID(), Long.valueOf(processSize)); |
| _unhydratedSizes.remove(process.getPID()); |
| } else { |
| _hydratedSizes.remove(process.getPID()); |
| _unhydratedSizes.put(process.getPID(), Long.valueOf(processSize)); |
| } |
| } |
| |
| /** |
| * Returns true if the last used process was dehydrated because it was not in-use. |
| */ |
| public boolean dehydrateLastUnusedProcess() { |
| BpelProcess lastUnusedProcess = null; |
| long lastUsedMinimum = Long.MAX_VALUE; |
| for (BpelProcess process : _activeProcesses.values()) { |
| if (process.hintIsHydrated() |
| && process.getLastUsed() < lastUsedMinimum |
| && process.getInstanceInUseCount() == 0) { |
| lastUsedMinimum = process.getLastUsed(); |
| lastUnusedProcess = process; |
| } |
| } |
| if (lastUnusedProcess != null) { |
| lastUnusedProcess.dehydrate(); |
| return true; |
| } |
| return false; |
| } |
| |
| public void sendMyRoleFault(BpelProcess process, JobDetails we, int causeCode) { |
| MessageExchange mex = (MessageExchange) getMessageExchange(we.getMexId()); |
| if (!(mex instanceof MyRoleMessageExchange)) { |
| return; |
| } |
| QName faultQName = null; |
| OConstants constants = process.getOProcess().getConstants(); |
| if (constants != null) { |
| Document document = DOMUtils.newDocument(); |
| Element faultElement = document.createElementNS(Namespaces.SOAP_ENV_NS, "Fault"); |
| Element faultDetail = document.createElementNS(Namespaces.ODE_EXTENSION_NS, "fault"); |
| faultElement.appendChild(faultDetail); |
| switch (causeCode) { |
| case InvalidProcessException.TOO_MANY_PROCESSES_CAUSE_CODE: |
| faultQName = constants.getQnTooManyProcesses(); |
| faultDetail.setTextContent("The total number of processes in use is over the limit."); |
| break; |
| case InvalidProcessException.TOO_HUGE_PROCESSES_CAUSE_CODE: |
| faultQName = constants.getQnTooHugeProcesses(); |
| faultDetail.setTextContent("The total size of processes in use is over the limit"); |
| break; |
| case InvalidProcessException.TOO_MANY_INSTANCES_CAUSE_CODE: |
| faultQName = constants.getQnTooManyInstances(); |
| faultDetail.setTextContent("No more instances of the process allowed at start at this time."); |
| break; |
| case InvalidProcessException.RETIRED_CAUSE_CODE: |
| // we're invoking a target process, trying to see if we can retarget the message |
| // to the current version (only applies when it's a new process creation) |
| for (BpelProcess activeProcess : _activeProcesses.values()) { |
| if (activeProcess.getConf().getState().equals(org.apache.ode.bpel.iapi.ProcessState.ACTIVE) |
| && activeProcess.getConf().getType().equals(process.getConf().getType())) { |
| we.setProcessId(activeProcess._pid); |
| ((MyRoleMessageExchangeImpl) mex)._process = activeProcess; |
| process.handleJobDetails(we); |
| return; |
| } |
| } |
| faultQName = constants.getQnRetiredProcess(); |
| faultDetail.setTextContent("The process you're trying to instantiate has been retired."); |
| break; |
| case InvalidProcessException.DEFAULT_CAUSE_CODE: |
| default: |
| faultQName = constants.getQnUnknownFault(); |
| break; |
| } |
| MexDaoUtil.setFaulted((MessageExchangeImpl) mex, faultQName, faultElement); |
| } |
| } |
| |
| private void sendPartnerRoleFailure(JobDetails we, FailureType failureType) { |
| MessageExchange mex = (MessageExchange) getMessageExchange(we.getMexId()); |
| if (mex instanceof PartnerRoleMessageExchange) { |
| if (mex.getStatus() == MessageExchange.Status.ASYNC || mex.getStatus() == MessageExchange.Status.REQUEST) { |
| String msg = "No response received for invoke (mexId=" + we.getMexId() + "), forcing it into a failed state."; |
| if (__log.isDebugEnabled()) __log.debug(msg); |
| MexDaoUtil.setFailure((PartnerRoleMessageExchangeImpl) mex, failureType, msg, null); |
| } |
| } |
| } |
| |
| public BpelProcess getNewestProcessByType(QName processType) { |
| long v = -1; |
| BpelProcess q = null; |
| for (BpelProcess p : _activeProcesses.values()) { |
| if (p.getProcessType().equals(processType) && v < p.getVersion()) { |
| v = p.getVersion(); |
| q = p; |
| } |
| } |
| return q; |
| } |
| } |
| |
