Added classes to manage thread-safe queues and to execute operations at

regular intervals on a separate thread. git-svn-id: https://samskivert.googlecode.com/svn/trunk@65 6335cc39-0255-0410-8fd6-9bcaacd3b74c
2001-03-02 00:47:10 +00:00
parent 7e723c6ddb
commit 4fc7d1dbe7
4 changed files with 663 additions and 3 deletions
@@ -0,0 +1,22 @@
+//
+// $Id: Interval.java,v 1.1 2001/03/02 00:47:10 mdb Exp $
+
+package com.samskivert.util;
+
+import java.util.*;
+
+/**
+ * An interface for doing operations after some delay. Normally,
+ * <code>intervalExpired()</code> should not do anything that will take
+ * very long. If you want to use the thread to do some serious stuff, look
+ * at <code>incrementHelperThreads</code> and
+ * <code>setMaxHelperThreads</code> in <code>IntervalManager</code>.
+ *
+ * @see go2net.util.IntervalManager
+ * @see go2net.util.IntervalManager#incrementHelperThreads
+ * @see go2net.util.IntervalManager#setMaxHelperThreads
+ */
+public interface Interval
+{
+    public void intervalExpired (int id, Object arg);
+}
@@ -0,0 +1,345 @@
+//
+// $Id: IntervalManager.java,v 1.1 2001/03/02 00:47:10 mdb Exp $
+
+package com.samskivert.util;
+
+import java.util.*;
+import com.samskivert.Log;
+
+/**
+ * Interval: can be used to register an object that is to be called once
+ * or repeatedly after an interval has expired.
+ *
+ * <p> A caveat: be careful about deadlocks kids. The number of helper
+ * threads should exceed the number of times you nest calls to Interval
+ * stuff from inside the intervalExpired method of an Interval.  Normally,
+ * if all you're using this for is interrupting other threads or calling
+ * repaint you can get by with 0 helpers. If you want to do some more
+ * complicated stuff in intervalExpired() then you probably want a few
+ * threads in the pool. If you're doing extensive processing and setting
+ * up other timeouts on that thread then you want to make sure you've got
+ * plenty of threads.
+ */
+public class IntervalManager extends Thread
+{
+    /**
+     * Sets the maximum number of helper threads to run methods in
+     * Intervals.  Defaults to 0, meaning that the main thread does all
+     * the work.  Setting this to a nonzero value makes it such that a
+     * pool of helper threads is created to do all the actual work.
+     */
+    public static synchronized void setMaxHelperThreads (int newmax)
+    {
+	newmax = Math.max(newmax, 0);
+
+	// if we are moving down, we need to kill some threads.
+	if (_helpers > newmax) {
+	    for (int ii=0; ii < _helpers - newmax; ii++) {
+		_queue.addItem(KILLHELPER);
+	    }
+	    _helpers = newmax;
+	}
+
+	// finally, set the new maximum thread pool size.
+	_maxhelpers = newmax;
+    }
+
+    /**
+     * Increment the number of maximum helper threads.  This is useful if
+     * you may have many packages which use the IntervalMgr as a
+     * threadpool to do work, each package can statically initialize the
+     * number of threads they want to be available by calling this method
+     * and they'll all be added together.
+     */
+    public static synchronized void incrementHelperThreads (int amt)
+    {
+	setMaxHelperThreads(_maxhelpers + amt);
+    }
+
+    /**
+     * Schedule the intervaled object to get called after an interval.
+     *
+     * @param i the intervaled object
+     * @param timout # of ms until interval is up.
+     * @param arg object to be passed when interval expires
+     * @param recur if true, interval gets called every timeout until
+     *              removed.
+     *
+     * @return an ID number that will passed to the intervalExpired method
+     * of i along with arg
+     */
+    public static int register (Interval i, long delay, Object arg,
+				boolean recur)
+    {
+	synchronized (_mgr) {
+	    IntervalItem item = new IntervalItem(i, delay, arg, recur);
+	    _hash.put(item.id, item);
+	    _schedule.add(item);
+	    Collections.sort(_schedule);
+	    if (item.endtime < _nextwake) {
+		_mgr.notify();
+	    }
+	    return item.id;
+	}
+    }
+
+    /**
+     * Non-recurring intervals are removed automatically after they are
+     * run!  This method is only useful if you want to remove a recurring
+     * Interval or if you want to remove a non-recurring Interval before
+     * it gets run.
+     */
+    public static void remove (int id)
+    {
+	synchronized (_mgr) {
+	    IntervalItem item = (IntervalItem) _hash.remove(id);
+	    if (item != null) {
+		_schedule.remove(item);
+		return;
+	    }
+	}
+
+	Log.warning("remove() called on non-registered " +
+		    "interval [id=" + id + "].");
+	Thread.dumpStack();
+    }
+
+    /**
+     * Do our interval thing.
+     */
+    public void run ()
+    {
+	while (_mgr == Thread.currentThread()) {
+
+	    // check to see if an interval has expired, if so call
+	    // expired from an unsynchronized position.
+//  	    TrackedThread.setState("Checking intervals");
+	    IntervalItem item = checkInterval();
+	    if (item != null) {
+		if (_maxhelpers > 0) {
+		    helperHandle(item);
+		} else {
+		    item.expired();  // we have no helpers, do it ourselves.
+		}
+	    }
+
+	    // now attempt to sleep
+//  	    TrackedThread.setState("Waiting for Interval event...");
+	    doWait();
+	}
+    }
+
+    /**
+     * Have a helper thread handle the expiration of this interval.
+     */
+    protected static synchronized void helperHandle (IntervalItem item)
+    {
+	// put the item on the queue...
+	_busyhelpers++;
+	_queue.addItem(item);
+
+	// possibly create a new thread in the pool to do this work.
+	if ((_busyhelpers > _helpers) && (_helpers < _maxhelpers)) {
+	    IntervalExpirer helper = new IntervalExpirer(_queue);
+	    _helpers++;
+	    helper.start();
+	}
+    }
+
+    /**
+     * A helper thread lets us know when it has finished its work.
+     */
+    protected static synchronized void helperFinished ()
+    {
+	_busyhelpers--;
+    }
+
+    /**
+     * Check to see if anything needs calling.
+     */
+    private synchronized IntervalItem checkInterval ()
+    {
+//  	TrackedThread.setState("Checking intervals");
+	if (_schedule.size() > 0) {
+	    IntervalItem item = (IntervalItem) _schedule.get(0);
+
+	    // it's totally valid for us to wake up early..
+	    // so make sure we really want to run the first item on the queue.
+	    if (item.endtime <= System.currentTimeMillis()) {
+		// we gotta deal with this monster!
+		// first, remove it.
+		_schedule.remove(0);
+		if (item.checkRecur()) {
+		    // since we're definitionally not sleeping, we can just
+		    // insert into the queue without checking wait times..
+		    _schedule.add(item);
+		    Collections.sort(_schedule);
+
+		} else {
+		    // otherwise, get rid of this interval altogether
+		    _hash.remove(item.id);
+		}
+
+		return item;
+	    }
+	}
+
+	return null;
+    }
+
+    /**
+     * Sleep until the next Interval needs to be attended to.
+     */
+    private synchronized void doWait ()
+    {
+//  	TrackedThread.setState("Waiting for Interval event...");
+	if (_schedule.size() == 0) {
+	    _nextwake = Long.MAX_VALUE;
+	} else {
+	    IntervalItem item = (IntervalItem) _schedule.get(0);
+	    _nextwake = item.endtime;
+	}
+
+	long waittime = _nextwake - System.currentTimeMillis();
+	if (waittime > 0L) {
+	    try {
+		wait(waittime);
+	    } catch (InterruptedException e) {
+	    }
+	}
+    }
+
+    private IntervalManager ()
+    {
+	super("IntervalManager");
+	setDaemon(true);
+	start();
+    }
+
+    // there can be only one!
+    protected static IntervalManager _mgr = new IntervalManager();
+
+    // We just use a sorted list for now since we aren't likely to have
+    // too many monitorables at any time. Insertions are O(log n),
+    // removals are O(n) since we search then entire queue to find the
+    // intervaleds to remove.
+    // 
+    // If we someday find that we have a lot of intervaleds, we may want
+    // to rewrite this such that we can add and remove intervaleds much
+    // faster.
+    protected static ArrayList _schedule = new ArrayList();
+    protected static IntMap _hash = new IntMap();
+    protected static long _nextwake = Long.MAX_VALUE;
+
+    protected static int _helpers = 0;       // # of created helpers
+    protected static int _maxhelpers = 0;    // max # of helpers we can create
+    protected static int _busyhelpers = 0;   // # of outstanding requests
+    protected static Queue _queue = new Queue();
+
+    protected static final Object KILLHELPER = new Object();
+}
+
+class IntervalItem implements Comparable
+{
+    public long endtime;
+    public int id;
+
+    protected long _timeout;
+    protected boolean _recur;
+    protected Interval _i;
+    protected Object _arg;
+
+    public IntervalItem (Interval i, long timeout, Object arg, boolean recur)
+    {
+	_i = i;
+	_timeout = Math.max(timeout, 0);
+	_arg = arg;
+	_recur = recur;
+
+	id = nextID();
+	endtime = System.currentTimeMillis() + timeout;
+    }
+
+    public int compareTo (Object other)
+    {
+	return (int) (endtime - ((IntervalItem) other).endtime);
+    }
+
+    /**
+     * Test-n-set recurring stuff. If we do recur, increment the time we
+     * are to next wake.
+     */
+    public boolean checkRecur ()
+    {
+	if (_recur) {
+	    endtime += _timeout;
+	}
+	return _recur;
+    }
+
+    /**
+     * Run the interval. It's synchronized so that if we someday have
+     * multiple threads, the same interval won't be called multiple times.
+     */
+    public synchronized void expired ()
+    {
+	// protect our ass.
+	try {
+	    _i.intervalExpired(id, _arg);
+	} catch (Exception e) {
+	    Log.warning("Exception while expiring interval: " + e);
+	    Log.logStackTrace(e);
+	}
+    }
+
+    /**
+     * For debugging.
+     */
+    public String toString()
+    {
+	return "Interval: " + _i + " (wakes in " +
+		(endtime - System.currentTimeMillis()) + "ms)" +
+		(_recur ? (" [recurring every " + _timeout + "ms]") : "");
+    }
+
+    /**
+     * Unique ids for each interval item.
+     */
+    private static synchronized int nextID ()
+    {
+	return _idseq++;
+    }
+
+    private static int _idseq = 0;
+}
+
+/**
+ * These are the helper threads for the IntervalManager.
+ */
+class IntervalExpirer extends Thread
+{
+    public IntervalExpirer (Queue queue)
+    {
+	super("IntervalExpirer");
+	setDaemon(true);
+	_queue = queue;
+    }
+
+    public void run ()
+    {
+	while (true) {
+//  	    TrackedThread.setState("Waiting for Interval to run...");
+	    Object o = _queue.getItem();
+	    if (o == IntervalManager.KILLHELPER) {
+		break; //exit
+	    }
+
+	    // otherwise run the bashtard!
+	    ((IntervalItem) o).expired();
+
+	    IntervalManager.helperFinished();
+	}
+    }
+
+    protected Queue _queue;
+}
@@ -1,17 +1,20 @@
 #
-# $Id: Makefile,v 1.7 2001/03/01 22:44:42 mdb Exp $
+# $Id: Makefile,v 1.8 2001/03/02 00:47:10 mdb Exp $

 ROOT = ../../..

 SRCS = \
-	Crypt.java \
 	ConfigUtil.java \
+	Crypt.java \
 	DefaultLogProvider.java \
-	IntMap.java \
 	IntIntMap.java \
+	IntMap.java \
+	Interval.java \
+	IntervalManager.java \
 	Log.java \
 	LogProvider.java \
 	PropertiesUtil.java \
+	Queue.java \
 	StringUtil.java \

 include $(ROOT)/build/Makefile.java
@@ -0,0 +1,290 @@
+//
+// $Id: Queue.java,v 1.1 2001/03/02 00:47:10 mdb Exp $
+
+package com.samskivert.util;
+
+/**
+ * A queue implementation that is more efficient than a wrapper around
+ * java.util.Vector. Allows adding and removing elements to/from the
+ * beginning, without the unneccessary System.arraycopy overhead of
+ * java.util.Vector.
+ */
+public class Queue
+{
+    public Queue (int suggestedSize)
+    {
+        _size = _suggestedSize = suggestedSize;
+        _items = new Object[_size];
+    }
+
+    public Queue ()
+    {
+        this(4);
+    }
+
+    public synchronized void clear ()
+    {
+        _count = _start = _end = 0;
+        _size = _suggestedSize;
+        _items = new Object[_size];
+    }
+
+    public synchronized boolean hasElements ()
+    {
+        return (_count != 0);
+    }
+
+    public synchronized int size ()
+    {
+        return _count;
+    }
+
+    public void appendItem (Object item)
+    {
+        addItem(item);
+    }
+
+    public synchronized void prependItem (Object item)
+    {
+        if (_count == _size) makeMoreRoom();
+
+        if (_start == 0) {
+            _start = _size - 1;
+        } else {
+            _start--;
+        }
+
+        _items[_start] = item;
+        _count++;
+
+        if (_count == 1) notify();
+    }
+
+    public synchronized void addItem (Object item)
+    {
+        if (_count == _size) makeMoreRoom();
+
+        _items[_end] = item;
+        _end = (_end + 1) % _size;
+        _count++;
+
+        if (_count == 1) notify();
+    }
+
+    /**
+     * Adds an item to the queue without notifying anyone. Useful for
+     * adding a bunch of items and then waking up the listener.
+     */
+    public synchronized void addItemSilent (Object item)
+    {
+        if (_count == _size) makeMoreRoom();
+        _items[_end] = item;
+        _end = (_end + 1) % _size;
+        _count++;
+    }
+
+    /**
+     * Adds an item to the queue and notify any listeners regardless of
+     * how many items are on the queue. Use this for the last item you add
+     * to a queue in a batch via addItemSilent() because the regular
+     * addItem() will think it doesn't need to notify anyone because the
+     * queue size isn't zero prior to this add.
+     */
+    public synchronized void addItemLoud (Object item)
+    {
+        if (_count == _size) makeMoreRoom();
+        _items[_end] = item;
+        _end = (_end + 1) % _size;
+        _count++;
+        notify();
+    }
+
+    public synchronized Object getItemNonBlocking ()
+    {
+        if (_count == 0) return null;
+
+        // pull the object off, and clear our reference to it
+        Object retval = _items[_start];
+        _items[_start] = null;
+
+        _start = (_start + 1) % _size;
+        _count--;
+
+        return retval;
+    }
+
+    public synchronized void waitForItem ()
+    {
+        while (_count == 0) {
+            try { wait(); } catch (InterruptedException e) {}
+        }
+    }
+
+    public synchronized Object getItem (long maxwait)
+    {
+	if (_count == 0) {
+	    try { wait(maxwait); } catch (InterruptedException e) {}
+
+	    // if count's still null when we pull out, we waited
+	    // ourmaxwait time.
+	    if (_count == 0) {
+		return null;
+	    }
+	}
+
+	return getItem();
+    }
+
+    public synchronized Object getItem ()
+    {
+        while (_count == 0) {
+            try { wait(); } catch (InterruptedException e) {}
+        }
+
+        // pull the object off, and clear our reference to it
+        Object retval = _items[_start];
+        _items[_start] = null;
+
+        _start = (_start + 1) % _size;
+        _count--;
+
+        // if we are only filling 1/8th of the space, shrink by half
+        if ((_size > MIN_SHRINK_SIZE) && (_size > _suggestedSize) &&
+            (_count < (_size >> 3))) shrink();
+
+        return retval;
+    }
+
+    private void makeMoreRoom ()
+    {
+        Object[] items = new Object[_size * 2];
+        System.arraycopy(_items, _start, items, 0, _size - _start);
+        System.arraycopy(_items, 0, items, _size - _start, _end);
+        _start = 0;
+        _end = _size;
+        _size *= 2;
+        _items = items;
+    }
+
+    // shrink by half
+    private void shrink ()
+    {
+        Object[] items = new Object[_size / 2];
+
+        if (_start > _end) {
+            // the data wraps around
+            System.arraycopy(_items, _start, items, 0, _size - _start);
+            System.arraycopy(_items, 0, items, _size - _start, _end + 1);
+
+        } else {
+            // the data does not wrap around
+            System.arraycopy(_items, _start, items, 0, _end - _start+1);
+        }
+
+        _size = _size / 2;
+        _start = 0;
+        _end = _count;
+        _items = items;
+    }
+
+    public String toString ()
+    {
+        StringBuffer buf = new StringBuffer();
+
+        buf.append("[count=").append(_count);
+        buf.append(", size=").append(_size);
+        buf.append(", start=").append(_start);
+        buf.append(", end=").append(_end);
+        buf.append(", elements={");
+
+        for (int i = 0; i < _count; i++) {
+            int pos = (i + _start) % _size;
+            if (i > 0) buf.append(", ");
+            buf.append(_items[pos]);
+        }
+
+        return buf.append("}]").toString();
+    }
+
+//     public static void main (String[] args)
+//     {
+//         Queue queue = new Queue();
+//         int value = 0;
+
+//         // add three items and dump the queue
+//         for (int i = 0; i < 3; i++) {
+//             queue.addItem(new Integer(value++));
+//         }
+//         System.out.println("Add three: " + queue);
+
+//         // now add three more items and cause it to expand and see how it
+//         // does
+//         for (int i = 0; i < 3; i++) {
+//             queue.addItem(new Integer(value++));
+//         }
+//         System.out.println("Add three more: " + queue);
+
+//         // now remove three items and move the queue pointer up a bit
+//         for (int i = 0; i < 3; i++) {
+//             queue.getItem();
+//         }
+//         System.out.println("Remove three: " + queue);
+
+//         // now add three again and cause the queue to wrap around
+//         for (int i = 0; i < 3; i++) {
+//             queue.addItem(new Integer(value++));
+//         }
+//         System.out.println("Add three more: " + queue);
+
+//         // now add three more and cause it to expand while wrapped
+//         for (int i = 0; i < 3; i++) {
+//             queue.addItem(new Integer(value++));
+//         }
+//         System.out.println("Add three more: " + queue);
+
+//         // now remove three elements and add 8 to cause it to wrap around
+//         // again
+//         for (int i = 0; i < 3; i++) {
+//             queue.getItem();
+//         }
+//         for (int i = 0; i < 8; i++) {
+//             queue.addItem(new Integer(value++));
+//         }
+//         System.out.println("-3 and +8: " + queue);
+
+//         // now add 2030 elements, cause it to expand a great deal
+//         for (int i = 0; i < 2030; i++) {
+//             queue.addItem(new Integer(value++));
+//         }
+//         // now remove some from the front so that we may wrap
+//         for (int i = 0; i < 8; i++) {
+//             queue.getItem();
+//         }
+//         // now add a few more to the end to cause us to wrap
+//         for (int i = 0; i < 8; i++) {
+//             queue.addItem(new Integer(value++));
+//         }
+//         System.out.println("+2030 -8 and +8: " + queue);
+
+//         // finally, remove 2030 and see where we end up after the shrink
+//         for (int i = 0; i < 2030; i++) {
+//             queue.getItem();
+//         }
+//         System.out.println("Remove 2030: " + queue);
+
+//         // now add and remove two elements to make sure the append and
+//         // remove pointers are in the right place
+//         for (int i = 0; i < 2; i++) {
+//             queue.addItem(new Integer(value++));
+//             queue.getItem();
+//         }
+//         System.out.println("Add two and remove two: " + queue);
+//     }
+
+    protected final static int MIN_SHRINK_SIZE = 1024;
+
+    protected Object[] _items;
+    protected int _count = 0;
+    protected int _start = 0, _end = 0;
+    protected int _suggestedSize, _size = 0;
+}