os/dtrace/LocalConsumer.java

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License (the "License").
 * You may not use this file except in compliance with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */

/*
 * Copyright 2006 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 *
 * ident	"%Z%%M%	%I%	%E% SMI"
 */
package org.opensolaris.os.dtrace;

import java.io.*;
import java.util.*;
import java.net.InetAddress;
import java.net.UnknownHostException;
import javax.swing.event.EventListenerList;
import java.util.logging.*;

/**
 * Interface to the native DTrace library, each instance is a single
 * DTrace consumer.
 *
 * @author Tom Erickson
 */
public class LocalConsumer implements Consumer {
    //
    // Implementation notes:
    //
    // libdtrace is *not* thread-safe.  You cannot make multiple calls
    // into it simultaneously from different threads, even if those
    // threads are operating on different dtrace_hdl_t's.  Calls to
    // libdtrace are synchronized on a global lock, LocalConsumer.class.

    static Logger logger = Logger.getLogger(LocalConsumer.class.getName());

    private static final int DTRACE_JNI_VERSION = 1;

    private static final Option[] DEFAULT_OPTIONS = new Option[] {
	new Option(Option.bufsize, Option.kb(256)),
	new Option(Option.aggsize, Option.kb(256)),
    };

    private static native void _loadJniTable();

    // Undocumented configuration options
    private static boolean debug;
    private static int maxConsumers;

    static {
	LocalConsumer.configureLogging();
	// Undocumented configuration options settable using
	// java -Doption=value
	LocalConsumer.getConfigurationOptions();

	Utility.loadLibrary("libdtrace_jni.so.1", debug);

	_checkVersion(DTRACE_JNI_VERSION);
	_setDebug(debug);
	if (maxConsumers > 0) {
	    _setMaximumConsumers(maxConsumers);
	}

	//
	// Last of all in case configuration options affect the loading
	// of the JNI table.
	//
	_loadJniTable();
    }

    // Native JNI interface (see lib/libdtrace_jni/dtrace_jni.c)
    private static native void _checkVersion(int version);
    private native void _open(OpenFlag[] flags) throws DTraceException;
    private native Program _compileString(String program, String[] args)
	    throws DTraceException;
    private native Program.File _compileFile(String path, String[] args)
	    throws DTraceException;
    private native void _exec(Program program) throws DTraceException;
    private native void _getProgramInfo(Program program)
	    throws DTraceException;
    private native void _setOption(String option, String value)
	    throws DTraceException;
    private native long _getOption(String option) throws DTraceException;
    private native boolean _isEnabled();
    private native void _checkProgramEnabling();
    private native void _go() throws DTraceException;
    private native void _stop() throws DTraceException;
    private native void _consume() throws DTraceException;
    private native void _interrupt();
    private native void _close();
    private native Aggregate _getAggregate(AggregateSpec spec)
	    throws DTraceException;
    private native int _createProcess(String cmd) throws DTraceException;
    private native void _grabProcess(int pid) throws DTraceException;
    private native void _listProbes(List <ProbeDescription> probeList,
	    ProbeDescription filter);
    private native void _listProbeDetail(List <Probe> probeList,
	    ProbeDescription filter);
    private native void _listCompiledProbes(
	    List <ProbeDescription> probeList, Program program);
    private native void _listCompiledProbeDetail(
	    List <Probe> probeList, Program program);
    private static native String _getVersion();
    private static native int _openCount();
    //
    // Releases memory held in the JNI layer after dtrace_close() has
    // released critical system resources like file descriptors, and
    // calls to libdtrace are no longer needed (or possible).
    //
    private native void _destroy();
    // Called by LogDistribution
    static native long _quantizeBucket(int i);
    //
    // Cannot be static because the necessary dtrace handle is specific
    // to this Consumer.
    //
    private native String _lookupKernelFunction(Number address);
    private native String _lookupUserFunction(int pid, Number address);
    private static native String _getExecutableName();

    // Undocumented configuration options
    private static native void _setMaximumConsumers(int max);
    private static native void _setDebug(boolean debug);

    protected EventListenerList listenerList;
    protected ExceptionHandler exceptionHandler;

    private int _handle = -1;    // native C identifier (do not modify)
    private final Identifier id; // java identifier

    private enum State {
	INIT,
	OPEN,
	COMPILED,
	GO,
	STARTED,
	STOPPED,
	CLOSED
    }

    private State state = State.INIT;
    private boolean stopCalled;
    private boolean abortCalled;

    //
    // Per-consumer lock used in native code to prevent conflict between
    // the native consumer loop and the getAggregate() thread without
    // locking this LocalConsumer.  A distinct per-consumer lock allows
    // the stop() method to be synchronized without causing deadlock
    // when the consumer loop grabs the per-consumer lock before
    // dtrace_work().
    //
    private Object consumerLock;

    //
    // stopLock is a synchronization lock used to ensure that the stop()
    // method does not return until this consumer has actually stopped.
    // Correct lock ordering is needed to ensure that listeners cannot
    // deadlock this consumer:
    // 1. stop() grabs the lock on this consumer before determining if
    //    this consumer is running (to ensure valid state).
    // 2. Once stop() determines that this consumer is actually running,
    //    it releases the lock on this consumer.  Failing to release the
    //    lock makes it possible for a ConsumerListener to deadlock this
    //    consumer by calling any synchronized LocalConcumer method
    //    (because the listener called by the worker thread prevents the
    //    worker thread from finishing while it waits for stop() to
    //    release the lock, which it will never do until the worker
    //    thread finishes).
    // 3. stop() interrupts this consumer and grabs the stopLock, then
    //    waits on the stopLock for this consumer to stop (i.e. for the
    //    worker thread to finish).
    // 4. The interrupted worker thread grabs the stopLock when it
    //    finishes so it can notify waiters on the stopLock (in this
    //    case the stop() method) that the worker thread is finished.
    //    The workEnded flag (whose access is protected by the
    //    stopLock), is used in case the interrupted worker thread
    //    finishes and grabs the stopLock before the stop() method does.
    //    Setting the flag in that case tells the stop() method it has
    //    nothing to wait for (otherwise stop() would wait forever,
    //    since there is no one left after the worker thread finishes to
    //    notify the stop() method to stop waiting).
    // 5. The worker thread updates the state member to STOPPED and
    //    notifies listeners while it holds the stopLock and before it
    //    notifies waiters on the stopLock.  This is to ensure that
    //    state has been updated to STOPPED and that listeners have
    //    executed consumerStopped() before the stop() method returns,
    //    to ensure valid state and in case the caller of stop() is
    //    relying on anything having been done by consumerStopped()
    //    before it proceeds to the next statement.
    // 6. The worker thread notifies waiters on the stopLock before
    //    releasing it.  stop() returns.
    //
    private Object stopLock;
    private boolean workEnded;

    private static int sequence = 0;

    private static void
    configureLogging()
    {
	logger.setUseParentHandlers(false);
	Handler handler = new ConsoleHandler();
	handler.setLevel(Level.ALL);
	logger.addHandler(handler);
        logger.setLevel(Level.OFF);
    }

    private static Integer
    getIntegerProperty(String name)
    {
	Integer value = null;
	String property = System.getProperty(name);
	if (property != null && property.length() != 0) {
	    try {
		value = Integer.parseInt(property);
		System.out.println(name + "=" + value);
	    } catch (NumberFormatException e) {
		System.err.println("Warning: property ignored: " +
			name + "=" + property);
	    }
	}
	return value;
    }

    private static void
    getConfigurationOptions()
    {
	Integer property;
	property = getIntegerProperty("JAVA_DTRACE_API_DEBUG");
	if (property != null) {
	    debug = (property != 0);
	}
	property = getIntegerProperty("JAVA_DTRACE_MAX_CONSUMERS");
	if (property != null) {
	    maxConsumers = property;
	}
    }

    /**
     * Creates a consumer that interacts with the native DTrace library
     * on the local system.
     */
    public
    LocalConsumer()
    {
	id = new LocalConsumer.Identifier(this);
	consumerLock = new Object();
	stopLock = new Object();
	listenerList = new EventListenerList();
    }

    /**
     * Called by native C code only
     */
    private int
    getHandle()
    {
	return _handle;
    }

    /**
     * Called by native C code only
     */
    private void
    setHandle(int n)
    {
	_handle = n;
    }

    public synchronized void
    open(OpenFlag ... flags) throws DTraceException
    {
	if (state == State.CLOSED) {
	    throw new IllegalStateException("cannot reopen a closed consumer");
	}
	if (state != State.INIT) {
	    throw new IllegalStateException("consumer already open");
	}

	for (OpenFlag flag : flags) {
	    if (flag == null) {
		throw new NullPointerException("open flag is null");
	    }
	}

	synchronized (LocalConsumer.class) {
	    _open(flags);
	}

	state = State.OPEN;
	setOptions(DEFAULT_OPTIONS);

	if (abortCalled) {
	    _interrupt();
	}

	if (logger.isLoggable(Level.INFO)) {
	    logger.info("consumer table count: " + _openCount());
	}
    }

    private synchronized void
    checkCompile()
    {
	switch (state) {
	    case INIT:
		throw new IllegalStateException("consumer not open");
	    case OPEN:
	    case COMPILED: // caller may compile more than one program
		break;
	    case GO:
	    case STARTED:
		throw new IllegalStateException("go() already called");
	    case STOPPED:
		throw new IllegalStateException("consumer stopped");
	    case CLOSED:
		throw new IllegalStateException("consumer closed");
	}
    }

    public synchronized Program
    compile(String program, String ... macroArgs) throws DTraceException
    {
	if (program == null) {
	    throw new NullPointerException("program string is null");
	}
	checkCompile();
	Program p = null;

	String[] argv = null;
	if (macroArgs != null) {
	    for (String macroArg : macroArgs) {
		if (macroArg == null) {
		    throw new NullPointerException("macro argument is null");
		}
	    }
	    argv = new String[macroArgs.length + 1];
	    synchronized (LocalConsumer.class) {
		//
		// Could be an application with an embedded JVM, not
		// necessarily "java".
		//
		argv[0] = _getExecutableName();
	    }
	    System.arraycopy(macroArgs, 0, argv, 1, macroArgs.length);
	} else {
	    synchronized (LocalConsumer.class) {
		argv = new String[] { _getExecutableName() };
	    }
	}
	synchronized (LocalConsumer.class) {
	    p = _compileString(program, argv);
	}
	p.consumerID = id;
	p.contents = program;
	p.validate();
	state = State.COMPILED;

	return p;
    }

    public synchronized Program
    compile(File program, String ... macroArgs) throws DTraceException,
            IOException, SecurityException
    {
	if (program == null) {
	    throw new NullPointerException("program file is null");
	}
	if (!program.canRead()) {
	    throw new FileNotFoundException("failed to open " +
		    program.getName());
	}
	checkCompile();
	Program.File p = null;

	String[] argv = null;
	if (macroArgs != null) {
	    for (String macroArg : macroArgs) {
		if (macroArg == null) {
		    throw new NullPointerException("macro argument is null");
		}
	    }
	    argv = new String[macroArgs.length + 1];
	    argv[0] = program.getPath();
	    System.arraycopy(macroArgs, 0, argv, 1, macroArgs.length);
	} else {
	    macroArgs = new String[] { program.getPath() };
	}
	synchronized (LocalConsumer.class) {
	    p = _compileFile(program.getPath(), argv);
	}
	p.consumerID = id;
	p.contents = Program.getProgramString(program);
	p.file = program;
	p.validate();
	state = State.COMPILED;

	return p;
    }

    private synchronized void
    checkProgram(Program program)
    {
	if (program == null) {
	    throw new NullPointerException("program is null");
	}
	if (!id.equals(program.consumerID)) {
	    throw new IllegalArgumentException("program not compiled " +
		    "by this consumer");
	}
    }

    public void
    enable() throws DTraceException
    {
	enable(null);
    }

    public synchronized void
    enable(Program program) throws DTraceException
    {
	switch (state) {
	    case INIT:
		throw new IllegalStateException("consumer not open");
	    case OPEN:
		throw new IllegalStateException("no compiled program");
	    case COMPILED:
		break;
	    case GO:
	    case STARTED:
		throw new IllegalStateException("go() already called");
	    case STOPPED:
		throw new IllegalStateException("consumer stopped");
	    case CLOSED:
		throw new IllegalStateException("consumer closed");
	}

	// Compile all programs if null
	if (program != null) {
	    checkProgram(program);
	}

	//
	// Left to native code to throw IllegalArgumentException if the
	// program is already enabled, since only the native code knows
	// the enabled state.
	//
	synchronized (LocalConsumer.class) {
	    _exec(program);
	}
    }

    public synchronized void
    getProgramInfo(Program program) throws DTraceException
    {
	checkProgram(program);
	if (state == State.CLOSED) {
	    throw new IllegalStateException("consumer closed");
	}

	//
	// The given program was compiled by this consumer, so we can
	// assert the following:
	//
	assert ((state != State.INIT) && (state != State.OPEN));

	synchronized (LocalConsumer.class) {
	    _getProgramInfo(program);
	}
    }

    private void
    setOptions(Option[] options) throws DTraceException
    {
	for (Option o : options) {
	    setOption(o.getName(), o.getValue());
	}
    }

    public void
    setOption(String option) throws DTraceException
    {
	setOption(option, Option.VALUE_SET);
    }

    public void
    unsetOption(String option) throws DTraceException
    {
	setOption(option, Option.VALUE_UNSET);
    }

    public synchronized void
    setOption(String option, String value) throws DTraceException
    {
	if (option == null) {
	    throw new NullPointerException("option is null");
	}
	if (value == null) {
	    throw new NullPointerException("option value is null");
	}

	switch (state) {
	    case INIT:
		throw new IllegalStateException("consumer not open");
	    case OPEN:
	    case COMPILED:
	    case GO:
	    case STARTED: // Some options can be set on a running consumer
	    case STOPPED: // Allowed (may affect getAggregate())
		break;
	    case CLOSED:
		throw new IllegalStateException("consumer closed");
	}

	synchronized (LocalConsumer.class) {
	    _setOption(option, value);
	}
    }

    public synchronized long
    getOption(String option) throws DTraceException
    {
	if (option == null) {
	    throw new NullPointerException("option is null");
	}

	switch (state) {
	    case INIT:
		throw new IllegalStateException("consumer not open");
	    case OPEN:
	    case COMPILED:
	    case GO:
	    case STARTED:
	    case STOPPED:
		break;
	    case CLOSED:
		throw new IllegalStateException("consumer closed");
	}

	long value;
	synchronized (LocalConsumer.class) {
	    value = _getOption(option);
	}
	return value;
    }

    public final synchronized boolean
    isOpen()
    {
	return ((state != State.INIT) && (state != State.CLOSED));
    }

    public final synchronized boolean
    isEnabled()
    {
	if (state != State.COMPILED) {
	    return false;
	}

	return _isEnabled();
    }

    public final synchronized boolean
    isRunning()
    {
	return (state == State.STARTED);
    }

    public final synchronized boolean
    isClosed()
    {
	return (state == State.CLOSED);
    }

    /**
     * Called in the runnable target of the thread returned by {@link
     * #createThread()} to run this DTrace consumer.
     *
     * @see #createThread()
     */
    protected final void
    work()
    {
	try {
	    synchronized (this) {
		if (state != State.GO) {
		    //
		    // stop() was called after go() but before the
		    // consumer started
		    //
		    return; // executes finally block before returning
		}

		state = State.STARTED;
		fireConsumerStarted(new ConsumerEvent(this,
			System.nanoTime()));
	    }

	    //
	    // We should not prevent other consumers from running
	    // concurrently while this consumer blocks on the native
	    // consumer loop.  Instead, native code will acquire the
	    // LocalConsumer.class monitor as needed before calling
	    // libdtrace functions.
	    //
	    _consume();

	} catch (Throwable e) {
	    if (exceptionHandler != null) {
		exceptionHandler.handleException(e);
	    } else {
		e.printStackTrace();
	    }
	} finally {
	    synchronized (stopLock) {
		// Notify listeners while holding stopLock to guarantee
		// that listeners finish executing consumerStopped()
		// before the stop() method returns.
		synchronized (this) {
		    state = State.STOPPED;
		    fireConsumerStopped(new ConsumerEvent(this,
			    System.nanoTime()));
		}

		// Notify the stop() method to stop waiting
		workEnded = true;
		stopLock.notifyAll();
	    }
	}
    }

    /**
     * Creates the background thread started by {@link #go()} to run
     * this consumer.  Override this method if you need to set
     * non-default {@code Thread} options or create the thread in a
     * {@code ThreadGroup}.  If you don't need to create the thread
     * yourself, set the desired options on {@code super.createThread()}
     * before returning it.  Otherwise, the {@code Runnable} target of
     * the created thread must call {@link #work()} in order to run this
     * DTrace consumer.  For example, to modify the default background
     * consumer thread:
     * <pre><code>
     *	protected Thread
     *	createThread()
     *	{
     *		Thread t = super.createThread();
     *		t.setPriority(Thread.MIN_PRIORITY);
     *		return t;
     *	}
     * </code></pre>
     * Or if you need to create your own thread:
     * <pre></code>
     *	protected Thread
     *	createThread()
     *	{
     *		Runnable target = new Runnable() {
     *			public void run() {
     *				work();
     *			}
     *		};
     *		String name = "Consumer " + UserApplication.sequence++;
     *		Thread t = new Thread(UserApplication.threadGroup,
     *			target, name);
     *		return t;
     *	}
     * </code></pre>
     * Do not start the returned thread, otherwise {@code go()} will
     * throw an {@link IllegalThreadStateException} when it tries to
     * start the returned thread a second time.
     */
    protected Thread
    createThread()
    {
	Thread t = new Thread(new Runnable() {
	    public void run() {
		work();
	    }
	}, "DTrace consumer " + id);
	return t;
    }

    /**
     * @inheritDoc
     * @throws IllegalThreadStateException if a subclass calls {@link
     * Thread#start()} on the value of {@link #createThread()}
     * @see #createThread()
     */
    public void
    go() throws DTraceException
    {
	go(null);
    }

    /**
     * @inheritDoc
     * @throws IllegalThreadStateException if a subclass calls {@link
     * Thread#start()} on the value of {@link #createThread()}
     * @see #createThread()
     */
    public synchronized void
    go(ExceptionHandler h) throws DTraceException
    {
	switch (state) {
	    case INIT:
		throw new IllegalStateException("consumer not open");
	    case OPEN:
		throw new IllegalStateException("no compiled program");
	    case COMPILED:
		//
		// Throws IllegalStateException if not all compiled programs are
		// also enabled.  Does not make any calls to libdtrace.
		//
		_checkProgramEnabling();
		break;
	    case GO:
	    case STARTED:
		throw new IllegalStateException("go() already called");
	    case STOPPED:
		throw new IllegalStateException("consumer stopped");
	    case CLOSED:
		throw new IllegalStateException("consumer closed");
	    default:
		throw new IllegalArgumentException("unknown state: " + state);
	}

	synchronized (LocalConsumer.class) {
	    _go();
	}

	state = State.GO;
	exceptionHandler = h;
	Thread t = createThread();
	t.start();
    }

    /**
     * @inheritDoc
     *
     * @throws IllegalThreadStateException if attempting to {@code
     * stop()} a running consumer while holding the lock on that
     * consumer
     */
    public void
    stop()
    {
	boolean running = false;

	synchronized (this) {
	    switch (state) {
		case INIT:
		    throw new IllegalStateException("consumer not open");
		case OPEN:
		case COMPILED:
		    throw new IllegalStateException("go() not called");
		case GO:
		    try {
			synchronized (LocalConsumer.class) {
			    _stop();
			}
			state = State.STOPPED;
		    } catch (DTraceException e) {
			if (exceptionHandler != null) {
			    exceptionHandler.handleException(e);
			} else {
			    e.printStackTrace();
			}
		    }
		    break;
		case STARTED:
		    running = true;
		    break;
		case STOPPED:
		    //
		    // The work() thread that runs the native consumer
		    // loop may have terminated because of the exit()
		    // action in a DTrace program.  In that case, a
		    // RuntimeException is inappropriate because there
		    // is no misuse of the API.  Creating a new checked
		    // exception type to handle this case seems to offer
		    // no benefit for the trouble to the caller.
		    // Instead, the situation calls for stop() to be
		    // quietly tolerant.
		    //
		    if (stopCalled) {
			throw new IllegalStateException(
				"consumer already stopped");
		    }
		    logger.fine("consumer already stopped");
		    break;
		case CLOSED:
		    throw new IllegalStateException("consumer closed");
		default:
		    throw new IllegalArgumentException("unknown state: " +
			    state);
	    }

	    stopCalled = true;
	}

	if (running) {
	    if (Thread.holdsLock(this)) {
		throw new IllegalThreadStateException("The current " +
			"thread cannot stop this LocalConsumer while " +
			"holding the lock on this LocalConsumer");
	    }

	    //
	    // Calls no libdtrace methods, so no synchronization is
	    // needed.  Sets a native flag that causes the consumer
	    // thread to exit the consumer loop and call native
	    // dtrace_stop() at the end of the current interval (after
	    // grabbing the global Consumer.class lock required for any
	    // libdtrace call).
	    //
	    _interrupt();

	    synchronized (stopLock) {
		//
		// Wait for work() to set workEnded.  If the work()
		// thread got the stopLock first, then workEnded is
		// already set.
		//
		while (!workEnded) {
		    try {
			stopLock.wait();
		    } catch (InterruptedException e) {
			logger.warning(e.toString());
			// do nothing but re-check the condition for
			// waiting
		    }
		}
	    }
	}
    }

    public synchronized void
    abort()
    {
	if ((state != State.INIT) && (state != State.CLOSED)) {
	    _interrupt();
	}
	abortCalled = true;
    }

    /**
     * @inheritDoc
     *
     * @throws IllegalThreadStateException if attempting to {@code
     * close()} a running consumer while holding the lock on that
     * consumer
     */
    public void
    close()
    {
	synchronized (this) {
	    if ((state == State.INIT) || (state == State.CLOSED)) {
		state = State.CLOSED;
		return;
	    }
	}

	try {
	    stop();
	} catch (IllegalStateException e) {
	    // ignore (we don't have synchronized state access because
	    // it is illegal to call stop() while holding the lock on
	    // this consumer)
	}

	synchronized (this) {
	    if (state != State.CLOSED) {
		synchronized (LocalConsumer.class) {
		    _close();
		}
		_destroy();
		state = State.CLOSED;

		if (logger.isLoggable(Level.INFO)) {
		    logger.info("consumer table count: " + _openCount());
		}
	    }
	}
    }

    public void
    addConsumerListener(ConsumerListener l)
    {
        listenerList.add(ConsumerListener.class, l);
    }

    public void
    removeConsumerListener(ConsumerListener l)
    {
        listenerList.remove(ConsumerListener.class, l);
    }

    public Aggregate
    getAggregate() throws DTraceException
    {
	// include all, clear none
	return getAggregate(null, Collections. <String> emptySet());
    }

    public Aggregate
    getAggregate(Set <String> includedAggregationNames)
            throws DTraceException
    {
	return getAggregate(includedAggregationNames,
		Collections. <String> emptySet());
    }

    public Aggregate
    getAggregate(Set <String> includedAggregationNames,
	    Set <String> clearedAggregationNames)
            throws DTraceException
    {
	AggregateSpec spec = new AggregateSpec();

	if (includedAggregationNames == null) {
	    spec.setIncludeByDefault(true);
	} else {
	    spec.setIncludeByDefault(false);
	    for (String included : includedAggregationNames) {
		spec.addIncludedAggregationName(included);
	    }
	}

	if (clearedAggregationNames == null) {
	    spec.setClearByDefault(true);
	} else {
	    spec.setClearByDefault(false);
	    for (String cleared : clearedAggregationNames) {
		spec.addClearedAggregationName(cleared);
	    }
	}

	return getAggregate(spec);
    }

    private synchronized Aggregate
    getAggregate(AggregateSpec spec) throws DTraceException
    {
	//
	// It should be possible to request aggregation data after a
	// consumer has stopped but not after it has been closed.
	//
	checkGoCalled();

	//
	// Getting the aggregate is a time-consuming request that should not
	// prevent other consumers from running concurrently.  Instead,
	// native code will acquire the LocalConsumer.class monitor as
	// needed before calling libdtrace functions.
	//
	Aggregate aggregate = _getAggregate(spec);
	return aggregate;
    }

    private synchronized void
    checkGoCalled()
    {
	switch (state) {
	    case INIT:
		throw new IllegalStateException("consumer not open");
	    case OPEN:
	    case COMPILED:
		throw new IllegalStateException("go() not called");
	    case GO:
	    case STARTED:
	    case STOPPED:
		break;
	    case CLOSED:
		throw new IllegalStateException("consumer closed");
	}
    }

    private synchronized void
    checkGoNotCalled()
    {
	switch (state) {
	    case INIT:
		throw new IllegalStateException("consumer not open");
	    case OPEN:
	    case COMPILED:
		break;
	    case GO:
	    case STARTED:
		throw new IllegalStateException("go() already called");
	    case STOPPED:
		throw new IllegalStateException("consumer stopped");
	    case CLOSED:
		throw new IllegalStateException("consumer closed");
	}
    }

    public synchronized int
    createProcess(String command) throws DTraceException
    {
	if (command == null) {
	    throw new NullPointerException("command is null");
	}

	checkGoNotCalled();

	int pid;
	synchronized (LocalConsumer.class) {
	    pid = _createProcess(command);
	}
	return pid;
    }

    public synchronized void
    grabProcess(int pid) throws DTraceException
    {
	checkGoNotCalled();

	synchronized (LocalConsumer.class) {
	    _grabProcess(pid);
	}
    }

    public synchronized List <ProbeDescription>
    listProbes(ProbeDescription filter) throws DTraceException
    {
	checkGoNotCalled();
	List <ProbeDescription> probeList =
		new LinkedList <ProbeDescription> ();
	if (filter == ProbeDescription.EMPTY) {
	    filter = null;
	}
	synchronized (LocalConsumer.class) {
	    _listProbes(probeList, filter);
	}
	return probeList;
    }

    public synchronized List <Probe>
    listProbeDetail(ProbeDescription filter) throws DTraceException
    {
	checkGoNotCalled();
	List <Probe> probeList = new LinkedList <Probe> ();
	if (filter == ProbeDescription.EMPTY) {
	    filter = null;
	}
	synchronized (LocalConsumer.class) {
	    _listProbeDetail(probeList, filter);
	}
	return probeList;
    }

    public synchronized List <ProbeDescription>
    listProgramProbes(Program program) throws DTraceException
    {
	checkProgram(program);
	checkGoNotCalled();
	List <ProbeDescription> probeList =
		new LinkedList <ProbeDescription> ();
	synchronized (LocalConsumer.class) {
	    _listCompiledProbes(probeList, program);
	}
	return probeList;
    }

    public synchronized List <Probe>
    listProgramProbeDetail(Program program) throws DTraceException
    {
	checkProgram(program);
	checkGoNotCalled();
	List <Probe> probeList = new LinkedList <Probe> ();
	synchronized (LocalConsumer.class) {
	    _listCompiledProbeDetail(probeList, program);
	}
	return probeList;
    }

    public synchronized String
    lookupKernelFunction(int address)
    {
	checkGoCalled();
	synchronized (LocalConsumer.class) {
	    return _lookupKernelFunction(new Integer(address));
	}
    }

    public synchronized String
    lookupKernelFunction(long address)
    {
	checkGoCalled();
	synchronized (LocalConsumer.class) {
	    return _lookupKernelFunction(new Long(address));
	}
    }

    public synchronized String
    lookupUserFunction(int pid, int address)
    {
	checkGoCalled();
	synchronized (LocalConsumer.class) {
	    return _lookupUserFunction(pid, new Integer(address));
	}
    }

    public synchronized String
    lookupUserFunction(int pid, long address)
    {
	checkGoCalled();
	synchronized (LocalConsumer.class) {
	    return _lookupUserFunction(pid, new Long(address));
	}
    }

    public String
    getVersion()
    {
	synchronized (LocalConsumer.class) {
	    return LocalConsumer._getVersion();
	}
    }

    /**
     * Called by native code.
     */
    private void
    nextProbeData(ProbeData probeData) throws ConsumerException
    {
	fireDataReceived(new DataEvent(this, probeData));
    }

    /**
     * Called by native code.
     */
    private void
    dataDropped(Drop drop) throws ConsumerException
    {
	fireDataDropped(new DropEvent(this, drop));
    }

    /**
     * Called by native code.
     */
    private void
    errorEncountered(Error error) throws ConsumerException
    {
	fireErrorEncountered(new ErrorEvent(this, error));
    }

    /**
     * Called by native code.
     */
    private void
    processStateChanged(ProcessState processState) throws ConsumerException
    {
	fireProcessStateChanged(new ProcessEvent(this, processState));
    }

    protected void
    fireDataReceived(DataEvent e) throws ConsumerException
    {
        // Guaranteed to return a non-null array
        Object[] listeners = listenerList.getListenerList();
        // Process the listeners last to first, notifying
        // those that are interested in this event
        for (int i = listeners.length - 2; i >= 0; i -= 2) {
            if (listeners[i] == ConsumerListener.class) {
                ((ConsumerListener)listeners[i + 1]).dataReceived(e);
            }
        }
    }

    protected void
    fireDataDropped(DropEvent e) throws ConsumerException
    {
        // Guaranteed to return a non-null array
        Object[] listeners = listenerList.getListenerList();
        // Process the listeners last to first, notifying
        // those that are interested in this event
        for (int i = listeners.length - 2; i >= 0; i -= 2) {
            if (listeners[i] == ConsumerListener.class) {
                ((ConsumerListener)listeners[i + 1]).dataDropped(e);
            }
        }
    }

    protected void
    fireErrorEncountered(ErrorEvent e) throws ConsumerException
    {
        // Guaranteed to return a non-null array
        Object[] listeners = listenerList.getListenerList();
        // Process the listeners last to first, notifying
        // those that are interested in this event
        for (int i = listeners.length - 2; i >= 0; i -= 2) {
            if (listeners[i] == ConsumerListener.class) {
                ((ConsumerListener)listeners[i + 1]).errorEncountered(e);
            }
        }
    }

    protected void
    fireProcessStateChanged(ProcessEvent e) throws ConsumerException
    {
        // Guaranteed to return a non-null array
        Object[] listeners = listenerList.getListenerList();
        // Process the listeners last to first, notifying
        // those that are interested in this event
        for (int i = listeners.length - 2; i >= 0; i -= 2) {
            if (listeners[i] == ConsumerListener.class) {
                ((ConsumerListener)listeners[i + 1]).processStateChanged(e);
            }
        }
    }

    protected void
    fireConsumerStarted(ConsumerEvent e)
    {
        // Guaranteed to return a non-null array
        Object[] listeners = listenerList.getListenerList();
        // Process the listeners last to first, notifying
        // those that are interested in this event
        for (int i = listeners.length - 2; i >= 0; i -= 2) {
            if (listeners[i] == ConsumerListener.class) {
                ((ConsumerListener)listeners[i + 1]).consumerStarted(e);
            }
        }
    }

    protected void
    fireConsumerStopped(ConsumerEvent e)
    {
        // Guaranteed to return a non-null array
        Object[] listeners = listenerList.getListenerList();
        // Process the listeners last to first, notifying
        // those that are interested in this event
        for (int i = listeners.length - 2; i >= 0; i -= 2) {
            if (listeners[i] == ConsumerListener.class) {
                ((ConsumerListener)listeners[i + 1]).consumerStopped(e);
            }
        }
    }

    // Called by native code
    private void
    intervalBegan()
    {
	fireIntervalBegan(new ConsumerEvent(this, System.nanoTime()));
    }

    protected void
    fireIntervalBegan(ConsumerEvent e)
    {
        // Guaranteed to return a non-null array
        Object[] listeners = listenerList.getListenerList();
        // Process the listeners last to first, notifying
        // those that are interested in this event
        for (int i = listeners.length - 2; i >= 0; i -= 2) {
            if (listeners[i] == ConsumerListener.class) {
                ((ConsumerListener)listeners[i + 1]).intervalBegan(e);
            }
        }
    }

    // Called by native code
    private void
    intervalEnded()
    {
	fireIntervalEnded(new ConsumerEvent(this, System.nanoTime()));
    }

    protected void
    fireIntervalEnded(ConsumerEvent e)
    {
        // Guaranteed to return a non-null array
        Object[] listeners = listenerList.getListenerList();
        // Process the listeners last to first, notifying
        // those that are interested in this event
        for (int i = listeners.length - 2; i >= 0; i -= 2) {
            if (listeners[i] == ConsumerListener.class) {
                ((ConsumerListener)listeners[i + 1]).intervalEnded(e);
            }
        }
    }

    /**
     * Gets a string representation of this consumer useful for logging
     * and not intended for display.  The exact details of the
     * representation are unspecified and subject to change, but the
     * following format may be regarded as typical:
     * <pre><code>
     * class-name[property1 = value1, property2 = value2]
     * </code></pre>
     */
    public String
    toString()
    {
	StringBuffer buf = new StringBuffer(LocalConsumer.class.getName());
	synchronized (this) {
	    buf.append("[open = ");
	    buf.append(isOpen());
	    buf.append(", enabled = ");
	    buf.append(isEnabled());
	    buf.append(", running = ");
	    buf.append(isRunning());
	    buf.append(", closed = ");
	    buf.append(isClosed());
	}
	buf.append(']');
	return buf.toString();
    }

    /**
     * Ensures that the {@link #close()} method of this consumer has
     * been called before it is garbage-collected.  The intended safety
     * net is weak because the JVM does not guarantee that an object
     * will be garbage-collected when it is no longer referenced.  Users
     * of the API should call {@code close()} to ensure that all
     * resources associated with this consumer are reclaimed in a timely
     * manner.
     *
     * @see #close()
     */
    protected void
    finalize()
    {
	close();
    }

    private String
    getTag()
    {
	return super.toString();
    }

    //
    // Uniquely identifies a consumer across systems so it is possible
    // to validate that an object such as a Program passed to a remote
    // client over a socket was created by this consumer and no other.
    //
    static class Identifier implements Serializable {
	static final long serialVersionUID = 2183165132305302834L;

	// local identifier
	private int id;
	private long timestamp;
	// remote identifier
	private InetAddress localHost;
	private String tag; // in case localHost not available

	private
	Identifier(LocalConsumer consumer)
	{
	    id = LocalConsumer.sequence++;
	    timestamp = System.currentTimeMillis();
	    try {
		localHost = InetAddress.getLocalHost();
	    } catch (UnknownHostException e) {
		localHost = null;
	    }
	    tag = consumer.getTag();
	}

	@Override
	public boolean
	equals(Object o)
	{
	    if (o == this) {
		return true;
	    }
	    if (o instanceof Identifier) {
		Identifier i = (Identifier)o;
		return ((id == i.id) &&
			(timestamp == i.timestamp) &&
			((localHost == null) ? (i.localHost == null) :
			 localHost.equals(i.localHost)) &&
			tag.equals(i.tag));
	    }
	    return false;
	}

	@Override
	public int
	hashCode()
	{
	    int hash = 17;
	    hash = (37 * hash) + id;
	    hash = (37 * hash) + ((int)(timestamp ^ (timestamp >>> 32)));
	    hash = (37 * hash) + (localHost == null ? 0 :
		    localHost.hashCode());
	    hash = (37 * hash) + tag.hashCode();
	    return hash;
	}

	@Override
	public String
	toString()
	{
	    StringBuffer buf = new StringBuffer();
	    buf.append(Identifier.class.getName());
	    buf.append("[id = ");
	    buf.append(id);
	    buf.append(", timestamp = ");
	    buf.append(timestamp);
	    buf.append(", localHost = ");
	    buf.append(localHost);
	    buf.append(", tag = ");
	    buf.append(tag);
	    buf.append(']');
	    return buf.toString();
	}
    }
}