FlowPreparer { void setup(final ExecutableFlow flow) throws ExecutorManagerException { final ProjectFileHandler projectFileHandler = null; File tempDir = null; try { final ProjectDirectoryMetadata project = new ProjectDirectoryMetadata( flow.getProjectId(), flow.getVersion()); final long flowPrepStartTime = System.currentTimeMillis(); tempDir = downloadProjectIfNotExists(project, flow.getExecutionId()); LOGGER.info("Project is setup for execution {}", flow.getExecutionId()); long criticalSectionStartTime = -1; File execDir = null; synchronized (this) { LOGGER.info("Setting up execution dir for {}", flow.getExecutionId()); criticalSectionStartTime = System.currentTimeMillis(); if (!project.getInstalledDir().exists() && tempDir != null) { if (this.projectCacheCleaner.isPresent()) { this.projectCacheCleaner.get() .deleteProjectDirsIfNecessary(project.getDirSizeInByte()); } Files.move(tempDir.toPath(), project.getInstalledDir().toPath()); } final long start = System.currentTimeMillis(); execDir = setupExecutionDir(project.getInstalledDir(), flow); final long end = System.currentTimeMillis(); LOGGER.info("Setting up execution dir {} took {} sec(s)", execDir, (end - start) / 1000); } final long flowPrepCompletionTime = System.currentTimeMillis(); LOGGER.info("Flow preparation completed in {} sec(s), out ot which {} sec(s) was spent inside " + "critical section. [execid: {}, path: {}]", (flowPrepCompletionTime - flowPrepStartTime) / 1000, (flowPrepCompletionTime - criticalSectionStartTime) / 1000, flow.getExecutionId(), execDir.getPath()); } catch (final Exception ex) { FileIOUtils.deleteDirectorySilently(tempDir); LOGGER.error("Error in preparing flow execution {}", flow.getExecutionId(), ex); throw new ExecutorManagerException(ex); } finally { if (projectFileHandler != null) { projectFileHandler.deleteLocalFile(); } } } FlowPreparer(final ProjectStorageManager projectStorageManager, final DependencyTransferManager dependencyTransferManager,
      final File projectsDir, final ProjectCacheCleaner cleaner, final ProjectCacheHitRatio projectCacheHitRatio,
      final File executionsDir); void shutdown();  }
@Test public void testSetupFlowByMultipleThreads() { final int threadNum = 4; final ExecutableFlow[] executableFlows = new ExecutableFlow[]{ mockExecutableFlow(1, FAT_PROJECT_ID, 34), mockExecutableFlow(2, FAT_PROJECT_ID, 34), mockExecutableFlow(3, FAT_PROJECT_ID, 34), mockExecutableFlow(4, FAT_PROJECT_ID, 34) }; final ExecutorService service = Executors.newFixedThreadPool(threadNum); final List<Future> futures = new ArrayList<>(); for (int i = 0; i < threadNum; i++) { final int finalI = i; futures.add(service.submit(() -> { assertThatCode(() -> this.instance.setup(executableFlows[finalI]) ).doesNotThrowAnyException(); })); } for (final Future future : futures) { assertThatCode(() -> future.get()).doesNotThrowAnyException(); } service.shutdownNow(); for (final ExecutableFlow flow : executableFlows) { final File execDir = new File(this.executionsDir, String.valueOf(flow.getExecutionId())); assertTrue(execDir.exists()); assertTrue(new File(execDir, SAMPLE_FLOW_01).exists()); } }
@Test public void testSetupFlow() throws ExecutorManagerException { final ExecutableFlow executableFlow = mock(ExecutableFlow.class); when(executableFlow.getExecutionId()).thenReturn(12345); when(executableFlow.getProjectId()).thenReturn(FAT_PROJECT_ID); when(executableFlow.getVersion()).thenReturn(34); this.instance.setup(executableFlow); final File execDir = new File(this.executionsDir, "12345"); assertTrue(execDir.exists()); assertTrue(new File(execDir, SAMPLE_FLOW_01).exists()); }