Re: What you get is what you want

> we could add 
> a blocking_enqueue method to QueueIF that will guarantee that the 
> enqueued event will be delivered, or at least, will not be dropped. 

Not only that, but that is the only viable back-pressure mecanism I know of.
Otherwise, you need infine amount of threads downstream to choke an upstream stage,
and you would also need infinite memory because of infinite queues.
That theory wasn't tought through...

I defined a sub interface of SinkIF, so that stages are not forced to have a blocking_enqueue
capable queue. I just lack a blocking_enqueueMany(QueueElementIF[] enqueueMe, int timeout_millis)
because... well I can't remember... something not trivial back then.


///////////////
package seda.sandStorm.api;
public interface BlockingSinkIF extends SinkIF {
	/**
	 * This method blocks on the queue up until a timeout occurs (throws SinkFullException) or
	 * until an element can be inserted. The queue perform its blocking operation through the
	 * <code>ThreadManagerIF.doBlockingAction()</code> for avoiding useless adaptation of thread
management.
	 *
	 * @param enqueueMe the event to enqueue
	 * @param timeout_millis if timeout_millis is <code>0</code>, this method
	 *    will be non-blocking and will behave like stadard enqueue.
	 *    If timeout_millis is <code>-1</code>, this method blocks forever until it can enqueue.
	 *    If timeout_millis is positive, this method will wait 
	 *    about that number of milliseconds before returning/throwing, but possibly a
	 *    little more.
	 */
	public void blocking_enqueue(QueueElementIF enqueueMe, int timeout_millis) throws
SinkFullException, SinkClosedException, SinkIOException, InterruptedException;
}
///////////////


> This mean that others ThreadManager must be created, because actuals TM 
> are not conceived with this in mind and will not perform efficiently.

The queue imlementation NEEDS to see the thread manager in order to ask it to perform blocking
actions (the tm would take note of the blocking thread and not adjust priority nor thread pool or
anything else, because the stage actually isn't that heavy loaded (given it's queue would fill
up))


///////////////
public interface ThreadManagerIF {
	public interface BlockingAction {
		Object perform() throws Exception;
	}

	public void setManager(ManagerIF mgr);
	public void init();
	public Object doBlockingAction(BlockingAction action) throws Exception;
	public SandstormThread createThread(StageWrapperIF sw, ThreadGroup tg, Runnable r, String name);
	public void register(StageWrapperIF stage);
	public void deregister(StageWrapperIF stage);
	public void shutdown();//renamed method from deregisterAll() to shutdown()
}
///////////////

Given the blocking action in the FiniteQueue:

	static class SemWaitBlockingAction implements ThreadManagerIF.BlockingAction {
		Sem sem;//Object lock;
		long timeout_millis;
		public Object perform() throws Exception {
			if (timeout_millis <= 0)
				sem.listen();//lock.wait();
			else
				sem.listen(timeout_millis);//lock.wait(timeout_millis);
			return null;
		}
	}


In my FiniteQueue, anytime it has to block (for enqueue only)
you would call threadManager.doBlockingAction(swba);

Note that you could use the blocking action for other purpose, like blocking I/O,
to report to the TM that the thread is blocked.

For my TM, it implements like this:

	public Object doBlockingAction(ThreadManagerIF.BlockingAction action) throws Exception {
		Thread t = Thread.currentThread();
		StageGroupRunner sgr = null;

		if (t instanceof SGRThread) {
			sgr = ((SGRThread)t).sgr;
			blocking(sgr); //mark
		}

		try {
			return action.perform();
		} finally {
			if(sgr!=null)
				unblocking(sgr); //unmark
		}
	}

Hope this hint helps. Because to me, all these are mandatory to have blocking_enqueue (beside the
timeout implementation in it, of course...).



		
__________________________________ 
Do you Yahoo!? 
Check out the new Yahoo! Front Page. 
www.yahoo.com 
 



-------------------------------------------------------
This SF.Net email is sponsored by:
Sybase ASE Linux Express Edition - download now for FREE
LinuxWorld Reader's Choice Award Winner for best database on Linux.
http://ads.osdn.com/?ad_id=5588&alloc_id=12065&op=click