/*
    Copyright (C) 2006 Grid Systems, S.A.
   
    This library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Lesser General Public
    License as published by the Free Software Foundation; either
    version 2.1 of the License, or (at your option) any later version.
   
    This library 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
   Lesser General Public License for more details.
  
   You should have received a copy of the GNU Lesser General Public
   License along with this library; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
  */
  package com.gridsystems.nextgrid.api.pom;
  
  import java.util.BitSet;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.locks.Condition;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReentrantLock;
  
  import nextgrid.api.pem.ProcessEvent;
  import nextgrid.api.pem.ProcessListener;
  import nextgrid.api.pom.Process;
  import nextgrid.api.pom.ProcessController;
  import nextgrid.api.pom.ProcessException;
  import nextgrid.api.pom.Reference;
  import nextgrid.api.pom.UnorderedListProcess;
  
  /**
   * Control process that executes its children in a non-deterministic order,
   * but never concurrently.
   *
   * @author Rodrigo Ruiz
   */
  public final class UnorderedListProcessImpl extends ControlProcessImpl
    implements UnorderedListProcess {
  
    /**
     * <code>serialVersionUID</code> attribute.
     */
    private static final long serialVersionUID = -8766837087826444925L;
  
    /**
     * Timeout used for process notification wait.
     */
    private static final long WAIT_TIMEOUT = 200L;
  
    /**
     * Flag set for knowing which child processes have already been enacted.
     */
    private BitSet enacted = new BitSet();
  
    /**
     * Used for "process finished" notification.
     */
    private ProcessFinishedNotifier notifier = new ProcessFinishedNotifier();
  
    /**
     * Creates a new instance.
     */
    public UnorderedListProcessImpl() {
      super(-1);
    }
  
    /**
     * {@inheritDoc}
     */
    @Override
    protected void doValidate(ValidationType when) throws ProcessException {
      // No extra validation required here
    }
  
    /**
     * {@inheritDoc}
     */
    public void run(ProcessContext ctx) throws ProcessException, InterruptedException {
      fireProcessStarted();
      waitForInputs();
  
      Process[] children = getChildren();
      int count = (children == null) ? 0 : children.length;
  
      while (this.enacted.cardinality() < count) {
        if (!ctx.isRunning()) {
          return;
        }
  
        int index = selectChildToEnact();
        while (index == -1 && ctx.isRunning()) {
          if (waitForProcessFinalization(ctx)) {
            index = selectChildToEnact();
            if (ENACTOR_LOG.isDebugEnabled()) {
              ENACTOR_LOG.debug("Child at position " + index + " unlocked");
            }
         }
       }
 
       if (!ctx.isRunning()) {
         return;
       }
 
       ProcessController controller = children[index].enact(ctx);
       controller.run();
       this.enacted.set(index);
     }
 
     if (ctx.isRunning()) {
       fireProcessFinished();
     }
 
   }
 
   /**
    * Waits until a process in the same workflow as this instance has finished,
    * or the enaction is cancelled / paused.
    *
    * @param ctx The process context
    * @return <tt>true</tt> if a process has finished its execution
    * @throws InterruptedException If the thread has been interrupted
    */
   private boolean waitForProcessFinalization(ProcessContext ctx)
     throws InterruptedException {
     this.findRoot().addListener(this.notifier);
 
     while (ctx.isRunning()) {
       if (this.notifier.getCondition().await(WAIT_TIMEOUT, TimeUnit.MILLISECONDS)) {
         return true;
       }
     }
     return false;
   }
 
   /**
    * {@inheritDoc}
    */
   @Override protected void doReset() {
     this.enacted.clear();
   }
 
   /**
    * {@inheritDoc}
    */
   @Override public String toString() {
     return "UnorderedListProcess#" + this.getId();
   }
 
   /**
    * Searches a process still not enacted for which all its inputs are
    * available.
    * <p>
    * If none is found, it blocks until the process is cancelled, or
    * a child process is "unlocked".
    *
    * @return The index of the child process found, or -1 if none
    */
   private int selectChildToEnact() {
     Process[] children = getChildren();
     int count = (children == null) ? 0 : children.length;
 
     for (int i = 0; i < count; i++) {
       if (!this.enacted.get(i)) {
         if (!isLocked(children[i])) {
           return i;
         }
       }
     }
 
     return -1;
   }
 
   /**
    * Gets if the passed process is "locked". A process is locked if at least
    * one of its used inputs is not available.
    *
    * @param p The process to test
    * @return <tt>true</tt> if one of its used inputs is not available
    */
   private boolean isLocked(Process p) {
     String[] names = p.getUsedInputNames();
     for (String name : names) {
       Reference<?> ref = p.getInput(name);
       if (!ref.isAvailable()) {
         return true;
       }
     }
     return false;
   }
 
   /**
    * Notifies process finalisation events through a Condition.
    *
    * @author Rodrigo Ruiz
    */
   static class ProcessFinishedNotifier implements ProcessListener {
 
     /**
      * Serial Version UID.
      */
     private static final long serialVersionUID = -4488491121616206883L;
 
     /**
      * Lock used for creating the condition.
      */
     private Lock lock = new ReentrantLock();
 
     /**
      * Managed condition.
      */
     private Condition cond = lock.newCondition();
 
     /**
      * Gets the condition managed by this instance.
      *
      * @return The managed condition
      */
     public Condition getCondition() {
       return this.cond;
     }
 
     /**
      * {@inheritDoc}
      */
     public void processEvaluated(ProcessEvent event) {
     }
 
     /**
      * {@inheritDoc}
      */
     public void processFinished(ProcessEvent event) {
       this.cond.signalAll();
     }
 
     /**
      * {@inheritDoc}
      */
     public void processFailed(ProcessEvent event) {
     }
 
     /**
      * {@inheritDoc}
      */
     public void processSelected(ProcessEvent event) {
     }
 
     /**
      * {@inheritDoc}
      */
     public void processStarted(ProcessEvent event) {
     }
 
     /**
      * {@inheritDoc}
      */
     public boolean isListeningToDescendants() {
       return true;
     }
 
   }
 }