diff --git a/src/main/java/fi/csc/chipster/filebroker/FileBroker.java b/src/main/java/fi/csc/chipster/filebroker/FileBroker.java index e6572521..167a075f 100644 --- a/src/main/java/fi/csc/chipster/filebroker/FileBroker.java +++ b/src/main/java/fi/csc/chipster/filebroker/FileBroker.java @@ -140,11 +140,26 @@ public static void main(String[] args) throws Exception { public void close() { RestUtils.shutdown("file-broker-admin", adminServer); + try { - httpServer.stop(); - authService.close(); + if (httpServer != null) { + httpServer.stop(); + } } catch (Exception e) { logger.warn("failed to stop the file-broker", e); } + + try { + if (authService != null) { + authService.close(); + } + } catch (Exception e) { + logger.warn("failed to stop the file-broker auth client", e); + } + + // after httpServer.stop(), because requests submit tasks to its executor + if (storageDiscovery != null) { + storageDiscovery.close(); + } } } diff --git a/src/main/java/fi/csc/chipster/filestorage/client/FileStorageDiscovery.java b/src/main/java/fi/csc/chipster/filestorage/client/FileStorageDiscovery.java index 82462673..73f3feae 100644 --- a/src/main/java/fi/csc/chipster/filestorage/client/FileStorageDiscovery.java +++ b/src/main/java/fi/csc/chipster/filestorage/client/FileStorageDiscovery.java @@ -280,4 +280,8 @@ public Map getStorages() { return storages; } } + + public void close() { + updateExecutor.shutdown(); + } } diff --git a/src/main/java/fi/csc/chipster/sessionworker/ZipSessionServlet.java b/src/main/java/fi/csc/chipster/sessionworker/ZipSessionServlet.java index f17a5a62..43237774 100644 --- a/src/main/java/fi/csc/chipster/sessionworker/ZipSessionServlet.java +++ b/src/main/java/fi/csc/chipster/sessionworker/ZipSessionServlet.java @@ -17,6 +17,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import java.util.zip.ZipEntry; @@ -112,6 +113,11 @@ public ZipSessionServlet(ServiceLocatorClient serviceLocator) { this.executor = Executors.newCachedThreadPool(); } + @Override + public void destroy() { + executor.shutdown(); + } + private void packageSession(HttpServletResponse response, StaticCredentials credentials, UUID sessionId) throws IOException { @@ -123,6 +129,9 @@ private void packageSession(HttpServletResponse response, StaticCredentials cred ArrayList entries = new ArrayList<>(); + CountDownLatch latch = new CountDownLatch(1); + PipedInputStream in = null; + try { Session session = sessionDb.getSession(sessionId); @@ -137,8 +146,6 @@ private void packageSession(HttpServletResponse response, StaticCredentials cred ArrayList errors = new ArrayList<>(); - CountDownLatch latch = new CountDownLatch(1); - OutputStream respoonseOutput = response.getOutputStream(); keepAliveWithSpaces(respoonseOutput, latch); @@ -181,20 +188,30 @@ private void packageSession(HttpServletResponse response, StaticCredentials cred } OutputStream output2 = new PipedOutputStream(); - PipedInputStream in = new PipedInputStream((PipedOutputStream) output2); + in = new PipedInputStream((PipedOutputStream) output2); if (errors.isEmpty()) { // start creating the zip stream in background thread (may complete before all // data is uploaded) - executor.submit(() -> { - try { - streamZip(entries, output2); - } catch (IOException e) { - logger.error("failed to package zip session", e); - errors.add("failed to package zip session: " + e.getMessage()); - } - }); + try { + executor.submit(() -> { + try { + streamZip(entries, output2); + } catch (IOException e) { + logger.error("failed to package zip session", e); + errors.add("failed to package zip session: " + e.getMessage()); + } + }); + } catch (RejectedExecutionException e) { + // destroy() has shut down the executor, i.e. this session-worker is stopping + logger.error("failed to start zip packaging, session-worker is stopping", e); + errors.add("failed to package zip session: session-worker is stopping"); + } + } + // skipped when the zip thread wasn't started, because then the upload would + // block forever waiting for the zip stream + if (errors.isEmpty()) { // upload in the zip stream in this thread, so that we send response to this // servlet request only after the upload has really completed try { @@ -227,12 +244,23 @@ private void packageSession(HttpServletResponse response, StaticCredentials cred logger.info("response: " + RestUtils.asJson(json, true)); + // stop the keep-alive before writing the json. This doesn't wait for a write + // that is already in progress, but prevents all the following ones. latch.countDown(); respoonseOutput.write(RestUtils.asJson(json).getBytes()); respoonseOutput.close(); } catch (RestException e) { throw ServletUtils.extractRestException(e); + } finally { + // stop the keep-alive thread also when something unexpected was thrown. + // Otherwise it would keep writing spaces to the client forever. + latch.countDown(); + + // If the upload failed or was skipped, the zip thread may still be writing to + // the pipe. Its writes would block forever, because the reader thread stays + // alive in the Jetty pool. Closing the read end makes them fail instead. + IOUtils.closeQuietly(in); } } diff --git a/src/test/java/fi/csc/chipster/auth/OidcResourceTest.java b/src/test/java/fi/csc/chipster/auth/OidcResourceTest.java index 9bf491d0..0a0e2fc5 100644 --- a/src/test/java/fi/csc/chipster/auth/OidcResourceTest.java +++ b/src/test/java/fi/csc/chipster/auth/OidcResourceTest.java @@ -117,6 +117,7 @@ public static void setUp() throws Exception { @AfterAll public static void tearDown() throws Exception { + hibernate.getSessionFactory().close(); launcher.stop(); }