@Test public void testMultithreadedExecWithDependencies() throws InterruptedException { Assume.assumeTrue(shouldRunMaven); int executionsNum = 5; final CountDownLatch latch = new CountDownLatch(executionsNum); final String [] dependencies = { new File(getClass().getClassLoader().getResource(".m2/repository/python/math2/mult/1.0/mult-1.0.zip").getFile()).getAbsolutePath(), new File(getClass().getClassLoader().getResource(".m2/repository/python/math2/sum/2.1/sum-2.1.zip").getFile()).getAbsolutePath(), new File(getClass().getClassLoader().getResource(".m2/repository/python/math3/mult/1.2/mult-1.2.zip").getFile()).getAbsolutePath(), new File(getClass().getClassLoader().getResource(".m2/repository/python/math3/sum/4.1/sum-4.1.zip").getFile()).getAbsolutePath() }; for(int i = 0; i < executionsNum; i++) { final String executioId = String.valueOf(i); final String varName = "VAR"; final String script = "import sys\nimport time\nimport math_fake.utils.print_text as print_text\ntime.sleep(3)\n" + varName + " = print_text.foo('" + executioId + "')\nprint " + varName + "\n"; final String dependency = dependencies[i % 4]; int count = dependency.contains("math2") ? 2 : 3; String sign = dependency.contains("sum-") ? "+" : "*"; final StringBuilder expectedResult = new StringBuilder(executioId); while (--count > 0) { expectedResult.append(sign).append(executioId); } new Thread() { public void run() { try { PythonExecutionResult result = pythonRuntimeService.exec(new HashSet<>(Collections.singletonList(dependency)), script, EMPTY_CALL_ARGUMENTS); assertNotNull(result); assertNotNull(result.getExecutionResult().get(varName)); assertEquals(expectedResult.toString(), result.getExecutionResult().get(varName).toString()); } finally { latch.countDown(); } } }.start(); } latch.await(); }
public PythonExecutionResult exec(String script, Map<String, Serializable> callArguments) { checkValidInterpreter(); initInterpreter(); prepareInterpreterContext(callArguments); Exception originException = null; for(int i = 0; i < RETRIES_NUMBER_ON_THREADED_ISSUE; i++) { try { return exec(script); } catch (Exception e) { if(!isThreadsRelatedModuleIssue(e)) { throw new RuntimeException("Error executing python script: " + e, e); } if(originException == null) { originException = e; } } } throw new RuntimeException("Error executing python script: " + originException, originException); }
PythonExecutor implements Executor { public PythonExecutionResult exec(String script, Map<String, Serializable> callArguments) { checkValidInterpreter(); initInterpreter(); prepareInterpreterContext(callArguments); Exception originException = null; for(int i = 0; i < RETRIES_NUMBER_ON_THREADED_ISSUE; i++) { try { return exec(script); } catch (Exception e) { if(!isThreadsRelatedModuleIssue(e)) { throw new RuntimeException("Error executing python script: " + e, e); } if(originException == null) { originException = e; } } } throw new RuntimeException("Error executing python script: " + originException, originException); } }
PythonExecutor implements Executor { public PythonExecutionResult exec(String script, Map<String, Serializable> callArguments) { checkValidInterpreter(); initInterpreter(); prepareInterpreterContext(callArguments); Exception originException = null; for(int i = 0; i < RETRIES_NUMBER_ON_THREADED_ISSUE; i++) { try { return exec(script); } catch (Exception e) { if(!isThreadsRelatedModuleIssue(e)) { throw new RuntimeException("Error executing python script: " + e, e); } if(originException == null) { originException = e; } } } throw new RuntimeException("Error executing python script: " + originException, originException); } PythonExecutor();  PythonExecutor(Set<String> dependencies); }
PythonExecutor implements Executor { public PythonExecutionResult exec(String script, Map<String, Serializable> callArguments) { checkValidInterpreter(); initInterpreter(); prepareInterpreterContext(callArguments); Exception originException = null; for(int i = 0; i < RETRIES_NUMBER_ON_THREADED_ISSUE; i++) { try { return exec(script); } catch (Exception e) { if(!isThreadsRelatedModuleIssue(e)) { throw new RuntimeException("Error executing python script: " + e, e); } if(originException == null) { originException = e; } } } throw new RuntimeException("Error executing python script: " + originException, originException); } PythonExecutor();  PythonExecutor(Set<String> dependencies); PythonExecutionResult exec(String script, Map<String, Serializable> callArguments); PythonEvaluationResult eval(String prepareEnvironmentScript, String expr, Map<String, Serializable> context); @Override void allocate(); @Override void release(); @Override void close(); boolean isClosed(); }
PythonExecutor implements Executor { public PythonExecutionResult exec(String script, Map<String, Serializable> callArguments) { checkValidInterpreter(); initInterpreter(); prepareInterpreterContext(callArguments); Exception originException = null; for(int i = 0; i < RETRIES_NUMBER_ON_THREADED_ISSUE; i++) { try { return exec(script); } catch (Exception e) { if(!isThreadsRelatedModuleIssue(e)) { throw new RuntimeException("Error executing python script: " + e, e); } if(originException == null) { originException = e; } } } throw new RuntimeException("Error executing python script: " + originException, originException); } PythonExecutor();  PythonExecutor(Set<String> dependencies); PythonExecutionResult exec(String script, Map<String, Serializable> callArguments); PythonEvaluationResult eval(String prepareEnvironmentScript, String expr, Map<String, Serializable> context); @Override void allocate(); @Override void release(); @Override void close(); boolean isClosed(); static final String THREADED_MODULES_ISSUE; static final int RETRIES_NUMBER_ON_THREADED_ISSUE; static final int MAX_LENGTH; }