[all] Add slot based booking

VERSION.in

@@ -1 +1 @@
-1.29
+1.30

cuebot/src/main/java/com/imageworks/spcue/DispatchFrame.java

@@ -44,6 +44,11 @@ public class DispatchFrame extends FrameEntity implements FrameInterface {
     public int maxGpus;
     public long minGpuMemory;
 
+    // Concurrency slots each frame of this layer requires (slot-based scheduling).
+    // 0 = not slot-based. Only populated by the slot dispatch queries; the generic
+    // dispatch queries never return slot-based frames.
+    public int slotsRequired;
+
     // A comma separated list of services
     public String services;
 

cuebot/src/main/java/com/imageworks/spcue/DispatchHost.java

@@ -43,6 +43,19 @@ public class DispatchHost extends Entity
     // Basically an 0 = auto, 1 = all.
     public int threadMode;
 
+    /**
+     * Max concurrent frames (slots) this host may run. -1 marks a regular (cores/memory) host; >= 0
+     * marks a slot-based host that only runs slot-based layers, capped at this many concurrent
+     * slots.
+     */
+    public int concurrentSlotsLimit = -1;
+
+    /**
+     * Number of slots not currently reserved on this host (concurrentSlotsLimit minus
+     * SUM(proc.int_slots_reserved)). Only meaningful for slot-based hosts; -1 otherwise.
+     */
+    public int idleSlots = -1;
+
     public long memory;
     public long idleMemory;
     public long gpuMemory;
@@ -69,6 +82,14 @@ public String getHostId() {
         return id;
     }
 
+    /**
+     * True when this host is slot-based: it only runs slot-based layers, capped by
+     * concurrentSlotsLimit, and ignores cores/memory for booking decisions.
+     */
+    public boolean isSlotHost() {
+        return concurrentSlotsLimit >= 0;
+    }
+
     public String getAllocationId() {
         return allocationId;
     }

cuebot/src/main/java/com/imageworks/spcue/LayerDetail.java

@@ -38,6 +38,8 @@ public class LayerDetail extends LayerEntity implements LayerInterface {
     public int timeout_llu;
     public int dispatchOrder;
     public int totalFrameCount;
+    // Concurrency slots each frame requires (slot-based scheduling). 0 = not slot-based.
+    public int slotsRequired;
 
     public Set<String> tags = new LinkedHashSet<String>();
     public Set<String> services = new LinkedHashSet<String>();

cuebot/src/main/java/com/imageworks/spcue/VirtualProc.java

@@ -45,6 +45,9 @@ public class VirtualProc extends FrameEntity implements ProcInterface {
     public int gpusReserved;
     public long gpuMemoryReserved;
     public long gpuMemoryUsed;
+
+    // Concurrency slots reserved by this proc (slot-based scheduling). 0 for regular procs.
+    public int slotsReserved;
     public long gpuMemoryMax;
 
     public boolean unbooked;
@@ -207,6 +210,45 @@ && containsSelfishService(frame.services.split(","), selfishServices)) {
         return proc;
     }
 
+    /**
+     * Build a proc for a slot-based booking. Slot procs reserve 0 cores, 0 memory and 0 gpus - the
+     * only resource they consume is the frame's slot requirement, which counts against the host's
+     * concurrent slots limit and the subscription/folder/job max_slots caps.
+     *
+     * @param host a slot-based host (isSlotHost() == true)
+     * @param frame a slot-based frame (slotsRequired > 0)
+     * @return
+     */
+    public static final VirtualProc buildSlotProc(DispatchHost host, DispatchFrame frame) {
+        if (frame.slotsRequired <= 0) {
+            throw new EntityException(
+                    "Cannot build a slot proc for a frame that requires no slots.");
+        }
+
+        VirtualProc proc = new VirtualProc();
+        proc.allocationId = host.getAllocationId();
+        proc.hostId = host.getHostId();
+        proc.frameId = null;
+        proc.layerId = frame.getLayerId();
+        proc.jobId = frame.getJobId();
+        proc.showId = frame.getShowId();
+        proc.facilityId = frame.getFacilityId();
+        proc.os = frame.os;
+
+        proc.hostName = host.getName();
+        proc.unbooked = false;
+        proc.isLocalDispatch = false;
+        proc.canHandleNegativeCoresRequest = false;
+
+        proc.coresReserved = 0;
+        proc.memoryReserved = 0;
+        proc.gpusReserved = 0;
+        proc.gpuMemoryReserved = 0;
+        proc.slotsReserved = frame.slotsRequired;
+
+        return proc;
+    }
+
     private static final boolean containsSelfishService(String[] frameServices,
             String[] selfishServices) {
         for (String frameService : frameServices) {

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

@@ -153,6 +153,40 @@ public interface DispatcherDao {
      */
     List<DispatchFrame> findNextDispatchFrames(LayerInterface layer, DispatchHost host, int limit);
 
+    /**
+     * Return a list of jobs with pending slot-based work that could book on the specified
+     * slot-based host. Jobs over their job/folder/subscription max_slots caps are excluded.
+     *
+     * @param host a slot-based host (concurrentSlotsLimit >= 0)
+     * @param numJobs
+     * @return a set of unique job ids ordered by priority
+     */
+    Set<String> findSlotDispatchJobs(DispatchHost host, int numJobs);
+
+    /**
+     * Return the next slot-based frames in the given job bookable on the specified slot-based host.
+     * Only frames of slot-based layers whose slot requirement fits the host's idle slots and whose
+     * job/folder/subscription max_slots caps allow another booking are returned.
+     *
+     * @param job
+     * @param host a slot-based host (concurrentSlotsLimit >= 0)
+     * @param limit
+     * @return
+     */
+    List<DispatchFrame> findNextSlotDispatchFrames(JobInterface job, DispatchHost host, int limit);
+
+    /**
+     * Return how many more slots the given job may book on the given host's allocation: the
+     * smallest remaining allowance across the job, folder and subscription max_slots caps.
+     * Integer.MAX_VALUE when all three caps are unlimited (-1); 0 when the job's show has no
+     * subscription to the host's allocation.
+     *
+     * @param job
+     * @param host
+     * @return
+     */
+    int getSlotCapacityRemaining(JobInterface job, DispatchHost host);
+
     /**
      * Return Scheduling Mode selected
      *

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

@@ -125,6 +125,15 @@ public interface GroupDao {
      */
     public void updateMaxCores(GroupInterface group, int value);
 
+    /**
+     * Sets the max concurrent slots for slot-based layers in the group's folder. -1 unlimited, 0
+     * reject-all, N caps at N.
+     *
+     * @param group
+     * @param value
+     */
+    public void updateMaxSlots(GroupInterface group, int value);
+
     /**
      * Set the minimum number of cores for this group
      *

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

@@ -245,6 +245,15 @@ public interface HostDao {
      */
     void updateThreadMode(HostInterface host, ThreadMode mode);
 
+    /**
+     * Sets the host's concurrent slots limit. -1 disables slot mode (regular host); >= 0 makes the
+     * host slot-based, running only slot layers up to this many concurrent slots.
+     *
+     * @param host HostInterface
+     * @param limit int
+     */
+    void updateConcurrentSlotsLimit(HostInterface host, int limit);
+
     /**
      * Update the specified host's hardware information.
      *

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

@@ -45,6 +45,14 @@ public interface JobDao {
      */
     public void updateMaxCores(GroupInterface g, int cores);
 
+    /**
+     * Updates all jobs in the specified group to the max slots value.
+     *
+     * @param g
+     * @param slots
+     */
+    public void updateMaxSlots(GroupInterface g, int slots);
+
     /**
      * Updates all jobs in the specifid group to the min cores value.
      *
@@ -306,6 +314,14 @@ public interface JobDao {
      */
     void updateMaxCores(JobInterface j, int v);
 
+    /**
+     * Sets the job's max concurrent slots for slot-based layers. -1 unlimited, 0 reject-all.
+     *
+     * @param j
+     * @param v
+     */
+    void updateMaxSlots(JobInterface j, int v);
+
     /**
      * sets the jobs new min gpu value
      *

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

@@ -114,4 +114,13 @@ public interface SubscriptionDao {
      * @param size int
      */
     void updateSubscriptionBurst(SubscriptionInterface sub, int size);
+
+    /**
+     * update the subscription max concurrent slots for slot-based layers (-1 unlimited, 0
+     * reject-all, N cap)
+     *
+     * @param sub SubscriptionInterface
+     * @param maxSlots int
+     */
+    void updateSubscriptionMaxSlots(SubscriptionInterface sub, int maxSlots);
 }

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

@@ -363,11 +363,13 @@ public List<DispatchFrame> findNextDispatchFrames(JobInterface job, VirtualProc
         long lastTime = System.currentTimeMillis();
         List<DispatchFrame> frames;
         if (proc.isLocalDispatch) {
-            frames = getJdbcTemplate().query(FIND_LOCAL_DISPATCH_FRAME_BY_JOB_AND_PROC,
+            frames = getJdbcTemplate().query(
+                    SlotDispatchQuery.FIND_LOCAL_DISPATCH_FRAME_BY_JOB_AND_PROC_EXCLUDE_SLOT,
                     FrameDaoJdbc.DISPATCH_FRAME_MAPPER, proc.memoryReserved, proc.gpuMemoryReserved,
                     job.getJobId(), limit);
         } else {
-            frames = getJdbcTemplate().query(FIND_DISPATCH_FRAME_BY_JOB_AND_PROC,
+            frames = getJdbcTemplate().query(
+                    SlotDispatchQuery.FIND_DISPATCH_FRAME_BY_JOB_AND_PROC_EXCLUDE_SLOT,
                     FrameDaoJdbc.DISPATCH_FRAME_MAPPER, proc.coresReserved, proc.memoryReserved,
                     proc.gpusReserved, (proc.gpuMemoryReserved > 0) ? 1 : 0, proc.gpuMemoryReserved,
                     job.getJobId(), proc.hostName, job.getJobId(), limit);
@@ -386,12 +388,14 @@ public List<DispatchFrame> findNextDispatchFrames(JobInterface job, DispatchHost
         List<DispatchFrame> frames;
 
         if (host.isLocalDispatch) {
-            frames = getJdbcTemplate().query(FIND_LOCAL_DISPATCH_FRAME_BY_JOB_AND_HOST,
+            frames = getJdbcTemplate().query(
+                    SlotDispatchQuery.FIND_LOCAL_DISPATCH_FRAME_BY_JOB_AND_HOST_EXCLUDE_SLOT,
                     FrameDaoJdbc.DISPATCH_FRAME_MAPPER, host.idleMemory, host.idleGpuMemory,
                     job.getJobId(), limit);
 
         } else {
-            frames = getJdbcTemplate().query(FIND_DISPATCH_FRAME_BY_JOB_AND_HOST,
+            frames = getJdbcTemplate().query(
+                    SlotDispatchQuery.FIND_DISPATCH_FRAME_BY_JOB_AND_HOST_EXCLUDE_SLOT,
                     FrameDaoJdbc.DISPATCH_FRAME_MAPPER, host.idleCores, host.idleMemory,
                     threadMode(host.threadMode), host.idleGpus, (host.idleGpuMemory > 0) ? 1 : 0,
                     host.idleGpuMemory, job.getJobId(), host.getName(), job.getJobId(), limit);
@@ -409,11 +413,13 @@ public List<DispatchFrame> findNextDispatchFrames(LayerInterface layer, VirtualP
         List<DispatchFrame> frames;
 
         if (proc.isLocalDispatch) {
-            frames = getJdbcTemplate().query(FIND_LOCAL_DISPATCH_FRAME_BY_LAYER_AND_PROC,
+            frames = getJdbcTemplate().query(
+                    SlotDispatchQuery.FIND_LOCAL_DISPATCH_FRAME_BY_LAYER_AND_PROC_EXCLUDE_SLOT,
                     FrameDaoJdbc.DISPATCH_FRAME_MAPPER, proc.memoryReserved, proc.gpuMemoryReserved,
                     layer.getLayerId(), limit);
         } else {
-            frames = getJdbcTemplate().query(FIND_DISPATCH_FRAME_BY_LAYER_AND_PROC,
+            frames = getJdbcTemplate().query(
+                    SlotDispatchQuery.FIND_DISPATCH_FRAME_BY_LAYER_AND_PROC_EXCLUDE_SLOT,
                     FrameDaoJdbc.DISPATCH_FRAME_MAPPER, proc.coresReserved, proc.memoryReserved,
                     proc.gpusReserved, proc.gpuMemoryReserved, layer.getLayerId(),
                     layer.getLayerId(), proc.hostName, limit);
@@ -432,12 +438,14 @@ public List<DispatchFrame> findNextDispatchFrames(LayerInterface layer, Dispatch
         List<DispatchFrame> frames;
 
         if (host.isLocalDispatch) {
-            frames = getJdbcTemplate().query(FIND_LOCAL_DISPATCH_FRAME_BY_LAYER_AND_HOST,
+            frames = getJdbcTemplate().query(
+                    SlotDispatchQuery.FIND_LOCAL_DISPATCH_FRAME_BY_LAYER_AND_HOST_EXCLUDE_SLOT,
                     FrameDaoJdbc.DISPATCH_FRAME_MAPPER, host.idleMemory, host.idleGpuMemory,
                     layer.getLayerId(), limit);
 
         } else {
-            frames = getJdbcTemplate().query(FIND_DISPATCH_FRAME_BY_LAYER_AND_HOST,
+            frames = getJdbcTemplate().query(
+                    SlotDispatchQuery.FIND_DISPATCH_FRAME_BY_LAYER_AND_HOST_EXCLUDE_SLOT,
                     FrameDaoJdbc.DISPATCH_FRAME_MAPPER, host.idleCores, host.idleMemory,
                     threadMode(host.threadMode), host.idleGpus, host.idleGpuMemory,
                     layer.getLayerId(), layer.getLayerId(), host.getName(), limit);
@@ -449,6 +457,77 @@ public List<DispatchFrame> findNextDispatchFrames(LayerInterface layer, Dispatch
         return frames;
     }
 
+    /**
+     * Maps rows of the slot dispatch frame queries; delegates to the generic mapper and adds the
+     * slot requirement column.
+     */
+    public static final RowMapper<DispatchFrame> SLOT_DISPATCH_FRAME_MAPPER =
+            new RowMapper<DispatchFrame>() {
+                public DispatchFrame mapRow(ResultSet rs, int rowNum) throws SQLException {
+                    DispatchFrame frame = FrameDaoJdbc.DISPATCH_FRAME_MAPPER.mapRow(rs, rowNum);
+                    frame.slotsRequired = rs.getInt("int_slots_required");
+                    return frame;
+                }
+            };
+
+    @Override
+    public Set<String> findSlotDispatchJobs(DispatchHost host, int numJobs) {
+        long lastTime = System.currentTimeMillis();
+        LinkedHashSet<String> result = new LinkedHashSet<String>();
+
+        result.addAll(getJdbcTemplate().query(new PreparedStatementCreator() {
+            @Override
+            public PreparedStatement createPreparedStatement(Connection conn) throws SQLException {
+                String query = handleInClause("str_os", SlotDispatchQuery.FIND_SLOT_DISPATCH_JOBS,
+                        host.getOs().length);
+                PreparedStatement findJobsStmt = conn.prepareStatement(query);
+                int index = 1;
+                findJobsStmt.setString(index++, host.getFacilityId());
+                for (String item : host.getOs()) {
+                    findJobsStmt.setString(index++, item);
+                }
+                findJobsStmt.setInt(index++, host.idleSlots);
+                findJobsStmt.setString(index++, host.getName());
+                findJobsStmt.setString(index++, host.getAllocationId());
+                findJobsStmt.setInt(index++, numJobs);
+                return findJobsStmt;
+            }
+        }, PKJOB_MAPPER));
+
+        prometheusMetrics.setBookingDurationMetric("findSlotDispatchJobs query",
+                System.currentTimeMillis() - lastTime);
+        return result;
+    }
+
+    @Override
+    public List<DispatchFrame> findNextSlotDispatchFrames(JobInterface job, DispatchHost host,
+            int limit) {
+        long lastTime = System.currentTimeMillis();
+        List<DispatchFrame> frames =
+                getJdbcTemplate().query(SlotDispatchQuery.FIND_SLOT_DISPATCH_FRAMES_BY_JOB_AND_HOST,
+                        SLOT_DISPATCH_FRAME_MAPPER, host.idleSlots, job.getJobId(), host.getName(),
+                        job.getJobId(), host.getAllocationId(), limit);
+        prometheusMetrics.setBookingDurationMetric("findNextSlotDispatchFrames query",
+                System.currentTimeMillis() - lastTime);
+        return frames;
+    }
+
+    @Override
+    public int getSlotCapacityRemaining(JobInterface job, DispatchHost host) {
+        try {
+            Integer capacity =
+                    getJdbcTemplate().queryForObject(SlotDispatchQuery.GET_SLOT_CAPACITY_REMAINING,
+                            Integer.class, host.getAllocationId(), job.getJobId());
+            if (capacity == null) {
+                return Integer.MAX_VALUE;
+            }
+            return Math.max(capacity, 0);
+        } catch (org.springframework.dao.EmptyResultDataAccessException e) {
+            // No subscription between the job's show and the host's allocation.
+            return 0;
+        }
+    }
+
     @Override
     public DispatchFrame findNextDispatchFrame(JobInterface job, VirtualProc proc) {
         return findNextDispatchFrames(job, proc, 1).get(0);

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

@@ -215,6 +215,22 @@ public void updateMaxCores(GroupInterface group, int value) {
         }
     }
 
+    @Override
+    public void updateMaxSlots(GroupInterface group, int value) {
+        // Slots are whole counts: -1 = unlimited, 0 = reject-all, N = cap. Normalize any
+        // negative to the -1 sentinel.
+        if (value < 0) {
+            value = CueUtil.FEATURE_DISABLED;
+        }
+
+        getJdbcTemplate().update("UPDATE folder_resource SET int_max_slots=? WHERE pk_folder=?",
+                value, group.getId());
+
+        if (accountingNotifier.isEnabled() && showDao.isSchedulerManaged(group.getShowId())) {
+            accountingNotifier.notifyFolderMaxSlots(group.getId(), value);
+        }
+    }
+
     @Override
     public void updateMinCores(GroupInterface group, int value) {
         if (value < 0) {