From 8642cab82ba605bff6a23f0b04f7e9dd6c822b88 Mon Sep 17 00:00:00 2001 From: mpoemsl <37959974+mpoemsl@users.noreply.github.com> Date: Fri, 17 Mar 2023 14:38:32 +0100 Subject: [PATCH 1/9] Make priorities into categorical values 'low', 'high', 'max' --- .../pytorch/serve/grpcimpl/InferenceImpl.java | 2 +- .../main/java/org/pytorch/serve/job/Job.java | 13 +-- .../java/org/pytorch/serve/util/ApiUtils.java | 6 +- .../org/pytorch/serve/util/ConfigManager.java | 14 ++- .../org/pytorch/serve/util/Prioritisable.java | 4 +- .../util/PriorityLinkedBlockingDeque.java | 85 ++++++++----------- .../java/org/pytorch/serve/wlm/Model.java | 10 +-- .../org/pytorch/serve/wlm/ModelManager.java | 4 +- 8 files changed, 62 insertions(+), 76 deletions(-) diff --git a/frontend/server/src/main/java/org/pytorch/serve/grpcimpl/InferenceImpl.java b/frontend/server/src/main/java/org/pytorch/serve/grpcimpl/InferenceImpl.java index 7b613fb121..c85422d76b 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/grpcimpl/InferenceImpl.java +++ b/frontend/server/src/main/java/org/pytorch/serve/grpcimpl/InferenceImpl.java @@ -103,7 +103,7 @@ public void predictions( try { if (!ModelManager.getInstance().addJob(job)) { - int priority = job.getPriority(); + String priority = job.getPriority(); String responseMessage = ApiUtils.getInferenceErrorResponseMessage(modelName, modelVersion, priority); InternalServerException e = new InternalServerException(responseMessage); diff --git a/frontend/server/src/main/java/org/pytorch/serve/job/Job.java b/frontend/server/src/main/java/org/pytorch/serve/job/Job.java index 5dc042f526..056a868536 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/job/Job.java +++ b/frontend/server/src/main/java/org/pytorch/serve/job/Job.java @@ -11,7 +11,7 @@ public abstract class Job implements Prioritisable { private String modelVersion; private WorkerCommands cmd; // Else its data msg or inf requests private RequestInput input; - private int priority; + private String priority; private long begin; private long scheduled; @@ -23,19 +23,14 @@ public Job(String modelName, String version, WorkerCommands cmd, RequestInput in begin = System.nanoTime(); scheduled = begin; - Map headers = input.getHeaders(); - if (headers.containsKey("X-TS-Priority")) { - this.priority = Integer.parseInt(headers.get("X-TS-Priority")); - } else { - this.priority = 0; - } + this.priority = input.getHeaders().getOrDefault("x-ts-priority", "max"); } - public int getPriority() { + public String getPriority() { return this.priority; } - public void setPriority(int priority) { + public void setPriority(String priority) { this.priority = priority; } diff --git a/frontend/server/src/main/java/org/pytorch/serve/util/ApiUtils.java b/frontend/server/src/main/java/org/pytorch/serve/util/ApiUtils.java index c485b67eb7..0dcdc3a0b8 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/util/ApiUtils.java +++ b/frontend/server/src/main/java/org/pytorch/serve/util/ApiUtils.java @@ -394,7 +394,7 @@ public static RestJob addRESTInferenceJob( throws ModelNotFoundException, ModelVersionNotFoundException { RestJob job = new RestJob(ctx, modelName, version, WorkerCommands.PREDICT, input); if (!ModelManager.getInstance().addJob(job)) { - int priority = job.getPriority(); + String priority = job.getPriority(); String responseMessage = getInferenceErrorResponseMessage(modelName, version, priority); throw new ServiceUnavailableException(responseMessage); } @@ -402,14 +402,14 @@ public static RestJob addRESTInferenceJob( } @SuppressWarnings("PMD") - public static String getInferenceErrorResponseMessage(String modelName, String modelVersion, int jobPriority) { + public static String getInferenceErrorResponseMessage(String modelName, String modelVersion, String jobPriority) { String responseMessage = "Model: " + modelName + "\n"; if (modelVersion != null) { responseMessage += "Version: " + modelVersion + "\n"; } - responseMessage += "Priority: " + String.valueOf(jobPriority) + "\n"; + responseMessage += "Priority: " + jobPriority + "\n"; responseMessage += "Reason: queue full"; diff --git a/frontend/server/src/main/java/org/pytorch/serve/util/ConfigManager.java b/frontend/server/src/main/java/org/pytorch/serve/util/ConfigManager.java index 7d0422853e..9d0a1179ee 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/util/ConfigManager.java +++ b/frontend/server/src/main/java/org/pytorch/serve/util/ConfigManager.java @@ -64,7 +64,7 @@ public final class ConfigManager { private static final String TS_NUMBER_OF_NETTY_THREADS = "number_of_netty_threads"; private static final String TS_NETTY_CLIENT_THREADS = "netty_client_threads"; private static final String TS_JOB_QUEUE_SIZE = "job_queue_size"; - private static final String TS_NUMBER_OF_PRIORITIES = "n_priorities"; + private static final String TS_HIGH_PRIORITY_PROBABILITY = "high_prio_prob"; private static final String TS_NUMBER_OF_GPU = "number_of_gpu"; private static final String TS_METRICS_CONFIG = "metrics_config"; @@ -362,8 +362,8 @@ public int getJobQueueSize() { return getIntProperty(TS_JOB_QUEUE_SIZE, 100); } - public int getNumberOfPriorities() { - return getIntProperty(TS_NUMBER_OF_PRIORITIES, 1); + public float getHighPrioProb() { + return getFloatProperty(TS_HIGH_PRIORITY_PROBABILITY, 0.67f); } public int getNumberOfGpu() { @@ -685,6 +685,14 @@ private int getIntProperty(String key, int def) { return Integer.parseInt(value); } + private float getFloatProperty(String key, float def) { + String value = prop.getProperty(key); + if (value == null) { + return def; + } + return Float.parseFloat(value); + } + public int getDefaultResponseTimeout() { return Integer.parseInt(prop.getProperty(TS_DEFAULT_RESPONSE_TIMEOUT, "120")); } diff --git a/frontend/server/src/main/java/org/pytorch/serve/util/Prioritisable.java b/frontend/server/src/main/java/org/pytorch/serve/util/Prioritisable.java index 4b4c76b264..88b109f213 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/util/Prioritisable.java +++ b/frontend/server/src/main/java/org/pytorch/serve/util/Prioritisable.java @@ -2,7 +2,7 @@ public interface Prioritisable { - public int getPriority(); - public void setPriority(int priority); + public String getPriority(); + public void setPriority(String priority); } diff --git a/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java b/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java index a76da66c43..00c38658a8 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java +++ b/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java @@ -8,6 +8,7 @@ import java.util.concurrent.TimeUnit; import java.util.function.BiFunction; import java.util.function.Function; +import java.util.Enumeration; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -19,70 +20,52 @@ public class PriorityLinkedBlockingDeque { final ReentrantLock lock = new ReentrantLock(); private final Condition notEmpty = lock.newCondition(); - private int nPriorities; - private int sumPriorityWeights; - private int[] weightedPriorityMap; - private ConcurrentHashMap> priorityDeques; - - public PriorityLinkedBlockingDeque(int nPriorities, int queueSize) { - - this.nPriorities = nPriorities; - this.priorityDeques = new ConcurrentHashMap>(); - - // note: the weight calculation logic may be adapted to the user's needs - // sum of 0 + 1 + 2 + ... + nPriorities - 1 via triangular sum - this.sumPriorityWeights = ((this.nPriorities - 1) * this.nPriorities) / 2; - this.weightedPriorityMap = new int[sumPriorityWeights]; - - // initialize priority deques and weight map - int keyStart = 0; - for (int priority = 0; priority < this.nPriorities; priority++) { - this.priorityDeques.put(priority, new LinkedBlockingDeque(queueSize)); - if (priority > 0) { - // priority weights are inverse priority values for now - int priorityWeight = this.nPriorities - priority; - for (int key = keyStart; key < keyStart + priorityWeight; key++) { - this.weightedPriorityMap[key] = priority; - } - keyStart += priorityWeight; - } - } + private int queueSize; + private float highPrioProb; + private ConcurrentHashMap> priorityDeques; + + public PriorityLinkedBlockingDeque(int queueSize, float highPrioProb) { + + this.queueSize = queueSize; + this.highPrioProb = highPrioProb; + this.priorityDeques = new ConcurrentHashMap>(); + + // initialize priority deques + this.priorityDeques.put("max", new LinkedBlockingDeque(queueSize)); + this.priorityDeques.put("high", new LinkedBlockingDeque(queueSize)); + this.priorityDeques.put("low", new LinkedBlockingDeque(queueSize)); + } private LinkedBlockingDeque getDequeForExtraction() { - // always select deque 0 first if non-empty - if (this.nPriorities == 1 || !this.priorityDeques.get(0).isEmpty()) { - return this.priorityDeques.get(0); + // always select deque max first if non-empty + if (!this.priorityDeques.get("max").isEmpty()) { + return this.priorityDeques.get("max"); } - // sample according to weight map - int randInt = ThreadLocalRandom.current().nextInt(sumPriorityWeights); - int randPriority = this.weightedPriorityMap[randInt]; - LinkedBlockingDeque dequeForExtraction = this.priorityDeques.get(randPriority); - - // if sampled deque is empty, scan deques according to priority - if (dequeForExtraction.isEmpty()) { - for (int priority = 1; priority < this.nPriorities; priority++) { - LinkedBlockingDeque priorityDeque = this.priorityDeques.get(priority); - if (!priorityDeque.isEmpty()) { - return priorityDeque; - } + // return random deque unless it is empty and the other is non-empty + if (ThreadLocalRandom.current().nextFloat() < this.highPrioProb) { + if (this.priorityDeques.get("high").isEmpty() && !this.priorityDeques.get("low").isEmpty()) { + return this.priorityDeques.get("low"); + } + return this.priorityDeques.get("high"); + } else { + if (this.priorityDeques.get("low").isEmpty() && !this.priorityDeques.get("high").isEmpty()) { + return this.priorityDeques.get("high"); } + return this.priorityDeques.get("low"); } - return dequeForExtraction; } private LinkedBlockingDeque getDequeForInsertion(T p) { - int priority = p.getPriority(); + String priority = p.getPriority(); LinkedBlockingDeque dequeForInsertion = this.priorityDeques.get(priority); if (dequeForInsertion == null) { - logger.warn("Priority value " + String.valueOf(priority) + " not valid, setting to highest valid priority value " - + String.valueOf(this.nPriorities - 1) + "."); - int newPriority = this.nPriorities - 1; - p.setPriority(newPriority); - dequeForInsertion = this.priorityDeques.get(newPriority); + logger.warn("Priority value '" + priority + "' not valid, setting to 'low'."); + p.setPriority("low"); + dequeForInsertion = this.priorityDeques.get("low"); } return dequeForInsertion; @@ -98,9 +81,9 @@ private T unlinkFirst() { } public boolean isEmpty() { + // return true iff all deques are empty Function, Boolean> getIsEmpty = (LinkedBlockingDeque deque) -> deque.isEmpty(); BiFunction logicalAnd = (Boolean a, Boolean b) -> a && b; - // return true iff all deques are empty return this.priorityDeques.reduceValues(Long.MAX_VALUE, getIsEmpty, logicalAnd); } diff --git a/frontend/server/src/main/java/org/pytorch/serve/wlm/Model.java b/frontend/server/src/main/java/org/pytorch/serve/wlm/Model.java index e4f5c80d12..565b301e53 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/wlm/Model.java +++ b/frontend/server/src/main/java/org/pytorch/serve/wlm/Model.java @@ -40,7 +40,7 @@ public class Model { private ReentrantLock lock; private int responseTimeout; private int queueSize; - private int nPriorities; + private float highPrioProb; private ModelVersionName modelVersionName; private boolean isWorkflowModel; @@ -51,15 +51,15 @@ public class Model { // Per worker thread job queue. This separates out the control queue from data queue private ConcurrentMap> jobsDb; - public Model(ModelArchive modelArchive, int queueSize, int nPriorities) { + public Model(ModelArchive modelArchive, int queueSize, float highPrioProb) { this.modelArchive = modelArchive; this.queueSize = queueSize; - this.nPriorities = nPriorities; + this.highPrioProb = highPrioProb; batchSize = 1; maxBatchDelay = 100; jobsDb = new ConcurrentHashMap<>(); // Always have a queue for data - jobsDb.putIfAbsent(DEFAULT_DATA_QUEUE, new PriorityLinkedBlockingDeque<>(this.nPriorities, this.queueSize)); + jobsDb.putIfAbsent(DEFAULT_DATA_QUEUE, new PriorityLinkedBlockingDeque<>(this.queueSize, this.highPrioProb)); failedInfReqs = new AtomicInteger(0); lock = new ReentrantLock(); modelVersionName = @@ -156,7 +156,7 @@ public void setWorkflowModel(boolean workflowModel) { public void addJob(String threadId, Job job) { PriorityLinkedBlockingDeque blockingDeque = jobsDb.get(threadId); if (blockingDeque == null) { - blockingDeque = new PriorityLinkedBlockingDeque<>(this.nPriorities, this.queueSize); + blockingDeque = new PriorityLinkedBlockingDeque<>(this.queueSize, this.highPrioProb); jobsDb.put(threadId, blockingDeque); } blockingDeque.offer(job); diff --git a/frontend/server/src/main/java/org/pytorch/serve/wlm/ModelManager.java b/frontend/server/src/main/java/org/pytorch/serve/wlm/ModelManager.java index 8450449e61..462d4d3023 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/wlm/ModelManager.java +++ b/frontend/server/src/main/java/org/pytorch/serve/wlm/ModelManager.java @@ -264,7 +264,7 @@ private Model createModel( int maxBatchDelay, int responseTimeout, boolean isWorkflowModel) { - Model model = new Model(archive, configManager.getJobQueueSize(), configManager.getNumberOfPriorities()); + Model model = new Model(archive, configManager.getJobQueueSize(), configManager.getHighPrioProb()); model.setBatchSize( configManager.getJsonIntValue( @@ -290,7 +290,7 @@ private Model createModel( } private Model createModel(ModelArchive archive, JsonObject modelInfo) { - Model model = new Model(archive, configManager.getJobQueueSize(), configManager.getNumberOfPriorities()); + Model model = new Model(archive, configManager.getJobQueueSize(), configManager.getHighPrioProb()); model.setModelState(modelInfo); model.setWorkflowModel(false); From c5d64754134120a618a9746f10d8ec7dd9bf57c0 Mon Sep 17 00:00:00 2001 From: mpoemsl <37959974+mpoemsl@users.noreply.github.com> Date: Mon, 20 Mar 2023 09:57:31 +0100 Subject: [PATCH 2/9] Adapt RequestPriority metric to categorical value --- .../server/src/main/java/org/pytorch/serve/job/GRPCJob.java | 4 ++-- .../server/src/main/java/org/pytorch/serve/job/RestJob.java | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/frontend/server/src/main/java/org/pytorch/serve/job/GRPCJob.java b/frontend/server/src/main/java/org/pytorch/serve/job/GRPCJob.java index 3a868f2354..a1cf12978f 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/job/GRPCJob.java +++ b/frontend/server/src/main/java/org/pytorch/serve/job/GRPCJob.java @@ -86,8 +86,8 @@ public void response( "{}", new Metric( "RequestPriority", - String.valueOf(this.getPriority()), - "int", + this.getPriority(), + "category", ConfigManager.getInstance().getHostName(), DIMENSION)); } else if (this.getCmd() == WorkerCommands.DESCRIBE) { diff --git a/frontend/server/src/main/java/org/pytorch/serve/job/RestJob.java b/frontend/server/src/main/java/org/pytorch/serve/job/RestJob.java index 5e0d919c04..94c4e5e288 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/job/RestJob.java +++ b/frontend/server/src/main/java/org/pytorch/serve/job/RestJob.java @@ -163,8 +163,8 @@ private void responseInference( "{}", new Metric( "RequestPriority", - String.valueOf(this.getPriority()), - "int", + this.getPriority(), + "category", ConfigManager.getInstance().getHostName(), DIMENSION)); } From c5a1c597ade6f9fe9b199474338e98355212ba11 Mon Sep 17 00:00:00 2001 From: mpoemsl <37959974+mpoemsl@users.noreply.github.com> Date: Mon, 20 Mar 2023 14:33:31 +0100 Subject: [PATCH 3/9] Lowercase all headers --- .../pytorch/serve/http/api/rest/InferenceRequestHandler.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/frontend/server/src/main/java/org/pytorch/serve/http/api/rest/InferenceRequestHandler.java b/frontend/server/src/main/java/org/pytorch/serve/http/api/rest/InferenceRequestHandler.java index 10774a1206..7cc6a21175 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/http/api/rest/InferenceRequestHandler.java +++ b/frontend/server/src/main/java/org/pytorch/serve/http/api/rest/InferenceRequestHandler.java @@ -274,7 +274,7 @@ private static RequestInput parseRequest( CharSequence contentType = HttpUtil.getMimeType(req); for (Map.Entry entry : req.headers().entries()) { - inputData.updateHeaders(entry.getKey(), entry.getValue()); + inputData.updateHeaders(entry.getKey().toLowerCase(), entry.getValue()); } if (HttpPostRequestDecoder.isMultipart(req) From 5821b1dc52b13ea704d1871ddc0b5d5258305c3b Mon Sep 17 00:00:00 2001 From: mpoemsl <37959974+mpoemsl@users.noreply.github.com> Date: Mon, 20 Mar 2023 14:47:17 +0100 Subject: [PATCH 4/9] Make priority enum --- .../pytorch/serve/grpcimpl/InferenceImpl.java | 2 +- .../java/org/pytorch/serve/job/GRPCJob.java | 2 +- .../main/java/org/pytorch/serve/job/Job.java | 10 +++++----- .../java/org/pytorch/serve/job/RestJob.java | 2 +- .../java/org/pytorch/serve/util/ApiUtils.java | 2 +- .../org/pytorch/serve/util/Prioritisable.java | 4 ++-- .../java/org/pytorch/serve/util/Priority.java | 5 +++++ .../util/PriorityLinkedBlockingDeque.java | 20 ++++++------------- 8 files changed, 22 insertions(+), 25 deletions(-) create mode 100644 frontend/server/src/main/java/org/pytorch/serve/util/Priority.java diff --git a/frontend/server/src/main/java/org/pytorch/serve/grpcimpl/InferenceImpl.java b/frontend/server/src/main/java/org/pytorch/serve/grpcimpl/InferenceImpl.java index c85422d76b..1a2242a2e7 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/grpcimpl/InferenceImpl.java +++ b/frontend/server/src/main/java/org/pytorch/serve/grpcimpl/InferenceImpl.java @@ -103,7 +103,7 @@ public void predictions( try { if (!ModelManager.getInstance().addJob(job)) { - String priority = job.getPriority(); + String priority = job.getPriority().toString(); String responseMessage = ApiUtils.getInferenceErrorResponseMessage(modelName, modelVersion, priority); InternalServerException e = new InternalServerException(responseMessage); diff --git a/frontend/server/src/main/java/org/pytorch/serve/job/GRPCJob.java b/frontend/server/src/main/java/org/pytorch/serve/job/GRPCJob.java index a1cf12978f..1b90af7c7d 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/job/GRPCJob.java +++ b/frontend/server/src/main/java/org/pytorch/serve/job/GRPCJob.java @@ -86,7 +86,7 @@ public void response( "{}", new Metric( "RequestPriority", - this.getPriority(), + this.getPriority().toString(), "category", ConfigManager.getInstance().getHostName(), DIMENSION)); diff --git a/frontend/server/src/main/java/org/pytorch/serve/job/Job.java b/frontend/server/src/main/java/org/pytorch/serve/job/Job.java index 056a868536..1c6cae0e38 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/job/Job.java +++ b/frontend/server/src/main/java/org/pytorch/serve/job/Job.java @@ -4,6 +4,7 @@ import org.pytorch.serve.util.messages.RequestInput; import org.pytorch.serve.util.messages.WorkerCommands; import org.pytorch.serve.util.Prioritisable; +import org.pytorch.serve.util.Priority; public abstract class Job implements Prioritisable { @@ -11,7 +12,7 @@ public abstract class Job implements Prioritisable { private String modelVersion; private WorkerCommands cmd; // Else its data msg or inf requests private RequestInput input; - private String priority; + private Priority priority; private long begin; private long scheduled; @@ -22,15 +23,14 @@ public Job(String modelName, String version, WorkerCommands cmd, RequestInput in this.modelVersion = version; begin = System.nanoTime(); scheduled = begin; - - this.priority = input.getHeaders().getOrDefault("x-ts-priority", "max"); + this.priority = Priority.valueOf(input.getHeaders().getOrDefault("x-ts-priority", "MAX").toUpperCase()); } - public String getPriority() { + public Priority getPriority() { return this.priority; } - public void setPriority(String priority) { + public void setPriority(Priority priority) { this.priority = priority; } diff --git a/frontend/server/src/main/java/org/pytorch/serve/job/RestJob.java b/frontend/server/src/main/java/org/pytorch/serve/job/RestJob.java index 94c4e5e288..adacba25c9 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/job/RestJob.java +++ b/frontend/server/src/main/java/org/pytorch/serve/job/RestJob.java @@ -163,7 +163,7 @@ private void responseInference( "{}", new Metric( "RequestPriority", - this.getPriority(), + this.getPriority().toString(), "category", ConfigManager.getInstance().getHostName(), DIMENSION)); diff --git a/frontend/server/src/main/java/org/pytorch/serve/util/ApiUtils.java b/frontend/server/src/main/java/org/pytorch/serve/util/ApiUtils.java index 0dcdc3a0b8..978eac4a37 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/util/ApiUtils.java +++ b/frontend/server/src/main/java/org/pytorch/serve/util/ApiUtils.java @@ -394,7 +394,7 @@ public static RestJob addRESTInferenceJob( throws ModelNotFoundException, ModelVersionNotFoundException { RestJob job = new RestJob(ctx, modelName, version, WorkerCommands.PREDICT, input); if (!ModelManager.getInstance().addJob(job)) { - String priority = job.getPriority(); + String priority = job.getPriority().toString(); String responseMessage = getInferenceErrorResponseMessage(modelName, version, priority); throw new ServiceUnavailableException(responseMessage); } diff --git a/frontend/server/src/main/java/org/pytorch/serve/util/Prioritisable.java b/frontend/server/src/main/java/org/pytorch/serve/util/Prioritisable.java index 88b109f213..752d64f9d6 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/util/Prioritisable.java +++ b/frontend/server/src/main/java/org/pytorch/serve/util/Prioritisable.java @@ -2,7 +2,7 @@ public interface Prioritisable { - public String getPriority(); - public void setPriority(String priority); + public Priority getPriority(); + public void setPriority(Priority priority); } diff --git a/frontend/server/src/main/java/org/pytorch/serve/util/Priority.java b/frontend/server/src/main/java/org/pytorch/serve/util/Priority.java new file mode 100644 index 0000000000..ce065431a8 --- /dev/null +++ b/frontend/server/src/main/java/org/pytorch/serve/util/Priority.java @@ -0,0 +1,5 @@ +package org.pytorch.serve.util; + +public enum Priority { + LOW, HIGH, MAX +} diff --git a/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java b/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java index 00c38658a8..1f1bfd85d4 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java +++ b/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java @@ -22,19 +22,18 @@ public class PriorityLinkedBlockingDeque { private int queueSize; private float highPrioProb; - private ConcurrentHashMap> priorityDeques; + private ConcurrentHashMap> priorityDeques; public PriorityLinkedBlockingDeque(int queueSize, float highPrioProb) { this.queueSize = queueSize; this.highPrioProb = highPrioProb; - this.priorityDeques = new ConcurrentHashMap>(); + this.priorityDeques = new ConcurrentHashMap>(); // initialize priority deques - this.priorityDeques.put("max", new LinkedBlockingDeque(queueSize)); - this.priorityDeques.put("high", new LinkedBlockingDeque(queueSize)); - this.priorityDeques.put("low", new LinkedBlockingDeque(queueSize)); - + for (Priority priority : Priority.values()) { + this.priorityDeques.put(priority, new LinkedBlockingDeque(queueSize)); + } } private LinkedBlockingDeque getDequeForExtraction() { @@ -59,15 +58,8 @@ private LinkedBlockingDeque getDequeForExtraction() { } private LinkedBlockingDeque getDequeForInsertion(T p) { - String priority = p.getPriority(); + Priority priority = p.getPriority(); LinkedBlockingDeque dequeForInsertion = this.priorityDeques.get(priority); - - if (dequeForInsertion == null) { - logger.warn("Priority value '" + priority + "' not valid, setting to 'low'."); - p.setPriority("low"); - dequeForInsertion = this.priorityDeques.get("low"); - } - return dequeForInsertion; } From f93ad0ed114a4774268f929dbd84d7167aa5c452 Mon Sep 17 00:00:00 2001 From: mpoemsl <37959974+mpoemsl@users.noreply.github.com> Date: Mon, 20 Mar 2023 14:53:24 +0100 Subject: [PATCH 5/9] Validate high priority probability --- .../main/java/org/pytorch/serve/util/ConfigManager.java | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/frontend/server/src/main/java/org/pytorch/serve/util/ConfigManager.java b/frontend/server/src/main/java/org/pytorch/serve/util/ConfigManager.java index 9d0a1179ee..8aaac8d2ca 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/util/ConfigManager.java +++ b/frontend/server/src/main/java/org/pytorch/serve/util/ConfigManager.java @@ -362,8 +362,13 @@ public int getJobQueueSize() { return getIntProperty(TS_JOB_QUEUE_SIZE, 100); } - public float getHighPrioProb() { - return getFloatProperty(TS_HIGH_PRIORITY_PROBABILITY, 0.67f); + public float getHighPrioProb() throws IllegalArgumentException { + float highPrioProb = getFloatProperty(TS_HIGH_PRIORITY_PROBABILITY, 0.67f); + if (highPrioProb < 0.00f || highPrioProb > 1.00f){ + throw new IllegalArgumentException("highPrioProb " + String.valueOf(highPrioProb) + + " is not a valid probability!"); + } + return highPrioProb; } public int getNumberOfGpu() { From 9866c091e5cab8c6a4fa4b8276a188229671d1e6 Mon Sep 17 00:00:00 2001 From: mpoemsl <37959974+mpoemsl@users.noreply.github.com> Date: Mon, 20 Mar 2023 14:54:39 +0100 Subject: [PATCH 6/9] Shorten empty check --- .../org/pytorch/serve/util/PriorityLinkedBlockingDeque.java | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java b/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java index 1f1bfd85d4..f9487f9291 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java +++ b/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java @@ -74,9 +74,7 @@ private T unlinkFirst() { public boolean isEmpty() { // return true iff all deques are empty - Function, Boolean> getIsEmpty = (LinkedBlockingDeque deque) -> deque.isEmpty(); - BiFunction logicalAnd = (Boolean a, Boolean b) -> a && b; - return this.priorityDeques.reduceValues(Long.MAX_VALUE, getIsEmpty, logicalAnd); + return this.priorityDeques.reduceValues(Long.MAX_VALUE, LinkedBlockingDeque::isEmpty, Boolean::logicalAnd); } public boolean offer(T p) { From 59de5e1e3186411bea2d32f9f26c5078f3f4deda Mon Sep 17 00:00:00 2001 From: mpoemsl <37959974+mpoemsl@users.noreply.github.com> Date: Mon, 20 Mar 2023 14:55:26 +0100 Subject: [PATCH 7/9] Remove unnecessary queueSize storage --- .../org/pytorch/serve/util/PriorityLinkedBlockingDeque.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java b/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java index f9487f9291..9cf957605a 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java +++ b/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java @@ -20,13 +20,11 @@ public class PriorityLinkedBlockingDeque { final ReentrantLock lock = new ReentrantLock(); private final Condition notEmpty = lock.newCondition(); - private int queueSize; private float highPrioProb; private ConcurrentHashMap> priorityDeques; public PriorityLinkedBlockingDeque(int queueSize, float highPrioProb) { - this.queueSize = queueSize; this.highPrioProb = highPrioProb; this.priorityDeques = new ConcurrentHashMap>(); From 75147729dd4497b4040ce0b7e1ec4eb51aa95e68 Mon Sep 17 00:00:00 2001 From: mpoemsl <37959974+mpoemsl@users.noreply.github.com> Date: Mon, 20 Mar 2023 14:56:30 +0100 Subject: [PATCH 8/9] Make private attributes final --- .../org/pytorch/serve/util/PriorityLinkedBlockingDeque.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java b/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java index 9cf957605a..cec2299d0e 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java +++ b/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java @@ -20,8 +20,8 @@ public class PriorityLinkedBlockingDeque { final ReentrantLock lock = new ReentrantLock(); private final Condition notEmpty = lock.newCondition(); - private float highPrioProb; - private ConcurrentHashMap> priorityDeques; + private final float highPrioProb; + private final ConcurrentHashMap> priorityDeques; public PriorityLinkedBlockingDeque(int queueSize, float highPrioProb) { From 5b93145329fc680a62adf756d13310d92ae328dc Mon Sep 17 00:00:00 2001 From: mpoemsl <37959974+mpoemsl@users.noreply.github.com> Date: Mon, 20 Mar 2023 15:18:32 +0100 Subject: [PATCH 9/9] Refactor random deque selection --- .../util/PriorityLinkedBlockingDeque.java | 29 +++++++++++-------- 1 file changed, 17 insertions(+), 12 deletions(-) diff --git a/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java b/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java index cec2299d0e..1d734796ea 100644 --- a/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java +++ b/frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java @@ -37,22 +37,27 @@ public PriorityLinkedBlockingDeque(int queueSize, float highPrioProb) { private LinkedBlockingDeque getDequeForExtraction() { // always select deque max first if non-empty - if (!this.priorityDeques.get("max").isEmpty()) { - return this.priorityDeques.get("max"); + if (!this.priorityDeques.get(Priority.MAX).isEmpty()) { + return this.priorityDeques.get(Priority.MAX); } - // return random deque unless it is empty and the other is non-empty - if (ThreadLocalRandom.current().nextFloat() < this.highPrioProb) { - if (this.priorityDeques.get("high").isEmpty() && !this.priorityDeques.get("low").isEmpty()) { - return this.priorityDeques.get("low"); - } - return this.priorityDeques.get("high"); - } else { - if (this.priorityDeques.get("low").isEmpty() && !this.priorityDeques.get("high").isEmpty()) { - return this.priorityDeques.get("high"); + boolean highNonEmpty = !this.priorityDeques.get(Priority.HIGH).isEmpty(); + + // if both high and low are non-empty, make random selection + if (highNonEmpty && !this.priorityDeques.get(Priority.LOW).isEmpty()) { + if (ThreadLocalRandom.current().nextFloat() < this.highPrioProb) { + return this.priorityDeques.get(Priority.HIGH); + } else { + return this.priorityDeques.get(Priority.LOW); } - return this.priorityDeques.get("low"); + // if only high is non-empty, return high + } else if (highNonEmpty) { + return this.priorityDeques.get(Priority.HIGH); } + + // if both empty or only low non-empty, return low + return this.priorityDeques.get(Priority.LOW); + } private LinkedBlockingDeque getDequeForInsertion(T p) {