ThreadLauncherDaemon.java
/*
* Copyright (c) 2002-2022, City of Paris
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
*
* 1. Redistributions of source code must retain the above copyright notice
* and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright notice
* and the following disclaimer in the documentation and/or other materials
* provided with the distribution.
*
* 3. Neither the name of 'Mairie de Paris' nor 'Lutece' nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*
* License 1.0
*/
package fr.paris.lutece.portal.service.daemon;
import fr.paris.lutece.portal.service.plugin.Plugin;
import fr.paris.lutece.portal.service.util.AppPropertiesService;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.commons.collections.CollectionUtils;
/**
* Daemon that manage a pool of threads to launch runnables.
*/
public class ThreadLauncherDaemon extends Daemon
{
private static final String THREAD_LAUNCHER_DAEMON_ID = "threadLauncherDaemon";
/**
* Runnable wrapper to signal the daemon on processing end
*/
private class RunnableWrapper implements Runnable
{
private final Runnable _wrapped;
/**
* Constructor
*
* @param wrapped
* the wrapped Runnable
*/
public RunnableWrapper( Runnable wrapped )
{
_wrapped = wrapped;
}
@Override
public void run( )
{
try
{
_wrapped.run( );
}
finally
{
// will clean up the dead thread
// and permit execution of blocked tasks
AppDaemonService.signalDaemon( THREAD_LAUNCHER_DAEMON_ID );
}
}
}
private static final String PROPERTY_MAX_NUMBER_THREAD = "daemon.threadLauncherDaemon.maxNumberOfThread";
private static Deque<RunnableQueueItem> _stackItems = new ArrayDeque<>( );
private Map<String, Thread> _mapThreadByKey = new HashMap<>( );
private List<Thread> _listThread = new ArrayList<>( );
/**
* {@inheritDoc}
*/
@Override
public void run( )
{
int nMaxNumberThread = AppPropertiesService.getPropertyInt( PROPERTY_MAX_NUMBER_THREAD, 5 );
// We remove dead threads from running thread collections
String logs = removeDeadThreads( );
RunnableQueueItem item = null;
int nCurrentNumberRunningThreads = _mapThreadByKey.size( ) + _listThread.size( );
List<RunnableQueueItem> listLockedItems = new ArrayList<>( );
while ( ( nCurrentNumberRunningThreads < nMaxNumberThread ) && ( ( item = popItemFromQueue( ) ) != null ) )
{
String key = item.computeKey( );
// If the item has a key, then we must make sure that another thread with the same key and plugin is not running
if ( key != null )
{
Thread thread = _mapThreadByKey.get( key );
if ( thread != null )
{
if ( thread.isAlive( ) )
{
// The thread is already running. We declare this item as locked for this run of the daemon.
listLockedItems.add( item );
}
else
{
// Dead threads are removed from collections at the beginning of the run of the daemon
// We still check again that the thread is alive just in case it died during the run
thread = getThread( item );
_mapThreadByKey.put( key, thread );
// We do not increase the number of running threads, because we removed and add one
}
}
else
{
thread = getThread( item );
_mapThreadByKey.put( key, thread );
nCurrentNumberRunningThreads++;
}
}
else
{
// If it has no key, or if the plugin has not been set, we create a thread in the keyless collection
Thread thread = getThread( item );
_mapThreadByKey.put( key, thread );
nCurrentNumberRunningThreads++;
}
}
// We replace in the queue items that was locked
listLockedItems.stream( ).forEach( ThreadLauncherDaemon::addItemToQueue );
// If the maximum number of running threads has been reached, we end this run
if ( nCurrentNumberRunningThreads >= nMaxNumberThread )
{
setLastRunLogs( logs + "Every threads are running. Daemon execution ending." );
return;
}
setLastRunLogs( logs + "There is no more runnable to launch." );
}
private Thread getThread( RunnableQueueItem item )
{
Thread thread = new Thread( new RunnableWrapper( item.getRunnable( ) ) );
thread.start( );
return thread;
}
private String removeDeadThreads( )
{
List<String> listDeadThreadKeys = new ArrayList<>( );
for ( Entry<String, Thread> threadEntry : _mapThreadByKey.entrySet( ) )
{
if ( !threadEntry.getValue( ).isAlive( ) )
{
listDeadThreadKeys.add( threadEntry.getKey( ) );
}
}
for ( String strThreadKey : listDeadThreadKeys )
{
_mapThreadByKey.remove( strThreadKey );
}
List<Thread> listDeadThreads = new ArrayList<>( );
for ( Thread thread : _listThread )
{
if ( !thread.isAlive( ) )
{
listDeadThreads.add( thread );
}
}
String logs = "";
if ( !listDeadThreads.isEmpty( ) )
{
logs = "Releasing " + listDeadThreads.size( ) + " dead threads.\n";
}
for ( Thread thread : listDeadThreads )
{
_listThread.remove( thread );
}
return logs;
}
/**
* Add a runnable to the launch queue. It will be launched as soon as a thread is available.
*
* @param runnable
* The runnable to execute
* @param strKey
* The key of the runnable. Runnables of a given plugin are ensured that they will not be executed at the same time if they have the same key.
* @param plugin
* The plugin the runnable is associated with
*/
public static void addItemToQueue( Runnable runnable, String strKey, Plugin plugin )
{
RunnableQueueItem runnableItem = new RunnableQueueItem( runnable, strKey, plugin );
synchronized( ThreadLauncherDaemon.class )
{
_stackItems.addLast( runnableItem );
}
AppDaemonService.signalDaemon( THREAD_LAUNCHER_DAEMON_ID );
}
/**
* Add a runnable item to the queue
*
* @param runnableItem
* The runnable item to add to the queue
*/
private static synchronized void addItemToQueue( RunnableQueueItem runnableItem )
{
_stackItems.addLast( runnableItem );
}
/**
* Pop the first item of the queue. The item is removed from the queue.
*
* @return The first item of the queue, or null if the queue is empty
*/
private static synchronized RunnableQueueItem popItemFromQueue( )
{
if ( CollectionUtils.isEmpty( _stackItems ) )
{
return null;
}
return _stackItems.pop( );
}
/**
* Count the number of items in the queue.
*
* @return The current number of items in the queue
*/
public static synchronized Integer countItemsInQueue( )
{
return _stackItems.size( );
}
}