[jbpm-commits] JBoss JBPM SVN: r5405 - jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor.

[do-not-reply at jboss.org]

Author: alex.guizar at jboss.com
Date: 2009-07-31 03:12:07 -0400 (Fri, 31 Jul 2009)
New Revision: 5405

Modified:
   jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutor.java
   jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobParcel.java
Log:
[JBPM-2402] verify job executor threads in case of errors 
pass bounded queue to job thread pool

Modified: jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutor.java
===================================================================
--- jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutor.java	2009-07-31 06:42:58 UTC (rev 5404)
+++ jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobExecutor.java	2009-07-31 07:12:07 UTC (rev 5405)
@@ -28,8 +28,11 @@
 import java.util.Collection;
 import java.util.Date;
 import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
 import org.jbpm.api.JbpmException;
@@ -37,7 +40,6 @@
 import org.jbpm.pvm.internal.cmd.Command;
 import org.jbpm.pvm.internal.cmd.CommandService;
 
-
 /** manager for job execution threads and their configuration.
  * 
  * @author Tom Baeyens, Guillaume Porcher
@@ -53,10 +55,10 @@
   CommandService commandService;
   String name = "JobExecutor-"+getHostName();
   int nbrOfThreads = 3;
-  int idleMillis = 5000; // default normal poll interval is 5 seconds
-  int idleMillisMax = 1000*60*2; // default max poll interval in case of continuous exceptions is 2 minutes
+  int idleMillis = 5*1000; // default normal poll interval is 5 seconds
+  int idleMillisMax = 5*60*1000; // default max poll interval in case of continuous exceptions is 5 minutes
   int historySize = 200;
-  int lockMillis = 1000*60*30; // default max lock time is 30 minutes
+  int lockMillis = 30*60*1000; // default max lock time is 30 minutes
 
   // runtime state
 
@@ -81,7 +83,8 @@
       
       isActive = true;
       log.trace("starting thread pool for job executor '"+name+"'...");
-      threadPool = Executors.newFixedThreadPool(nbrOfThreads);
+      threadPool = new ThreadPoolExecutor(nbrOfThreads, nbrOfThreads, 0L, TimeUnit.MILLISECONDS,
+          new ArrayBlockingQueue<Runnable>(nbrOfThreads), JobRejectionHandler.INSTANCE);
 
       log.trace("starting dispatcher thread for job executor '"+name+"'...");
       dispatcherThread = new DispatcherThread(this);
@@ -91,7 +94,20 @@
       log.trace("ignoring start: job executor '"+name+"' is already started'");
     }
   }
-  
+
+  static final class JobRejectionHandler implements RejectedExecutionHandler {
+    
+    static final JobRejectionHandler INSTANCE = new JobRejectionHandler();
+
+    public void rejectedExecution(Runnable task, ThreadPoolExecutor executor) {
+      try {
+        executor.getQueue().put(task);
+      } catch (InterruptedException e) {
+        throw new RejectedExecutionException("queuing " + task + " got interrupted", e);
+      }
+    }
+  }
+
   /** stops with join set to false.
    * @see #stop(boolean) */ 
   public synchronized void stop() {

Modified: jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobParcel.java
===================================================================
--- jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobParcel.java	2009-07-31 06:42:58 UTC (rev 5404)
+++ jbpm4/trunk/modules/pvm/src/main/java/org/jbpm/pvm/internal/jobexecutor/JobParcel.java	2009-07-31 07:12:07 UTC (rev 5405)
@@ -51,4 +51,9 @@
       log.error("exception in job block", e);
     }
   }
+
+  @Override
+  public String toString() {
+    return JobParcel.class.getSimpleName() + jobDbids;
+  }
 }