JepPythonEngine implements ApexPythonEngine { 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; } } JepPythonEngine(String threadGroupName,int numWorkerThreads); @Override void preInitInterpreter(Map<PythonInterpreterConfig, Object> preInitConfigs); @Override void startInterpreter(); @Override void postStartInterpreter(); @Override Map<String,PythonRequestResponse<Void>> runCommands(WorkerExecutionMode executionMode,long windowId,
      long requestId, PythonInterpreterRequest<Void> request); @Override Map<String,PythonRequestResponse<T>> executeMethodCall(WorkerExecutionMode executionMode,long windowId,
      long requestId, PythonInterpreterRequest<T> req); @Override Map<String,PythonRequestResponse<Void>> executeScript(WorkerExecutionMode executionMode,long windowId,
      long requestId, PythonInterpreterRequest<Void> req); @Override Map<String,PythonRequestResponse<T>> eval(WorkerExecutionMode executionMode,long windowId, long requestId,
      PythonInterpreterRequest<T> request); @Override void stopInterpreter(); int getNumWorkerThreads(); void setNumWorkerThreads(int numWorkerThreads); List<InterpreterWrapper> getWorkers(); void setWorkers(List<InterpreterWrapper> workers); @Override List<PythonRequestResponse> getCommandHistory(); @Override void setCommandHistory(List<PythonRequestResponse> commandHistory); long getSleepTimeAfterInterpreterStart(); void setSleepTimeAfterInterpreterStart(long sleepTimeAfterInterpreterStart); @Override BlockingQueue<PythonRequestResponse> getDelayedResponseQueue(); @Override void setDelayedResponseQueue(BlockingQueue<PythonRequestResponse> delayedResponseQueue); SpinPolicy getCpuSpinPolicyForWaitingInBuffer(); void setCpuSpinPolicyForWaitingInBuffer(SpinPolicy cpuSpinPolicyForWaitingInBuffer); int getBufferCapacity(); void setBufferCapacity(int bufferCapacity); @Override long getNumStarvedReturns(); @Override void setNumStarvedReturns(long numStarvedReturns);  }
@JepPythonTestContext(jepPythonBasedTest = true) @Test public void testSelectWorkerForCurrentCall() throws Exception { Set<String> overloadedWorkers = new HashSet<>(); List<String> busyCommands = new ArrayList<>(); busyCommands.add("import time"); busyCommands.add("time.sleep(5)"); PythonInterpreterRequest<Void> busyCommandRequest = buildRequestObjectForVoidGenericCommand(busyCommands, 2,TimeUnit.MILLISECONDS); for ( int i = 0; i < jepPythonEngine.getNumWorkerThreads() - 1; i++) { InterpreterWrapper aWrapper = jepPythonEngine.getWorkers().get(i); aWrapper.runCommands(1L,i,busyCommandRequest); assertTrue(aWrapper.isCurrentlyBusy()); overloadedWorkers.add(aWrapper.getInterpreterId()); } InterpreterWrapper candidateWrapperForExecution = null; InterpreterWrapper validCandidateWrapperForExecution = null; int counterForNullWorkers = 0; int counterForValidWorkers = 0; for ( int i = 0; i < jepPythonEngine.getNumWorkerThreads(); i++) { candidateWrapperForExecution = jepPythonEngine.selectWorkerForCurrentCall(i); if ( candidateWrapperForExecution == null) { counterForNullWorkers += 1; } else { counterForValidWorkers += 1; validCandidateWrapperForExecution = candidateWrapperForExecution; } } assertEquals(jepPythonEngine.getNumWorkerThreads(), counterForValidWorkers); assertEquals( 0, counterForNullWorkers); assertFalse(overloadedWorkers.contains(validCandidateWrapperForExecution.getInterpreterId())); Thread.sleep(5000); for ( int i = 0; i < jepPythonEngine.getNumWorkerThreads(); i++) { InterpreterWrapper aWrapper = jepPythonEngine.getWorkers().get(i); aWrapper.runCommands(1L,i,busyCommandRequest); } for (int i = 0; i < jepPythonEngine.getNumWorkerThreads(); i++) { candidateWrapperForExecution = jepPythonEngine.selectWorkerForCurrentCall(i); assertNull(candidateWrapperForExecution); } Thread.sleep(5000); }