| /** |
| * 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.apex.malhar.python.base.jep; |
| |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.BlockingQueue; |
| import java.util.concurrent.TimeUnit; |
| |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import org.apache.apex.malhar.python.base.ApexPythonEngine; |
| import org.apache.apex.malhar.python.base.ApexPythonInterpreterException; |
| import org.apache.apex.malhar.python.base.PythonInterpreterConfig; |
| import org.apache.apex.malhar.python.base.WorkerExecutionMode; |
| import org.apache.apex.malhar.python.base.requestresponse.PythonCommandType; |
| import org.apache.apex.malhar.python.base.requestresponse.PythonInterpreterRequest; |
| import org.apache.apex.malhar.python.base.requestresponse.PythonRequestResponse; |
| |
| import com.conversantmedia.util.concurrent.DisruptorBlockingQueue; |
| import com.conversantmedia.util.concurrent.SpinPolicy; |
| import com.google.common.primitives.Ints; |
| |
| import static com.google.common.base.Preconditions.checkNotNull; |
| |
| /*** |
| * <p>Implements the {@link ApexPythonEngine} interface by using the JEP ( Java Embedded Python ) engine. It is an |
| * in-memory interpreter and has the following characteristics: |
| * <ol> |
| * <li>The python engine allows for 4 major API patterns |
| * <ul> |
| * <li>Execute a method call by accepting parameters to pass to the interpreter</li> |
| * <li>Execute a python script as given in a file path</li> |
| * <li>Evaluate an expression and allows for passing of variables between the java code and the python |
| * in-memory interpreter bridge</li> |
| * <li>A handy method wherein a series of instructions can be passed in one single java call ( executed as a |
| * sequence of python eval instructions under the hood ) </li> |
| * </ul> |
| * <li>Automatic garbage collection of the variables that are passed from java code to the in memory python |
| * interpreter</li> |
| * <li>Support for all major python libraries. Tensorflow, Keras, Scikit, XGBoost</li> |
| * <li>The python engine uses the concept of a worker thread that is responsible for executing any of the 4 |
| * patterns mentioned above. The worker thread is implemented by {@link InterpreterThread}</li> |
| * <li>The implementation allows for SLA based execution model. i.e. the caller can stipulate that if the call is not |
| * complete within a time out, the engine code returns back null. See {@link InterpreterWrapper}</li> |
| * <li>Supports the concept of stragglers i.e. the processing of a request can complete eventually and the |
| * result available from a queue called as the delayed response queue</li> |
| * <li>Supports the concept of executing a call on all of the worker threads if required. This is to ensure the |
| * following use cases: |
| * <ul> |
| * <li>Since this is an interpreter, the users can make use of an earlier calls variable definition if |
| * need be. In such cases, the caller will have the need for a sticky thread i.e. all such calls need to |
| * end up on the same thread.</li> |
| * <li>Another reason is to implement the concept of Dynamic partitioning. Since interpreter accumulates |
| * state due to commands run on it, if a new partition is introduced at runtime, this can failures for all |
| * subsequent commands as they might depend on variables created in previous windows</li> |
| * </ul> |
| * </li> |
| * </li> |
| * </ol> |
| * </p> |
| * |
| * <p> Note that the engine implementation can be used independent of an Operator i.e. as a utility stack if need be. |
| * Some of the API signatures need a window ID and request ID but they do not necessarily mean that the API |
| * signatures are bound to an operator lifecycle. These parameters are used for efficient thread usage only and |
| * the API only needs a monotonically increasing number in true sense. |
| * </p> |
| * |
| * <p> |
| * JEP needs to be installed on all of the YARN nodes prior to the use of the JEP engine until docker support is |
| * available for Apex. Virtual environments are not supported yet. If multiple versions of python are present |
| * on the YARN nodes, ensure the JVM option java.library.path is pointing to the right version of JEP which in |
| * turn will ensure the right version of python to be used at runtime. |
| * </p> |
| * |
| * <p> |
| * JEPPythonEngine works on the concept of a worker pool. The engine maintains a configurable number of workers and |
| * each worker has a dedicated request and response queue. While this class is responsible for choosing the |
| * right worker from the pool of workers for a given request , the {@link InterpreterWrapper} class is responsible |
| * for maintaining the time bound SLAs. |
| * </p> |
| * |
| */ |
| public class JepPythonEngine implements ApexPythonEngine |
| { |
| private static final Logger LOG = LoggerFactory.getLogger(JepPythonEngine.class); |
| |
| /* Size of the worker pool */ |
| private int numWorkerThreads = 3; |
| |
| /* A name that can be used while logging messages and also used to set thread names */ |
| private String threadGroupName; |
| |
| private static final String JEP_LIBRARY_NAME = "jep"; |
| |
| private transient List<PythonRequestResponse> commandHistory = new ArrayList<>(); |
| |
| /* Spin policy for the disruptor queue implementation */ |
| private transient SpinPolicy cpuSpinPolicyForWaitingInBuffer = SpinPolicy.WAITING; |
| |
| // Represents the number of responses that can be held in the queue |
| private int bufferCapacity = 64; |
| |
| /* Time used to sleep in the beginning of the interpreter threads run i.e. start while initializing the interpreter. |
| Note that booting of the memory interpreter can be a really heavy process depending on the libraries that |
| are being loaded and hence this variable */ |
| private long sleepTimeAfterInterpreterStart = 3000; // 3 secs |
| |
| /** |
| * Represents the queue into which all the stragglers are drained into |
| */ |
| private transient BlockingQueue<PythonRequestResponse> delayedResponseQueue = |
| new DisruptorBlockingQueue<>(bufferCapacity,cpuSpinPolicyForWaitingInBuffer); |
| |
| private List<InterpreterWrapper> workers = new ArrayList<>(); |
| |
| private Map<PythonInterpreterConfig, Object> preInitConfigs; |
| |
| private long numStarvedReturns = 0; |
| |
| /*** |
| * Created the JEP Python engine instance but does not start the interpreters yet |
| * @param threadGroupName A name that represents all the workers in this thread |
| * @param numWorkerThreads Number of workers in the work pool |
| */ |
| public JepPythonEngine(String threadGroupName,int numWorkerThreads) |
| { |
| this.numWorkerThreads = numWorkerThreads; |
| this.threadGroupName = threadGroupName; |
| } |
| |
| private void initWorkers() throws ApexPythonInterpreterException |
| { |
| LOG.info("Attempting to load the JEP dynamic library"); |
| System.loadLibrary(JEP_LIBRARY_NAME); |
| LOG.info("Successfully loaded the JEP dynamic library in memory"); |
| SpinPolicy spinPolicyForReqQueue = SpinPolicy.WAITING; |
| if (preInitConfigs.containsKey(PythonInterpreterConfig.REQUEST_QUEUE_WAIT_SPIN_POLICY)) { |
| spinPolicyForReqQueue = (SpinPolicy)preInitConfigs.get(PythonInterpreterConfig.REQUEST_QUEUE_WAIT_SPIN_POLICY); |
| } |
| for ( int i = 0; i < numWorkerThreads; i++) { |
| InterpreterWrapper aWorker = new InterpreterWrapper(threadGroupName + "-" + i,delayedResponseQueue, |
| spinPolicyForReqQueue); |
| aWorker.preInitInterpreter(preInitConfigs); |
| aWorker.startInterpreter(); |
| workers.add(aWorker); |
| } |
| } |
| |
| /*** |
| * Used to select the right worker from the work pool. The goal is to round robin the workers as far as possible. |
| * Factors like busy workers can mean that the next available worker is chosen |
| * @param requestId Used to round robin the requests. Need not necessarily mean only an operator can use this engine. |
| * @return A worker from the worker pool. Null if all workers are busy. |
| */ |
| protected InterpreterWrapper selectWorkerForCurrentCall(long requestId) |
| { |
| int slotToLookFor = Ints.saturatedCast(requestId) % numWorkerThreads; |
| LOG.debug("Slot that is being looked for in the worker pool " + slotToLookFor); |
| boolean isWorkerFound = false; |
| int numWorkersScannedForAvailability = 0; |
| InterpreterWrapper aWorker = null; |
| while ( (!isWorkerFound) && (numWorkersScannedForAvailability < numWorkerThreads)) { |
| aWorker = workers.get(slotToLookFor); |
| numWorkersScannedForAvailability = numWorkersScannedForAvailability + 1; |
| if (!aWorker.isCurrentlyBusy()) { |
| isWorkerFound = true; |
| LOG.debug("Found worker with index as " + slotToLookFor); |
| break; |
| } else { |
| LOG.debug("Thread ID is currently busy " + aWorker.getInterpreterId()); |
| slotToLookFor = slotToLookFor + 1; |
| if ( slotToLookFor == numWorkerThreads) { |
| slotToLookFor = 0; |
| } |
| } |
| } |
| if (isWorkerFound) { |
| return aWorker; |
| } else { |
| numStarvedReturns += 1; |
| return null; |
| } |
| } |
| |
| /*** |
| * See {@link ApexPythonEngine#preInitInterpreter(Map)} for more details |
| * @param preInitConfigs The configuration that is going to be used by the interpreter.See constants |
| * defined in {@link PythonInterpreterConfig} for a list of keys available |
| * @throws ApexPythonInterpreterException if an issue while executing the pre interpreter logic |
| */ |
| @Override |
| public void preInitInterpreter(Map<PythonInterpreterConfig, Object> preInitConfigs) |
| throws ApexPythonInterpreterException |
| { |
| this.preInitConfigs = preInitConfigs; |
| } |
| |
| /*** |
| * Starts all of the worker threads. Also sleeps for a few moments to ensure "fat" frameworks like Tensorflow can |
| * be allowed to boot completely. |
| * @throws ApexPythonInterpreterException |
| */ |
| @Override |
| public void startInterpreter() throws ApexPythonInterpreterException |
| { |
| initWorkers(); |
| try { |
| if (sleepTimeAfterInterpreterStart > 0) { |
| LOG.debug("Sleeping to let the interpreter boot up in memory"); |
| Thread.sleep(sleepTimeAfterInterpreterStart); |
| } |
| } catch (InterruptedException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| /*** |
| * Used to execute all of the commands from the command history when an operator is instantiating a new instance of |
| * the engine. Used by the dynamic partitioner to let a newly provisioned operator to catch up to the state of all of |
| * the remaining operator instances |
| * @throws ApexPythonInterpreterException |
| */ |
| @Override |
| public void postStartInterpreter() throws ApexPythonInterpreterException |
| { |
| for ( InterpreterWrapper wrapper : workers) { |
| for (PythonRequestResponse requestResponse : commandHistory) { |
| PythonInterpreterRequest requestPayload = requestResponse.getPythonInterpreterRequest(); |
| try { |
| wrapper.processRequest(requestResponse,requestPayload); |
| } catch (InterruptedException e) { |
| throw new ApexPythonInterpreterException(e); |
| } |
| } |
| } |
| } |
| |
| /*** |
| * See {@link ApexPythonEngine#runCommands(WorkerExecutionMode, long, long, PythonInterpreterRequest)} for more |
| * details. Note that if the worker execution mode {@link WorkerExecutionMode} is BROADCAST, then the time SLA |
| * set is the total time for all workers i.e. each worker is given a ( total time / N ) where N is the current |
| * number of worker threads |
| * @param executionMode Whether these commands need to be run on all worker nodes or any of the worker node |
| * @param windowId used to select the worker from the worker pool.Can be any long if an operator is not using this. |
| * @param requestId used to select the worker from the worker pool. Can be any long if an operator is not using this. |
| * @param request Represents the request to be processed. |
| * @return A map containing the command as key and boolean representing success or failure as the value. |
| * @throws ApexPythonInterpreterException |
| */ |
| @Override |
| public Map<String,PythonRequestResponse<Void>> runCommands(WorkerExecutionMode executionMode,long windowId, |
| long requestId, PythonInterpreterRequest<Void> request) throws ApexPythonInterpreterException |
| { |
| checkNotNullConditions(request); |
| checkNotNull(request.getGenericCommandsRequestPayload(), "Run commands payload not set"); |
| checkNotNull(request.getGenericCommandsRequestPayload().getGenericCommands(), |
| "Commands that need to be run not set"); |
| Map<String,PythonRequestResponse<Void>> returnStatus = new HashMap<>(); |
| PythonRequestResponse lastSuccessfullySubmittedRequest = null; |
| try { |
| if (executionMode.equals(WorkerExecutionMode.BROADCAST)) { |
| LOG.debug("Executing run commands on all of the interpreter worker threads"); |
| long timeOutPerWorker = TimeUnit.NANOSECONDS.convert(request.getTimeout(),request.getTimeUnit()) / |
| numWorkerThreads; |
| LOG.debug("Allocating " + timeOutPerWorker + " nanoseconds for each of the worker threads"); |
| if ( timeOutPerWorker == 0) { |
| timeOutPerWorker = 1; |
| } |
| request.setTimeout(timeOutPerWorker); |
| request.setTimeUnit(TimeUnit.NANOSECONDS); |
| for ( InterpreterWrapper wrapper : workers) { |
| lastSuccessfullySubmittedRequest = wrapper.runCommands(windowId,requestId,request); |
| if (lastSuccessfullySubmittedRequest != null) { |
| returnStatus.put(wrapper.getInterpreterId(), lastSuccessfullySubmittedRequest); |
| } |
| } |
| if ( returnStatus.size() > 0) { |
| commandHistory.add(lastSuccessfullySubmittedRequest); |
| } |
| } else { |
| InterpreterWrapper currentThread = null; |
| if (executionMode.equals(WorkerExecutionMode.ANY)) { |
| LOG.debug("Executing run commands on a single interpreter worker thread"); |
| currentThread = selectWorkerForCurrentCall(requestId); |
| } |
| if (executionMode.equals(WorkerExecutionMode.STICKY)) { |
| currentThread = workers.get(request.hashCode() % numWorkerThreads); |
| LOG.debug(" Choosing sticky worker " + currentThread.getInterpreterId()); |
| } |
| if (currentThread != null) { |
| lastSuccessfullySubmittedRequest = currentThread.runCommands(windowId, requestId, request); |
| if (lastSuccessfullySubmittedRequest != null) { |
| returnStatus.put(currentThread.getInterpreterId(), lastSuccessfullySubmittedRequest); |
| } |
| } else { |
| throw new ApexPythonInterpreterException("No free interpreter threads available." + |
| " Consider increasing workers and relaunch"); |
| } |
| } |
| } catch (InterruptedException e) { |
| throw new ApexPythonInterpreterException(e); |
| } |
| return returnStatus; |
| } |
| |
| /*** |
| * See {@link ApexPythonEngine#executeMethodCall(WorkerExecutionMode, long, long, PythonInterpreterRequest)} for more |
| * details. Note that if the worker execution mode {@link WorkerExecutionMode} is BROADCAST, then the time SLA |
| * set is the total time for all workers i.e. each worker is given a ( total time / N ) where N is the current |
| * number of worker threads |
| * |
| * @param executionMode If the method call needs to be invoked on all workers or any single worker |
| * @param windowId used to select the worker from the worker pool.Can be any long if an operator is not using this. |
| * @param requestId used to select the worker from the worker pool. Can be any long if an operator is not using this. |
| * @param req Represents the request to be processed. |
| * @param <T> |
| * @return |
| * @throws ApexPythonInterpreterException |
| */ |
| @Override |
| public <T> Map<String,PythonRequestResponse<T>> executeMethodCall(WorkerExecutionMode executionMode,long windowId, |
| long requestId, PythonInterpreterRequest<T> req) throws ApexPythonInterpreterException |
| { |
| checkNotNullConditions(req); |
| checkNotNull(req.getMethodCallRequest(), "Method call info not set"); |
| checkNotNull(req.getMethodCallRequest().getNameOfMethod(), "Method name not set"); |
| Map<String,PythonRequestResponse<T>> returnStatus = new HashMap<>(); |
| req.setCommandType(PythonCommandType.METHOD_INVOCATION_COMMAND); |
| PythonRequestResponse lastSuccessfullySubmittedRequest = null; |
| try { |
| if (executionMode.equals(WorkerExecutionMode.BROADCAST)) { |
| long timeOutPerWorker = TimeUnit.NANOSECONDS.convert(req.getTimeout(), req.getTimeUnit()) / |
| numWorkerThreads; |
| if ( timeOutPerWorker == 0) { |
| timeOutPerWorker = 1; |
| } |
| req.setTimeout(timeOutPerWorker); |
| req.setTimeUnit(TimeUnit.NANOSECONDS); |
| for ( InterpreterWrapper wrapper : workers) { |
| lastSuccessfullySubmittedRequest = wrapper.executeMethodCall(windowId,requestId,req); |
| if ( lastSuccessfullySubmittedRequest != null) { |
| returnStatus.put(wrapper.getInterpreterId(), lastSuccessfullySubmittedRequest); |
| } |
| } |
| if ( returnStatus.size() > 0) { |
| commandHistory.add(lastSuccessfullySubmittedRequest); |
| } |
| } else { |
| InterpreterWrapper currentThread = null; |
| if (executionMode.equals(WorkerExecutionMode.ANY)) { |
| currentThread = selectWorkerForCurrentCall(requestId); |
| } |
| if (executionMode.equals(WorkerExecutionMode.STICKY)) { |
| currentThread = workers.get(req.hashCode() % numWorkerThreads); |
| LOG.debug(" Choosing sticky worker " + currentThread.getInterpreterId()); |
| } |
| if (currentThread != null) { |
| lastSuccessfullySubmittedRequest = currentThread.executeMethodCall(windowId, requestId, req); |
| if (lastSuccessfullySubmittedRequest != null) { |
| returnStatus.put(currentThread.getInterpreterId(), lastSuccessfullySubmittedRequest); |
| } else { |
| throw new ApexPythonInterpreterException("No free interpreter threads available." + |
| " Consider increasing workers and relaunch"); |
| } |
| } |
| } |
| } catch (InterruptedException e) { |
| throw new ApexPythonInterpreterException(e); |
| } |
| return returnStatus; |
| } |
| |
| /*** |
| * See {@link ApexPythonEngine#executeScript(WorkerExecutionMode, long, long, PythonInterpreterRequest)} for more |
| * details. Note that if the worker execution mode {@link WorkerExecutionMode} is BROADCAST, then the time SLA |
| * set is the total time for all workers i.e. each worker is given a ( total time / N ) where N is the current |
| * number of worker threads |
| * @param executionMode If the method call needs to be invoked on all workers or any single worker |
| * @param windowId used to select the worker from the worker pool.Can be any long if an operator is not using this. |
| * @param requestId used to select the worker from the worker pool. Can be any long if an operator is not using this. |
| * @param req Represents the request to be processed. |
| * @return |
| * @throws ApexPythonInterpreterException |
| */ |
| @Override |
| public Map<String,PythonRequestResponse<Void>> executeScript(WorkerExecutionMode executionMode,long windowId, |
| long requestId, PythonInterpreterRequest<Void> req) |
| throws ApexPythonInterpreterException |
| { |
| checkNotNullConditions(req); |
| checkNotNull(req.getScriptExecutionRequestPayload(), "Script execution info not set"); |
| checkNotNull(req.getScriptExecutionRequestPayload().getScriptName(), "Script name not set"); |
| Map<String,PythonRequestResponse<Void>> returnStatus = new HashMap<>(); |
| PythonRequestResponse lastSuccessfullySubmittedRequest = null; |
| try { |
| if (executionMode.equals(WorkerExecutionMode.BROADCAST)) { |
| long timeOutPerWorker = TimeUnit.NANOSECONDS.convert(req.getTimeout(), req.getTimeUnit()) / |
| numWorkerThreads; |
| if ( timeOutPerWorker == 0) { |
| timeOutPerWorker = 1; |
| } |
| req.setTimeout(timeOutPerWorker); |
| req.setTimeUnit(TimeUnit.NANOSECONDS); |
| for ( InterpreterWrapper wrapper : workers) { |
| lastSuccessfullySubmittedRequest = wrapper.executeScript(windowId,requestId,req); |
| if (lastSuccessfullySubmittedRequest != null) { |
| returnStatus.put(wrapper.getInterpreterId(),lastSuccessfullySubmittedRequest); |
| } |
| } |
| if ( returnStatus.size() > 0) { |
| commandHistory.add(lastSuccessfullySubmittedRequest); |
| } |
| } else { |
| InterpreterWrapper currentThread = null; |
| if (executionMode.equals(WorkerExecutionMode.ANY)) { |
| currentThread = selectWorkerForCurrentCall(requestId); |
| } |
| if (executionMode.equals(WorkerExecutionMode.STICKY)) { |
| currentThread = workers.get(req.hashCode() % numWorkerThreads); |
| LOG.debug(" Choosing sticky worker " + currentThread.getInterpreterId()); |
| } |
| if (currentThread != null) { |
| lastSuccessfullySubmittedRequest = currentThread.executeScript(windowId, requestId, req); |
| if (lastSuccessfullySubmittedRequest != null) { |
| returnStatus.put(currentThread.getInterpreterId(), lastSuccessfullySubmittedRequest); |
| } |
| } else { |
| throw new ApexPythonInterpreterException("No free interpreter threads available." + |
| " Consider increasing workers and relaunch"); |
| } |
| } |
| } catch (InterruptedException e) { |
| throw new ApexPythonInterpreterException(e); |
| } |
| return returnStatus; |
| } |
| |
| private void checkNotNullConditions(PythonInterpreterRequest request) |
| { |
| checkNotNull(request, "Request object cannnot be null"); |
| checkNotNull(request.getTimeout(), "Time out value not set"); |
| checkNotNull(request.getTimeUnit(), "Time out unit not set"); |
| } |
| |
| /*** |
| * See {@link ApexPythonEngine#eval(WorkerExecutionMode, long, long, PythonInterpreterRequest)} for more |
| * details. Note that if the worker execution mode {@link WorkerExecutionMode} is BROADCAST, then the time SLA |
| * set is the total time for all workers i.e. each worker is given a ( total time / N ) where N is the current |
| * number of worker threads |
| * @param executionMode If the method call needs to be invoked on all workers or any single worker |
| * @param windowId used to select the worker from the worker pool.Can be any long if an operator is not using this. |
| * @param requestId used to select the worker from the worker pool. Can be any long if an operator is not using this. |
| * @param request |
| * @param <T> |
| * @return |
| * @throws ApexPythonInterpreterException |
| */ |
| @Override |
| public <T> Map<String,PythonRequestResponse<T>> eval(WorkerExecutionMode executionMode,long windowId, long requestId, |
| PythonInterpreterRequest<T> request) throws ApexPythonInterpreterException |
| { |
| checkNotNullConditions(request); |
| checkNotNull(request.getEvalCommandRequestPayload(), "Eval command info not set"); |
| checkNotNull(request.getEvalCommandRequestPayload().getEvalCommand(),"Eval command not set"); |
| Map<String,PythonRequestResponse<T>> statusOfEval = new HashMap<>(); |
| PythonRequestResponse lastSuccessfullySubmittedRequest = null; |
| try { |
| if (executionMode.equals(WorkerExecutionMode.BROADCAST)) { |
| long timeOutPerWorker = TimeUnit.NANOSECONDS.convert(request.getTimeout(), request.getTimeUnit()) / |
| numWorkerThreads; |
| if ( timeOutPerWorker == 0) { |
| timeOutPerWorker = 1; |
| } |
| request.setTimeout(timeOutPerWorker); |
| request.setTimeUnit(TimeUnit.NANOSECONDS); |
| for ( InterpreterWrapper wrapper : workers) { |
| lastSuccessfullySubmittedRequest = wrapper.eval(windowId,requestId,request); |
| if (lastSuccessfullySubmittedRequest != null) { |
| statusOfEval.put(wrapper.getInterpreterId(), lastSuccessfullySubmittedRequest); |
| } |
| } |
| commandHistory.add(lastSuccessfullySubmittedRequest); |
| } else { |
| InterpreterWrapper currentThread = null; |
| if (executionMode.equals(WorkerExecutionMode.ANY)) { |
| currentThread = selectWorkerForCurrentCall(requestId); |
| } |
| if (executionMode.equals(WorkerExecutionMode.STICKY)) { |
| currentThread = workers.get(request.hashCode() % numWorkerThreads); |
| LOG.debug(" Choosing sticky worker " + currentThread.getInterpreterId()); |
| } |
| if (currentThread != null) { |
| lastSuccessfullySubmittedRequest = currentThread.eval(windowId, requestId, request); |
| if (lastSuccessfullySubmittedRequest != null) { |
| statusOfEval.put(currentThread.getInterpreterId(), lastSuccessfullySubmittedRequest); |
| } |
| } else { |
| throw new ApexPythonInterpreterException("No free interpreter threads available." + |
| " Consider increasing workers and relaunch"); |
| } |
| } |
| } catch (InterruptedException e) { |
| throw new ApexPythonInterpreterException(e); |
| } |
| return statusOfEval; |
| } |
| |
| @Override |
| public void stopInterpreter() throws ApexPythonInterpreterException |
| { |
| for ( InterpreterWrapper wrapper : workers) { |
| wrapper.stopInterpreter(); |
| } |
| } |
| |
| public int getNumWorkerThreads() |
| { |
| return numWorkerThreads; |
| } |
| |
| public void setNumWorkerThreads(int numWorkerThreads) |
| { |
| this.numWorkerThreads = numWorkerThreads; |
| } |
| |
| public List<InterpreterWrapper> getWorkers() |
| { |
| return workers; |
| } |
| |
| public void setWorkers(List<InterpreterWrapper> workers) |
| { |
| this.workers = workers; |
| } |
| |
| @Override |
| public List<PythonRequestResponse> getCommandHistory() |
| { |
| return commandHistory; |
| } |
| |
| @Override |
| public void setCommandHistory(List<PythonRequestResponse> commandHistory) |
| { |
| this.commandHistory = commandHistory; |
| } |
| |
| public long getSleepTimeAfterInterpreterStart() |
| { |
| return sleepTimeAfterInterpreterStart; |
| } |
| |
| public void setSleepTimeAfterInterpreterStart(long sleepTimeAfterInterpreterStart) |
| { |
| this.sleepTimeAfterInterpreterStart = sleepTimeAfterInterpreterStart; |
| } |
| |
| @Override |
| public BlockingQueue<PythonRequestResponse> getDelayedResponseQueue() |
| { |
| return delayedResponseQueue; |
| } |
| |
| @Override |
| public void setDelayedResponseQueue(BlockingQueue<PythonRequestResponse> delayedResponseQueue) |
| { |
| this.delayedResponseQueue = delayedResponseQueue; |
| } |
| |
| public SpinPolicy getCpuSpinPolicyForWaitingInBuffer() |
| { |
| return cpuSpinPolicyForWaitingInBuffer; |
| } |
| |
| public void setCpuSpinPolicyForWaitingInBuffer(SpinPolicy cpuSpinPolicyForWaitingInBuffer) |
| { |
| this.cpuSpinPolicyForWaitingInBuffer = cpuSpinPolicyForWaitingInBuffer; |
| } |
| |
| public int getBufferCapacity() |
| { |
| return bufferCapacity; |
| } |
| |
| public void setBufferCapacity(int bufferCapacity) |
| { |
| this.bufferCapacity = bufferCapacity; |
| } |
| |
| @Override |
| public long getNumStarvedReturns() |
| { |
| return numStarvedReturns; |
| } |
| |
| @Override |
| public void setNumStarvedReturns(long numStarvedReturns) |
| { |
| this.numStarvedReturns = numStarvedReturns; |
| } |
| } |
