package de.renew.watch;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import de.renew.engine.searcher.Searcher;
import de.renew.engine.searcher.TriggerCollection;
import de.renew.engine.searcher.TriggerCollectionImpl;
import de.renew.engine.searcher.Triggerable;
import de.renew.engine.simulator.ExecuteFinder;
import de.renew.engine.thread.SimulationThreadPool;
import de.renew.net.NetInstance;
import de.renew.simulatorontology.simulation.StepIdentifier;
import de.renew.unify.ICalculationChecker;
import de.renew.unify.Impossible;
import de.renew.unify.Tuple;
import de.renew.unify.Variable;
import de.renew.util.DelayedFieldOwner;
import de.renew.util.EmptyIterator;
import de.renew.util.RenewObjectInputStream;
import de.renew.util.RenewObjectOutputStream;
import de.renew.util.Semaphor;


/**
 * Allows monitoring of synchronous channel uplinks in a given net instance from Java code. Users of this class must implement the {@link ChannelWatcher}
 * interface.
 *
 * @author Olaf Kummer
 * @author Michael Duvigneau (serialization)
 * @serial This class is not intended to be serialized.  However, serialization will occur when a simulation state is stored.  Hence, instances of this
 *     class do not store state or references to other objects with one notable exception.  The simulation-relevant collection of triggers is stored in order to
 *     cleanly cut the connection on deserialization.
 **/
public class ChannelSupervisor implements Triggerable, Serializable, DelayedFieldOwner {

    static org.apache.log4j.Logger _logger =
        org.apache.log4j.Logger.getLogger(ChannelSupervisor.class);
    static final long serialVersionUID = 8023504578713732407L;

    /**
     * A supervisor that might not have informed its watcher about the possible bindings correctly will be inserted into this list. Its set of bindings will be
     * cleaned up by the search thread.
     * <p>
     * Because supervisors may be inserted and remove from the dirty list asynchronously, it is required to synchronize on the
     * <code>allSupervisors</code> list before modifying it.
     **/
    private static final LinkedList<ChannelSupervisor> DIRTY_SUPERVISORS =
        new LinkedList<ChannelSupervisor>();
    private static final List<ChannelSupervisor> ALL_SUPERVISORS =
        Collections.synchronizedList(new LinkedList<ChannelSupervisor>());

    /**
     * This semaphor counts the number of elements in the dirty list. It controls the search process.
     **/
    private static Semaphor _dirtySem = new Semaphor();
    private static SupervisorThread _supervisorThread;

    private static class SupervisorThread implements Runnable {

        private boolean _active = true;

        @Override
        public void run() {
            if (_logger.isDebugEnabled()) {
                _logger.debug("ChannelSupervisor: Thread " + Thread.currentThread() + " started.");
            }
            while (_active) {
                // Wait until the dirty list is non-empty.
                _dirtySem.P();
                if (!_active) {
                    // The thread has been marked for termination inbetween.
                    // Therefore we should not handle this event.
                    // Put it back to the Semaphor...
                    _dirtySem.V();
                } else {
                    // Extract one element.
                    ChannelSupervisor supervisor;
                    synchronized (ALL_SUPERVISORS) {
                        supervisor = DIRTY_SUPERVISORS.getFirst();
                        DIRTY_SUPERVISORS.removeFirst();
                    }

                    // Pass control to that supervisor for an update.
                    supervisor.update();
                }
            }
            if (_logger.isDebugEnabled()) {
                _logger
                    .debug("ChannelSupervisor: Thread " + Thread.currentThread() + " terminated.");
            }
        }

        public void terminate() {
            _logger.debug("ChannelSupervisor: Thread marked for termination.");
            _active = false;
        }
    }

    private transient final NetInstance _netInstance;
    private transient final ChannelWatcher _watcher;
    private transient final String _channel;
    private transient final Class<?> _lateClass;

    /**
     * Since the <code>ChannelSupervisor</code> acts as a {@link Triggerable} within the simulation, it needs to keep a list of the triggers it is interested
     * in.
     *
     * @serial The contents of this collection is needed to cut off all connections to other simulation elements after deserialization.
     **/
    private TriggerCollection _triggers;

    /**
     * Marks whether this object belongs to the current simulation. The value is always <code>true</code>, unless the object has been deserialized from a
     * previous simulation state. Invalid ChannelSupervisors do nothing except cutting all connections to other simulation objects on their first invocation.
     * <p>
     * This field may not be modified.  Unfortunately, an assignment on deserialization requires omission of the <code>final</code> modifier.
     * </p>
     **/
    private transient boolean _valid;

    /**
     * Constructs a new instance of {@link ChannelSupervisor}. This constructor also adds the object instance to the list of all supervisors, initializes its
     * triggers collection, and invokes a proposal search if valid.
     *
     * @param netInstance the corresponding {@link NetInstance} for this supervisor, representing the net system.
     * @param watcher the {@link ChannelWatcher} that oversees channel interactions and bindings.
     * @param channel the name of the associated communication channel.
     * @param lateClass the Class object representing the specification for any late evaluation interactions.
     */
    public ChannelSupervisor(
        NetInstance netInstance, ChannelWatcher watcher, String channel, Class<?> lateClass)
    {
        this._netInstance = netInstance;
        this._watcher = watcher;
        this._channel = channel;
        this._lateClass = lateClass;
        this._valid = true;

        _triggers = new TriggerCollectionImpl(this);
        synchronized (ALL_SUPERVISORS) {
            ALL_SUPERVISORS.add(0, this);
        }
        if (_logger.isDebugEnabled()) {
            _logger.debug("New " + toString() + ".");
        }
        proposeSearch();
    }

    /**
     * Executes a process based on the provided parameters. It leverages a finder mechanism to initiate a search and perform actions if the operation is
     * successful.
     *
     * @param early an object representing the early stage data or metadata required for execution.
     * @param late an object representing the late stage data or metadata required for execution.
     * @param stepIdentifier identifies a specific step within a simulation or process.
     * @return true if the execution was successful, false otherwise.
     */
    // This method is currently unused.
    public boolean execute(
        final Object early, final Object late, final StepIdentifier stepIdentifier)
    {
        if (_valid) {
            Future<Boolean> future =
                SimulationThreadPool.getCurrent().submitAndWait(new Callable<Boolean>()
                {
                    @Override
                    public Boolean call() throws Exception {
                        ExecuteFinder finder = new ExecuteFinder();
                        Tuple paramTuple = new Tuple(new Object[] { early, late }, null);
                        Variable variable = new Variable(paramTuple, null);

                        Searcher searcher = new Searcher();
                        searcher
                            .initiatedSearch(_netInstance, _channel, variable, false, finder, null);

                        boolean success = finder.isCompleted();
                        if (success) {
                            finder.execute(stepIdentifier, true);
                        }

                        return success;
                    }
                });
            try {
                return future.get();
            } catch (InterruptedException e) {
                _logger.error("Timeout while waiting for simulation thread to finish", e);
            } catch (ExecutionException e) {
                _logger.error("Simulation thread threw an exception", e);
            }
        } else {
            if (_logger.isDebugEnabled()) {
                _logger.debug(
                    "Call to execute on invalid ChannelSupervisor:" + toString()
                        + ", parameters: early=" + early + ", late=" + late + ", step="
                        + stepIdentifier,
                    new Throwable("DEBUG STACKTRACE"));
            }
        }

        // We should never return nothing but some error occured before.
        return false;

    }

    @Override
    public TriggerCollection triggers() {
        return _triggers;
    }

    @Override
    public void proposeSearch() {
        if (_valid) {
            synchronized (ALL_SUPERVISORS) {
                DIRTY_SUPERVISORS.addLast(this);
                _triggers.clear();
            }
            _dirtySem.V();
        } else {
            if (_logger.isDebugEnabled()) {
                _logger.debug(
                    "Call to proposeSearch on invalid ChannelSupervisor:" + toString()
                        + ". Clearing trigger collection.");
            }
            _triggers.clear();
        }
    }

    void update() {
        if (_valid) {
            final ChannelSupervisor object = this;
            if (_logger.isDebugEnabled()) {
                _logger.debug("Initiating update on " + toString() + "...");
            }
            Future<Object> future =
                SimulationThreadPool.getCurrent().submitAndWait(new Callable<Object>()
                {
                    @Override
                    public Object call() throws Exception {
                        // Try to invoke the channel
                        //   :channel(early,late);
                        // where the early argument must be known after the search
                        // and the late argument must be freely assignable.
                        Tuple arguments = new Tuple(2);
                        Variable earlyVariable = new Variable(arguments.getComponent(0), null);

                        // We create a new searcher for this search process.
                        Searcher searcher = new Searcher();

                        // The searcher has to be rigged to satisfy the requirements
                        // on the channels: the first channel must be completely known
                        // after the search and the second channel must be freely assignable.
                        try {
                            ICalculationChecker calculationChecker =
                                searcher.getCalculationChecker();
                            // The first argument must be computed early
                            // during the search for a binding.
                            calculationChecker.addEarlyVariable(earlyVariable, null);

                            // The last argument must not be computed early.
                            calculationChecker
                                .addCalculated(_lateClass, arguments.getComponent(1), null, null);
                        } catch (Impossible e) {
                            throw new RuntimeException("Error in unification algorithm.");
                        }

                        // We must not reuse the old finder, because its set of
                        // permissible values cannot be reset.
                        ParamFinder finder = new ParamFinder(earlyVariable);

                        // We want to find a transition within the specified
                        // net instance that can fire.
                        searcher.initiatedSearch(
                            _netInstance, _channel, new Variable(arguments, null), false, finder,
                            object);
                        synchronized (this) {
                            if (_watcher != null) {
                                _watcher.bindingsCalculated(finder.iterator());
                            }
                        }
                        return null;
                    }
                });
            try {
                future.get();
            } catch (InterruptedException e) {
                _logger.error("Timeout while waiting for simulation thread to finish", e);
            } catch (ExecutionException e) {
                _logger.error("Simulation thread threw an exception", e);
            }
        } else {
            if (_logger.isDebugEnabled()) {
                _logger.debug(
                    "Call to update on invalid ChannelSupervisor:" + toString(),
                    new Throwable("DEBUG STACKTRACE"));
            }
        }
    }

    private synchronized void discard() {
        if (_watcher != null) {
            _watcher.bindingsCalculated(EmptyIterator.INSTANCE);
        }
    }

    /**
     * Resets the state of all registered {@link ChannelSupervisor} instances and their associated resources.
     */
    public static void reset() {
        synchronized (ALL_SUPERVISORS) {
            _logger.debug("Resetting all ChannelSupervisors.");
            DIRTY_SUPERVISORS.clear();
            Iterator<ChannelSupervisor> iterator = ALL_SUPERVISORS.iterator();
            while (iterator.hasNext()) {
                ChannelSupervisor supervisor = iterator.next();
                supervisor.discard();
            }
            ALL_SUPERVISORS.clear();
            if (_supervisorThread != null) {
                _supervisorThread.terminate();
                _supervisorThread = null;
            }
        }
    }

    /**
     * Activates the {@link ChannelSupervisor} system by ensuring a new {@link SupervisorThread} instance is created and started, if not already active.
     */
    public static void activate() {
        synchronized (ALL_SUPERVISORS) {
            reset();
            _logger.debug("Activating new ChannelSupervisor thread.");
            if (_supervisorThread == null) {
                _supervisorThread = new SupervisorThread();
                SimulationThreadPool.getCurrent().execute(_supervisorThread);
            }
        }
    }


    /**
     * Serialization method, behaves like default writeObject method except storing the not-really-transient field triggers. If the stream used is a
     * RenewObjectOutputStream, this field is delayed to cut down recursion depth. The domain trace feature of this special stream is also used.
     * {@link de.renew.util.RenewObjectOutputStream}
     * @param out the {@link ObjectOutputStream} used for serialization.
     * @throws IOException if an I/O error occurs.
     **/
    private void writeObject(ObjectOutputStream out) throws IOException {
        RenewObjectOutputStream rOut = null;
        if (out instanceof RenewObjectOutputStream) {
            rOut = (RenewObjectOutputStream) out;
        }
        if (rOut != null) {
            rOut.beginDomain(this);
        }
        out.defaultWriteObject();
        if (rOut != null) {
            rOut.delayedWriteObject(_triggers, this);
            rOut.endDomain(this);
        } else {
            out.writeObject(_triggers);
        }
    }


    /**
     * After deserialization, this object is not usable any more.
     * <p>
     * Most fields contain null references since they have not been stored on serialization. The deserialized <code>ChannelSupervisor</code> instance is
     * <em>not</em> registered with the {@link #ALL_SUPERVISORS} list.
     * </p>
     * The method also restores the not-really-transient field
     * <code>triggers</code>, <b>if</b> the
     * stream used is <b>not</b> a RenewObjectInputStream.
     * @param in the {@link ObjectInputStream} used for deserialization.
     * @throws IOException if an I/O error occurs.
     * @throws ClassNotFoundException if the class of a serialized object could not be found.
     **/
    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
        in.defaultReadObject();
        _valid = false;
        if (_logger.isDebugEnabled()) {
            _logger.debug("Deserializing " + toString() + "...");
        }
        if (in instanceof RenewObjectInputStream) {
            // Do nothing, the fields will be
            // reassigned by the stream soon.
        } else {
            reassignField(in.readObject());
        }
    }

    /**
     * Method used on deserialization by RenewObjectInputStream. Reassigns values to the not-really-transient fields, one at a time.
     * <p>
     * The contents of the {@link TriggerCollection _triggers} field is used to break connections from this object to all other simulation elements.
     * </p>
     **/
    @Override
    public void reassignField(Object value) throws java.io.IOException {
        if (value instanceof TriggerCollection) {
            _triggers = (TriggerCollection) value;
            if (_logger.isDebugEnabled()) {
                _logger.debug("Deserialization completed:  " + toString());
            }
        }
    }

    @Override
    public String toString() {
        final int sbSize = 1000;
        final String variableSeparator = ", ";
        final StringBuffer sb = new StringBuffer(sbSize);
        sb.append(this.getClass().getName());
        sb.append("(");
        if (_valid) {
            sb.append("valid");
            sb.append(variableSeparator);
            sb.append("netInstance=").append(_netInstance);
            sb.append(variableSeparator);
            sb.append("watcher=").append(_watcher);
            sb.append(variableSeparator);
            sb.append("channel=").append(_channel);
            sb.append(variableSeparator);
            sb.append("lateClass=").append(_lateClass);
        } else {
            sb.append("invalid");
        }
        sb.append(variableSeparator);
        sb.append("triggers=").append(_triggers);
        sb.append(")");
        return sb.toString();
    }
}
