intervalExpired() should not do anything that will take
+ * very long. If you want to use the thread to do some serious stuff, look
+ * at incrementHelperThreads and
+ * setMaxHelperThreads in IntervalManager.
+ *
+ * @see go2net.util.IntervalManager
+ * @see go2net.util.IntervalManager#incrementHelperThreads
+ * @see go2net.util.IntervalManager#setMaxHelperThreads
+ */
+public interface Interval
+{
+ public void intervalExpired (int id, Object arg);
+}
diff --git a/projects/samskivert/src/java/com/samskivert/util/IntervalManager.java b/projects/samskivert/src/java/com/samskivert/util/IntervalManager.java
new file mode 100644
index 00000000..f30852f1
--- /dev/null
+++ b/projects/samskivert/src/java/com/samskivert/util/IntervalManager.java
@@ -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.
+ *
+ * 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; +} diff --git a/projects/samskivert/src/java/com/samskivert/util/Makefile b/projects/samskivert/src/java/com/samskivert/util/Makefile index afbbc80e..637ff662 100644 --- a/projects/samskivert/src/java/com/samskivert/util/Makefile +++ b/projects/samskivert/src/java/com/samskivert/util/Makefile @@ -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 diff --git a/projects/samskivert/src/java/com/samskivert/util/Queue.java b/projects/samskivert/src/java/com/samskivert/util/Queue.java new file mode 100644 index 00000000..e6ad99ab --- /dev/null +++ b/projects/samskivert/src/java/com/samskivert/util/Queue.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; +}