A new queue has come
"Jean Morissette" <[email protected]> Thu, 02 Dec 2004 10:32:42 -0500
| Newsgroups | gmane.comp.java.seda.user |
|---|---|
| Message-ID | <[email protected]> |
<html><div style='background-color:'><DIV class=RTE>I post here a queue prototype that this backed by a resizable array. I don't have tested it a lot yet, but it should be more scalable than actual FiniteQueue and consume less memory. Please let me know if you find bugs or have comments.</DIV>
<DIV class=RTE>Jean M</DIV>
<DIV class=RTE> </DIV>
<DIV class=RTE><BR>/**<BR> * QueueIF implementation that use an array, so no memory allocation occur<BR> * on queue operations apart from situations where array needs to be resized.<BR> */<BR>public class DynamicArrayBlockingQueue implements BlockingQueueIF {</DIV>
<DIV class=RTE> protected QueueElementIF[] array; // the elements</DIV>
<DIV class=RTE> protected int takePtr = 0; // circular indices<BR> protected int putPtr = 0;</DIV>
<DIV class=RTE> protected int usedSlots = 0; // length<BR> protected int emptySlots; // capacity - length</DIV>
<DIV class=RTE> /**<BR> * Helper monitor to handle puts.<BR> */<BR> protected final Object putMonitor = new Object();</DIV>
<DIV class=RTE> protected int maxCap;</DIV>
<DIV class=RTE> // transactional map<BR> protected Map txnMap = Collections.synchronizedMap(new HashMap());</DIV>
<DIV class=RTE> protected EnqueuePredicateIF pred;</DIV>
<DIV class=RTE> /**<BR> * Create a BoundedBuffer with the given capacity.<BR> *<BR> * @throws IllegalArgumentException if capacity less or equal to zero<BR> */<BR> public DynamicArrayBlockingQueue(int capacity) throws IllegalArgumentException {<BR> if (capacity <= 0) throw new IllegalArgumentException();<BR> array = new QueueElementIF[capacity];<BR> emptySlots = capacity;<BR> this.maxCap = capacity;<BR> }</DIV>
<DIV class=RTE> /**<BR> * Create a queue with the current default capacity<BR> */<BR> public DynamicArrayBlockingQueue() {<BR> this(10);<BR> }</DIV>
<DIV class=RTE> /**<BR> * Return the number of elements in the buffer.<BR> * This is only a snapshot value, that may change<BR> * immediately after returning.<BR> */<BR> public synchronized int size() {<BR> return usedSlots;<BR> }</DIV>
<DIV class=RTE> public int capacity() {<BR> return maxCap;<BR> }</DIV>
<DIV class=RTE><BR> public void setCapacity(int newCap) {<BR> if (newCap <= 0) throw new IllegalArgumentException();<BR> if (newCap == maxCap) return;<BR> synchronized(putMonitor) {<BR> synchronized(this) {<BR> if (newCap > maxCap) {<BR> QueueElementIF[] newBuf = new QueueElementIF[newCap];<BR> if (takePtr <= putPtr) {<BR> System.arraycopy(array, takePtr, newBuf, 0, takePtr - putPtr);<BR> putPtr -= takePtr;<BR> takePtr =
0;<BR> }<BR> else {<BR> int length = array.length - takePtr;<BR> System.arraycopy(array, takePtr, newBuf, 0, length);<BR> System.arraycopy(array, 0, newBuf, length, putPtr);<BR> putPtr = length + putPtr;<BR> takePtr = 0;<BR> }<BR> array = newBuf;<BR> }<BR> emptySlots +=
newCap - maxCap;<BR> maxCap = newCap;<BR> }<BR> }<BR> }</DIV>
<DIV class=RTE> public void trimToSize() {<BR> throw new UnsupportedOperationException("Not implemented yet");<BR> }</DIV>
<DIV class=RTE> public QueueElementIF peek() {<BR> synchronized (this) {<BR> if (usedSlots > 0)<BR> return array[takePtr];<BR> else<BR> return null;<BR> }<BR> }</DIV>
<DIV class=RTE> public void blocking_enqueue(QueueElementIF x) throws InterruptedException {<BR> if (x == null) throw new IllegalArgumentException();<BR> if (Thread.interrupted()) throw new InterruptedException();</DIV>
<DIV class=RTE> synchronized (putMonitor) {<BR> while (emptySlots <= 0) {<BR> try {<BR> putMonitor.wait();<BR> } catch (InterruptedException ex) {<BR> putMonitor.notify();<BR> throw ex;<BR> }<BR> }<BR> insert(x);<BR> }<BR> incUsedSlots();<BR> }</DIV>
<DIV class=RTE> public boolean enqueue_lossy(QueueElementIF x, long msecs) throws InterruptedException {<BR> if (x == null) throw new IllegalArgumentException();<BR> if (Thread.interrupted()) throw new InterruptedException();</DIV>
<DIV class=RTE> synchronized (putMonitor) {<BR> long start = (msecs <= 0) ? 0 : System.currentTimeMillis();<BR> long waitTime = msecs;<BR> while (emptySlots <= 0) {<BR> if (waitTime <= 0) return false;<BR> try {<BR> putMonitor.wait(waitTime);<BR> } catch (InterruptedException ex) {<BR> putMonitor.notify();<BR> throw ex;<BR> }<BR> waitTime = msecs - (System.currentTimeMillis() -
start);<BR> }<BR> insert(x);<BR> }<BR> incUsedSlots();<BR> return true;<BR> }</DIV>
<DIV class=RTE> public QueueElementIF blocking_dequeue(int timeout_millis) {<BR> try {<BR> if (timeout_millis == -1)<BR> return blocking_dequeue();<BR> else<BR> return blocking_dequeue0(timeout_millis);<BR> } catch (InterruptedException e) {<BR> e.printStackTrace();<BR> return null;<BR> }<BR> }</DIV>
<DIV class=RTE> private QueueElementIF blocking_dequeue() throws InterruptedException {<BR> if (Thread.interrupted()) throw new InterruptedException();<BR> QueueElementIF old = null;<BR> synchronized (this) {<BR> while (usedSlots <= 0) {<BR> try {<BR> wait();<BR> } catch (InterruptedException ex) {<BR> notify();<BR> throw ex;<BR> }<BR> }<BR> old = extract();<BR> }<BR> incEmptySlots();<BR> return old;<B
R> }</DIV>
<DIV class=RTE> private QueueElementIF blocking_dequeue0(long msecs) throws InterruptedException {<BR> if (Thread.interrupted()) throw new InterruptedException();<BR> QueueElementIF old = null;<BR> synchronized (this) {<BR> long start = (msecs <= 0) ? 0 : System.currentTimeMillis();<BR> long waitTime = msecs;</DIV>
<DIV class=RTE> while (usedSlots <= 0) {<BR> if (waitTime <= 0) return null;<BR> try {<BR> wait(waitTime);<BR> } catch (InterruptedException ex) {<BR> notify();<BR> throw ex;<BR> }<BR> waitTime = msecs - (System.currentTimeMillis() - start);</DIV>
<DIV class=RTE> }<BR> old = extract();<BR> }<BR> incEmptySlots();<BR> return old;<BR> }</DIV>
<DIV class=RTE> public Object enqueue_prepare(QueueElementIF[] elements) throws SinkException {<BR> synchronized (putMonitor) {<BR> if (emptySlots < elements.length) {<BR> throw new SinkFullException(); /////// The queue is not necessary full ...<BR> }<BR> emptySlots -= elements.length;<BR> }<BR> Object key = new Object();<BR> txnMap.put(key, elements);<BR> return key;<BR> }</DIV>
<DIV class=RTE> public void enqueue_commit(Object enqueue_key) {<BR> QueueElementIF[] elements = (QueueElementIF[]) txnMap.get(enqueue_key);<BR> synchronized (putMonitor) {<BR> for (int i = 0; i < elements.length; i++) {<BR> array[putPtr] = elements[i];<BR> if (++putPtr >= array.length) putPtr = 0;<BR> }<BR> }<BR> synchronized (this) {<BR> usedSlots += elements.length;<BR> notify(); /////////// notify or notifyAll ???<BR> }<BR> }</DIV>
<DIV class=RTE> public void enqueue_abort(Object enqueue_key) {<BR> QueueElementIF[] elements = (QueueElementIF[]) txnMap.remove(enqueue_key);<BR> if (elements == null) throw new IllegalArgumentException("unknown key");<BR> synchronized (putMonitor) {<BR> emptySlots += elements.length;<BR> putMonitor.notify(); /////////// notify or notifyAll ???<BR> }<BR> }</DIV>
<DIV class=RTE> public void enqueue(QueueElementIF element) throws SinkException {<BR> try {<BR> if (!enqueue_lossy(element, 0))<BR> throw new SinkFullException();<BR> } catch (InterruptedException e) {<BR> e.printStackTrace();<BR> throw new SinkFullException();<BR> }<BR> }</DIV>
<DIV class=RTE> public boolean enqueue_lossy(QueueElementIF element) {<BR> try {<BR> return enqueue_lossy(element, 0);<BR> } catch (InterruptedException e) {<BR> e.printStackTrace();<BR> return false;<BR> }<BR> }</DIV>
<DIV class=RTE> public void enqueue_many(QueueElementIF[] elements) throws SinkException {<BR> synchronized (putMonitor) {<BR> if (emptySlots < elements.length) {<BR> throw new SinkFullException(); /////// The queue is not necessary full ...<BR> }<BR> insertMany(elements);<BR> }<BR> synchronized (this) {<BR> usedSlots += elements.length;<BR> notify(); /////////// notify or notifyAll ???<BR> }</DIV>
<DIV class=RTE> }</DIV>
<DIV class=RTE> public QueueElementIF[] dequeue_all() {<BR> throw new UnsupportedOperationException("Not implemented yet");<BR> }</DIV>
<DIV class=RTE> public QueueElementIF[] dequeue(int num) {<BR> throw new UnsupportedOperationException("Not implemented yet");<BR> }</DIV>
<DIV class=RTE> public QueueElementIF[] blocking_dequeue_all(int timeout_millis) {<BR> throw new UnsupportedOperationException("Not implemented yet");<BR> }</DIV>
<DIV class=RTE> public QueueElementIF[] blocking_dequeue(int timeout_millis, int num) {<BR> throw new UnsupportedOperationException("Not implemented yet");<BR> }</DIV>
<DIV class=RTE> public void setEnqueuePredicate(EnqueuePredicateIF pred) {<BR> this.pred = pred;<BR> }</DIV>
<DIV class=RTE> public EnqueuePredicateIF getEnqueuePredicate() {<BR> return pred;<BR> }</DIV>
<DIV class=RTE> protected void incEmptySlots() {<BR> synchronized (putMonitor) {<BR> ++emptySlots;<BR> putMonitor.notify();<BR> }<BR> }</DIV>
<DIV class=RTE> protected synchronized void incUsedSlots() {<BR> ++usedSlots;<BR> notify();<BR> }</DIV>
<DIV class=RTE> protected final void insert(QueueElementIF x) { // mechanics of put<BR> --emptySlots;<BR> array[putPtr] = x;<BR> if (++putPtr >= array.length) putPtr = 0;<BR> }</DIV>
<DIV class=RTE> protected final void insertMany(QueueElementIF[] elements) {<BR> emptySlots -= elements.length;<BR> for (int i = 0; i < elements.length; i++) {<BR> array[putPtr] = elements[i];<BR> if (++putPtr >= array.length) putPtr = 0;<BR> }<BR> }</DIV>
<DIV class=RTE> protected final QueueElementIF extract() { // mechanics of take<BR> --usedSlots;<BR> QueueElementIF old = array[takePtr];<BR> array[takePtr] = null;<BR> if (++takePtr >= array.length) takePtr = 0;<BR> return old;<BR> }</DIV>
<DIV class=RTE> public static void main(String[] args) throws InterruptedException {<BR> new Thread(new Consumer()).start();<BR> new Thread(new Producer()).start();<BR> while (true) {<BR> Thread.sleep(1);<BR> q.setCapacity(10);<BR> }<BR> }</DIV>
<DIV class=RTE> static DynamicArrayBlockingQueue q = new DynamicArrayBlockingQueue();</DIV>
<DIV class=RTE> static class Producer implements Runnable {<BR> public void run() {<BR> while (true) {<BR> try {<BR> q.blocking_enqueue(new QueueElementIF() {});<BR> } catch (InterruptedException e) {<BR> e.printStackTrace();<BR> }<BR> }<BR> }<BR> }</DIV>
<DIV class=RTE> static class Consumer implements Runnable {<BR> public void run() {<BR> while (true) {<BR> try {<BR> q.blocking_dequeue();<BR> } catch (InterruptedException e) {<BR> e.printStackTrace();<BR> }<BR> }<BR> }<BR> }<BR>}<BR></DIV>
<DIV class=RTE> </DIV>
<DIV class=RTE> </DIV></div></html>
-------------------------------------------------------
SF email is sponsored by - The IT Product Guide
Read honest & candid reviews on hundreds of IT Products from real users.
Discover which products truly live up to the hype. Start reading now.
http://productguide.itmanagersjournal.com/