Cuebot reserve all cores

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

@@ -24,9 +24,14 @@
 import com.imageworks.spcue.grpc.host.LockState;
 import com.imageworks.spcue.util.CueUtil;
 
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.LogManager;
+
 public class DispatchHost extends Entity
     implements HostInterface, FacilityInterface, ResourceContainer {
 
+    private static final Logger logger = LogManager.getLogger(DispatchHost.class);
+
     public String facilityId;
     public String allocationId;
     public LockState lockState;
@@ -76,12 +81,52 @@ public String getFacilityId() {
         return facilityId;
     }
 
+    public boolean canHandleNegativeCoresRequest(int requestedCores) {
+        // Request is positive, no need to test further.
+        if (requestedCores > 0) {
+            logger.debug(getName() + " can handle the job with " + requestedCores + " cores.");
+            return true;
+        }
+        // All cores are available, validate the request.
+        if (cores == idleCores) {
+            logger.debug(getName() + " can handle the job with " + requestedCores + " cores.");
+            return true;
+        }
+        // Some or all cores are busy, avoid booking again.
+        logger.debug(getName() + " cannot handle the job with " + requestedCores + " cores.");
+        return false;
+    }
+
+    public int handleNegativeCoresRequirement(int requestedCores) {
+        // If we request a <=0 amount of cores, return positive core count.
+
+        if (requestedCores > 0) {
+            // Do not process positive core requests.
+            logger.debug("Requested " + requestedCores + " cores.");
+            return requestedCores;
+        }
+        if (requestedCores <=0 && idleCores < cores) {
+            // If request is negative but cores are already used, return 0.
+            // We don't want to overbook the host.
+            logger.debug("Requested " + requestedCores + " cores, but the host is busy and cannot book more jobs.");
+            return 0;
+        }
+        // Book all cores minus the request
+        int totalCores = idleCores + requestedCores;
+        logger.debug("Requested " + requestedCores + " cores  <= 0, " +
+                     idleCores + " cores are free, booking " + totalCores + " cores");
+        return totalCores;
+    }
+
     @Override
     public boolean hasAdditionalResources(int minCores, long minMemory, int minGpus, long minGpuMemory) {
-
+        minCores = handleNegativeCoresRequirement(minCores);
         if (idleCores < minCores) {
             return false;
         }
+        if (minCores <= 0) {
+            return false;
+        }
         else if (idleMemory <  minMemory) {
             return false;
         }

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

@@ -22,6 +22,9 @@
 import com.imageworks.spcue.dispatcher.ResourceContainer;
 import com.imageworks.spcue.grpc.renderpartition.RenderPartitionType;
 
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.LogManager;
+
 /**
  * Contains information about local desktop cores a user has
  * assigned to the given job.
@@ -33,6 +36,8 @@
 public class LocalHostAssignment extends Entity
     implements ResourceContainer {
 
+    private static final Logger logger = LogManager.getLogger(LocalHostAssignment.class);
+
     private int idleCoreUnits;
     private long idleMemory;
     private int idleGpuUnits;
@@ -62,12 +67,36 @@ public LocalHostAssignment(int maxCores, int threads, long maxMemory, int maxGpu
         this.maxGpuMemory = maxGpuMemory;
     }
 
+    public int handleNegativeCoresRequirement(int requestedCores) {
+        // If we request a <=0 amount of cores, return positive core count.
+
+        if (requestedCores > 0) {
+            // Do not process positive core requests.
+            logger.debug("Requested " + requestedCores + " cores.");
+            return requestedCores;
+        }
+        if (requestedCores <=0 && idleCoreUnits < threads) {
+            // If request is negative but cores are already used, return 0.
+            // We don't want to overbook the host.
+            logger.debug("Requested " + requestedCores + " cores, but the host is busy and cannot book more jobs.");
+            return 0;
+        }
+        // Book all cores minus the request
+        int totalCores = idleCoreUnits + requestedCores;
+        logger.debug("Requested " + requestedCores + " cores  <= 0, " +
+                     idleCoreUnits + " cores are free, booking " + totalCores + " cores");
+        return totalCores;
+    }
+
     @Override
     public boolean hasAdditionalResources(int minCores, long minMemory, int minGpus, long minGpuMemory) {
-
+        minCores = handleNegativeCoresRequirement(minCores);
         if (idleCoreUnits < minCores) {
             return false;
         }
+        if (minCores <= 0) {
+            return false;
+        }
         else if (idleMemory <  minMemory) {
             return false;
         }

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

@@ -54,12 +54,12 @@ public boolean isSkipped(String tags, long cores, long memory) {
         try {
             if (failed.containsKey(tags)) {
                 long [] mark = failed.get(tags);
-                if (cores <= mark[0]) {
-                    logger.info("skipped due to not enough cores " + cores + " <= " + mark[0]);
+                if (cores < mark[0]) {
+                    logger.info("skipped due to not enough cores " + cores + " < " + mark[0]);
                     return true;
                 }
-                else if (memory <= mark[1]) {
-                    logger.info("skipped due to not enough memory " + memory + " <= " + mark[1]);
+                else if (memory < mark[1]) {
+                    logger.info("skipped due to not enough memory " + memory + " < " + mark[1]);
                     return true;
                 }
             }

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

@@ -22,15 +22,21 @@
 import com.imageworks.spcue.dispatcher.Dispatcher;
 import com.imageworks.spcue.grpc.host.ThreadMode;
 
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.LogManager;
+
 public class VirtualProc extends FrameEntity implements ProcInterface {
 
+    private static final Logger logger = LogManager.getLogger(VirtualProc.class);
+
     public String hostId;
     public String allocationId;
     public String frameId;
     public String hostName;
     public String os;
     public byte[] childProcesses;
 
+    public boolean canHandleNegativeCoresRequest;
     public int coresReserved;
     public long memoryReserved;
     public long memoryUsed;
@@ -111,7 +117,17 @@ public static final VirtualProc build(DispatchHost host, DispatchFrame frame) {
             proc.coresReserved = proc.coresReserved + host.strandedCores;
         }
 
-        if (proc.coresReserved >= 100) {
+        proc.canHandleNegativeCoresRequest = host.canHandleNegativeCoresRequest(proc.coresReserved);
+
+        if (proc.coresReserved == 0) {
+            logger.debug("Reserving all cores");
+            proc.coresReserved = host.cores;
+        }
+        else if (proc.coresReserved < 0) {
+            logger.debug("Reserving all cores minus " + proc.coresReserved);
+            proc.coresReserved = host.cores + proc.coresReserved;
+        }
+        else if (proc.coresReserved >= 100) {
 
             int originalCores = proc.coresReserved;
 

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

@@ -59,7 +59,7 @@ public interface LayerDao {
     public List<LayerDetail> getLayerDetails(JobInterface job);
 
     /**
-     * Returns true if supplied layer is compelte.
+     * Returns true if supplied layer is complete.
      *
      * @param layer
      * @return boolean
@@ -82,7 +82,7 @@ public interface LayerDao {
     void insertLayerDetail(LayerDetail l);
 
     /**
-     * gets a layer detail from an object that implments layer
+     * gets a layer detail from an object that implements layer
      *
      * @param layer
      * @return LayerDetail
@@ -167,7 +167,7 @@ public interface LayerDao {
     void updateLayerTags(LayerInterface layer, Set<String> tags);
 
     /**
-     * Insert a key/valye pair into the layer environment
+     * Insert a key/value pair into the layer environment
      *
      * @param layer
      * @param key
@@ -292,7 +292,7 @@ public interface LayerDao {
 
     /**
      * Update all layers of the set type in the specified job
-     * with the new min cores requirement.
+     * with the new min gpu requirement.
      *
      * @param job
      * @param gpus
@@ -304,17 +304,16 @@ public interface LayerDao {
      * Update a layer's max cores value, which limits how
      * much threading can go on.
      *
-     * @param job
-     * @param cores
-     * @param type
+     * @param layer
+     * @param threadable
      */
     void updateThreadable(LayerInterface layer, boolean threadable);
 
     /**
      * Update a layer's timeout value, which limits how
      * much the frame can run on a host.
      *
-     * @param job
+     * @param layer
      * @param timeout
      */
     void updateTimeout(LayerInterface layer, int timeout);
@@ -323,8 +322,8 @@ public interface LayerDao {
      * Update a layer's LLU timeout value, which limits how
      * much the frame can run on a host without updates in the log file.
      *
-     * @param job
-     * @param timeout
+     * @param layer
+     * @param timeout_llu
      */
     void updateTimeoutLLU(LayerInterface layer, int timeout_llu);
 
@@ -341,7 +340,7 @@ public interface LayerDao {
 
     /**
      * Appends a tag to the current set of tags.  If the tag
-     * already exists than nothing happens.
+     * already exists then nothing happens.
      *
      * @param layer
      * @param val
@@ -363,8 +362,9 @@ public interface LayerDao {
      * Update layer usage with processor time usage.
      * This happens when the proc has completed or failed some work.
      *
-     * @param proc
+     * @param layer
      * @param newState
+     * @param exitStatus
      */
     void updateUsage(LayerInterface layer, ResourceUsage usage, int exitStatus);
 
@@ -387,6 +387,9 @@ public interface LayerDao {
 
     /**
      * Enable/disable memory optimizer.
+     *
+     * @param layer
+     * @param state
      */
     void enableMemoryOptimizer(LayerInterface layer, boolean state);
 

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

@@ -51,8 +51,12 @@
 import com.imageworks.spcue.util.CueUtil;
 import com.imageworks.spcue.util.SqlUtil;
 
-public class LayerDaoJdbc extends JdbcDaoSupport implements LayerDao {
 
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.LogManager;
+
+public class LayerDaoJdbc extends JdbcDaoSupport implements LayerDao {
+    private static final Logger logger = LogManager.getLogger(LayerDaoJdbc.class);
     private static final String INSERT_OUTPUT_PATH =
         "INSERT INTO " +
             "layer_output " +

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

@@ -259,13 +259,16 @@ public List<VirtualProc> dispatchHost(DispatchHost host, JobInterface job) {
                 " on job " + job.getName());
 
         for (DispatchFrame frame: frames) {
-
             VirtualProc proc =  VirtualProc.build(host, frame);
-
-            if (host.idleCores < frame.minCores ||
+            if (frame.minCores <= 0 && !proc.canHandleNegativeCoresRequest) {
+                logger.debug("Cannot dispatch job, host is busy.");
+                break;
+            }
+            if (host.idleCores < host.handleNegativeCoresRequirement(frame.minCores) ||
                     host.idleMemory < frame.minMemory ||
                     host.idleGpus < frame.minGpus ||
                     host.idleGpuMemory < frame.minGpuMemory) {
+                    logger.debug("Cannot dispatch, insufficient resources.");
                 break;
             }
 
@@ -281,6 +284,8 @@ public List<VirtualProc> dispatchHost(DispatchHost host, JobInterface job) {
 
             boolean success = new DispatchFrameTemplate(proc, job, frame, false) {
                 public void wrapDispatchFrame() {
+                    logger.debug("Dispatching frame with " + frame.minCores + " minCores on proc with " +
+                                 proc.coresReserved + " coresReserved");
                     dispatch(frame, proc);
                     dispatchSummary(proc, frame, "Booking");
                     return;

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

@@ -220,8 +220,8 @@ public void handleHostReport(HostReport report, boolean isBoot) {
                     bookingManager.removeInactiveLocalHostAssignment(lca);
                 }
             }
-
-            if (host.idleCores < Dispatcher.CORE_POINTS_RESERVED_MIN) {
+            int cores_to_reserve = host.handleNegativeCoresRequirement(Dispatcher.CORE_POINTS_RESERVED_MIN);
+            if (cores_to_reserve <= 0 || host.idleCores < Dispatcher.CORE_POINTS_RESERVED_MIN) {
                 msg = String.format("%s doesn't have enough idle cores, %d needs %d",
                     host.name,  host.idleCores, Dispatcher.CORE_POINTS_RESERVED_MIN);
             }

cuebot/src/main/java/com/imageworks/spcue/service/JobManagerService.java

@@ -274,7 +274,7 @@ public JobDetail createJob(BuildableJob buildableJob) {
                     }
                 }
 
-                if (layer.minimumCores < Dispatcher.CORE_POINTS_RESERVED_MIN) {
+                if (layer.minimumCores > 0 && layer.minimumCores < Dispatcher.CORE_POINTS_RESERVED_MIN) {
                     layer.minimumCores =  Dispatcher.CORE_POINTS_RESERVED_MIN;
                 }
 

cuebot/src/main/java/com/imageworks/spcue/service/JobSpec.java

@@ -117,7 +117,7 @@ public class JobSpec {
     public JobSpec() {
     }
 
-    public static final String NAME_REGEX = "^([\\w\\.]{3,})$";
+    public static final String NAME_REGEX = "^([\\w\\.-]{3,})$";
 
     public static final Pattern NAME_PATTERN = Pattern.compile(NAME_REGEX);
 
@@ -612,12 +612,16 @@ private void determineMinimumCores(Element layerTag, LayerDetail layer) {
         int corePoints = layer.minimumCores;
 
         if (cores.contains(".")) {
-            corePoints = (int) (Double.valueOf(cores) * 100 + .5);
+            if (cores.contains("-")) {
+                corePoints = (int) (Double.valueOf(cores) * 100 - .5);
+            } else {
+                corePoints = (int) (Double.valueOf(cores) * 100 + .5);
+            }
         } else {
             corePoints = Integer.valueOf(cores);
         }
 
-        if (corePoints < Dispatcher.CORE_POINTS_RESERVED_MIN) {
+        if (corePoints > 0 && corePoints < Dispatcher.CORE_POINTS_RESERVED_MIN) {
             corePoints = Dispatcher.CORE_POINTS_RESERVED_DEFAULT;
         }
 
@@ -651,7 +655,7 @@ private void determineChunkSize(Element layerTag, LayerDetail layer) {
      */
     private void determineThreadable(Element layerTag, LayerDetail layer) {
         // Must have at least 1 core to thread.
-        if (layer.minimumCores < 100) {
+        if (layer.minimumCores > 0 && layer.minimumCores < 100) {
             layer.isThreadable = false;
         }
         else if (layerTag.getChildTextTrim("threadable") != null) {

cuegui/cuegui/FilterDialog.py

@@ -454,7 +454,7 @@ def createAction(self):
                         "Create Action",
                         "What value should this property be set to?",
                         0,
-                        0,
+                        -8,  # Minimum core value can be <=0, booking all cores minus this value.
                         50000,
                         2)
                     value = float(value)

cuesubmit/cuesubmit/Submission.py

@@ -87,7 +87,7 @@ def buildLayer(layerData, command, lastLayer=None):
     @type lastLayer: outline.layer.Layer
     @param lastLayer: layer that this new layer should be dependent on if dependType is set.
     """
-    threadable = float(layerData.cores) >= 2
+    threadable = float(layerData.cores) >= 2 or float(layerData.cores) <= 0
     layer = outline.modules.shell.Shell(
         layerData.name, command=command.split(), chunk=layerData.chunk,
         threads=float(layerData.cores), range=str(layerData.layerRange), threadable=threadable)