Converting threadgroup to executorservice

[John Vorwald]

The following code seems to run correctly. The goal is to start four threads on two threadgroups, and then cancel the threads after some calculations are finished, but before the calculation is completed.

I would like to change the code to limit the number of threads submitted at one time.

One approach to limiting the number of thread submitted at one time is to use executor service. But, when I try to convert to executor service, then the code doesn't produce any results.

How should the code be modified to use executor service?
Is there another way to limit the number of threads running at one time? The worker threads are computationally intensive, and would typically run for several minutes.

import java.util.ArrayList;
import java.util.List;


public class NextGenerationWorker extends Thread
{
    public static class MyThread extends StoppableWorker
    {
        boolean suspended;
        double score;
        List<Double> scores;

        MyThread(ThreadGroup tgOb, String threadname)
        {
            super(tgOb, threadname);
            System.out.println("New thread: " + this);
            suspended = false;
            score = -1e10;
            scores = new ArrayList<Double>();
            start(); // Start the thread
        }

        public void run()
        {
            try
            {
                final int minIter = 3;
                final int maxIter = 10;
                final int nIter = minIter + (int)(Math.random() * ((maxIter - minIter) + 1));
                final int minSleep =  5 * 1000;
                final int maxSleep = 30 * 1000;

                for (int i = 1; i<= nIter; i++)
                {
                    // System.out.println(getName() + ": " + i);
                    // pick random sleep time between 5 and 30 seconds
                    final int sleepTime = minSleep + (int)(Math.random() * ((maxSleep - minSleep) + 1));

                    Thread.sleep(sleepTime);
                    synchronized (this)
                    {
                        while (suspended)
                        {
                            wait();
                        }
                    }
                    if (this.shouldProcessStop()) {
                        System.out.printf("On iteration %d of %d, %s slept for %6.2f seconds, stop flag is set, exiting.\n",
                                          i, nIter, getName(), sleepTime/1000.);
                        return;
                    } else {
                        double minScore = 0;
                        double maxScore = 20;
                        double tmpScore = minScore + (Math.random() * (maxScore - minScore));
                        scores.add( tmpScore );
                        if ( scores.size() == 1) {
                            System.out.printf("On iteration %2d of %2d, %s slept for %6.2f seconds, score initalized to %7.2f.\n",
                                              i, nIter, getName(), sleepTime/1000., tmpScore);
                            score = tmpScore; 
                        } else if ( tmpScore > score ) {
                            System.out.printf("On iteration %2d of %2d, %s slept for %6.2f seconds, score %7.2f is updated to %7.2f.\n",
                                              i, nIter, getName(), sleepTime/1000., score, tmpScore);
                            score = tmpScore;
                        } else {
                            System.out.printf("On iteration %2d of %2d, %s slept for %6.2f seconds, score %7.2f is kept since it is greater than %7.2f.\n",
                                              i, nIter, getName(), sleepTime/1000., score, tmpScore);
                        }
                    }

                }
            }
            catch (Exception e)
            {
                System.out.printf("\n*** Exception %s in %s\n\n", 
                                  e.getMessage(), getName() );//Thread.currentThread().getStackTrace());
                e.printStackTrace();
                System.exit( 1 );
            }
            System.out.printf("%s exiting with score of %f\n\n", getName(), score );
        }

        public List<Double> getScores() {
            List<Double> rtn = new ArrayList<Double>();
            for (Double d:scores) {
                rtn.add( d );
            }
            return rtn;
        }

        void suspendMe()
        {
            suspended = true;
        }

        synchronized void resumeMe()
        {
            suspended = false;
            notify();
        }

        public double getScore()
        {
            return this.score;
        }
    }

    public void run()
    {
        /* 
         * How many threads? 
         */
        final int nProcessors = Runtime.getRuntime().availableProcessors();
        final int nThreadsAllowed = nProcessors; 
        System.out.printf( "The number of avaiable processors is %d, allowing %d threads\n", 
                           nProcessors, nThreadsAllowed );
        // first define two groups
        ThreadGroup groupA = new ThreadGroup("my group");
        ThreadGroup groupB = new ThreadGroup("your group");

        // then define some threads and assign them a group
        MyThread t1 = new MyThread(groupA, "t1");
        MyThread t2 = new MyThread(groupA, "t2");
        MyThread t3 = new MyThread(groupB, "t3");
        MyThread t4 = new MyThread(groupA, "t4");

        System.out.println("\nHere are the outputs from list():");
        System.out.println("Group A");
        groupA.list();
        System.out.println("Group B");
        groupB.list();

        System.out.println("Suspending Group A");
        Thread tga[] = new Thread[groupA.activeCount()];
        groupA.enumerate(tga); // get threads in group
        for (int i = 0; i < tga.length; i++)
        {
            ((MyThread) tga[i]).suspendMe(); // suspend each thread
        }

        try
        {
            Thread.sleep(1000);
        } catch (InterruptedException e)
        {
            System.out.println("Main thread interrupted.");
        }

        System.out.println("Resuming Group A");
        for (int i = 0; i < tga.length; i++)
        {
            ((MyThread) tga[i]).resumeMe();
        }

        final double nSecsToSleep = 50;
        System.out.printf( "Sleeping %5.1f seconds in main before cancelling\n", 
                           nSecsToSleep);
        try
        {
            Thread.sleep( (long) ( nSecsToSleep*1000 ) );
        } catch (InterruptedException e)
        {
            System.out.println("Main thread interrupted.");
        }

        final int minThreadQuit = 0;
        final int maxThreadQuit = tga.length-1;
        final int nThreadQuit = minThreadQuit + (int)(Math.random() * ((maxThreadQuit - minThreadQuit) + 1));
        System.out.printf("Calling stopProcess on thread %s\n", tga[nThreadQuit].getName());
        ((StoppableWorker) tga[nThreadQuit] ).stopProcesses();

        try
        {
            System.out.println("Waiting for threads to finish.");
            t1.join();
            t2.join();
            t3.join();
            t4.join();
        } catch (Exception e) {
            System.out.println("Exception in Main thread");
        }
        double max_score_t1 = t1.getScore();
        double max_score_t2 = t2.getScore();
        double max_score_t3 = t3.getScore();
        double max_score_t4 = t4.getScore();
        List<Double> scores_t1 = t1.getScores();
        List<Double> scores_t2 = t2.getScores();
        List<Double> scores_t3 = t3.getScores();
        List<Double> scores_t4 = t4.getScores();
        if (scores_t1.isEmpty()) {
            System.out.printf("  t1 was not assigned a score, so maximum score is initial value of %6.2f\n", max_score_t1);
        } else {
            System.out.printf("  t1 maximum score was %6.2f\n  Individual scores are:\n", max_score_t1);
            for (Double d:scores_t1) {
                System.out.printf( "    %6.2f\n", d.doubleValue() );
            }
        }
        if (scores_t2.isEmpty()) {
            System.out.printf("  t2 was not assigned a score, so maximum score is initial value of %6.2f\n", max_score_t2);
        } else {
            System.out.printf("  t2 maximum score was %6.2f\n  Individual scores are:\n", max_score_t2);
            for (Double d:scores_t2) {
                System.out.printf( "    %6.2f\n", d.doubleValue() );
            }
        }
        if (scores_t3.isEmpty()) {
            System.out.printf("  t3 was not assigned a score, so maximum score is initial value of %6.2f\n", max_score_t3);
        } else {
            System.out.printf("  t3 maximum score was %6.2f\n  Individual scores are:\n", max_score_t3);
            for (Double d:scores_t3) {
                System.out.printf( "    %6.2f\n", d.doubleValue() );
            }
        }
        if (scores_t4.isEmpty()) {
            System.out.printf("  t4 was not assigned a score, so maximum score is initial value of %6.2f\n", max_score_t4);
        } else {
            System.out.printf("  t4 maximum score was %6.2f, the individual scores are:\n", max_score_t4);
            for (Double d:scores_t4) {
                System.out.printf( "    %6.2f\n", d.doubleValue() );
            }
        }
        double final_score;
        final_score = Math.max( max_score_t1, max_score_t2 );
        final_score = Math.max( final_score, max_score_t3 );
        final_score = Math.max( final_score, max_score_t4 );
        System.out.printf("Absolute maximum score was %6.2f\n",final_score);
        System.out.printf("Main thread exiting\n");

    }

    public static void main(String[] args)
    {
        final Thread runme = new NextGenerationWorker();
        runme.start();
    }


}

abstract class StoppableWorker extends Thread
{
    /**
     * Flag indicating if thread processing should stop.
     * Variable is volatile so that all threads see most up to date value.
     */
    private static volatile boolean stopRunning;

    /**
     * @param aTgOb
     * @param aThreadname
     */
    public StoppableWorker( ThreadGroup aTgOb, String aThreadname )
    {
        super(aTgOb, aThreadname);
    }

    /**
     * @param aName
     */
    public StoppableWorker( String aName )
    {
        super(aName);
    }

    /**
     * Has a stop been requested?
     * 
     * @concurrency Returns a volatile flag that indicates if the processes should stop.
     */
    public boolean shouldProcessStop()
    {
        return stopRunning;
    }

    /**
     * Request that all processes stop.
     * 
     * @concurrency Returns a volatile flag that indicates if the processes should stop.
     */
    public void stopProcesses()
    {
        synchronized (this) {
            StoppableWorker.stopRunning = true;
            notify();
        }
    }
}

[John Vorwald]

Somedays I think I know what I'm doing, other days ...

I played with some example codes, and came up with the following implementation of an executor service that has a cancel flag, which is what the previous code was on.

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class NextGenerationExecutor extends Thread
{
    private final ExecutorService poolA;
    private final ExecutorService poolB;

    public NextGenerationExecutor() {
        /* 
         * How many threads? 
         */
        final int nProcessors = Runtime.getRuntime().availableProcessors();
        final int nThreadsAllowed = nProcessors; 
        System.out.printf( "The number of avaiable processors is %d, allowing %d threads\n", 
                           nProcessors, nThreadsAllowed );
        poolA = Executors.newFixedThreadPool(nThreadsAllowed);
        poolB = Executors.newFixedThreadPool(nThreadsAllowed);

    }

    public static class MyThread implements Callable<List<Double>>
    {
        boolean suspended;
        double score;
        List<Double> scores;
        private final String name;
        public static boolean stop;

        public String getName() { return name; }
        MyThread(String threadname)
        {
            name = threadname;
            System.out.println("New thread: " + this);
            suspended = false;
            score = -1e10;
            scores = new ArrayList<Double>();
            stop = false;
        }

        public void run()
        {
            try
            {
                final int minIter = 3;
                final int maxIter = 10;
                final int nIter = minIter + (int)(Math.random() * ((maxIter - minIter) + 1));
                final int minSleep =  5 * 1000;
                final int maxSleep = 30 * 1000;

                for (int i = 1; i<= nIter; i++)
                {
                    // System.out.println(getName() + ": " + i);
                    // pick random sleep time between 5 and 30 seconds
                    final int sleepTime = minSleep + (int)(Math.random() * ((maxSleep - minSleep) + 1));

                    Thread.sleep(sleepTime);
                    synchronized (this)
                    {
                        while (suspended)
                        {
                            wait();
                        }
                    }
                    if (this.shouldProcessStop()) {
                        System.out.printf("On iteration %d of %d, %s slept for %6.2f seconds, stop flag is set, exiting.\n",
                                          i, nIter, name, sleepTime/1000.);
                        return;
                    } else {
                        double minScore = 0;
                        double maxScore = 20;
                        double tmpScore = minScore + (Math.random() * (maxScore - minScore));
                        scores.add( tmpScore );
                        if ( scores.size() == 1) {
                            System.out.printf("On iteration %2d of %2d, %s slept for %6.2f seconds, score initalized to %7.2f.\n",
                                              i, nIter, name, sleepTime/1000., tmpScore);
                            score = tmpScore; 
                        } else if ( tmpScore > score ) {
                            System.out.printf("On iteration %2d of %2d, %s slept for %6.2f seconds, score %7.2f is updated to %7.2f.\n",
                                              i, nIter, name, sleepTime/1000., score, tmpScore);
                            score = tmpScore;
                        } else {
                            System.out.printf("On iteration %2d of %2d, %s slept for %6.2f seconds, score %7.2f is kept since it is greater than %7.2f.\n",
                                              i, nIter, name, sleepTime/1000., score, tmpScore);
                        }
                    }

                }
            }
            catch (Exception e)
            {
                System.out.printf("\n*** Exception %s in %s\n\n", 
                                  e.getMessage(), name );//Thread.currentThread().getStackTrace());
                e.printStackTrace();
                System.exit( 1 );
            }
            System.out.printf("%s exiting with score of %f\n\n", name, score );
        }

        /**
         */
        private boolean shouldProcessStop()
        {
            return worker.NextGenerationExecutor.MyThread.stop;
        }
        public List<Double> getScores() {
            List<Double> rtn = new ArrayList<Double>();
            for (Double d:scores) {
                rtn.add( d );
            }
            return rtn;
        }

        void suspendMe()
        {
            suspended = true;
        }

        synchronized void resumeMe()
        {
            suspended = false;
            notify();
        }

        /* (non-Javadoc)
         * @see java.util.concurrent.Callable#call()
         */
        @Override
        public List<Double> call()
        throws Exception
        {
            run();
            return this.scores;
        }

        public void stopProcesses() {
            synchronized(this) {
                worker.NextGenerationExecutor.MyThread.stop = true;
            }
        }
    }


    public void run() {

        // then define some threads and assign them a group
        MyThread t1 = new MyThread("t1");
        MyThread t2 = new MyThread("t2");
        MyThread t3 = new MyThread("t3");
        MyThread t4 = new MyThread("t4");

        List<MyThread> tasks_A = new ArrayList<MyThread>();
        tasks_A.add( t1 );
        tasks_A.add( t2 );
        tasks_A.add( t4 );
        
        List<MyThread> tasks_B = new ArrayList<MyThread>();
        tasks_B.add( t3 );

        List<Future<List<Double>>> results_A = new ArrayList<Future<List<Double>>>();
        List<Future<List<Double>>> results_B = new ArrayList<Future<List<Double>>>();
        
        
        for (MyThread task:tasks_A) {
            
            Future<List<Double>> fut = poolA.submit( task );
            results_A.add( fut );
        }
        for (MyThread task:tasks_B) {
            Future<List<Double>> fut = poolA.submit( task );
            results_B.add( fut );
        }
        
        
        System.out.println("Group A");
        for (Future<List<Double>> p:results_A) {
            System.out.printf("  %s\n",p.toString());     
        }
        System.out.println("Group B");
        for (Future<List<Double>> p:results_B) {
            System.out.printf("  %s\n",p.toString());            
        }


        System.out.println("Resuming Group A");

        final double nSecsToSleep = 50;
        System.out.printf( "Sleeping %5.1f seconds in main before cancelling\n", 
                           nSecsToSleep);
        try
        {
            Thread.sleep( (long) ( nSecsToSleep*1000 ) );
        } catch (InterruptedException e)
        {
            System.out.println("Main thread interrupted.");
        }

        final int minThreadQuit = 0;
        final int maxThreadQuit = tasks_A.size()-1;
        final int nThreadQuit = minThreadQuit + (int)(Math.random() * ((maxThreadQuit - minThreadQuit) + 1));
        System.out.printf("Calling stopProcess on thread %s\n", tasks_A.get( nThreadQuit ).name);
        tasks_A.get( nThreadQuit ).stopProcesses();

        try
        {
            System.out.println("Waiting for threads to finish.");
            poolA.shutdown();
            poolB.shutdown();
        } catch (Exception e) {
            System.out.println("Exception in Main thread");
        }

        double final_score = 0;
        boolean isFirst = true;
        System.out.println("Group A Results");
        for (Future<List<Double>> p:results_A) {
            List<Double> t=null;
            try
            {
                t = p.get();
            }
            catch ( InterruptedException e )
            {
                e.printStackTrace();
            }
            catch ( ExecutionException e )
            {
                e.printStackTrace();
            }
            int ic=0;
            for (Double tt:t ) {
                if (isFirst) {
                    isFirst = false;
                    final_score = tt.doubleValue();
                } else {
                    final_score = Math.max( final_score,tt.doubleValue() );
                }
                System.out.printf("  %d %6.2f\n", ++ic, tt.doubleValue());     
            }
        }
        System.out.println("Group B Results");
        for (Future<List<Double>> p:results_B) {
            List<Double> t = null;
            try
            {
                t = p.get();
            }
            catch ( InterruptedException e )
            {
                e.printStackTrace();
            }
            catch ( ExecutionException e )
            {
                e.printStackTrace();
            }
            int ic=0;
            for (Double tt:t ) {
                if (isFirst) {
                    isFirst = false;
                    final_score = tt.doubleValue();
                } else {
                    final_score = Math.max( final_score,tt.doubleValue() );
                }
                System.out.printf("  %d %6.2f\n", ++ic, tt.doubleValue());     
            }
        }
        System.out.printf("Absolute maximum score was %6.2f\n",final_score);
        System.out.printf("Main thread exiting\n");

    }

    public static void main(String[] args)
    {
        final Thread runme = new NextGenerationExecutor();
        runme.start();
    }

}