Producer Consumer Controller

[Kasi Viswan]

Hello,

I need some help with this Producer Consumer Controller code.

I need to perform Task A on all the records of an input file. Input file may contain millions of records.
I need to be able to run this process at a steady thread count of say 20 threads.

I wrote three classes - Producer, Consumer and Controller

    
class Producer extends Thread
{
   BlockingQueue queue;

   public Producer(BlockingQueue queue)
   {
      this.queue = queue;
   }

   public void run()
   {
	  while(records left in the input file)
	  {
	      read a record
  		  Runnable run = new Runnable()
		  {
			public void run()
			{
			  TaskA.process(inputRecord)			
			}
		  }
		  queue.put(run)
	  }
   }
}

    
class Consumer extends Thread
{
   BlockingQueue queue;
   ThreadPoolExecutor tpex;
   
   public Consumer(BlockingQueue queue, int threadPoolSize)
   {
      this.queue = queue;
	  tpex = new ThreadPoolExecutor(threadPoolSize, threadPoolSize, 60, TimeUnit.SECONDS, queue);
   }
   
   public void run()
   {
      while(!tpex.isShutdown())
		{
			try {
				tpex.execute(queue.take());			
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
   }
   
   public void shutdown()
	{
		while (tpex.getActiveCount() != 0)
		{
			try {
				sleep(1000);
			} catch (InterruptedException e) {
				Thread.currentThread().interrupt();
			}						
		}
		tpex.shutdown();
	}	
}

class Controller
{
	BlockingQueue queue;
	Producer p;
	Consumer q;
	
   public Controller()
   {
	  threadCount = 5;
      queue = new LinkedBlockingQueue<Runnable>(threadCount*2);
	   p = new Producer(queue);
	   c = new Consumer(queue, threadCount)
   }
   
   public void process()
   {
      p.start()
	  c.start()
	  p.join()
	  c.join()
	  c.shutdown()
   }
}

I do not know how to make the program wait till the producer and consumer completes their task and then exit the VM.
If I set the threadcount less than the number of records in the file, the program exits properly after processing all the records.
If I set threadcount equal to the number of records in file, the program never exits, even though all the records were processed.

Can someone please tell me how to get past this, Thank you.

Regards
Kasi

[Mike Peters]

Maybe removing c.join() will speed up your shutdown process. You have a deadlock there.

[Jim Hoglund]

Kasi : Where are you with your project? Have you solved your
problem or do you still need some pointers?

Jim ... ...

[Kasi Viswan]

Thank you Jim, Mike was right on the mark, removing join statement on the consumer made it work.

It would be great, if you can help me in the same app how to vary number of consumer threads at run time.

E.g. Increase/Decrease the consumer count based on TPS.
Lets say, i started this process, with 10 consumers and i find that based on some parameters that i can start few more consumers or reduce the number of consumers, how can i do that, Please advice.

Thanks
Kasi

[Sandeep Sanaboyina]

Try using Executors. Check out this link.

http://cupi2.uniandes.edu.co/site/images/recursos/javadoc/j2se/1.5.0/docs/api/java/util/concurrent/Executors.html#newCachedThreadPool(java.util.concurrent.ThreadFactory)

[Kasi Viswan]

Hi Sandeep,

Sorry, I couldn't figure out how to use newCachedThreadPool() for this scenario.

There is a set of producers injecting tasks into a blockingqueue of size 10.
There is a set of consumers removing these tasks from the queue and executing these tasks.

I want to know how to increase the number of consumers while the application is running.
Do you have a sample code or pseudo code that i can refer?

Thanks
Kasi

[Sandeep Sanaboyina]

I'm giving the sample code for ThreadPoolExecutor. Run it once and see how new threads are created.

Also look up javadocs for ThreadPoolExecutor.

Hope it helps.

  
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

//This is the thread factory that can be used to create consumer threads. A default ThreadFactory can also be used
class SimpleThreadFact implements ThreadFactory {
	public Thread newThread(Runnable arg0) {
		// TODO Auto-generated method stub
		return new Thread(arg0);
	}
}

//This is the actual thread that does the execution
class SimpleThread implements Runnable{
	private String data;
	public SimpleThread(String data){
		this.data = data;
	}
	public void run() {
		// TODO Auto-generated method stub
		System.out.println("Starting Thread with data:"+data);
		try {
			Thread.sleep(10000);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}

//This gives the statistics of the ThreadPoolExecutor so that we can see the creation of new threads and closing of idle threads.
class MonitorThread implements Runnable{
	
	private ThreadPoolExecutor tpe;
	public MonitorThread(ThreadPoolExecutor tpe){
		this.tpe = tpe;
	}
	
	public void run() {
		// TODO Auto-generated method stub
		while(true){
			try {
				System.out.println("Stats---->Active Count:"+tpe.getActiveCount()+"  Total Count:"+tpe.getPoolSize()+"  Queue Size:"+tpe.getQueue().size());
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
	
}

public class TPEExample {
	
	public static void main(String[] args) {
		try {
			//Creating a ThreadPoolExecutor with corePoolSize 10 (min threads) , maximum 30 threads (maxPoolSize), Maximum size of the queue 10 and a custom ThreadFactory
			/*
				1) If the number of threads is less than the corePoolSize, create a new Thread to run a new task
				2) If the number of threads is equal (or greater than) the corePoolSize, put the task into the queue
				3) If the queue is filled, and the number of threads is less than the maxPoolSize, create a new thread to run tasks in.
				4) If the queue is filled, and the number of threads is greater than or equal to maxPoolSize, reject the task.

			 */
			ThreadPoolExecutor tpe = new ThreadPoolExecutor(10, 30, 3000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(10),new SimpleThreadFact());
			Thread monitor = new Thread(new MonitorThread(tpe));
			monitor.setDaemon(true);
			monitor.start();
			//I have simulated three different block with different loads to simulate increasing TPS.
			for(int i=0;i<10;i++){
				try {
					//This is where you could put the code- getting the data from the queue and giving it to your Consumer thread pool 
					tpe.execute(new SimpleThread("Data"+i));
				} catch (RejectedExecutionException e) {
					// This exception comes up when the queue limit is reached. No new tasks are put in the queue. This can be handled as per the requirement
					System.out.println("SORRY !!! Unable to execute task");
					//e.printStackTrace();
				}
			}
			
			Thread.sleep(10000);
			
			for(int i=0;i<20;i++){
				try {
					tpe.execute(new SimpleThread("Data"+i));
				} catch (RejectedExecutionException e) {
					// TODO Auto-generated catch block
					System.out.println("SORRY !!! Unable to execute task");
					//e.printStackTrace();
				}
			}

			Thread.sleep(2000);

			for(int i=0;i<35;i++){
				try{
					tpe.execute(new SimpleThread("Data"+i));
				} catch (RejectedExecutionException e) {
					// TODO Auto-generated catch block
					System.out.println("SORRY !!! Unable to execute task");
					//e.printStackTrace();
				}
			}
			
			Thread.sleep(30000);
			tpe.shutdown();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
}

[Kasi Viswan]

Thank you Sandeep for your detailed explanation. I understood your suggestion to use for loop and keep calling more execute() in the executor to increase number of consumers.

If there is a way to increase the blockingqueue size at runtime, I need not change any of my producer consumer setup.

Thanks again,
Kasi

[Jim Hoglund]

Kasi : Can you read a parameter file periodically, or pass in a queue size on the command
line at start-up? You could then construct a new (larger/smaller) blocking queue, transfer
any pending jobs to it, and shut down the old queue. What do you think?

Jim ... ...

[Kasi Viswan]

Good Idea Jim, I like the periodic parameter file reading and transferring the tasks to new queue.

Thanks
Kasi