You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemml.apache.org by ni...@apache.org on 2019/03/29 17:36:03 UTC
[systemml] branch master updated: [SYSTEMML-2525] Initial
implementation of RESTful model serving system
This is an automated email from the ASF dual-hosted git repository.
niketanpansare pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/systemml.git
The following commit(s) were added to refs/heads/master by this push:
new 863c9d5 [SYSTEMML-2525] Initial implementation of RESTful model serving system
863c9d5 is described below
commit 863c9d5cb1752b0e50140f5c6673968b57c2f9d0
Author: Anthony Thomas <ah...@eng.ucsd.edu>
AuthorDate: Fri Mar 29 10:27:54 2019 -0700
[SYSTEMML-2525] Initial implementation of RESTful model serving system
- The current implementation extends JMLC's readMatrix and GPUContext API.
- The serving system is implemented in Scala using Akka and is available in the org.apache.sysml.api.ml.serving.
- Minor cleanup and refactoring required before it's ready to be used by the general public will be done in subsequent commits.
- It still remains unclear whether CUDA and Serving code should be included in future standalone releases. If yes, it will greatly simplify the deployment, else the user will have to build standalone jar before deployment.
- The serving system can be started by:
```
mvn -Djcuda.scope=compile -Dserving.scope=compile package -P standalone-jar
java -jar systemml-*-standalone.jar org.apache.sysml.api.ml.serving.PredictionService -port 8099 -scheduler scheduler -admin_password admin
```
- The model can registered using http://localhost:8099/register-model and user can invoke prediction using http://localhost:8099/predict service.
Closes #860.
---
.travis.yml | 5 +-
pom.xml | 31 ++
.../java/org/apache/sysml/api/jmlc/Connection.java | 51 +++
.../org/apache/sysml/api/jmlc/PreparedScript.java | 18 +
.../org/apache/sysml/parser/DataExpression.java | 1 +
.../runtime/controlprogram/LocalVariableMap.java | 4 +
.../org/apache/sysml/utils/PersistentLRUCache.java | 97 ++--
.../api/ml/serving/BasicBatchingScheduler.scala | 93 ++++
.../sysml/api/ml/serving/BatchingScheduler.scala | 99 +++++
.../sysml/api/ml/serving/BatchingUtils.scala | 57 +++
.../org/apache/sysml/api/ml/serving/Executor.scala | 155 +++++++
.../api/ml/serving/LocalityAwareScheduler.scala | 218 +++++++++
.../apache/sysml/api/ml/serving/ModelManager.scala | 176 ++++++++
.../api/ml/serving/NonBatchingScheduler.scala | 69 +++
.../sysml/api/ml/serving/PredictionService.scala | 490 +++++++++++++++++++++
.../apache/sysml/api/ml/serving/RLSEstimator.scala | 91 ++++
.../apache/sysml/api/ml/serving/Scheduler.scala | 133 ++++++
.../sysml/api/ml/serving/SchedulerFactory.scala | 29 ++
18 files changed, 1756 insertions(+), 61 deletions(-)
diff --git a/.travis.yml b/.travis.yml
index a0c308b..3ce9d06 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -46,7 +46,8 @@ before_script:
script:
# - mvn clean verify jacoco:report coveralls:report
- - mvn clean verify
+# The -q parameter tells mvn not to display anything other than ERROR level log messages. This is required because travis kills the job after the log length exceeds its maximum log length (usually 4 MB).
+ - mvn -q clean verify
after_success:
-# - mvn test jacoco:report coveralls:report
\ No newline at end of file
+# - mvn test jacoco:report coveralls:report
diff --git a/pom.xml b/pom.xml
index ad74276..4b5dd29 100644
--- a/pom.xml
+++ b/pom.xml
@@ -72,6 +72,7 @@
<maven.build.timestamp.format>yyyy-MM-dd HH:mm:ss z</maven.build.timestamp.format>
<enableGPU>false</enableGPU>
<jcuda.scope>provided</jcuda.scope>
+ <serving.scope>provided</serving.scope>
<jcuda.version>0.9.0d</jcuda.version>
<!-- OS-specific JVM arguments for running integration tests -->
<integrationTestExtraJVMArgs />
@@ -1259,6 +1260,36 @@
<version>3.2.0</version>
</dependency>
<dependency>
+ <groupId>com.typesafe.akka</groupId>
+ <artifactId>akka-http_2.11</artifactId>
+ <version>10.1.3</version>
+ <scope>${serving.scope}</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.typesafe.akka</groupId>
+ <artifactId>akka-actor_2.11</artifactId>
+ <version>2.5.14</version>
+ <scope>${serving.scope}</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.typesafe.akka</groupId>
+ <artifactId>akka-stream_2.11</artifactId>
+ <version>2.5.14</version>
+ <scope>${serving.scope}</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.typesafe</groupId>
+ <artifactId>config</artifactId>
+ <version>1.2.0</version>
+ <scope>${serving.scope}</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.typesafe.akka</groupId>
+ <artifactId>akka-http-spray-json-experimental_2.11</artifactId>
+ <version>2.4.11.2</version>
+ <scope>${serving.scope}</scope>
+ </dependency>
+ <dependency>
<groupId>org.jcuda</groupId>
<artifactId>jcuda</artifactId>
<version>${jcuda.version}</version>
diff --git a/src/main/java/org/apache/sysml/api/jmlc/Connection.java b/src/main/java/org/apache/sysml/api/jmlc/Connection.java
index 53b7d04..29df4c0 100644
--- a/src/main/java/org/apache/sysml/api/jmlc/Connection.java
+++ b/src/main/java/org/apache/sysml/api/jmlc/Connection.java
@@ -370,6 +370,57 @@ public class Connection implements Closeable
// Read matrices
////////////////////////////////////////////
+ public MatrixBlock readMatrix(String fname) throws IOException {
+ try {
+ String fnamemtd = DataExpression.getMTDFileName(fname);
+ JSONObject jmtd = new DataExpression().readMetadataFile(fnamemtd, false);
+
+ //parse json meta data
+ long rows = jmtd.getLong(DataExpression.READROWPARAM);
+ long cols = jmtd.getLong(DataExpression.READCOLPARAM);
+ int brlen = jmtd.containsKey(DataExpression.ROWBLOCKCOUNTPARAM)?
+ jmtd.getInt(DataExpression.ROWBLOCKCOUNTPARAM) : -1;
+ int bclen = jmtd.containsKey(DataExpression.COLUMNBLOCKCOUNTPARAM)?
+ jmtd.getInt(DataExpression.COLUMNBLOCKCOUNTPARAM) : -1;
+ long nnz = jmtd.containsKey(DataExpression.READNNZPARAM)?
+ jmtd.getLong(DataExpression.READNNZPARAM) : -1;
+ String format = jmtd.getString(DataExpression.FORMAT_TYPE);
+ InputInfo iinfo = InputInfo.stringExternalToInputInfo(format);
+ return readMatrix(fname, iinfo, rows, cols, brlen, bclen, nnz);
+ } catch (Exception ex) {
+ throw new IOException(ex);
+ }
+ }
+
+ /**
+ * Reads an input matrix in arbitrary format from HDFS into a dense double array.
+ * NOTE: this call currently only supports default configurations for CSV.
+ *
+ * @param fname the filename of the input matrix
+ * @param iinfo InputInfo object
+ * @param rows number of rows in the matrix
+ * @param cols number of columns in the matrix
+ * @param brlen number of rows per block
+ * @param bclen number of columns per block
+ * @param nnz number of non-zero values, -1 indicates unknown
+ * @return matrix as a two-dimensional double array
+ * @throws IOException if IOException occurs
+ */
+ public MatrixBlock readMatrix(String fname, InputInfo iinfo, long rows, long cols, int brlen, int bclen, long nnz)
+ throws IOException
+ {
+ setLocalConfigs();
+
+ try {
+ MatrixReader reader = MatrixReaderFactory.createMatrixReader(iinfo);
+ return reader.readMatrixFromHDFS(fname, rows, cols, brlen, bclen, nnz);
+
+ }
+ catch(Exception ex) {
+ throw new IOException(ex);
+ }
+ }
+
/**
* Reads an input matrix in arbitrary format from HDFS into a dense double array.
* NOTE: this call currently only supports default configurations for CSV.
diff --git a/src/main/java/org/apache/sysml/api/jmlc/PreparedScript.java b/src/main/java/org/apache/sysml/api/jmlc/PreparedScript.java
index 701af30..5926bcc 100644
--- a/src/main/java/org/apache/sysml/api/jmlc/PreparedScript.java
+++ b/src/main/java/org/apache/sysml/api/jmlc/PreparedScript.java
@@ -75,6 +75,11 @@ public class PreparedScript implements ConfigurableAPI
private final HashSet<String> _outVarnames;
private final HashMap<String,Data> _inVarReuse;
+ private String name = "";
+ public void setName(String name) {
+ this.name = name;
+ }
+
//internal state (reused)
private final Program _prog;
private final LocalVariableMap _vars;
@@ -131,6 +136,19 @@ public class PreparedScript implements ConfigurableAPI
_cconf = cconf;
}
+ public void clearPinnedData() {
+ this._inVarReuse.clear();
+ }
+
+ public boolean hasPinnedData() {
+ return _inVarReuse.keySet().size() > 0;
+ }
+
+ public void setGpuContext(GPUContext gCtx) {
+ this._gpuCtx.set(0, gCtx);
+ }
+
+
/**
* Sets a boolean flag indicating if runtime statistics should be gathered
* Same behavior as in "MLContext.setStatistics()"
diff --git a/src/main/java/org/apache/sysml/parser/DataExpression.java b/src/main/java/org/apache/sysml/parser/DataExpression.java
index 44f368e..7b64922 100644
--- a/src/main/java/org/apache/sysml/parser/DataExpression.java
+++ b/src/main/java/org/apache/sysml/parser/DataExpression.java
@@ -44,6 +44,7 @@ import org.apache.sysml.runtime.util.UtilFunctions;
import org.apache.sysml.utils.JSONHelper;
import org.apache.wink.json4j.JSONArray;
import org.apache.wink.json4j.JSONObject;
+import org.apache.sysml.parser.Expression.DataOp;
public class DataExpression extends DataIdentifier
diff --git a/src/main/java/org/apache/sysml/runtime/controlprogram/LocalVariableMap.java b/src/main/java/org/apache/sysml/runtime/controlprogram/LocalVariableMap.java
index cf9e79a..bb06849 100644
--- a/src/main/java/org/apache/sysml/runtime/controlprogram/LocalVariableMap.java
+++ b/src/main/java/org/apache/sysml/runtime/controlprogram/LocalVariableMap.java
@@ -131,6 +131,10 @@ public class LocalVariableMap implements Cloneable
put(kv.getKey(), kv.getValue());
}
}
+
+ public void putAll(LocalVariableMap vars) {
+ putAll(vars.localMap);
+ }
public Data remove( String name ) {
Data ret = localMap.remove( name );
diff --git a/src/main/java/org/apache/sysml/utils/PersistentLRUCache.java b/src/main/java/org/apache/sysml/utils/PersistentLRUCache.java
index d9d9337..24e685b 100644
--- a/src/main/java/org/apache/sysml/utils/PersistentLRUCache.java
+++ b/src/main/java/org/apache/sysml/utils/PersistentLRUCache.java
@@ -86,7 +86,7 @@ public class PersistentLRUCache extends LinkedHashMap<String, ValueWrapper> {
private String _prefixFilePath;
final AtomicLong _currentNumBytes = new AtomicLong();
private final long _maxNumBytes;
- private static final Random _rand = new Random();
+ Random _rand = new Random();
boolean isInReadOnlyMode;
HashSet<String> persistedKeys = new HashSet<>();
@@ -101,9 +101,6 @@ public class PersistentLRUCache extends LinkedHashMap<String, ValueWrapper> {
for(long i = 0; i < numIter; ++i) {
LOG.debug("Putting a double array of size 50MB.");
cache.put("file_" + i, new double[numDoubleIn50MB]);
- try {
- Thread.sleep(100);
- } catch (InterruptedException e) {}
}
cache.clear();
}
@@ -130,13 +127,13 @@ public class PersistentLRUCache extends LinkedHashMap<String, ValueWrapper> {
_prefixFilePath = tmp.getAbsolutePath();
}
public ValueWrapper put(String key, double[] value) throws FileNotFoundException, IOException {
- return putImplm(key, new ValueWrapper(new DataWrapper(key, value, this), isInReadOnlyMode), value.length*Double.BYTES);
+ return putImplm(key, new ValueWrapper(new DataWrapper(key, value, this)), value.length*Double.BYTES);
}
public ValueWrapper put(String key, float[] value) throws FileNotFoundException, IOException {
- return putImplm(key, new ValueWrapper(new DataWrapper(key, value, this), isInReadOnlyMode), value.length*Float.BYTES);
+ return putImplm(key, new ValueWrapper(new DataWrapper(key, value, this)), value.length*Float.BYTES);
}
public ValueWrapper put(String key, MatrixBlock value) throws FileNotFoundException, IOException {
- return putImplm(key, new ValueWrapper(new DataWrapper(key, value, this), isInReadOnlyMode), value.getInMemorySize());
+ return putImplm(key, new ValueWrapper(new DataWrapper(key, value, this)), value.getInMemorySize());
}
private ValueWrapper putImplm(String key, ValueWrapper value, long sizeInBytes) throws FileNotFoundException, IOException {
@@ -209,7 +206,7 @@ public class PersistentLRUCache extends LinkedHashMap<String, ValueWrapper> {
}
float [] tmp = new float[0];
- static String dummyKey = "RAND_KEY_" + Math.abs(_rand.nextLong()) + "_" + Math.abs(_rand.nextLong());
+ String dummyKey = "RAND_KEY_" + Math.abs(_rand.nextLong()) + "_" + Math.abs(_rand.nextLong());
void ensureCapacity(long newNumBytes) throws FileNotFoundException, IOException {
if(newNumBytes > _maxNumBytes) {
throw new DMLRuntimeException("Exceeds maximum capacity. Cannot put a value of size " + newNumBytes +
@@ -220,7 +217,7 @@ public class PersistentLRUCache extends LinkedHashMap<String, ValueWrapper> {
synchronized(this) {
if(LOG.isDebugEnabled())
LOG.debug("The required capacity (" + newCapacity + ") is greater than max capacity:" + _maxNumBytes);
- ValueWrapper dummyValue = new ValueWrapper(new DataWrapper(dummyKey, tmp, this), isInReadOnlyMode);
+ ValueWrapper dummyValue = new ValueWrapper(new DataWrapper(dummyKey, tmp, this));
int maxIter = size();
while(_currentNumBytes.get() > _maxNumBytes && maxIter > 0) {
super.put(dummyKey, dummyValue); // This will invoke removeEldestEntry, which will set _eldest
@@ -351,13 +348,17 @@ class DataWrapper {
_mo = value;
_cache = cache;
}
+ @Override
+ protected void finalize() throws Throwable {
+ super.finalize();
+ write(true);
+ }
- public synchronized void write(boolean forceAggresiveWrites) throws FileNotFoundException, IOException {
- if(_key.equals(PersistentLRUCache.dummyKey))
+ public synchronized void write(boolean isBeingGarbageCollected) throws FileNotFoundException, IOException {
+ if(_key.equals(_cache.dummyKey))
return;
-
- // Prepare for writing
_cache.makeRecent(_key); // Make it recent.
+
if(_dArr != null || _fArr != null || _mb != null || _mo != null) {
_cache._currentNumBytes.addAndGet(-getSize());
}
@@ -365,16 +366,14 @@ class DataWrapper {
if(!_cache.isInReadOnlyMode) {
String debugSuffix = null;
if(PersistentLRUCache.LOG.isDebugEnabled()) {
- if(forceAggresiveWrites)
- debugSuffix = " (aggressively written).";
+ if(isBeingGarbageCollected)
+ debugSuffix = " (is being garbage collected).";
else
debugSuffix = " (capacity exceeded).";
}
if(_dArr != null) {
- File file = new File(_cache.getFilePath(_key));
- file.deleteOnExit();
- try (ObjectOutputStream os = new ObjectOutputStream(new FileOutputStream(file))) {
+ try (ObjectOutputStream os = new ObjectOutputStream(new FileOutputStream(_cache.getFilePath(_key)))) {
os.writeInt(_dArr.length);
for(int i = 0; i < _dArr.length; i++) {
os.writeDouble(_dArr[i]);
@@ -385,9 +384,7 @@ class DataWrapper {
PersistentLRUCache.LOG.debug("Writing value (double[] of size " + getSize() + " bytes) for the key " + _key + " to disk" + debugSuffix);
}
else if(_fArr != null) {
- File file = new File(_cache.getFilePath(_key));
- file.deleteOnExit();
- try (ObjectOutputStream os = new ObjectOutputStream(new FileOutputStream(file))) {
+ try (ObjectOutputStream os = new ObjectOutputStream(new FileOutputStream(_cache.getFilePath(_key)))) {
os.writeInt(_fArr.length);
for(int i = 0; i < _fArr.length; i++) {
os.writeFloat(_fArr[i]);
@@ -398,13 +395,12 @@ class DataWrapper {
PersistentLRUCache.LOG.debug("Writing value (float[] of size " + getSize() + " bytes) for the key " + _key + " to disk" + debugSuffix);
}
else if(_mb != null) {
- File file = new File(_cache.getFilePath(_key));
- file.deleteOnExit();
- try(FastBufferedDataOutputStream os = new FastBufferedDataOutputStream(new ObjectOutputStream(new FileOutputStream(file)))) {
+ try(FastBufferedDataOutputStream os = new FastBufferedDataOutputStream(new ObjectOutputStream(new FileOutputStream(_cache.getFilePath(_key))))) {
os.writeLong(_mb.getInMemorySize());
_mb.write(os);
}
_cache.persistedKeys.add(_key);
+ System.err.println("Writing value (MatrixBlock of size " + getSize() + " bytes) for the key " + _key + " to disk" + debugSuffix);
if(PersistentLRUCache.LOG.isDebugEnabled())
PersistentLRUCache.LOG.debug("Writing value (MatrixBlock of size " + getSize() + " bytes) for the key " + _key + " to disk" + debugSuffix);
}
@@ -513,63 +509,46 @@ class DataWrapper {
// Internal helper class
class ValueWrapper {
final Object _lock;
- final boolean _isInReadOnlyMode;
- private SoftReference<DataWrapper> _softRef;
+ private SoftReference<DataWrapper> _ref;
long _rlen;
long _clen;
long _nnz;
- ValueWrapper(DataWrapper data, boolean isInReadOnlyMode) {
+ ValueWrapper(DataWrapper _data) {
_lock = new Object();
- _isInReadOnlyMode = isInReadOnlyMode;
- boolean isDummyValue = (data._key == PersistentLRUCache.dummyKey);
- if(!_isInReadOnlyMode && !isDummyValue) {
- // Aggressive write to disk when the cache is used in the write-mode.
- // This avoids the need to depend on finalize to perform writing.
- Thread t = new Thread() {
- public void run() {
- try {
- data.write(true);
- } catch (IOException e) {
- throw new DMLRuntimeException("Error occured while aggressively writing the value to disk.", e);
- }
- }
- };
- t.start();
- }
- _softRef = new SoftReference<>(data);
- if(data._mb != null) {
- _rlen = data._mb.getNumRows();
- _clen = data._mb.getNumColumns();
- _nnz = data._mb.getNonZeros();
+ _ref = new SoftReference<>(_data);
+ if(_data._mb != null) {
+ _rlen = _data._mb.getNumRows();
+ _clen = _data._mb.getNumColumns();
+ _nnz = _data._mb.getNonZeros();
}
}
- void update(DataWrapper data) {
- _softRef = new SoftReference<>(data);
- if(data._mb != null) {
- _rlen = data._mb.getNumRows();
- _clen = data._mb.getNumColumns();
- _nnz = data._mb.getNonZeros();
+ void update(DataWrapper _data) {
+ _ref = new SoftReference<>(_data);
+ if(_data._mb != null) {
+ _rlen = _data._mb.getNumRows();
+ _clen = _data._mb.getNumColumns();
+ _nnz = _data._mb.getNonZeros();
}
}
boolean isAvailable() {
- DataWrapper data = _softRef.get();
+ DataWrapper data = _ref.get();
return data != null && data.isAvailable();
}
DataWrapper get() {
- return _softRef.get();
+ return _ref.get();
}
long getSize() {
- DataWrapper data = _softRef.get();
+ DataWrapper data = _ref.get();
if(data != null)
return data.getSize();
else
return 0;
}
void remove() {
- DataWrapper data = _softRef.get();
+ DataWrapper data = _ref.get();
if(data != null) {
data.remove();
}
}
-}
+}
\ No newline at end of file
diff --git a/src/main/scala/org/apache/sysml/api/ml/serving/BasicBatchingScheduler.scala b/src/main/scala/org/apache/sysml/api/ml/serving/BasicBatchingScheduler.scala
new file mode 100644
index 0000000..6f19cce
--- /dev/null
+++ b/src/main/scala/org/apache/sysml/api/ml/serving/BasicBatchingScheduler.scala
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sysml.api.ml.serving
+
+import java.util.concurrent.{ConcurrentHashMap, CountDownLatch}
+
+import scala.concurrent.Future
+import scala.concurrent.duration.Duration
+import scala.math.min
+
+object BasicBatchingScheduler extends BatchingScheduler {
+
+ override def start(numCores: Int, cpuMemoryBudgetInBytes: Long, gpus: String): Unit = {
+ LOG.info(s"Starting Basic Batching Scheduler with: ${numCores} CPUs and ${gpus} GPUs")
+ super.start(numCores, cpuMemoryBudgetInBytes, gpus)
+ }
+
+ /**
+ * Returns a list of requests to execute. If the list contains more than one element, they will be batched
+ * by the executor. Returns an empty list when there are no models to be scheduled.
+ * @param executor an Executor instance
+ * @return a list of model requests to process
+ */
+ override def schedule(executor: JmlcExecutor) : Array[SchedulingRequest] = {
+ var ret = Array[SchedulingRequest]()
+ val execType = executor.getExecType
+ dummyResponse.synchronized {
+ val schedulableModels = getSchedulableModels(execType)
+ if (schedulableModels.nonEmpty) {
+ val (nextModel, nextBatchSize) = getNextModelAndBatchSize(schedulableModels, execType)
+ for (_ <- 0 until nextBatchSize) {
+ val next = modelQueues.get(nextModel).poll()
+ assert(next != null, "Something is wrong. Next model should not be null")
+ ret :+= next
+ }
+ }
+ }
+ ret
+ }
+
+ /**
+ * Helper method which gets the next model to schedule and the optimal batchsize
+ * @param models A list of models to schedule
+ * @return The model to schedule next
+ */
+ def getNextModelAndBatchSize(models : Iterable[String], execType: String) : (String, Int) = {
+ val nextModel = models.map(m =>
+ (getOptimalBatchSize(m, execType)*getExpectedExecutionTime(m), m)).minBy(x => x._1)._2
+
+ val nextBatchSize = min(modelQueues.get(nextModel).size(),
+ getOptimalBatchSize(nextModel, execType))
+ (nextModel, nextBatchSize)
+ }
+
+ /**
+ * Enqueues a request for processing. The scheduler will read from these queues to determine which
+ * models to execute next
+ * @param request A PredictionRequest object containing the data for which a prediction is desired
+ * @param model The model object for which prediction
+ * @return
+ */
+ override private[serving] def enqueue(request: PredictionRequest, model: Model): Future[PredictionResponse] = Future {
+ val statistics = if (_statistics) RequestStatistics() else null
+ val schedulingRequest = SchedulingRequest(
+ request, model, new CountDownLatch(1), System.nanoTime(), null, statistics)
+ statistics.queueSize = modelQueues.get(model.name).size
+ modelQueues.get(model.name).add(schedulingRequest)
+ counter += 1
+ try {
+ schedulingRequest.latch.await(timeout.length, timeout.unit)
+ schedulingRequest.response
+ } catch {
+ case e : scala.concurrent.TimeoutException => dummyResponse
+ }
+ }
+
+}
\ No newline at end of file
diff --git a/src/main/scala/org/apache/sysml/api/ml/serving/BatchingScheduler.scala b/src/main/scala/org/apache/sysml/api/ml/serving/BatchingScheduler.scala
new file mode 100644
index 0000000..e62f4ef
--- /dev/null
+++ b/src/main/scala/org/apache/sysml/api/ml/serving/BatchingScheduler.scala
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sysml.api.ml.serving
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.LongAdder
+
+import scala.math.{floor, max}
+
+trait BatchingScheduler extends Scheduler {
+
+ val modelBatchSizes = new ConcurrentHashMap[String, ConcurrentHashMap[String,Int]]()
+ val expectedExecutionTimes = new ConcurrentHashMap[String, (LongAdder, LongAdder)]()
+
+ def getOptimalBatchSize(model : String, execType: String) : Int = {
+ modelBatchSizes.putIfAbsent(execType, new ConcurrentHashMap[String,Int]())
+ modelBatchSizes.get(execType).putIfAbsent(model, 2)
+ modelBatchSizes.get(execType).get(model)
+ }
+
+ override def onCompleteCallback(model: String,
+ latency: Double,
+ batchSize: Int,
+ execType: String,
+ execTime: Long): Unit = {
+ if (batchSize > 1) {
+ val latencyObjective = latencyObjectives.get(model)
+ val prevSize = modelBatchSizes.get(execType).get(model)
+ val decreaseSize = if (prevSize > 10) max(floor(prevSize * 0.90).toInt, 1) else prevSize - 1
+ modelBatchSizes.get(execType).put(model,
+ if (latency < latencyObjective.toNanos) prevSize + 1 else decreaseSize)
+
+ // update expected execution times. For now we just assume this is a simple average
+ val execTimeData = expectedExecutionTimes.get(model)
+ execTimeData._1.add(execTime / batchSize)
+ execTimeData._2.increment()
+ }
+ }
+
+ def getExpectedExecutionTime(model: String) : Long = {
+ expectedExecutionTimes.putIfAbsent(model, (new LongAdder(), new LongAdder()))
+ val execTime = expectedExecutionTimes.get(model)
+ val totalNumRequests = execTime._2.longValue()
+ if (totalNumRequests > 0) execTime._1.longValue() / execTime._2.longValue() else 0
+ }
+
+ /**
+ * Gets a list of models that are eligible to be run. A model is eligible to be run if it
+ * has a greater number of requests enqueued than its optimal batch size.
+ * @return A list of models which may be scheduled
+ */
+ def getSchedulableModels(execType: String) : Set[String] = {
+ var batchableModels = Set[String]()
+ var shortFuse = Set[String]()
+ val keyIterator = modelQueues.keys()
+ while (keyIterator.hasMoreElements) {
+ val name = keyIterator.nextElement()
+ val qsize = modelQueues.get(name).size()
+ if (qsize > 0) {
+ val nextRequest = modelQueues.get(name).peek()
+ assert(nextRequest != null, "Something is wrong. Next request should not be null")
+
+ if (checkShortFuse(nextRequest, qsize)) {
+ LOG.info("Model: " + name + " is near violating threshold. Scheduling immediately.")
+ shortFuse += name
+ }
+
+ if (qsize >= getOptimalBatchSize(name, execType)) {
+ batchableModels += name
+ }
+ }
+ }
+
+ if (shortFuse.nonEmpty) shortFuse else batchableModels
+ }
+
+ /**
+ * Returns a boolean value if it would violate the latency threshold to execute the current number of models
+ */
+ def checkShortFuse(request: SchedulingRequest, numRequests: Int) : Boolean = {
+ val elapsed = System.nanoTime() - request.receivedTime
+ (elapsed + 1.1*numRequests*getExpectedExecutionTime(request.model.name)) > request.model.latencyObjective.toNanos
+ }
+}
\ No newline at end of file
diff --git a/src/main/scala/org/apache/sysml/api/ml/serving/BatchingUtils.scala b/src/main/scala/org/apache/sysml/api/ml/serving/BatchingUtils.scala
new file mode 100644
index 0000000..ca28e7f
--- /dev/null
+++ b/src/main/scala/org/apache/sysml/api/ml/serving/BatchingUtils.scala
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sysml.api.ml.serving
+import org.apache.sysml.runtime.matrix.data.MatrixBlock
+
+object BatchingUtils {
+ def batchRequests(requests: Array[SchedulingRequest]) : MatrixBlock = {
+ if (requests.length == 1) {
+ return requests(0).request.data
+ }
+ val ncol = requests(0).request.data.getNumColumns
+ val res = new MatrixBlock(requests.length, ncol, -1).allocateDenseBlock()
+ val doubles = res.getDenseBlockValues
+ var start = 0
+ for (req <- requests) {
+ System.arraycopy(req.request.data.getDenseBlockValues, 0, doubles, start, ncol)
+ start += ncol
+ }
+ res.setNonZeros(-1)
+ res
+ }
+
+ def unbatchRequests(requests: Array[SchedulingRequest],
+ batchedResults: MatrixBlock) : Array[PredictionResponse] = {
+ var responses = Array[PredictionResponse]()
+ val start = 0
+ for (req <- requests) {
+ val unbatchStart = System.nanoTime()
+ val resp = PredictionResponse(batchedResults.slice(
+ start, (start + req.request.requestSize)-1),
+ batchedResults.getNumRows, req.statistics)
+ val unbatchingTime = System.nanoTime() - unbatchStart
+ if (req.statistics != null)
+ req.statistics.unbatchingTime = unbatchingTime
+
+ responses :+= resp
+ }
+
+ responses
+ }
+}
\ No newline at end of file
diff --git a/src/main/scala/org/apache/sysml/api/ml/serving/Executor.scala b/src/main/scala/org/apache/sysml/api/ml/serving/Executor.scala
new file mode 100644
index 0000000..c353e07
--- /dev/null
+++ b/src/main/scala/org/apache/sysml/api/ml/serving/Executor.scala
@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysml.api.ml.serving
+import java.util.concurrent.PriorityBlockingQueue
+import java.util.concurrent.atomic.LongAdder
+
+import org.apache.commons.logging.{Log, LogFactory}
+import org.apache.sysml.runtime.instructions.gpu.context.GPUContext
+
+
+case class Batch(size: Int, expectedTime: Long, priority: Double, modelName: String) extends Comparable[Batch] {
+ override def compareTo(that: Batch): Int = {
+ this.priority.compareTo(that.priority)
+ }
+}
+
+class BatchQueue(execType: String, name: String) extends PriorityBlockingQueue[Batch] {
+ val LOG: Log = LogFactory.getLog(classOf[BatchQueue].getName)
+ private val expectedExecutionTime = new LongAdder()
+ private var prevFirstRequest= Map[String, SchedulingRequest]()
+
+ def getName : String = { name }
+
+ def updatePrevRequest(name: String, request: SchedulingRequest) : Unit = {
+ prevFirstRequest += (name -> request)
+ }
+
+ def getPrevRequest(name: String) : SchedulingRequest = { prevFirstRequest.getOrElse(name, null) }
+
+ def enqueue(batch: Batch) : Unit = {
+ LOG.debug("Enqueuing onto: " + getName)
+ synchronized {
+ this.add(batch)
+ expectedExecutionTime.add(batch.expectedTime)
+ }
+ }
+
+ def dequeue() : Batch = {
+ if (this.isEmpty)
+ return Batch(-1, -1, -1, "NO NAME")
+ synchronized {
+ val nextBatch = this.poll()
+ expectedExecutionTime.add(-1*nextBatch.expectedTime)
+ return nextBatch
+ }
+ }
+
+ def getExpectedExecutionTime : Long = { expectedExecutionTime.longValue() }
+
+ def getExecType : String = { execType }
+}
+
+class JmlcExecutor(scheduler: Scheduler, execType: String, name: String, gCtx: GPUContext) extends Runnable {
+ @volatile protected var _shouldShutdown: Boolean = false
+ val LOG: Log = LogFactory.getLog(classOf[JmlcExecutor].getName)
+ var prevModel = ""
+
+ def shutdown(): Unit = {
+ _shouldShutdown = true
+ }
+
+ def getExecType: String = { execType }
+
+ def getName: String = { name }
+
+ def run(): Unit = {
+ Thread.sleep(1000)
+ while (!_shouldShutdown) {
+ val requests = scheduler.schedule(this)
+ if (requests.nonEmpty) {
+ val responses = execute(requests)
+ for ((req, resp) <- requests zip responses) {
+ req.response = resp
+ req.latch.countDown()
+ }
+ }
+ }
+ }
+
+ def execute(requests: Array[SchedulingRequest]): Array[PredictionResponse] = {
+ var responses = Array[PredictionResponse]()
+ if (requests.nonEmpty) {
+ try {
+ val start = System.nanoTime()
+ val batchedMatrixData = BatchingUtils.batchRequests(requests)
+ val batchingTime = System.nanoTime() - start
+ val req = requests(0)
+ LOG.info("Executing: " + req.model.name + " with batch size: " + batchedMatrixData.getNumRows + " on " + name)
+ val modelAcquireStart = System.nanoTime()
+ val script = scheduler.modelManager.acquire(req.model.name, this)
+ script.setName(this.getName)
+ val modelAcquireTime = System.nanoTime() - modelAcquireStart
+ script.setMatrix(req.model.inputVarName, batchedMatrixData, false)
+ val execStart = System.nanoTime()
+ val res = script.executeScript().getMatrixBlock(req.model.outputVarName)
+ val execTime = System.nanoTime() - execStart
+ responses = BatchingUtils.unbatchRequests(requests, res)
+
+ val modelReleaseStart = System.nanoTime()
+ scheduler.modelManager.release(req.model.name)
+ scheduler.modelManager.releaseMemory(req.memUse)
+ val modelReleaseTime = System.nanoTime() - modelReleaseStart
+ scheduler.onCompleteCallback(req.model.name,
+ System.nanoTime() - req.receivedTime,
+ requests.length,
+ execType, System.nanoTime() - start)
+ if (req.statistics != null)
+ setStatistics(requests, start, batchingTime, execTime, modelAcquireTime, modelReleaseTime)
+ if (prevModel.nonEmpty)
+ scheduler.modelManager.unsetModelLocality(prevModel, this)
+ scheduler.modelManager.setModelLocality(req.model.name, this)
+ prevModel = req.model.name
+
+ LOG.info("Done executing request for: " + req.model.name + " on " + name)
+ } catch {
+ case e: Exception => println("AN ERROR OCCURRED: " + e.getMessage + e.printStackTrace())
+ }
+ }
+ responses
+ }
+
+ def setStatistics(requests: Array[SchedulingRequest],
+ processingStartTime: Long,
+ batchingTime: Long,
+ execTime: Long,
+ modelAcquireTime: Long,
+ modelReleaseTime: Long): Unit = {
+ for (req <- requests) {
+ req.statistics.batchingTime = batchingTime
+ req.statistics.execType = getExecType
+ req.statistics.batchSize = requests.length
+ req.statistics.queueWaitTime = processingStartTime - req.receivedTime
+ req.statistics.execTime = execTime
+ req.statistics.modelAcquireTime = modelAcquireTime
+ req.statistics.modelReleaseTime = modelReleaseTime
+ }
+ }
+}
\ No newline at end of file
diff --git a/src/main/scala/org/apache/sysml/api/ml/serving/LocalityAwareScheduler.scala b/src/main/scala/org/apache/sysml/api/ml/serving/LocalityAwareScheduler.scala
new file mode 100644
index 0000000..61fc84f
--- /dev/null
+++ b/src/main/scala/org/apache/sysml/api/ml/serving/LocalityAwareScheduler.scala
@@ -0,0 +1,218 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sysml.api.ml.serving
+
+import java.util.concurrent.{ConcurrentHashMap, CountDownLatch}
+
+import org.apache.commons.logging.{Log, LogFactory}
+
+import scala.concurrent.Future
+import scala.math.min
+
+object ExecutorQueueManager extends Runnable {
+ val LOG: Log = LogFactory.getLog(ExecutorQueueManager.getClass.getName)
+ var _shutDown = false
+ var _scheduler = LocalityAwareScheduler
+ def shutdown(): Unit = { _shutDown = true }
+
+ override def run() : Unit = {
+ while (!_shutDown) {
+ _scheduler.dummyResponse.synchronized {
+ val schedulableModels = _scheduler.executorTypes.map(
+ x => _scheduler.getSchedulableModels(x)).reduce(_ union _)
+ if (schedulableModels.nonEmpty) {
+ for (m <- schedulableModels) {
+ // every request batch can go to up to three queues
+
+ // 1. Every batch goes to the global disk queue since the model might get evicted
+ val diskQueues = _scheduler.executorTypes.map(x => _scheduler.globalDiskQueues.get(x))
+
+ // 2. If the model is cached in memory, then also put it on the cache queue
+ var cacheQueues = Array[BatchQueue]()
+ if (_scheduler.modelManager.isCached(m))
+ cacheQueues = _scheduler.executorTypes.map(x => _scheduler.globalCacheQueues.get(x))
+
+ // 3. If the model is local to an executor, then put it on the lowest utilizaiton queue
+ val localExecutionQueues = getLocalExecutionQueues(m)
+ val localQueue = if (localExecutionQueues.nonEmpty)
+ Array[BatchQueue](localExecutionQueues.minBy(x => x.getExpectedExecutionTime))
+ else Array[BatchQueue]()
+
+ val queues = diskQueues ++ cacheQueues ++ localQueue
+ val nextRequest = _scheduler.modelQueues.get(m).peek()
+ queues.foreach ( queue => {
+ val qsize = _scheduler.modelQueues.get(m).size()
+ if (nextRequest ne queue.getPrevRequest(m)) {
+ val nextBatchSize = min(qsize, _scheduler.getOptimalBatchSize(m, queue.getExecType))
+ assert(nextBatchSize > 0, "An error occurred - batch size should not be zero")
+ LOG.debug("Enqueuing: " + nextBatchSize + " for: " + m + " onto: " + queue.getName)
+ val nextBatch = Batch(
+ nextBatchSize, nextBatchSize*_scheduler.getExpectedExecutionTime(m),
+ nextRequest.receivedTime - System.nanoTime(), nextRequest.model.name)
+ queue.enqueue(nextBatch)
+ LOG.debug("Batch enqueued onto: " + queue.getName)
+ }
+ queue.updatePrevRequest(m, nextRequest) } )
+ }
+ }
+ }
+ }
+ }
+
+ def getLocalExecutionQueues(model: String) : Array[BatchQueue] = {
+ val execs = _scheduler.modelManager.getModelLocality(model)
+ var queues = Array[BatchQueue]()
+ if (execs == null)
+ return queues
+
+ _scheduler.modelManager.synchronized({
+ for (ix <- 0 until execs.size()) { _scheduler.executorQueues.get(execs.get(ix)) }
+ })
+
+ queues
+ }
+}
+
+object ExecMode extends Enumeration {
+ type MODE = Value
+ val LOCAL, GLOBAL_MEM, GLOBAL_DISK = Value
+}
+
+object LocalityAwareScheduler extends BatchingScheduler {
+ var queueManager : Thread = _
+
+ val globalCacheQueues = new ConcurrentHashMap[String, BatchQueue]()
+ val globalDiskQueues = new ConcurrentHashMap[String, BatchQueue]()
+
+ override def start(numCores: Int, cpuMemoryBudgetInBytes: Long, gpus: String): Unit = {
+ super.start(numCores, cpuMemoryBudgetInBytes, gpus)
+
+ executorTypes.foreach ( x => {
+ globalCacheQueues.putIfAbsent(x, new BatchQueue(x, x + "-CACHE"))
+ globalDiskQueues.putIfAbsent(x, new BatchQueue(x, x + "-DISK"))
+ } )
+
+ queueManager = new Thread(ExecutorQueueManager)
+ queueManager.start()
+ }
+
+ override def addModel(model: Model): Unit = {
+ super.addModel(model)
+ }
+
+ override def schedule(executor: JmlcExecutor) : Array[SchedulingRequest] = {
+ var ret = Array[SchedulingRequest]()
+ val localQueue = executorQueues.get(executor)
+ val globalDiskQueue = globalDiskQueues.get(executor.getExecType)
+ val globalMemQueue = globalCacheQueues.get(executor.getExecType)
+ if (localQueue.size() > 0 || globalDiskQueue.size() > 0 || globalMemQueue.size() > 0) {
+ dummyResponse.synchronized {
+ if (localQueue.size() > 0 || globalDiskQueue.size() > 0 || globalMemQueue.size() > 0) {
+ LOG.debug("Begin scheduling for executor: " + executor.getName)
+ val execMode = Array[(BatchQueue, ExecMode.MODE)](
+ (localQueue, ExecMode.LOCAL),
+ (globalDiskQueue, ExecMode.GLOBAL_DISK),
+ (globalMemQueue, ExecMode.GLOBAL_MEM)
+ ).filter(x => x._1.size() > 0).maxBy(x => x._1.getExpectedExecutionTime)._2
+
+ val batch = execMode match {
+ case ExecMode.LOCAL => localQueue.peek()
+ case ExecMode.GLOBAL_MEM => globalMemQueue.peek()
+ case ExecMode.GLOBAL_DISK => globalDiskQueue.peek()
+ }
+ assert(batch != null, "Something is wrong. Batch should not be null!")
+
+ // now we need to ask the resource manager if there's enough memory to execute the batch
+ val model = modelManager.get(batch.modelName)
+
+ // If there's enough memory we can actually remove the requests from the queue and
+ // submit them for processing
+ val mqueue = modelQueues.get(batch.modelName)
+ val numToDequeue = min(batch.size, mqueue.size())
+
+ // if this value is zero there are no more requests and the batch is stale
+ if (numToDequeue == 0) {
+ execMode match {
+ case ExecMode.LOCAL => localQueue.poll()
+ case ExecMode.GLOBAL_DISK => globalDiskQueue.poll()
+ case ExecMode.GLOBAL_MEM => globalMemQueue.poll()
+ }
+ } else {
+ val memReceived = modelManager.tryAllocMem(model.name, batch.size)
+ if (memReceived < 0) {
+ return ret
+ }
+
+ // now we need to actually remove the request from the queue since it's going to be processed
+ execMode match {
+ case ExecMode.LOCAL => localQueue.poll()
+ case ExecMode.GLOBAL_DISK => globalDiskQueue.poll()
+ case ExecMode.GLOBAL_MEM => globalMemQueue.poll()
+ }
+
+ // now we can actually take the original requests out of the model queues
+ LOG.debug("Scheduling: " + numToDequeue + " for " + batch.modelName + " on " + executor.getName)
+ for (_ <- 0 until numToDequeue) {
+ val nextRequest = mqueue.poll()
+ assert(nextRequest != null, "Something is wrong - request should not be null!")
+
+ nextRequest.memUse = memReceived
+ nextRequest.statistics.execMode = execMode match {
+ case ExecMode.LOCAL => 0
+ case ExecMode.GLOBAL_MEM => 1
+ case ExecMode.GLOBAL_DISK => 2
+ case _ => -1
+ }
+ ret :+= nextRequest
+ }
+ LOG.debug("Done scheduling on: " + executor.getName)
+ }
+ }
+ }
+ }
+ ret
+ }
+
+ /**
+ * Enqueues a request for processing. The scheduler will read from these queues to determine which
+ * models to execute next
+ * @param request A PredictionRequest object containing the data for which a prediction is desired
+ * @param model The model object for which prediction
+ * @return
+ */
+ override private[serving] def enqueue(request: PredictionRequest, model: Model): Future[PredictionResponse] = Future {
+ val statistics = if (_statistics) RequestStatistics() else null
+ val schedulingRequest = SchedulingRequest(
+ request, model, new CountDownLatch(1), System.nanoTime(), null, statistics)
+
+ if (_statistics) {
+ statistics.queueSize = modelQueues.get(model.name).size
+ statistics.preprocWaitTime = System.nanoTime() - request.receivedTime
+ }
+
+ modelQueues.get(model.name).add(schedulingRequest)
+
+ try {
+ schedulingRequest.latch.await(timeout.length, timeout.unit)
+ schedulingRequest.response
+ } catch {
+ case _ : scala.concurrent.TimeoutException => dummyResponse
+ }
+ }
+}
\ No newline at end of file
diff --git a/src/main/scala/org/apache/sysml/api/ml/serving/ModelManager.scala b/src/main/scala/org/apache/sysml/api/ml/serving/ModelManager.scala
new file mode 100644
index 0000000..b67d367
--- /dev/null
+++ b/src/main/scala/org/apache/sysml/api/ml/serving/ModelManager.scala
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sysml.api.ml.serving
+
+import java.util
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.LongAdder
+
+import org.apache.commons.logging.{Log, LogFactory}
+import org.apache.sysml.api.jmlc.{Connection, PreparedScript}
+import org.apache.sysml.runtime.matrix.data.MatrixBlock
+import org.apache.sysml.utils.PersistentLRUCache
+
+trait ModelManager {
+ val LOG: Log = LogFactory.getLog(classOf[ModelManager].getName)
+ var modelLocality = new ConcurrentHashMap[String, util.ArrayList[JmlcExecutor]]()
+ val conn: Connection = new Connection()
+ val availableMemory = new LongAdder
+ var totalMemory = 0L
+ var cleanupEnabled = true
+ var memCheckEnabled = true
+ var models: Map[String, Model] = Map()
+
+ def setAvailableMemory(memBytes: Long) : Unit = {
+ LOG.info("Setting total memory to: " + memBytes + " bytes")
+ totalMemory = memBytes
+ availableMemory.reset()
+ availableMemory.add(memBytes)
+ }
+
+ def getAvailableMemory : Long = { availableMemory.longValue() }
+
+ def acquireMemory(bytes: Long) : Long = {
+ // if memory checking is not enabled just always say they get the memory
+ if (!memCheckEnabled || bytes == 0)
+ return bytes
+ LOG.debug("Requested: " + bytes)
+
+ // otherwise check to see if there is enough memory to meet the request
+ if (bytes <= availableMemory.longValue()) {
+ availableMemory.add(-1 * bytes)
+ LOG.debug("Granted: " + bytes + "/" + availableMemory.longValue())
+ return bytes
+ }
+ // not enough memory available :(
+
+ LOG.debug("Insufficient memory. Request was not granted")
+ -1
+ }
+
+ def releaseMemory(bytes: Long) : Unit = {
+ if (bytes > 0) {
+ LOG.debug("Releasing: " + bytes)
+ availableMemory.add(bytes)
+ LOG.debug("Available memory is now: " + availableMemory.longValue())
+ }
+ }
+
+ def setModelLocality(model: String, exec: JmlcExecutor) : Unit = {
+ this.synchronized({
+ modelLocality.putIfAbsent(model, new util.ArrayList[JmlcExecutor]())
+ modelLocality.get(model).add(exec)
+ })
+ }
+
+ def unsetModelLocality(model: String, exec: JmlcExecutor) : Unit = {
+ this.synchronized({ modelLocality.get(model).remove(exec) })
+ }
+
+ def getModelLocality(model: String) : util.ArrayList[JmlcExecutor] = { modelLocality.get(model) }
+
+ def isModelLocal(model: String, exec: JmlcExecutor) : Boolean = { getModelLocality(model).contains(exec) }
+
+ def disableCleanup() : Unit = { cleanupEnabled = false }
+
+ def disableMemcheck() : Unit = { memCheckEnabled = false }
+
+ def put(model: Model): Unit
+
+ def get(name: String): Model
+
+ def putWeight(name: String, weight: MatrixBlock) : Unit
+
+ def acquire(name: String, executor: JmlcExecutor) : PreparedScript
+
+ def release(name: String) : Unit
+}
+
+object ReferenceCountedModelManager extends ModelManager {
+ var modelRefCounts: Map[String,LongAdder] = Map()
+ var weightCache : PersistentLRUCache = _
+
+ override def setAvailableMemory(maxBytes: Long) : Unit = {
+ super.setAvailableMemory(maxBytes)
+ weightCache = new PersistentLRUCache((0.80*maxBytes).toLong)
+ weightCache.enableReadOnlyMode(true)
+ }
+
+ def tryAllocMem(name: String, batchSize: Int) : Long = {
+ // TODO: More sophisticated memory management
+ val extraMem = (0.5*models(name).weightMem).toLong
+ val weightMem = if (modelRefCounts(name).longValue() > 0) 0L else models(name).weightMem
+ val memReceived = acquireMemory(extraMem + weightMem)
+ if (memReceived < 0) memReceived else extraMem
+ }
+
+ def isCached(name: String) : Boolean = { modelRefCounts(name).longValue() > 0 }
+
+ def acquire(name: String, executor: JmlcExecutor) : PreparedScript = {
+ LOG.debug("Acquiring model: " + name + " Ref count: " + modelRefCounts(name).longValue())
+
+ val execName = if (executor.getExecType == "GPU") executor.getName else executor.getExecType
+ val ps = models(name).script(execName)
+ if (modelRefCounts(name).longValue() > 0 && ps.hasPinnedData) {
+ modelRefCounts(name).increment()
+ return ps.clone(false)
+ }
+
+ // otherwise we need to re-pin the weights, possibly reading them from disk
+ val model = models(name)
+ model.synchronized {
+ LOG.debug("Pinning weights for: " + name)
+ model.weightFiles.foreach(x => ps.setMatrix(x._1, weightCache.getAsMatrixBlock(x._2), true))
+ modelRefCounts(name).increment()
+ }
+ LOG.debug("Done acquiring model: " + name)
+ ps.clone(false)
+ }
+
+ override def disableCleanup(): Unit = {
+ super.disableCleanup()
+ LOG.debug("Cleanup is disabled")
+ }
+
+ def release(name: String) : Unit = {
+ modelRefCounts(name).decrement()
+ releaseMemory(models(name).weightMem)
+
+ LOG.debug("Releasing model: " + name + " Ref count: " + modelRefCounts(name).longValue())
+ if (modelRefCounts(name).longValue() == 0) {
+ models(name).script.synchronized {
+ if (modelRefCounts(name).longValue() == 0) {
+ models(name).script.foreach { x => x._2.clearPinnedData() }
+ }
+ }
+ }
+ }
+
+ def put(model: Model) : Unit = {
+ models += (model.name -> model)
+ modelRefCounts += (model.name -> new LongAdder())
+ }
+
+ def putWeight(name: String, weight: MatrixBlock) : Unit = {
+ weightCache.put(name, weight)
+ }
+
+ def get(name: String) : Model = { models(name) }
+
+}
\ No newline at end of file
diff --git a/src/main/scala/org/apache/sysml/api/ml/serving/NonBatchingScheduler.scala b/src/main/scala/org/apache/sysml/api/ml/serving/NonBatchingScheduler.scala
new file mode 100644
index 0000000..44ff26f
--- /dev/null
+++ b/src/main/scala/org/apache/sysml/api/ml/serving/NonBatchingScheduler.scala
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sysml.api.ml.serving
+
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.atomic.LongAdder
+
+import scala.concurrent.Future
+import scala.concurrent.duration.Duration
+
+object NonBatchingScheduler extends Scheduler {
+
+ override def start(numCores: Int, cpuMemoryBudgetInBytes: Long, gpus: String): Unit = {
+ LOG.info(s"Starting Non Batching Scheduler with: ${numCores} CPUs and ${gpus} GPUs")
+ super.start(numCores, cpuMemoryBudgetInBytes, gpus)
+ }
+
+ override def schedule(executor: JmlcExecutor): Array[SchedulingRequest] = {
+ var ret = Array[SchedulingRequest]()
+ dummyResponse.synchronized {
+ if (requestQueue.size() > 0) {
+ val request = requestQueue.poll()
+ ret :+= request
+ }
+ }
+ ret
+ }
+
+ var requestNum = new LongAdder
+ /**
+ * Enqueues a request for processing. The scheduler will read from these queues to determine which
+ * models to execute next
+ * @param request A PredictionRequest object containing the data for which a prediction is desired
+ * @param model The model object for which prediction is desired
+ * @return
+ */
+ override private[serving] def enqueue(request: PredictionRequest, model: Model): Future[PredictionResponse] = Future {
+ val statistics = if (_statistics) RequestStatistics() else null
+ val schedulingRequest = SchedulingRequest(
+ request, model, new CountDownLatch(1), System.nanoTime(), null, statistics)
+ if (_statistics) statistics.queueSize = requestQueue.size()
+ requestQueue.add(schedulingRequest)
+ counter += 1
+ try {
+ schedulingRequest.latch.await(timeout.length, timeout.unit)
+ schedulingRequest.response
+ } catch {
+ case e : scala.concurrent.TimeoutException => dummyResponse
+ }
+ }
+
+ override def onCompleteCallback(model: String, latency: Double, batchSize: Int, execType: String, execTime: Long): Unit = {}
+}
\ No newline at end of file
diff --git a/src/main/scala/org/apache/sysml/api/ml/serving/PredictionService.scala b/src/main/scala/org/apache/sysml/api/ml/serving/PredictionService.scala
new file mode 100644
index 0000000..f8c2345
--- /dev/null
+++ b/src/main/scala/org/apache/sysml/api/ml/serving/PredictionService.scala
@@ -0,0 +1,490 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sysml.api.ml.serving
+
+import java.io.File
+
+import akka.http.scaladsl.server.StandardRoute
+import akka.http.scaladsl.server.Directives._
+import akka.http.scaladsl.model.StatusCodes
+import akka.http.scaladsl.Http
+import akka.actor.ActorSystem
+import akka.stream.ActorMaterializer
+import org.apache.commons.cli.PosixParser
+import com.typesafe.config.ConfigFactory
+
+import scala.concurrent.duration._
+import java.util.HashMap
+
+import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
+import spray.json._
+import java.util.concurrent.atomic.LongAdder
+
+import scala.concurrent.{Await, Future}
+import scala.math.{max, pow}
+import org.apache.sysml.runtime.matrix.data.{MatrixBlock, OutputInfo}
+import org.apache.sysml.parser.DataExpression
+import org.apache.sysml.runtime.io.IOUtilFunctions
+import org.apache.sysml.api.jmlc.Connection
+import org.apache.sysml.api.jmlc.PreparedScript
+import org.apache.sysml.conf.ConfigurationManager
+import org.apache.sysml.runtime.instructions.gpu.context.GPUContextPool
+import org.apache.sysml.runtime.matrix.MatrixCharacteristics
+import org.apache.sysml.runtime.util.DataConverter
+import org.apache.commons.logging.Log
+import org.apache.commons.logging.LogFactory
+
+import scala.concurrent.ExecutionContext
+
+// format: can be file, binary, csv, ijv, jpeg, ...
+
+case class RequestStatistics(var batchSize: Int = -1,
+ var execTime: Long = -1,
+ var execType: String = "",
+ var requestDeserializationTime: Long = -1,
+ var responseSerializationTime: Long = -1,
+ var modelAcquireTime: Long = -1,
+ var modelReleaseTime: Long = -1,
+ var batchingTime: Long = -1,
+ var unbatchingTime: Long = -1,
+ var queueWaitTime: Long = -1,
+ var queueSize: Int = -1,
+ var execMode: Int = 0,
+ var preprocWaitTime: Long = -1)
+case class PredictionRequestExternal(name: String, data: Array[Double], rows: Int, cols: Int)
+case class PredictionResponseExternal(response: Array[Double], rows: Int, cols: Int, statistics: RequestStatistics)
+
+case class AddModelRequest(name: String, dml: String, inputVarName: String,
+ outputVarName: String, weightsDir: String,
+ latencyObjective: String, batchSize: Array[Int], memUse: Array[Long])
+
+case class Model(name: String,
+ script: Map[String,PreparedScript],
+ inputVarName: String,
+ outputVarName: String,
+ latencyObjective: Duration,
+ weightFiles: Map[String, String],
+ coeffs: (Double, Double),
+ weightMem: Long)
+case class PredictionRequest(data : MatrixBlock, modelName : String, requestSize : Int, receivedTime : Long)
+case class PredictionResponse(response: MatrixBlock, batchSize: Int, statistics: RequestStatistics)
+case class MatrixBlockContainer(numRows: Long, numCols: Long, nnz: Long, sum: Double, data: MatrixBlock)
+
+trait PredictionJsonProtocol extends SprayJsonSupport with DefaultJsonProtocol {
+ implicit val RequestStatisticsFormat = jsonFormat13(RequestStatistics)
+ implicit val predictionRequestExternalFormat = jsonFormat4(PredictionRequestExternal)
+ implicit val predictionResponseExternalFormat = jsonFormat4(PredictionResponseExternal)
+}
+
+trait AddModelJsonProtocol extends SprayJsonSupport with DefaultJsonProtocol {
+ implicit val AddModelRequetFormat = jsonFormat8(AddModelRequest)
+}
+
+class PredictionService {
+
+}
+
+/*
+Usage:
+1. Compiling a fat jar with maven assembly plugin in our standalone jar created lot of issues.
+Hence, for time being, we recommend downloading jar using the below script:
+SCALA_VERSION="2.11"
+AKKA_HTTP_VERSION="10.1.3"
+AKKA_VERSION="2.5.14"
+PREFIX="http://central.maven.org/maven2/com/typesafe/akka/"
+JARS=""
+for PKG in actor stream protobuf
+do
+ PKG_NAME="akka-"$PKG"_"$SCALA_VERSION
+ JAR_FILE=$PKG_NAME"-"$AKKA_VERSION".jar"
+ wget $PREFIX$PKG_NAME"/"$AKKA_VERSION"/"$JAR_FILE
+ JARS=$JARS$JAR_FILE":"
+done
+for PKG in http http-core parsing
+do
+ PKG_NAME="akka-"$PKG"_"$SCALA_VERSION
+ JAR_FILE=$PKG_NAME"-"$AKKA_HTTP_VERSION".jar"
+ wget $PREFIX$PKG_NAME"/"$AKKA_HTTP_VERSION"/"$JAR_FILE
+ JARS=$JARS$JAR_FILE":"
+done
+wget http://central.maven.org/maven2/com/typesafe/config/1.3.3/config-1.3.3.jar
+wget http://central.maven.org/maven2/com/typesafe/ssl-config-core_2.11/0.2.4/ssl-config-core_2.11-0.2.4.jar
+wget http://central.maven.org/maven2/org/reactivestreams/reactive-streams/1.0.2/reactive-streams-1.0.2.jar
+wget http://central.maven.org/maven2/org/scala-lang/scala-library/2.11.12/scala-library-2.11.12.jar
+wget http://central.maven.org/maven2/org/scala-lang/scala-parser-combinators/2.11.0-M4/scala-parser-combinators-2.11.0-M4.jar
+wget http://central.maven.org/maven2/commons-cli/commons-cli/1.4/commons-cli-1.4.jar
+wget http://central.maven.org/maven2/com/typesafe/akka/akka-http-spray-json-experimental_2.11/2.4.11.2/akka-http-spray-json-experimental_2.11-2.4.11.2.jar
+wget http://central.maven.org/maven2/io/spray/spray-json_2.11/1.3.2/spray-json_2.11-1.3.2.jar
+JARS=$JARS"config-1.3.3.jar:ssl-config-core_2.11-0.2.4.jar:reactive-streams-1.0.2.jar:commons-cli-1.4.jar:scala-parser-combinators-2.11.0-M4.jar:scala-library-2.11.12.jar:akka-http-spray-json-experimental_2.11-2.4.11.2.jar:spray-json_2.11-1.3.2.jar"
+echo "Include the following jars into the classpath: "$JARS
+
+
+2. Copy SystemML.jar and systemml-1.2.0-SNAPSHOT-extra.jar into the directory where akka jars are placed
+
+3. Start the server:
+java -cp $JARS org.apache.sysml.api.ml.serving.PredictionService -port 9000 -admin_password admin
+
+4. Check the health of the server:
+curl -u admin -XGET localhost:9000/health
+
+5. Perform prediction
+curl -XPOST -H "Content-Type:application/json" -d '{ "inputs":"1,2,3", "format":"csv", "model":"test", "num_input":1 }' localhost:9000/predict
+
+6. Shutdown the server:
+curl -u admin -XGET localhost:9000/shutdown
+
+ */
+
+object PredictionService extends PredictionJsonProtocol with AddModelJsonProtocol {
+ val __DEBUG__ = false
+
+ val LOG = LogFactory.getLog(classOf[PredictionService].getName)
+ val customConf = ConfigFactory.parseString("""
+ akka.http.server.idle-timeout=infinite
+ akka.http.client.idle-timeout=infinite
+ akka.http.host-connection-pool.idle-timeout=infinite
+ akka.http.host-connection-pool.client.idle-timeout=infinite
+ akka.http.server.max-connections=100000
+ """)
+ val basicConf = ConfigFactory.load()
+ val combined = customConf.withFallback(basicConf)
+ implicit val system = ActorSystem("systemml-prediction-service", ConfigFactory.load(combined))
+ implicit val materializer = ActorMaterializer()
+ implicit val executionContext = ExecutionContext.global
+ implicit val timeout = akka.util.Timeout(300.seconds)
+ val userPassword = new HashMap[String, String]()
+ var bindingFuture: Future[Http.ServerBinding] = null
+ var scheduler: Scheduler = null
+ val conn = new Connection()
+ var existantMatrixBlocks = Array[MatrixBlockContainer]()
+
+ def getCommandLineOptions(): org.apache.commons.cli.Options = {
+ val hostOption = new org.apache.commons.cli.Option("ip", true, "IP address")
+ val portOption = new org.apache.commons.cli.Option("port", true, "Port number")
+ val numRequestOption = new org.apache.commons.cli.Option("max_requests", true, "Maximum number of requests")
+ val timeoutOption = new org.apache.commons.cli.Option("timeout", true, "Timeout in milliseconds")
+ val passwdOption = new org.apache.commons.cli.Option("admin_password", true, "Admin password. Default: admin")
+ val helpOption = new org.apache.commons.cli.Option("help", false, "Show usage message")
+ val maxSizeOption = new org.apache.commons.cli.Option("max_bytes", true, "Maximum size of request in bytes")
+ val statisticsOption = new org.apache.commons.cli.Option("statistics", true, "Gather statistics on request execution")
+ val numCpuOption = new org.apache.commons.cli.Option("num_cpus", true, "How many CPUs should be allocated to the prediction service. Default nproc-1")
+ val gpusOption = new org.apache.commons.cli.Option("gpus", true, "GPUs available to this process. Default: 0")
+ val schedulerOption = new org.apache.commons.cli.Option("scheduler", true, "Scheduler implementation to use. Default: locality-aware")
+
+ // Only port is required option
+ portOption.setRequired(true)
+
+ return new org.apache.commons.cli.Options()
+ .addOption(hostOption).addOption(portOption).addOption(numRequestOption)
+ .addOption(passwdOption).addOption(timeoutOption).addOption(helpOption)
+ .addOption(maxSizeOption).addOption(statisticsOption).addOption(numCpuOption)
+ .addOption(gpusOption).addOption(schedulerOption)
+ }
+
+ def main(args: Array[String]): Unit = {
+ // Parse commandline variables:
+ val options = getCommandLineOptions
+ val line = new PosixParser().parse(getCommandLineOptions, args)
+ if (line.hasOption("help")) {
+ new org.apache.commons.cli.HelpFormatter().printHelp("systemml-prediction-service", options)
+ return
+ }
+ userPassword.put("admin", line.getOptionValue("admin_password", "admin"))
+ val currNumRequests = new LongAdder
+ val maxNumRequests = if (line.hasOption("max_requests"))
+ line.getOptionValue("max_requests").toLong else Long.MaxValue
+ val timeout = if (line.hasOption("timeout"))
+ Duration(line.getOptionValue("timeout").toLong, MILLISECONDS) else 300.seconds
+ val sizeDirective = if (line.hasOption("max_bytes"))
+ withSizeLimit(line.getOptionValue("max_bytes").toLong) else withoutSizeLimit
+ val numCores = if (line.hasOption("num_cpus"))
+ line.getOptionValue("num_cpus").toInt else Runtime.getRuntime.availableProcessors() - 1
+ val gpus = if (line.hasOption("gpus")) line.getOptionValue("gpus") else null
+ val schedulerType = line.getOptionValue("scheduler", "locality-aware")
+
+ // Initialize statistics counters
+ val numTimeouts = new LongAdder
+ val numFailures = new LongAdder
+ val totalTime = new LongAdder
+ val numCompletedPredictions = new LongAdder
+
+ // For now the models need to be loaded every time. TODO: pass the local to serialized models via commandline
+ var models = Map[String, Model]()
+
+ // TODO: Set the scheduler using factory
+ scheduler = SchedulerFactory.getScheduler(schedulerType)
+ val maxMemory = Runtime.getRuntime.maxMemory() // total memory is just what the JVM has currently allocated
+
+ LOG.info("Total memory allocated to server: " + maxMemory)
+ scheduler.start(numCores, maxMemory, gpus)
+
+ // Define unsecured routes: /predict and /health
+ val unsecuredRoutes = {
+ path("predict") {
+ withoutRequestTimeout {
+ post {
+ validate(currNumRequests.longValue() < maxNumRequests, "The prediction server received too many requests. Ignoring the current request.") {
+ entity(as[PredictionRequestExternal]) { request =>
+ validate(models.contains(request.name), "The model is not available.") {
+ try {
+ currNumRequests.increment()
+ val start = System.nanoTime()
+ val processedRequest = processPredictionRequest(request)
+ val deserializationTime = System.nanoTime() - start
+
+ val response = Await.result(
+ scheduler.enqueue(processedRequest, models(request.name)), timeout)
+ totalTime.add(System.nanoTime() - start)
+
+ numCompletedPredictions.increment()
+ complete(StatusCodes.OK, processPredictionResponse(response, "NOT IMPLEMENTED", deserializationTime))
+ } catch {
+ case e: scala.concurrent.TimeoutException => {
+ numTimeouts.increment()
+ complete(StatusCodes.RequestTimeout, "Timeout occured")
+ }
+ case e: Exception => {
+ numFailures.increment()
+ e.printStackTrace()
+ val msg = "Exception occured while executing the prediction request:"
+ complete(StatusCodes.InternalServerError, msg + e.getMessage)
+ }
+ } finally {
+ currNumRequests.decrement()
+ }
+ }
+ }
+ }
+ }
+ }
+ } ~ path("health") {
+ get {
+ val stats = "Number of requests (total/completed/timeout/failures):" + currNumRequests.longValue() + "/" + numCompletedPredictions.longValue() + "/"
+ numTimeouts.longValue() + "/" + numFailures.longValue() + ".\n" +
+ "Average prediction time:" + ((totalTime.doubleValue() * 1e-6) / numCompletedPredictions.longValue()) + " ms.\n"
+ complete(StatusCodes.OK, stats)
+ }
+ }
+ }
+
+ // For administration: This can be later extended for supporting multiple users.
+ val securedRoutes = {
+ authenticateBasicAsync(realm = "secure site", userAuthenticate) {
+ user =>
+ path("shutdown") {
+ get {
+ shutdownService(user, scheduler)
+ }
+ } ~
+ path("register-model") {
+ withoutRequestTimeout {
+ post {
+ entity(as[AddModelRequest]) { request =>
+ validate(!models.contains(request.name), "The model is already loaded") {
+ try {
+ val weightsInfo = processWeights(request.weightsDir)
+ val inputs = weightsInfo._1.keys.toArray ++ Array[String](request.inputVarName)
+
+ // compile for executor types
+ val scriptCpu = conn.prepareScript(
+ request.dml, inputs, Array[String](request.outputVarName))
+ var scripts = Map("CPU" -> scriptCpu)
+
+ if (gpus != null) {
+ GPUContextPool.AVAILABLE_GPUS = gpus
+ for (ix <- 0 until GPUContextPool.getAvailableCount) {
+ LOG.info("Compiling script for GPU: " + ix)
+ scripts += (s"GPU${ix}" -> conn.prepareScript(
+ request.dml, inputs, Array[String](request.outputVarName),
+ true, true, ix))
+ }
+ }
+
+ // b = cov(x,y) / var(x)
+ // a = mean(y) - b*mean(x)
+ val n = max(request.batchSize.length, 1).toDouble
+ val x = request.batchSize
+ val y = request.memUse
+ val mux = x.sum / n
+ val muy = y.sum / n
+ val vx = (1 / n) * x.map(v => pow(v - mux, 2.0)).sum
+ val b = ((1 / n) * (x.map(v => v - mux) zip y.map(v => v - muy)
+ ).map(v => v._1 * v._2).sum) * (1 / vx)
+ val a = muy - b * mux
+
+ // now register the created model
+ val model = Model(request.name,
+ scripts,
+ request.inputVarName,
+ request.outputVarName,
+ Duration(request.latencyObjective),
+ weightsInfo._1, (a, b), weightsInfo._2)
+ models += (request.name -> model)
+ scheduler.addModel(model)
+ complete(StatusCodes.OK)
+ } catch {
+ case e: Exception => {
+ numFailures.increment()
+ e.printStackTrace()
+ complete(StatusCodes.InternalServerError,
+ "Exception occured while trying to add model:" + e.getMessage)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ bindingFuture = Http().bindAndHandle(
+ sizeDirective { // Both secured and unsecured routes need to respect the size restriction
+ unsecuredRoutes ~ securedRoutes
+ },
+ line.getOptionValue("ip", "localhost"), line.getOptionValue("port").toInt)
+
+ println(s"Prediction Server online.")
+ while (true) Thread.sleep(100)
+ bindingFuture
+ .flatMap(_.unbind())
+ .onComplete(_ ⇒ system.terminate())
+ }
+
+ def processPredictionResponse(response : PredictionResponse,
+ format : String,
+ deserializationTime: Long) : PredictionResponseExternal = {
+ if (response != null) {
+ val start = System.nanoTime()
+ val dataArray = response.response.getDenseBlockValues
+ val rows = response.response.getNumRows
+ val cols = response.response.getNumColumns
+ val serializationTime = System.nanoTime() - start
+ if (response.statistics != null) {
+ response.statistics.requestDeserializationTime = deserializationTime
+ response.statistics.responseSerializationTime = serializationTime
+ }
+ PredictionResponseExternal(dataArray, rows, cols, response.statistics)
+ } else {
+ PredictionResponseExternal(null, -1, -1, null)
+ }
+ }
+
+ def processWeights(dirname: String) : (Map[String, String], Long) = {
+ val dir = new File(dirname)
+ if (!(dir.exists && dir.isDirectory))
+ throw new Exception("Weight directory: " + dirname + " is invalid")
+
+ val weightsWithSize = dir.listFiles().filter(
+ x => !(x.isDirectory && (x.toString contains "binary"))).map(_.toString).filter(
+ x => (x.slice(x.length-3, x.length) != "mtd") &&
+ !(x contains "_bin.mtx")).
+ map(x => getNameFromPath(x) -> registerWeight(x, dirname)).toMap
+
+ val weightMap = weightsWithSize.map(x => x._1 -> x._2._1)
+ val totalSize = weightsWithSize.map(x => x._2._2).sum
+
+ (weightMap, totalSize)
+ }
+
+ def getNameFromPath(path: String) : String = {
+ path.split("/").last.split("\\.")(0)
+ }
+
+ def registerWeight(path: String, dir: String) : (String, Long) = {
+ val res = convertToBinaryIfNecessary(path, dir)
+ scheduler.modelManager.putWeight(res._2, res._1)
+ (res._2, res._1.getInMemorySize)
+ }
+
+ def convertToBinaryIfNecessary(path: String, dir: String) : (MatrixBlock, String) = {
+ var pathActual = path
+ LOG.info("Reading weight: " + path)
+ val data = conn.readMatrix(path)
+
+ if (!isBinaryFormat(path)) {
+ LOG.info("Converting weight to binary format")
+ data.getMatrixCharacteristics
+ val binPath = dir + "/binary/" + getNameFromPath(path) + ".mtx"
+ DataConverter.writeMatrixToHDFS(data, binPath,
+ OutputInfo.BinaryBlockOutputInfo,
+ new MatrixCharacteristics(data.getNumRows, data.getNumColumns, ConfigurationManager.getBlocksize,
+ ConfigurationManager.getBlocksize, data.getNonZeros))
+ pathActual = binPath
+ }
+ (data, pathActual)
+ }
+
+ def isBinaryFormat(path: String) : Boolean = {
+ val mtdName = DataExpression.getMTDFileName(path)
+ val mtd = new DataExpression().readMetadataFile(mtdName, false)
+ if (mtd.containsKey("format")) mtd.getString("format") == "binary" else false
+ }
+
+ def processPredictionRequest(request : PredictionRequestExternal) : PredictionRequest = {
+ val mat = new MatrixBlock(request.rows, request.cols, false)
+ mat.init(request.data, request.rows, request.cols)
+ PredictionRequest(mat, request.name, request.rows, System.nanoTime())
+ }
+
+ def processMatrixInput(data : String, rows : Int, cols : Int, format : String) : MatrixBlock = {
+ val result = format match {
+ case "csv" => processTextInput(data, rows, cols, DataExpression.FORMAT_TYPE_VALUE_CSV)
+ case _ => throw new Exception("Only CSV Input currently supported")
+ }
+ result
+ }
+
+ def processTextInput(data : String, rows : Int, cols : Int, format : String) : MatrixBlock = {
+ val is = IOUtilFunctions.toInputStream(data)
+ conn.convertToMatrix(is, rows, cols, format)
+ }
+
+ def userAuthenticate(credentials: akka.http.scaladsl.server.directives.Credentials): Future[Option[String]] = {
+ credentials match {
+ case p@akka.http.scaladsl.server.directives.Credentials.Provided(id) =>
+ Future {
+ if (userPassword.containsKey(id) && p.verify(userPassword.get(id))) Some(id)
+ else None
+ }
+ case _ => Future.successful(None)
+ }
+ }
+
+ def shutdownService(user: String, scheduler: Scheduler): StandardRoute = {
+ if (user.equals("admin")) {
+ try {
+ Http().shutdownAllConnectionPools() andThen { case _ => bindingFuture.flatMap(_.unbind()).onComplete(_ ⇒ system.terminate()) }
+ scheduler.shutdown()
+ complete(StatusCodes.OK, "Shutting down the server.")
+ } finally {
+ new Thread(new Runnable {
+ def run() {
+ Thread.sleep(100) // wait for 100ms to send reply and then kill the prediction JVM so that we don't wait scala.io.StdIn.readLine()
+ System.exit(0)
+ }
+ }).start();
+ }
+ }
+ else {
+ complete(StatusCodes.BadRequest, "Only admin can shutdown the service.")
+ }
+ }
+
+}
\ No newline at end of file
diff --git a/src/main/scala/org/apache/sysml/api/ml/serving/RLSEstimator.scala b/src/main/scala/org/apache/sysml/api/ml/serving/RLSEstimator.scala
new file mode 100644
index 0000000..03bc8cd
--- /dev/null
+++ b/src/main/scala/org/apache/sysml/api/ml/serving/RLSEstimator.scala
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sysml.api.ml.serving
+
+import java.util.concurrent.LinkedBlockingQueue
+
+import breeze.linalg._
+import breeze.numerics._
+import breeze.stats._
+
+class RLSEstimator {
+ val dataQueue = new LinkedBlockingQueue[(Double, Double)]()
+ val chunkSize = 2
+
+ var isInitialized = false
+ var isFinalized = false
+ var Q : DenseMatrix[Double] = _
+ var b : DenseMatrix[Double] = _
+ var n = 0
+ val lda = 0.98
+ val eps = 0.00000001
+ var sigma = -1.0
+
+ def enqueueExample(batchSize: Int, latency: Double) : Unit = {
+ if (!isFinalized) {
+ println("ENQUEUING => " + dataQueue.size())
+ dataQueue.add((batchSize.toDouble, latency))
+ if (dataQueue.size() >= chunkSize)
+ update()
+ }
+ }
+
+ def dequeueExamples() : (DenseMatrix[Double], DenseMatrix[Double]) = {
+ val X = DenseMatrix.zeros[Double](chunkSize,4)
+ val y = DenseMatrix.zeros[Double](chunkSize, 1)
+
+ for (ix <- 0 until chunkSize) {
+ val (x_ex, y_ex) = dataQueue.poll()
+ X(ix,::) := DenseVector[Double](1.0, x_ex, pow(x_ex,2), pow(x_ex,3)).t
+ y(ix,0) = y_ex
+ }
+ (X, y)
+ }
+
+ def update() : Unit = {
+ val s = pow(lda, n)
+ val R = dequeueExamples()
+ val X = R._1
+ val y = R._2
+ if (!isInitialized) {
+ Q = X.t * X
+ b = Q \ (X.t * y)
+ isInitialized = true
+ } else if (s >= eps) {
+ val Q_new = Q + (X.t * X)
+ val S = pow(lda, n) * DenseMatrix.eye[Double](chunkSize)
+ val K = inv(Q_new) * (X.t * S) // Kalman filter gain
+ val V = y - (X * b) // Innovations
+ b :+= K * V
+ Q = Q_new
+ } else {
+ isFinalized = true
+ dataQueue.clear()
+ }
+ sigma = variance(y - (X*b))
+ n += 1
+ }
+
+ def predict(batchSize: Int) : (Double,Double) = {
+ val x = DenseMatrix(1.0, batchSize, pow(batchSize,2), pow(batchSize,3)).reshape(1,4)
+ val y_hat = x*b
+ (max(y_hat(0,0), 0.0), sigma)
+ }
+
+}
\ No newline at end of file
diff --git a/src/main/scala/org/apache/sysml/api/ml/serving/Scheduler.scala b/src/main/scala/org/apache/sysml/api/ml/serving/Scheduler.scala
new file mode 100644
index 0000000..39e85c0
--- /dev/null
+++ b/src/main/scala/org/apache/sysml/api/ml/serving/Scheduler.scala
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sysml.api.ml.serving
+
+import scala.concurrent.Future
+import scala.concurrent.duration._
+import java.util.concurrent._
+import java.util.List
+
+import org.apache.sysml.runtime.instructions.gpu.context.GPUContextPool
+import org.apache.sysml.runtime.instructions.gpu.context.GPUContext
+import org.apache.commons.logging.Log
+import org.apache.commons.logging.LogFactory
+
+import scala.concurrent.ExecutionContext
+
+case class SchedulingRequest(request: PredictionRequest,
+ model: Model,
+ latch: CountDownLatch,
+ receivedTime: Long,
+ var response: PredictionResponse = null,
+ statistics: RequestStatistics = null,
+ var memUse: Long = 0)
+
+trait Scheduler {
+ val LOG: Log = LogFactory.getLog(classOf[Scheduler].getName)
+ var executorService: ExecutorService = _
+ protected var _statistics = true
+ implicit val ec = ExecutionContext.global
+ var executorTypes = Array[String]()
+ var modelManager = ReferenceCountedModelManager
+
+ def start(numCores: Int, cpuMemoryBudgetInBytes: Long, gpus: String): Unit = {
+ LOG.info(s"Starting Scheduler with ${numCores} CPUs and ${gpus} GPUs")
+ var numGpus = 0
+ var gCtxs: List[GPUContext] = null
+ if (gpus != null) {
+ GPUContextPool.AVAILABLE_GPUS = gpus
+ gCtxs = GPUContextPool.getAllGPUContexts
+ numGpus = gCtxs.size
+ }
+
+ executorService = Executors.newFixedThreadPool(numCores + numGpus)
+ modelManager.setAvailableMemory((cpuMemoryBudgetInBytes*0.80).toLong)
+
+ if (numCores > 0)
+ executorTypes :+= "CPU"
+ if (numGpus > 0)
+ executorTypes :+= "GPU"
+
+ LOG.debug("STARTING SCHEDULER WITH: " + numCores + " CPU => " + numGpus + " GPUS")
+ for (i <- 0 until numCores) {
+ val exec = new JmlcExecutor(this, "CPU", "CPU" + i, null)
+ executorQueues.put(exec, new BatchQueue("CPU", "CPU" + i))
+ executorService.submit(exec)
+ }
+ for (i <- 0 until numGpus) {
+ val exec = new JmlcExecutor(this, "GPU","GPU" + i, gCtxs.get(i))
+ executorQueues.put(exec, new BatchQueue("GPU", "GPU" + i))
+ executorService.submit(exec)
+ }
+ }
+
+ def shutdown(): Unit = {
+ executorService.shutdown()
+ }
+
+ def schedule(executor: JmlcExecutor): Array[SchedulingRequest]
+
+ /**
+ * Registers a model with this scheduler. This should be called before enqueueing requests
+ * @param model Model object to be registered
+ */
+ def addModel(model: Model): Unit = {
+ modelQueues.putIfAbsent(model.name, new LinkedBlockingDeque[SchedulingRequest]())
+ latencyObjectives.putIfAbsent(model.name, model.latencyObjective)
+ modelManager.put(model)
+ }
+
+ /**
+ * Sets a flag indicating if detailed statistics should be gathered which profile the time spent
+ * in various stages of the execution pipeline
+ * @param flag Boolean flag indicating whether statistics should be gathered
+ */
+ def setStatistics(flag: Boolean): Unit = { _statistics = flag }
+
+ def timeout: Duration = 300.seconds
+
+ /**
+ * Method which is used to update scheduler state of execution of a batch. If necessary
+ * objects implementing the Scheduler trait should override this method and implement any logic needed
+ * to post-process execution after a batch
+ *
+ * @param model String indicating the name of the model which was just executed
+ * @param latency A measure of latency for this batch
+ * @param batchSize The number of examples in the batch
+ * @param execType The device type on which the batch was executed
+ */
+ def onCompleteCallback(model: String, latency: Double, batchSize: Int, execType: String, execTime: Long) : Unit
+
+ val requestQueue = new LinkedBlockingDeque[SchedulingRequest]()
+ val globalSchedulingQueues = new ConcurrentHashMap[String, BatchQueue]()
+ var modelQueues = new ConcurrentHashMap[String, BlockingQueue[SchedulingRequest]]()
+ var executorQueues = new ConcurrentHashMap[JmlcExecutor, BatchQueue]()
+ val dummyResponse = PredictionResponse(null, -1, null)
+ val latencyObjectives = new ConcurrentHashMap[String, Duration]()
+ var counter = 0
+
+ /**
+ * Enqueues a request for processing. The scheduler will read from these queues to determine which
+ * models to execute next
+ * @param request A PredictionRequest object containing the data for which a prediction is desired
+ * @param model The model object for which prediction
+ * @return
+ */
+ private[serving] def enqueue(request: PredictionRequest, model: Model): Future[PredictionResponse]
+}
diff --git a/src/main/scala/org/apache/sysml/api/ml/serving/SchedulerFactory.scala b/src/main/scala/org/apache/sysml/api/ml/serving/SchedulerFactory.scala
new file mode 100644
index 0000000..6fa6007
--- /dev/null
+++ b/src/main/scala/org/apache/sysml/api/ml/serving/SchedulerFactory.scala
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sysml.api.ml.serving
+
+object SchedulerFactory {
+ def getScheduler(schedulerType: String) : Scheduler = {
+ schedulerType match {
+ case "non-batching" => NonBatchingScheduler
+ case "basic-batching" => BasicBatchingScheduler
+ case "locality-aware" => LocalityAwareScheduler
+ }
+ }
+}
\ No newline at end of file