scheduledthreadpool.java

来自「RESIN 3.2 最新源码」· Java 代码 · 共 855 行 · 第 1/2 页

/*
 * Copyright (c) 1998-2008 Caucho Technology -- all rights reserved
 *
 * This file is part of Resin(R) Open Source
 *
 * Each copy or derived work must preserve the copyright notice and this
 * notice unmodified.
 *
 * Resin Open Source is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * Resin Open Source is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE, or any warranty
 * of NON-INFRINGEMENT.  See the GNU General Public License for more
 * details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Resin Open Source; if not, write to the
 *
 *   Free Software Foundation, Inc.
 *   59 Temple Place, Suite 330
 *   Boston, MA 02111-1307  USA
 *
 * @author Scott Ferguson
 */

package com.caucho.server.util;

import java.util.*;
import java.util.logging.*;
import java.util.concurrent.*;

import com.caucho.loader.*;
import com.caucho.util.*;
import com.caucho.webbeans.component.WebBeansHandle;

/**
 * A wrapper for Caucho system variables, allowing tests to override
 * the default variables.
 */
public class ScheduledThreadPool
  implements ScheduledExecutorService, EnvironmentListener,
	     java.io.Serializable
{
  private static Logger log
    = Logger.getLogger(ScheduledThreadPool.class.getName());
  private static L10N L = new L10N(ScheduledThreadPool.class);

  private static EnvironmentLocal<ScheduledThreadPool> _local
    = new EnvironmentLocal<ScheduledThreadPool>();

  private ThreadPool _threadPool;

  private boolean _isShutdown;
  private boolean _isTerminated;

  private ClassLoader _loader;

  private final HashSet<Future> _futureSet
    = new HashSet<Future>();

  private ScheduledThreadPool()
  {
    _loader = Thread.currentThread().getContextClassLoader();
    _threadPool = ThreadPool.getThreadPool();

    Environment.addEnvironmentListener(this);
  }

  public static ScheduledThreadPool getLocal()
  {
    synchronized (_local) {
      ScheduledThreadPool pool = _local.getLevel();

      if (pool == null) {
	pool = new ScheduledThreadPool();
	_local.set(pool);
      }

      return pool;
    }
  }
  
  //
  // Executor
  //

  /**
   * Launches a thread to execute a command.
   */
  public void execute(Runnable command)
  {
    if (_isShutdown)
      throw new IllegalStateException("ThreadPool has closed");

    TaskFuture future = new TaskFuture(_loader, command, null);

    synchronized (_futureSet) {
      _futureSet.add(future);
    
      _threadPool.scheduleExecutorTask(future);
    }
  }

  //
  // ExecutorService
  //

  /**
   * Blocks until the tasks complete.
   */
  public boolean awaitTermination(long timeout, TimeUnit unit)
  {
    throw new UnsupportedOperationException();
  }

  /**
   * Invokes a set of tasks.
   */
  public List invokeAll(Collection tasks)
  {
    throw new UnsupportedOperationException();
  }

  /**
   * Invokes a set of tasks.
   */
  public List invokeAll(Collection tasks, long timeout, TimeUnit unit)
  {
    // XXX: todo
    throw new UnsupportedOperationException();
  }

  /**
   * Invokes a set of tasks.
   */
  public Object invokeAny(Collection tasks)
  {
    // XXX: todo
    throw new UnsupportedOperationException();
  }

  /**
   * Invokes a set of tasks.
   */
  public Object invokeAny(Collection tasks,
			  long timeout,
			  TimeUnit unit)
  {
    // XXX: todo
    throw new UnsupportedOperationException();
  }

  /**
   * Return true if the executor is shut down.
   */
  public boolean isShutdown()
  {
    return _isShutdown;
  }

  /**
   * Return true if the executor has completed shutting down.
   */
  public boolean isTerminated()
  {
    return _isTerminated;
  }

  /**
   * Starts the shutdown.
   */
  public void shutdown()
  {
    throw new UnsupportedOperationException();
  }

  /**
   * Starts the shutdown.
   */
  public List<Runnable> shutdownNow()
  {
    throw new UnsupportedOperationException();
  }

  /**
   * Submits a task for execution.
   */
  public <T> Future<T> submit(Callable<T> task)
  {
    if (_isShutdown)
      throw new IllegalStateException("ThreadPool has closed");

    TaskFuture<T> future = new TaskFuture<T>(_loader, task);

    synchronized (_futureSet) {
      _futureSet.add(future);
    
      _threadPool.scheduleExecutorTask(future);
    }

    return future;
  }

  /**
   * Submits a task for execution.
   */
  public Future<?> submit(Runnable command)
  {
    if (_isShutdown)
      throw new IllegalStateException(L.l("Can't submit after ThreadPool has closed"));

    TaskFuture future = new TaskFuture(_loader, command, null);

    synchronized (_futureSet) {
      _futureSet.add(future);
    
      _threadPool.scheduleExecutorTask(future);
    }

    return future;
  }

  /**
   * Submits a task for execution.
   */
  public <T> Future<T> submit(Runnable task, T result)
  {
    if (_isShutdown)
      throw new IllegalStateException(L.l("Can't submit after ThreadPool has closed"));

    TaskFuture<T> future = new TaskFuture<T>(_loader, task, result);

    synchronized (_futureSet) {
      _futureSet.add(future);
    
      _threadPool.scheduleExecutorTask(future);
    }

    return future;
  }

  //
  // ScheduledExecutorService
  //

  /**
   * Schedules a future task.
   */
  public <V> ScheduledFuture<V> schedule(Callable<V> callable,
					 long delay,
					 TimeUnit unit)
  {
    if (_isShutdown)
      throw new IllegalStateException(L.l("Can't submit after ThreadPool has closed"));

    long initialExpires = Alarm.getCurrentTime() + unit.toMillis(delay);
    
    AlarmFuture future = new AlarmFuture(_loader,
					 callable,
					 initialExpires,
					 0,
					 0);

    synchronized (_futureSet) {
      _futureSet.add(future);
    }

    future.queue();

    return future;
  }

  /**
   * Schedules a future task.
   */
  public ScheduledFuture<?> schedule(Runnable command,
				     long delay,
				     TimeUnit unit)
  {
    if (_isShutdown)
      throw new IllegalStateException(L.l("Can't submit after ThreadPool has closed"));

    long initialExpires = Alarm.getCurrentTime() + unit.toMillis(delay);
    
    AlarmFuture future = new AlarmFuture(_loader,
					 command,
					 initialExpires,
					 0,
					 0);

    synchronized (_futureSet) {
      _futureSet.add(future);
    }

    future.queue();

    return future;
  }

  /**
   * Schedules a future task.
   */
  public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
						long initialDelay,
						long period,
						TimeUnit unit)
  {
    if (_isShutdown)
      throw new IllegalStateException(L.l("Can't submit after ThreadPool has closed"));

    long initialExpires = Alarm.getExactTime() + unit.toMillis(initialDelay);
    
    AlarmFuture future = new AlarmFuture(_loader,
					 command,
					 initialExpires,
					 unit.toMillis(period),
					 0);

    synchronized (_futureSet) {
      _futureSet.add(future);
    }

    future.queue();

    return future;
  }

  /**
   * Schedules with fixed delay
   */
  public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
						   long initialDelay,
						   long delay,
						   TimeUnit unit)
  {
    if (_isShutdown)
      throw new IllegalStateException(L.l("Can't submit after ThreadPool has closed"));

    long initialExpires = Alarm.getCurrentTime() + unit.toMillis(initialDelay);
    
    AlarmFuture future = new AlarmFuture(_loader,
					 command,
					 initialExpires,
					 0,
					 unit.toMillis(delay));

    synchronized (_futureSet) {
      _futureSet.add(future);
    }

    future.queue();

    return future;
  }

  //
  // Timer
  //

  /**
   * Returns the Timer for this pool.
   */
  public Timer getTimer()
  {
    throw new UnsupportedOperationException();
  }

  //
  // lifecycle
  //

  /**
   * Stops the pool on environment shutdown.
   */
  private void stop()
  {
    _isShutdown = true;

    while (true) {
      Future future = null;

      synchronized (_futureSet) {
	Iterator<Future> iter = _futureSet.iterator();

	if (iter.hasNext()) {
	  future = iter.next();

	  _futureSet.remove(future);
	}
      }

      if (future == null)
	break;
      
      future.cancel(true);
    }
  }

  void removeFuture(Future future)
  {
    synchronized (_futureSet) {
      _futureSet.remove(future);
    }
  }

  //
  // Environment callbacks.
  //

  /**
   * Called when the environment config phase
   */
  public void environmentConfigure(EnvironmentClassLoader loader)
  {
  }

  /**
   * Called when the environment bind phase
   */
  public void environmentBind(EnvironmentClassLoader loader)
  {
  }

  /**