You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@continuum.apache.org by ba...@apache.org on 2015/04/09 00:51:10 UTC
svn commit: r1672215 - in /continuum/trunk:
continuum-api/src/main/java/org/apache/continuum/taskqueue/
continuum-api/src/main/java/org/apache/continuum/utils/
continuum-buildagent/continuum-buildagent-core/src/main/resources/META-INF/plexus/
continuum...
Author: batkinson
Date: Wed Apr 8 22:51:10 2015
New Revision: 1672215
URL: http://svn.apache.org/r1672215
Log:
Replaced default plexus task queue with enhanced version, better thread names and shutdown.
Added:
continuum/trunk/continuum-api/src/main/java/org/apache/continuum/taskqueue/ThreadedTaskQueueExecutor.java
continuum/trunk/continuum-api/src/main/java/org/apache/continuum/utils/ThreadNames.java
Modified:
continuum/trunk/continuum-buildagent/continuum-buildagent-core/src/main/resources/META-INF/plexus/components-fragment.xml
continuum/trunk/continuum-purge/continuum-purge-core/src/main/resources/META-INF/plexus/components-fragment.xml
continuum/trunk/continuum-release/src/main/resources/META-INF/plexus/components.xml
Added: continuum/trunk/continuum-api/src/main/java/org/apache/continuum/taskqueue/ThreadedTaskQueueExecutor.java
URL: http://svn.apache.org/viewvc/continuum/trunk/continuum-api/src/main/java/org/apache/continuum/taskqueue/ThreadedTaskQueueExecutor.java?rev=1672215&view=auto
==============================================================================
--- continuum/trunk/continuum-api/src/main/java/org/apache/continuum/taskqueue/ThreadedTaskQueueExecutor.java (added)
+++ continuum/trunk/continuum-api/src/main/java/org/apache/continuum/taskqueue/ThreadedTaskQueueExecutor.java Wed Apr 8 22:51:10 2015
@@ -0,0 +1,352 @@
+package org.apache.continuum.taskqueue;
+
+/*
+ * 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.
+ */
+
+import edu.emory.mathcs.backport.java.util.concurrent.CancellationException;
+import edu.emory.mathcs.backport.java.util.concurrent.ExecutionException;
+import edu.emory.mathcs.backport.java.util.concurrent.ExecutorService;
+import edu.emory.mathcs.backport.java.util.concurrent.Executors;
+import edu.emory.mathcs.backport.java.util.concurrent.Future;
+import edu.emory.mathcs.backport.java.util.concurrent.TimeUnit;
+import edu.emory.mathcs.backport.java.util.concurrent.TimeoutException;
+import org.apache.continuum.utils.ThreadNames;
+import org.codehaus.plexus.component.annotations.Configuration;
+import org.codehaus.plexus.component.annotations.Requirement;
+import org.codehaus.plexus.logging.AbstractLogEnabled;
+import org.codehaus.plexus.personality.plexus.lifecycle.phase.Disposable;
+import org.codehaus.plexus.personality.plexus.lifecycle.phase.Initializable;
+import org.codehaus.plexus.personality.plexus.lifecycle.phase.InitializationException;
+import org.codehaus.plexus.personality.plexus.lifecycle.phase.Startable;
+import org.codehaus.plexus.personality.plexus.lifecycle.phase.StartingException;
+import org.codehaus.plexus.personality.plexus.lifecycle.phase.StoppingException;
+import org.codehaus.plexus.taskqueue.Task;
+import org.codehaus.plexus.taskqueue.TaskQueue;
+import org.codehaus.plexus.taskqueue.execution.TaskExecutionException;
+import org.codehaus.plexus.taskqueue.execution.TaskExecutor;
+import org.codehaus.plexus.taskqueue.execution.TaskQueueExecutor;
+import org.codehaus.plexus.util.StringUtils;
+
+/**
+ * Modified plexus ThreadedTaskQueueExecutor
+ */
+public class ThreadedTaskQueueExecutor
+ extends AbstractLogEnabled
+ implements TaskQueueExecutor, Initializable, Startable, Disposable
+{
+ private static final int SHUTDOWN = 1;
+
+ private static final int CANCEL_TASK = 2;
+
+ @Requirement
+ private TaskQueue queue;
+
+ @Requirement
+ private TaskExecutor executor;
+
+ @Configuration( "" )
+ private String name;
+
+ // ----------------------------------------------------------------------
+ //
+ // ----------------------------------------------------------------------
+
+ private ExecutorRunnable executorRunnable;
+
+ private ExecutorService executorService;
+
+ private Task currentTask;
+
+ private class ExecutorRunnable
+ extends Thread
+ {
+ private volatile int command;
+
+ private boolean done;
+
+ public ExecutorRunnable()
+ {
+ super( ThreadNames.formatNext( "%s-executor", name ) );
+ }
+
+ public void run()
+ {
+ while ( command != SHUTDOWN )
+ {
+ final Task task;
+
+ currentTask = null;
+
+ try
+ {
+ task = queue.poll( 100, TimeUnit.MILLISECONDS );
+ }
+ catch ( InterruptedException e )
+ {
+ getLogger().info( "Executor thread interrupted, command: "
+ + ( command == SHUTDOWN ? "Shutdown" :
+ command == CANCEL_TASK ? "Cancel task" : "Unknown" ) );
+ continue;
+ }
+
+ if ( task == null )
+ {
+ continue;
+ }
+
+ currentTask = task;
+
+ Future future = executorService.submit( new Runnable()
+ {
+ public void run()
+ {
+ try
+ {
+ executor.executeTask( task );
+ }
+ catch ( TaskExecutionException e )
+ {
+ getLogger().error( "Error executing task", e );
+ }
+ }
+ } );
+
+ try
+ {
+ waitForTask( task, future );
+ }
+ catch ( ExecutionException e )
+ {
+ getLogger().error( "Error executing task", e );
+ }
+ }
+
+ currentTask = null;
+
+ getLogger().info( "Executor thread '" + name + "' exited." );
+
+ done = true;
+
+ synchronized ( this )
+ {
+ notifyAll();
+ }
+ }
+
+ private void waitForTask( Task task, Future future )
+ throws ExecutionException
+ {
+ boolean stop = false;
+
+ while ( !stop )
+ {
+ try
+ {
+ if ( task.getMaxExecutionTime() == 0 )
+ {
+ getLogger().debug( "Waiting indefinitely for task to complete" );
+ future.get();
+ return;
+ }
+ else
+ {
+ getLogger().debug( "Waiting at most " + task.getMaxExecutionTime() + "ms for task completion" );
+ future.get( task.getMaxExecutionTime(), TimeUnit.MILLISECONDS );
+ getLogger().debug( "Task completed within " + task.getMaxExecutionTime() + "ms" );
+ return;
+ }
+ }
+ catch ( InterruptedException e )
+ {
+ switch ( command )
+ {
+ case SHUTDOWN:
+ {
+ getLogger().info( "Shutdown command received. Cancelling task." );
+ cancel( future );
+ return;
+ }
+
+ case CANCEL_TASK:
+ {
+ command = 0;
+ getLogger().info( "Cancelling task" );
+ cancel( future );
+ return;
+ }
+
+ default:
+ // when can this thread be interrupted, and should we ignore it if shutdown = false?
+ getLogger().warn( "Interrupted while waiting for task to complete; ignoring", e );
+ break;
+ }
+ }
+ catch ( TimeoutException e )
+ {
+ getLogger().warn( "Task " + task + " didn't complete within time, cancelling it." );
+ cancel( future );
+ return;
+ }
+ catch ( CancellationException e )
+ {
+ getLogger().warn( "The task was cancelled", e );
+ return;
+ }
+ }
+ }
+
+ private void cancel( Future future )
+ {
+ if ( !future.cancel( true ) )
+ {
+ if ( !future.isDone() && !future.isCancelled() )
+ {
+ getLogger().warn( "Unable to cancel task" );
+ }
+ else
+ {
+ getLogger().warn( "Task not cancelled (Flags: done: " + future.isDone() + " cancelled: "
+ + future.isCancelled() + ")" );
+ }
+ }
+ else
+ {
+ getLogger().debug( "Task successfully cancelled" );
+ }
+ }
+
+ public synchronized void shutdown()
+ {
+ getLogger().debug( "Signalling executor thread to shutdown" );
+
+ command = SHUTDOWN;
+
+ interrupt();
+ }
+
+ public synchronized boolean cancelTask( Task task )
+ {
+ if ( !task.equals( currentTask ) )
+ {
+ getLogger().debug( "Not cancelling task - it is not running" );
+ return false;
+ }
+
+ if ( command != SHUTDOWN )
+ {
+ getLogger().debug( "Signalling executor thread to cancel task" );
+
+ command = CANCEL_TASK;
+
+ interrupt();
+ }
+ else
+ {
+ getLogger().debug( "Executor thread already stopping; task will be cancelled automatically" );
+ }
+
+ return true;
+ }
+
+ public boolean isDone()
+ {
+ return done;
+ }
+ }
+
+ // ----------------------------------------------------------------------
+ // Component lifecycle
+ // ----------------------------------------------------------------------
+
+ public void initialize()
+ throws InitializationException
+ {
+ if ( StringUtils.isEmpty( name ) )
+ {
+ throw new IllegalArgumentException( "'name' must be set." );
+ }
+ }
+
+ public void start()
+ throws StartingException
+ {
+ getLogger().info( "Starting task executor, thread name '" + name + "'." );
+
+ this.executorService = Executors.newSingleThreadExecutor();
+
+ executorRunnable = new ExecutorRunnable();
+
+ executorRunnable.setDaemon( true );
+
+ executorRunnable.start();
+ }
+
+ public void stop()
+ throws StoppingException
+ {
+ executorRunnable.shutdown();
+
+ int maxSleep = 10 * 1000; // 10 seconds
+
+ int interval = 1000;
+
+ long endTime = System.currentTimeMillis() + maxSleep;
+
+ while ( !executorRunnable.isDone() && executorRunnable.isAlive() )
+ {
+ if ( System.currentTimeMillis() > endTime )
+ {
+ getLogger().warn( "Timeout waiting for executor thread '" + name + "' to stop, aborting" );
+ break;
+ }
+
+ getLogger().info( "Waiting until task executor '" + name + "' is idling..." );
+
+ try
+ {
+ synchronized ( executorRunnable )
+ {
+ executorRunnable.wait( interval );
+ }
+ }
+ catch ( InterruptedException ex )
+ {
+ // ignore
+ }
+
+ // notify again, just in case.
+ executorRunnable.shutdown();
+ }
+ }
+
+ public void dispose()
+ {
+ executorRunnable.shutdown();
+ }
+
+ public Task getCurrentTask()
+ {
+ return currentTask;
+ }
+
+ public synchronized boolean cancelTask( Task task )
+ {
+ return executorRunnable.cancelTask( task );
+ }
+}
Added: continuum/trunk/continuum-api/src/main/java/org/apache/continuum/utils/ThreadNames.java
URL: http://svn.apache.org/viewvc/continuum/trunk/continuum-api/src/main/java/org/apache/continuum/utils/ThreadNames.java?rev=1672215&view=auto
==============================================================================
--- continuum/trunk/continuum-api/src/main/java/org/apache/continuum/utils/ThreadNames.java (added)
+++ continuum/trunk/continuum-api/src/main/java/org/apache/continuum/utils/ThreadNames.java Wed Apr 8 22:51:10 2015
@@ -0,0 +1,31 @@
+package org.apache.continuum.utils;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * A simple utility for creating thread names uniformly.
+ */
+public class ThreadNames
+{
+ private static Map<String, Integer> threadCountByName = new HashMap<String, Integer>();
+
+ public static String formatNext( String format, Object... args )
+ {
+ int nextId;
+ String baseName = String.format( format, args );
+ synchronized ( threadCountByName )
+ {
+ if ( !threadCountByName.containsKey( baseName ) )
+ {
+ nextId = 0;
+ }
+ else
+ {
+ nextId = threadCountByName.get( baseName );
+ }
+ threadCountByName.put( baseName, nextId + 1 );
+ }
+ return String.format( "%s-%s", baseName, nextId );
+ }
+}
Modified: continuum/trunk/continuum-buildagent/continuum-buildagent-core/src/main/resources/META-INF/plexus/components-fragment.xml
URL: http://svn.apache.org/viewvc/continuum/trunk/continuum-buildagent/continuum-buildagent-core/src/main/resources/META-INF/plexus/components-fragment.xml?rev=1672215&r1=1672214&r2=1672215&view=diff
==============================================================================
--- continuum/trunk/continuum-buildagent/continuum-buildagent-core/src/main/resources/META-INF/plexus/components-fragment.xml (original)
+++ continuum/trunk/continuum-buildagent/continuum-buildagent-core/src/main/resources/META-INF/plexus/components-fragment.xml Wed Apr 8 22:51:10 2015
@@ -36,7 +36,7 @@
<component>
<role>org.codehaus.plexus.taskqueue.execution.TaskQueueExecutor</role>
<role-hint>build-agent</role-hint>
- <implementation>org.codehaus.plexus.taskqueue.execution.ThreadedTaskQueueExecutor</implementation>
+ <implementation>org.apache.continuum.taskqueue.ThreadedTaskQueueExecutor</implementation>
<instantiation-strategy>singleton</instantiation-strategy>
<requirements>
<requirement>
@@ -56,7 +56,7 @@
<component>
<role>org.codehaus.plexus.taskqueue.execution.TaskQueueExecutor</role>
<role-hint>prepare-build-agent</role-hint>
- <implementation>org.codehaus.plexus.taskqueue.execution.ThreadedTaskQueueExecutor</implementation>
+ <implementation>org.apache.continuum.taskqueue.ThreadedTaskQueueExecutor</implementation>
<instantiation-strategy>singleton</instantiation-strategy>
<requirements>
<requirement>
Modified: continuum/trunk/continuum-purge/continuum-purge-core/src/main/resources/META-INF/plexus/components-fragment.xml
URL: http://svn.apache.org/viewvc/continuum/trunk/continuum-purge/continuum-purge-core/src/main/resources/META-INF/plexus/components-fragment.xml?rev=1672215&r1=1672214&r2=1672215&view=diff
==============================================================================
--- continuum/trunk/continuum-purge/continuum-purge-core/src/main/resources/META-INF/plexus/components-fragment.xml (original)
+++ continuum/trunk/continuum-purge/continuum-purge-core/src/main/resources/META-INF/plexus/components-fragment.xml Wed Apr 8 22:51:10 2015
@@ -44,7 +44,7 @@
<component>
<role>org.codehaus.plexus.taskqueue.execution.TaskQueueExecutor</role>
<role-hint>purge</role-hint>
- <implementation>org.codehaus.plexus.taskqueue.execution.ThreadedTaskQueueExecutor</implementation>
+ <implementation>org.apache.continuum.taskqueue.ThreadedTaskQueueExecutor</implementation>
<instantiation-strategy>singleton</instantiation-strategy>
<requirements>
<requirement>
Modified: continuum/trunk/continuum-release/src/main/resources/META-INF/plexus/components.xml
URL: http://svn.apache.org/viewvc/continuum/trunk/continuum-release/src/main/resources/META-INF/plexus/components.xml?rev=1672215&r1=1672214&r2=1672215&view=diff
==============================================================================
--- continuum/trunk/continuum-release/src/main/resources/META-INF/plexus/components.xml (original)
+++ continuum/trunk/continuum-release/src/main/resources/META-INF/plexus/components.xml Wed Apr 8 22:51:10 2015
@@ -91,7 +91,7 @@
<component>
<role>org.codehaus.plexus.taskqueue.execution.TaskQueueExecutor</role>
<role-hint>prepare-release</role-hint>
- <implementation>org.codehaus.plexus.taskqueue.execution.ThreadedTaskQueueExecutor</implementation>
+ <implementation>org.apache.continuum.taskqueue.ThreadedTaskQueueExecutor</implementation>
<requirements>
<requirement>
<role>org.codehaus.plexus.taskqueue.execution.TaskExecutor</role>
@@ -152,7 +152,7 @@
<component>
<role>org.codehaus.plexus.taskqueue.execution.TaskQueueExecutor</role>
<role-hint>perform-release</role-hint>
- <implementation>org.codehaus.plexus.taskqueue.execution.ThreadedTaskQueueExecutor</implementation>
+ <implementation>org.apache.continuum.taskqueue.ThreadedTaskQueueExecutor</implementation>
<requirements>
<requirement>
<role>org.codehaus.plexus.taskqueue.execution.TaskExecutor</role>
@@ -210,7 +210,7 @@
<component>
<role>org.codehaus.plexus.taskqueue.execution.TaskQueueExecutor</role>
<role-hint>rollback-release</role-hint>
- <implementation>org.codehaus.plexus.taskqueue.execution.ThreadedTaskQueueExecutor</implementation>
+ <implementation>org.apache.continuum.taskqueue.ThreadedTaskQueueExecutor</implementation>
<requirements>
<requirement>
<role>org.codehaus.plexus.taskqueue.execution.TaskExecutor</role>