java - Doubts in the code of Thread Pool implementation

Question

After spending lots of time with threadpool concepts and by reading different codes on numbers of blogs and posting questions on Stackoverflow.com, now I got clear image of this concept. But in the meanwhile, I found some doubts in code.

1. When pool.assign(new TestWorkerThread()); executes in TestThreadPool Class, it calls done.workerBegin(); method that is in Done Class, where it increments _activeThreads variable. But what I thinks is, LOGICALLY that is not correct because if number of threads are less(in this case 2) than number of tasks (given in TestThreadPool Class)(in this case 5), it increments _activeThreads (i.e., _activeThreads = 5) counts unnecessarily.

2. What _started variable does in Done class?

3. How waitDone() and waitBegin() (in Done Class ) performs their functioning? (It is good if you explain these two methods step by step.)

Code is as follows. I am arranging the codes according to its flow.

TestThreadPool Class :-

package hitesh;

/**
 *
 * @author jhamb
 */

public class TestThreadPool {

 public static void main(String args[]) throws InterruptedException
 {
  ThreadPool pool = new ThreadPool(2);

  for (int i = 1;i <= 5;i++) {
   pool.assign(new TestWorkerThread());
  }
  System.out.println("All tasks are assigned");

  pool.complete();

  System.out.println("All tasks are done.");
 }
}

TestWorkerThread Class :-

package hitesh;

/**
 *
 * @author jhamb
 */
/**
 * This class shows an example worker thread that can
 * be used with the thread pool. It demonstrates the main
 * points that should be included in any worker thread. Use
 * this as a starting point for your own threads.
 */

public class TestWorkerThread implements Runnable {
 static private int count = 0;
 private int taskNumber;
 protected Done done;

 /**
  * 
  * @param done
  */
 TestWorkerThread()
 {
  count++;
  taskNumber = count;
  //System.out.println("tasknumber  --->  " + taskNumber);
 }

 public void run()
 {
  System.out.println("TWT run starts   -->  "  + this.toString());
  for (int i=0;i <= 100;i += 25) {
   System.out.println("Task number: " + taskNumber + 
             ",percent complete = " + i );
   try {
    Thread.sleep((int)(Math.random()*500));
   } catch (InterruptedException e) {
   }
  }
  System.out.println("task for thread --> " + this.toString() + "   completed");
 }
}

ThreadPool Class :-

package hitesh;

/**
 *
 * @author jhamb
 */
import java.util.*;


/* 
 * This is the main class for the thread pool. You should
 * create an instance of this class and assign tasks to it.
 */

public class ThreadPool {

 protected Thread threads[] = null;

 Collection assignments = new ArrayList(3);

 protected Done done = new Done();


 public ThreadPool(int size) throws InterruptedException
 {
   threads = new WorkerThread[size];
   for (int i=0;i<threads.length;i++) {
    threads[i] = new WorkerThread(this);
    threads[i].start();
    System.out.println ("thread " + i + " started");
    threads[i].sleep(1000);
  }

 }

 public synchronized void assign(Runnable r)
 {
  done.workerBegin();
  assignments.add(r);
  System.out.println("Collection size --->   " + assignments.size() +   "  Thread can work on this");
  notify();
 }

 public synchronized Runnable getAssignment()
 {
  try {
   while ( !assignments.iterator().hasNext() )
    wait();

   Runnable r = (Runnable)assignments.iterator().next();
   assignments.remove(r);
   return r;
  } catch (InterruptedException e) {
   done.workerEnd();
   return null;
  }
 }

 public void complete()
 {   
  done.waitBegin();
  done.waitDone();
 }

}

WorkerThread Class :-

package hitesh;
import java.util.*;
/**
 *
 * @author jhamb
 */

/**
 * The worker threads that make up the thread pool.
 */
class WorkerThread extends Thread {
 /**
  * True if this thread is currently processing.
  */
 public boolean busy;
 /**
  * The thread pool that this object belongs to.
  */
 public ThreadPool owner;

 /**
  * The constructor.
  * 
  * @param o the thread pool 
  */
 WorkerThread(ThreadPool o)
 {
  owner = o;
 }

 /**
  * Scan for and execute tasks.
  */
    //@Override
 public void run()
 {
  System.out.println("Threads name : "+ this.getName() + "  working.....");
  Runnable target = null;

  do {
   System.out.println("enter in do while " + this.getName() );
   target = owner.getAssignment();
   System.out.println("GetAssignment k aage aa gya mai "  +  target);
   if (target!=null) {
    target.run();
    //target.
    owner.done.workerEnd();
   }
  } while (target!=null);
  System.out.println("do while finishes for "+ this.getName());
 }
}

Done Class :-

package hitesh;

/**
 *
 * @author jhamb
 */
/**
 * 
 * This is a thread pool for Java, it is
 * simple to use and gets the job done. This program and
 * all supporting files are distributed under the Limited
 * GNU Public License (LGPL, http://www.gnu.org).
 * 
 * This is a very simple object that
 * allows the TheadPool to determine when 
 * it is done. This object implements
 * a simple lock that the ThreadPool class
 * can wait on to determine completion.
 * Done is defined as the ThreadPool having
 * no more work to complete.
 * 
 * Copyright 2001 by Jeff Heaton
 *
 * @author Jeff Heaton (http://www.jeffheaton.com)
 * @version 1.0
 */
public class Done {

 /**
  * The number of Worker object
  * threads that are currently working
  * on something.
  */
 private int _activeThreads = 0;

 /**
  * This boolean keeps track of if
  * the very first thread has started
  * or not. This prevents this object
  * from falsely reporting that the ThreadPool 
  * is done, just because the first thread
  * has not yet started.
  */
 private boolean _started = false;
 /**
  * This method can be called to block
  * the current thread until the ThreadPool
  * is done.
  */

 synchronized public void waitDone()
 {
  try {
   while ( _activeThreads>0 ) {
    wait();
   }
  } catch ( InterruptedException e ) {
  }
 }
 /**
  * Called to wait for the first thread to 
  * start. Once this method returns the
  * process has begun.
  */

 synchronized public void waitBegin()
 {
  try {
   while ( !_started ) {
    wait();
   }
  } catch ( InterruptedException e ) {
  }
 }


 /**
  * Called by a Worker object
  * to indicate that it has begun 
  * working on a workload.
  */
 synchronized public void workerBegin()
 {
  _activeThreads++;
  _started = true;
  notify();
 }

 /**
  * Called by a Worker object to 
  * indicate that it has completed a 
  * workload.
  */
 synchronized public void workerEnd()
 {
  _activeThreads--;
  notify();
 }

 /**
  * Called to reset this object to
  * its initial state.
  */
 synchronized public void reset()
 {
  _activeThreads = 0;
 }

}

Please help. Thanks in advance. Looking for your kind response.

Answer 1

Now I understand that whole code very perfectly. If you find any doubts in this code, then you can ask.

Answers of my questions are as follows after reading a lot on this.

1. Yes, you are right, it is logically wrong. Its better, if it would be _activeTasks . It is used to kill all the threads , when threadpool have no more work because waitDone() function executes successfully only when _activeTasks <= 0.

2. This Variable is used in waitBegin() method. Whenever any tasks starts, it updates _started by TRUE, means the tasks that are assigned by users is now in processing by threads, means threads starts working on these tasks. If tasks is not given by user, then all threads are still active , and waiting for tasks. This is the use of this variable here.

3. waitBegin() method executes successfully when threads starts working on tasks, because in that case only _started become true. Otherwise, threads keep on waiting for some tasks. waitDone() executes successfully only when _activeTasks become Zero, because this is the only situation when threadpool don't have any work to perform, means threadpool completes its work. Otherwise, it keep waiting until all tasks finish, means it waits until when _activeTasks becomes ZERO