Alex Miller lives in St. Louis. He writes code for a living and currently work for Terracotta Tech on the Terracotta open-source Java clustering product. Prior to Terracotta he worked at BEA Systems and was Chief Architect at MetaMatrix. His main language for the last decade has been Java, although Alex have been paid to program in several languages over the years (C++, Python, Pascal, etc). Alex has posted 43 posts at DZone. You can read more from them at their website. View Full User Profile

Coordinating Threads

02.07.2008
| 25717 views |
  • submit to reddit

Java 5 introduced many new concurrency primitives and collections, and this post is going to look at two classes that can be used to coordinate threads: CountDownLatch and CyclicBarrier.

A CountDownLatch is initialized with a counter. Threads can then either count down on the latch or wait for it to reach 0. When the latch reaches 0, all waiting threads are released.

A common idiom is to use a latch to trigger a coordinated start or end between threads:

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CountDownDemo {

public static void main(String[] args) throws Exception {
int threads = 3;
final CountDownLatch startLatch = new CountDownLatch(threads);
final CountDownLatch endLatch = new CountDownLatch(threads);

ExecutorService svc = Executors.newFixedThreadPool(threads);
for (int i = 0; i < threads; i++) {
svc.execute(new Runnable() {
public void run() {
try {
log("At run()");
startLatch.countDown();
startLatch.await();

log("Do work");
Thread.sleep((int) (Math.random() * 1000));

log("Wait for end");
endLatch.countDown();
endLatch.await();

log("Done");
} catch (Exception e) {
e.printStackTrace();
}
}
});
Thread.sleep(100);
}
}

private static void log(String msg) {
System.out.println(System.currentTimeMillis() + ": "
+ Thread.currentThread().getId() + " " + msg);
}
}

In this code, you'll see two latches get initialized. Each thread that starts up counts down on the latch and awaits the latch counting down to 0 (when all threads have been initialized). Similarly, each thread waits for all threads to complete at the same time.

Running this program yields:

1194812267416: 7 At run()
1194812267517: 8 At run()
1194812267618: 9 At run()
1194812267618: 9 Do work
1194812267618: 7 Do work
1194812267619: 8 Do work
1194812267673: 7 Wait for end
1194812267688: 8 Wait for end
1194812268023: 9 Wait for end
1194812268023: 9 Done
1194812268023: 7 Done
1194812268023: 8 Done

You can see that each thread hits run() at different times, but proceeds past the barrier at the same time. They each then do some random amount of work and wait for the latch, then proceed past it together.

In the example above, each thread waits forever for the latch to trigger. You can also choose to wait for a specified time period before giving up. And you can check the latch to see how many threads have arrived and are now waiting. Each CountDownLatch instance can only be used once and is then dead.

If you want a set of threads to repeatedly meet at a common point, you are better served by using a CyclicBarrier. A common use for this is in multi-threaded testing where it is typical to start a bunch of threads, meet, do some stuff, meet, validate some assertions, repeatedly.

The prior program can be simplified by replacing the two latches with a single barrier:

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CyclicBarrierDemo {

public static void main(String[] args) throws Exception {
int threads = 3;
final CyclicBarrier barrier = new CyclicBarrier(threads);

ExecutorService svc = Executors.newFixedThreadPool(threads);
for (int i = 0; i < threads; i++) {
svc.execute(new Runnable() {
public void run() {
try {
log("At run()");
barrier.await();

log("Do work");
Thread.sleep((int) (Math.random() * 1000));

log("Wait for end");
barrier.await();

log("Done");
} catch (Exception e) {
e.printStackTrace();
}
}
});
Thread.sleep(100);
}
}

private static void log(String msg) {
System.out.println(System.currentTimeMillis() + ": "
+ Thread.currentThread().getId() + " " + msg);
}
}

We can see here that the threads can repeatedly wait at the barrier, which implicitly counts down until all threads have arrived, then releases all threads.

Another nice trick with CyclicBarrier is that a Runnable action can be associated with the barrier to be run by the last thread reaching the barrier. You can very simply build a start/end timer for testing with this functionality:

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class TimerBarrierDemo {

public static void main(String[] args) throws Exception {
int threads = 3;
final CyclicBarrier barrier = new CyclicBarrier(threads, new BarrierTimer());

ExecutorService svc = Executors.newFixedThreadPool(threads);
for (int i = 0; i < threads; i++) {
svc.execute(new Runnable() {
public void run() {
try {
barrier.await();
long sleepTime = (int) (Math.random() * 1000);
System.out.println(Thread.currentThread().getId() + " working for " + sleepTime);
Thread.sleep(sleepTime);
barrier.await();
} catch (Exception e) {
e.printStackTrace();
}
}
});
}
}

private static class BarrierTimer implements Runnable {
private long start;

public void run() {
if (start == 0) {
start = System.currentTimeMillis();
} else {
long end = System.currentTimeMillis();
long elapsed = (end - start);
System.out.println("Completed in " + elapsed + " ms");
}
}
}
}

Here we rely on knowing that the barrier will be reached exactly twice - once at start and once at end. The first time it's reached, we record a timestamp and the second time it's reached we print out the timing. When we construct our barrier, we give it an instance of this timer class. Each thread then waits to start on the barrier, works for a random amount of time, and waits for the end barrier.

A run looks like this:

9 working for 35
7 working for 341
8 working for 371
Completed in 372 ms

Generally, you should expect the recorded elapsed time to be the maximum of the working time of any of the threads.

CyclicBarrier has a few additional tricks as well - threads can wait for a time period instead of forever, check whether a barrier has been broken (by interruption or forcibly with a reset() method), and determine the number of parties and the number currently waiting.

Published at DZone with permission of its author, Alex Miller.

(Note: Opinions expressed in this article and its replies are the opinions of their respective authors and not those of DZone, Inc.)

Comments

Alex Miller replied on Thu, 2008/02/07 - 2:13pm

Hey, I somehow published this with comments off - sorry about that!

A colleague of mine just coincidentally published a blog today about using distributed CyclicBarrier with Terracotta, which is an interesting companion to the article above. 

googlsa puyad replied on Sat, 2009/05/23 - 10:33am

Hey, I somehow published this with comments off . ED Hardy Cheap ED Hardy ED Hardy Clothes

jame jack replied on Wed, 2009/06/17 - 7:40pm

تحميل برامج برامج جوالات العاب بنات تكنولوجيا كتب تعليم UltraSurf Internet Download Manager ProgDVB برامج مجانية أفضل المواقع العربية مشاهدة محطات مشفرة Online TV Player 3.0.0.940 Internet Download Manager 5.17 Build 4 رقص شرقي anyTV Pro 4.32 OnLineLive 7.1.1 هزي يانواعم ProgDVB 6.06.2 SopCast 3.0.3 Falco Image Studio 3.6 لعبة تزلج على الجليد UltraSurf 9.4 كاثرين هيغل Katherine Heigl محطة غنوة FreeZ Online TV 1.0 Free Video to Mp3 Converter 3.1.3.51 Advanced MP3 Converter 2.10 Xilisoft Video to Audio Converter 5.1.23.0515 Blaze Media Pro 8.02 AKRAM Media Creator 1.11 DVD Audio Extractor 4.5.4 Free WMA to MP3 Converter 1.16 لعبة نينجا المتقدم لعبة قذف كرة لعبة دراجات البهلوانية لعبة اعداء الغابة تحميل برامج Download DivX Subtitles 2.0 BullGuard 8.5 Google Chrome 2.0.181.1 Dev Dell Studio XPS Desktop 435T Intel Matrix Storage Manager A00 Gigabyte GA-EP45-UD3P Bios F9 Ambush HDConvertToX 1.1.229.1764 MSI Wind Nettop CS 120 Realtek Audio Driver 5.10.0.5618 Biostar T41-A7 6.x Realtek On-Board Audio Driver 5.10.0.5735 for 2000/2003/XP TweakNow RegCleaner 4.1.1 SpeedItup Free 4.97 برامج العاب - Internet Download Manager - برامج جوالات - العاب - محطة غنوة - قنوات فضائية - بنات - تكنولوجيا - كتب تعليم - UltraSurf -

Ming Shing replied on Thu, 2010/11/25 - 4:34am

The CountDownLatch protects you against the case of a thread missing a signal which can occur if you use these other mechanisms for coordinating jobs.

pass4sure MB7-843, pass4sure MB7-846, pass4sure MB7-848

Ash Mughal replied on Thu, 2012/01/26 - 2:13am

When writing the code that runs when your thread is executed, you must consider the behavior of other threads that may be executing simultaneously. In particular, care must be taken to avoid two threads trying to use the same global object or variable at the same time. In addition, the code in one thread can depend on the results of tasks performed by other threads.

advanced java

Comment viewing options

Select your preferred way to display the comments and click "Save settings" to activate your changes.