<commit_before>/***************************************************************************
 *   Copyright (C) 2006-2013 by <NAME> <<EMAIL>>     *
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU Library 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 Library 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 details about the authors of this software, see the AUTHORS file. *
 ***************************************************************************/

package jolie;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.PrintStream;
import java.lang.ref.WeakReference;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Deque;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;
import jolie.lang.Constants;
import jolie.lang.parse.OLParseTreeOptimizer;
import jolie.lang.parse.OLParser;
import jolie.lang.parse.ParserException;
import jolie.lang.parse.Scanner;
import jolie.lang.parse.SemanticVerifier;
import jolie.lang.parse.TypeChecker;
import jolie.lang.parse.ast.Program;
import jolie.monitoring.MonitoringEvent;
import jolie.monitoring.events.MonitorAttachedEvent;
import jolie.monitoring.events.OperationStartedEvent;
import jolie.monitoring.events.SessionEndedEvent;
import jolie.monitoring.events.SessionStartedEvent;
import jolie.net.CommChannel;
import jolie.net.CommCore;
import jolie.net.CommMessage;
import jolie.net.SessionMessage;
import jolie.net.ports.OutputPort;
import jolie.process.DefinitionProcess;
import jolie.process.InputOperationProcess;
import jolie.process.SequentialProcess;
import jolie.process.TransformationReason;
import jolie.runtime.ExitingException;
import jolie.runtime.FaultException;
import jolie.runtime.InputOperation;
import jolie.runtime.InvalidIdException;
import jolie.runtime.OneWayOperation;
import jolie.runtime.RequestResponseOperation;
import jolie.runtime.TimeoutHandler;
import jolie.runtime.Value;
import jolie.runtime.ValueVector;
import jolie.runtime.correlation.CorrelationEngine;
import jolie.runtime.correlation.CorrelationError;
import jolie.runtime.correlation.CorrelationSet;
import jolie.runtime.embedding.EmbeddedServiceLoader;
import jolie.tracer.DummyTracer;
import jolie.tracer.PrintingTracer;
import jolie.tracer.Tracer;

/**
 * The Jolie interpreter engine.
 * Multiple Interpreter instances can be run in the same JavaVM;
 * this is used, e.g., for service embedding.
 * @author <NAME>
 */
public class Interpreter
{
	private class InitSessionThread extends SessionThread
	{
		public InitSessionThread( Interpreter interpreter, jolie.process.Process process )
		{
			super( interpreter, process );
		}
		
		@Override
		public boolean isInitialisingThread()
		{
			return true;
		}
		
		@Override
		public void run()
		{
			super.run();

			if ( executionMode == Constants.ExecutionMode.SINGLE ) {
				try {
					mainSession = new SessionThread( getDefinition( "main" ), initExecutionThread );
					correlationEngine.onSingleExecutionSessionStart( mainSession );
					mainSession.addSessionListener( correlationEngine );
				} catch( InvalidIdException e ) {
					assert false;
				}
			}
			
			/*
			 * We need to relay the messages we did not consume during the init procedure.
			 * We do this asynchronously, because calling correlationEngine.onMessageReceive
			 * will trigger a join() on this thread, leading to a deadlock if we
			 * were to call that directly from here.
			 */
			execute( new Runnable() {
				private void pushMessages( Deque< SessionMessage > queue )
				{
					for( SessionMessage message : queue ) {
						try {
							correlationEngine.onMessageReceive( message.message(), message.channel() );
						} catch( CorrelationError e ) {
							logWarning( e );
							try {
								message.channel().send( CommMessage.createFaultResponse( message.message(), new FaultException( "CorrelationError", "The message you sent can not be correlated with any session and can not be used to start a new session." ) ) );
							} catch( IOException ioe ) {
								logSevere( ioe );
							}
						}
					}
				}

				public void run()
				{
					for( Deque< SessionMessage > queue : messageQueues.values() ) {
						pushMessages( queue );
					}
					pushMessages( uncorrelatedMessageQueue );
				}
			});
		}
	}
	
	private static class JolieExecutionThreadFactory implements ThreadFactory {
		private final Interpreter interpreter;

		private static class JolieExecutionThread extends JolieThread
		{
			public JolieExecutionThread( Interpreter interpreter, Runnable r )
			{
				super( interpreter, r );
			}
		}
		
		public JolieExecutionThreadFactory( Interpreter interpreter )
		{
			this.interpreter = interpreter;
		}
		
		public Thread newThread( Runnable r )
		{
			return new JolieExecutionThread( interpreter, r );
		}
	}
	
	public static class SessionStarter
	{
		private final InputOperationProcess guard;
		private final jolie.process.Process body;
		private CorrelationSet correlationInitializer = null;

		public SessionStarter( InputOperationProcess guard, jolie.process.Process body )
		{
			this.guard = guard;
			this.body = body;
		}

		public InputOperationProcess guard()
		{
			return guard;
		}

		public jolie.process.Process body()
		{
			return body;
		}

		public void setCorrelationInitializer( CorrelationSet cset )
		{
			correlationInitializer = cset;
		}

		public CorrelationSet correlationInitializer()
		{
			return correlationInitializer;
		}
	}


	private CommCore commCore;
	private CommandLineParser cmdParser;

	private Map< String, SessionStarter > sessionStarters = new HashMap< String, SessionStarter >();
	private boolean exiting = false;
	private final Lock exitingLock;
	private final Condition exitingCondition;
	private final CorrelationEngine correlationEngine;
	private final List< CorrelationSet > correlationSets =
				new ArrayList< CorrelationSet > ();
	private final Map< String, CorrelationSet > operationCorrelationSetMap = new HashMap< String, CorrelationSet >();
	private Constants.ExecutionMode executionMode = Constants.ExecutionMode.SINGLE;
	private final Value globalValue = Value.createRootValue();
	private final String[] arguments;
	private final Collection< EmbeddedServiceLoader > embeddedServiceLoaders =
			new LinkedList< EmbeddedServiceLoader >();
	private static final Logger logger = Logger.getLogger( "Jolie" );
	
	private final Map< String, DefinitionProcess > definitions =
				new HashMap< String, DefinitionProcess >();
	private final Map< String, OutputPort > outputPorts = new HashMap< String, OutputPort >();
	private final Map< String, InputOperation > inputOperations =
				new HashMap< String, InputOperation>();
	
	private final HashMap< String, Object > locksMap =
				new HashMap< String, Object >();
	
	private final ClassLoader parentClassLoader;
	private final String[] includePaths;
	private final String[] optionArgs;
	private final String logPrefix;
	private final Tracer tracer;
	private final Timer timer;
	// private long inputMessageTimeout = 24 * 60 * 60 * 1000; // 1 day
	private final long persistentConnectionTimeout = 60 * 60 * 1000; // 1 hour
	private final long awaitTerminationTimeout = 60 * 1000; // 1 minute
	// private long persistentConnectionTimeout = 2 * 60 * 1000; // 4 minutes
	// private long persistentConnectionTimeout = 1;

	// 11 is the default initial capacity for PriorityQueue
	private final Queue< WeakReference< TimeoutHandler > > timeoutHandlerQueue =
		new PriorityQueue< WeakReference< TimeoutHandler > >( 11, new TimeoutHandler.Comparator() );
	
	private final ExecutorService timeoutHandlerExecutor = Executors.newSingleThreadExecutor();

	private final String programFilename;
	private final File programDirectory;
	private OutputPort monitor = null;

	public void setMonitor( OutputPort monitor )
	{
		this.monitor = monitor;
		fireMonitorEvent( new MonitorAttachedEvent() );
	}
	
	public boolean isMonitoring()
	{
		return monitor != null;
	}
	
	/*public long inputMessageTimeout()
	{
		return inputMessageTimeout;
	}*/
	
	public String logPrefix()
	{
		return logPrefix;
	}
	
	public Tracer tracer()
	{
		return tracer;
	}
	
	public void fireMonitorEvent( MonitoringEvent event )
	{
		if ( monitor != null ) {
			CommMessage m = CommMessage.createRequest( "pushEvent", "/", MonitoringEvent.toValue( event ) );
			CommChannel channel = null;
			try {
				channel = monitor.getCommChannel();
				channel.send( m );
				CommMessage response;
				do {
					response = channel.recvResponseFor( m );
				} while( response == null );
			} catch( URISyntaxException e ) {
				logWarning( e );
			} catch( IOException e ) {
				logWarning( e );
			} finally {
				if ( channel != null ) {
					try {
						channel.release();
					} catch( IOException e ) {
						logWarning( e );
					}
				}
			}
		}
	}

	public long persistentConnectionTimeout()
	{
		return persistentConnectionTimeout;
	}

	public CorrelationEngine correlationEngine()
	{
		return correlationEngine;
	}

	public void schedule( TimerTask task, long delay )
	{
		if ( exiting == false ) {
			timer.schedule( task, delay );
		}
	}

	public void addTimeoutHandler( TimeoutHandler handler )
	{
		synchronized( timeoutHandlerQueue ) {
			timeoutHandlerQueue.add( new WeakReference< TimeoutHandler >( handler ) );
			if ( timeoutHandlerQueue.size() == 1 ) {
				schedule( new TimerTask() {
					public void run()
					{
						synchronized( timeoutHandlerQueue ) {
							checkForExpiredTimeoutHandlers();
						}
					}
				}, handler.time() - System.currentTimeMillis() + 1 );
			} else {
				checkForExpiredTimeoutHandlers();
			}
		}
	}

	/*public void removeTimeoutHandler( TimeoutHandler handler )
	{
		synchronized( timeoutHandlerQueue ) {
			timeoutHandlerQueue.remove( handler );
			checkForExpiredTimeoutHandlers();
		}
	}*/

	private void checkForExpiredTimeoutHandlers()
	{
		long currentTime = System.currentTimeMillis();
		WeakReference< TimeoutHandler > whandler = timeoutHandlerQueue.peek();
		boolean keepRun = true;
		TimeoutHandler handler;
		while( whandler != null && keepRun ) {
			handler = whandler.get();
			if ( handler == null ) {
				timeoutHandlerQueue.remove();
				whandler = timeoutHandlerQueue.peek();
			} else if ( handler.time() < currentTime || exiting ) {
				// final TimeoutHandler h = handler;
				timeoutHandlerExecutor.execute( handler );
				timeoutHandlerQueue.remove();
				whandler = timeoutHandlerQueue.peek();
			} else {
				keepRun = false;
			}
		}
	}
	
	/**
	 * Returns the option arguments passed to this Interpreter.
	 * @return the option arguments passed to this Interpreter
	 */
	public String[] optionArgs()
	{
		return optionArgs;
	}
	
	/**
	 * Returns the include paths this Interpreter is considering.
	 * @return the include paths this Interpreter is considering
	 */
	public String[] includePaths()
	{
		return includePaths;
	}

	/**
	 * Registers a session starter on this <code>Interpreter</code>.
	 * @param guard the input guard for this session starter
	 * @param process the body of this session starter
	 */
	public void registerSessionStarter( InputOperationProcess guard, jolie.process.Process body )
	{
		sessionStarters.put( guard.inputOperation().id(), new SessionStarter( guard, body ) );
	}
	
	private JolieClassLoader classLoader;
	
	/**
	 * Returns the output ports of this Interpreter.
	 * @return the output ports of this Interpreter
	 */
	public Collection< OutputPort > outputPorts()
	{
		return outputPorts.values();
	}
		
	/**
	 * Returns the InputOperation identified by key.
	 * @param key the name of the InputOperation to return
	 * @return the InputOperation identified by key
	 * @throws jolie.runtime.InvalidIdException if this Interpreter does not own the requested InputOperation
	 * @see InputOperation
	 */
	public InputOperation getInputOperation( String key )
		throws InvalidIdException
	{
		InputOperation ret = inputOperations.get( key );
		if ( ret == null ) {
			throw new InvalidIdException( key );
		}
		return ret;
	}
	
	/**
	 * As {@link #getInputOperation(String) getInputOperation}, with the additional constraint that key must identify a OneWayOperation.
	 * @param key the name of the OneWayOperation to return
	 * @return the OneWayOperation identified by key
	 * @throws jolie.runtime.InvalidIdException if this Interpreter does not own the requested OneWayOperation
	 * @see OneWayOperation
	 */
	public OneWayOperation getOneWayOperation( String key )
		throws InvalidIdException
	{
		InputOperation ret;
		if ( (ret=inputOperations.get( key )) == null || !(ret instanceof OneWayOperation) )
			throw new InvalidIdException( key );
		return (OneWayOperation)ret;
	}
	
	/**
	 * As {@link #getInputOperation(String) getInputOperation}, with the additional constraint that key must identify a RequestResponseOperation.
	 * @param key the name of the RequestResponseOperation to return
	 * @return the RequestResponseOperation identified by key
	 * @throws jolie.runtime.InvalidIdException if this Interpreter does not own the requested RequestResponseOperation
	 * @see RequestResponseOperation
	 */
	public RequestResponseOperation getRequestResponseOperation( String key )
		throws InvalidIdException
	{
		InputOperation ret;
		if ( (ret=inputOperations.get( key )) == null || !(ret instanceof RequestResponseOperation) )
			throw new InvalidIdException( key );
		return (RequestResponseOperation)ret;
	}
	
	/**
	 * Returns the OutputPort identified by key.
	 * @param key the name of the OutputPort to return
	 * @return the OutputPort identified by key
	 * @throws jolie.runtime.InvalidIdException if this Interpreter does not own the requested OutputPort
	 */
	public synchronized OutputPort getOutputPort( String key )
		throws InvalidIdException
	{
		OutputPort ret;
		if ( (ret=outputPorts.get( key )) == null )
			throw new InvalidIdException( key );
		return (OutputPort)ret;
	}
	
	/**
	 * Removes a registered OutputPort.
	 * @param key the name of the OutputPort to remove
	 */
	public synchronized void removeOutputPort( String key )
	{
		outputPorts.remove( key );
	}

	/**
	 * Returns the Definition identified by key.
	 * @param key the name of the Definition to return
	 * @return the Definition identified by key
	 * @throws jolie.runtime.InvalidIdException if this Interpreter does not own the requested Definition
	 */
	public DefinitionProcess getDefinition( String key )
		throws InvalidIdException
	{
		DefinitionProcess ret;
		if ( (ret=definitions.get( key )) == null )
			throw new InvalidIdException( key );
		return ret;
	}

	/**
	 * Registers an <code>OutputPort</code> on this interpreter.
	 * @param key the name of the <code>OutputPort</code> to register
	 * @param value the <code>OutputPort</code> to register
	 */
	public void register( String key, OutputPort value )
	{
		outputPorts.put( key, value );
	}

	/**
	 * Registers a defined sub-routine on this interpreter.
	 * @param key the name of the defined sub-routine to register
	 * @param value the defined sub-routine to register
	 */
	public void register( String key, DefinitionProcess value )
	{
		definitions.put( key, value );
	}

	/**
	 * Registers an <code>InputOperation</code> on this interpreter.
	 * @param key the name of the <code>InputOperation</code> to register
	 * @param value the <code>InputOperation</code> to register
	 */
	public void register( String key, InputOperation value )
	{
		inputOperations.put( key, value );
	}

	/**
	 * Registers an <code>EmbeddedServiceLoader</code> on this interpreter.
	 * @param key the name of the <code>EmbeddedServiceLoader</code> to register
	 * @param value the <code>EmbeddedServiceLoader</code> to register
	 */
	public void addEmbeddedServiceLoader( EmbeddedServiceLoader n )
	{
		embeddedServiceLoaders.add( n );
	}

	/**
	 * Returns the <code>EmbeddedServiceLoader</code> instances registered on this interpreter.
	 * @return the <code>EmbeddedServiceLoader</code> instances registered on this interpreter
	 */
	public Collection< EmbeddedServiceLoader > embeddedServiceLoaders()
	{
		return embeddedServiceLoaders;
	}
	
	/**
	 * Makes this <code>Interpreter</code> entering in exiting mode.
	 * When in exiting mode, an interpreter waits for each session to finish
	 * its execution and then terminates gracefully the execution of the entire program.
	 * An interpreter in exiting mode cannot receive any more messages.
	 *
	 * Multiple calls of this method are redundant.
	 *
	 * The fact that the interpreter cannot receive any more messages after
	 * entering exiting mode can cause deadlocks if a session is waiting for a
	 * message to finish its execution. Use this method with caution.
	 */
	public void exit()
	{
		exit( awaitTerminationTimeout );
	}

	/**
	 * Makes this <code>Interpreter</code> entering in exiting mode.
	 * When in exiting mode, an interpreter waits for each session to finish
	 * its execution and then terminates gracefully the execution of the entire program.
	 * An interpreter in exiting mode cannot receive any more messages.
	 *
	 * Multiple calls of this method are redundant.
	 *
	 * The fact that the interpreter cannot receive any more messages after
	 * entering exiting mode can cause deadlocks if a session is waiting for a
	 * message to finish its execution. Use this method with caution.
	 * @param terminationTimeout the timeout for the wait of the termination of running processes
	 */
	public void exit( long terminationTimeout )
	{
		synchronized( this ) {
			if ( exiting ) {
				return;
			} else {
				exiting = true;
			}
		}
		exitingLock.lock();
		try {
			exitingCondition.signalAll();
		} finally {
			exitingLock.unlock();
		}
		timer.cancel();
		checkForExpiredTimeoutHandlers();
		executorService.shutdown();
		commCore.shutdown();
		try {
			executorService.awaitTermination( terminationTimeout, TimeUnit.MILLISECONDS );
		} catch ( InterruptedException e ) {}
	}

	/**
	 * Returns <code>true</code> if this interpreter is in exiting mode, <code>false</code> otherwise.
	 * @return <code>true</code> if this interpreter is in exiting mode, <code>false</code> otherwise
	 * @see #exit()
	 */
	public boolean exiting()
	{
		return exiting;
	}

	/**
	 * Logs an unhandled fault using the logger of this interpreter.
	 * This method is used by sessions that had to terminate due to a fault
	 * which could not be handled, due to a missing fault handler.
	 * @param fault the <code>FaultException</code> that could not be handled
	 */
	public void logUnhandledFault( FaultException fault )
	{
		logInfo( "Thrown unhandled fault: " + fault.faultName() );
	}

	/**
	 * Logs an information message using the logger of this interpreter.
	 * @param message the message to log
	 */
	public void logInfo( String message )
	{
		logger.info( logPrefix + message );
	}
	
	/**
	 * Logs an information message using the logger of this interpreter (logger level: fine).
	 * @param message the message to log
	 */
	public void logFine( String message )
	{
		logger.fine( logPrefix + message );
	}
	
	/**
	 * Logs an information message using the logger of this interpreter (logger level: fine).
	 * @param t the <code>Throwable</code> object whose stack trace has to be logged
	 */
	public void logFine( Throwable t )
	{
		ByteArrayOutputStream bs = new ByteArrayOutputStream();
		t.printStackTrace( new PrintStream( bs ) );
		logger.severe( logPrefix + bs.toString() );
	}

	/**
	 * Logs a severe error message using the logger of this interpreter.
	 * @param message the message to log
	 */<commit_msg>Bugfix: logFine was outputting SEVERE logs.<commit_after>/***************************************************************************
 *   Copyright (C) 2006-2013 by <NAME> <<EMAIL>>     *
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU Library 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 Library 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 details about the authors of this software, see the AUTHORS file. *
 ***************************************************************************/

package jolie;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.PrintStream;
import java.lang.ref.WeakReference;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Deque;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;
import jolie.lang.Constants;
import jolie.lang.parse.OLParseTreeOptimizer;
import jolie.lang.parse.OLParser;
import jolie.lang.parse.ParserException;
import jolie.lang.parse.Scanner;
import jolie.lang.parse.SemanticVerifier;
import jolie.lang.parse.TypeChecker;
import jolie.lang.parse.ast.Program;
import jolie.monitoring.MonitoringEvent;
import jolie.monitoring.events.MonitorAttachedEvent;
import jolie.monitoring.events.OperationStartedEvent;
import jolie.monitoring.events.SessionEndedEvent;
import jolie.monitoring.events.SessionStartedEvent;
import jolie.net.CommChannel;
import jolie.net.CommCore;
import jolie.net.CommMessage;
import jolie.net.SessionMessage;
import jolie.net.ports.OutputPort;
import jolie.process.DefinitionProcess;
import jolie.process.InputOperationProcess;
import jolie.process.SequentialProcess;
import jolie.process.TransformationReason;
import jolie.runtime.ExitingException;
import jolie.runtime.FaultException;
import jolie.runtime.InputOperation;
import jolie.runtime.InvalidIdException;
import jolie.runtime.OneWayOperation;
import jolie.runtime.RequestResponseOperation;
import jolie.runtime.TimeoutHandler;
import jolie.runtime.Value;
import jolie.runtime.ValueVector;
import jolie.runtime.correlation.CorrelationEngine;
import jolie.runtime.correlation.CorrelationError;
import jolie.runtime.correlation.CorrelationSet;
import jolie.runtime.embedding.EmbeddedServiceLoader;
import jolie.tracer.DummyTracer;
import jolie.tracer.PrintingTracer;
import jolie.tracer.Tracer;

/**
 * The Jolie interpreter engine.
 * Multiple Interpreter instances can be run in the same JavaVM;
 * this is used, e.g., for service embedding.
 * @author <NAME>
 */
public class Interpreter
{
	private class InitSessionThread extends SessionThread
	{
		public InitSessionThread( Interpreter interpreter, jolie.process.Process process )
		{
			super( interpreter, process );
		}
		
		@Override
		public boolean isInitialisingThread()
		{
			return true;
		}
		
		@Override
		public void run()
		{
			super.run();

			if ( executionMode == Constants.ExecutionMode.SINGLE ) {
				try {
					mainSession = new SessionThread( getDefinition( "main" ), initExecutionThread );
					correlationEngine.onSingleExecutionSessionStart( mainSession );
					mainSession.addSessionListener( correlationEngine );
				} catch( InvalidIdException e ) {
					assert false;
				}
			}
			
			/*
			 * We need to relay the messages we did not consume during the init procedure.
			 * We do this asynchronously, because calling correlationEngine.onMessageReceive
			 * will trigger a join() on this thread, leading to a deadlock if we
			 * were to call that directly from here.
			 */
			execute( new Runnable() {
				private void pushMessages( Deque< SessionMessage > queue )
				{
					for( SessionMessage message : queue ) {
						try {
							correlationEngine.onMessageReceive( message.message(), message.channel() );
						} catch( CorrelationError e ) {
							logWarning( e );
							try {
								message.channel().send( CommMessage.createFaultResponse( message.message(), new FaultException( "CorrelationError", "The message you sent can not be correlated with any session and can not be used to start a new session." ) ) );
							} catch( IOException ioe ) {
								logSevere( ioe );
							}
						}
					}
				}

				public void run()
				{
					for( Deque< SessionMessage > queue : messageQueues.values() ) {
						pushMessages( queue );
					}
					pushMessages( uncorrelatedMessageQueue );
				}
			});
		}
	}
	
	private static class JolieExecutionThreadFactory implements ThreadFactory {
		private final Interpreter interpreter;

		private static class JolieExecutionThread extends JolieThread
		{
			public JolieExecutionThread( Interpreter interpreter, Runnable r )
			{
				super( interpreter, r );
			}
		}
		
		public JolieExecutionThreadFactory( Interpreter interpreter )
		{
			this.interpreter = interpreter;
		}
		
		public Thread newThread( Runnable r )
		{
			return new JolieExecutionThread( interpreter, r );
		}
	}
	
	public static class SessionStarter
	{
		private final InputOperationProcess guard;
		private final jolie.process.Process body;
		private CorrelationSet correlationInitializer = null;

		public SessionStarter( InputOperationProcess guard, jolie.process.Process body )
		{
			this.guard = guard;
			this.body = body;
		}

		public InputOperationProcess guard()
		{
			return guard;
		}

		public jolie.process.Process body()
		{
			return body;
		}

		public void setCorrelationInitializer( CorrelationSet cset )
		{
			correlationInitializer = cset;
		}

		public CorrelationSet correlationInitializer()
		{
			return correlationInitializer;
		}
	}


	private CommCore commCore;
	private CommandLineParser cmdParser;

	private Map< String, SessionStarter > sessionStarters = new HashMap< String, SessionStarter >();
	private boolean exiting = false;
	private final Lock exitingLock;
	private final Condition exitingCondition;
	private final CorrelationEngine correlationEngine;
	private final List< CorrelationSet > correlationSets =
				new ArrayList< CorrelationSet > ();
	private final Map< String, CorrelationSet > operationCorrelationSetMap = new HashMap< String, CorrelationSet >();
	private Constants.ExecutionMode executionMode = Constants.ExecutionMode.SINGLE;
	private final Value globalValue = Value.createRootValue();
	private final String[] arguments;
	private final Collection< EmbeddedServiceLoader > embeddedServiceLoaders =
			new LinkedList< EmbeddedServiceLoader >();
	private static final Logger logger = Logger.getLogger( "Jolie" );
	
	private final Map< String, DefinitionProcess > definitions =
				new HashMap< String, DefinitionProcess >();
	private final Map< String, OutputPort > outputPorts = new HashMap< String, OutputPort >();
	private final Map< String, InputOperation > inputOperations =
				new HashMap< String, InputOperation>();
	
	private final HashMap< String, Object > locksMap =
				new HashMap< String, Object >();
	
	private final ClassLoader parentClassLoader;
	private final String[] includePaths;
	private final String[] optionArgs;
	private final String logPrefix;
	private final Tracer tracer;
	private final Timer timer;
	// private long inputMessageTimeout = 24 * 60 * 60 * 1000; // 1 day
	private final long persistentConnectionTimeout = 60 * 60 * 1000; // 1 hour
	private final long awaitTerminationTimeout = 60 * 1000; // 1 minute
	// private long persistentConnectionTimeout = 2 * 60 * 1000; // 4 minutes
	// private long persistentConnectionTimeout = 1;

	// 11 is the default initial capacity for PriorityQueue
	private final Queue< WeakReference< TimeoutHandler > > timeoutHandlerQueue =
		new PriorityQueue< WeakReference< TimeoutHandler > >( 11, new TimeoutHandler.Comparator() );
	
	private final ExecutorService timeoutHandlerExecutor = Executors.newSingleThreadExecutor();

	private final String programFilename;
	private final File programDirectory;
	private OutputPort monitor = null;

	public void setMonitor( OutputPort monitor )
	{
		this.monitor = monitor;
		fireMonitorEvent( new MonitorAttachedEvent() );
	}
	
	public boolean isMonitoring()
	{
		return monitor != null;
	}
	
	/*public long inputMessageTimeout()
	{
		return inputMessageTimeout;
	}*/
	
	public String logPrefix()
	{
		return logPrefix;
	}
	
	public Tracer tracer()
	{
		return tracer;
	}
	
	public void fireMonitorEvent( MonitoringEvent event )
	{
		if ( monitor != null ) {
			CommMessage m = CommMessage.createRequest( "pushEvent", "/", MonitoringEvent.toValue( event ) );
			CommChannel channel = null;
			try {
				channel = monitor.getCommChannel();
				channel.send( m );
				CommMessage response;
				do {
					response = channel.recvResponseFor( m );
				} while( response == null );
			} catch( URISyntaxException e ) {
				logWarning( e );
			} catch( IOException e ) {
				logWarning( e );
			} finally {
				if ( channel != null ) {
					try {
						channel.release();
					} catch( IOException e ) {
						logWarning( e );
					}
				}
			}
		}
	}

	public long persistentConnectionTimeout()
	{
		return persistentConnectionTimeout;
	}

	public CorrelationEngine correlationEngine()
	{
		return correlationEngine;
	}

	public void schedule( TimerTask task, long delay )
	{
		if ( exiting == false ) {
			timer.schedule( task, delay );
		}
	}

	public void addTimeoutHandler( TimeoutHandler handler )
	{
		synchronized( timeoutHandlerQueue ) {
			timeoutHandlerQueue.add( new WeakReference< TimeoutHandler >( handler ) );
			if ( timeoutHandlerQueue.size() == 1 ) {
				schedule( new TimerTask() {
					public void run()
					{
						synchronized( timeoutHandlerQueue ) {
							checkForExpiredTimeoutHandlers();
						}
					}
				}, handler.time() - System.currentTimeMillis() + 1 );
			} else {
				checkForExpiredTimeoutHandlers();
			}
		}
	}

	/*public void removeTimeoutHandler( TimeoutHandler handler )
	{
		synchronized( timeoutHandlerQueue ) {
			timeoutHandlerQueue.remove( handler );
			checkForExpiredTimeoutHandlers();
		}
	}*/

	private void checkForExpiredTimeoutHandlers()
	{
		long currentTime = System.currentTimeMillis();
		WeakReference< TimeoutHandler > whandler = timeoutHandlerQueue.peek();
		boolean keepRun = true;
		TimeoutHandler handler;
		while( whandler != null && keepRun ) {
			handler = whandler.get();
			if ( handler == null ) {
				timeoutHandlerQueue.remove();
				whandler = timeoutHandlerQueue.peek();
			} else if ( handler.time() < currentTime || exiting ) {
				// final TimeoutHandler h = handler;
				timeoutHandlerExecutor.execute( handler );
				timeoutHandlerQueue.remove();
				whandler = timeoutHandlerQueue.peek();
			} else {
				keepRun = false;
			}
		}
	}
	
	/**
	 * Returns the option arguments passed to this Interpreter.
	 * @return the option arguments passed to this Interpreter
	 */
	public String[] optionArgs()
	{
		return optionArgs;
	}
	
	/**
	 * Returns the include paths this Interpreter is considering.
	 * @return the include paths this Interpreter is considering
	 */
	public String[] includePaths()
	{
		return includePaths;
	}

	/**
	 * Registers a session starter on this <code>Interpreter</code>.
	 * @param guard the input guard for this session starter
	 * @param process the body of this session starter
	 */
	public void registerSessionStarter( InputOperationProcess guard, jolie.process.Process body )
	{
		sessionStarters.put( guard.inputOperation().id(), new SessionStarter( guard, body ) );
	}
	
	private JolieClassLoader classLoader;
	
	/**
	 * Returns the output ports of this Interpreter.
	 * @return the output ports of this Interpreter
	 */
	public Collection< OutputPort > outputPorts()
	{
		return outputPorts.values();
	}
		
	/**
	 * Returns the InputOperation identified by key.
	 * @param key the name of the InputOperation to return
	 * @return the InputOperation identified by key
	 * @throws jolie.runtime.InvalidIdException if this Interpreter does not own the requested InputOperation
	 * @see InputOperation
	 */
	public InputOperation getInputOperation( String key )
		throws InvalidIdException
	{
		InputOperation ret = inputOperations.get( key );
		if ( ret == null ) {
			throw new InvalidIdException( key );
		}
		return ret;
	}
	
	/**
	 * As {@link #getInputOperation(String) getInputOperation}, with the additional constraint that key must identify a OneWayOperation.
	 * @param key the name of the OneWayOperation to return
	 * @return the OneWayOperation identified by key
	 * @throws jolie.runtime.InvalidIdException if this Interpreter does not own the requested OneWayOperation
	 * @see OneWayOperation
	 */
	public OneWayOperation getOneWayOperation( String key )
		throws InvalidIdException
	{
		InputOperation ret;
		if ( (ret=inputOperations.get( key )) == null || !(ret instanceof OneWayOperation) )
			throw new InvalidIdException( key );
		return (OneWayOperation)ret;
	}
	
	/**
	 * As {@link #getInputOperation(String) getInputOperation}, with the additional constraint that key must identify a RequestResponseOperation.
	 * @param key the name of the RequestResponseOperation to return
	 * @return the RequestResponseOperation identified by key
	 * @throws jolie.runtime.InvalidIdException if this Interpreter does not own the requested RequestResponseOperation
	 * @see RequestResponseOperation
	 */
	public RequestResponseOperation getRequestResponseOperation( String key )
		throws InvalidIdException
	{
		InputOperation ret;
		if ( (ret=inputOperations.get( key )) == null || !(ret instanceof RequestResponseOperation) )
			throw new InvalidIdException( key );
		return (RequestResponseOperation)ret;
	}
	
	/**
	 * Returns the OutputPort identified by key.
	 * @param key the name of the OutputPort to return
	 * @return the OutputPort identified by key
	 * @throws jolie.runtime.InvalidIdException if this Interpreter does not own the requested OutputPort
	 */
	public synchronized OutputPort getOutputPort( String key )
		throws InvalidIdException
	{
		OutputPort ret;
		if ( (ret=outputPorts.get( key )) == null )
			throw new InvalidIdException( key );
		return (OutputPort)ret;
	}
	
	/**
	 * Removes a registered OutputPort.
	 * @param key the name of the OutputPort to remove
	 */
	public synchronized void removeOutputPort( String key )
	{
		outputPorts.remove( key );
	}

	/**
	 * Returns the Definition identified by key.
	 * @param key the name of the Definition to return
	 * @return the Definition identified by key
	 * @throws jolie.runtime.InvalidIdException if this Interpreter does not own the requested Definition
	 */
	public DefinitionProcess getDefinition( String key )
		throws InvalidIdException
	{
		DefinitionProcess ret;
		if ( (ret=definitions.get( key )) == null )
			throw new InvalidIdException( key );
		return ret;
	}

	/**
	 * Registers an <code>OutputPort</code> on this interpreter.
	 * @param key the name of the <code>OutputPort</code> to register
	 * @param value the <code>OutputPort</code> to register
	 */
	public void register( String key, OutputPort value )
	{
		outputPorts.put( key, value );
	}

	/**
	 * Registers a defined sub-routine on this interpreter.
	 * @param key the name of the defined sub-routine to register
	 * @param value the defined sub-routine to register
	 */
	public void register( String key, DefinitionProcess value )
	{
		definitions.put( key, value );
	}

	/**
	 * Registers an <code>InputOperation</code> on this interpreter.
	 * @param key the name of the <code>InputOperation</code> to register
	 * @param value the <code>InputOperation</code> to register
	 */
	public void register( String key, InputOperation value )
	{
		inputOperations.put( key, value );
	}

	/**
	 * Registers an <code>EmbeddedServiceLoader</code> on this interpreter.
	 * @param key the name of the <code>EmbeddedServiceLoader</code> to register
	 * @param value the <code>EmbeddedServiceLoader</code> to register
	 */
	public void addEmbeddedServiceLoader( EmbeddedServiceLoader n )
	{
		embeddedServiceLoaders.add( n );
	}

	/**
	 * Returns the <code>EmbeddedServiceLoader</code> instances registered on this interpreter.
	 * @return the <code>EmbeddedServiceLoader</code> instances registered on this interpreter
	 */
	public Collection< EmbeddedServiceLoader > embeddedServiceLoaders()
	{
		return embeddedServiceLoaders;
	}
	
	/**
	 * Makes this <code>Interpreter</code> entering in exiting mode.
	 * When in exiting mode, an interpreter waits for each session to finish
	 * its execution and then terminates gracefully the execution of the entire program.
	 * An interpreter in exiting mode cannot receive any more messages.
	 *
	 * Multiple calls of this method are redundant.
	 *
	 * The fact that the interpreter cannot receive any more messages after
	 * entering exiting mode can cause deadlocks if a session is waiting for a
	 * message to finish its execution. Use this method with caution.
	 */
	public void exit()
	{
		exit( awaitTerminationTimeout );
	}

	/**
	 * Makes this <code>Interpreter</code> entering in exiting mode.
	 * When in exiting mode, an interpreter waits for each session to finish
	 * its execution and then terminates gracefully the execution of the entire program.
	 * An interpreter in exiting mode cannot receive any more messages.
	 *
	 * Multiple calls of this method are redundant.
	 *
	 * The fact that the interpreter cannot receive any more messages after
	 * entering exiting mode can cause deadlocks if a session is waiting for a
	 * message to finish its execution. Use this method with caution.
	 * @param terminationTimeout the timeout for the wait of the termination of running processes
	 */
	public void exit( long terminationTimeout )
	{
		synchronized( this ) {
			if ( exiting ) {
				return;
			} else {
				exiting = true;
			}
		}
		exitingLock.lock();
		try {
			exitingCondition.signalAll();
		} finally {
			exitingLock.unlock();
		}
		timer.cancel();
		checkForExpiredTimeoutHandlers();
		executorService.shutdown();
		commCore.shutdown();
		try {
			executorService.awaitTermination( terminationTimeout, TimeUnit.MILLISECONDS );
		} catch ( InterruptedException e ) {}
	}

	/**
	 * Returns <code>true</code> if this interpreter is in exiting mode, <code>false</code> otherwise.
	 * @return <code>true</code> if this interpreter is in exiting mode, <code>false</code> otherwise
	 * @see #exit()
	 */
	public boolean exiting()
	{
		return exiting;
	}

	/**
	 * Logs an unhandled fault using the logger of this interpreter.
	 * This method is used by sessions that had to terminate due to a fault
	 * which could not be handled, due to a missing fault handler.
	 * @param fault the <code>FaultException</code> that could not be handled
	 */
	public void logUnhandledFault( FaultException fault )
	{
		logInfo( "Thrown unhandled fault: " + fault.faultName() );
	}

	/**
	 * Logs an information message using the logger of this interpreter.
	 * @param message the message to log
	 */
	public void logInfo( String message )
	{
		logger.info( logPrefix + message );
	}
	
	/**
	 * Logs an information message using the logger of this interpreter (logger level: fine).
	 * @param message the message to log
	 */
	public void logFine( String message )
	{
		logger.fine( logPrefix + message );
	}
	
	/**
	 * Logs an information message using the logger of this interpreter (logger level: fine).
	 * @param t the <code>Throwable</code> object whose stack trace has to be logged
	 */
	public void logFine( Throwable t )
	{
		ByteArrayOutputStream bs = new ByteArrayOutputStream();
		t.printStackTrace( new PrintStream( bs ) );
		logger.fine( logPrefix + bs.toString() );
	}

	/**
	 * Logs a severe error message using the logger of this interpreter.
	 * @param message the message to log
	 */
{'fraction_non_alphanumeric': 0.0915011244377811, 'fraction_numerical': 0.003045352323838081, 'mean_word_length': 3.617522985397512, 'pattern_counts': {'":': 0, '<': 145, '<?xml version=': 0, '>': 143, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 26, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5754917', 'n_tokens_mistral': 13064, 'n_tokens_neox': 11343, 'n_words': 3825}