ParallelAsynchronousTaskExecutor.java

Go to the documentation of this file.

/*
 *   DENOPTIM
 *   Copyright (C) 2019 Marco Foscato <marco.foscato@uib.no>
 * 
 *   This program is free software: you can redistribute it and/or modify
 *   it under the terms of the GNU Affero General Public License as published
 *   by the Free Software Foundation, either version 3 of the License, or
 *   (at your option) any later version.
 *
 *   This program is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU Affero General Public License for more details.
 *
 *   You should have received a copy of the GNU Affero General Public License
 *   along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
package denoptim.task;
 
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
 
import org.apache.commons.lang3.time.StopWatch;
 
import denoptim.constants.DENOPTIMConstants;
import denoptim.exception.DENOPTIMException;
 
 
abstract public class ParallelAsynchronousTaskExecutor
{
    private final Map<Task,Future<Object>> futures;
 
    private final List<Task> submitted;
    
    protected final List<Object> results;
 
    final ThreadPoolExecutor tpe;
 
    private Throwable thrownByTask;
    
    private Logger logger;
    
    private int numThreads = 1;
 
    
//-----------------------------------------------------------------------------
 
    public ParallelAsynchronousTaskExecutor(int numberOfTasks, Logger logger)
    {
        this.logger = logger;
        futures = new HashMap<Task,Future<Object>>();
        submitted = new ArrayList<>();
        results = new ArrayList<>();
        numThreads = numberOfTasks;
 
        tpe = new ThreadPoolExecutor(numThreads,
                numThreads,
                Long.MAX_VALUE,
                TimeUnit.NANOSECONDS,
                new ArrayBlockingQueue<Runnable>(1));
 
        Runtime.getRuntime().addShutdownHook(new Thread()
        {
            @Override
            public void run()
            {
                tpe.shutdown(); // Disable new tasks from being submitted
                try
                {
                    // Wait a while for existing tasks to terminate
                    if (!tpe.awaitTermination(30, TimeUnit.SECONDS))
                    {
                        tpe.shutdownNow(); // Cancel currently executing tasks
                    }
 
                    if (!tpe.awaitTermination(60, TimeUnit.SECONDS))
                    {
                        // pool didn't terminate after the second try
                    }
                }
                catch (InterruptedException ie)
                {
                    cleanupCompleted();
                    cleanup();
                    // (Re-)Cancel if current thread also interrupted
                    tpe.shutdownNow();
                    // Preserve interrupt status
                    Thread.currentThread().interrupt();
                }
            }
        });
 
        // by default the ThreadPoolExecutor will throw an exception
        tpe.setRejectedExecutionHandler(new RejectedExecutionHandler()
        {
            @Override
            public void rejectedExecution(Runnable r, 
                    ThreadPoolExecutor executor)
            {
                try
                {
                    // this will block if the queue is full
                    executor.getQueue().put(r);
                }
                catch (InterruptedException ex)
                {
                    ex.printStackTrace();
                    String msg = "EXCEPTION in rejectedExecution.";
                    logger.log(Level.WARNING,msg);
                }
            }
        });
    }
 
//------------------------------------------------------------------------------
 
    public void stopRun()
    {
        cleanupCompleted();
        cleanup();
        tpe.shutdown();
    }
 
//------------------------------------------------------------------------------
 
    protected boolean subtaskHasException()
    {
        boolean hasException = false;
        for (Task tsk : submitted)
        {
            if (tsk.foundException())
            {
                hasException = true;
                thrownByTask = tsk.getException();
                break;
            }
        }
        return hasException;
    }
    
//------------------------------------------------------------------------------
    
    protected Throwable getExceptionFromSubTask()
    {
        return thrownByTask;
    }
 
//------------------------------------------------------------------------------
 
    protected boolean allTasksCompleted()
    {
        boolean allDone = true;
        for (Task tsk : submitted)
        {
            if (!tsk.isCompleted())
            {
                allDone = false;
                break;
            }
        }
        return allDone;
    }
 
//------------------------------------------------------------------------------
 
    public void run() throws DENOPTIMException, IOException
    {
        String msg = "";
        StopWatch watch = new StopWatch();
        watch.start();
        
        if (!doPreFlightOperations())
            return;
        
        // Start parallel threads for generating and/or manipulating fragments
        tpe.prestartAllCoreThreads();
        
        createAndSubmitTasks();
        
        // This sounds weird when reading the doc, but the following does wait
        // for the threads to complete.
        // TODO try ExecutorCompletionService
        
        // shutdown thread pool
        tpe.shutdown();
        try
        {
            // wait a bit for pending tasks to finish
            while (!tpe.awaitTermination(5, TimeUnit.SECONDS))
            {
                // do nothing
            }
        }
        catch (InterruptedException ex)
        {
            //Do nothing
        }
        
        // This collects also the results and frees-up memory
        cleanupCompleted();
        
        if (!doPostFlightOperations())
            return;
        
        // closing messages
        watch.stop();
        msg = "Overall time: " + watch.toString() + ". " 
            + DENOPTIMConstants.EOL
            + this.getClass().getSimpleName() + " run completed." 
            + DENOPTIMConstants.EOL;
        logger.log(Level.INFO, msg);
    }
 
//------------------------------------------------------------------------------
    
    protected void submitTask(Task task, String logFilePathname)
    {
        submitted.add(task);
        futures.put(task, tpe.submit(task));
        String msg = task.getClass().getSimpleName() + " "
                + task.getId() + " submitted.";
        if (logFilePathname!=null && !logFilePathname.isBlank())
        {
            msg = msg + " Log file: " + logFilePathname;
        }
        logger.log(Level.INFO, msg);
        if (submitted.size()>numThreads*2)
        {
            cleanupCompleted();
        }
    }
    
//------------------------------------------------------------------------------
 
    abstract protected void createAndSubmitTasks();
 
//------------------------------------------------------------------------------
    
    abstract protected boolean doPostFlightOperations();
    
//------------------------------------------------------------------------------
 
    abstract protected boolean doPreFlightOperations();
 
//------------------------------------------------------------------------------
 
    private void cleanupCompleted()
    {
        List<Task> completed = new ArrayList<Task>();
 
        for (Task t : submitted)
        {
            if (t.isCompleted())
                completed.add(t);
        }
 
        for (Task t : completed)
        {
            try
            {
                results.add(futures.get(t).get());
            } catch (InterruptedException | ExecutionException e)
            {
                e.printStackTrace();
                String msg = "EXCEPTION upon retrieving results from completed "
                        + "task. The result from task '" + t.toString() + "' "
                        + "are lost.";
                logger.log(Level.WARNING,msg);
            }
            submitted.remove(t);
            futures.get(t).cancel(true);
            futures.remove(t);
        }
    }
    
//------------------------------------------------------------------------------
    
    private void cleanup()
    {
        for (Task tsk : submitted)
        {
            tsk.stopTask();
        }
        for (Task tsk : futures.keySet())
        {
            futures.get(tsk).cancel(true);
            futures.remove(tsk);
        }
        submitted.clear();
        tpe.getQueue().clear();
    }
 
//------------------------------------------------------------------------------    
 
}