cms/util/Queue.java

//---PACKAGE DECLARATION---
package uk.ac.ukc.iscream.util;

//---IMPORTS---
import java.util.LinkedList;
import java.util.NoSuchElementException;
import java.util.Random;
import uk.ac.ukc.iscream.util.*;

/**
 * A Queue class designed to operate in a multi-threaded environment, with
 * added support for multiple "consumer" threads. Also offers blocking on
 * the get() methods, which ensures the consumer waits until the queue
 * actually contains some elements.
 *
 * @author  $Author: tdb1 $
 * @version $Id: Queue.java,v 1.13 2001/02/25 23:29:06 tdb1 Exp $
 */
public class Queue {

//---FINAL ATTRIBUTES---

    /**
     * The current CVS revision of this class
     */
    public static final String REVISION = "$Revision: 1.13 $";
    
    /**
     * Pass to constructor to remove a RANDOM item from
     * the Queue upon reaching the maximum limit.
     */
    public static final int RANDOM = 0;
    
    /**
     * Pass to constructor to remove the FIRST item from
     * the Queue upon reaching the maximum limit.
     */
    public static final int FIRST = 1;
    
    /**
     * Pass to constructor to remove the LAST item from
     * the Queue upon reaching the maximum limit.
     */
    public static final int LAST = 2;
    
    /**
     * Pass to constructor to drop the new item upon reaching
     * the maximum Queue limit.
     */
    public static final int DROP = 3;
    
//---STATIC METHODS---

//---CONSTRUCTORS---  
    
    /**
     * Constructs a new Queue with a maximum size limit on
     * any individual queue. This should be used to stop
     * conditions where the Queue cannot be guaranteed to
     * be emptied as quick as it's filled.
     *
     * An algorithm will be used to remove data when new data
     * arrives. There may be choices of algorithms later on.
     *
     * @param maxSize the upper limit for a queue
     * @param removeAlgorithm the remove algorithm to use upon reaching the maxSize
     */
    public Queue(int maxSize, int removeAlgorithm) {
        _maxSize = maxSize;
        _removeAlgorithm = removeAlgorithm;
    }
    
    /**
     * Constructs a Queue with no maximum size.
     */
    public Queue() {
        _maxSize = -1;
    }

//---PUBLIC METHODS---
    
    /**
     * This method adds a given object to every queue. It will notify
     * any waiting consumers (on an empty queue) during this process.
     *
     * @param o An Object to be added to the queues.
     */
    public void add(Object o) {
        for(int i=0; i < _lists.size(); i++) {
            // skip over any gaps left in the list
            if(_lists.get(i) != null) {
                // get size before adding to the Queue
                int s = ((LinkedList) _lists.get(i)).size();
                // check whether we need to remove an item from the current Queue
                if(_maxSize!=-1 && s==_maxSize && _removeAlgorithm!=DROP) {
                    // we need to remove an item
                    removeQueueItem((LinkedList) _lists.get(i));
                }
                // check if we should add (not if Queue full, and using DROP algorithm)
                if(!(s==_maxSize && _removeAlgorithm==DROP)) {
                    // add the next item, ensuring we lock
                    synchronized(this) {
                        // LinkedList.add() does the same thing, but this ensures behaviour
                        ((LinkedList) _lists.get(i)).addLast(o);
                    }
                }
                // if the queue was empty before the add it is possible
                // that a consumer is waiting... so we notify them
                if (s == 0) {
                    synchronized(((LinkedList) _lists.get(i))) {
                        ((LinkedList) _lists.get(i)).notifyAll();
                    }
                }
            }
        }
        // we keep track of the total additions for the status() method
        _count++;
    }
    
    /**
     * This method returns an object from the front of a given queue.
     * It will block until data exists in the queue if required.
     *
     * @param The queue to retrieve data from.
     * @return The object from the front of the queue.
     * @throws InvalidQueueException if the queue does not exist.
     */
    public Object get(int queue) throws InvalidQueueException {
        // make sure queue exists
        if (queue >= _lists.size() || _lists.get(queue) == null) {
            throw new InvalidQueueException("Requested queue "+queue+" does not exist");
        }
        // block if the queue is empty
        if (((LinkedList) _lists.get(queue)).size() == 0) {
            synchronized(((LinkedList) _lists.get(queue))) {
                try { ((LinkedList) _lists.get(queue)).wait(); } catch(Exception e) {}
            }
        }
        // get an item, it should never be null due to the blocking above
        Object o = null;
        synchronized(this) {
            try {
                o = ((LinkedList) _lists.get(queue)).removeFirst();
            }
            catch (NoSuchElementException e) {
                // This should never happen !
            }
        }
        return o;
    }
    
    /**
     * This method releases a get() method that's currently
     * waiting on an empty queue. This was designed for
     * shutdown() type methods that may have problems closing
     * if the thread of control is waiting on a queue.
     *
     * @param queue the queue to release.
     */
    public void releaseQueue(int queue) {
        synchronized(((LinkedList) _lists.get(queue))) {
                ((LinkedList) _lists.get(queue)).notifyAll();
        }
    }

    /**
     * This method erases the contents of a given queue. This
     * method should be used with care. It can only empty one
     * internal queue, not all of them. This must be called
     * multiple times to empty all queues.
     *
     * @param queue the queue to empty.
     */
    public void clearQueue(int queue) {
        synchronized(this) {
            ((LinkedList) _lists.get(queue)).clear();
        }
    }
    
    /**
     * This method returns an XML textual status of the queues.
     * It is merely for observation, and would most likely be
     * used by a larger "monitoring" component. Information
     * returned includes the current size of each queue, and
     * the total items passed through.
     *
     * @return A String message containing status information in XML format
     */
    public String xmlStatus() {
        String status = "<queue ";
        for(int i=0; i < _lists.size(); i++) {
            // check for null entries
            if(_lists.get(i) != null) {
                status += "queue"+i+"=\""+((LinkedList) _lists.get(i)).size()+"\" ";
            }
            else {
                status += "queue"+i+"=\"[deleted]\" ";
            }
        }
        status += "total=\""+_count+"\"";
        if(_maxSize != -1) {
            status += " maxSize=\""+_maxSize+"\"";
        }
        status += "</queue>";
        return status;
    }
    
    /**
     * Returns the size of a given queue. A consumer can use
     * this to see how big their queue is at any given time.
     * they should use their queue number as the parameter.
     *
     * @param queue The queue number to query.
     * @return the current size of the queue.
     * @throws InvalidQueueException if the queue does not exist.
     */
    public int queueSize(int queue) throws InvalidQueueException {
        if (queue >= _lists.size() || _lists.get(queue) == null) {
            throw new InvalidQueueException("Requested queue "+queue+" does not exist");
        }
        return ((LinkedList) _lists.get(queue)).size();
    }
    
    /**
     * Returns the total numer of elements to have passed
     * through this queue (ie. a counter on the add method).
     *
     * @return the element-ometer.
     */
    public int elementCount() {
        return _count;
    }
    
    /**
     * This method assigns a queue to a consumer. The idea behind
     * this is to ensure that only 1 consumer can be associated with
     * a given queue, otherwise the whole "blocking" thing fails
     * miserably. Queues are created upon requested.
     *
     * It is IMPORTANT that removeQueue() is used when the queue is
     * no longer required.
     *
     * @return An integer to be passed to the get() method.
     */
    public int getQueue() {
        int pos = -1;
        for(int i=0; i < _lists.size(); i++) {
            if(_lists.get(i) == null) {
                // found a gap, re-use it
                pos = i;
                _lists.set(i, new LinkedList());
            }
        }
        if(pos == -1) {
            //we didn't find a gap, add at end
            pos = _lists.size();
            _lists.add(pos, new LinkedList());
        }
        return pos;
    }
    
    /**
     * This method sets a entry to null in the list. This ensures
     * that it will no longer be added to after it is no longer
     * required be a consumer.
     *
     * @param queue The integer identifier for the queue, given by getQueue().
     */
    public void removeQueue(int queue) {
        _lists.set(queue, null);
    }
    
    /**
     * Start a monitor on our own Queue. This will log XML
     * statistics about our Queue to a given Queue (could be
     * the one being monitored).
     *
     * @param interval The long interval, in milliseconds, at which to take samples
     * @param destQueue The queue to monitor to
     * @param name A name to identify this Queue with
     * @return whether we succeeded
     */
    public boolean startMonitor(long interval, Queue destQueue, String name) {
        if(_queueMon == null) {
            // start a monitor
            _queueMon = new QueueMonitor(this, destQueue, interval, name);
            _queueMon.start();
            return true;
        }
        else {
            // already have a monitor running
            return false;
        }
    }
    
    /**
     * Start a monitor on our own Queue. This will log XML
     * statistics about our Queue to this Queue.
     *
     * @param interval The long interval, in milliseconds, at which to take samples
     * @param name A name to identify this Queue with
     * @return whether we succeeded
     */
    public boolean startMonitor(long interval, String name) {
        return startMonitor(interval, this, name);
    }
    
    /**
     * Stop a monitor on our Queue if we have on running.
     *
     * @return whether we succeeded
     */
    public boolean stopMonitor() {
        if(_queueMon != null) {
            // stop a monitor
            _queueMon.shutdown();
            _queueMon = null;
            return true;
        }
        else {
            // no monitor running
            return false;
        }
    }
    
    /**
     * Overrides the {@link java.lang.Object#toString() Object.toString()}
     * method to provide clean logging (every class should have this).
     *
     * This uses the uk.ac.ukc.iscream.util.FormatName class
     * to format the toString()
     *
     * @return the name of this class and its CVS revision.
     */
    public String toString() {
        return FormatName.getName(
            _name,
            getClass().getName(),
            REVISION);
    }

//---PRIVATE METHODS---
    
    /**
     * This method removes an item from a Queue, using a method
     * requested at construction.
     *
     * @param list The LinkedList from which to remove an item.
     */
    private void removeQueueItem(LinkedList list) {
        // look at our algorithm
        // remove a random item from the list
        if(_removeAlgorithm==RANDOM) {
            // new Random, with a good seed
            Random rand = new Random(System.currentTimeMillis());
            int i = rand.nextInt(_maxSize);
            synchronized(this) {
                list.remove(i);
            }
        }
        // remove the first item from the list
        else if(_removeAlgorithm==FIRST) {
            synchronized(this) {
                list.removeFirst();
            }
        }
        // remove the last item from the list
        else if(_removeAlgorithm==LAST) {
            synchronized(this) {
                list.removeLast();
            }
        }
    }
    
//---ACCESSOR/MUTATOR METHODS---

//---ATTRIBUTES---
    
    /**
     * The LinkedLists of queues.
     */
    private LinkedList _lists = new LinkedList();
    
    /**
     * A counter so we know how many data items have been
     * passed through, for statistical purposes.
     */
    private int _count = 0;
    
    /**
     * A reference to our QueueMonitor, if we have one.
     */
    private QueueMonitor _queueMon = null;
    
    /**
     * The maximum size of any Queue.
     */
    private int _maxSize = -1;
    
    /**
     * The remove algorithm to use upon a Queue reaching
     * it's maximum size.
     */
    private int _removeAlgorithm = -1;
    
    /**
     * This is the friendly identifier of the
     * component this class is running in.
     * eg, a Filter may be called "filter1",
     * If this class does not have an owning
     * component,  a name from the configuration
     * can be placed here.  This name could also
     * be changed to null for utility classes.
     */
    private String _name = null;

//---STATIC ATTRIBUTES---

}
Revision:	1.14
Committed:	Thu Mar 1 01:05:49 2001 UTC (23 years, 2 months ago) by tdb
Branch:	MAIN
Changes since 1.13:	+112 -7 lines
Log Message:	Finally implemented maximum size on Queue's. This is an optional feature, and is only used if the correct constructor is called. For normal use (with no limit), just use the empty constructor; new Queue() For using a limit, use the following constructor; new Queue(int maxSize, int removeAlgorithm) Where; maxSize = the maximum size of a queue removeAlgorithm = Queue.RANDOM, Queue.FIRST, Queue.LAST, or Queue.DROP Queue.RANDOM = Remove a RANDOM item from a full queue when adding. Queue.FIRST = Remove the FIRST item from a full queue when adding. Queue.LAST = Remove the LAST item from a full queue when adding. Queue.DROP = Don't remove, just drop the new item when adding.
#	Content
1	//---PACKAGE DECLARATION---
2	package uk.ac.ukc.iscream.util;
3
4	//---IMPORTS---
5	import java.util.LinkedList;
6	import java.util.NoSuchElementException;
7	import java.util.Random;
8	import uk.ac.ukc.iscream.util.*;
9
10	/**
11	* A Queue class designed to operate in a multi-threaded environment, with
12	* added support for multiple "consumer" threads. Also offers blocking on
13	* the get() methods, which ensures the consumer waits until the queue
14	* actually contains some elements.
15	*
16	* @author $Author: tdb1 $
17	* @version $Id: Queue.java,v 1.13 2001/02/25 23:29:06 tdb1 Exp $
18	*/
19	public class Queue {
20
21	//---FINAL ATTRIBUTES---
22
23	/**
24	* The current CVS revision of this class
25	*/
26	public static final String REVISION = "$Revision: 1.13 $";
27
28	/**
29	* Pass to constructor to remove a RANDOM item from
30	* the Queue upon reaching the maximum limit.
31	*/
32	public static final int RANDOM = 0;
33
34	/**
35	* Pass to constructor to remove the FIRST item from
36	* the Queue upon reaching the maximum limit.
37	*/
38	public static final int FIRST = 1;
39
40	/**
41	* Pass to constructor to remove the LAST item from
42	* the Queue upon reaching the maximum limit.
43	*/
44	public static final int LAST = 2;
45
46	/**
47	* Pass to constructor to drop the new item upon reaching
48	* the maximum Queue limit.
49	*/
50	public static final int DROP = 3;
51
52	//---STATIC METHODS---
53
54	//---CONSTRUCTORS---
55
56	/**
57	* Constructs a new Queue with a maximum size limit on
58	* any individual queue. This should be used to stop
59	* conditions where the Queue cannot be guaranteed to
60	* be emptied as quick as it's filled.
61	*
62	* An algorithm will be used to remove data when new data
63	* arrives. There may be choices of algorithms later on.
64	*
65	* @param maxSize the upper limit for a queue
66	* @param removeAlgorithm the remove algorithm to use upon reaching the maxSize
67	*/
68	public Queue(int maxSize, int removeAlgorithm) {
69	_maxSize = maxSize;
70	_removeAlgorithm = removeAlgorithm;
71	}
72
73	/**
74	* Constructs a Queue with no maximum size.
75	*/
76	public Queue() {
77	_maxSize = -1;
78	}
79
80	//---PUBLIC METHODS---
81
82	/**
83	* This method adds a given object to every queue. It will notify
84	* any waiting consumers (on an empty queue) during this process.
85	*
86	* @param o An Object to be added to the queues.
87	*/
88	public void add(Object o) {
89	for(int i=0; i < _lists.size(); i++) {
90	// skip over any gaps left in the list
91	if(_lists.get(i) != null) {
92	// get size before adding to the Queue
93	int s = ((LinkedList) _lists.get(i)).size();
94	// check whether we need to remove an item from the current Queue
95	if(_maxSize!=-1 && s==_maxSize && _removeAlgorithm!=DROP) {
96	// we need to remove an item
97	removeQueueItem((LinkedList) _lists.get(i));
98	}
99	// check if we should add (not if Queue full, and using DROP algorithm)
100	if(!(s==_maxSize && _removeAlgorithm==DROP)) {
101	// add the next item, ensuring we lock
102	synchronized(this) {
103	// LinkedList.add() does the same thing, but this ensures behaviour
104	((LinkedList) _lists.get(i)).addLast(o);
105	}
106	}
107	// if the queue was empty before the add it is possible
108	// that a consumer is waiting... so we notify them
109	if (s == 0) {
110	synchronized(((LinkedList) _lists.get(i))) {
111	((LinkedList) _lists.get(i)).notifyAll();
112	}
113	}
114	}
115	}
116	// we keep track of the total additions for the status() method
117	_count++;
118	}
119
120	/**
121	* This method returns an object from the front of a given queue.
122	* It will block until data exists in the queue if required.
123	*
124	* @param The queue to retrieve data from.
125	* @return The object from the front of the queue.
126	* @throws InvalidQueueException if the queue does not exist.
127	*/
128	public Object get(int queue) throws InvalidQueueException {
129	// make sure queue exists
130	if (queue >= _lists.size() \|\| _lists.get(queue) == null) {
131	throw new InvalidQueueException("Requested queue "+queue+" does not exist");
132	}
133	// block if the queue is empty
134	if (((LinkedList) _lists.get(queue)).size() == 0) {
135	synchronized(((LinkedList) _lists.get(queue))) {
136	try { ((LinkedList) _lists.get(queue)).wait(); } catch(Exception e) {}
137	}
138	}
139	// get an item, it should never be null due to the blocking above
140	Object o = null;
141	synchronized(this) {
142	try {
143	o = ((LinkedList) _lists.get(queue)).removeFirst();
144	}
145	catch (NoSuchElementException e) {
146	// This should never happen !
147	}
148	}
149	return o;
150	}
151
152	/**
153	* This method releases a get() method that's currently
154	* waiting on an empty queue. This was designed for
155	* shutdown() type methods that may have problems closing
156	* if the thread of control is waiting on a queue.
157	*
158	* @param queue the queue to release.
159	*/
160	public void releaseQueue(int queue) {
161	synchronized(((LinkedList) _lists.get(queue))) {
162	((LinkedList) _lists.get(queue)).notifyAll();
163	}
164	}
165
166	/**
167	* This method erases the contents of a given queue. This
168	* method should be used with care. It can only empty one
169	* internal queue, not all of them. This must be called
170	* multiple times to empty all queues.
171	*
172	* @param queue the queue to empty.
173	*/
174	public void clearQueue(int queue) {
175	synchronized(this) {
176	((LinkedList) _lists.get(queue)).clear();
177	}
178	}
179
180	/**
181	* This method returns an XML textual status of the queues.
182	* It is merely for observation, and would most likely be
183	* used by a larger "monitoring" component. Information
184	* returned includes the current size of each queue, and
185	* the total items passed through.
186	*
187	* @return A String message containing status information in XML format
188	*/
189	public String xmlStatus() {
190	String status = "<queue ";
191	for(int i=0; i < _lists.size(); i++) {
192	// check for null entries
193	if(_lists.get(i) != null) {
194	status += "queue"+i+"=\""+((LinkedList) _lists.get(i)).size()+"\" ";
195	}
196	else {
197	status += "queue"+i+"=\"[deleted]\" ";
198	}
199	}
200	status += "total=\""+_count+"\"";
201	if(_maxSize != -1) {
202	status += " maxSize=\""+_maxSize+"\"";
203	}
204	status += "</queue>";
205	return status;
206	}
207
208	/**
209	* Returns the size of a given queue. A consumer can use
210	* this to see how big their queue is at any given time.
211	* they should use their queue number as the parameter.
212	*
213	* @param queue The queue number to query.
214	* @return the current size of the queue.
215	* @throws InvalidQueueException if the queue does not exist.
216	*/
217	public int queueSize(int queue) throws InvalidQueueException {
218	if (queue >= _lists.size() \|\| _lists.get(queue) == null) {
219	throw new InvalidQueueException("Requested queue "+queue+" does not exist");
220	}
221	return ((LinkedList) _lists.get(queue)).size();
222	}
223
224	/**
225	* Returns the total numer of elements to have passed
226	* through this queue (ie. a counter on the add method).
227	*
228	* @return the element-ometer.
229	*/
230	public int elementCount() {
231	return _count;
232	}
233
234	/**
235	* This method assigns a queue to a consumer. The idea behind
236	* this is to ensure that only 1 consumer can be associated with
237	* a given queue, otherwise the whole "blocking" thing fails
238	* miserably. Queues are created upon requested.
239	*
240	* It is IMPORTANT that removeQueue() is used when the queue is
241	* no longer required.
242	*
243	* @return An integer to be passed to the get() method.
244	*/
245	public int getQueue() {
246	int pos = -1;
247	for(int i=0; i < _lists.size(); i++) {
248	if(_lists.get(i) == null) {
249	// found a gap, re-use it
250	pos = i;
251	_lists.set(i, new LinkedList());
252	}
253	}
254	if(pos == -1) {
255	//we didn't find a gap, add at end
256	pos = _lists.size();
257	_lists.add(pos, new LinkedList());
258	}
259	return pos;
260	}
261
262	/**
263	* This method sets a entry to null in the list. This ensures
264	* that it will no longer be added to after it is no longer
265	* required be a consumer.
266	*
267	* @param queue The integer identifier for the queue, given by getQueue().
268	*/
269	public void removeQueue(int queue) {
270	_lists.set(queue, null);
271	}
272
273	/**
274	* Start a monitor on our own Queue. This will log XML
275	* statistics about our Queue to a given Queue (could be
276	* the one being monitored).
277	*
278	* @param interval The long interval, in milliseconds, at which to take samples
279	* @param destQueue The queue to monitor to
280	* @param name A name to identify this Queue with
281	* @return whether we succeeded
282	*/
283	public boolean startMonitor(long interval, Queue destQueue, String name) {
284	if(_queueMon == null) {
285	// start a monitor
286	_queueMon = new QueueMonitor(this, destQueue, interval, name);
287	_queueMon.start();
288	return true;
289	}
290	else {
291	// already have a monitor running
292	return false;
293	}
294	}
295
296	/**
297	* Start a monitor on our own Queue. This will log XML
298	* statistics about our Queue to this Queue.
299	*
300	* @param interval The long interval, in milliseconds, at which to take samples
301	* @param name A name to identify this Queue with
302	* @return whether we succeeded
303	*/
304	public boolean startMonitor(long interval, String name) {
305	return startMonitor(interval, this, name);
306	}
307
308	/**
309	* Stop a monitor on our Queue if we have on running.
310	*
311	* @return whether we succeeded
312	*/
313	public boolean stopMonitor() {
314	if(_queueMon != null) {
315	// stop a monitor
316	_queueMon.shutdown();
317	_queueMon = null;
318	return true;
319	}
320	else {
321	// no monitor running
322	return false;
323	}
324	}
325
326	/**
327	* Overrides the {@link java.lang.Object#toString() Object.toString()}
328	* method to provide clean logging (every class should have this).
329	*
330	* This uses the uk.ac.ukc.iscream.util.FormatName class
331	* to format the toString()
332	*
333	* @return the name of this class and its CVS revision.
334	*/
335	public String toString() {
336	return FormatName.getName(
337	_name,
338	getClass().getName(),
339	REVISION);
340	}
341
342	//---PRIVATE METHODS---
343
344	/**
345	* This method removes an item from a Queue, using a method
346	* requested at construction.
347	*
348	* @param list The LinkedList from which to remove an item.
349	*/
350	private void removeQueueItem(LinkedList list) {
351	// look at our algorithm
352	// remove a random item from the list
353	if(_removeAlgorithm==RANDOM) {
354	// new Random, with a good seed
355	Random rand = new Random(System.currentTimeMillis());
356	int i = rand.nextInt(_maxSize);
357	synchronized(this) {
358	list.remove(i);
359	}
360	}
361	// remove the first item from the list
362	else if(_removeAlgorithm==FIRST) {
363	synchronized(this) {
364	list.removeFirst();
365	}
366	}
367	// remove the last item from the list
368	else if(_removeAlgorithm==LAST) {
369	synchronized(this) {
370	list.removeLast();
371	}
372	}
373	}
374
375	//---ACCESSOR/MUTATOR METHODS---
376
377	//---ATTRIBUTES---
378
379	/**
380	* The LinkedLists of queues.
381	*/
382	private LinkedList _lists = new LinkedList();
383
384	/**
385	* A counter so we know how many data items have been
386	* passed through, for statistical purposes.
387	*/
388	private int _count = 0;
389
390	/**
391	* A reference to our QueueMonitor, if we have one.
392	*/
393	private QueueMonitor _queueMon = null;
394
395	/**
396	* The maximum size of any Queue.
397	*/
398	private int _maxSize = -1;
399
400	/**
401	* The remove algorithm to use upon a Queue reaching
402	* it's maximum size.
403	*/
404	private int _removeAlgorithm = -1;
405
406	/**
407	* This is the friendly identifier of the
408	* component this class is running in.
409	* eg, a Filter may be called "filter1",
410	* If this class does not have an owning
411	* component, a name from the configuration
412	* can be placed here. This name could also
413	* be changed to null for utility classes.
414	*/
415	private String _name = null;
416
417	//---STATIC ATTRIBUTES---
418
419	}