[Cuebot] Add FIFO scheduling capability

cuebot/src/main/java/com/imageworks/spcue/dao/DispatcherDao.java

@@ -82,7 +82,7 @@ public interface DispatcherDao {
      * @param numJobs
      * @return
      */
-    Set<String> findDispatchJobsForAllShows(DispatchHost host, int numJobs);
+    List<String> findDispatchJobsForAllShows(DispatchHost host, int numJobs);
 
     /**
      * Return a list of jobs which could use resources of the specified
@@ -92,7 +92,7 @@ public interface DispatcherDao {
      * @param numJobs
      * @return
      */
-    Set<String> findDispatchJobs(DispatchHost host, int numJobs);
+    List<String> findDispatchJobs(DispatchHost host, int numJobs);
 
     /**
     * Return a list of jobs which could use resources of the specified
@@ -102,7 +102,7 @@ public interface DispatcherDao {
     * @param numJobs
     * @return
     */
-    Set<String> findDispatchJobs(DispatchHost host, GroupInterface g);
+    List<String> findDispatchJobs(DispatchHost host, GroupInterface g);
 
     /**
      * Finds an under proced job if one exists and returns it,
@@ -131,7 +131,7 @@ public interface DispatcherDao {
     * @param numJobs
     * @return
     */
-   Set<String> findDispatchJobs(DispatchHost host, ShowInterface show, int numJobs);
+   List<String> findDispatchJobs(DispatchHost host, ShowInterface show, int numJobs);
 
    /**
     * Find a list of local dispatch jobs.
@@ -162,6 +162,20 @@ List<DispatchFrame> findNextDispatchFrames(LayerInterface layer, VirtualProc pro
     */
    List<DispatchFrame> findNextDispatchFrames(LayerInterface layer, DispatchHost host,
                                               int limit);
+
+   /**
+    * Return whether FIFO scheduling is enabled or not in the same priority for unittest.
+    *
+    * @return
+    */
+   boolean getFifoSchedulingEnabled();
+
+   /**
+    * Set whether FIFO scheduling is enabled or not in the same priority for unittest.
+    *
+    * @param fifoSchedulingEnabled
+    */
+   void setFifoSchedulingEnabled(boolean fifoSchedulingEnabled);
 }
 
 

cuebot/src/main/java/com/imageworks/spcue/dao/postgres/DispatchQuery.java

@@ -112,6 +112,22 @@ public class DispatchQuery {
                 "AND job.pk_folder                  = ? ");
 
 
+    private static final String replaceQueryForFifo(String query) {
+        return query
+            .replace(
+                "JOBS_BY",
+                "JOBS_FIFO_BY")
+            .replace(
+                "ORDER BY job_resource.int_priority DESC",
+                "ORDER BY job_resource.int_priority DESC, job.ts_started ASC")
+            .replace(
+                "WHERE rank < ?",
+                "WHERE rank < ? ORDER BY rank");
+    }
+
+    public static final String FIND_JOBS_FIFO_BY_SHOW = replaceQueryForFifo(FIND_JOBS_BY_SHOW);
+    public static final String FIND_JOBS_FIFO_BY_GROUP = replaceQueryForFifo(FIND_JOBS_BY_GROUP);
+
     /**
      * Dispatch a host in local booking mode.
      */

cuebot/src/main/java/com/imageworks/spcue/dao/postgres/DispatcherDaoJdbc.java

@@ -20,6 +20,7 @@
 
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.LinkedHashSet;
 import java.util.LinkedList;
@@ -28,6 +29,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.log4j.Logger;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.env.Environment;
 import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.core.support.JdbcDaoSupport;
 
@@ -52,6 +55,8 @@
 import static com.imageworks.spcue.dao.postgres.DispatchQuery.FIND_JOBS_BY_GROUP;
 import static com.imageworks.spcue.dao.postgres.DispatchQuery.FIND_JOBS_BY_LOCAL;
 import static com.imageworks.spcue.dao.postgres.DispatchQuery.FIND_JOBS_BY_SHOW;
+import static com.imageworks.spcue.dao.postgres.DispatchQuery.FIND_JOBS_FIFO_BY_GROUP;
+import static com.imageworks.spcue.dao.postgres.DispatchQuery.FIND_JOBS_FIFO_BY_SHOW;
 import static com.imageworks.spcue.dao.postgres.DispatchQuery.FIND_LOCAL_DISPATCH_FRAME_BY_JOB_AND_HOST;
 import static com.imageworks.spcue.dao.postgres.DispatchQuery.FIND_LOCAL_DISPATCH_FRAME_BY_JOB_AND_PROC;
 import static com.imageworks.spcue.dao.postgres.DispatchQuery.FIND_LOCAL_DISPATCH_FRAME_BY_LAYER_AND_HOST;
@@ -124,6 +129,27 @@ public List<SortableShow> getShows() {
     private final ConcurrentHashMap<String, ShowCache> bookableShows =
         new ConcurrentHashMap<String, ShowCache>();
 
+    /**
+     * Whether or not to enable FIFO scheduling in the same priority.
+     */
+    private boolean fifoSchedulingEnabled;
+
+    @Autowired
+    public DispatcherDaoJdbc(Environment env) {
+        fifoSchedulingEnabled = env.getProperty(
+            "dispatcher.fifo_scheduling_enabled", Boolean.class, false);
+    }
+
+    @Override
+    public boolean getFifoSchedulingEnabled() {
+        return fifoSchedulingEnabled;
+    }
+
+    @Override
+    public void setFifoSchedulingEnabled(boolean fifoSchedulingEnabled) {
+        this.fifoSchedulingEnabled = fifoSchedulingEnabled;
+    }
+
     /**
      * Returns a sorted list of shows that have pending jobs
      * which could benefit from the specified allocation.
@@ -149,8 +175,8 @@ else if (cached.isExpired()) {
         return bookableShows.get(key).shows;
     }
 
-    private Set<String> findDispatchJobs(DispatchHost host, int numJobs, boolean shuffleShows) {
-        LinkedHashSet<String> result = new LinkedHashSet<String>();
+    private List<String> findDispatchJobs(DispatchHost host, int numJobs, boolean shuffleShows) {
+        ArrayList<String> result = new ArrayList<String>();
         List<SortableShow> shows = new LinkedList<SortableShow>(getBookableShows(host));
         // shows were sorted. If we want it in random sequence, we need to shuffle it.
         if (shuffleShows) {
@@ -185,7 +211,7 @@ private Set<String> findDispatchJobs(DispatchHost host, int numJobs, boolean shu
             }
 
             result.addAll(getJdbcTemplate().query(
-                    FIND_JOBS_BY_SHOW,
+                    fifoSchedulingEnabled ? FIND_JOBS_FIFO_BY_SHOW : FIND_JOBS_BY_SHOW,
                     PKJOB_MAPPER,
                     s.getShowId(), host.getFacilityId(), host.os,
                     host.idleCores, host.idleMemory,
@@ -208,27 +234,26 @@ private Set<String> findDispatchJobs(DispatchHost host, int numJobs, boolean shu
     }
 
     @Override
-    public Set<String> findDispatchJobsForAllShows(DispatchHost host, int numJobs) {
+    public List<String> findDispatchJobsForAllShows(DispatchHost host, int numJobs) {
         return findDispatchJobs(host, numJobs, true);
     }
 
     @Override
-    public Set<String> findDispatchJobs(DispatchHost host, int numJobs) {
+    public List<String> findDispatchJobs(DispatchHost host, int numJobs) {
         return findDispatchJobs(host, numJobs, false);
     }
 
     @Override
-    public Set<String> findDispatchJobs(DispatchHost host, GroupInterface g) {
-        LinkedHashSet<String> result = new LinkedHashSet<String>(5);
-        result.addAll(getJdbcTemplate().query(
-                FIND_JOBS_BY_GROUP,
+    public List<String> findDispatchJobs(DispatchHost host, GroupInterface g) {
+        List<String> result = getJdbcTemplate().query(
+                fifoSchedulingEnabled ? FIND_JOBS_FIFO_BY_GROUP : FIND_JOBS_BY_GROUP,
                 PKJOB_MAPPER,
                 g.getGroupId(),host.getFacilityId(), host.os,
                 host.idleCores, host.idleMemory,
                 threadMode(host.threadMode),
                 host.idleGpus,
                 (host.idleGpuMemory > 0) ? 1 : 0, host.idleGpuMemory,
-                host.getName(), 50));
+                host.getName(), 50);
 
         return result;
     }
@@ -378,19 +403,17 @@ public boolean higherPriorityJobExists(JobDetail baseJob, VirtualProc proc) {
     }
 
     @Override
-    public Set<String> findDispatchJobs(DispatchHost host,
+    public List<String> findDispatchJobs(DispatchHost host,
             ShowInterface show, int numJobs) {
-        LinkedHashSet<String> result = new LinkedHashSet<String>(numJobs);
-
-        result.addAll(getJdbcTemplate().query(
-                FIND_JOBS_BY_SHOW,
+        List<String> result = getJdbcTemplate().query(
+                fifoSchedulingEnabled ? FIND_JOBS_FIFO_BY_SHOW : FIND_JOBS_BY_SHOW,
                 PKJOB_MAPPER,
                 show.getShowId(), host.getFacilityId(), host.os,
                 host.idleCores, host.idleMemory,
                 threadMode(host.threadMode),
                 host.idleGpus,
                 (host.idleGpuMemory > 0) ? 1 : 0, host.idleGpuMemory,
-                host.getName(), numJobs * 10));
+                host.getName(), numJobs * 10);
 
         return result;
     }

cuebot/src/main/java/com/imageworks/spcue/dispatcher/CoreUnitDispatcher.java

@@ -21,7 +21,6 @@
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 import com.google.common.cache.Cache;
@@ -126,7 +125,7 @@ private Cache<String, String> getOrCreateJobLock() {
     }
 
 
-    private List<VirtualProc> dispatchJobs(DispatchHost host, Set<String> jobs) {
+    private List<VirtualProc> dispatchJobs(DispatchHost host, List<String> jobs) {
         List<VirtualProc> procs = new ArrayList<VirtualProc>();
 
         try {
@@ -170,8 +169,8 @@ private List<VirtualProc> dispatchJobs(DispatchHost host, Set<String> jobs) {
         return procs;
     }
 
-    private Set<String> getGpuJobs(DispatchHost host, ShowInterface show) {
-        Set<String> jobs = null;
+    private List<String> getGpuJobs(DispatchHost host, ShowInterface show) {
+        List<String> jobs = null;
 
         // TODO: GPU: make index with the 4 components instead of just 3, replace the just 3
 
@@ -200,7 +199,7 @@ private Set<String> getGpuJobs(DispatchHost host, ShowInterface show) {
 
     @Override
     public List<VirtualProc> dispatchHostToAllShows(DispatchHost host) {
-        Set<String> jobs = dispatchSupport.findDispatchJobsForAllShows(
+        List<String> jobs = dispatchSupport.findDispatchJobsForAllShows(
                 host,
                 getIntProperty("dispatcher.job_query_max"));
 
@@ -210,7 +209,7 @@ public List<VirtualProc> dispatchHostToAllShows(DispatchHost host) {
     @Override
     public List<VirtualProc> dispatchHost(DispatchHost host) {
 
-        Set<String> jobs = getGpuJobs(host, null);
+        List<String> jobs = getGpuJobs(host, null);
 
         if (jobs == null)
             jobs = dispatchSupport.findDispatchJobs(host, getIntProperty("dispatcher.job_query_max"));
@@ -221,7 +220,7 @@ public List<VirtualProc> dispatchHost(DispatchHost host) {
     @Override
     public List<VirtualProc> dispatchHost(DispatchHost host, ShowInterface show) {
 
-        Set<String> jobs = getGpuJobs(host, show);
+        List<String> jobs = getGpuJobs(host, show);
 
         if (jobs == null)
             jobs = dispatchSupport.findDispatchJobs(host, show,
@@ -233,7 +232,7 @@ public List<VirtualProc> dispatchHost(DispatchHost host, ShowInterface show) {
     @Override
     public List<VirtualProc> dispatchHost(DispatchHost host, GroupInterface group) {
 
-        Set<String> jobs = getGpuJobs(host, null);
+        List<String> jobs = getGpuJobs(host, null);
 
         if (jobs == null)
             jobs = dispatchSupport.findDispatchJobs(host, group);

cuebot/src/main/java/com/imageworks/spcue/dispatcher/DispatchSupport.java

@@ -306,7 +306,7 @@ List<DispatchFrame> findNextDispatchFrames(LayerInterface layer, VirtualProc pro
      * @param host
      * @return
      */
-    Set<String> findDispatchJobsForAllShows(DispatchHost host, int numJobs);
+    List<String> findDispatchJobsForAllShows(DispatchHost host, int numJobs);
 
     /**
      * Returns the highest priority job that can utilize
@@ -315,7 +315,7 @@ List<DispatchFrame> findNextDispatchFrames(LayerInterface layer, VirtualProc pro
      * @param host
      * @return
      */
-    Set<String> findDispatchJobs(DispatchHost host, int numJobs);
+    List<String> findDispatchJobs(DispatchHost host, int numJobs);
 
     /**
      * Returns the highest priority jobs that can utilize
@@ -324,7 +324,7 @@ List<DispatchFrame> findNextDispatchFrames(LayerInterface layer, VirtualProc pro
      * @param host
      * @return  A set of unique job ids.
      */
-    Set<String> findDispatchJobs(DispatchHost host, GroupInterface p);
+    List<String> findDispatchJobs(DispatchHost host, GroupInterface p);
 
     /**
      *
@@ -523,14 +523,14 @@ void updateProcMemoryUsage(FrameInterface frame, long rss, long maxRss, long vsi
     void determineIdleCores(DispatchHost host, int load);
 
     /**
-     * Return a set of job IDs that can take the given host.
+     * Return a list of job IDs that can take the given host.
      *
      * @param host
      * @param show
      * @param numJobs
      * @return
      */
-    Set<String> findDispatchJobs(DispatchHost host, ShowInterface show, int numJobs);
+    List<String> findDispatchJobs(DispatchHost host, ShowInterface show, int numJobs);
 
     /**
      * Return true of the job has pending frames.

cuebot/src/main/java/com/imageworks/spcue/dispatcher/DispatchSupportService.java

@@ -148,17 +148,17 @@ public boolean higherPriorityJobExists(JobDetail baseJob, VirtualProc proc) {
     }
 
     @Transactional(readOnly = true)
-    public Set<String> findDispatchJobsForAllShows(DispatchHost host, int numJobs) {
+    public List<String> findDispatchJobsForAllShows(DispatchHost host, int numJobs) {
         return dispatcherDao.findDispatchJobsForAllShows(host, numJobs);
     }
 
     @Transactional(readOnly = true)
-    public Set<String> findDispatchJobs(DispatchHost host, int numJobs) {
+    public List<String> findDispatchJobs(DispatchHost host, int numJobs) {
         return dispatcherDao.findDispatchJobs(host, numJobs);
     }
 
     @Transactional(readOnly = true)
-    public Set<String> findDispatchJobs(DispatchHost host, GroupInterface g) {
+    public List<String> findDispatchJobs(DispatchHost host, GroupInterface g) {
         return dispatcherDao.findDispatchJobs(host, g);
     }
 
@@ -170,7 +170,7 @@ public Set<String> findLocalDispatchJobs(DispatchHost host) {
 
     @Override
     @Transactional(readOnly = true)
-    public Set<String> findDispatchJobs(DispatchHost host, ShowInterface show,
+    public List<String> findDispatchJobs(DispatchHost host, ShowInterface show,
             int numJobs) {
         return dispatcherDao.findDispatchJobs(host, show, numJobs);
     }

cuebot/src/main/resources/opencue.properties

@@ -43,6 +43,8 @@ dispatcher.frame_query_max=20
 dispatcher.job_frame_dispatch_max=8
 # Maximum number of frames to dispatch from a host at one time.
 dispatcher.host_frame_dispatch_max=12
+# Whether or not to enable FIFO scheduling in the same priority.
+dispatcher.fifo_scheduling_enabled=false
 
 # Jobs will be archived to the history tables after being completed for this long.
 history.archive_jobs_cutoff_hours=72