StaticTaskManager.java

Go to the documentation of this file.

/*
 *   DENOPTIM
 *   Copyright (C) 2022 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.util.HashMap;
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 javax.swing.JProgressBar;
 
import denoptim.programs.combinatorial.FragSpaceExplorer;
import denoptim.programs.denovo.GARunner;
 
public class StaticTaskManager
{
    private static final StaticTaskManager instance = new StaticTaskManager();
    
    private static ThreadPoolExecutor tpe; 
 
    private static Map<Task,Future<?>> subToFutureMap = 
            new HashMap<Task,Future<?>>();
    
    private static final int maxParallelPrograms = 4;
    
    private static final int queueSize = 10;
    
    public static int valProgressBar = 1; //On startup the bar is filled.up.
    
    public static int maxProgressBar = 1;
 
    public static JProgressBar queueStatusBar = null;
    
    public static final String PROGRESSBARUPDATE = "UPDATE-PROGRESS-BAR";
 
    private static final Object LOCK = new Object();
    
//------------------------------------------------------------------------------
    
    private StaticTaskManager()
    {
        tpe = new ThreadPoolExecutor(maxParallelPrograms, maxParallelPrograms, 
                Long.MAX_VALUE, 
                TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(queueSize));
        
        // What to do in case of filled-up resources and queue rejecting tasks
        tpe.setRejectedExecutionHandler(new RejectedExecutionHandler()
        {
            @Override
            public void rejectedExecution(Runnable r,
                    ThreadPoolExecutor executor)
            {
                try
                {
                    // re-submit to the queue any rejected job
                    executor.getQueue().put(r);
                }
                catch (InterruptedException ex)
                {
                    ex.printStackTrace();
                }
            }
        });
        
        // What to do in case of shutdown signal
        Runtime.getRuntime().addShutdownHook(new ShutDownHook());
 
        tpe.prestartAllCoreThreads();
    }
    
//------------------------------------------------------------------------------
    
    private class ShutDownHook extends Thread
    {
        @Override
        public void run()
        {
            tpe.shutdown();
            try
            {
                if (!tpe.awaitTermination(1, TimeUnit.SECONDS))
                {
                    tpe.shutdownNow(); // Cancel running asks
                }
            }
            catch (InterruptedException ie)
            {
                stopAll();
                subToFutureMap.clear();
                tpe.purge();
                tpe.getQueue().clear();
                tpe.shutdownNow();
                
                // and stop possibly alive thread
                //TODO is this needed? It should not
                Thread.currentThread().interrupt();
            }
        }
    }
    
//------------------------------------------------------------------------------
    
    public static StaticTaskManager getInstance()
    {
        return instance;
    }
 
//------------------------------------------------------------------------------
    
    public static void addTodoTask()
    {
        addTodoTasks(1);
    }
    
//------------------------------------------------------------------------------
    
    public static void addTodoTasks(int addedTasksCount)
    {
        synchronized (LOCK) {
            int max = maxProgressBar;
            int val = valProgressBar;
            if (max==1 && val==1)
            {
                // NB: the progress bar is initialized to max=1/val=1
                // Thus it looks "filled-up", i.e., no pending/running task
                maxProgressBar = addedTasksCount;
                valProgressBar = 0;
            } else {
                maxProgressBar = max+addedTasksCount;
                valProgressBar = val;
            }
        }
        if (queueStatusBar!=null)
        {
            queueStatusBar.setMaximum(maxProgressBar);
            queueStatusBar.setValue(valProgressBar);
            queueStatusBar.repaint();
        }
    }
 
//------------------------------------------------------------------------------
    
    public static void subtractDoneTask()
    {
        subtractDoneTasks(1);
    }
    
//------------------------------------------------------------------------------
    
    public static void subtractDoneTasks(int doneTasksCount)
    {
        synchronized (LOCK) {
            int val = valProgressBar;
            valProgressBar = val+doneTasksCount;
            if (valProgressBar == maxProgressBar)
            {
                maxProgressBar = 1;
                valProgressBar = 1;
            }
        }
        if (queueStatusBar!=null)
        {
            queueStatusBar.setMaximum(maxProgressBar);
            queueStatusBar.setValue(valProgressBar);
            queueStatusBar.repaint();
        }
    }
    
//------------------------------------------------------------------------------
    
    public static String getQueueSnapshot()
    {
        String s = "Approximate queue status:<ul>"
                + "<li>Running tasks = " + tpe.getActiveCount() + "/"
                + tpe.getPoolSize() + "</li>"
                + "<li>Queue Size    = " + tpe.getQueue().size()+ "</li></ul>";
        return s;
    }
    
//------------------------------------------------------------------------------
    
    public static void submitAndWait(Task task) throws InterruptedException, 
    ExecutionException
    {
        tpe.submit(task).get();
    }
    
//------------------------------------------------------------------------------
    
    public static void submit(Task task)
    {
        task.setNotify(true);
        addTodoTask();
        Future<?> future = tpe.submit(task);
        subToFutureMap.put(task,future);
    }
    
//------------------------------------------------------------------------------
    
    public static void stop(Task task)
    {
        task.stopTask();
        if (subToFutureMap.containsKey(task))
        {
            subToFutureMap.get(task).cancel(true);
            subToFutureMap.remove(task);
        }
    }
    
//------------------------------------------------------------------------------
    
    public static void stopAll() 
    {
        for (Task task : subToFutureMap.keySet())
        {
            Future<?> expectation = subToFutureMap.get(task);
            expectation.cancel(true);
            task.stopTask();
        }
    }
    
//------------------------------------------------------------------------------
 
    public static boolean hasActive()
    {
        return tpe.getActiveCount() > 0;
    }
 
//------------------------------------------------------------------------------
 
    public static void setLinkToProgressBar(JProgressBar bar)
    {
        queueStatusBar = bar;
    }
 
//------------------------------------------------------------------------------
    
}