/*
** Module   : JobQueue.java
** Abstract : queue tests and allow them to be dequeued based on dependencies
**
** Copyright (c) 2009-2024, Golden Code Development Corporation.
**
** -#- -I- --Date-- -----------------------Description------------------------
** 001 GES 20090318 Queue tests and allow them to be dequeued based on
**                  dependencies.
** 002 TJD 20220809 Support for aborting the queue
** 003 TJD 20220817 Implement performance test mode
** 004 TJD 20230119 Allow restarting of Driver threads
** 005 TJD 20240201 Benchmark mode enhancements
*/

/*
** This program is free software: you can redistribute it and/or modify
** it under the terms of the GNU Affero General Public License as
** published by the Free Software Foundation, either version 3 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 Affero General Public License for more details.
**
** You should have received a copy of the GNU Affero General Public License
** along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

package com.goldencode.harness;

import java.util.*;
import java.util.concurrent.atomic.*;

/**
 * Queue objects and allow them to be dequeued based on dependencies. The
 * dependency processing is isolated into {@link SubQueue} instances, which
 * can each implement a portion of the given queue with specific or arbitrary
 * dependency processing. The responsibility of the job queue itself is to
 * process the list of sub-queues to answer queries or to dequeue elements.
 * 
 * @param   <E>
 *          Type of job elements in the queue.
 */
public class JobQueue<E>
{
   /** Queue containing the elements. */
   private final List<SubQueue<E>> queue;
   
   /** Synchronization primitive (more granular than "this"). */
   private final Object lock = new Object();
   
   /** Flag indicating queue abort was requested */
   private final AtomicBoolean abort = new AtomicBoolean(false);
   
   /** Flag preventing starting new threads */
   volatile private boolean isEmptyForNewThreads = false;
   
   /**
    * Construct an instance.
    *
    * @param    queue
    *           Queue from which elements are obtained.
    */
   public JobQueue(List<SubQueue<E>> queue)
   {
      this.queue = queue;
   }
   
   /**
    * Obtain the next element from the queue, blocking until an element is
    * available. Elements can exist but be unavailable due to dependencies that
    * have not yet completed.
    *
    * @return   The next element in the queue or <code>null</code> if there
    *           are no elements to return.
    */
   public E dequeue()
   {
      SubQueue<E> ready = null;
      
      int sz = 0;
      
      // check if queue abort was requested
      if (abort.get())
         return null;
      
      synchronized (lock)
      {
         if (queue == null)
            return null;
         
         sz = queue.size();
      }
         
      outer:
      while (sz > 0)
      {
         // make a single pass through the entire queue to see if there is
         // anything that can be dequeued
         synchronized (lock)
         {
            ListIterator<SubQueue<E>> list = queue.listIterator();
            int ignoredSubqueues = 0;
            // walk through all sub-queues
            while (list.hasNext())
            {
               SubQueue<E> subQueue = list.next();
               
               // cleanup dead queue elements
               if (subQueue.isEmpty())
               {
                  // eliminates the current sub-queue (this can be safely done
                  // during iteration because we are using the list iterator)
                  list.remove();
                  
                  continue;
               }
               
               // do not go past any barrier sub-queues
               if (subQueue.isBarrier())
               {
                  if (queue.indexOf(subQueue) == ignoredSubqueues)
                  {
                     // all previous sub-queues are finished, remove the
                     // barrier and allow processing to continue
                     list.remove();
                     
                     continue;
                  }
                  
                  // make the thread wait
                  break;
               }
               
               // sub-queues may contain elements which cannot be executed
               // immediately due to waiting on dependencies to complete
               if (subQueue.isReady())
               {
                  ready = subQueue;
                  
                  // reserve the next element and allow us to safely exit the
                  // synchronized block
                  if (ready.reserve())
                  {
                     break outer;
                  }
                  // if not ready - repeat the cycle until its ready
               }
               // check if subqueue can be ignored for this thread
               if (subQueue.isIgnored())
               {
                  ignoredSubqueues ++;
               }
            }
            
            sz = queue.size() - ignoredSubqueues;
         }
            
         // no sub-queues are ready yet, the current thread must wait for an
         // element to be available; this is done outside the synchronized
         // block so that other threads can enter and look too (otherwise
         // any sub-queue blocking on something thread-specific may deadlock
         // everything)
         if (sz > 0)
         {
            try
            {
               if (!Harness.isBenchmarkMode())
               {
                  Thread.sleep(500);
               }
               else
               {
                  Thread.sleep(1);
               }
               
            }
            
            catch (InterruptedException ie)
            {
               // allow null to be returned
               break;
            }
         }
         else
         {
            // if any of thread has reached here no new threads should be started
            isEmptyForNewThreads = true;
         }
      }
      
      // the dequeuing is intentionally done outside of the synchronized block
      // since the dequeue() is a potentially blocking operation; that is why
      // there is a 2-step process where the reserve() is called inside the
      // synchronized block and the dequeue() is called outside; if this thread
      // called dequeue() inside the synchronized block and it did block inside
      // this call, then all other threads that call this method to dequeue
      // from the job queue would block (since we would own the monitor for
      // the synchronized block)
      return (ready == null) ? null : ready.dequeue();
   }
   
   /**
    * Request abort of queue in case of global test set errors
    */
   public void abort()
   {
      abort.set(true);
   }

   /**
    * Check if given queue is empty or it should treated like empty for starting new threads
    *
    * @return   <code>true</code> if the queue is empty
    */
   public boolean isEmpty()
   {
      synchronized (lock)
      {
         return (queue == null || (queue.size() == 0) || isEmptyForNewThreads);
      }
   }
}
