Fork Join : Work Stealing

Load Balancing vs. Synchronization

One of the key challenges in parallelizing any type of workload is the partitioning step: ideally we want to partition the work such that every piece will take the exact same amount of time. In reality, we often have to guess at what the partition should be, which means that some parts of the problem will take longer, either because of the inefficient partitioning scheme, or due to some other, unanticipated reasons (e.g. external service, slow disk access, etc).

This is where work-stealing comes in. If some of the CPU cores finish their jobs early, then we want them to help to finish the problem. However, now we have to be careful: trying to "steal" work from another worker will require synchronization, which will slowdown the processing. Hence, we want work-stealing, but with minimal synchronization.

Fork/Join Work-Stealing

The Fork-Join framework (docs) solves this problem in a clever way: recursive job partitioning, and a double-ended queue (deque) structure for holding the tasks.


Given a problem, we divide the problem into N large pieces, and hand each piece to one of the workers (2 in the diagram above). Each worker then recursively subdivides the first problem at the head of the deque and appends the split tasks to the head of the same deque. After a few iterations we will end up with some number of smaller tasks at the front of the deque, and a few larger and yet to be partitioned tasks on end. So far so good, but what do we get?

Imagine the second worker has finished all of its work, while the first worker is busy. To minimize synchronization the second worker grabs a job from the end of the deque (hence the reason for efficient head and tail access). By doing so, it will get the largest available block of work, allowing it to minimize the number of times it has to interact with the other worker (aka, minimize synchronization). Simple, but a very clever technique!

Work stealing would be like this: Worker B has finished his work. He is a kind one, so he looks around and sees Worker A still working very hard. He strolls over and asks: "Hey lad, I could give you a hand." A replies. "Cool, I have this task of 1000 units. So far I have finished 345 leaving 655. Could you please work on number 673 to 1000, I'll do the 346 to 672." B says "OK, let's start so we can go to the pub earlier."

You see - the workers must communicate between each other even when they started the real work. This is the missing part in the examples.

The only remaining difference between Fork/Join and splitting the task upfront is this: When splitting upfront you have the work queue full right from start. Example: 1000 units, the threshold is 10, so the queue has 100 entries. These packets are distributed to the threadpool members.
Fork/Join is more complex and tries to keep the number of packets in the queue smaller:

• Step 1: Put one packet containing (1...1000) into queue
• Step 2: One worker pops the packet(1...1000) and replaces it with two packets: (1...500) and (501...1000).
• Step 3: One worker pops packet (500...1000) and pushes (500...750) and (751...1000).
• Step n: The stack contains these packets: (1..500), (500...750), (750...875)... (991..1000)
• Step n+1: Packet (991..1000) is popped and executed
• Step n+2: Packet (981..990) is popped and executed
• Step n+3: Packet (961..980) is popped and split into (961...970) and (971..980). ....

You see: in Fork/Join the queue is smaller (6 in the example) and the "split" and "work" phases are interleaved.

CyclicBarrier

What does CyclicBarrier do ?

Await will suspend itself until N number of threads have invoked await on the barrier. So if you define new CyclicBarrier(3) than once 3 threads invoke await the barrier will allow threads to continue.

Basically it is used to Synchronize tasks in a common point

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.

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.

Example : Suppose T1 prints 1,3,5 etc and T2 prints 2,5,6 etc and T3 prints 3,6,9 how does main print 1,2,3,4,5,6,

class ThreadTest {
 private CyclicBarrier cyclicBarrier = new CyclicBarrier(2, new Runnable() {
      @Override
     public void run() {
           System.out.println(oddNumberGenerator.result);
         System.out.println(evenNumberGenerator.result);
     }
 });

 private NumberGenerator oddNumberGenerator = new NumberGenerator(1,11,2);
 private NumberGenerator evenNumberGenerator = new NumberGenerator(2,10,2);

 public void generateSeries(){
     oddNumberGenerator.generateNumbers();
     evenNumberGenerator.generateNumbers();
 }

 class NumberGenerator {
     private Thread thread;
     private int result;

     private NumberGenerator(final int initialValue, final int maxValue,final int stepSize) {
         this.thread = new Thread(new Runnable() {
             @Override
             public void run() {
                 for (int i = initialValue; i <= maxValue; i = i + stepSize) {
                     try {
                         result = i;
                         cyclicBarrier.await();
                     } catch (InterruptedException e) {
                         e.printStackTrace();
                     } catch (BrokenBarrierException e) {
                         e.printStackTrace();
                     }
     }
             }
         });
     }
     public void generateNumbers() {
          thread.start();
     }
 }
 main(String[] args){
     new ThreadTest().generateSeries();
 }
}

ok

CountDownLatch

CountDownLatch works in latch principle, main thread will wait until gate is open. One thread waits for n number of threads specified while creating CountDownLatch in Java.

Any thread, usually main thread of application, which calls CountDownLatch.await( ) will wait until count reaches zero or its interrupted by another thread. All other thread are required to do count down by calling CountDownLatch.countDown( ) once they are completed or ready.

As soon as count reaches zero, Thread awaiting starts running. One of the disadvantages/advantages of CountDownLatch is that its not reusable once count reaches to zero you can not use CountDownLatch any more.

public class Audioconference implements Runnable{
 // This class uses a CountDownLatch to control the arrivel of all the participants
 private final CountDownLatch controller;
 
 // Constructor of the class. Initializes the CountDownLatch @param number The number of participants in the Audioconference
 public Audioconference(int number) {
  controller=new CountDownLatch(number);
 }

 // This method is called by every participant when he incorporates to the Audioconference * @param participant
 public void arrive(String name){
  System.out.printf("%s has arrived.\n",name);
  // This method uses the countDown method to decrement the internal counter of the
  // CountDownLatch
  controller.countDown();
  System.out.printf("Audioconference: Waiting for %d participants.\n",controller.getCount());
 }
 
 // This is the main method of the Controller of the Audioconference. It waits for all the participants and the, starts the conference
 @Override
 public void run() {
  System.out.printf("Audioconference: Initialization: %d participants.\n",controller.getCount());
  try {
   // Wait for all the participants
   controller.await();
   // Starts the conference
   System.out.printf("Audioconference: All the participants have come\n");
   System.out.printf("Audioconference: Let's start...\n");
  } catch (InterruptedException e) {
   e.printStackTrace();
  }
 }
 
}

And Dialer

public class Dialer implements Runnable {

 // Audioconference in which this Dialer will take part off
 private Audioconference conference;
 
 // Name of the Dialer. For log purposes only
 private String name;
 
 /**
  * Constructor of the class. Initialize its attributes
  * @param conference Audioconference in which is going to take part off
  * @param name Name of the Dialer
  */
 public Dialer(Audioconference conference, String name) {
  this.conference=conference;
  this.name=name;
 }

 // Core method of the Dialer. Waits a random time and joins the Audioconference 
 @Override
 public void run() {
  Long duration=(long)(Math.random()*10);
  try {
   TimeUnit.SECONDS.sleep(duration);
  } catch (InterruptedException e) {
   e.printStackTrace();
  } 
  conference.arrive(name);
 }
}

Main

 public static void main(String[] args) {

  // Creates a Audioconference with 10 Dialers.
  Audioconference conference=new Audioconference(10);
  // Creates a thread to run the Audioconference and start it.
  Thread threadConference=new Thread(conference);
  threadConference.start();
  
  // Creates ten Dialers, a thread for each one and starts them
  for (int i=0; i<10; i++){
   Dialer p=new Dialer(conference, "Dialer "+i);
   Thread t=new Thread(p);
   t.start();
  }

 }

Semaphores

Semaphores , can be of 2 types

1. Couting
2. Binary

Semaphores which allow an arbitrary resource count are called counting semaphores, while semaphores which are restricted to the values 0 and 1 (or locked/unlocked, unavailable/available) are called binary semaphores.

Library Analogy

Suppose a library has 10 identical study rooms, to be used by one student at a time. To prevent disputes, students must request a room from the front desk if they wish to make use of a study room.

The clerk at the front desk does not keep track of which room is occupied or who is using it, nor does he or she know if the room is actually being used, only the number of free rooms available, which she only knows correctly if all of the students actually use their room and return them when they're done. When a student requests a room, the clerk decreases this number. When a student releases a room, the clerk increases this number. Once access to a room is granted, the room can be used for as long as desired, and so it is not possible to book rooms ahead of time.

In this scenario the front desk represents a semaphore, the rooms are the resources, and the students represent processes. The value of the semaphore in this scenario is initially 10. When a student requests a room he or she is granted access and the value of the semaphore is changed to 9. After the next student comes, it drops to 8, then 7 and so on.

Important observations

When used for a pool of resources, a semaphore tracks only how many resources are free; it does not keep track of which of the resources are free. Some other mechanism (possibly involving more semaphores) may be required to select a particular free resource.

 Let's  see some sample code that represents same

public class LibraryRoomQueue {
 // Semaphore to control the access to the reading room
 private Semaphore semaphore;
 
 // Array to control what room is free
 private boolean freeRooms[];
 
 // Lock to control the access to the freeRooms array
 private Lock lockRooms;
 
 //Constructor of the class. It initializes the three objects
 public LibraryRoomQueue(){
  semaphore=new Semaphore(3);
  freeRooms=new boolean[3];
  for (int i=0; i<3; i++){
   freeRooms[i]=true;
  }
  lockRooms=new ReentrantLock();
 }
 
 public void bookRoom (Object document){
  try {
   // Get access to the semaphore. If there is one or more rooms free,
   // it will get the access to one of the rooms
   semaphore.acquire();
   
   // Get the number of the free room
   int assignedroom=getroom();
   
   Long duration=(long)(Math.random()*10);
   System.out.printf("%s: Booking a room %d during %d seconds\n",Thread.currentThread().getName(),assignedroom,duration);
   TimeUnit.SECONDS.sleep(duration);
   
   // Free the room
   freeRooms[assignedroom]=true;
  } catch (InterruptedException e) {
   e.printStackTrace();
  } finally {
   // Free the semaphore
   semaphore.release();   
  }
 }

 private int getroom() {
  int ret=-1;
  try {
   // Get the access to the array
   lockRooms.lock();
   // Look for the first free room
   for (int i=0; i<freeRooms.length; i++) {
    if (freeRooms[i]){
     ret=i;
     freeRooms[i]=false;
     break;
    }
   }
  } catch (Exception e) {
   e.printStackTrace();
  } finally {
   // Free the access to the array
   lockRooms.unlock();
  }
  return ret;
 }

}

:)