// ----------------------------------------------------------------------
// This code is developed as part of the Java CoG Kit project
// The terms of the license can be found at http://www.cogkit.org/license
// This message may not be removed or altered.
// ----------------------------------------------------------------------

/*
 * Created on Jul 8, 2003
 */
package org.globus.cog.karajan.scheduler.submitQueue;

import org.apache.log4j.Logger;
import org.globus.cog.abstraction.impl.common.StatusImpl;
import org.globus.cog.abstraction.interfaces.Status;
import org.globus.cog.abstraction.interfaces.Task;
import org.globus.cog.abstraction.interfaces.TaskHandler;

import edu.emory.mathcs.backport.java.util.concurrent.ExecutorService;
import edu.emory.mathcs.backport.java.util.concurrent.Executors;

public class NonBlockingSubmit implements Runnable {
	private static final Logger logger = Logger.getLogger(NonBlockingSubmit.class);

    private static ExecutorService pool = Executors.newCachedThreadPool();
	private final TaskHandler taskHandler;
	private final Task task;
	private SubmitQueue[] queues;
	private int currentQueue;
	private int attempts;
    private int id;
    
    private static volatile int sid = 0;

	public NonBlockingSubmit(TaskHandler th, Task task, SubmitQueue[] queues) {
		this.taskHandler = th;
		this.task = task;
		this.queues = queues;
        id = sid++;
		attempts = 0;
	}

	public void go() {
		nextQueue();
	}

	public void nextQueue() {
		if (queues != null && queues.length > currentQueue) {
			queues[currentQueue++].queue(this);
		}
		else {
            pool.submit(this);
		}
	}

	public void notifyPreviousQueue(Exception ex) {
		if (queues != null && currentQueue > 0) {
			try {
				queues[--currentQueue].submitCompleted(this, ex);
			}
			catch (Exception e) {
				logger.warn("Exception caught while notifying queue of job submission status", e);
			}
		}
		else {
			if (ex != null) {
				Status st = task.getStatus();
				if (st.getStatusCode() == Status.FAILED) {
					logger.warn("Warning: Task handler throws exception and also sets status");
				}
				else {
				    Status ns = new StatusImpl();
				    ns.setStatusCode(Status.FAILED);
				    ns.setException(ex);
					task.setStatus(ns);
				}
			}
		}
	}

	public void run() {
		try {
			attempts++;
			taskHandler.submit(task);
			notifyPreviousQueue(null);
		}
		catch (Exception e) {
			notifyPreviousQueue(e);
		}
		catch (ThreadDeath td) {
			throw td;
		}
		catch (Throwable t) {
			notifyPreviousQueue(new Exception(t));
			t.printStackTrace();
		}
	}

	public int getAttempts() {
		return attempts;
	}

	public Task getTask() {
		return task;
	}

	public TaskHandler getTaskHandler() {
		return taskHandler;
	}
	
	
}

@@ -801,16 +801,85 @@
 ecutors;
+%0Aimport edu.emory.mathcs.backport.java.util.concurrent.ThreadFactory;
 %0A%0Apublic
@@ -1007,20 +1007,17 @@
 lass);%0A%0A
-    
+%09
 private 
@@ -1076,19 +1076,81 @@
 eadPool(
+new DaemonThreadFactory(%0A%09%09%09Executors.defaultThreadFactory())
 );%0A
+%0A
 %09private
@@ -1288,20 +1288,17 @@
 tempts;%0A
-    
+%09
 private 
@@ -1309,17 +1309,10 @@
 id;%0A
-    %0A    
+%0A%09
 priv
@@ -1491,24 +1491,18 @@
 queues;%0A
-        
+%09%09
 id = sid
@@ -1703,20 +1703,11 @@
 e %7B%0A
-            
+%09%09%09
 pool
@@ -2224,20 +2224,17 @@
 e %7B%0A%09%09%09%09
-    
+%09
 Status n
@@ -2255,28 +2255,25 @@
 Impl();%0A%09%09%09%09
-    
+%09
 ns.setStatus
@@ -2301,12 +2301,9 @@
 %09%09%09%09
-    
+%09
 ns.s
@@ -2830,10 +2830,312 @@
 %0A%09%7D%0A
-%09
 %0A%09
+static class DaemonThreadFactory implements ThreadFactory %7B%0A%09%09private ThreadFactory delegate;%0A%0A%09%09public DaemonThreadFactory(ThreadFactory delegate) %7B%0A%09%09%09this.delegate = delegate;%0A%09%09%7D%0A%0A%09%09public Thread newThread(Runnable r) %7B%0A%09%09%09Thread t = delegate.newThread(r);%0A%09%09%09t.setDaemon(true);%0A%09%09%09return t;%0A%09%09%7D%0A%0A%09%7D
 %0A%7D%0A

thread pool threads should be daemon threads