Tracking Executing Threads

Question

I am trying to figure out how I can track all the threads that my application is spawning. Initially, I thought I had it figured out using a CyclicBarrier, however I am seeing threads executing after my await call.

Below is the working pseudo code:

public class ThreadTesterRunner {

    public static void main(String[] args) throws InterruptedException {

        final CyclicBarrier cb = new CyclicBarrier(1);
        ThreadRunner tr = new ThreadRunner(cb);
        Thread t = new Thread(tr, "Thread Runner");
        t.start();

        boolean process = true;
        // wait until all threads process, then print reports
        while (process){
            if(tr.getIsFinished()){
                System.out.println("Print metrics");
                process = false;
            }
            Thread.sleep(1000);
        }
    }
}


class ThreadRunner implements Runnable {
    static int timeOutTime = 2;
    private ExecutorService executorService = Executors.newFixedThreadPool(10);
    private final CyclicBarrier barrier;
    private boolean isFinished=false;

    public ThreadRunner(CyclicBarrier cb) {
        this.barrier = cb;
    }

    public void run(){
        try {
            boolean stillLoop = true; int i = 0;
            while (stillLoop){
                int size;
                Future<Integer> future = null;
                try {
                    future = executorService.submit(new Reader()); // sleeps
                    size = future.get();
                } catch (InterruptedException | ExecutionException ex) {
                    // handle Errs
                }

                if(i == 3){
                    stillLoop = false;
                    this.barrier.await();
                    this.isFinished=true;
                }
                //System.out.println("i = "+i+"  Size is: "+size+"\r");
                i++;
            }
        } catch (InterruptedException | BrokenBarrierException e1) {
            e1.printStackTrace();
        }
    }

    public boolean getIsFinished(){
        return this.isFinished;
    }
}

class Reader implements Callable {
    private ExecutorService executorService = Executors.newFixedThreadPool(1);

    @Override
    public Object call() throws Exception {
        System.out.println("Reading...");
        Thread.sleep(2000);
        executorService.submit(new Writer());
        return 1000;
    }
}

class Writer implements Callable {
    @Override
    public Void call() throws Exception {
        Thread.sleep(4000);
        System.out.println("Wrote");    
        return null;
    }
}

Can anyone suggest a way to ONLY print "print metrics" after all threads have run?

Answer 1

It doesn't seem like you're doing anything to coordinate with your Reader and Writer threads, which are the ones you want to wait for. If you pass your synchronization barrier through to those threads so that they can register and signal when they are done, it works just fine.

Here's a version rewritten to do so, using a Phaser instead of a CyclicBarrier. Note that each Reader and Writer registers itself upon construction, and notifies the synchronization barrier when it is done executing:

public class ThreadTesterRunner {
    public static void main(String[] args) throws InterruptedException {
        final Phaser cb = new Phaser();
        ThreadRunner tr = new ThreadRunner(cb);
        Thread t = new Thread(tr, "Thread Runner");
        t.start();

        boolean process = true;
        // wait until all threads process, then print reports
        while (process){
            if(tr.getIsFinished()){
                System.out.println("Print metrics");
                process = false;
            }
            //else {
            //  System.out.println("Waiting:  registered=" + cb.getRegisteredParties() + ", arrived=" + cb.getArrivedParties() + ", unarrived=" + cb.getUnarrivedParties());
            //}
            Thread.sleep(1000);
        }
    }
}


class ThreadRunner implements Runnable {
    static int timeOutTime = 2;
    private ExecutorService executorService = Executors.newFixedThreadPool(10);
    private final Phaser barrier;
    private boolean isFinished=false;

    public ThreadRunner(Phaser phaser) {
        this.barrier = phaser;
    }

    public void run(){
        try {
            boolean stillLoop = true; int i = 0;
            while (stillLoop){
                int size;
                Future<Integer> future = null;
                try {
                    future = executorService.submit(new Reader(this.barrier)); // sleeps
                    size = future.get();
                } catch (InterruptedException | ExecutionException ex) {
                    // handle Errs
                }

                if(i == 3){
                    stillLoop = false;
                    this.barrier.awaitAdvance(0);
                    this.isFinished=true;
                }
                //System.out.println("i = "+i+"  Size is: "+size+"\r");
                i++;
            }
        } catch (Exception e1) {
            e1.printStackTrace();
        }
    }

    public boolean getIsFinished(){
        return this.isFinished;
    }
}

class Reader implements Callable {
    private Phaser barrier;
    private ExecutorService executorService = Executors.newFixedThreadPool(1);

    public Reader(Phaser phase) {
        phase.register();
        this.barrier = phase;
    }

    @Override
    public Object call() throws Exception {
        System.out.println("Reading...");
        Thread.sleep(2000);
        executorService.submit(new Writer(this.barrier));
        this.barrier.arrive();
        return 1000;
    }
}

class Writer implements Callable {
    private Phaser barrier;

    public Writer(Phaser phase) {
        phase.register();
        this.barrier = phase;
    }

    @Override
    public Void call() throws Exception {
        Thread.sleep(4000);
        System.out.println("Wrote");
        this.barrier.arrive();
        return null;
    }
}

Answer 2

From what I can see you aren't waiting for the Writer to finish in the Reader. Is that the problem you are seeing?

You are also accessing isFinished from more than one thread without synchronization (which however, merely may delay the termination of the loop in this situation).

I don't see CyclicBarrier doing anything.

Not sure what you are trying to do, but I'd think about how simpler I can make it. For example, can Reader and Writer be combined into one task? Then, waiting for them to finish would merely be:

executorService.invokeAll(tasks);
System.out.println("Print metrics");

where tasks is a collection of tasks (see also this javadoc)