[freenet-cvs] r14600 - trunk/freenet/src/freenet/support

[toad at freenetproject.org]

Author: toad
Date: 2007-08-11 13:49:20 +0000 (Sat, 11 Aug 2007)
New Revision: 14600

Added:
   trunk/freenet/src/freenet/support/Executor.java
   trunk/freenet/src/freenet/support/PooledExecutor.java
Log:
Add Executor interface (clone of java.util.concurrent) and homegrown PooledExecutor impl (we create a thread when we need one, they die after 5 minutes of inactivity)

Added: trunk/freenet/src/freenet/support/Executor.java
===================================================================
--- trunk/freenet/src/freenet/support/Executor.java	                        (rev 0)
+++ trunk/freenet/src/freenet/support/Executor.java	2007-08-11 13:49:20 UTC (rev 14600)
@@ -0,0 +1,14 @@
+/**
+ * @author toad
+ * To the extent that this is copyrightable, it's part of Freenet and licensed 
+ * under GPL2 or later. However, it's a trivial interface taken from Sun JDK 1.5,
+ * and we will use that when we migrate to 1.5.
+ */
+package freenet.support;
+
+public interface Executor {
+	
+	/** Execute a job. */
+	public void execute(Runnable job);
+
+}

Added: trunk/freenet/src/freenet/support/PooledExecutor.java
===================================================================
--- trunk/freenet/src/freenet/support/PooledExecutor.java	                        (rev 0)
+++ trunk/freenet/src/freenet/support/PooledExecutor.java	2007-08-11 13:49:20 UTC (rev 14600)
@@ -0,0 +1,101 @@
+package freenet.support;
+
+import java.util.ArrayList;
+
+public class PooledExecutor implements Executor {
+
+	private final ArrayList runningThreads /* <MyThread> */ = new ArrayList();
+	private final ArrayList waitingThreads /* <MyThread> */ = new ArrayList();
+	long threadCounter = 0;
+	
+	/** Maximum time a thread will wait for a job */
+	static final int TIMEOUT = 5*60*1000;
+	
+	public void execute(Runnable job) {
+		while(true) {
+			MyThread t;
+			boolean mustStart = false;
+			synchronized(this) {
+				if(!waitingThreads.isEmpty()) {
+					t = (MyThread) waitingThreads.remove(waitingThreads.size()-1);
+				} else {
+					t = new MyThread("Pooled thread "+(threadCounter++));
+					t.setDaemon(true);
+					mustStart = true;
+				}
+			}
+			synchronized(t) {
+				if(!t.alive) continue;
+				if(t.nextJob != null) continue;
+				t.nextJob = job;
+				if(!mustStart)
+					t.notify();
+			}
+			if(mustStart) {
+				t.start();
+				synchronized(this) {
+					runningThreads.add(t);
+				}
+			}
+			return;
+		}
+	}
+
+	class MyThread extends Thread {
+		
+		boolean alive = true;
+		Runnable nextJob;
+		
+		public MyThread(String string) {
+			super(string);
+		}
+
+		public void run() {
+			while(true) {
+				Runnable job;
+				
+				synchronized(this) {
+					job = nextJob;
+					nextJob = null;
+				}
+				
+				if(job == null) {
+					synchronized(PooledExecutor.this) {
+						waitingThreads.add(this);
+					}
+					synchronized(this) {
+						if(nextJob == null) {
+							try {
+								wait(TIMEOUT);
+							} catch (InterruptedException e) {
+								// Ignore
+							}
+						}
+						job = nextJob;
+						nextJob = null;
+						if(job == null) {
+							alive = false;
+							// execute() won't give us another one if alive = false
+						}
+					}
+					synchronized(PooledExecutor.this) {
+						waitingThreads.remove(this);
+						if(!alive) {
+							runningThreads.remove(this);
+							return;
+						}
+					}
+				}
+				
+				// Run the job
+				try {
+					job.run();
+				} catch (Throwable t) {
+					Logger.error(this, "Caught "+t+" running job "+job, t);
+				}
+			}
+		}
+		
+	}
+	
+}