executor.java

来自「opennms得相关源码 请大家看看」· Java 代码 · 共 614 行 · 第 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/
//
// Tab Size = 8
//

package org.opennms.netmgt.actiond;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

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

/**
 * This class is used as a thread for launching and executing actions as they
 * are discovered by the action daemon. The actions are read from an execution
 * queue and the processes are created by the fiber. Each created process is
 * added to garbage collection list that is periodically polled and culled based
 * upon the status of the process or how long the process is run. If the process
 * has run long than allocated it is terminated during collection.
 * 
 * @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>
 * 
 */
final class Executor implements Runnable, PausableFiber {
    /**
     * The input queue of runnable commands.
     */
    private FifoQueue m_execQ;

    /**
     * The list of outstanding commands.
     */
    private List m_processes;

    /**
     * The maximum time that a command can execute.
     */
    private long m_maxWait;

    /**
     * The maximum number of outstanding processes.
     */
    private int m_maxProcCount;

    /**
     * The process garbage collection thread.
     */
    private Thread m_reaper;

    /**
     * The garbage collection instance.
     */
    private Runnable m_reaperRun;

    /**
     * The worker thread that executes the <code>run</code> method.
     */
    private Thread m_worker;

    /**
     * The name of this Fiber
     */
    private String m_name;

    /**
     * The status of this fiber.
     */
    private int m_status;

    /**
     * This class is designed to encapsulated a process and its start time. The
     * start time is based upon the system clock and the runtime is the
     * difference between the current time and the started time.
     * 
     * @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>
     * 
     */
    private static final class DatedProc {
        /**
         * The executable running
         */
        private final String m_cmd;

        /**
         * The process returned from the {@link java.lang.Runtime Runtime}
         * instance.
         */
        private final Process m_proc;

        /**
         * The time the process was started.
         */
        private final long m_started;

        /**
         * Constructs a new dated process.
         * 
         * @param cmd
         *            The command used to start the process.
         * @param p
         *            The running process.
         */
        DatedProc(String cmd, Process p) {
            m_cmd = cmd;
            m_proc = p;
            m_started = System.currentTimeMillis();
        }

        /**
         * Returns the encapsulated process.
         * 
         */
        Process getProcess() {
            return m_proc;
        }

        /**
         * Returns the current runtime of the process.
         */
        long getRunTime() {
            return System.currentTimeMillis() - m_started;
        }

        /**
         * Returns the command being run by the dated process.
         */
        public String toString() {
            return m_cmd;
        }
    } // end class DatedProc

    /**
     * This class encapsules a singular run method that is used to
     * <em>garbage collect</em> expired processes. If a process has exceeded
     * its maximum runtime then it is killed and removed from the process queue.
     * 
     * @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>
     * 
     */
    private final class Reaper implements Runnable {
        /**
         * The reaper execution enviroment. This method scans the process array
         * and removes expired and completed commands from the array on a
         * periodic basis. In that respect it is a garbage collection thread for
         * processes.
         * 
         */
        public void run() {
            // Wait for a maximum of 15 seconds between checks!
            //
            long waitPeriod = m_maxWait / 5;
            if (waitPeriod > 15000) {
                waitPeriod = 15000;
            }

            Category log = ThreadCategory.getInstance(Executor.class);

            // Begin the checking process.
            //
            // Make sure to leave the 'this' keyword associated with the
            // getClass() call or jikes will complain. The 'this' keyword
            // removes all ambiguity in the call.
            //
            for (;;) {
                // run and check the queue once about
                // 1/5 of the maximum run time.
                //
                synchronized (m_processes) {
                    Iterator i = m_processes.iterator();
                    while (i.hasNext()) {
                        DatedProc dp = (DatedProc) i.next();
                        try {
                            int rc = dp.getProcess().exitValue();

                            if (log.isDebugEnabled()) {
                                log.debug("Process " + dp + " completed, rc = " + rc);
                            }

                            i.remove();
                            continue;
                        } catch (IllegalThreadStateException ex) {
                        } // still running

                        if (dp.getRunTime() > m_maxWait) {
                            if (log.isInfoEnabled())
                                log.info("Process " + dp + " did not complete in the alloted time, terminating.");

                            dp.getProcess().destroy();
                            i.remove();
                        }
                    }
                }

                synchronized (this) {
                    // the 'this' keyword should not be removed
                    // or else jikes will complain about an ambiguous
                    // call.
                    this.notifyAll();

                    // sleep for 1/5 of wait time or
                    // 15 seconds, which ever is smaller.
                    //
                    try {
                        // the 'this' keyword should not be removed
                        // or else jikes will complain about an ambiguous
                        // call.
                        this.wait(waitPeriod);
                    } catch (InterruptedException ex) {
                        // this is used as a shutdown mechinism
                        break;
                    }
                }

            } // end for(;;)

        } // end run

    } // end class Reaper

    /**
     * <p>
     * Converts a single command to an array that can be passed to the
     * {@link java.lang.Runtime#exec(java.lang.String[]) exec}system call. The
     * element at index zero of the array is the name of the executable to run.
     * Indexs [1..length) are the arguments passed to the executable command.
     * </p>
     * 
     * <p>
     * The input command has is white space trimmed before processing. The basic
     * processing is to split on spaces, except when a double quote or single
     * quote is encountered. Also backspaces(\) should also be handled correctly
     * both in and out of the quotes. Shell escapes with <em>$</em> are not
     * supported.
     * </p>
     * 
     * @param cmd
     *            The command to split into an array.
     * 
     * @return The execution array.
     * 
     */
    private static String[] getExecArguments(String cmd) {
        Category log = ThreadCategory.getInstance(Executor.class);

        // make sure we get rid of excess white space.
        //
        cmd = cmd.trim();

        // get the processing elements.
        //
        StringBuffer buf = new StringBuffer();
        List args = new ArrayList(5);
        char[] chars = cmd.toCharArray();

        boolean dquoted = false;
        boolean squoted = false;
        for (int x = 0; x < chars.length; x++) {
            if (chars[x] == '\\') {
                if (squoted) {
                    buf.append(chars[x]).append(chars[x + 1]);