cms/util/Queue.java

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

//---IMPORTS---
import java.util.LinkedList;
import java.util.NoSuchElementException;
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.5 2001/01/19 01:11:55 tdb1 Exp $
 */
public class Queue {

//---FINAL ATTRIBUTES---

    /**
     * The current CVS revision of this class
     */
    public static final String REVISION = "$Revision: 1.5 $";
    
//---STATIC METHODS---

//---CONSTRUCTORS---  

//---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) {
                int s = ((LinkedList) _lists.get(i)).size();
                synchronized(this) {
                    // 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.
     *
     * @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 returns a 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.
     */
    public String status() {
        String status = "";
        for(int i=0; i < _lists.size(); i++) {
            // check for null entries
            if(_lists.get(i) != null) {
                status += "Queue number "+i+" contains "+((LinkedList) _lists.get(i)).size()+" elements";
                status += "\n";
            }
            else {
                status += "Slot number "+i+" is currently empty";
                status += "\n";
            }
        }
        status += "A total of "+_count+" elements have been added to the queues";
        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.
     */
    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);
    }
    
    /**
     * 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---

//---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;
    
    /**
     * 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.6
Committed:	Tue Jan 23 18:22:28 2001 UTC (23 years, 3 months ago) by tdb
Branch:	MAIN
Changes since 1.5:	+17 -3 lines
Log Message:	Added a releaseQueue() method that will "nudge" a get() method that's blocking.
#	User	Rev	Content
1	tdb	1.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 uk.ac.ukc.iscream.util.*;
8
9			/**
10			* A Queue class designed to operate in a multi-threaded environment, with
11			* added support for multiple "consumer" threads. Also offers blocking on
12			* the get() methods, which ensures the consumer waits until the queue
13			* actually contains some elements.
14			*
15			* @author $Author: tdb1 $
16	tdb	1.6	* @version $Id: Queue.java,v 1.5 2001/01/19 01:11:55 tdb1 Exp $
17	tdb	1.1	*/
18	tdb	1.2	public class Queue {
19	tdb	1.1
20			//---FINAL ATTRIBUTES---
21
22			/**
23			* The current CVS revision of this class
24			*/
25	tdb	1.6	public static final String REVISION = "$Revision: 1.5 $";
26	tdb	1.1
27			//---STATIC METHODS---
28
29			//---CONSTRUCTORS---
30
31			//---PUBLIC METHODS---
32
33			/**
34			* This method adds a given object to every queue. It will notify
35			* any waiting consumers (on an empty queue) during this process.
36			*
37			* @param o An Object to be added to the queues.
38			*/
39			public void add(Object o) {
40			for(int i=0; i < _lists.size(); i++) {
41			// skip over any gaps left in the list
42			if(_lists.get(i) != null) {
43			int s = ((LinkedList) _lists.get(i)).size();
44			synchronized(this) {
45			// add() does the same thing, but this ensures behaviour
46			((LinkedList) _lists.get(i)).addLast(o);
47			}
48			// if the queue was empty before the add it is possible
49			// that a consumer is waiting... so we notify them
50			if (s == 0) {
51			synchronized(((LinkedList) _lists.get(i))) {
52			((LinkedList) _lists.get(i)).notifyAll();
53			}
54			}
55			}
56			}
57			// we keep track of the total additions for the status() method
58			_count++;
59			}
60
61			/**
62			* This method returns an object from the front of a given queue.
63			* It will block until data exists in the queue if required.
64			*
65			* @return The object from the front of the queue.
66			* @throws InvalidQueueException if the queue does not exist.
67			*/
68			public Object get(int queue) throws InvalidQueueException {
69			// make sure queue exists
70			if (queue >= _lists.size() \|\| _lists.get(queue) == null) {
71			throw new InvalidQueueException("Requested queue "+queue+" does not exist");
72			}
73			// block if the queue is empty
74			if (((LinkedList) _lists.get(queue)).size() == 0) {
75			synchronized(((LinkedList) _lists.get(queue))) {
76			try { ((LinkedList) _lists.get(queue)).wait(); } catch(Exception e) {}
77			}
78			}
79			// get an item, it should never be null due to the blocking above
80			Object o = null;
81			synchronized(this) {
82			try {
83			o = ((LinkedList) _lists.get(queue)).removeFirst();
84			}
85			catch (NoSuchElementException e) {
86			// This should never happen !
87			}
88			}
89			return o;
90			}
91	tdb	1.6
92			/**
93			* This method releases a get() method that's currently
94			* waiting on an empty queue. This was designed for
95			* shutdown() type methods that may have problems closing
96			* if the thread of control is waiting on a queue.
97			*
98			* @param queue the queue to release
99			*/
100			public void releaseQueue(int queue) {
101			synchronized(((LinkedList) _lists.get(queue))) {
102			((LinkedList) _lists.get(queue)).notifyAll();
103			}
104			}
105
106	tdb	1.1	/**
107			* This method returns a textual status of the queues. It
108			* is merely for observation, and would most likely be used
109			* by a larger "monitoring" component. Information returned
110			* includes the current size of each queue, and the total
111			* items passed through.
112			*
113			* @return A String message containing status information.
114			*/
115			public String status() {
116			String status = "";
117			for(int i=0; i < _lists.size(); i++) {
118			// check for null entries
119			if(_lists.get(i) != null) {
120			status += "Queue number "+i+" contains "+((LinkedList) _lists.get(i)).size()+" elements";
121			status += "\n";
122			}
123			else {
124			status += "Slot number "+i+" is currently empty";
125			status += "\n";
126			}
127			}
128			status += "A total of "+_count+" elements have been added to the queues";
129			return status;
130	tdb	1.4	}
131
132			/**
133			* Returns the size of a given queue. A consumer can use
134			* this to see how big their queue is at any given time.
135			* they should use their queue number as the parameter.
136			*
137			* @param queue The queue number to query.
138			* @return the current size of the queue.
139			*/
140			public int queueSize(int queue) throws InvalidQueueException {
141	tdb	1.5	if (queue >= _lists.size() \|\| _lists.get(queue) == null) {
142	tdb	1.4	throw new InvalidQueueException("Requested queue "+queue+" does not exist");
143			}
144	tdb	1.5	return ((LinkedList) _lists.get(queue)).size();
145	tdb	1.4	}
146
147			/**
148			* Returns the total numer of elements to have passed
149			* through this queue (ie. a counter on the add method).
150			*
151			* @return the element-ometer
152			*/
153			public int elementCount() {
154	tdb	1.5	return _count;
155	tdb	1.1	}
156
157			/**
158			* This method assigns a queue to a consumer. The idea behind
159			* this is to ensure that only 1 consumer can be associated with
160			* a given queue, otherwise the whole "blocking" thing fails
161			* miserably. Queues are created upon requested.
162			*
163			* It is IMPORTANT that removeQueue() is used when the queue is
164			* no longer required.
165			*
166			* @return An integer to be passed to the get() method.
167			*/
168			public int getQueue() {
169			int pos = -1;
170			for(int i=0; i < _lists.size(); i++) {
171			if(_lists.get(i) == null) {
172			// found a gap, re-use it
173			pos = i;
174			_lists.set(i, new LinkedList());
175			}
176			}
177			if(pos == -1) {
178			//we didn't find a gap, add at end
179			pos = _lists.size();
180			_lists.add(pos, new LinkedList());
181			}
182			return pos;
183			}
184
185			/**
186			* This method sets a entry to null in the list. This ensures
187			* that it will no longer be added to after it is no longer
188			* required be a consumer.
189			*
190			* @param queue The integer identifier for the queue, given by getQueue().
191			*/
192			public void removeQueue(int queue) {
193			_lists.set(queue, null);
194			}
195
196			/**
197			* Overrides the {@link java.lang.Object#toString() Object.toString()}
198			* method to provide clean logging (every class should have this).
199			*
200			* This uses the uk.ac.ukc.iscream.util.FormatName class
201			* to format the toString()
202			*
203			* @return the name of this class and its CVS revision
204			*/
205	tdb	1.3	public String toString() {
206	tdb	1.1	return FormatName.getName(
207			_name,
208			getClass().getName(),
209			REVISION);
210	tdb	1.3	}
211	tdb	1.1
212			//---PRIVATE METHODS---
213
214			//---ACCESSOR/MUTATOR METHODS---
215
216			//---ATTRIBUTES---
217
218			/**
219			* The LinkedLists of queues.
220			*/
221			private LinkedList _lists = new LinkedList();
222
223			/**
224			* A counter so we know how many data items have been
225			* passed through, for statistical purposes.
226			*/
227			private int _count = 0;
228
229			/**
230			* This is the friendly identifier of the
231			* component this class is running in.
232			* eg, a Filter may be called "filter1",
233			* If this class does not have an owning
234			* component, a name from the configuration
235			* can be placed here. This name could also
236			* be changed to null for utility classes.
237			*/
238	tdb	1.3	private String _name = null;
239	tdb	1.1
240			//---STATIC ATTRIBUTES---
241
242			}