From ab82089aed8329d23e1315a84984061a61ab0ce6 Mon Sep 17 00:00:00 2001 From: Simon Rettberg Date: Tue, 5 Jan 2016 09:35:04 +0100 Subject: [server] Allow httpd to handle more than two concurrent connections Had I read the javadocs properly I would have noticed that the thread pool only starts growing to its maximum size if the queue is full, not the other way round. So now we use a modified thread pool borrowed from SO that does exactly that. --- .../src/main/java/fi/iki/elonen/NanoHTTPD.java | 23 +++++++-- .../bwlp/sat/util/GrowingThreadPoolExecutor.java | 60 ++++++++++++++++++++++ 2 files changed, 79 insertions(+), 4 deletions(-) create mode 100644 dozentenmodulserver/src/main/java/org/openslx/bwlp/sat/util/GrowingThreadPoolExecutor.java diff --git a/dozentenmodulserver/src/main/java/fi/iki/elonen/NanoHTTPD.java b/dozentenmodulserver/src/main/java/fi/iki/elonen/NanoHTTPD.java index ff552e83..7b437414 100644 --- a/dozentenmodulserver/src/main/java/fi/iki/elonen/NanoHTTPD.java +++ b/dozentenmodulserver/src/main/java/fi/iki/elonen/NanoHTTPD.java @@ -63,13 +63,13 @@ import java.util.StringTokenizer; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import org.apache.commons.io.output.ByteArrayOutputStream; import org.apache.log4j.Logger; import org.joda.time.format.DateTimeFormat; import org.joda.time.format.DateTimeFormatter; +import org.openslx.bwlp.sat.util.GrowingThreadPoolExecutor; /** * A simple, tiny, nicely embeddable HTTP server in Java @@ -131,7 +131,7 @@ public abstract class NanoHTTPD implements Runnable { * This is required as the Keep-Alive HTTP connections would otherwise * block the socket reading thread forever (or as long the browser is open). */ - public static final int SOCKET_READ_TIMEOUT = 15000; + public static final int SOCKET_READ_TIMEOUT = 10000; /** * Common MIME type for dynamic content: plain text */ @@ -201,21 +201,26 @@ public abstract class NanoHTTPD implements Runnable { do { try { final Socket finalAccept = myServerSocket.accept(); + LOGGER.info("Accepted connection"); registerConnection(finalAccept); finalAccept.setSoTimeout(SOCKET_READ_TIMEOUT); final InputStream inputStream = finalAccept.getInputStream(); asyncRunner.exec(new Runnable() { @Override public void run() { + LOGGER.info("Starting worker"); OutputStream outputStream = null; try { outputStream = finalAccept.getOutputStream(); HTTPSession session = new HTTPSession(inputStream, outputStream, finalAccept.getInetAddress()); - while (!finalAccept.isClosed()) { + while (!finalAccept.isClosed() && !finalAccept.isInputShutdown()) { + LOGGER.info("Pre-execute"); session.execute(); + LOGGER.info("Post-execute"); } } catch (Exception e) { + LOGGER.info("Post-execute (ex)"); // When the socket is closed by the client, we throw our own SocketException // to break the "keep alive" loop above. if (!(e instanceof SocketTimeoutException) @@ -227,6 +232,7 @@ public abstract class NanoHTTPD implements Runnable { safeClose(inputStream); safeClose(finalAccept); unRegisterConnection(finalAccept); + LOGGER.info("Finished worker"); } } }); @@ -457,7 +463,7 @@ public abstract class NanoHTTPD implements Runnable { *

*/ public static class DefaultAsyncRunner implements AsyncRunner { - private ExecutorService pool = new ThreadPoolExecutor(2, 16, 1, TimeUnit.MINUTES, + private ExecutorService pool = new GrowingThreadPoolExecutor(2, 16, 1, TimeUnit.MINUTES, new ArrayBlockingQueue(16)); @Override @@ -624,6 +630,11 @@ public abstract class NanoHTTPD implements Runnable { if (!headerAlreadySent(header, "connection")) { sb.append("Connection: keep-alive\r\n"); } + if (!headerAlreadySent(header, "keep-alive")) { + sb.append("Keep-Alive: timeout="); + sb.append(SOCKET_READ_TIMEOUT / 1000 - 1); + sb.append("\r\n"); + } } private boolean headerAlreadySent(Map header, String name) { @@ -837,8 +848,11 @@ public abstract class NanoHTTPD implements Runnable { { int read = -1; try { + LOGGER.info("A " + inputStream.available()); read = inputStream.read(buf, 0, BUFSIZE); + LOGGER.info("B"); } catch (Exception e) { + LOGGER.info("C"); safeClose(inputStream); safeClose(outputStream); throw e; @@ -1009,6 +1023,7 @@ public abstract class NanoHTTPD implements Runnable { } String uri = st.nextToken(); + LOGGER.info("URI: " + uri); // Decode parameters from the URI int qmi = uri.indexOf('?'); diff --git a/dozentenmodulserver/src/main/java/org/openslx/bwlp/sat/util/GrowingThreadPoolExecutor.java b/dozentenmodulserver/src/main/java/org/openslx/bwlp/sat/util/GrowingThreadPoolExecutor.java new file mode 100644 index 00000000..becccf2f --- /dev/null +++ b/dozentenmodulserver/src/main/java/org/openslx/bwlp/sat/util/GrowingThreadPoolExecutor.java @@ -0,0 +1,60 @@ +package org.openslx.bwlp.sat.util; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +/** + * Grows to maximum pool size before queueing. See + * http://stackoverflow.com/a/20153234/2043481 + */ +public class GrowingThreadPoolExecutor extends ThreadPoolExecutor { + private int userSpecifiedCorePoolSize; + private int taskCount; + + public GrowingThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, + TimeUnit unit, BlockingQueue workQueue) { + super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue); + userSpecifiedCorePoolSize = corePoolSize; + } + + @Override + public void execute(Runnable runnable) { + synchronized (this) { + taskCount++; + setCorePoolSizeToTaskCountWithinBounds(); + } + super.execute(runnable); + } + + @Override + protected void afterExecute(Runnable runnable, Throwable throwable) { + super.afterExecute(runnable, throwable); + synchronized (this) { + taskCount--; + setCorePoolSizeToTaskCountWithinBounds(); + } + } + + private void setCorePoolSizeToTaskCountWithinBounds() { + int threads = taskCount; + if (threads < userSpecifiedCorePoolSize) + threads = userSpecifiedCorePoolSize; + if (threads > getMaximumPoolSize()) + threads = getMaximumPoolSize(); + super.setCorePoolSize(threads); + } + + public void setCorePoolSize(int corePoolSize) { + synchronized (this) { + userSpecifiedCorePoolSize = corePoolSize; + } + } + + @Override + public int getCorePoolSize() { + synchronized (this) { + return userSpecifiedCorePoolSize; + } + } +} \ No newline at end of file -- cgit v1.2.3-55-g7522