scheduler.java

来自「opennms得相关源码 请大家看看」· Java 代码 · 共 545 行 · 第 1/2 页

//
// This file is part of the OpenNMS(R) Application.
//
// OpenNMS(R) is Copyright (C) 2002-2003 The OpenNMS Group, Inc.  All rights reserved.
// OpenNMS(R) is a derivative work, containing both original code, included code and modified
// code that was published under the GNU General Public License. Copyrights for modified 
// and included code are below.
//
// OpenNMS(R) is a registered trademark of The OpenNMS Group, Inc.
//
// Modifications:
//
// 2003 Jan 31: Cleaned up some unused imports.
//
// Original code base Copyright (C) 1999-2001 Oculan Corp.  All rights reserved.
//
// This program 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.
//
// This program 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.  See the
// GNU General Public License for more details.                                                            
//
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
//       
// For more information contact: 
//      OpenNMS Licensing       <[e-mail]>
//      http://www.opennms.org/
//      http://www.opennms.com/
//

package org.opennms.netmgt.scheduler;

import java.lang.reflect.UndeclaredThrowableException;
import java.util.Collections;
import java.util.Iterator;
import java.util.Map;
import java.util.TreeMap;

import org.apache.log4j.Category;
import org.opennms.core.concurrent.RunnableConsumerThreadPool;
import org.opennms.core.fiber.PausableFiber;
import org.opennms.core.queue.FifoQueue;
import org.opennms.core.queue.FifoQueueException;
import org.opennms.core.queue.FifoQueueImpl;
import org.opennms.core.utils.ThreadCategory;

/**
 * This class implements a simple scheduler to ensure the polling occurs at the
 * expected intervals. The scheduler employees a dynamic thread pool that adjust
 * to the load until a maximum thread count is reached.
 * 
 * @author <a href="mailto:[e-mail]">Mike Davidson </a>
 * @author <a href="mailto:[e-mail]">Brian Weaver </a>
 * @author <a href="http://www.opennms.org/">OpenNMS </a>
 * 
 */
public class Scheduler implements Runnable, PausableFiber, ScheduleTimer {
    /**
     * The map of queue that contain {@link ReadyRunnable ready runnable}
     * instances. The queues are mapped according to the interval of scheduling.
     */
    public Map m_queues;

    /**
     * The total number of elements currently scheduled. This should be the sum
     * of all the elements in the various queues.
     */
    public int m_scheduled;

    /**
     * The pool of threads that are used to executed the runnable instances
     * scheduled by the class' instance.
     */
    public RunnableConsumerThreadPool m_runner;

    /**
     * The name of this fiber.
     */
    private String m_name;

    /**
     * The status for this fiber.
     */
    public int m_status;

    /**
     * The worker thread that executes this instance.
     */
    private Thread m_worker;

    /**
     * This queue extends the standard FIFO queue instance so that it is
     * possible to peek at an instance without removing it from the queue.
     * 
     */
    public static final class PeekableFifoQueue extends FifoQueueImpl {
        /**
         * The object hold. This holds the last object peeked at by the
         * application.
         */
        private Object m_hold;

        /**
         * Default constructor.
         */
        PeekableFifoQueue() {
            m_hold = null;
        }

        /**
         * This method allows the caller to peek at the next object that would
         * be returned on a <code>remove</code> call. If the queue is
         * currently empty then the caller is blocked until an object is put
         * into the queue.
         * 
         * @return The object that would be returned on the next call to
         *         <code>remove</code>.
         * 
         * @throws java.lang.InterruptedException
         *             Thrown if the thread is interrupted.
         * @throws org.opennms.core.queue.FifoQueueException
         *             Thrown if an error occurs removing an item from the
         *             queue.
         */
        public synchronized Object peek() throws InterruptedException, FifoQueueException {
            if (m_hold == null)
                m_hold = super.remove(1L);

            return m_hold;
        }

        /**
         * Removes the next element from the queue and returns it to the caller.
         * If there is no objects available then the caller is blocked until an
         * item is available.
         * 
         * @return The next element in the queue.
         * 
         * @throws java.lang.InterruptedException
         *             Thrown if the thread is interrupted.
         * @throws org.opennms.core.queue.FifoQueueException
         *             Thrown if an error occurs removing an item from the
         *             queue.
         */
        public synchronized Object remove() throws InterruptedException, FifoQueueException {
            Object rval = null;
            if (m_hold != null) {
                rval = m_hold;
                m_hold = null;
            } else {
                rval = super.remove();
            }

            return rval;
        }

        /**
         * Removes the next element from the queue and returns it to the caller.
         * If there is no objects available then the caller is blocked until an
         * item is available. If an object is not available within the time
         * frame specified by <code>timeout</code>.
         * 
         * @param timeout
         *            The maximum time to wait.
         * 
         * @return The next element in the queue.
         * 
         * @throws java.lang.InterruptedException
         *             Thrown if the thread is interrupted.
         * @throws org.opennms.core.queue.FifoQueueException
         *             Thrown if an error occurs removing an item from the
         *             queue.
         */
        public synchronized Object remove(long timeout) throws InterruptedException, FifoQueueException {
            Object rval = null;
            if (m_hold != null) {
                rval = m_hold;
                m_hold = null;
            } else {
                rval = super.remove(timeout);
            }

            return rval;
        }
    }

    /**
     * Constructs a new instance of the scheduler. The maximum number of
     * executable threads is specified in the constructor. The executable
     * threads are part of a runnable thread pool where the scheduled runnables
     * are executed.
     * 
     * @param parent
     *            String prepended to "Scheduler" to create fiber name
     * @param maxSize
     *            The maximum size of the thread pool.
     * 
     */
    public Scheduler(String parent, int maxSize) {
        m_name = parent + "Scheduler-" + maxSize;
        m_status = START_PENDING;
        m_runner = new RunnableConsumerThreadPool(m_name + " Pool", 0.6f, 1.0f, maxSize);
        m_queues = Collections.synchronizedMap(new TreeMap());
        m_scheduled = 0;
        m_worker = null;
    }

    /**
     * Constructs a new instance of the scheduler. The maximum number of
     * executable threads is specified in the constructor. The executable
     * threads are part of a runnable thread pool where the scheduled runnables
     * are executed.
     * 
     * @param parent
     *            String prepended to "Scheduler" to create fiber name
     * @param maxSize
     *            The maximum size of the thread pool.
     * @param lowMark
     *            The low water mark ratios of thread size to threads when
     *            threads are stopped.
     * @param hiMark
     *            The high water mark ratio of thread size to threads when
     *            threads are started.
     * 
     */
    public Scheduler(String parent, int maxSize, float lowMark, float hiMark) {
        m_name = parent + "Scheduler-" + maxSize;
        m_status = START_PENDING;
        m_runner = new RunnableConsumerThreadPool(m_name + " Pool", lowMark, hiMark, maxSize);
        m_queues = Collections.synchronizedMap(new TreeMap());
        m_scheduled = 0;
        m_worker = null;
    }

    /**
     * This method is used to schedule a ready runnable in the system. The
     * interval is used as the key for determining which queue to add the
     * runnable.
     * 
     * @param runnable
     *            The element to run when interval expires.
     * @param interval
     *            The queue to add the runnable to.
     * 
     * @throws java.lang.RuntimeException
     *             Thrown if an error occurs adding the element to the queue.
     */
    public synchronized void schedule(ReadyRunnable runnable, long interval) {
        Category log = ThreadCategory.getInstance(getClass());

        if (log.isDebugEnabled()) {
            log.debug("schedule: Adding ready runnable "+runnable+" at interval " + interval);
        }

        Long key = new Long(interval);
        if (!m_queues.containsKey(key)) {
            if (log.isDebugEnabled())
                log.debug("schedule: interval queue did not exist, a new one has been created");
            m_queues.put(key, new PeekableFifoQueue());
        }

        try {
            ((FifoQueue) m_queues.get(key)).add(runnable);
            if (m_scheduled++ == 0) {
                if (log.isDebugEnabled())
                    log.debug("schedule: queue element added, calling notify all since none were scheduled");
                notifyAll();