threadpool.java

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

	  if (startNew) {
	    synchronized (_launcher) {
	      _launcher.notifyAll();
	    }
	    
	    if (poolItem == null) {
	      if (queueIfFull) {
		synchronized (_taskQueue) {
		  _taskQueue.add(task);
		  _loaderQueue.add(loader);
		  _taskQueue.notifyAll();
		}
		
		return false;
	      }

	      if (expireTime < Alarm.getCurrentTime())
		return false;
		  
	      _scheduleWaitCount++;
		
	      try {
		// clear interrupted flag
		Thread.interrupted();
		
		_idleLock.wait(1000);
	      } finally {
		_scheduleWaitCount--;
	      }
	    }
	  }
	}
      } catch (OutOfMemoryError e) {
	try {
	  System.err.println("Exiting due to OutOfMemoryError");
	} finally {
	  System.exit(11);
	}
      } catch (Throwable e) {
	e.printStackTrace();
      }
    }

    poolItem.start(task, loader);

    return true;
  }

  private void init()
  {
    if (_threadMax < 0)
      _threadMax = DEFAULT_THREAD_MAX;

    if (_threadIdleMin < 0 && _threadIdleMax < 0) {
      _threadIdleMin = DEFAULT_THREAD_IDLE_MIN;
      _threadIdleMax = _threadIdleMin + DEFAULT_THREAD_IDLE_GAP;
    }
    else if (_threadIdleMax < 0) {
      _threadIdleMax = _threadIdleMin + DEFAULT_THREAD_IDLE_GAP;
    }
    else if (_threadIdleMin < 0) {
      _threadIdleMin = DEFAULT_THREAD_IDLE_MIN;

      if (_threadIdleMax < _threadIdleMin)
	_threadIdleMin = 1;
    }

    calculateThreadPriority();
  }

  private void calculateThreadPriority()
  {
    if (_threadPrioritySet) {
    }
    else if (_threadIdleMin <= 0)
      _threadPriority = 0;
    else if (_threadIdleMin <= 2)
      _threadPriority = _threadIdleMin;
    else
      _threadPriority = (_threadIdleMin + 1) / 2;
  }

  public String toString()
  {
    return getClass().getSimpleName() + "[]";
  }

  class Item implements Runnable {
    final int _id;
    final String _name;

    Thread _thread;
    Thread _queueThread;

    Item _prev;
    Item _next;
    boolean _isIdle;

    long _threadResetCount;
  
    Runnable _task;
    ClassLoader _classLoader;

    private Item()
    {
      synchronized (Item.class) {
	_id = _g_id++;
	_name = "resin-" + _id;
      }
    }

    /**
     * Returns the id.
     */
    int getId()
    {
      return _id;
    }

    /**
     * Returns the name.
     */
    public String getName()
    {
      return _name;
    }

    /**
     * Returns the thread id.
     */
    public long getThreadId()
    {
      return _thread.getId();
    }

    /**
     * Returns the thread.
     */
    Thread getThread()
    {
      return _thread;
    }

    /**
     * Starts the thread.
     */
    private boolean start(Runnable task, ClassLoader loader)
    {
      synchronized (this) {
	_task = task;
	_classLoader = loader;

	notifyAll();
      }

      return true;
    }

    /**
     * The main thread execution method.
     */
    public void run()
    {
      _thread = Thread.currentThread();

      synchronized (_idleLock) {
	_threadCount++;
	_startCount--;
	_threads.add(this);

	if (_startCount < 0) {
	  _startCount = 0;
	}
      }
      
      try {
	runTasks();
      } finally {
	synchronized (_idleLock) {
	  _threadCount--;

	  _threads.remove(this);
	}

	if (_threadCount < _threadIdleMin) {
	  synchronized (_launcher) {
	    _launcher.notifyAll();
	  }
	}
      }
    }

    private void runTasks()
    {
      _threadResetCount = _resetCount;
    
      Thread thread = Thread.currentThread();
      ClassLoader systemClassLoader = ClassLoader.getSystemClassLoader();
      boolean isIdle = false;

      while (true) {
	try {
	  // put the thread into the idle ring
	  if (! isIdle) {
	    _isQueuePriority = true;
	  
	    isIdle = true;
	  
	    synchronized (_idleLock) {
	      if (_threadIdleMax < _idleCount) {
		return;
	      }
	      
	      _next = _idleHead;
	      _prev = null;
	      _isIdle = true;

	      if (_idleHead != null)
		_idleHead._prev = this;

	      _idleHead = this;
	      _idleCount++;

	      if (_scheduleWaitCount > 0)
		_idleLock.notifyAll();
	    }
	  }

	  Runnable task = null;
	  ClassLoader classLoader = null;

	  // clear interrupted flag
	  Thread.interrupted();
	
	  // wait for the next available task
	  synchronized (this) {
	    if (_task == null) {
	      thread.setContextClassLoader(systemClassLoader);
	      wait(60000L);
	    }

	    task = _task;
	    _task = null;

	    classLoader = _classLoader;
	    _classLoader = null;
	  }

	  // if the task is available, run it in the proper context
	  if (task != null) {
	    isIdle = false;

	    thread.setContextClassLoader(classLoader);
	    try {
	      task.run();
	    } catch (Throwable e) {
	      log.log(Level.WARNING, e.toString(), e);
	    } finally {
	      thread.setContextClassLoader(ClassLoader.getSystemClassLoader());
	    }
	  }
	  else {
	    boolean isDead = false;
	    boolean isReset = false;

	    // check to see if we're over the idle thread limit
	    synchronized (_idleLock) {
	      if (_isIdle &&
		  (_threadIdleMax < _idleCount ||
		   _resetCount != _threadResetCount)) {
		isDead = true;

		isReset = _resetCount != _threadResetCount;
	      
		Item next = _next;
		Item prev = _prev;

		_next = null;
		_prev = null;
		_isIdle = false;

		if (next != null)
		  next._prev = prev;

		if (prev != null)
		  prev._next = next;
		else
		  _idleHead = next;

		_idleCount--;
	      }
	    }

	    if (isReset) {
	      synchronized (_launcher) {
		_launcher.notifyAll();
	      }
	    }
	  
	    if (isDead)
	      return;
	  }
	} catch (Throwable e) {
	}
      }
    }
  }

  class OverflowItem implements Runnable {
    Runnable _task;
    ClassLoader _loader;

    OverflowItem(Runnable task)
    {
      _task = task;
      _loader = Thread.currentThread().getContextClassLoader();
    }

    void start()
    {
      Thread thread = new Thread(this, _task.getClass().getSimpleName() + "-Overflow");
      thread.setDaemon(true);
      thread.start();
    }

    /**
     * The main thread execution method.
     */
    public void run()
    {
      Thread thread = Thread.currentThread();
      thread.setContextClassLoader(_loader);

      _task.run();
    }
  }

  class ThreadLauncher implements Runnable {
    private ThreadLauncher()
    {
      Thread thread = new Thread(this);
      thread.setName("resin-thread-launcher");
      thread.setDaemon(true);

      thread.start();
    }

    /**
     * Starts a new connection
     */
    private boolean startConnection(long waitTime)
      throws InterruptedException
    {
      boolean doStart = true;
      
      synchronized (_idleLock) {
	int idleCount = _idleCount;

	if (_threadMax < _threadCount + _startCount)
	  doStart = false;
	else if (_threadIdleMin < idleCount + _startCount)
	  doStart = false;

	if (doStart)
	  _startCount++;
      }

      if (doStart) {
	try {
	  Item poolItem = new Item();
    
	  Thread thread = new Thread(poolItem, poolItem.getName());
	  thread.setDaemon(true);

	  thread.start();
	} catch (Throwable e) {
	  _startCount--;

	  e.printStackTrace();
	  if (_startCount < 0) {
	    Thread.dumpStack();
	    _startCount = 0;
	  }
	}

	// Thread.yield();
      }
      else {
	Thread.interrupted();
	synchronized (this) {
	  wait(waitTime);
	  return false;
	}
      }

      return true;
    }
    
    public void run()
    {
      ClassLoader systemLoader = ClassLoader.getSystemClassLoader();
      
      Thread.currentThread().setContextClassLoader(systemLoader);

      try {
	for (int i = 0; i < _threadIdleMin; i++)
	  startConnection(0);
      } catch (Throwable e) {
	e.printStackTrace();
      }
      
      while (true) {
	try {
	  startConnection(10000);

	  //Thread.currentThread().sleep(5);
	  Thread.currentThread().yield();
	} catch (OutOfMemoryError e) {
	  System.exit(10);
	} catch (Throwable e) {
	  e.printStackTrace();
	}
      }
    }
  }

  class ScheduleThread implements Runnable {
    private ScheduleThread()
    {
      Thread thread = new Thread(this);
      thread.setName("resin-thread-scheduler");
      thread.setDaemon(true);

      thread.start();
    }
    
    public void run()
    {
      ClassLoader systemLoader = ClassLoader.getSystemClassLoader();

      Thread thread = Thread.currentThread();
      thread.setContextClassLoader(systemLoader);
      
      while (true) {
	try {
	  Runnable task = null;
	  ClassLoader loader = null;

	  Thread.interrupted();
	  
	  synchronized (_taskQueue) {
	    if (_taskQueue.size() > 0) {
	      task = _taskQueue.remove(0);
	      loader = _loaderQueue.remove(0);
	    }
	    else {
	      try {
		_taskQueue.wait(60000);
	      } catch (Throwable e) {
		thread.interrupted();
		log.finer(e.toString());
	      }
	    }
	  }

	  if (task != null) {
	    schedule(task, loader, _threadIdleMin, MAX_EXPIRE, false);
	  }
	} catch (OutOfMemoryError e) {
	  System.exit(10);
	} catch (Throwable e) {
	  e.printStackTrace();
	}
      }
    }
  }

  static class ExecutorQueueItem {
    Runnable _runnable;
    ClassLoader _loader;

    ExecutorQueueItem _next;

    ExecutorQueueItem(Runnable runnable, ClassLoader loader)
    {
      _runnable = runnable;
      _loader = loader;
    }

    Runnable getRunnable()
    {
      return _runnable;
    }

    ClassLoader getLoader()
    {
      return _loader;
    }
  }
}