Re: svn commit: r1362422 - in /ofbiz/trunk/framework: service/config/ service/dtd/ service/src/org/ofbiz/service/
Locked
Re: svn commit: r1362422 - in /ofbiz/trunk/framework: service/config/ service/dtd/ service/src/org/ofbiz/service/
 
Administrator
|
Hi,
While reviewing Eclipse found that JobPoller misses some Map and List types arguments Eclipse autocompletion suggests the following change: Index: framework/service/src/org/ofbiz/service/job/JobPoller.java =================================================================== --- framework/service/src/org/ofbiz/service/job/JobPoller.java (revision 1369293) +++ framework/service/src/org/ofbiz/service/job/JobPoller.java (working copy) @@ -83,7 +83,7 @@ } public Map<String, Object> getPoolState() { - Map poolState = new HashMap(); + Map<String, Object> poolState = new HashMap<String, Object>(); poolState.put("pollerName", this.name); poolState.put("pollerThreadName", "OFBiz-JobPoller-" + this.name); poolState.put("invokerThreadNameFormat", "OFBiz-JobInvoker-" + this.name + "-<SEQ>"); @@ -95,12 +95,12 @@ poolState.put("greatestNumberOfInvokerThreads", this.executor.getLargestPoolSize()); poolState.put("numberOfCompletedTasks", this.executor.getCompletedTaskCount()); BlockingQueue<Runnable> queue = this.executor.getQueue(); - List taskList = new ArrayList(); - Map taskInfo = null; + List<Map> taskList = new ArrayList<Map>(); + Map<String, Comparable> taskInfo = null; for (Runnable task : queue) { if (task instanceof JobInvoker) { JobInvoker jobInvoker = (JobInvoker) task; - taskInfo = new HashMap(); + taskInfo = new HashMap<String, Comparable>(); taskInfo.put("id", jobInvoker.getJobId()); taskInfo.put("name", jobInvoker.getJobName()); taskInfo.put("serviceName", jobInvoker.getServiceName()); Methods like executor.getCorePoolSize() return int/s and methods like jobInvoker.getTime() return long/s. It should be ok, thanks to autoboxing, and it compiles of course. But I wonder why Jacopo did not do it, any reasons or just a miss? Jacques From: <[hidden email]> > Author: jacopoc > Date: Tue Jul 17 09:20:42 2012 > New Revision: 1362422 > > URL: http://svn.apache.org/viewvc?rev=1362422&view=rev > Log: > Refactored the very old code in the org.ofbiz.service.job package to leverage the Executor framework of java.util.concurrent > rather than relying on ad-hoc synchronization and thread management: the new code is simpler, smaller and it is now easier to > review and change the execution policy; two attributes of service-config.xsd have been deprecated as they are not used in the new > version ("jobs" and "wait-millis"). > > Added: > ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThread.java > ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThreadFactory.java > Modified: > ofbiz/trunk/framework/service/config/serviceengine.xml > ofbiz/trunk/framework/service/dtd/service-config.xsd > ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvoker.java > ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobManager.java > ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobPoller.java > ofbiz/trunk/framework/webtools/webapp/webtools/WEB-INF/actions/service/Threads.groovy > ofbiz/trunk/framework/webtools/widget/ServiceForms.xml > ofbiz/trunk/framework/webtools/widget/ServiceScreens.xml > > Modified: ofbiz/trunk/framework/service/config/serviceengine.xml > URL: > http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/config/serviceengine.xml?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/service/config/serviceengine.xml (original) > +++ ofbiz/trunk/framework/service/config/serviceengine.xml Tue Jul 17 09:20:42 2012 > @@ -30,8 +30,6 @@ under the License. > purge-job-days="4" > failed-retry-min="3" > ttl="18000000" > - wait-millis="750" > - jobs="10" > min-threads="5" > max-threads="15" > poll-enabled="true" > > Modified: ofbiz/trunk/framework/service/dtd/service-config.xsd > URL: http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/dtd/service-config.xsd?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/service/dtd/service-config.xsd (original) > +++ ofbiz/trunk/framework/service/dtd/service-config.xsd Tue Jul 17 09:20:42 2012 > @@ -68,8 +68,8 @@ under the License. > <xs:attribute type="xs:nonNegativeInteger" name="purge-job-days" default="30"/> > <xs:attribute type="xs:nonNegativeInteger" name="failed-retry-min" default="30"/> > <xs:attribute type="xs:nonNegativeInteger" name="ttl" use="required"/> > - <xs:attribute type="xs:nonNegativeInteger" name="wait-millis" use="required"/> > - <xs:attribute type="xs:nonNegativeInteger" name="jobs" use="required"/> > + <xs:attribute type="xs:nonNegativeInteger" name="wait-millis"/> <!-- deprecated --> > + <xs:attribute type="xs:nonNegativeInteger" name="jobs"/> <!-- deprecated --> > <xs:attribute type="xs:nonNegativeInteger" name="min-threads" use="required"/> > <xs:attribute type="xs:nonNegativeInteger" name="max-threads" use="required"/> > <xs:attribute name="poll-enabled" default="true"> > > Modified: ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvoker.java > URL: > http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvoker.java?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvoker.java (original) > +++ ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvoker.java Tue Jul 17 09:20:42 2012 > @@ -20,12 +20,9 @@ package org.ofbiz.service.job; > > import java.util.Date; > > -import org.apache.commons.lang.math.NumberUtils; > import org.ofbiz.base.util.Debug; > -import org.ofbiz.base.util.UtilDateTime; > import org.ofbiz.entity.transaction.GenericTransactionException; > import org.ofbiz.entity.transaction.TransactionUtil; > -import org.ofbiz.service.config.ServiceConfigUtil; > > /** > * JobInvoker > @@ -33,82 +30,20 @@ import org.ofbiz.service.config.ServiceC > public class JobInvoker implements Runnable { > > public static final String module = JobInvoker.class.getName(); > - public static final long THREAD_TTL = 18000000; > - public static final int WAIT_TIME = 750; > > - private JobPoller jp = null; > - private Thread thread = null; > private Date created = null; > - private String name = null; > - private int count = 0; > - private int wait = 0; > - > - private volatile boolean run = false; > - private volatile Job currentJob = null; > - private volatile int statusCode = 0; > - private volatile long jobStart = 0; > + private long jobStart; > > - public JobInvoker(JobPoller jp) { > - this(jp, WAIT_TIME); > - } > + private Job currentJob = null; > > - public JobInvoker(JobPoller jp, int wait) { > + public JobInvoker(Job job) { > this.created = new Date(); > - this.run = true; > - this.count = 0; > - this.jp = jp; > - this.wait = wait; > - > - // service dispatcher delegator name (for thread name) > - String delegatorName = jp.getManager().getDelegator().getDelegatorName(); > - > - // get a new thread > - this.thread = new Thread(this); > - this.name = delegatorName + "-invoker-" + this.thread.getName(); > - > - this.thread.setDaemon(false); > - this.thread.setName(this.name); > - > - if (Debug.verboseOn()) Debug.logVerbose("JobInvoker: Starting Invoker Thread -- " + thread.getName(), module); > - this.thread.start(); > + this.currentJob = job; > } > > protected JobInvoker() {} > > /** > - * Tells the thread to stop after the next job. > - */ > - public void stop() { > - run = false; > - } > - > - /** > - * Wakes up this thread. > - */ > - public void wakeUp() { > - notifyAll(); > - } > - > - /** > - * Gets the number of times this thread was used. > - * @return The number of times used. > - */ > - public int getUsage() { > - return count; > - } > - > - /** > - * Gets the remaining time this thread has before it is killed > - * @return Time in millis remaining > - */ > - public long getTimeRemaining() { > - long now = UtilDateTime.nowTimestamp().getTime(); > - long time = getTime(); > - long ttl = getTTL(); > - return (time + ttl) - now; > - } > - > - /** > * Gets the time when this thread was created. > * @return Time in milliseconds when this was created. > */ > @@ -117,24 +52,8 @@ public class JobInvoker implements Runna > } > > /** > - * Gets the name of this JobInvoker. > - * @return Name of the invoker. > - */ > - public String getName() { > - return this.name; > - } > - > - /** > - * Gets the status code for this thread (0 = sleeping, 1 = running job) > - * @return 0 for sleeping or 1 when running a job. > - */ > - public int getCurrentStatus() { > - return this.statusCode; > - } > - > - /** > * Gets the total time the current job has been running or 0 when sleeping. > - * @return Total time the curent job has been running. > + * @return Total time the current job has been running. > */ > public long getCurrentRuntime() { > if (this.jobStart > 0) { > @@ -145,27 +64,15 @@ public class JobInvoker implements Runna > } > } > > - public Long getThreadId() { > - if (this.thread != null) { > - return this.thread.getId(); > - } else { > - return null; > - } > - } > - > /** > * Get the current running job's ID. > * @return String ID of the current running job. > */ > public String getJobId() { > - if (this.statusCode == 1) { > - if (this.currentJob != null) { > - return this.currentJob.getJobId(); > - } else { > - return "WARNING: Invalid Job!"; > - } > + if (this.currentJob != null) { > + return this.currentJob.getJobId(); > } else { > - return null; > + return "WARNING: Invalid Job!"; > } > } > > @@ -174,14 +81,10 @@ public class JobInvoker implements Runna > * @return String name of the current running job. > */ > public String getJobName() { > - if (this.statusCode == 1) { > - if (this.currentJob != null) { > - return this.currentJob.getJobName(); > - } else { > - return "WARNING: Invalid Job!"; > - } > + if (this.currentJob != null) { > + return this.currentJob.getJobName(); > } else { > - return null; > + return "WARNING: Invalid Job!"; > } > } > > @@ -191,101 +94,45 @@ public class JobInvoker implements Runna > */ > public String getServiceName() { > String serviceName = null; > - if (this.statusCode == 1) { > - if (this.currentJob != null) { > - if (this.currentJob instanceof GenericServiceJob) { > - GenericServiceJob gsj = (GenericServiceJob) this.currentJob; > - try { > - serviceName = gsj.getServiceName(); > - } catch (InvalidJobException e) { > - Debug.logError(e, module); > - } > + if (this.currentJob != null) { > + if (this.currentJob instanceof GenericServiceJob) { > + GenericServiceJob gsj = (GenericServiceJob) this.currentJob; > + try { > + serviceName = gsj.getServiceName(); > + } catch (InvalidJobException e) { > + Debug.logError(e, module); > } > } > } > return serviceName; > } > > - /** > - * Kill this invoker thread.s > - */ > - public void kill() { > - this.stop(); > - this.statusCode = -1; > - this.thread.interrupt(); > - this.thread = null; > - } > + public void run() { > + // setup the current job settings > + this.jobStart = System.currentTimeMillis(); > > - public synchronized void run() { > - while (run) { > - Job job = jp.next(); > - > - if (job == null) { > - try { > - java.lang.Thread.sleep(wait); > - } catch (InterruptedException ie) { > - Debug.logError(ie, "JobInvoker.run() : InterruptedException", module); > - stop(); > - } > - } else { > - Debug.logInfo("Invoker [" + thread.getName() + "] received job [" + job.getJobName() + "] from poller [" + > jp.toString() + "]", module); > - > - // setup the current job settings > - this.currentJob = job; > - this.statusCode = 1; > - this.jobStart = System.currentTimeMillis(); > - > - // execute the job > - if (Debug.verboseOn()) Debug.logVerbose("Invoker: " + thread.getName() + " executing job -- " + job.getJobName(), > module); > - try { > - job.exec(); > - } catch (InvalidJobException e) { > - Debug.logWarning(e.getMessage(), module); > - } > - if (Debug.verboseOn()) Debug.logVerbose("Invoker: " + thread.getName() + " finished executing job -- " + > job.getJobName(), module); > - > - // clear the current job settings > - this.currentJob = null; > - this.statusCode = 0; > - this.jobStart = 0; > - > - // sanity check; make sure we don't have any transactions in place > - try { > - // roll back current TX first > - if (TransactionUtil.isTransactionInPlace()) { > - Debug.logWarning("*** NOTICE: JobInvoker finished w/ a transaction in place! Rolling back.", module); > - TransactionUtil.rollback(); > - } > - > - // now resume/rollback any suspended txs > - if (TransactionUtil.suspendedTransactionsHeld()) { > - int suspended = TransactionUtil.cleanSuspendedTransactions(); > - Debug.logWarning("Resumed/Rolled Back [" + suspended + "] transactions.", module); > - } > - } catch (GenericTransactionException e) { > - Debug.logWarning(e, module); > - } > - > - // increment the count > - count++; > - if (Debug.verboseOn()) Debug.logVerbose("Invoker: " + thread.getName() + " (" + count + ") total.", module); > - } > - long diff = (new Date().getTime() - this.getTime()); > - > - if (getTTL() > 0 && diff > getTTL()) > - jp.removeThread(this); > + // execute the job > + try { > + this.currentJob.exec(); > + } catch (InvalidJobException e) { > + Debug.logWarning(e.getMessage(), module); > } > - if (Debug.verboseOn()) Debug.logVerbose("Invoker: " + thread.getName() + " dead -- " + UtilDateTime.nowTimestamp(), > module); > - } > - > - private long getTTL() { > - long ttl = THREAD_TTL; > > + // sanity check; make sure we don't have any transactions in place > try { > - ttl = NumberUtils.toLong(ServiceConfigUtil.getElementAttr("thread-pool", "ttl")); > - } catch (NumberFormatException nfe) { > - Debug.logError("Problems reading value from attribute [ttl] of element [thread-pool] in serviceengine.xml file [" + > nfe.toString() + "]. Using default (" + THREAD_TTL + ").", module); > + // roll back current TX first > + if (TransactionUtil.isTransactionInPlace()) { > + Debug.logWarning("*** NOTICE: JobInvoker finished w/ a transaction in place! Rolling back.", module); > + TransactionUtil.rollback(); > + } > + > + // now resume/rollback any suspended txs > + if (TransactionUtil.suspendedTransactionsHeld()) { > + int suspended = TransactionUtil.cleanSuspendedTransactions(); > + Debug.logWarning("Resumed/Rolled Back [" + suspended + "] transactions.", module); > + } > + } catch (GenericTransactionException e) { > + Debug.logWarning(e, module); > } > - return ttl; > } > } > > Added: ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThread.java > URL: > http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThread.java?rev=1362422&view=auto > ============================================================================== > --- ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThread.java (added) > +++ ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThread.java Tue Jul 17 09:20:42 2012 > @@ -0,0 +1,29 @@ > +/******************************************************************************* > + * 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.ofbiz.service.job; > + > +import java.util.concurrent.atomic.AtomicInteger; > + > +public class JobInvokerThread extends Thread { > + private static final AtomicInteger created = new AtomicInteger(); > + > + public JobInvokerThread(Runnable runnable, String poolName) { > + super(runnable, "OFBiz-JobInvoker-" + poolName + "-" + created.getAndIncrement()); > + } > +} > > Added: ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThreadFactory.java > URL: > http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThreadFactory.java?rev=1362422&view=auto > ============================================================================== > --- ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThreadFactory.java (added) > +++ ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThreadFactory.java Tue Jul 17 09:20:42 2012 > @@ -0,0 +1,33 @@ > +/******************************************************************************* > + * 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.ofbiz.service.job; > + > +import java.util.concurrent.ThreadFactory; > + > +public class JobInvokerThreadFactory implements ThreadFactory { > + private final String poolName; > + > + public JobInvokerThreadFactory(String poolName) { > + this.poolName = poolName; > + } > + > + public Thread newThread(Runnable runnable) { > + return new JobInvokerThread(runnable, poolName); > + } > +} > > Modified: ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobManager.java > URL: > http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobManager.java?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobManager.java (original) > +++ ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobManager.java Tue Jul 17 09:20:42 2012 > @@ -423,18 +423,10 @@ public class JobManager { > } > > /** > - * Kill a JobInvoker Thread. > - * @param threadName Name of the JobInvoker Thread to kill. > - */ > - public void killThread(String threadName) { > - jp.killThread(threadName); > - } > - > - /** > * Get a List of each threads current state. > * @return List containing a Map of each thread's state. > */ > - public List<Map<String, Object>> processList() { > + public Map<String, Object> getPoolState() { > return jp.getPoolState(); > } > > > Modified: ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobPoller.java > URL: > http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobPoller.java?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobPoller.java (original) > +++ ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobPoller.java Tue Jul 17 09:20:42 2012 > @@ -18,15 +18,20 @@ > *******************************************************************************/ > package org.ofbiz.service.job; > > +import java.util.ArrayList; > +import java.util.HashMap; > import java.util.List; > import java.util.Map; > - > -import javolution.util.FastList; > -import javolution.util.FastMap; > +import java.util.concurrent.BlockingQueue; > +import java.util.concurrent.LinkedBlockingQueue; > +import java.util.concurrent.ThreadPoolExecutor; > +import java.util.concurrent.TimeUnit; > > import org.ofbiz.base.util.Debug; > import org.ofbiz.service.config.ServiceConfigUtil; > > +import org.apache.commons.lang.math.NumberUtils; > + > /** > * JobPoller - Polls for persisted jobs to run. > */ > @@ -36,27 +41,28 @@ public class JobPoller implements Runnab > > public static final int MIN_THREADS = 1; > public static final int MAX_THREADS = 15; > - public static final int MAX_JOBS = 3; > public static final int POLL_WAIT = 20000; > - //public static final long MAX_TTL = 18000000; > + public static final long THREAD_TTL = 18000000; > > - protected Thread thread = null; > - protected List<JobInvoker> pool = null; > - protected List<Job> run = null; > - protected JobManager jm = null; > - > - protected volatile boolean isRunning = false; > + private Thread thread = null; > + private JobManager jm = null; > + private ThreadPoolExecutor executor = null; > + private String name = null; > > /** > * Creates a new JobScheduler > * @param jm JobManager associated with this scheduler > */ > public JobPoller(JobManager jm, boolean enabled) { > + this.name = (jm.getDelegator() != null? jm.getDelegator().getDelegatorName(): "NA"); > this.jm = jm; > - this.run = FastList.newInstance(); > - > - // create the thread pool > - this.pool = createThreadPool(); > + this.executor = new ThreadPoolExecutor(minThreads(), > + maxThreads(), > + getTTL(), > + TimeUnit.MILLISECONDS, > + new LinkedBlockingQueue<Runnable>(), > + new JobInvokerThreadFactory(this.name), > + new ThreadPoolExecutor.AbortPolicy()); > > if (enabled) { > // re-load crashed jobs > @@ -66,11 +72,10 @@ public class JobPoller implements Runnab > if (pollEnabled()) { > > // create the poller thread > - thread = new Thread(this, this.toString()); > + thread = new Thread(this, "OFBiz-JobPoller-" + this.name); > thread.setDaemon(false); > > // start the poller > - this.isRunning = true; > thread.start(); > } > } > @@ -84,7 +89,7 @@ public class JobPoller implements Runnab > java.lang.Thread.sleep(30000); > } catch (InterruptedException e) { > } > - while (isRunning) { > + while (!executor.isShutdown()) { > try { > // grab a list of jobs to run. > List<Job> pollList = jm.poll(); > @@ -106,140 +111,77 @@ public class JobPoller implements Runnab > } > > /** > - * Returns the JobManager > + * Adds a job to the RUN queue > */ > - public JobManager getManager() { > - return jm; > + public void queueNow(Job job) { > + this.executor.execute(new JobInvoker(job)); > } > > /** > * Stops the JobPoller > */ > - public void stop() { > - isRunning = false; > - destroyThreadPool(); > - } > - > - public List<Map<String, Object>> getPoolState() { > - List<Map<String, Object>> stateList = FastList.newInstance(); > - for (JobInvoker invoker: this.pool) { > - Map<String, Object> stateMap = FastMap.newInstance(); > - stateMap.put("threadName", invoker.getName()); > - stateMap.put("threadId", invoker.getThreadId()); > - stateMap.put("jobName", invoker.getJobName()); > - stateMap.put("serviceName", invoker.getServiceName()); > - stateMap.put("usage", invoker.getUsage()); > - stateMap.put("ttl", invoker.getTimeRemaining()); > - stateMap.put("runTime", invoker.getCurrentRuntime()); > - stateMap.put("status", invoker.getCurrentStatus()); > - stateList.add(stateMap); > - } > - return stateList; > - } > - > - /** > - * Stops all threads in the threadPool and clears > - * the pool as final step. > - */ > - private void destroyThreadPool() { > - Debug.logInfo("Destroying thread pool...", module); > - for (JobInvoker ji: pool) { > - ji.stop(); > - } > - pool.clear(); > - } > - > - public synchronized void killThread(String threadName) { > - JobInvoker inv = findThread(threadName); > - if (inv != null) { > - inv.kill(); > - this.pool.remove(inv); > - } > - } > - > - private JobInvoker findThread(String threadName) { > - for (JobInvoker inv: pool) { > - if (threadName.equals(inv.getName())) { > - return inv; > - } > - } > - return null; > - } > - > - /** > - * Returns the next job to run > - */ > - public Job next() { > - if (run.size() > 0) { > - // NOTE: this syncrhonized isn't really necessary as the only method that calls it is already synchronized (the > JobInvoker.run method), so this is here as an added protection especially for the case where it might be used differently in the > future > - synchronized (run) { > - // make sure the size is still greater than zero > - if (run.size() > 0) { > - return run.remove(0); > + void stop() { > + Debug.logInfo("Shutting down thread pool for " + this.name, module); > + this.executor.shutdown(); > + try { > + // Wait 60 seconds for existing tasks to terminate > + if (!this.executor.awaitTermination(60, TimeUnit.SECONDS)) { > + // abrupt shutdown (cancel currently executing tasks) > + Debug.logInfo("Attempting abrupt shut down of thread pool for " + this.name, module); > + this.executor.shutdownNow(); > + // Wait 60 seconds for tasks to respond to being cancelled > + if (!this.executor.awaitTermination(60, TimeUnit.SECONDS)) { > + Debug.logWarning("Unable to shutdown the thread pool for " + this.name, module); > } > } > + } catch (InterruptedException ie) { > + // re cancel if current thread was also interrupted > + this.executor.shutdownNow(); > + // preserve interrupt status > + Thread.currentThread().interrupt(); > } > - return null; > + Debug.logInfo("Shutdown completed of thread pool for " + this.name, module); > } > > /** > - * Adds a job to the RUN queue > + * Returns the JobManager > */ > - public void queueNow(Job job) { > - //Debug.logInfo("[" + Thread.currentThread().getId() + "] Begin queueNow; holds run lock? " + Thread.holdsLock(run), > module); > - > - // NOTE DEJ20071201 MUST use a different object for the lock here because the "this" object is always held by the poller > thread in the run method above (which sleeps and runs) > - synchronized (run) { > - run.add(job); > - } > - if (Debug.verboseOn()) Debug.logVerbose("New run queue size: " + run.size(), module); > - if (run.size() > pool.size() && pool.size() < maxThreads()) { > - synchronized (pool) { > - if (run.size() > pool.size() && pool.size() < maxThreads()) { > - int calcSize = (run.size() / jobsPerThread()) - (pool.size()); > - int addSize = calcSize > maxThreads() ? maxThreads() : calcSize; > - > - for (int i = 0; i < addSize; i++) { > - JobInvoker iv = new JobInvoker(this, invokerWaitTime()); > - pool.add(iv); > - } > - } > - } > - } > + public JobManager getManager() { > + return jm; > } > > - /** > - * Removes a thread from the pool. > - * @param invoker The invoker to remove. > - */ > - public void removeThread(JobInvoker invoker) { > - if (pool != null) { > - synchronized (pool) { > - pool.remove(invoker); > - invoker.stop(); > - } > - } > - > - if (pool != null && pool.size() < minThreads()) { > - synchronized (pool) { > - for (int i = 0; i < minThreads() - pool.size(); i++) { > - JobInvoker iv = new JobInvoker(this, invokerWaitTime()); > - pool.add(iv); > - } > + public Map<String, Object> getPoolState() { > + Map poolState = new HashMap(); > + poolState.put("pollerName", this.name); > + poolState.put("pollerThreadName", "OFBiz-JobPoller-" + this.name); > + poolState.put("invokerThreadNameFormat", "OFBiz-JobInvoker-" + this.name + "-<SEQ>"); > + poolState.put("keepAliveTimeInSeconds", this.executor.getKeepAliveTime(TimeUnit.SECONDS)); > + > + poolState.put("numberOfCoreInvokerThreads", this.executor.getCorePoolSize()); > + poolState.put("currentNumberOfInvokerThreads", this.executor.getPoolSize()); > + poolState.put("numberOfActiveInvokerThreads", this.executor.getActiveCount()); > + poolState.put("maxNumberOfInvokerThreads", this.executor.getMaximumPoolSize()); > + poolState.put("greatestNumberOfInvokerThreads", this.executor.getLargestPoolSize()); > + > + poolState.put("numberOfCompletedTasks", this.executor.getCompletedTaskCount()); > + > + BlockingQueue<Runnable> queue = this.executor.getQueue(); > + List taskList = new ArrayList(); > + Map taskInfo = null; > + for (Runnable task: queue) { > + if (task instanceof JobInvoker) { > + JobInvoker jobInvoker = (JobInvoker)task; > + taskInfo = new HashMap(); > + taskInfo.put("id", jobInvoker.getJobId()); > + taskInfo.put("name", jobInvoker.getJobName()); > + taskInfo.put("serviceName", jobInvoker.getServiceName()); > + taskInfo.put("time", jobInvoker.getTime()); > + taskInfo.put("runtime", jobInvoker.getCurrentRuntime()); > + taskList.add(taskInfo); > } > } > - } > - > - // Creates the invoker pool > - private List<JobInvoker> createThreadPool() { > - List<JobInvoker> threadPool = FastList.newInstance(); > - > - while (threadPool.size() < minThreads()) { > - JobInvoker iv = new JobInvoker(this, invokerWaitTime()); > - threadPool.add(iv); > - } > - > - return threadPool; > + poolState.put("taskList", taskList); > + return poolState; > } > > private int maxThreads() { > @@ -264,37 +206,26 @@ public class JobPoller implements Runnab > return min; > } > > - private int jobsPerThread() { > - int jobs = MAX_JOBS; > - > - try { > - jobs = Integer.parseInt(ServiceConfigUtil.getElementAttr("thread-pool", "jobs")); > - } catch (NumberFormatException nfe) { > - Debug.logError("Problems reading values from serviceengine.xml file [" + nfe.toString() + "]. Using defaults.", > module); > - } > - return jobs; > - } > - > - private int invokerWaitTime() { > - int wait = JobInvoker.WAIT_TIME; > + private int pollWaitTime() { > + int poll = POLL_WAIT; > > try { > - wait = Integer.parseInt(ServiceConfigUtil.getElementAttr("thread-pool", "wait-millis")); > + poll = Integer.parseInt(ServiceConfigUtil.getElementAttr("thread-pool", "poll-db-millis")); > } catch (NumberFormatException nfe) { > Debug.logError("Problems reading values from serviceengine.xml file [" + nfe.toString() + "]. Using defaults.", > module); > } > - return wait; > + return poll; > } > > - private int pollWaitTime() { > - int poll = POLL_WAIT; > + private long getTTL() { > + long ttl = THREAD_TTL; > > try { > - poll = Integer.parseInt(ServiceConfigUtil.getElementAttr("thread-pool", "poll-db-millis")); > + ttl = NumberUtils.toLong(ServiceConfigUtil.getElementAttr("thread-pool", "ttl")); > } catch (NumberFormatException nfe) { > - Debug.logError("Problems reading values from serviceengine.xml file [" + nfe.toString() + "]. Using defaults.", > module); > + Debug.logError("Problems reading value from attribute [ttl] of element [thread-pool] in serviceengine.xml file [" + > nfe.toString() + "]. Using default (" + THREAD_TTL + ").", module); > } > - return poll; > + return ttl; > } > > private boolean pollEnabled() { > > Modified: ofbiz/trunk/framework/webtools/webapp/webtools/WEB-INF/actions/service/Threads.groovy > URL: > http://svn.apache.org/viewvc/ofbiz/trunk/framework/webtools/webapp/webtools/WEB-INF/actions/service/Threads.groovy?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/webtools/webapp/webtools/WEB-INF/actions/service/Threads.groovy (original) > +++ ofbiz/trunk/framework/webtools/webapp/webtools/WEB-INF/actions/service/Threads.groovy Tue Jul 17 09:20:42 2012 > @@ -35,19 +35,9 @@ uiLabelMap = UtilProperties.getResourceB > uiLabelMap.addBottomResourceBundle("CommonUiLabels"); > > threads = []; > -jobs = dispatcher.getJobManager().processList(); > -jobs.each { job -> > - state = job.status; > - switch (state) { > - case 0 : status = uiLabelMap.WebtoolsStatusSleeping; break; > - case 1 : status = uiLabelMap.WebtoolsStatusRunning; break; > - case -1: status = uiLabelMap.WebtoolsStatusShuttingDown; break; > - default: status = uiLabelMap.WebtoolsStatusInvalid; break; > - } > - job.status = status; > - threads.add(job); > -} > -context.threads = threads; > +poolState = dispatcher.getJobManager().getPoolState(); > +context.poolState = poolState; > +context.threads = poolState.taskList; > > // Some stuff for general threads on the server > currentThread = Thread.currentThread(); > > Modified: ofbiz/trunk/framework/webtools/widget/ServiceForms.xml > URL: > http://svn.apache.org/viewvc/ofbiz/trunk/framework/webtools/widget/ServiceForms.xml?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/webtools/widget/ServiceForms.xml (original) > +++ ofbiz/trunk/framework/webtools/widget/ServiceForms.xml Tue Jul 17 09:20:42 2012 > @@ -110,14 +110,24 @@ under the License. > <field name="key"><display/></field> > <field name="value"><display/></field> > </form> > + <form name="PoolState" type="single" default-map-name="poolState"> > + <field name="pollerName"><display/></field> > + <field name="pollerThreadName"><display/></field> > + <field name="invokerThreadNameFormat"><display/></field> > + <field name="keepAliveTimeInSeconds"><display/></field> > + <field name="numberOfCoreInvokerThreads"><display/></field> > + <field name="currentNumberOfInvokerThreads"><display/></field> > + <field name="numberOfActiveInvokerThreads"><display/></field> > + <field name="maxNumberOfInvokerThreads"><display/></field> > + <field name="greatestNumberOfInvokerThreads"><display/></field> > + <field name="numberOfCompletedTasks"><display/></field> > + </form> > <form name="ListJavaThread" type="list" list-name="threads" paginate-target="threadList" separate-columns="true" > odd-row-style="alternate-row" default-table-style="basic-table hover-bar"> > - <field name="threadId" title="${uiLabelMap.WebtoolsThread}"><display description="${threadId} ${threadName}"/></field> > - <field name="status" title="${uiLabelMap.CommonStatus}"><display/></field> > - <field name="jobName" title="${uiLabelMap.WebtoolsJob}"><display default-value="${uiLabelMap.CommonNone}"/></field> > + <field name="id" title="${uiLabelMap.WebtoolsThread}"><display description="${threadId} ${threadName}"/></field> > + <field name="name" title="${uiLabelMap.WebtoolsJob}"><display default-value="${uiLabelMap.CommonNone}"/></field> > <field name="serviceName" title="${uiLabelMap.WebtoolsService}"><display > default-value="${uiLabelMap.CommonNone}"/></field> > - <field name="usage" title="${uiLabelMap.WebtoolsUsage}"><display/></field> > - <field name="ttl" title="${uiLabelMap.WebtoolsTTL} (ms)"><display/></field> > + <field name="time"><display/></field> > <field name="runTime" title="${uiLabelMap.CommonTime} (ms)"><display/></field> > </form> > <form name="ListServices" type="list" list-name="services" paginate-target="ServiceLog" separate-columns="true" > > Modified: ofbiz/trunk/framework/webtools/widget/ServiceScreens.xml > URL: > http://svn.apache.org/viewvc/ofbiz/trunk/framework/webtools/widget/ServiceScreens.xml?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/webtools/widget/ServiceScreens.xml (original) > +++ ofbiz/trunk/framework/webtools/widget/ServiceScreens.xml Tue Jul 17 09:20:42 2012 > @@ -103,6 +103,7 @@ under the License. > <decorator-screen name="CommonServiceDecorator" location="${parameters.mainDecoratorLocation}"> > <decorator-section name="body"> > <screenlet title="${uiLabelMap.WebtoolsServiceEngineThreads}"> > + <include-form name="PoolState" location="component://webtools/widget/ServiceForms.xml"/> > <include-form name="ListJavaThread" location="component://webtools/widget/ServiceForms.xml"/> > </screenlet> > <screenlet title="${uiLabelMap.WebtoolsGeneralJavaThreads}"> > > |
Re: svn commit: r1362422 - in /ofbiz/trunk/framework: service/config/ service/dtd/ service/src/org/ofbiz/service/
|
I'm working on the Job Scheduler code right now. I fixed it already.
-Adrian On 8/4/2012 11:21 AM, Jacques Le Roux wrote: > Hi, > > While reviewing Eclipse found that JobPoller misses some Map and List > types arguments > Eclipse autocompletion suggests the following change: > > Index: framework/service/src/org/ofbiz/service/job/JobPoller.java > =================================================================== > --- framework/service/src/org/ofbiz/service/job/JobPoller.java > (revision 1369293) > +++ framework/service/src/org/ofbiz/service/job/JobPoller.java > (working copy) > @@ -83,7 +83,7 @@ > } > > public Map<String, Object> getPoolState() { > - Map poolState = new HashMap(); > + Map<String, Object> poolState = new HashMap<String, Object>(); > poolState.put("pollerName", this.name); > poolState.put("pollerThreadName", "OFBiz-JobPoller-" + > this.name); > poolState.put("invokerThreadNameFormat", "OFBiz-JobInvoker-" + > this.name + "-<SEQ>"); > @@ -95,12 +95,12 @@ > poolState.put("greatestNumberOfInvokerThreads", > this.executor.getLargestPoolSize()); > poolState.put("numberOfCompletedTasks", > this.executor.getCompletedTaskCount()); > BlockingQueue<Runnable> queue = this.executor.getQueue(); > - List taskList = new ArrayList(); > - Map taskInfo = null; > + List<Map> taskList = new ArrayList<Map>(); > + Map<String, Comparable> taskInfo = null; > for (Runnable task : queue) { > if (task instanceof JobInvoker) { > JobInvoker jobInvoker = (JobInvoker) task; > - taskInfo = new HashMap(); > + taskInfo = new HashMap<String, Comparable>(); > taskInfo.put("id", jobInvoker.getJobId()); > taskInfo.put("name", jobInvoker.getJobName()); > taskInfo.put("serviceName", jobInvoker.getServiceName()); > > Methods like executor.getCorePoolSize() return int/s and methods like > jobInvoker.getTime() return long/s. It should be ok, thanks to > autoboxing, and it compiles of course. > But I wonder why Jacopo did not do it, any reasons or just a miss? > > Jacques > > From: <[hidden email]> >> Author: jacopoc >> Date: Tue Jul 17 09:20:42 2012 >> New Revision: 1362422 >> >> URL: http://svn.apache.org/viewvc?rev=1362422&view=rev >> Log: >> Refactored the very old code in the org.ofbiz.service.job package to >> leverage the Executor framework of java.util.concurrent rather than >> relying on ad-hoc synchronization and thread management: the new code >> is simpler, smaller and it is now easier to review and change the >> execution policy; two attributes of service-config.xsd have been >> deprecated as they are not used in the new version ("jobs" and >> "wait-millis"). >> >> Added: >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThread.java >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThreadFactory.java >> Modified: >> ofbiz/trunk/framework/service/config/serviceengine.xml >> ofbiz/trunk/framework/service/dtd/service-config.xsd >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvoker.java >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobManager.java >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobPoller.java >> ofbiz/trunk/framework/webtools/webapp/webtools/WEB-INF/actions/service/Threads.groovy >> ofbiz/trunk/framework/webtools/widget/ServiceForms.xml >> ofbiz/trunk/framework/webtools/widget/ServiceScreens.xml >> >> Modified: ofbiz/trunk/framework/service/config/serviceengine.xml >> URL: >> http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/config/serviceengine.xml?rev=1362422&r1=1362421&r2=1362422&view=diff >> ============================================================================== >> >> --- ofbiz/trunk/framework/service/config/serviceengine.xml (original) >> +++ ofbiz/trunk/framework/service/config/serviceengine.xml Tue Jul 17 >> 09:20:42 2012 >> @@ -30,8 +30,6 @@ under the License. >> purge-job-days="4" >> failed-retry-min="3" >> ttl="18000000" >> - wait-millis="750" >> - jobs="10" >> min-threads="5" >> max-threads="15" >> poll-enabled="true" >> >> Modified: ofbiz/trunk/framework/service/dtd/service-config.xsd >> URL: >> http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/dtd/service-config.xsd?rev=1362422&r1=1362421&r2=1362422&view=diff >> ============================================================================== >> >> --- ofbiz/trunk/framework/service/dtd/service-config.xsd (original) >> +++ ofbiz/trunk/framework/service/dtd/service-config.xsd Tue Jul 17 >> 09:20:42 2012 >> @@ -68,8 +68,8 @@ under the License. >> <xs:attribute type="xs:nonNegativeInteger" >> name="purge-job-days" default="30"/> >> <xs:attribute type="xs:nonNegativeInteger" >> name="failed-retry-min" default="30"/> >> <xs:attribute type="xs:nonNegativeInteger" name="ttl" >> use="required"/> >> - <xs:attribute type="xs:nonNegativeInteger" >> name="wait-millis" use="required"/> >> - <xs:attribute type="xs:nonNegativeInteger" name="jobs" >> use="required"/> >> + <xs:attribute type="xs:nonNegativeInteger" >> name="wait-millis"/> <!-- deprecated --> >> + <xs:attribute type="xs:nonNegativeInteger" name="jobs"/> >> <!-- deprecated --> >> <xs:attribute type="xs:nonNegativeInteger" name="min-threads" >> use="required"/> >> <xs:attribute type="xs:nonNegativeInteger" name="max-threads" >> use="required"/> >> <xs:attribute name="poll-enabled" default="true"> >> >> Modified: >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvoker.java >> URL: >> http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvoker.java?rev=1362422&r1=1362421&r2=1362422&view=diff >> ============================================================================== >> >> --- >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvoker.java >> (original) >> +++ >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvoker.java >> Tue Jul 17 09:20:42 2012 >> @@ -20,12 +20,9 @@ package org.ofbiz.service.job; >> >> import java.util.Date; >> >> -import org.apache.commons.lang.math.NumberUtils; >> import org.ofbiz.base.util.Debug; >> -import org.ofbiz.base.util.UtilDateTime; >> import org.ofbiz.entity.transaction.GenericTransactionException; >> import org.ofbiz.entity.transaction.TransactionUtil; >> -import org.ofbiz.service.config.ServiceConfigUtil; >> >> /** >> * JobInvoker >> @@ -33,82 +30,20 @@ import org.ofbiz.service.config.ServiceC >> public class JobInvoker implements Runnable { >> >> public static final String module = JobInvoker.class.getName(); >> - public static final long THREAD_TTL = 18000000; >> - public static final int WAIT_TIME = 750; >> >> - private JobPoller jp = null; >> - private Thread thread = null; >> private Date created = null; >> - private String name = null; >> - private int count = 0; >> - private int wait = 0; >> - >> - private volatile boolean run = false; >> - private volatile Job currentJob = null; >> - private volatile int statusCode = 0; >> - private volatile long jobStart = 0; >> + private long jobStart; >> >> - public JobInvoker(JobPoller jp) { >> - this(jp, WAIT_TIME); >> - } >> + private Job currentJob = null; >> >> - public JobInvoker(JobPoller jp, int wait) { >> + public JobInvoker(Job job) { >> this.created = new Date(); >> - this.run = true; >> - this.count = 0; >> - this.jp = jp; >> - this.wait = wait; >> - >> - // service dispatcher delegator name (for thread name) >> - String delegatorName = >> jp.getManager().getDelegator().getDelegatorName(); >> - >> - // get a new thread >> - this.thread = new Thread(this); >> - this.name = delegatorName + "-invoker-" + >> this.thread.getName(); >> - >> - this.thread.setDaemon(false); >> - this.thread.setName(this.name); >> - >> - if (Debug.verboseOn()) Debug.logVerbose("JobInvoker: >> Starting Invoker Thread -- " + thread.getName(), module); >> - this.thread.start(); >> + this.currentJob = job; >> } >> >> protected JobInvoker() {} >> >> /** >> - * Tells the thread to stop after the next job. >> - */ >> - public void stop() { >> - run = false; >> - } >> - >> - /** >> - * Wakes up this thread. >> - */ >> - public void wakeUp() { >> - notifyAll(); >> - } >> - >> - /** >> - * Gets the number of times this thread was used. >> - * @return The number of times used. >> - */ >> - public int getUsage() { >> - return count; >> - } >> - >> - /** >> - * Gets the remaining time this thread has before it is killed >> - * @return Time in millis remaining >> - */ >> - public long getTimeRemaining() { >> - long now = UtilDateTime.nowTimestamp().getTime(); >> - long time = getTime(); >> - long ttl = getTTL(); >> - return (time + ttl) - now; >> - } >> - >> - /** >> * Gets the time when this thread was created. >> * @return Time in milliseconds when this was created. >> */ >> @@ -117,24 +52,8 @@ public class JobInvoker implements Runna >> } >> >> /** >> - * Gets the name of this JobInvoker. >> - * @return Name of the invoker. >> - */ >> - public String getName() { >> - return this.name; >> - } >> - >> - /** >> - * Gets the status code for this thread (0 = sleeping, 1 = >> running job) >> - * @return 0 for sleeping or 1 when running a job. >> - */ >> - public int getCurrentStatus() { >> - return this.statusCode; >> - } >> - >> - /** >> * Gets the total time the current job has been running or 0 when >> sleeping. >> - * @return Total time the curent job has been running. >> + * @return Total time the current job has been running. >> */ >> public long getCurrentRuntime() { >> if (this.jobStart > 0) { >> @@ -145,27 +64,15 @@ public class JobInvoker implements Runna >> } >> } >> >> - public Long getThreadId() { >> - if (this.thread != null) { >> - return this.thread.getId(); >> - } else { >> - return null; >> - } >> - } >> - >> /** >> * Get the current running job's ID. >> * @return String ID of the current running job. >> */ >> public String getJobId() { >> - if (this.statusCode == 1) { >> - if (this.currentJob != null) { >> - return this.currentJob.getJobId(); >> - } else { >> - return "WARNING: Invalid Job!"; >> - } >> + if (this.currentJob != null) { >> + return this.currentJob.getJobId(); >> } else { >> - return null; >> + return "WARNING: Invalid Job!"; >> } >> } >> >> @@ -174,14 +81,10 @@ public class JobInvoker implements Runna >> * @return String name of the current running job. >> */ >> public String getJobName() { >> - if (this.statusCode == 1) { >> - if (this.currentJob != null) { >> - return this.currentJob.getJobName(); >> - } else { >> - return "WARNING: Invalid Job!"; >> - } >> + if (this.currentJob != null) { >> + return this.currentJob.getJobName(); >> } else { >> - return null; >> + return "WARNING: Invalid Job!"; >> } >> } >> >> @@ -191,101 +94,45 @@ public class JobInvoker implements Runna >> */ >> public String getServiceName() { >> String serviceName = null; >> - if (this.statusCode == 1) { >> - if (this.currentJob != null) { >> - if (this.currentJob instanceof GenericServiceJob) { >> - GenericServiceJob gsj = (GenericServiceJob) >> this.currentJob; >> - try { >> - serviceName = gsj.getServiceName(); >> - } catch (InvalidJobException e) { >> - Debug.logError(e, module); >> - } >> + if (this.currentJob != null) { >> + if (this.currentJob instanceof GenericServiceJob) { >> + GenericServiceJob gsj = (GenericServiceJob) >> this.currentJob; >> + try { >> + serviceName = gsj.getServiceName(); >> + } catch (InvalidJobException e) { >> + Debug.logError(e, module); >> } >> } >> } >> return serviceName; >> } >> >> - /** >> - * Kill this invoker thread.s >> - */ >> - public void kill() { >> - this.stop(); >> - this.statusCode = -1; >> - this.thread.interrupt(); >> - this.thread = null; >> - } >> + public void run() { >> + // setup the current job settings >> + this.jobStart = System.currentTimeMillis(); >> >> - public synchronized void run() { >> - while (run) { >> - Job job = jp.next(); >> - >> - if (job == null) { >> - try { >> - java.lang.Thread.sleep(wait); >> - } catch (InterruptedException ie) { >> - Debug.logError(ie, "JobInvoker.run() : >> InterruptedException", module); >> - stop(); >> - } >> - } else { >> - Debug.logInfo("Invoker [" + thread.getName() + "] >> received job [" + job.getJobName() + "] from poller [" + >> jp.toString() + "]", module); >> - >> - // setup the current job settings >> - this.currentJob = job; >> - this.statusCode = 1; >> - this.jobStart = System.currentTimeMillis(); >> - >> - // execute the job >> - if (Debug.verboseOn()) Debug.logVerbose("Invoker: " >> + thread.getName() + " executing job -- " + job.getJobName(), module); >> - try { >> - job.exec(); >> - } catch (InvalidJobException e) { >> - Debug.logWarning(e.getMessage(), module); >> - } >> - if (Debug.verboseOn()) Debug.logVerbose("Invoker: " >> + thread.getName() + " finished executing job -- " + >> job.getJobName(), module); >> - >> - // clear the current job settings >> - this.currentJob = null; >> - this.statusCode = 0; >> - this.jobStart = 0; >> - >> - // sanity check; make sure we don't have any >> transactions in place >> - try { >> - // roll back current TX first >> - if (TransactionUtil.isTransactionInPlace()) { >> - Debug.logWarning("*** NOTICE: JobInvoker >> finished w/ a transaction in place! Rolling back.", module); >> - TransactionUtil.rollback(); >> - } >> - >> - // now resume/rollback any suspended txs >> - if (TransactionUtil.suspendedTransactionsHeld()) { >> - int suspended = >> TransactionUtil.cleanSuspendedTransactions(); >> - Debug.logWarning("Resumed/Rolled Back [" + >> suspended + "] transactions.", module); >> - } >> - } catch (GenericTransactionException e) { >> - Debug.logWarning(e, module); >> - } >> - >> - // increment the count >> - count++; >> - if (Debug.verboseOn()) Debug.logVerbose("Invoker: " >> + thread.getName() + " (" + count + ") total.", module); >> - } >> - long diff = (new Date().getTime() - this.getTime()); >> - >> - if (getTTL() > 0 && diff > getTTL()) >> - jp.removeThread(this); >> + // execute the job >> + try { >> + this.currentJob.exec(); >> + } catch (InvalidJobException e) { >> + Debug.logWarning(e.getMessage(), module); >> } >> - if (Debug.verboseOn()) Debug.logVerbose("Invoker: " + >> thread.getName() + " dead -- " + UtilDateTime.nowTimestamp(), module); >> - } >> - >> - private long getTTL() { >> - long ttl = THREAD_TTL; >> >> + // sanity check; make sure we don't have any transactions in >> place >> try { >> - ttl = >> NumberUtils.toLong(ServiceConfigUtil.getElementAttr("thread-pool", >> "ttl")); >> - } catch (NumberFormatException nfe) { >> - Debug.logError("Problems reading value from attribute >> [ttl] of element [thread-pool] in serviceengine.xml file [" + >> nfe.toString() + "]. Using default (" + THREAD_TTL + ").", module); >> + // roll back current TX first >> + if (TransactionUtil.isTransactionInPlace()) { >> + Debug.logWarning("*** NOTICE: JobInvoker finished w/ >> a transaction in place! Rolling back.", module); >> + TransactionUtil.rollback(); >> + } >> + >> + // now resume/rollback any suspended txs >> + if (TransactionUtil.suspendedTransactionsHeld()) { >> + int suspended = >> TransactionUtil.cleanSuspendedTransactions(); >> + Debug.logWarning("Resumed/Rolled Back [" + suspended >> + "] transactions.", module); >> + } >> + } catch (GenericTransactionException e) { >> + Debug.logWarning(e, module); >> } >> - return ttl; >> } >> } >> >> Added: >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThread.java >> URL: >> http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThread.java?rev=1362422&view=auto >> ============================================================================== >> >> --- >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThread.java >> (added) >> +++ >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThread.java >> Tue Jul 17 09:20:42 2012 >> @@ -0,0 +1,29 @@ >> +/******************************************************************************* >> >> + * 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.ofbiz.service.job; >> + >> +import java.util.concurrent.atomic.AtomicInteger; >> + >> +public class JobInvokerThread extends Thread { >> + private static final AtomicInteger created = new AtomicInteger(); >> + >> + public JobInvokerThread(Runnable runnable, String poolName) { >> + super(runnable, "OFBiz-JobInvoker-" + poolName + "-" + >> created.getAndIncrement()); >> + } >> +} >> >> Added: >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThreadFactory.java >> URL: >> http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThreadFactory.java?rev=1362422&view=auto >> ============================================================================== >> >> --- >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThreadFactory.java >> (added) >> +++ >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThreadFactory.java >> Tue Jul 17 09:20:42 2012 >> @@ -0,0 +1,33 @@ >> +/******************************************************************************* >> >> + * 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.ofbiz.service.job; >> + >> +import java.util.concurrent.ThreadFactory; >> + >> +public class JobInvokerThreadFactory implements ThreadFactory { >> + private final String poolName; >> + >> + public JobInvokerThreadFactory(String poolName) { >> + this.poolName = poolName; >> + } >> + >> + public Thread newThread(Runnable runnable) { >> + return new JobInvokerThread(runnable, poolName); >> + } >> +} >> >> Modified: >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobManager.java >> URL: >> http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobManager.java?rev=1362422&r1=1362421&r2=1362422&view=diff >> ============================================================================== >> >> --- >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobManager.java >> (original) >> +++ >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobManager.java >> Tue Jul 17 09:20:42 2012 >> @@ -423,18 +423,10 @@ public class JobManager { >> } >> >> /** >> - * Kill a JobInvoker Thread. >> - * @param threadName Name of the JobInvoker Thread to kill. >> - */ >> - public void killThread(String threadName) { >> - jp.killThread(threadName); >> - } >> - >> - /** >> * Get a List of each threads current state. >> * @return List containing a Map of each thread's state. >> */ >> - public List<Map<String, Object>> processList() { >> + public Map<String, Object> getPoolState() { >> return jp.getPoolState(); >> } >> >> >> Modified: >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobPoller.java >> URL: >> http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobPoller.java?rev=1362422&r1=1362421&r2=1362422&view=diff >> ============================================================================== >> >> --- >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobPoller.java (original) >> >> +++ >> ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobPoller.java Tue >> Jul 17 09:20:42 2012 >> @@ -18,15 +18,20 @@ >> *******************************************************************************/ >> >> package org.ofbiz.service.job; >> >> +import java.util.ArrayList; >> +import java.util.HashMap; >> import java.util.List; >> import java.util.Map; >> - >> -import javolution.util.FastList; >> -import javolution.util.FastMap; >> +import java.util.concurrent.BlockingQueue; >> +import java.util.concurrent.LinkedBlockingQueue; >> +import java.util.concurrent.ThreadPoolExecutor; >> +import java.util.concurrent.TimeUnit; >> >> import org.ofbiz.base.util.Debug; >> import org.ofbiz.service.config.ServiceConfigUtil; >> >> +import org.apache.commons.lang.math.NumberUtils; >> + >> /** >> * JobPoller - Polls for persisted jobs to run. >> */ >> @@ -36,27 +41,28 @@ public class JobPoller implements Runnab >> >> public static final int MIN_THREADS = 1; >> public static final int MAX_THREADS = 15; >> - public static final int MAX_JOBS = 3; >> public static final int POLL_WAIT = 20000; >> - //public static final long MAX_TTL = 18000000; >> + public static final long THREAD_TTL = 18000000; >> >> - protected Thread thread = null; >> - protected List<JobInvoker> pool = null; >> - protected List<Job> run = null; >> - protected JobManager jm = null; >> - >> - protected volatile boolean isRunning = false; >> + private Thread thread = null; >> + private JobManager jm = null; >> + private ThreadPoolExecutor executor = null; >> + private String name = null; >> >> /** >> * Creates a new JobScheduler >> * @param jm JobManager associated with this scheduler >> */ >> public JobPoller(JobManager jm, boolean enabled) { >> + this.name = (jm.getDelegator() != null? >> jm.getDelegator().getDelegatorName(): "NA"); >> this.jm = jm; >> - this.run = FastList.newInstance(); >> - >> - // create the thread pool >> - this.pool = createThreadPool(); >> + this.executor = new ThreadPoolExecutor(minThreads(), >> + maxThreads(), >> + getTTL(), >> + TimeUnit.MILLISECONDS, >> + new >> LinkedBlockingQueue<Runnable>(), >> + new >> JobInvokerThreadFactory(this.name), >> + new >> ThreadPoolExecutor.AbortPolicy()); >> >> if (enabled) { >> // re-load crashed jobs >> @@ -66,11 +72,10 @@ public class JobPoller implements Runnab >> if (pollEnabled()) { >> >> // create the poller thread >> - thread = new Thread(this, this.toString()); >> + thread = new Thread(this, "OFBiz-JobPoller-" + >> this.name); >> thread.setDaemon(false); >> >> // start the poller >> - this.isRunning = true; >> thread.start(); >> } >> } >> @@ -84,7 +89,7 @@ public class JobPoller implements Runnab >> java.lang.Thread.sleep(30000); >> } catch (InterruptedException e) { >> } >> - while (isRunning) { >> + while (!executor.isShutdown()) { >> try { >> // grab a list of jobs to run. >> List<Job> pollList = jm.poll(); >> @@ -106,140 +111,77 @@ public class JobPoller implements Runnab >> } >> >> /** >> - * Returns the JobManager >> + * Adds a job to the RUN queue >> */ >> - public JobManager getManager() { >> - return jm; >> + public void queueNow(Job job) { >> + this.executor.execute(new JobInvoker(job)); >> } >> >> /** >> * Stops the JobPoller >> */ >> - public void stop() { >> - isRunning = false; >> - destroyThreadPool(); >> - } >> - >> - public List<Map<String, Object>> getPoolState() { >> - List<Map<String, Object>> stateList = FastList.newInstance(); >> - for (JobInvoker invoker: this.pool) { >> - Map<String, Object> stateMap = FastMap.newInstance(); >> - stateMap.put("threadName", invoker.getName()); >> - stateMap.put("threadId", invoker.getThreadId()); >> - stateMap.put("jobName", invoker.getJobName()); >> - stateMap.put("serviceName", invoker.getServiceName()); >> - stateMap.put("usage", invoker.getUsage()); >> - stateMap.put("ttl", invoker.getTimeRemaining()); >> - stateMap.put("runTime", invoker.getCurrentRuntime()); >> - stateMap.put("status", invoker.getCurrentStatus()); >> - stateList.add(stateMap); >> - } >> - return stateList; >> - } >> - >> - /** >> - * Stops all threads in the threadPool and clears >> - * the pool as final step. >> - */ >> - private void destroyThreadPool() { >> - Debug.logInfo("Destroying thread pool...", module); >> - for (JobInvoker ji: pool) { >> - ji.stop(); >> - } >> - pool.clear(); >> - } >> - >> - public synchronized void killThread(String threadName) { >> - JobInvoker inv = findThread(threadName); >> - if (inv != null) { >> - inv.kill(); >> - this.pool.remove(inv); >> - } >> - } >> - >> - private JobInvoker findThread(String threadName) { >> - for (JobInvoker inv: pool) { >> - if (threadName.equals(inv.getName())) { >> - return inv; >> - } >> - } >> - return null; >> - } >> - >> - /** >> - * Returns the next job to run >> - */ >> - public Job next() { >> - if (run.size() > 0) { >> - // NOTE: this syncrhonized isn't really necessary as the >> only method that calls it is already synchronized (the JobInvoker.run >> method), so this is here as an added protection especially for the >> case where it might be used differently in the future >> - synchronized (run) { >> - // make sure the size is still greater than zero >> - if (run.size() > 0) { >> - return run.remove(0); >> + void stop() { >> + Debug.logInfo("Shutting down thread pool for " + this.name, >> module); >> + this.executor.shutdown(); >> + try { >> + // Wait 60 seconds for existing tasks to terminate >> + if (!this.executor.awaitTermination(60, >> TimeUnit.SECONDS)) { >> + // abrupt shutdown (cancel currently executing tasks) >> + Debug.logInfo("Attempting abrupt shut down of thread >> pool for " + this.name, module); >> + this.executor.shutdownNow(); >> + // Wait 60 seconds for tasks to respond to being >> cancelled >> + if (!this.executor.awaitTermination(60, >> TimeUnit.SECONDS)) { >> + Debug.logWarning("Unable to shutdown the thread >> pool for " + this.name, module); >> } >> } >> + } catch (InterruptedException ie) { >> + // re cancel if current thread was also interrupted >> + this.executor.shutdownNow(); >> + // preserve interrupt status >> + Thread.currentThread().interrupt(); >> } >> - return null; >> + Debug.logInfo("Shutdown completed of thread pool for " + >> this.name, module); >> } >> >> /** >> - * Adds a job to the RUN queue >> + * Returns the JobManager >> */ >> - public void queueNow(Job job) { >> - //Debug.logInfo("[" + Thread.currentThread().getId() + "] >> Begin queueNow; holds run lock? " + Thread.holdsLock(run), module); >> - >> - // NOTE DEJ20071201 MUST use a different object for the lock >> here because the "this" object is always held by the poller thread in >> the run method above (which sleeps and runs) >> - synchronized (run) { >> - run.add(job); >> - } >> - if (Debug.verboseOn()) Debug.logVerbose("New run queue size: >> " + run.size(), module); >> - if (run.size() > pool.size() && pool.size() < maxThreads()) { >> - synchronized (pool) { >> - if (run.size() > pool.size() && pool.size() < >> maxThreads()) { >> - int calcSize = (run.size() / jobsPerThread()) - >> (pool.size()); >> - int addSize = calcSize > maxThreads() ? >> maxThreads() : calcSize; >> - >> - for (int i = 0; i < addSize; i++) { >> - JobInvoker iv = new JobInvoker(this, >> invokerWaitTime()); >> - pool.add(iv); >> - } >> - } >> - } >> - } >> + public JobManager getManager() { >> + return jm; >> } >> >> - /** >> - * Removes a thread from the pool. >> - * @param invoker The invoker to remove. >> - */ >> - public void removeThread(JobInvoker invoker) { >> - if (pool != null) { >> - synchronized (pool) { >> - pool.remove(invoker); >> - invoker.stop(); >> - } >> - } >> - >> - if (pool != null && pool.size() < minThreads()) { >> - synchronized (pool) { >> - for (int i = 0; i < minThreads() - pool.size(); i++) { >> - JobInvoker iv = new JobInvoker(this, >> invokerWaitTime()); >> - pool.add(iv); >> - } >> + public Map<String, Object> getPoolState() { >> + Map poolState = new HashMap(); >> + poolState.put("pollerName", this.name); >> + poolState.put("pollerThreadName", "OFBiz-JobPoller-" + >> this.name); >> + poolState.put("invokerThreadNameFormat", "OFBiz-JobInvoker-" >> + this.name + "-<SEQ>"); >> + poolState.put("keepAliveTimeInSeconds", >> this.executor.getKeepAliveTime(TimeUnit.SECONDS)); >> + >> + poolState.put("numberOfCoreInvokerThreads", >> this.executor.getCorePoolSize()); >> + poolState.put("currentNumberOfInvokerThreads", >> this.executor.getPoolSize()); >> + poolState.put("numberOfActiveInvokerThreads", >> this.executor.getActiveCount()); >> + poolState.put("maxNumberOfInvokerThreads", >> this.executor.getMaximumPoolSize()); >> + poolState.put("greatestNumberOfInvokerThreads", >> this.executor.getLargestPoolSize()); >> + >> + poolState.put("numberOfCompletedTasks", >> this.executor.getCompletedTaskCount()); >> + >> + BlockingQueue<Runnable> queue = this.executor.getQueue(); >> + List taskList = new ArrayList(); >> + Map taskInfo = null; >> + for (Runnable task: queue) { >> + if (task instanceof JobInvoker) { >> + JobInvoker jobInvoker = (JobInvoker)task; >> + taskInfo = new HashMap(); >> + taskInfo.put("id", jobInvoker.getJobId()); >> + taskInfo.put("name", jobInvoker.getJobName()); >> + taskInfo.put("serviceName", >> jobInvoker.getServiceName()); >> + taskInfo.put("time", jobInvoker.getTime()); >> + taskInfo.put("runtime", >> jobInvoker.getCurrentRuntime()); >> + taskList.add(taskInfo); >> } >> } >> - } >> - >> - // Creates the invoker pool >> - private List<JobInvoker> createThreadPool() { >> - List<JobInvoker> threadPool = FastList.newInstance(); >> - >> - while (threadPool.size() < minThreads()) { >> - JobInvoker iv = new JobInvoker(this, invokerWaitTime()); >> - threadPool.add(iv); >> - } >> - >> - return threadPool; >> + poolState.put("taskList", taskList); >> + return poolState; >> } >> >> private int maxThreads() { >> @@ -264,37 +206,26 @@ public class JobPoller implements Runnab >> return min; >> } >> >> - private int jobsPerThread() { >> - int jobs = MAX_JOBS; >> - >> - try { >> - jobs = >> Integer.parseInt(ServiceConfigUtil.getElementAttr("thread-pool", >> "jobs")); >> - } catch (NumberFormatException nfe) { >> - Debug.logError("Problems reading values from >> serviceengine.xml file [" + nfe.toString() + "]. Using defaults.", >> module); >> - } >> - return jobs; >> - } >> - >> - private int invokerWaitTime() { >> - int wait = JobInvoker.WAIT_TIME; >> + private int pollWaitTime() { >> + int poll = POLL_WAIT; >> >> try { >> - wait = >> Integer.parseInt(ServiceConfigUtil.getElementAttr("thread-pool", >> "wait-millis")); >> + poll = >> Integer.parseInt(ServiceConfigUtil.getElementAttr("thread-pool", >> "poll-db-millis")); >> } catch (NumberFormatException nfe) { >> Debug.logError("Problems reading values from >> serviceengine.xml file [" + nfe.toString() + "]. Using defaults.", >> module); >> } >> - return wait; >> + return poll; >> } >> >> - private int pollWaitTime() { >> - int poll = POLL_WAIT; >> + private long getTTL() { >> + long ttl = THREAD_TTL; >> >> try { >> - poll = >> Integer.parseInt(ServiceConfigUtil.getElementAttr("thread-pool", >> "poll-db-millis")); >> + ttl = >> NumberUtils.toLong(ServiceConfigUtil.getElementAttr("thread-pool", >> "ttl")); >> } catch (NumberFormatException nfe) { >> - Debug.logError("Problems reading values from >> serviceengine.xml file [" + nfe.toString() + "]. Using defaults.", >> module); >> + Debug.logError("Problems reading value from attribute >> [ttl] of element [thread-pool] in serviceengine.xml file [" + >> nfe.toString() + "]. Using default (" + THREAD_TTL + ").", module); >> } >> - return poll; >> + return ttl; >> } >> >> private boolean pollEnabled() { >> >> Modified: >> ofbiz/trunk/framework/webtools/webapp/webtools/WEB-INF/actions/service/Threads.groovy >> URL: >> http://svn.apache.org/viewvc/ofbiz/trunk/framework/webtools/webapp/webtools/WEB-INF/actions/service/Threads.groovy?rev=1362422&r1=1362421&r2=1362422&view=diff >> ============================================================================== >> >> --- >> ofbiz/trunk/framework/webtools/webapp/webtools/WEB-INF/actions/service/Threads.groovy >> (original) >> +++ >> ofbiz/trunk/framework/webtools/webapp/webtools/WEB-INF/actions/service/Threads.groovy >> Tue Jul 17 09:20:42 2012 >> @@ -35,19 +35,9 @@ uiLabelMap = UtilProperties.getResourceB >> uiLabelMap.addBottomResourceBundle("CommonUiLabels"); >> >> threads = []; >> -jobs = dispatcher.getJobManager().processList(); >> -jobs.each { job -> >> - state = job.status; >> - switch (state) { >> - case 0 : status = uiLabelMap.WebtoolsStatusSleeping; break; >> - case 1 : status = uiLabelMap.WebtoolsStatusRunning; break; >> - case -1: status = uiLabelMap.WebtoolsStatusShuttingDown; break; >> - default: status = uiLabelMap.WebtoolsStatusInvalid; break; >> - } >> - job.status = status; >> - threads.add(job); >> -} >> -context.threads = threads; >> +poolState = dispatcher.getJobManager().getPoolState(); >> +context.poolState = poolState; >> +context.threads = poolState.taskList; >> >> // Some stuff for general threads on the server >> currentThread = Thread.currentThread(); >> >> Modified: ofbiz/trunk/framework/webtools/widget/ServiceForms.xml >> URL: >> http://svn.apache.org/viewvc/ofbiz/trunk/framework/webtools/widget/ServiceForms.xml?rev=1362422&r1=1362421&r2=1362422&view=diff >> ============================================================================== >> >> --- ofbiz/trunk/framework/webtools/widget/ServiceForms.xml (original) >> +++ ofbiz/trunk/framework/webtools/widget/ServiceForms.xml Tue Jul 17 >> 09:20:42 2012 >> @@ -110,14 +110,24 @@ under the License. >> <field name="key"><display/></field> >> <field name="value"><display/></field> >> </form> >> + <form name="PoolState" type="single" default-map-name="poolState"> >> + <field name="pollerName"><display/></field> >> + <field name="pollerThreadName"><display/></field> >> + <field name="invokerThreadNameFormat"><display/></field> >> + <field name="keepAliveTimeInSeconds"><display/></field> >> + <field name="numberOfCoreInvokerThreads"><display/></field> >> + <field name="currentNumberOfInvokerThreads"><display/></field> >> + <field name="numberOfActiveInvokerThreads"><display/></field> >> + <field name="maxNumberOfInvokerThreads"><display/></field> >> + <field name="greatestNumberOfInvokerThreads"><display/></field> >> + <field name="numberOfCompletedTasks"><display/></field> >> + </form> >> <form name="ListJavaThread" type="list" list-name="threads" >> paginate-target="threadList" separate-columns="true" >> odd-row-style="alternate-row" >> default-table-style="basic-table hover-bar"> >> - <field name="threadId" >> title="${uiLabelMap.WebtoolsThread}"><display >> description="${threadId} ${threadName}"/></field> >> - <field name="status" >> title="${uiLabelMap.CommonStatus}"><display/></field> >> - <field name="jobName" >> title="${uiLabelMap.WebtoolsJob}"><display >> default-value="${uiLabelMap.CommonNone}"/></field> >> + <field name="id" >> title="${uiLabelMap.WebtoolsThread}"><display >> description="${threadId} ${threadName}"/></field> >> + <field name="name" >> title="${uiLabelMap.WebtoolsJob}"><display >> default-value="${uiLabelMap.CommonNone}"/></field> >> <field name="serviceName" >> title="${uiLabelMap.WebtoolsService}"><display >> default-value="${uiLabelMap.CommonNone}"/></field> >> - <field name="usage" >> title="${uiLabelMap.WebtoolsUsage}"><display/></field> >> - <field name="ttl" title="${uiLabelMap.WebtoolsTTL} >> (ms)"><display/></field> >> + <field name="time"><display/></field> >> <field name="runTime" title="${uiLabelMap.CommonTime} >> (ms)"><display/></field> >> </form> >> <form name="ListServices" type="list" list-name="services" >> paginate-target="ServiceLog" separate-columns="true" >> >> Modified: ofbiz/trunk/framework/webtools/widget/ServiceScreens.xml >> URL: >> http://svn.apache.org/viewvc/ofbiz/trunk/framework/webtools/widget/ServiceScreens.xml?rev=1362422&r1=1362421&r2=1362422&view=diff >> ============================================================================== >> >> --- ofbiz/trunk/framework/webtools/widget/ServiceScreens.xml (original) >> +++ ofbiz/trunk/framework/webtools/widget/ServiceScreens.xml Tue Jul >> 17 09:20:42 2012 >> @@ -103,6 +103,7 @@ under the License. >> <decorator-screen name="CommonServiceDecorator" >> location="${parameters.mainDecoratorLocation}"> >> <decorator-section name="body"> >> <screenlet >> title="${uiLabelMap.WebtoolsServiceEngineThreads}"> >> + <include-form name="PoolState" >> location="component://webtools/widget/ServiceForms.xml"/> >> <include-form name="ListJavaThread" >> location="component://webtools/widget/ServiceForms.xml"/> >> </screenlet> >> <screenlet >> title="${uiLabelMap.WebtoolsGeneralJavaThreads}"> >> >> |
Re: svn commit: r1362422 - in /ofbiz/trunk/framework: service/config/ service/dtd/ service/src/org/ofbiz/service/
|
|
In reply to this post by Jacques Le Roux
There is a problem with this change - the queue size has no limit. So,
the new code could cause an out-of-memory condition on a busy server. -Adrian On 7/17/2012 10:20 AM, [hidden email] wrote: > Author: jacopoc > Date: Tue Jul 17 09:20:42 2012 > New Revision: 1362422 > > URL: http://svn.apache.org/viewvc?rev=1362422&view=rev > Log: > Refactored the very old code in the org.ofbiz.service.job package to leverage the Executor framework of java.util.concurrent rather than relying on ad-hoc synchronization and thread management: the new code is simpler, smaller and it is now easier to review and change the execution policy; two attributes of service-config.xsd have been deprecated as they are not used in the new version ("jobs" and "wait-millis"). > > Added: > ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThread.java > ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThreadFactory.java > Modified: > ofbiz/trunk/framework/service/config/serviceengine.xml > ofbiz/trunk/framework/service/dtd/service-config.xsd > ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvoker.java > ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobManager.java > ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobPoller.java > ofbiz/trunk/framework/webtools/webapp/webtools/WEB-INF/actions/service/Threads.groovy > ofbiz/trunk/framework/webtools/widget/ServiceForms.xml > ofbiz/trunk/framework/webtools/widget/ServiceScreens.xml > > Modified: ofbiz/trunk/framework/service/config/serviceengine.xml > URL: http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/config/serviceengine.xml?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/service/config/serviceengine.xml (original) > +++ ofbiz/trunk/framework/service/config/serviceengine.xml Tue Jul 17 09:20:42 2012 > @@ -30,8 +30,6 @@ under the License. > purge-job-days="4" > failed-retry-min="3" > ttl="18000000" > - wait-millis="750" > - jobs="10" > min-threads="5" > max-threads="15" > poll-enabled="true" > > Modified: ofbiz/trunk/framework/service/dtd/service-config.xsd > URL: http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/dtd/service-config.xsd?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/service/dtd/service-config.xsd (original) > +++ ofbiz/trunk/framework/service/dtd/service-config.xsd Tue Jul 17 09:20:42 2012 > @@ -68,8 +68,8 @@ under the License. > <xs:attribute type="xs:nonNegativeInteger" name="purge-job-days" default="30"/> > <xs:attribute type="xs:nonNegativeInteger" name="failed-retry-min" default="30"/> > <xs:attribute type="xs:nonNegativeInteger" name="ttl" use="required"/> > - <xs:attribute type="xs:nonNegativeInteger" name="wait-millis" use="required"/> > - <xs:attribute type="xs:nonNegativeInteger" name="jobs" use="required"/> > + <xs:attribute type="xs:nonNegativeInteger" name="wait-millis"/> <!-- deprecated --> > + <xs:attribute type="xs:nonNegativeInteger" name="jobs"/> <!-- deprecated --> > <xs:attribute type="xs:nonNegativeInteger" name="min-threads" use="required"/> > <xs:attribute type="xs:nonNegativeInteger" name="max-threads" use="required"/> > <xs:attribute name="poll-enabled" default="true"> > > Modified: ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvoker.java > URL: http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvoker.java?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvoker.java (original) > +++ ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvoker.java Tue Jul 17 09:20:42 2012 > @@ -20,12 +20,9 @@ package org.ofbiz.service.job; > > import java.util.Date; > > -import org.apache.commons.lang.math.NumberUtils; > import org.ofbiz.base.util.Debug; > -import org.ofbiz.base.util.UtilDateTime; > import org.ofbiz.entity.transaction.GenericTransactionException; > import org.ofbiz.entity.transaction.TransactionUtil; > -import org.ofbiz.service.config.ServiceConfigUtil; > > /** > * JobInvoker > @@ -33,82 +30,20 @@ import org.ofbiz.service.config.ServiceC > public class JobInvoker implements Runnable { > > public static final String module = JobInvoker.class.getName(); > - public static final long THREAD_TTL = 18000000; > - public static final int WAIT_TIME = 750; > > - private JobPoller jp = null; > - private Thread thread = null; > private Date created = null; > - private String name = null; > - private int count = 0; > - private int wait = 0; > - > - private volatile boolean run = false; > - private volatile Job currentJob = null; > - private volatile int statusCode = 0; > - private volatile long jobStart = 0; > + private long jobStart; > > - public JobInvoker(JobPoller jp) { > - this(jp, WAIT_TIME); > - } > + private Job currentJob = null; > > - public JobInvoker(JobPoller jp, int wait) { > + public JobInvoker(Job job) { > this.created = new Date(); > - this.run = true; > - this.count = 0; > - this.jp = jp; > - this.wait = wait; > - > - // service dispatcher delegator name (for thread name) > - String delegatorName = jp.getManager().getDelegator().getDelegatorName(); > - > - // get a new thread > - this.thread = new Thread(this); > - this.name = delegatorName + "-invoker-" + this.thread.getName(); > - > - this.thread.setDaemon(false); > - this.thread.setName(this.name); > - > - if (Debug.verboseOn()) Debug.logVerbose("JobInvoker: Starting Invoker Thread -- " + thread.getName(), module); > - this.thread.start(); > + this.currentJob = job; > } > > protected JobInvoker() {} > > /** > - * Tells the thread to stop after the next job. > - */ > - public void stop() { > - run = false; > - } > - > - /** > - * Wakes up this thread. > - */ > - public void wakeUp() { > - notifyAll(); > - } > - > - /** > - * Gets the number of times this thread was used. > - * @return The number of times used. > - */ > - public int getUsage() { > - return count; > - } > - > - /** > - * Gets the remaining time this thread has before it is killed > - * @return Time in millis remaining > - */ > - public long getTimeRemaining() { > - long now = UtilDateTime.nowTimestamp().getTime(); > - long time = getTime(); > - long ttl = getTTL(); > - return (time + ttl) - now; > - } > - > - /** > * Gets the time when this thread was created. > * @return Time in milliseconds when this was created. > */ > @@ -117,24 +52,8 @@ public class JobInvoker implements Runna > } > > /** > - * Gets the name of this JobInvoker. > - * @return Name of the invoker. > - */ > - public String getName() { > - return this.name; > - } > - > - /** > - * Gets the status code for this thread (0 = sleeping, 1 = running job) > - * @return 0 for sleeping or 1 when running a job. > - */ > - public int getCurrentStatus() { > - return this.statusCode; > - } > - > - /** > * Gets the total time the current job has been running or 0 when sleeping. > - * @return Total time the curent job has been running. > + * @return Total time the current job has been running. > */ > public long getCurrentRuntime() { > if (this.jobStart > 0) { > @@ -145,27 +64,15 @@ public class JobInvoker implements Runna > } > } > > - public Long getThreadId() { > - if (this.thread != null) { > - return this.thread.getId(); > - } else { > - return null; > - } > - } > - > /** > * Get the current running job's ID. > * @return String ID of the current running job. > */ > public String getJobId() { > - if (this.statusCode == 1) { > - if (this.currentJob != null) { > - return this.currentJob.getJobId(); > - } else { > - return "WARNING: Invalid Job!"; > - } > + if (this.currentJob != null) { > + return this.currentJob.getJobId(); > } else { > - return null; > + return "WARNING: Invalid Job!"; > } > } > > @@ -174,14 +81,10 @@ public class JobInvoker implements Runna > * @return String name of the current running job. > */ > public String getJobName() { > - if (this.statusCode == 1) { > - if (this.currentJob != null) { > - return this.currentJob.getJobName(); > - } else { > - return "WARNING: Invalid Job!"; > - } > + if (this.currentJob != null) { > + return this.currentJob.getJobName(); > } else { > - return null; > + return "WARNING: Invalid Job!"; > } > } > > @@ -191,101 +94,45 @@ public class JobInvoker implements Runna > */ > public String getServiceName() { > String serviceName = null; > - if (this.statusCode == 1) { > - if (this.currentJob != null) { > - if (this.currentJob instanceof GenericServiceJob) { > - GenericServiceJob gsj = (GenericServiceJob) this.currentJob; > - try { > - serviceName = gsj.getServiceName(); > - } catch (InvalidJobException e) { > - Debug.logError(e, module); > - } > + if (this.currentJob != null) { > + if (this.currentJob instanceof GenericServiceJob) { > + GenericServiceJob gsj = (GenericServiceJob) this.currentJob; > + try { > + serviceName = gsj.getServiceName(); > + } catch (InvalidJobException e) { > + Debug.logError(e, module); > } > } > } > return serviceName; > } > > - /** > - * Kill this invoker thread.s > - */ > - public void kill() { > - this.stop(); > - this.statusCode = -1; > - this.thread.interrupt(); > - this.thread = null; > - } > + public void run() { > + // setup the current job settings > + this.jobStart = System.currentTimeMillis(); > > - public synchronized void run() { > - while (run) { > - Job job = jp.next(); > - > - if (job == null) { > - try { > - java.lang.Thread.sleep(wait); > - } catch (InterruptedException ie) { > - Debug.logError(ie, "JobInvoker.run() : InterruptedException", module); > - stop(); > - } > - } else { > - Debug.logInfo("Invoker [" + thread.getName() + "] received job [" + job.getJobName() + "] from poller [" + jp.toString() + "]", module); > - > - // setup the current job settings > - this.currentJob = job; > - this.statusCode = 1; > - this.jobStart = System.currentTimeMillis(); > - > - // execute the job > - if (Debug.verboseOn()) Debug.logVerbose("Invoker: " + thread.getName() + " executing job -- " + job.getJobName(), module); > - try { > - job.exec(); > - } catch (InvalidJobException e) { > - Debug.logWarning(e.getMessage(), module); > - } > - if (Debug.verboseOn()) Debug.logVerbose("Invoker: " + thread.getName() + " finished executing job -- " + job.getJobName(), module); > - > - // clear the current job settings > - this.currentJob = null; > - this.statusCode = 0; > - this.jobStart = 0; > - > - // sanity check; make sure we don't have any transactions in place > - try { > - // roll back current TX first > - if (TransactionUtil.isTransactionInPlace()) { > - Debug.logWarning("*** NOTICE: JobInvoker finished w/ a transaction in place! Rolling back.", module); > - TransactionUtil.rollback(); > - } > - > - // now resume/rollback any suspended txs > - if (TransactionUtil.suspendedTransactionsHeld()) { > - int suspended = TransactionUtil.cleanSuspendedTransactions(); > - Debug.logWarning("Resumed/Rolled Back [" + suspended + "] transactions.", module); > - } > - } catch (GenericTransactionException e) { > - Debug.logWarning(e, module); > - } > - > - // increment the count > - count++; > - if (Debug.verboseOn()) Debug.logVerbose("Invoker: " + thread.getName() + " (" + count + ") total.", module); > - } > - long diff = (new Date().getTime() - this.getTime()); > - > - if (getTTL() > 0 && diff > getTTL()) > - jp.removeThread(this); > + // execute the job > + try { > + this.currentJob.exec(); > + } catch (InvalidJobException e) { > + Debug.logWarning(e.getMessage(), module); > } > - if (Debug.verboseOn()) Debug.logVerbose("Invoker: " + thread.getName() + " dead -- " + UtilDateTime.nowTimestamp(), module); > - } > - > - private long getTTL() { > - long ttl = THREAD_TTL; > > + // sanity check; make sure we don't have any transactions in place > try { > - ttl = NumberUtils.toLong(ServiceConfigUtil.getElementAttr("thread-pool", "ttl")); > - } catch (NumberFormatException nfe) { > - Debug.logError("Problems reading value from attribute [ttl] of element [thread-pool] in serviceengine.xml file [" + nfe.toString() + "]. Using default (" + THREAD_TTL + ").", module); > + // roll back current TX first > + if (TransactionUtil.isTransactionInPlace()) { > + Debug.logWarning("*** NOTICE: JobInvoker finished w/ a transaction in place! Rolling back.", module); > + TransactionUtil.rollback(); > + } > + > + // now resume/rollback any suspended txs > + if (TransactionUtil.suspendedTransactionsHeld()) { > + int suspended = TransactionUtil.cleanSuspendedTransactions(); > + Debug.logWarning("Resumed/Rolled Back [" + suspended + "] transactions.", module); > + } > + } catch (GenericTransactionException e) { > + Debug.logWarning(e, module); > } > - return ttl; > } > } > > Added: ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThread.java > URL: http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThread.java?rev=1362422&view=auto > ============================================================================== > --- ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThread.java (added) > +++ ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThread.java Tue Jul 17 09:20:42 2012 > @@ -0,0 +1,29 @@ > +/******************************************************************************* > + * 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.ofbiz.service.job; > + > +import java.util.concurrent.atomic.AtomicInteger; > + > +public class JobInvokerThread extends Thread { > + private static final AtomicInteger created = new AtomicInteger(); > + > + public JobInvokerThread(Runnable runnable, String poolName) { > + super(runnable, "OFBiz-JobInvoker-" + poolName + "-" + created.getAndIncrement()); > + } > +} > > Added: ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThreadFactory.java > URL: http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThreadFactory.java?rev=1362422&view=auto > ============================================================================== > --- ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThreadFactory.java (added) > +++ ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobInvokerThreadFactory.java Tue Jul 17 09:20:42 2012 > @@ -0,0 +1,33 @@ > +/******************************************************************************* > + * 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.ofbiz.service.job; > + > +import java.util.concurrent.ThreadFactory; > + > +public class JobInvokerThreadFactory implements ThreadFactory { > + private final String poolName; > + > + public JobInvokerThreadFactory(String poolName) { > + this.poolName = poolName; > + } > + > + public Thread newThread(Runnable runnable) { > + return new JobInvokerThread(runnable, poolName); > + } > +} > > Modified: ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobManager.java > URL: http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobManager.java?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobManager.java (original) > +++ ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobManager.java Tue Jul 17 09:20:42 2012 > @@ -423,18 +423,10 @@ public class JobManager { > } > > /** > - * Kill a JobInvoker Thread. > - * @param threadName Name of the JobInvoker Thread to kill. > - */ > - public void killThread(String threadName) { > - jp.killThread(threadName); > - } > - > - /** > * Get a List of each threads current state. > * @return List containing a Map of each thread's state. > */ > - public List<Map<String, Object>> processList() { > + public Map<String, Object> getPoolState() { > return jp.getPoolState(); > } > > > Modified: ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobPoller.java > URL: http://svn.apache.org/viewvc/ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobPoller.java?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobPoller.java (original) > +++ ofbiz/trunk/framework/service/src/org/ofbiz/service/job/JobPoller.java Tue Jul 17 09:20:42 2012 > @@ -18,15 +18,20 @@ > *******************************************************************************/ > package org.ofbiz.service.job; > > +import java.util.ArrayList; > +import java.util.HashMap; > import java.util.List; > import java.util.Map; > - > -import javolution.util.FastList; > -import javolution.util.FastMap; > +import java.util.concurrent.BlockingQueue; > +import java.util.concurrent.LinkedBlockingQueue; > +import java.util.concurrent.ThreadPoolExecutor; > +import java.util.concurrent.TimeUnit; > > import org.ofbiz.base.util.Debug; > import org.ofbiz.service.config.ServiceConfigUtil; > > +import org.apache.commons.lang.math.NumberUtils; > + > /** > * JobPoller - Polls for persisted jobs to run. > */ > @@ -36,27 +41,28 @@ public class JobPoller implements Runnab > > public static final int MIN_THREADS = 1; > public static final int MAX_THREADS = 15; > - public static final int MAX_JOBS = 3; > public static final int POLL_WAIT = 20000; > - //public static final long MAX_TTL = 18000000; > + public static final long THREAD_TTL = 18000000; > > - protected Thread thread = null; > - protected List<JobInvoker> pool = null; > - protected List<Job> run = null; > - protected JobManager jm = null; > - > - protected volatile boolean isRunning = false; > + private Thread thread = null; > + private JobManager jm = null; > + private ThreadPoolExecutor executor = null; > + private String name = null; > > /** > * Creates a new JobScheduler > * @param jm JobManager associated with this scheduler > */ > public JobPoller(JobManager jm, boolean enabled) { > + this.name = (jm.getDelegator() != null? jm.getDelegator().getDelegatorName(): "NA"); > this.jm = jm; > - this.run = FastList.newInstance(); > - > - // create the thread pool > - this.pool = createThreadPool(); > + this.executor = new ThreadPoolExecutor(minThreads(), > + maxThreads(), > + getTTL(), > + TimeUnit.MILLISECONDS, > + new LinkedBlockingQueue<Runnable>(), > + new JobInvokerThreadFactory(this.name), > + new ThreadPoolExecutor.AbortPolicy()); > > if (enabled) { > // re-load crashed jobs > @@ -66,11 +72,10 @@ public class JobPoller implements Runnab > if (pollEnabled()) { > > // create the poller thread > - thread = new Thread(this, this.toString()); > + thread = new Thread(this, "OFBiz-JobPoller-" + this.name); > thread.setDaemon(false); > > // start the poller > - this.isRunning = true; > thread.start(); > } > } > @@ -84,7 +89,7 @@ public class JobPoller implements Runnab > java.lang.Thread.sleep(30000); > } catch (InterruptedException e) { > } > - while (isRunning) { > + while (!executor.isShutdown()) { > try { > // grab a list of jobs to run. > List<Job> pollList = jm.poll(); > @@ -106,140 +111,77 @@ public class JobPoller implements Runnab > } > > /** > - * Returns the JobManager > + * Adds a job to the RUN queue > */ > - public JobManager getManager() { > - return jm; > + public void queueNow(Job job) { > + this.executor.execute(new JobInvoker(job)); > } > > /** > * Stops the JobPoller > */ > - public void stop() { > - isRunning = false; > - destroyThreadPool(); > - } > - > - public List<Map<String, Object>> getPoolState() { > - List<Map<String, Object>> stateList = FastList.newInstance(); > - for (JobInvoker invoker: this.pool) { > - Map<String, Object> stateMap = FastMap.newInstance(); > - stateMap.put("threadName", invoker.getName()); > - stateMap.put("threadId", invoker.getThreadId()); > - stateMap.put("jobName", invoker.getJobName()); > - stateMap.put("serviceName", invoker.getServiceName()); > - stateMap.put("usage", invoker.getUsage()); > - stateMap.put("ttl", invoker.getTimeRemaining()); > - stateMap.put("runTime", invoker.getCurrentRuntime()); > - stateMap.put("status", invoker.getCurrentStatus()); > - stateList.add(stateMap); > - } > - return stateList; > - } > - > - /** > - * Stops all threads in the threadPool and clears > - * the pool as final step. > - */ > - private void destroyThreadPool() { > - Debug.logInfo("Destroying thread pool...", module); > - for (JobInvoker ji: pool) { > - ji.stop(); > - } > - pool.clear(); > - } > - > - public synchronized void killThread(String threadName) { > - JobInvoker inv = findThread(threadName); > - if (inv != null) { > - inv.kill(); > - this.pool.remove(inv); > - } > - } > - > - private JobInvoker findThread(String threadName) { > - for (JobInvoker inv: pool) { > - if (threadName.equals(inv.getName())) { > - return inv; > - } > - } > - return null; > - } > - > - /** > - * Returns the next job to run > - */ > - public Job next() { > - if (run.size() > 0) { > - // NOTE: this syncrhonized isn't really necessary as the only method that calls it is already synchronized (the JobInvoker.run method), so this is here as an added protection especially for the case where it might be used differently in the future > - synchronized (run) { > - // make sure the size is still greater than zero > - if (run.size() > 0) { > - return run.remove(0); > + void stop() { > + Debug.logInfo("Shutting down thread pool for " + this.name, module); > + this.executor.shutdown(); > + try { > + // Wait 60 seconds for existing tasks to terminate > + if (!this.executor.awaitTermination(60, TimeUnit.SECONDS)) { > + // abrupt shutdown (cancel currently executing tasks) > + Debug.logInfo("Attempting abrupt shut down of thread pool for " + this.name, module); > + this.executor.shutdownNow(); > + // Wait 60 seconds for tasks to respond to being cancelled > + if (!this.executor.awaitTermination(60, TimeUnit.SECONDS)) { > + Debug.logWarning("Unable to shutdown the thread pool for " + this.name, module); > } > } > + } catch (InterruptedException ie) { > + // re cancel if current thread was also interrupted > + this.executor.shutdownNow(); > + // preserve interrupt status > + Thread.currentThread().interrupt(); > } > - return null; > + Debug.logInfo("Shutdown completed of thread pool for " + this.name, module); > } > > /** > - * Adds a job to the RUN queue > + * Returns the JobManager > */ > - public void queueNow(Job job) { > - //Debug.logInfo("[" + Thread.currentThread().getId() + "] Begin queueNow; holds run lock? " + Thread.holdsLock(run), module); > - > - // NOTE DEJ20071201 MUST use a different object for the lock here because the "this" object is always held by the poller thread in the run method above (which sleeps and runs) > - synchronized (run) { > - run.add(job); > - } > - if (Debug.verboseOn()) Debug.logVerbose("New run queue size: " + run.size(), module); > - if (run.size() > pool.size() && pool.size() < maxThreads()) { > - synchronized (pool) { > - if (run.size() > pool.size() && pool.size() < maxThreads()) { > - int calcSize = (run.size() / jobsPerThread()) - (pool.size()); > - int addSize = calcSize > maxThreads() ? maxThreads() : calcSize; > - > - for (int i = 0; i < addSize; i++) { > - JobInvoker iv = new JobInvoker(this, invokerWaitTime()); > - pool.add(iv); > - } > - } > - } > - } > + public JobManager getManager() { > + return jm; > } > > - /** > - * Removes a thread from the pool. > - * @param invoker The invoker to remove. > - */ > - public void removeThread(JobInvoker invoker) { > - if (pool != null) { > - synchronized (pool) { > - pool.remove(invoker); > - invoker.stop(); > - } > - } > - > - if (pool != null && pool.size() < minThreads()) { > - synchronized (pool) { > - for (int i = 0; i < minThreads() - pool.size(); i++) { > - JobInvoker iv = new JobInvoker(this, invokerWaitTime()); > - pool.add(iv); > - } > + public Map<String, Object> getPoolState() { > + Map poolState = new HashMap(); > + poolState.put("pollerName", this.name); > + poolState.put("pollerThreadName", "OFBiz-JobPoller-" + this.name); > + poolState.put("invokerThreadNameFormat", "OFBiz-JobInvoker-" + this.name + "-<SEQ>"); > + poolState.put("keepAliveTimeInSeconds", this.executor.getKeepAliveTime(TimeUnit.SECONDS)); > + > + poolState.put("numberOfCoreInvokerThreads", this.executor.getCorePoolSize()); > + poolState.put("currentNumberOfInvokerThreads", this.executor.getPoolSize()); > + poolState.put("numberOfActiveInvokerThreads", this.executor.getActiveCount()); > + poolState.put("maxNumberOfInvokerThreads", this.executor.getMaximumPoolSize()); > + poolState.put("greatestNumberOfInvokerThreads", this.executor.getLargestPoolSize()); > + > + poolState.put("numberOfCompletedTasks", this.executor.getCompletedTaskCount()); > + > + BlockingQueue<Runnable> queue = this.executor.getQueue(); > + List taskList = new ArrayList(); > + Map taskInfo = null; > + for (Runnable task: queue) { > + if (task instanceof JobInvoker) { > + JobInvoker jobInvoker = (JobInvoker)task; > + taskInfo = new HashMap(); > + taskInfo.put("id", jobInvoker.getJobId()); > + taskInfo.put("name", jobInvoker.getJobName()); > + taskInfo.put("serviceName", jobInvoker.getServiceName()); > + taskInfo.put("time", jobInvoker.getTime()); > + taskInfo.put("runtime", jobInvoker.getCurrentRuntime()); > + taskList.add(taskInfo); > } > } > - } > - > - // Creates the invoker pool > - private List<JobInvoker> createThreadPool() { > - List<JobInvoker> threadPool = FastList.newInstance(); > - > - while (threadPool.size() < minThreads()) { > - JobInvoker iv = new JobInvoker(this, invokerWaitTime()); > - threadPool.add(iv); > - } > - > - return threadPool; > + poolState.put("taskList", taskList); > + return poolState; > } > > private int maxThreads() { > @@ -264,37 +206,26 @@ public class JobPoller implements Runnab > return min; > } > > - private int jobsPerThread() { > - int jobs = MAX_JOBS; > - > - try { > - jobs = Integer.parseInt(ServiceConfigUtil.getElementAttr("thread-pool", "jobs")); > - } catch (NumberFormatException nfe) { > - Debug.logError("Problems reading values from serviceengine.xml file [" + nfe.toString() + "]. Using defaults.", module); > - } > - return jobs; > - } > - > - private int invokerWaitTime() { > - int wait = JobInvoker.WAIT_TIME; > + private int pollWaitTime() { > + int poll = POLL_WAIT; > > try { > - wait = Integer.parseInt(ServiceConfigUtil.getElementAttr("thread-pool", "wait-millis")); > + poll = Integer.parseInt(ServiceConfigUtil.getElementAttr("thread-pool", "poll-db-millis")); > } catch (NumberFormatException nfe) { > Debug.logError("Problems reading values from serviceengine.xml file [" + nfe.toString() + "]. Using defaults.", module); > } > - return wait; > + return poll; > } > > - private int pollWaitTime() { > - int poll = POLL_WAIT; > + private long getTTL() { > + long ttl = THREAD_TTL; > > try { > - poll = Integer.parseInt(ServiceConfigUtil.getElementAttr("thread-pool", "poll-db-millis")); > + ttl = NumberUtils.toLong(ServiceConfigUtil.getElementAttr("thread-pool", "ttl")); > } catch (NumberFormatException nfe) { > - Debug.logError("Problems reading values from serviceengine.xml file [" + nfe.toString() + "]. Using defaults.", module); > + Debug.logError("Problems reading value from attribute [ttl] of element [thread-pool] in serviceengine.xml file [" + nfe.toString() + "]. Using default (" + THREAD_TTL + ").", module); > } > - return poll; > + return ttl; > } > > private boolean pollEnabled() { > > Modified: ofbiz/trunk/framework/webtools/webapp/webtools/WEB-INF/actions/service/Threads.groovy > URL: http://svn.apache.org/viewvc/ofbiz/trunk/framework/webtools/webapp/webtools/WEB-INF/actions/service/Threads.groovy?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/webtools/webapp/webtools/WEB-INF/actions/service/Threads.groovy (original) > +++ ofbiz/trunk/framework/webtools/webapp/webtools/WEB-INF/actions/service/Threads.groovy Tue Jul 17 09:20:42 2012 > @@ -35,19 +35,9 @@ uiLabelMap = UtilProperties.getResourceB > uiLabelMap.addBottomResourceBundle("CommonUiLabels"); > > threads = []; > -jobs = dispatcher.getJobManager().processList(); > -jobs.each { job -> > - state = job.status; > - switch (state) { > - case 0 : status = uiLabelMap.WebtoolsStatusSleeping; break; > - case 1 : status = uiLabelMap.WebtoolsStatusRunning; break; > - case -1: status = uiLabelMap.WebtoolsStatusShuttingDown; break; > - default: status = uiLabelMap.WebtoolsStatusInvalid; break; > - } > - job.status = status; > - threads.add(job); > -} > -context.threads = threads; > +poolState = dispatcher.getJobManager().getPoolState(); > +context.poolState = poolState; > +context.threads = poolState.taskList; > > // Some stuff for general threads on the server > currentThread = Thread.currentThread(); > > Modified: ofbiz/trunk/framework/webtools/widget/ServiceForms.xml > URL: http://svn.apache.org/viewvc/ofbiz/trunk/framework/webtools/widget/ServiceForms.xml?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/webtools/widget/ServiceForms.xml (original) > +++ ofbiz/trunk/framework/webtools/widget/ServiceForms.xml Tue Jul 17 09:20:42 2012 > @@ -110,14 +110,24 @@ under the License. > <field name="key"><display/></field> > <field name="value"><display/></field> > </form> > + <form name="PoolState" type="single" default-map-name="poolState"> > + <field name="pollerName"><display/></field> > + <field name="pollerThreadName"><display/></field> > + <field name="invokerThreadNameFormat"><display/></field> > + <field name="keepAliveTimeInSeconds"><display/></field> > + <field name="numberOfCoreInvokerThreads"><display/></field> > + <field name="currentNumberOfInvokerThreads"><display/></field> > + <field name="numberOfActiveInvokerThreads"><display/></field> > + <field name="maxNumberOfInvokerThreads"><display/></field> > + <field name="greatestNumberOfInvokerThreads"><display/></field> > + <field name="numberOfCompletedTasks"><display/></field> > + </form> > <form name="ListJavaThread" type="list" list-name="threads" paginate-target="threadList" separate-columns="true" > odd-row-style="alternate-row" default-table-style="basic-table hover-bar"> > - <field name="threadId" title="${uiLabelMap.WebtoolsThread}"><display description="${threadId} ${threadName}"/></field> > - <field name="status" title="${uiLabelMap.CommonStatus}"><display/></field> > - <field name="jobName" title="${uiLabelMap.WebtoolsJob}"><display default-value="${uiLabelMap.CommonNone}"/></field> > + <field name="id" title="${uiLabelMap.WebtoolsThread}"><display description="${threadId} ${threadName}"/></field> > + <field name="name" title="${uiLabelMap.WebtoolsJob}"><display default-value="${uiLabelMap.CommonNone}"/></field> > <field name="serviceName" title="${uiLabelMap.WebtoolsService}"><display default-value="${uiLabelMap.CommonNone}"/></field> > - <field name="usage" title="${uiLabelMap.WebtoolsUsage}"><display/></field> > - <field name="ttl" title="${uiLabelMap.WebtoolsTTL} (ms)"><display/></field> > + <field name="time"><display/></field> > <field name="runTime" title="${uiLabelMap.CommonTime} (ms)"><display/></field> > </form> > <form name="ListServices" type="list" list-name="services" paginate-target="ServiceLog" separate-columns="true" > > Modified: ofbiz/trunk/framework/webtools/widget/ServiceScreens.xml > URL: http://svn.apache.org/viewvc/ofbiz/trunk/framework/webtools/widget/ServiceScreens.xml?rev=1362422&r1=1362421&r2=1362422&view=diff > ============================================================================== > --- ofbiz/trunk/framework/webtools/widget/ServiceScreens.xml (original) > +++ ofbiz/trunk/framework/webtools/widget/ServiceScreens.xml Tue Jul 17 09:20:42 2012 > @@ -103,6 +103,7 @@ under the License. > <decorator-screen name="CommonServiceDecorator" location="${parameters.mainDecoratorLocation}"> > <decorator-section name="body"> > <screenlet title="${uiLabelMap.WebtoolsServiceEngineThreads}"> > + <include-form name="PoolState" location="component://webtools/widget/ServiceForms.xml"/> > <include-form name="ListJavaThread" location="component://webtools/widget/ServiceForms.xml"/> > </screenlet> > <screenlet title="${uiLabelMap.WebtoolsGeneralJavaThreads}"> > > |
«
Return to OFBiz - Dev
|
1 view|%1 views
| Free forum by Nabble | Edit this page |