Cyclic Barrier in java

Question

I have a list which needs to be populated by three parties(threads,lets say).I am using cyclic barrier to achieve this functionality. Everything works fine except that I am not able to use the resulted list without inducing a forced sleep. Below is the code :

public class Test{

List<Integer> item = new Vector<Integer>();

public void returnTheList(){
       CyclicBarrier cb = new CyclicBarrier(3, new Runnable() {

                @Override
                public void run() {

                    System.out.println("All parties are arrived at barrier, lets play -- : " + CyclicBarrierTest.getTheList().size());
                    //Here I am able to access my resulted list

                }
            });


            CyclicBarrierTest sw1 = new CyclicBarrierTest(cb, new ZetaCode(1500), s);
            CyclicBarrierTest sw2 = new CyclicBarrierTest(cb, new ZetaCode(1500),s);
            CyclicBarrierTest sw3 = new CyclicBarrierTest(cb, new ZetaCode(1500),s);
            Thread th1 = new Thread(sw1, "ZetaCode1");
            Thread th2 = new Thread(sw2, "ZetaCode2");
            Thread th3 = new Thread(sw3, "ZetaCode3");
            th1.start();
            th2.start();
            th3.start();

    }

public static void main(String args[]){
    System.out.println("asdfasd");
    Test test = new Test();
    //ActionClass ac = new ActionClass();
    test.returnTheList();
    System.out.println("Inside the main method...size of the final list : " +test.item.size() );
}

Below is my CyclicBrrierTest class :

public class CyclicBarrierTest implements Runnable{

private CyclicBarrier barrier;
private Object obj;
 static volatile String s = "";
 volatile List<Integer> finalIntList = new Vector<Integer>();

public CyclicBarrierTest(CyclicBarrier barrier, Object obj, String s){
    this.barrier = barrier;
    this.obj = obj;
}

@Override
public void run(){
    try{
        System.out.println(Thread.currentThread().getName() + " is waiting on barrier and s is now  : " + finalIntList.size());
        ZetaCode simple = (ZetaCode)obj;

        finalIntList.addAll(simple.getTheItemList());
        barrier.await();

        System.out.println(Thread.currentThread().getName() + " has crossed the barrier");

    }catch(InterruptedException ex){
        System.out.println("Error.." + ex.getMessage());

    }catch(Exception e){
        System.out.println("Error.." + e.getMessage());
    }
}
    public  List<Integer> getTheList(){
    return finalIntList;
}

So if I run this code without giving any delay the print statement in my main method gives me the length of my list as zero,however after giving an appropriate sleep it gives me the expected output.I want to achieve the same without giving any delay.Any help would be appreciated. Thanks in advance.

Answer 1

It seems you'd want to use a CountDownLatch, not a CyclicBarrier here. The CyclicBarrier is working exactly as intended - your main method just isn't waiting for it to be tripped by all 3 threads. When you give it a sleep statement, the other 3 threads just happen to finish before main wakes up again.

A CyclicBarrier is useful when you need N workers to all reach the same 'checkpoint' before proceeding, and the workers themselves are the only ones who care. However, you have an N + 1 user here, the main thread, who wants to know when they're all done, and CyclicBarrier doesn't support that use case.

Note, of course that you can also use both of them.

Answer 2

In this code we have 4 tasks . Task1, Task2, Task3 producing int values and Task4 will add all the int values . Task4 is waiting after calling await() for Task1, Task2, Task3 to produce values.When they produce values they call await() method and Task 4 will add their values and print the o/p and call reset() method so the barrier will reset. After reset this process will continue again

package practice;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicbarrierExample {

    public static void main(String[] args) {

        CyclicBarrier c = new CyclicBarrier(4);
        Task1 t1 = new Task1(c);
        Task2 t2 = new Task2(c);
        Task3 t3 = new Task3(c);
        Task4 t4 = new Task4(c);
        t1.start();
        t2.start();
        t3.start();
        t4.start();
    }
 }


class Task1 extends Thread {
    CyclicBarrier c;
    static int t1 ;

    public Task1(CyclicBarrier c) {
        this.c = c;
    }

    @Override
    public void run() {
        while (true) {
            t1 = t1 + 1;
            try {
                c.await();
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }

}

class Task2 extends Thread {
    CyclicBarrier c;

    static int t2;

    public Task2(CyclicBarrier c) {
        this.c = c;
    }

    @Override
    public void run() {
        while (true) {
            t2 = t2 + 1;
            try {
                c.await();
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }

}

class Task3 extends Thread {
    CyclicBarrier c;
    static int t3;

    public Task3(CyclicBarrier c) {
        this.c = c;
    }

    @Override
    public void run() {
        while (true) {
            t3 = t3 + 1;
            try {
                c.await();
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }

}

class Task4 extends Thread {
    CyclicBarrier c;
     static int t4;
    static int count=0;

    public Task4(CyclicBarrier c) {
        this.c = c;
    }

    @Override
    public void run() {
        while (count<10) {
            try {
                c.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
            t4 = Task1.t1 + Task2.t2 + Task3.t3;
            System.out.println(t4);
            try {
                c.reset();
            } catch (Exception e) {
                System.out.println("yo");
            }
            count++;

        }
    }

}