Merge branch 'v1.1' of git.openslx.org:openslx-ng/tutor-module into v1.1

2 files changed, 79 insertions, 4 deletions

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 {
 	 * </p>
 	 */
 	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<Runnable>(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<String, String> 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<Runnable> 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