You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sling.apache.org by cz...@apache.org on 2008/02/18 11:40:17 UTC
svn commit: r628678 - in /incubator/sling/trunk/sling:
event/src/main/java/org/apache/sling/event/impl/
event/src/test/java/org/apache/sling/event/impl/ scheduler/
scheduler/src/main/java/org/apache/sling/scheduler/impl/
threads/src/main/java/org/apach...
Author: cziegeler
Date: Mon Feb 18 02:40:12 2008
New Revision: 628678
URL: http://svn.apache.org/viewvc?rev=628678&view=rev
Log:
Use thread pooling for scheduler as well. Create own pool for eventing.
Modified:
incubator/sling/trunk/sling/event/src/main/java/org/apache/sling/event/impl/AbstractRepositoryEventHandler.java
incubator/sling/trunk/sling/event/src/test/java/org/apache/sling/event/impl/AbstractRepositoryEventHandlerTest.java
incubator/sling/trunk/sling/scheduler/pom.xml
incubator/sling/trunk/sling/scheduler/src/main/java/org/apache/sling/scheduler/impl/QuartzScheduler.java
incubator/sling/trunk/sling/threads/src/main/java/org/apache/sling/threads/ThreadPool.java
incubator/sling/trunk/sling/threads/src/main/java/org/apache/sling/threads/impl/DefaultThreadFactory.java
incubator/sling/trunk/sling/threads/src/main/java/org/apache/sling/threads/impl/DefaultThreadPool.java
Modified: incubator/sling/trunk/sling/event/src/main/java/org/apache/sling/event/impl/AbstractRepositoryEventHandler.java
URL: http://svn.apache.org/viewvc/incubator/sling/trunk/sling/event/src/main/java/org/apache/sling/event/impl/AbstractRepositoryEventHandler.java?rev=628678&r1=628677&r2=628678&view=diff
==============================================================================
--- incubator/sling/trunk/sling/event/src/main/java/org/apache/sling/event/impl/AbstractRepositoryEventHandler.java (original)
+++ incubator/sling/trunk/sling/event/src/main/java/org/apache/sling/event/impl/AbstractRepositoryEventHandler.java Mon Feb 18 02:40:12 2008
@@ -39,7 +39,6 @@
import org.apache.sling.jcr.api.SlingRepository;
import org.apache.sling.threads.ThreadPool;
import org.apache.sling.threads.ThreadPoolManager;
-import org.osgi.framework.ServiceReference;
import org.osgi.service.component.ComponentContext;
import org.osgi.service.event.Event;
import org.osgi.service.event.EventAdmin;
@@ -92,8 +91,8 @@
/** A local queue for writing received events into the repository. */
protected final BlockingQueue<Event> writeQueue = new LinkedBlockingQueue<Event>();
- /** The reference to the thread pool manager. */
- protected ServiceReference threadPoolManagerRef;
+ /** @scr.reference */
+ protected ThreadPoolManager threadPoolManager;
/** Our thread pool. */
protected ThreadPool threadPool;
@@ -109,57 +108,46 @@
this.repositoryPath = (String)context.getProperties().get(CONFIG_PROPERTY_REPO_PATH);
// start background threads
- this.threadPoolManagerRef = context.getBundleContext().getServiceReference(ThreadPoolManager.class.getName());
- if ( this.threadPoolManagerRef == null ) {
+ if ( this.threadPoolManager == null ) {
throw new Exception("No ThreadPoolManager found.");
}
- try {
- final ThreadPoolManager threadPoolManager = (ThreadPoolManager) context.getBundleContext().getService(this.threadPoolManagerRef);
- if ( threadPoolManager == null ) {
- throw new Exception("No ThreadPoolManager found.");
- }
- threadPoolManager.create(EventHelper.THREAD_POOL_NAME,
- 10, 30, 30, -1,
- ThreadPoolManager.DEFAULT_BLOCK_POLICY, true, 5000, null, 0, false);
- this.threadPool = threadPoolManager.get(EventHelper.THREAD_POOL_NAME);
- if ( this.threadPool == null ) {
- throw new Exception("No thread pool found.");
- }
- this.running = true;
- // start writer thread
- this.threadPool.execute(new Runnable() {
- public void run() {
- try {
- startWriterSession();
- } catch (RepositoryException e) {
- // there is nothing we can do except log!
- logger.error("Error during session starting.", e);
- running = false;
- }
- try {
- processWriteQueue();
- } catch (Throwable t) {
- logger.error("Writer thread stopped with exception: " + t.getMessage(), t);
- running = false;
- }
- stopWriterSession();
+ threadPoolManager.create(EventHelper.THREAD_POOL_NAME,
+ 10, 30, 30, -1,
+ ThreadPoolManager.DEFAULT_BLOCK_POLICY, true, 5000, null, Thread.NORM_PRIORITY, false);
+ this.threadPool = threadPoolManager.get(EventHelper.THREAD_POOL_NAME);
+ if ( this.threadPool == null ) {
+ throw new Exception("No thread pool found.");
+ }
+ this.running = true;
+ // start writer thread
+ this.threadPool.execute(new Runnable() {
+ public void run() {
+ try {
+ startWriterSession();
+ } catch (RepositoryException e) {
+ // there is nothing we can do except log!
+ logger.error("Error during session starting.", e);
+ running = false;
}
- });
- this.threadPool.execute(new Runnable() {
- public void run() {
- try {
- runInBackground();
- } catch (Throwable t) {
- logger.error("Background thread stopped with exception: " + t.getMessage(), t);
- running = false;
- }
+ try {
+ processWriteQueue();
+ } catch (Throwable t) {
+ logger.error("Writer thread stopped with exception: " + t.getMessage(), t);
+ running = false;
}
- });
- } catch (Exception e) {
- context.getBundleContext().ungetService(this.threadPoolManagerRef);
- this.threadPoolManagerRef = null;
- throw e;
- }
+ stopWriterSession();
+ }
+ });
+ this.threadPool.execute(new Runnable() {
+ public void run() {
+ try {
+ runInBackground();
+ } catch (Throwable t) {
+ logger.error("Background thread stopped with exception: " + t.getMessage(), t);
+ running = false;
+ }
+ }
+ });
}
protected abstract void runInBackground() throws RepositoryException;
@@ -182,10 +170,6 @@
this.queue.put(new EventInfo());
} catch (InterruptedException e) {
this.ignoreException(e);
- }
- if ( this.threadPoolManagerRef != null ) {
- context.getBundleContext().ungetService(this.threadPoolManagerRef);
- this.threadPoolManagerRef = null;
}
this.threadPool = null;
}
Modified: incubator/sling/trunk/sling/event/src/test/java/org/apache/sling/event/impl/AbstractRepositoryEventHandlerTest.java
URL: http://svn.apache.org/viewvc/incubator/sling/trunk/sling/event/src/test/java/org/apache/sling/event/impl/AbstractRepositoryEventHandlerTest.java?rev=628678&r1=628677&r2=628678&view=diff
==============================================================================
--- incubator/sling/trunk/sling/event/src/test/java/org/apache/sling/event/impl/AbstractRepositoryEventHandlerTest.java (original)
+++ incubator/sling/trunk/sling/event/src/test/java/org/apache/sling/event/impl/AbstractRepositoryEventHandlerTest.java Mon Feb 18 02:40:12 2008
@@ -40,7 +40,6 @@
import org.jmock.integration.junit4.JMock;
import org.junit.runner.RunWith;
import org.osgi.framework.BundleContext;
-import org.osgi.framework.ServiceReference;
import org.osgi.service.component.ComponentContext;
import org.osgi.service.event.Event;
import org.osgi.service.event.EventAdmin;
@@ -57,8 +56,6 @@
protected abstract Mockery getMockery();
- protected ServiceReference thpmSR;
-
protected Dictionary<String, Object> getComponentConfig() {
final Dictionary<String, Object> config = new Hashtable<String, Object>();
config.put(AbstractRepositoryEventHandler.CONFIG_PROPERTY_REPO_PATH, REPO_PATH);
@@ -88,16 +85,14 @@
allowing(eventAdmin).sendEvent(with(any(Event.class)));
}});
- // we need a service reference for the thread pool manager
- // and a thpm
- thpmSR = this.getMockery().mock(ServiceReference.class);
+ // we need a thread pool manager
+ this.handler.threadPoolManager = this.getMockery().mock(ThreadPoolManager.class);
final ThreadPool pool = new ThreadPoolImpl();
- final ThreadPoolManager thpm = this.getMockery().mock(ThreadPoolManager.class);
this.getMockery().checking(new Expectations() {{
- allowing(thpm).get(EventHelper.THREAD_POOL_NAME);
+ allowing(handler.threadPoolManager).get(EventHelper.THREAD_POOL_NAME);
will(returnValue(pool));
- allowing(thpm).create(EventHelper.THREAD_POOL_NAME, 10, 30, 30, -1,
- ThreadPoolManager.DEFAULT_BLOCK_POLICY, true, 5000, null, 0, false);
+ allowing(handler.threadPoolManager).create(EventHelper.THREAD_POOL_NAME, 10, 30, 30, -1,
+ ThreadPoolManager.DEFAULT_BLOCK_POLICY, true, 5000, null, Thread.NORM_PRIORITY, false);
will(returnValue(null));
}});
@@ -106,10 +101,6 @@
this.getMockery().checking(new Expectations() {{
allowing(bundleContext).getProperty(AbstractRepositoryEventHandler.SLING_ID);
will(returnValue(SLING_ID));
- allowing(bundleContext).getServiceReference(ThreadPoolManager.class.getName());
- will(returnValue(thpmSR));
- allowing(bundleContext).getService(thpmSR);
- will(returnValue(thpm));
}});
// lets set up the component configuration
@@ -140,9 +131,6 @@
rootNode.save();
// lets set up the bundle context with the sling id
final BundleContext bundleContext = this.getMockery().mock(BundleContext.class);
- this.getMockery().checking(new Expectations() {{
- allowing(bundleContext).ungetService(thpmSR);
- }});
final ComponentContext componentContext = this.getMockery().mock(ComponentContext.class);
this.getMockery().checking(new Expectations() {{
@@ -177,14 +165,15 @@
}
public String getName() {
- // TODO Auto-generated method stub
- return null;
+ return EventHelper.THREAD_POOL_NAME;
}
public void shutdown() {
- // TODO Auto-generated method stub
-
+ // nothing to do
}
+ public int getMaxPoolSize() {
+ return 30;
+ }
}
}
Modified: incubator/sling/trunk/sling/scheduler/pom.xml
URL: http://svn.apache.org/viewvc/incubator/sling/trunk/sling/scheduler/pom.xml?rev=628678&r1=628677&r2=628678&view=diff
==============================================================================
--- incubator/sling/trunk/sling/scheduler/pom.xml (original)
+++ incubator/sling/trunk/sling/scheduler/pom.xml Mon Feb 18 02:40:12 2008
@@ -105,6 +105,11 @@
<artifactId>org.apache.sling.api</artifactId>
<version>2.0.0-incubator-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.sling</groupId>
+ <artifactId>org.apache.sling.threads</artifactId>
+ <version>2.0.0-incubator-SNAPSHOT</version>
+ </dependency>
<!-- The following dependencies will be replaced by using the quartz wrapper bundle -->
<dependency>
<groupId>quartz</groupId>
Modified: incubator/sling/trunk/sling/scheduler/src/main/java/org/apache/sling/scheduler/impl/QuartzScheduler.java
URL: http://svn.apache.org/viewvc/incubator/sling/trunk/sling/scheduler/src/main/java/org/apache/sling/scheduler/impl/QuartzScheduler.java?rev=628678&r1=628677&r2=628678&view=diff
==============================================================================
--- incubator/sling/trunk/sling/scheduler/src/main/java/org/apache/sling/scheduler/impl/QuartzScheduler.java (original)
+++ incubator/sling/trunk/sling/scheduler/src/main/java/org/apache/sling/scheduler/impl/QuartzScheduler.java Mon Feb 18 02:40:12 2008
@@ -27,6 +27,8 @@
import org.apache.sling.scheduler.Job;
import org.apache.sling.scheduler.Scheduler;
+import org.apache.sling.threads.ThreadPool;
+import org.apache.sling.threads.ThreadPoolManager;
import org.osgi.framework.Constants;
import org.osgi.framework.ServiceReference;
import org.osgi.service.component.ComponentContext;
@@ -37,7 +39,9 @@
import org.quartz.SchedulerFactory;
import org.quartz.SimpleTrigger;
import org.quartz.Trigger;
+import org.quartz.impl.DirectSchedulerFactory;
import org.quartz.impl.StdSchedulerFactory;
+import org.quartz.simpl.RAMJobStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -56,6 +60,8 @@
protected static final String DEFAULT_QUARTZ_JOB_GROUP = "Sling";
+ protected static final String THREAD_POOL_NAME = "SLING_SCHEDULER";
+
/** Map key for the job object */
static final String DATA_MAP_OBJECT = "QuartzJobScheduler.Object";
@@ -83,6 +89,9 @@
protected ComponentContext context;
+ /** @scr.reference */
+ protected ThreadPoolManager threadPoolManager;
+
/**
* Activate this component.
* Start the scheduler.
@@ -119,8 +128,18 @@
}
protected void init() throws SchedulerException {
- final SchedulerFactory factory = new StdSchedulerFactory();
- this.scheduler = factory.getScheduler();
+ // if we don't have a thread pool manager, we use the default thread pool
+ final ThreadPoolManager tpm = this.threadPoolManager;
+ if ( tpm == null ) {
+ final SchedulerFactory factory = new StdSchedulerFactory();
+ this.scheduler = factory.getScheduler();
+ } else {
+ final ThreadPool pool = tpm.get(THREAD_POOL_NAME);
+ final QuartzThreadPool quartzPool = new QuartzThreadPool(pool);
+ final DirectSchedulerFactory factory = DirectSchedulerFactory.getInstance();
+ factory.createScheduler(quartzPool, new RAMJobStore());
+ this.scheduler = factory.getScheduler();
+ }
this.scheduler.start();
if ( this.logger.isDebugEnabled() ) {
this.logger.debug("Scheduler started.");
@@ -394,4 +413,53 @@
}
}
}
+
+ private static final class QuartzThreadPool implements org.quartz.spi.ThreadPool {
+
+ /** Default log. */
+ protected final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+ /** Our executor thread pool */
+ private ThreadPool executor;
+
+ /**
+ *
+ */
+ public QuartzThreadPool(final ThreadPool executor) {
+ this.executor = executor;
+ }
+
+ /* (non-Javadoc)
+ * @see org.quartz.spi.QuartzThreadPool#getPoolSize()
+ */
+ public int getPoolSize() {
+ return this.executor.getMaxPoolSize();
+ }
+
+ /* (non-Javadoc)
+ * @see org.quartz.spi.QuartzThreadPool#initialize()
+ */
+ public void initialize() {
+ // nothing to do
+ }
+
+ /* (non-Javadoc)
+ * @see org.quartz.spi.QuartzThreadPool#runInThread(java.lang.Runnable)
+ */
+ public boolean runInThread(final Runnable job) {
+ this.executor.execute(job);
+
+ return true;
+ }
+
+ /* (non-Javadoc)
+ * @see org.quartz.spi.QuartzThreadPool#shutdown(boolean)
+ */
+ public void shutdown(final boolean waitForJobsToComplete) {
+ // the pool is managed by the thread pool manager,
+ // so we can just return
+ this.executor = null;
+ }
+ }
+
}
Modified: incubator/sling/trunk/sling/threads/src/main/java/org/apache/sling/threads/ThreadPool.java
URL: http://svn.apache.org/viewvc/incubator/sling/trunk/sling/threads/src/main/java/org/apache/sling/threads/ThreadPool.java?rev=628678&r1=628677&r2=628678&view=diff
==============================================================================
--- incubator/sling/trunk/sling/threads/src/main/java/org/apache/sling/threads/ThreadPool.java (original)
+++ incubator/sling/trunk/sling/threads/src/main/java/org/apache/sling/threads/ThreadPool.java Mon Feb 18 02:40:12 2008
@@ -39,4 +39,6 @@
* Shut down the thread pool.
*/
void shutdown();
+
+ int getMaxPoolSize();
}
Modified: incubator/sling/trunk/sling/threads/src/main/java/org/apache/sling/threads/impl/DefaultThreadFactory.java
URL: http://svn.apache.org/viewvc/incubator/sling/trunk/sling/threads/src/main/java/org/apache/sling/threads/impl/DefaultThreadFactory.java?rev=628678&r1=628677&r2=628678&view=diff
==============================================================================
--- incubator/sling/trunk/sling/threads/src/main/java/org/apache/sling/threads/impl/DefaultThreadFactory.java (original)
+++ incubator/sling/trunk/sling/threads/src/main/java/org/apache/sling/threads/impl/DefaultThreadFactory.java Mon Feb 18 02:40:12 2008
@@ -55,7 +55,7 @@
( Thread.NORM_PRIORITY == priority ) ) {
this.priority = priority;
} else {
- throw new IllegalStateException("Unknown priority " + this.priority);
+ throw new IllegalStateException("Unknown priority " + priority);
}
}
Modified: incubator/sling/trunk/sling/threads/src/main/java/org/apache/sling/threads/impl/DefaultThreadPool.java
URL: http://svn.apache.org/viewvc/incubator/sling/trunk/sling/threads/src/main/java/org/apache/sling/threads/impl/DefaultThreadPool.java?rev=628678&r1=628677&r2=628678&view=diff
==============================================================================
--- incubator/sling/trunk/sling/threads/src/main/java/org/apache/sling/threads/impl/DefaultThreadPool.java (original)
+++ incubator/sling/trunk/sling/threads/src/main/java/org/apache/sling/threads/impl/DefaultThreadPool.java Mon Feb 18 02:40:12 2008
@@ -172,6 +172,13 @@
}
/**
+ * @see org.apache.sling.threads.ThreadPool#getMaxPoolSize()
+ */
+ public int getMaxPoolSize() {
+ return this.executor.getMaximumPoolSize();
+ }
+
+ /**
* @see org.apache.sling.threads.ThreadPool#execute(java.lang.Runnable)
*/
public void execute(Runnable runnable) {