[riftsaw-commits] riftsaw SVN: r1463 - in branches/RiftSaw-ODE-2.3.x: scheduler-simple/src/main/java/org/apache/ode/scheduler/simple and 1 other directory.

Author: jeff.yuchang Date: 2011-10-18 23:23:12 -0400 (Tue, 18 Oct 2011) New Revision: 1463 Modified: branches/RiftSaw-ODE-2.3.x/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelEngineImpl.java branches/RiftSaw-ODE-2.3.x/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelProcess.java branches/RiftSaw-ODE-2.3.x/scheduler-simple/src/main/java/org/apache/ode/scheduler/simple/SimpleScheduler.java Log: *Riftsaw-455, removte the AcquireInstanceLock logic. * Added a NPE check for the dequeue jobs. Modified: branches/RiftSaw-ODE-2.3.x/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelEngineImpl.java =================================================================== --- branches/RiftSaw-ODE-2.3.x/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelEngineImpl.java 2011-10-19 02:16:14 UTC (rev 1462) +++ branches/RiftSaw-ODE-2.3.x/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelEngineImpl.java 2011-10-19 03:23:12 UTC (rev 1463) @@ -375,6 +375,7 @@ return process.getOProcess(); } + @Deprecated 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 @@ -402,8 +403,9 @@ if( __log.isTraceEnabled() ) __log.trace("[JOB] onScheduledJob " + jobInfo + "" + we.getInstanceId()); - acquireInstanceLock(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 Modified: branches/RiftSaw-ODE-2.3.x/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelProcess.java =================================================================== --- branches/RiftSaw-ODE-2.3.x/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelProcess.java 2011-10-19 02:16:14 UTC (rev 1462) +++ branches/RiftSaw-ODE-2.3.x/bpel-runtime/src/main/java/org/apache/ode/bpel/engine/BpelProcess.java 2011-10-19 03:23:12 UTC (rev 1463) @@ -299,7 +299,7 @@ return true; } else if (routing.messageRoute != null) { // Found a route, hitting it - _engine.acquireInstanceLock(routing.messageRoute.getTargetInstance().getInstanceId()); + //_engine.acquireInstanceLock(routing.messageRoute.getTargetInstance().getInstanceId()); target.invokeInstance(mex, routing); return true; } Modified: branches/RiftSaw-ODE-2.3.x/scheduler-simple/src/main/java/org/apache/ode/scheduler/simple/SimpleScheduler.java =================================================================== --- branches/RiftSaw-ODE-2.3.x/scheduler-simple/src/main/java/org/apache/ode/scheduler/simple/SimpleScheduler.java 2011-10-19 02:16:14 UTC (rev 1462) +++ branches/RiftSaw-ODE-2.3.x/scheduler-simple/src/main/java/org/apache/ode/scheduler/simple/SimpleScheduler.java 2011-10-19 03:23:12 UTC (rev 1463) @@ -762,23 +762,25 @@ int delayedCount = 0; boolean runningLate; AbsoluteTimeDateFormat f = new AbsoluteTimeDateFormat(); - for (JobDAO j : jobs) { - // jobs might have been enqueued by #addTodoOnCommit meanwhile - if (_outstandingJobs.size() >= _todoLimit){ - if (__log.isDebugEnabled()) __log.debug("Max capacity reached: "+_outstandingJobs.size()+" jobs dispacthed i.e. queued or being executed"); - break; - } - runningLate = (j.getScheduledDate() <= delayedTime); - if (runningLate) { - delayedCount++; - } - if (__log.isDebugEnabled()) - __log.debug("todo.enqueue job from db: " + j.getJobId().trim() + " for " + j.getScheduledDate() + "(" + f.format(j.getScheduledDate())+") "+(runningLate?" delayed=true":"")); - enqueue(j); + if (jobs != null && jobs.size() > 0) { + for (JobDAO j : jobs) { + // jobs might have been enqueued by #addTodoOnCommit meanwhile + if (_outstandingJobs.size() >= _todoLimit){ + if (__log.isDebugEnabled()) __log.debug("Max capacity reached: "+_outstandingJobs.size()+" jobs dispacthed i.e. queued or being executed"); + break; + } + runningLate = (j.getScheduledDate() <= delayedTime); + if (runningLate) { + delayedCount++; + } + if (__log.isDebugEnabled()) + __log.debug("todo.enqueue job from db: " + j.getJobId().trim() + " for " + j.getScheduledDate() + "(" + f.format(j.getScheduledDate())+") "+(runningLate?" delayed=true":"")); + enqueue(j); + } + if (delayedCount > 0) { + __log.warn("Dispatching jobs with more than "+(_warningDelay/60000)+" minutes delay. Either the server was down for some time or the job load is greater than available capacity"); + } } - if (delayedCount > 0) { - __log.warn("Dispatching jobs with more than "+(_warningDelay/60000)+" minutes delay. Either the server was down for some time or the job load is greater than available capacity"); - } // clear only if the batch succeeded _processedSinceLastLoadTask.clear();