threadpool.java

snmp4j

/*_############################################################################
  _##
  _##  SNMP4J - ThreadPool.java
  _##
  _##  Copyright 2003-2005  Frank Fock and Jochen Katz (SNMP4J.org)
  _##
  _##  Licensed under the Apache License, Version 2.0 (the "License");
  _##  you may not use this file except in compliance with the License.
  _##  You may obtain a copy of the License at
  _##
  _##      http://www.apache.org/licenses/LICENSE-2.0
  _##
  _##  Unless required by applicable law or agreed to in writing, software
  _##  distributed under the License is distributed on an "AS IS" BASIS,
  _##  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  _##  See the License for the specific language governing permissions and
  _##  limitations under the License.
  _##
  _##########################################################################*/

package org.snmp4j.util;

import java.util.*;
import org.snmp4j.util.ThreadPool.TaskManager;

/**
 * The <code>ThreadPool</code> provides a pool of a fixed number of threads
 * that are capable to execute tasks that implement the <code>Runnable</code>
 * interface concurrently. The ThreadPool blocks when all threads are busy
 * with tasks and an additional task is added.
 *
 * @author Frank Fock
 * @version 1.0.2
 * @since 1.0.2
 */
public class ThreadPool {

  protected Vector taskManagers;
  protected String name = "ThreadPool";
  protected volatile boolean stop = false;
  protected boolean respawnThreads = false;

  protected ThreadPool() {
  }

  protected String getTaskManagerName(String prefix, int index) {
    return prefix+"."+index;
  }

  protected void setup(String name, int size) {
    this.name = name;
    taskManagers = new Vector(size);
    for (int i=0; i<size; i++) {
      TaskManager tm = new TaskManager(getTaskManagerName(name, i));
      taskManagers.add(tm);
      tm.start();
    }
  }

  /**
   * Creates a thread pool with the supplied name and size.
   * @param name
   *    the name prefix for the threads in this pool.
   * @param size
   *    the number of threads in this pool. This number also specifies the
   *    number of concurrent tasks that can be executed with this pool.
   * @return
   *    a <code>ThreadPool</code> instance.
   */
  public static ThreadPool create(String name, int size) {
    ThreadPool pool = new ThreadPool();
    pool.setup(name, size);
    return pool;
  }

  /**
   * Executes a task on behalf of this thread pool. If all threads are currently
   * busy, this method call blocks until a thread gets idle again which is when
   * the call returns immediately.
   * @param task
   *    a <code>Runnable</code> to execute.
   */
  public synchronized void execute(Runnable task) {
    TaskManager tm = null;
    while (tm == null) {
      for (int i=0; i<taskManagers.size(); i++) {
        tm = (TaskManager) taskManagers.get(i);
        if ((respawnThreads) && (!tm.isAlive())) {
          tm = new TaskManager(getTaskManagerName(name, i));
        }
        if (tm.isIdle()) {
          tm.execute(task);
          return;
        }
        else {
          tm = null;
        }
      }
      try {
        wait();
      }
      catch (InterruptedException ex) {
        // ignore
      }
    }
  }

  /**
   * Tests if the threads are respawn (recreates) when they have been stopped
   * or canceled.
   * @return
   *    <code>true</code> if threads are respawn.
   */
  public boolean isRespawnThreads() {
    return respawnThreads;
  }

  /**
   * Specifies whether threads are respawned by this thread pool after they
   * have been stopped or not. Default is no respawning.
   * @param respawnThreads
   *    if <code>true</code> then threads will be respawn.
   */
  public void setRespawnThreads(boolean respawnThreads) {
    this.respawnThreads = respawnThreads;
  }

  /**
   * Returns the name of the thread pool.
   * @return
   *    the name of this thread pool.
   */
  public String getName() {
    return name;
  }

  /**
   * Stops all threads in this thread pool gracefully. This method will not
   * return until all threads have been terminated and joined successfully.
   */
  public synchronized void stop() {
    stop = true;
    for (int i=0; i<taskManagers.size(); i++) {
      TaskManager tm = (TaskManager) taskManagers.get(i);
      tm.terminate();
      synchronized (tm) {
        tm.notify();
      }
      try {
        tm.join();
      }
      catch (InterruptedException ex) {
        //ignore
      }
    }
  }

  /**
   * Cancels all threads non-blocking by interrupting them.
   */
  public synchronized void cancel() {
    stop = true;
    for (int i=0; i<taskManagers.size(); i++) {
      TaskManager tm = (TaskManager) taskManagers.get(i);
      tm.terminate();
      tm.interrupt();
    }
  }

  /**
   * The <code>TaskManager</code> executes tasks in a thread.
   *
   * @author Frank Fock
   * @version 1.0.2
   * @since 1.0.2
   */
  class TaskManager extends Thread {

    private Runnable task = null;
    private volatile boolean run = true;

    public TaskManager(String name) {
      super(name);
    }

    public synchronized void run() {
      while (run) {
        if (task != null) {
          task.run();
          synchronized (ThreadPool.this) {
            task = null;
            ThreadPool.this.notify();
          }
        }
        else {
          try {
            wait();
          }
          catch (InterruptedException ex) {
            run = false;
            break;
          }
        }
      }
    }

    public boolean isIdle() {
      return ((task == null) && run);
    }

    public boolean isStopped() {
      return stop;
    }

    public void terminate() {
      stop = true;
    }

    public synchronized void execute(Runnable task) {
      if (this.task == null) {
        this.task = task;
        notify();
      }
      else {
        throw new IllegalStateException("TaskManager is not idle");
      }
    }
  }
}