content: /**
   * 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;
    }
  }
  