Use a managed thread to kill pipes, avoiding creating too many threads when lots of pipes are expired

zhourenjian · zhourenjian · commit c53ee5127cc8 · 2012-02-15T14:49:27.000Z
diff --git a/sources/net.sf.j2s.ajax/ajaxpipe/net/sf/j2s/ajax/SimplePipeHelper.java b/sources/net.sf.j2s.ajax/ajaxpipe/net/sf/j2s/ajax/SimplePipeHelper.java
@@ -17,6 +17,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Vector;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
 
 import net.sf.j2s.ajax.SimpleSerializable;
 import net.sf.j2s.annotation.J2SIgnore;
@@ -50,6 +53,9 @@ public static interface IPipeClosing {
 	
 	static Map<String, SimplePipeRunnable> pipes;
 
+	@J2SIgnore
+	private static BlockingQueue<SimplePipeRunnable> toBeDestroyedPipes = new LinkedBlockingQueue<SimplePipeRunnable>();
+	
 	@J2SIgnore
 	private SimplePipeHelper() {
 		//
@@ -417,9 +423,41 @@ private static void monitoringAllPipes() {
 		}
 		System.err.println("Pipe sessions are null or empty! Pipe session monitor exited!");
 	}
-
+	
+	@J2SIgnore
+	private static void killingPipes() {
+		while (true) {
+			SimplePipeRunnable pipe = null;
+			try {
+				pipe = toBeDestroyedPipes.poll(30, TimeUnit.SECONDS);
+			} catch (InterruptedException e1) {
+				e1.printStackTrace();
+			}
+			if (pipe != null) {
+				//System.out.println("Killing pipe " + pipe + " :// " + System.currentTimeMillis());
+    			try {
+    				if (pipe.closer != null) {
+    					pipe.closer.helpClosing(pipe); // will call pipe.pipeDestroy()
+    				} else {
+    					pipe.pipeClosed(); // will call pipe.pipeDestroy()
+    				}
+    			} catch (Throwable e) {
+    				e.printStackTrace();
+    			}
+				//System.out.println("Killed pipe " + pipe + " :// " + System.currentTimeMillis());
+			} else {
+				//System.out.println("?? Empty loop?");
+			}
+		}
+	}
+	
 	@J2SIgnore
 	static void asyncDestroyPipe(final SimplePipeRunnable pipe) {
+		//System.out.println("To destroy pipe " + pipe);
+		if (!toBeDestroyedPipes.contains(pipe)) {
+			toBeDestroyedPipes.offer(pipe);
+		}
+		/*
     	(new Thread("Destroy Pipe Thread") {
     		@Override
     		public void run() {
@@ -434,6 +472,7 @@ public void run() {
     			}
     		}
     	}).start();
+    	// */
 	}
 	
 	@J2SIgnore
@@ -451,6 +490,13 @@ public void run() {
 		};
 		thread.setDaemon(true);
 		thread.start();
+		Thread killerThread = new Thread("Managed Pipe Session Killer") {
+			public void run() {
+				killingPipes();
+			}
+		};
+		killerThread.setDaemon(true);
+		killerThread.start();
 	}
 	
 }
