Agent.java

/**
 *   Copyright (c) Rich Hickey. All rights reserved.
 *   The use and distribution terms for this software are covered by the
 *   Eclipse Public License 1.0 (http://opensource.org/licenses/eclipse-1.0.php)
 *   which can be found in the file epl-v10.html at the root of this distribution.
 *   By using this software in any fashion, you are agreeing to be bound by
 * 	 the terms of this license.
 *   You must not remove this notice, or any other, from this software.
 **/

/* rich Nov 17, 2007 */

package clojure.lang;

import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

public class Agent extends ARef {

static class ActionQueue {
	public final IPersistentStack q;
	public final Throwable error; // non-null indicates fail state
	static final ActionQueue EMPTY = new ActionQueue(PersistentQueue.EMPTY, null);

	public ActionQueue( IPersistentStack q, Throwable error )
		{
		this.q = q;
		this.error = error;
		}
}

static final Keyword CONTINUE = Keyword.intern(null, "continue");
static final Keyword FAIL = Keyword.intern(null, "fail");

volatile Object state;
    AtomicReference<ActionQueue> aq = new AtomicReference<ActionQueue>(ActionQueue.EMPTY);

    volatile Keyword errorMode = CONTINUE;
    volatile IFn errorHandler = null;

final private static AtomicLong sendThreadPoolCounter = new AtomicLong(0);

final private static AtomicLong sendOffThreadPoolCounter = new AtomicLong(0);

volatile public static ExecutorService pooledExecutor =
	Executors.newFixedThreadPool(2 + Runtime.getRuntime().availableProcessors(), 
		createThreadFactory("clojure-agent-send-pool-%d", sendThreadPoolCounter));

volatile public static ExecutorService soloExecutor = Executors.newCachedThreadPool(
	createThreadFactory("clojure-agent-send-off-pool-%d", sendOffThreadPoolCounter));

final static ThreadLocal<IPersistentVector> nested = new ThreadLocal<IPersistentVector>();

private static ThreadFactory createThreadFactory(final String format, final AtomicLong threadPoolCounter) {
	return new ThreadFactory() {
		public Thread newThread(Runnable runnable) {
			Thread thread = new Thread(runnable);
			thread.setName(String.format(format, threadPoolCounter.getAndIncrement()));
			return thread;
		}
	};
}

public static void shutdown(){
	soloExecutor.shutdown();
	pooledExecutor.shutdown();
}

static class Action implements Runnable{
	final Agent agent;
	final IFn fn;
	final ISeq args;
	final Executor exec;


	public Action(Agent agent, IFn fn, ISeq args, Executor exec){
		this.agent = agent;
		this.args = args;
		this.fn = fn;
		this.exec = exec;
	}

	void execute(){
		try
			{
			exec.execute(this);
			}
		catch(Throwable error)
			{
			if(agent.errorHandler != null)
				{
				try
					{
					agent.errorHandler.invoke(agent, error);
					}
				catch(Throwable e) {} // ignore errorHandler errors
				}
			}
	}

	static void doRun(Action action){
		try
			{
			nested.set(PersistentVector.EMPTY);

			Throwable error = null;
			try
				{
				Object oldval = action.agent.state;
				Object newval =  action.fn.applyTo(RT.cons(action.agent.state, action.args));
				action.agent.setState(newval);
                action.agent.notifyWatches(oldval,newval);
				}
			catch(Throwable e)
				{
				error = e;
				}

			if(error == null)
				{
				releasePendingSends();
				}
			else
				{
				nested.set(null); // allow errorHandler to send
				if(action.agent.errorHandler != null)
					{
					try
						{
						action.agent.errorHandler.invoke(action.agent, error);
						}
					catch(Throwable e) {} // ignore errorHandler errors
					}
				if(action.agent.errorMode == CONTINUE)
					{
					error = null;
					}
				}

			boolean popped = false;
			ActionQueue next = null;
			while(!popped)
				{
				ActionQueue prior = action.agent.aq.get();
				next = new ActionQueue(prior.q.pop(), error);
				popped = action.agent.aq.compareAndSet(prior, next);
				}

			if(error == null && next.q.count() > 0)
				((Action) next.q.peek()).execute();
			}
		finally
			{
			nested.set(null);
			}
	}

	public void run(){
		doRun(this);
	}
}

public Agent(Object state) {
	this(state,null);
}

public Agent(Object state, IPersistentMap meta)  {
    super(meta);
    setState(state);
}

boolean setState(Object newState) {
	validate(newState);
	boolean ret = state != newState;
	state = newState;
	return ret;
}

public Object deref() {
	return state;
}

public Throwable getError(){
	return aq.get().error;
}

public void setErrorMode(Keyword k){
	errorMode = k;
}

public Keyword getErrorMode(){
	return errorMode;
}

public void setErrorHandler(IFn f){
	errorHandler = f;
}

public IFn getErrorHandler(){
	return errorHandler;
}

synchronized public Object restart(Object newState, boolean clearActions){
	if(getError() == null)
		{
		throw Util.runtimeException("Agent does not need a restart");
		}
	validate(newState);
	state = newState;

	if(clearActions)
		aq.set(ActionQueue.EMPTY);
	else
		{
		boolean restarted = false;
		ActionQueue prior = null;
		while(!restarted)
			{
			prior = aq.get();
			restarted = aq.compareAndSet(prior, new ActionQueue(prior.q, null));
			}

		if(prior.q.count() > 0)
			((Action) prior.q.peek()).execute();
		}

	return newState;
}

public Object dispatch(IFn fn, ISeq args, Executor exec) {
	Throwable error = getError();
	if(error != null)
		{
		throw Util.runtimeException("Agent is failed, needs restart", error);
		}
	Action action = new Action(this, fn, args, exec);
	dispatchAction(action);

	return this;
}

static void dispatchAction(Action action){
	LockingTransaction trans = LockingTransaction.getRunning();
	if(trans != null)
		trans.enqueue(action);
	else if(nested.get() != null)
		{
		nested.set(nested.get().cons(action));
		}
	else
		action.agent.enqueue(action);
}

void enqueue(Action action){
	boolean queued = false;
	ActionQueue prior = null;
	while(!queued)
		{
		prior = aq.get();
		queued = aq.compareAndSet(prior, new ActionQueue((IPersistentStack)prior.q.cons(action), prior.error));
		}

	if(prior.q.count() == 0 && prior.error == null)
		action.execute();
}

public int getQueueCount(){
	return aq.get().q.count();
}

static public int releasePendingSends(){
	IPersistentVector sends = nested.get();
	if(sends == null)
		return 0;
	for(int i=0;i<sends.count();i++)
		{
		Action a = (Action) sends.valAt(i);
		a.agent.enqueue(a);
		}
	nested.set(PersistentVector.EMPTY);
	return sends.count();
}
}