cms/util/Queue.java

//---PACKAGE DECLARATION---
package uk.org.iscream.cms.server.util;

//---IMPORTS---
import java.util.LinkedList;
import java.util.NoSuchElementException;
import java.util.Random;
import uk.org.iscream.cms.server.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.21 2001/04/04 20:51:16 tdb1 Exp $
 */
public class Queue {

//---FINAL ATTRIBUTES---

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