CyclicBarrier in java : Concurrency

The Java concurrency API provides a synchronizing utility that allows the synchronization of two or more threads in a determined point. It's the CyclicBarrier class. This class is similar to the CountDownLatch class but presents some differences that make them a more powerful class.
There are many situations in concurrent programming where threads may need to wait at a predefined execution point until all other threads reach that point. CyclicBarrier helps provide such a synchronization point. The barrier is called cyclic because it can be re-used after the waiting threads are released.

Easy definition to remember CyclicBarrier
" If you and your friends are going to a Road Trip, and you need to first meet at some common/fixed point from where you all will start your road journey. "

The CyclicBarrier class is initialized with an integer number, which is the number of threads that will be synchronized in a determined point. When one of those threads arrives to the determined point, it calls the await() method to wait for the other threads. When the thread calls that method, the CyclicBarrier class blocks the thread that is sleeping until the other threads arrive. When the last thread calls the await() method of the CyclicBarrier class, it wakes up all the threads that were waiting and continues with its job.

In general,A barrier is used to make a group of threads meet at a barrier point. A thread from a group arriving at the barrier waits until all threads in that group arrive. Once the last thread from the group arrives at the barrier, all threads in the group are released. You can use a barrier when you have a task that can be divided into subtasks; each subtask can be performed in a separate thread and each thread must meet at a common point to combine their results. Following Figure depict how a barrier synchronizer lets a group of three threads meet at the barrier point and lets them proceed.

One interesting advantage of the CyclicBarrier class is that you can pass an additional Runnable object as an initialization parameter, and the CyclicBarrier class executes this object as a thread when all the threads have arrived to the common point. This characteristic makes this class adequate for the parallelization of tasks using the divide and conquer programming technique.

This is some sort similar to CountDownLatch. Cyclic barrier is re-used where as CountDownLatch not. We'll see in the following example before going further.

CyclicBarrier methods:

• CyclicBarrier(int parties,Runnable barrierAction) : Creates a CyclicBarrier object with the number of threads waiting on it specified. Throws IllegalArgumentException if numThreads is negative or zero.
• int await() : Blocks until the specified number of threads have called await() on this barrier. The method returns the arrival index of this thread. This method can throw an InterruptedException if the thread is interrupted while waiting for other threads or a BrokenBarrierException if the barrier was broken for some reason

Example 1

Suppose you organized a racing event and in this event exactly 4 cars are required to start a start. So until all cars will come to the race track, race will not start. We simulate this scenario with the help of cyclic barrier.

Sample Output 
4 lanes are created, race will startas soon as all player arrives

Player Ferrai is ready to race
Player Ford is ready to race
Player Skoda is ready to race
Player BMW is ready to race
All player are ready to race
Player BMW k1200S is ready to race
Player Ducati 1098s is ready to race
Player Aprilla RSV 1000R is ready to race
Player Yamaha YZF R1 is ready to race
All player are ready to race


In the main() method you create a CyclicBarrier object. The constructor takes two arguments: the number of threads to wait for, and the thread to invoke when all the threads reach the barrier.
In this case, you have four players to wait for, so you create four threads, with each thread representing a player. The second argument for the CyclicBarrier constructor is the Race object since this thread represents the game, which will start once all four players are ready.
Inside the run() method for each Player thread, you call the await() method on the CyclicBarrier object. Once the number of awaiting threads for the CyclicBarrier object reaches four, the run() method in Race is called

Example 2
Suppose you have split a large job into a n * m tasks, distributed over n threads. m corresponds to a matrix row and you have a total to compute for each row. In that case, threads must be synchronized after each task ending so that the total for the row is compute. In that case, a CyclicBarrier initialized with the number of threads n is used to wait for the end of each row computation (m times in fact).

Sample Output
sum of 1 row: 22
sum of 2 row: 46
sum of 0 row: 10
sum of 3 row: 46
Sum of rows of matrix: 124

The CyclicBarrier uses an all-or-none breakage model for failed synchronization attempts: If a thread leaves a barrier point prematurely because of interruption, failure, or timeout, all other threads waiting at that barrier point will also leave abnormally via BrokenBarrierException (or InterruptedException if they too were interrupted at about the same time).

There's more...
The CyclicBarrier class has another version of the await() method:

• await(long time, TimeUnit unit): The thread will be sleeping until it's interrupted; the internal counter of CyclicBarrier arrives to 0 or specified time passes. 

This class also provides the getNumberWaiting() method that returns the number of threads that are blocked in the await() method, and the getParties() method that returns the number of tasks that are going to be synchronized with CyclicBarrier.

Resetting a CyclicBarrier object
The CyclicBarrier class has some points in common with the CountDownLatch class, but they also have some differences. One of the most important differences is that a CyclicBarrier object can be reset to its initial state, assigning to its internal counter the value with which it was initialized.

This reset operation can be done using the reset() method of the CyclicBarrier class. When this occurs, all the threads that were waiting in the await() method receive a brokenBarrierException exception. This exception was processed in the example presented in this recipe by printing the stack trace, but in a more complex application, it could perform some other operation, such as restarting their execution or recovering their operation at the point it was interrupted.

Broken CyclicBarrier objects
A CyclicBarrier object can be in a special state denoted by broken. When there are various threads waiting in the await() method and one of them is interrupted, this thread receives an InterruptedException exception, but the other threads that were waiting receive a BrokenBarrierException exception and CyclicBarrier is placed in the broken state.
The CyclicBarrier class provides the isBroken() method, then returns true if the object is in the broken state; otherwise it returns false.



Related Post
Semaphore vs CountDownLatch in Java
How thread exchange data with Exchanger class
Daemon Thread in Java

If you know anyone who has started learning Java, why not help them out! Just share this post with them. Thanks for studying today!...

Semaphore in Java : Concurrency

In this post we'll see one of the synchronizers that is used in  java.util.concurrent library i.e, Semaphore. We'll see a simplest and easiest way to use Semaphore so that it is easy to remember and understand easily.

What is Semaphore?

A semaphore controls access to shared resources. A semaphore maintains a counter to specify the number of resources that the semaphore controls.
Access to the resource is allowed if the counter is greater than zero,while a zero value of the counter indicates that no resource is available at the moment and so the access is denied. Semaphore: n-member access to a resource






A semaphore acts as a limiter of available resource pool depth; for example, a semaphore with a capacity of 10 allows a maximum of 10 threads to acquire it at once, and any further threads that attempt to acquire it will block until one of the other threads releases it.

This is somewhat different from ordinary mutual exclusion or monitor locking, which is typically used to prevent more than one thread from simultaneously modifying the same variables and causing inconsistent results or program state.

Synchronization vs Semaphore
Synchronized allows only one thread of execution to access the resource at the same time. Semaphore allows up to n threads of execution to access the resource at the same time.


Basically two method that used are for acquiring and releasing resources from a semaphore

• acquire() : decrement the value
• release() : increment the value

If a thread calls acquire() and the counter is zero (i.e., resources are unavailable), the thread waits until the counter is non-zero and then gets the resource for use. Once the thread is done using the resource, it calls release() to increment the resource availability counter.

Note if the number of resources is 1, then at a given time only one thread can access the resource; in this case, using the semaphore is similar to using a lock

A semaphore initialized to one, and which is used such that it only has at most one permit available, can serve as a mutual exclusion lock.This is more commonly known as a binary semaphore, because it only has two states.
Semaphores are often used to restrict the number of threads than can access some (physical or logical) resource.

• Semaphore(int permits) : This method create Semaphore objects with a given number of permits, here permits means the number of threads that can access the resource at a time.
• void acquire() : Acquires a permit if available; otherwise, it blocks until a permit becomes available
• void release() : Releases a permit from the semaphore basically decrement the value

Example First we see example of binary semaphore.To show how to use it, we are going to implement a print queue that can be used by concurrent tasks to print their jobs. This print queue will be protected by a binary semaphore, so only one thread can print at a time.

Output

How it works...
The key to this example is in the printJob() method of the PrintQueue class. This method shows the three steps you must follow when you use a semaphore to implement a critical section, and protect the access to a shared resource:

• First, you acquire the semaphore, with the acquire() method.
• Then, you do the necessary operations with the shared resource.
•  Finally, release the semaphore with the release() method.

Another important point in this example is the constructor of the PrintQueue class and the initialization of the Semaphore object. You pass the value 1 as the parameter of this constructor, so you are creating a binary semaphore. The initial value of the internal counter is 1, so you will protect the access to one shared resource, in this case, the print queue.

The Semaphore class has two additional versions of the acquire() method:

• acquireUninterruptibly(): The acquire() method; when the internal counter of the semaphore is 0, blocks the thread until the semaphore is released. During this blocked time, the thread may be interrupted and then this method throws an InterruptedException exception. This version of the acquire operation ignores the interruption of the thread and doesn't throw any exceptions.
• tryAcquire(): This method tries to acquire the semaphore. If it can, the method returns the true value. But if it can't, the method returns the false value instead of being blocked and waits for the release of the semaphore. It's your responsibility to take the correct action based on the return value.

Fairness in semaphores
The concept of fairness is used by the Java language in all classes that can have various threads blocked waiting for the release of a synchronization resource (for example, a semaphore). The default mode is called the non-fair mode. In this mode, when the synchronization resource is released, one of the waiting threads is selected to get this resource, but it's selected without any criteria. The fair mode changes this behavior and forces to select the thread that has been waiting for more time.
As occurs with other classes, the Semaphore class admits a second parameter in its constructor. This parameter must take a Boolean value. If you give it the false value, you are creating a semaphore that will work in non-fair mode. You will get the same behavior if you don't use this parameter. If you give it the true value, you are creating a semaphore that will work in fair mode.


Controlling Concurrent access to multiple copies of a resource 
In that recipe, you implemented an example using binary semaphores. These kinds of semaphores are used to protect the access to one shared resource, or to a critical section that can only be executed by one thread at a time. But semaphores can also be used when you need to protect various copies of a resource, or when you have a critical section that can be executed by more than one thread at the same time.
In this Example, you will learn how to use a semaphore to protect more than one copy of a resource. You are going to implement an example, which has one print queue that can print documents in three different printers.

Output

How it works...
The key of this example is in the PrintQueue class. The Semaphore object is created using 3 as the parameter of the constructor. The first three threads that call the acquire() method will get the access to the critical section of this example, while the rest will be blocked. When a thread finishes the critical section and releases the semaphore, another thread will acquire it.
In this critical section, the thread gets the index of the printer assigned to print this job. This part of the example is used to give more realism to the example, but it doesn't use any code related with semaphores.

The acquire(), acquireUninterruptibly(), tryAcquire(), and release() methods have an additional version which has an int parameter. This parameter represents the number of permits that the thread that uses them wants to acquire or release, so as to say, the number of units that this thread wants to delete or to add to the internal counter of the semaphore. In the case of the acquire(), acquireUninterruptibly(), and tryAcquire() methods, if the value of this counter is less this value, the thread will be blocked until the counter gets this value or a greater one.




If you know anyone who has started learning Java, why not help them out! Just share this post with them. Thanks for studying today!...