util/concurrent/AbstractBlockingQueueFromQueue.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain. Use, modify, and
 * redistribute this code in any way without acknowledgement.
 */

package java.util.concurrent;
import java.util.*;

/**
 * An AbstractBlockingQueueFromQueue places blocking concurrency control
 * around a non-synchronized, non-thread-safe Queue.
 **/
abstract class AbstractBlockingQueueFromQueue<E> extends AbstractQueue<E> implements  BlockingQueue<E>, java.io.Serializable {

    /*
     * Concurrency control via the classic two-condition algorithm
     * found in any textbook.
     */

    private transient final FairReentrantLock lock = new FairReentrantLock();
    private transient final Condition notEmpty = lock.newCondition();
    private transient final Condition notFull =  lock.newCondition();
    private final Queue<E> q;
    private final int capacity;

    protected AbstractBlockingQueueFromQueue(Queue<E> queue, int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        q = queue;
    }


    public void put(E x) throws InterruptedException {
        if (x == null) throw new IllegalArgumentException();
        lock.lockInterruptibly();
        try {
            try {
                while (q.size() == capacity)
                    notFull.await();
            }
            catch (InterruptedException ie) {
                notFull.signal(); // propagate to non-interrupted thread
                throw ie;
            }
            boolean ok = q.offer(x);
            assert ok;
            notEmpty.signal();
        }
        finally {
            lock.unlock();
        }
    }

    public E take() throws InterruptedException {
        lock.lockInterruptibly();
        try {
            try {
                while (q.size() == 0)
                    notEmpty.await();
            }
            catch (InterruptedException ie) {
                notEmpty.signal(); // propagate to non-interrupted thread
                throw ie;
            }
            E x = q.poll();
            assert x != null;
            notFull.signal();
            return x;
        }
        finally {
            lock.unlock();
        }
    }

    public boolean offer(E x) {
        if (x == null) throw new IllegalArgumentException();
        lock.lock();
        try {
            if (q.size() == capacity) 
                return false;
            else {
                boolean ok = q.offer(x);
                assert ok;
                notEmpty.signal();
                return true;
            }
        }
        finally {
            lock.unlock(); 
        }
    }

    public E poll() {
        lock.lock();
        try {
            E x = q.poll();
            if (x != null)
                notFull.signal();
            return x;
        }
        finally {
            lock.unlock(); 
        }
    }

    public boolean offer(E x, long timeout, TimeUnit unit) throws InterruptedException {
        if (x == null) throw new IllegalArgumentException();
        lock.lockInterruptibly();
        long nanos = unit.toNanos(timeout);
        try {
            for (;;) {
                if (q.size() != capacity) {
                    boolean ok = q.offer(x);
                    assert ok;
                    notEmpty.signal();
                    return true;
                }
                if (nanos <= 0)
                    return false;
                try {
                    nanos = notFull.awaitNanos(nanos);
                }
                catch (InterruptedException ie) {
                    notFull.signal(); // propagate to non-interrupted thread
                    throw ie;
                }
            }
        }
        finally {
            lock.unlock();
        }
    }

    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        lock.lockInterruptibly();
        long nanos = unit.toNanos(timeout);
        try {
            for (;;) {
                E x = q.poll();
                if (x != null) {
                    notFull.signal();
                    return x;
                }
                if (nanos <= 0)
                    return null;
                try {
                    nanos = notEmpty.awaitNanos(nanos);
                }
                catch (InterruptedException ie) {
                    notEmpty.signal(); // propagate to non-interrupted thread
                    throw ie;
                }

            }
        }
        finally {
            lock.unlock();
        }
    }

    public E peek() {
        lock.lock();
        try {
            return q.peek();
        }
        finally {
            lock.unlock(); 
        }
    }

    public int size() {
        lock.lock();
        try {
            return q.size();
        }
        finally {
            lock.unlock();
        }
    }

    public int remainingCapacity() {
        lock.lock();
        try {
            return capacity - q.size();
        }
        finally {
            lock.unlock();
        }
    }

    public boolean remove(Object x) {
        lock.lock();
        try {
            boolean removed = q.remove(x);
            if (removed)
                notFull.signal();
            return removed;
        }
        finally {
            lock.unlock();
        }
    }

    public boolean contains(Object x) {
        lock.lock();
        try {
            return q.contains(x);
        }
        finally {
            lock.unlock();
        }
    }

    public Object[] toArray() {
        lock.lock();
        try {
            return q.toArray();
        }
        finally {
            lock.unlock();
        }
    }


    public String toString() {
        lock.lock();
        try {
            return q.toString();
        }
        finally {
            lock.unlock();
        }
    }

    public <T> T[] toArray(T[] a) {
        lock.lock();
        try {
            return q.toArray(a);
        }
        finally {
            lock.unlock();
        }
    }

    public Iterator<E> iterator() {
        lock.lock();
        try {
            return new Itr(q.iterator());
        }
        finally {
            lock.unlock();
        }
    }

    private class Itr<E> implements Iterator<E> {
        private final Iterator<E> iter;
        Itr(Iterator<E> i) { 
            iter = i; 
        }

        public boolean hasNext() {
            /*
             * No sync -- we rely on underlying hasNext to be
             * stateless, in which case we can return true by mistake
             * only when next() willl subsequently throw
             * ConcurrentModificationException.
             */
            return iter.hasNext();
        }
        
        public E next() {
            lock.lock();
            try {
                return iter.next();
            }
            finally {
                lock.unlock();
            }
        }
        
        public void remove() {
            lock.lock();
            try {
                iter.remove();
                notFull.signal();
            }
            finally {
                lock.unlock();
            }
        }
    }

    /**
     * Save the state to a stream (that is, serialize it).  This
     * merely wraps default serialization within lock.  The
     * serialization strategy for items is left to underlying
     * Queue. Note that locking is not needed on deserialization, so
     * readObject is not defined, just relying on default.
     */
    private void writeObject(java.io.ObjectOutputStream s)
        throws java.io.IOException {
        lock.lock();
        try {
            s.defaultWriteObject();
        }
        finally {
            lock.unlock();
        }
    }
}

        
Revision:	1.2
Committed:	Sun Jun 22 21:47:17 2003 UTC (20 years, 11 months ago) by dl
Branch:	MAIN
CVS Tags:	HEAD
Changes since 1.1:	+0 -0 lines
State:	*FILE REMOVED*
Log Message:	Split ArrayBlockingQueue and PriorityBlockingQueue to no longer subclass AbstractBlockingQueueFromQueue
#	User	Rev	Content
1	dl	1.1	/*
2			* Written by Doug Lea with assistance from members of JCP JSR-166
3			* Expert Group and released to the public domain. Use, modify, and
4			* redistribute this code in any way without acknowledgement.
5			*/
6
7			package java.util.concurrent;
8			import java.util.*;
9
10			/**
11			* An AbstractBlockingQueueFromQueue places blocking concurrency control
12			* around a non-synchronized, non-thread-safe Queue.
13			**/
14			abstract class AbstractBlockingQueueFromQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {
15
16			/*
17			* Concurrency control via the classic two-condition algorithm
18			* found in any textbook.
19			*/
20
21			private transient final FairReentrantLock lock = new FairReentrantLock();
22			private transient final Condition notEmpty = lock.newCondition();
23			private transient final Condition notFull = lock.newCondition();
24			private final Queue<E> q;
25			private final int capacity;
26
27			protected AbstractBlockingQueueFromQueue(Queue<E> queue, int capacity) {
28			if (capacity <= 0) throw new IllegalArgumentException();
29			this.capacity = capacity;
30			q = queue;
31			}
32
33
34			public void put(E x) throws InterruptedException {
35			if (x == null) throw new IllegalArgumentException();
36			lock.lockInterruptibly();
37			try {
38			try {
39			while (q.size() == capacity)
40			notFull.await();
41			}
42			catch (InterruptedException ie) {
43			notFull.signal(); // propagate to non-interrupted thread
44			throw ie;
45			}
46			boolean ok = q.offer(x);
47			assert ok;
48			notEmpty.signal();
49			}
50			finally {
51			lock.unlock();
52			}
53			}
54
55			public E take() throws InterruptedException {
56			lock.lockInterruptibly();
57			try {
58			try {
59			while (q.size() == 0)
60			notEmpty.await();
61			}
62			catch (InterruptedException ie) {
63			notEmpty.signal(); // propagate to non-interrupted thread
64			throw ie;
65			}
66			E x = q.poll();
67			assert x != null;
68			notFull.signal();
69			return x;
70			}
71			finally {
72			lock.unlock();
73			}
74			}
75
76			public boolean offer(E x) {
77			if (x == null) throw new IllegalArgumentException();
78			lock.lock();
79			try {
80			if (q.size() == capacity)
81			return false;
82			else {
83			boolean ok = q.offer(x);
84			assert ok;
85			notEmpty.signal();
86			return true;
87			}
88			}
89			finally {
90			lock.unlock();
91			}
92			}
93
94			public E poll() {
95			lock.lock();
96			try {
97			E x = q.poll();
98			if (x != null)
99			notFull.signal();
100			return x;
101			}
102			finally {
103			lock.unlock();
104			}
105			}
106
107			public boolean offer(E x, long timeout, TimeUnit unit) throws InterruptedException {
108			if (x == null) throw new IllegalArgumentException();
109			lock.lockInterruptibly();
110			long nanos = unit.toNanos(timeout);
111			try {
112			for (;;) {
113			if (q.size() != capacity) {
114			boolean ok = q.offer(x);
115			assert ok;
116			notEmpty.signal();
117			return true;
118			}
119			if (nanos <= 0)
120			return false;
121			try {
122			nanos = notFull.awaitNanos(nanos);
123			}
124			catch (InterruptedException ie) {
125			notFull.signal(); // propagate to non-interrupted thread
126			throw ie;
127			}
128			}
129			}
130			finally {
131			lock.unlock();
132			}
133			}
134
135			public E poll(long timeout, TimeUnit unit) throws InterruptedException {
136			lock.lockInterruptibly();
137			long nanos = unit.toNanos(timeout);
138			try {
139			for (;;) {
140			E x = q.poll();
141			if (x != null) {
142			notFull.signal();
143			return x;
144			}
145			if (nanos <= 0)
146			return null;
147			try {
148			nanos = notEmpty.awaitNanos(nanos);
149			}
150			catch (InterruptedException ie) {
151			notEmpty.signal(); // propagate to non-interrupted thread
152			throw ie;
153			}
154
155			}
156			}
157			finally {
158			lock.unlock();
159			}
160			}
161
162			public E peek() {
163			lock.lock();
164			try {
165			return q.peek();
166			}
167			finally {
168			lock.unlock();
169			}
170			}
171
172			public int size() {
173			lock.lock();
174			try {
175			return q.size();
176			}
177			finally {
178			lock.unlock();
179			}
180			}
181
182			public int remainingCapacity() {
183			lock.lock();
184			try {
185			return capacity - q.size();
186			}
187			finally {
188			lock.unlock();
189			}
190			}
191
192			public boolean remove(Object x) {
193			lock.lock();
194			try {
195			boolean removed = q.remove(x);
196			if (removed)
197			notFull.signal();
198			return removed;
199			}
200			finally {
201			lock.unlock();
202			}
203			}
204
205			public boolean contains(Object x) {
206			lock.lock();
207			try {
208			return q.contains(x);
209			}
210			finally {
211			lock.unlock();
212			}
213			}
214
215			public Object[] toArray() {
216			lock.lock();
217			try {
218			return q.toArray();
219			}
220			finally {
221			lock.unlock();
222			}
223			}
224
225
226			public String toString() {
227			lock.lock();
228			try {
229			return q.toString();
230			}
231			finally {
232			lock.unlock();
233			}
234			}
235
236			public <T> T[] toArray(T[] a) {
237			lock.lock();
238			try {
239			return q.toArray(a);
240			}
241			finally {
242			lock.unlock();
243			}
244			}
245
246			public Iterator<E> iterator() {
247			lock.lock();
248			try {
249			return new Itr(q.iterator());
250			}
251			finally {
252			lock.unlock();
253			}
254			}
255
256			private class Itr<E> implements Iterator<E> {
257			private final Iterator<E> iter;
258			Itr(Iterator<E> i) {
259			iter = i;
260			}
261
262			public boolean hasNext() {
263			/*
264			* No sync -- we rely on underlying hasNext to be
265			* stateless, in which case we can return true by mistake
266			* only when next() willl subsequently throw
267			* ConcurrentModificationException.
268			*/
269			return iter.hasNext();
270			}
271
272			public E next() {
273			lock.lock();
274			try {
275			return iter.next();
276			}
277			finally {
278			lock.unlock();
279			}
280			}
281
282			public void remove() {
283			lock.lock();
284			try {
285			iter.remove();
286			notFull.signal();
287			}
288			finally {
289			lock.unlock();
290			}
291			}
292			}
293
294			/**
295			* Save the state to a stream (that is, serialize it). This
296			* merely wraps default serialization within lock. The
297			* serialization strategy for items is left to underlying
298			* Queue. Note that locking is not needed on deserialization, so
299			* readObject is not defined, just relying on default.
300			*/
301			private void writeObject(java.io.ObjectOutputStream s)
302			throws java.io.IOException {
303			lock.lock();
304			try {
305			s.defaultWriteObject();
306			}
307			finally {
308			lock.unlock();
309			}
310			}
311			}
312
313
314