ViewVC Help

View File | Revision Log | Show Annotations | Download File | Root Listing

root/jsr166/jsr166/src/jsr166y/LinkedTransferQueue.java

Comparing jsr166/src/jsr166y/LinkedTransferQueue.java (file contents):
Revision 1.30 by jsr166, Mon Jul 27 03:22:39 2009 UTC vs.
Revision 1.40 by jsr166, Sat Aug 1 20:26:50 2009 UTC

#	Line 10 | Line 10 | import java.util.concurrent.*;
10
11		import java.util.AbstractQueue;
12		import java.util.Collection;
13	+	import java.util.ConcurrentModificationException;
14		import java.util.Iterator;
15		import java.util.NoSuchElementException;
16	+	import java.util.Queue;
17		import java.util.concurrent.locks.LockSupport;
18		import java.util.concurrent.atomic.AtomicReference;
17	–	import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
19
20		/**
21		* An unbounded {@linkplain TransferQueue} based on linked nodes.
#	Line 115 | Line 116 | public class LinkedTransferQueue<E> exte
116		this.isData = isData;
117		}
118
119	<	@SuppressWarnings("rawtypes")
120	<	static final AtomicReferenceFieldUpdater<Node, Node>
121	<	nextUpdater = AtomicReferenceFieldUpdater.newUpdater
122	<	(Node.class, Node.class, "next");
119	>	// Unsafe mechanics
120	>
121	>	private static final sun.misc.Unsafe UNSAFE = getUnsafe();
122	>	private static final long nextOffset =
123	>	objectFieldOffset(UNSAFE, "next", Node.class);
124
125		final boolean casNext(Node<E> cmp, Node<E> val) {
126	<	return nextUpdater.compareAndSet(this, cmp, val);
126	>	return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
127		}
128
129		final void clearNext() {
130	<	nextUpdater.lazySet(this, this);
130	>	UNSAFE.putOrderedObject(this, nextOffset, this);
131	>	}
132	>
133	>	/**
134	>	* Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
135	>	* Replace with a simple call to Unsafe.getUnsafe when integrating
136	>	* into a jdk.
137	>	*
138	>	* @return a sun.misc.Unsafe
139	>	*/
140	>	private static sun.misc.Unsafe getUnsafe() {
141	>	try {
142	>	return sun.misc.Unsafe.getUnsafe();
143	>	} catch (SecurityException se) {
144	>	try {
145	>	return java.security.AccessController.doPrivileged
146	>	(new java.security
147	>	.PrivilegedExceptionAction<sun.misc.Unsafe>() {
148	>	public sun.misc.Unsafe run() throws Exception {
149	>	java.lang.reflect.Field f = sun.misc
150	>	.Unsafe.class.getDeclaredField("theUnsafe");
151	>	f.setAccessible(true);
152	>	return (sun.misc.Unsafe) f.get(null);
153	>	}});
154	>	} catch (java.security.PrivilegedActionException e) {
155	>	throw new RuntimeException("Could not initialize intrinsics",
156	>	e.getCause());
157	>	}
158	>	}
159		}
160
161		private static final long serialVersionUID = -3375979862319811754L;
#	Line 264 | Line 294 | public class LinkedTransferQueue<E> exte
294		* @param e the comparison value for checking match
295		* @param mode mode
296		* @param nanos timeout value
297	<	* @return matched item, or s if cancelled
297	>	* @return matched item, or null if cancelled
298		*/
299		private E awaitFulfill(Node<E> pred, Node<E> s, E e,
300		int mode, long nanos) {
#	Line 330 | Line 360 | public class LinkedTransferQueue<E> exte
360		for (;;) {
361		Node<E> h = head.get();
362		Node<E> first = h.next;
363	<	if (first != null && first.next == first) { // help advance
363	>	if (first != null && first.get() == first) { // help advance
364		advanceHead(h, first);
365		continue;
366		}
#	Line 443 | Line 473 | public class LinkedTransferQueue<E> exte
473		}
474
475		/**
476	<	* @throws InterruptedException {@inheritDoc}
477	<	* @throws NullPointerException {@inheritDoc}
476	>	* Inserts the specified element at the tail of this queue.
477	>	* As the queue is unbounded, this method will never block.
478	>	*
479	>	* @throws NullPointerException if the specified element is null
480		*/
481	<	public void put(E e) throws InterruptedException {
482	<	if (e == null) throw new NullPointerException();
451	<	if (Thread.interrupted()) throw new InterruptedException();
452	<	xfer(e, NOWAIT, 0);
481	>	public void put(E e) {
482	>	offer(e);
483		}
484
485		/**
486	<	* @throws InterruptedException {@inheritDoc}
487	<	* @throws NullPointerException {@inheritDoc}
486	>	* Inserts the specified element at the tail of this queue.
487	>	* As the queue is unbounded, this method will never block or
488	>	* return {@code false}.
489	>	*
490	>	* @return {@code true} (as specified by
491	>	*  {@link BlockingQueue#offer(Object,long,TimeUnit) BlockingQueue.offer})
492	>	* @throws NullPointerException if the specified element is null
493		*/
494	<	public boolean offer(E e, long timeout, TimeUnit unit)
495	<	throws InterruptedException {
461	<	if (e == null) throw new NullPointerException();
462	<	if (Thread.interrupted()) throw new InterruptedException();
463	<	xfer(e, NOWAIT, 0);
464	<	return true;
494	>	public boolean offer(E e, long timeout, TimeUnit unit) {
495	>	return offer(e);
496		}
497
498		/**
499	<	* @throws NullPointerException {@inheritDoc}
499	>	* Inserts the specified element at the tail of this queue.
500	>	* As the queue is unbounded, this method will never return {@code false}.
501	>	*
502	>	* @return {@code true} (as specified by
503	>	*         {@link BlockingQueue#offer(Object) BlockingQueue.offer})
504	>	* @throws NullPointerException if the specified element is null
505		*/
506		public boolean offer(E e) {
507		if (e == null) throw new NullPointerException();
#	Line 474 | Line 510 | public class LinkedTransferQueue<E> exte
510		}
511
512		/**
513	<	* @throws NullPointerException {@inheritDoc}
513	>	* Inserts the specified element at the tail of this queue.
514	>	* As the queue is unbounded, this method will never throw
515	>	* {@link IllegalStateException} or return {@code false}.
516	>	*
517	>	* @return {@code true} (as specified by {@link Collection#add})
518	>	* @throws NullPointerException if the specified element is null
519		*/
520		public boolean add(E e) {
521	+	return offer(e);
522	+	}
523	+
524	+	/**
525	+	* Transfers the element to a waiting consumer immediately, if possible.
526	+	*
527	+	* <p>More precisely, transfers the specified element immediately
528	+	* if there exists a consumer already waiting to receive it (in
529	+	* {@link #take} or timed {@link #poll(long,TimeUnit) poll}),
530	+	* otherwise returning {@code false} without enqueuing the element.
531	+	*
532	+	* @throws NullPointerException if the specified element is null
533	+	*/
534	+	public boolean tryTransfer(E e) {
535		if (e == null) throw new NullPointerException();
536	<	xfer(e, NOWAIT, 0);
482	<	return true;
536	>	return fulfill(e) != null;
537		}
538
539		/**
540	<	* @throws InterruptedException {@inheritDoc}
541	<	* @throws NullPointerException {@inheritDoc}
540	>	* Transfers the element to a consumer, waiting if necessary to do so.
541	>	*
542	>	* <p>More precisely, transfers the specified element immediately
543	>	* if there exists a consumer already waiting to receive it (in
544	>	* {@link #take} or timed {@link #poll(long,TimeUnit) poll}),
545	>	* else inserts the specified element at the tail of this queue
546	>	* and waits until the element is received by a consumer.
547	>	*
548	>	* @throws NullPointerException if the specified element is null
549		*/
550		public void transfer(E e) throws InterruptedException {
551		if (e == null) throw new NullPointerException();
#	Line 495 | Line 556 | public class LinkedTransferQueue<E> exte
556		}
557
558		/**
559	<	* @throws InterruptedException {@inheritDoc}
560	<	* @throws NullPointerException {@inheritDoc}
559	>	* Transfers the element to a consumer if it is possible to do so
560	>	* before the timeout elapses.
561	>	*
562	>	* <p>More precisely, transfers the specified element immediately
563	>	* if there exists a consumer already waiting to receive it (in
564	>	* {@link #take} or timed {@link #poll(long,TimeUnit) poll}),
565	>	* else inserts the specified element at the tail of this queue
566	>	* and waits until the element is received by a consumer,
567	>	* returning {@code false} if the specified wait time elapses
568	>	* before the element can be transferred.
569	>	*
570	>	* @throws NullPointerException if the specified element is null
571		*/
572		public boolean tryTransfer(E e, long timeout, TimeUnit unit)
573		throws InterruptedException {
#	Line 508 | Line 579 | public class LinkedTransferQueue<E> exte
579		throw new InterruptedException();
580		}
581
511	–	/**
512	–	* @throws NullPointerException {@inheritDoc}
513	–	*/
514	–	public boolean tryTransfer(E e) {
515	–	if (e == null) throw new NullPointerException();
516	–	return fulfill(e) != null;
517	–	}
518	–
519	–	/**
520	–	* @throws InterruptedException {@inheritDoc}
521	–	*/
582		public E take() throws InterruptedException {
583	<	Object e = xfer(null, WAIT, 0);
583	>	E e = xfer(null, WAIT, 0);
584		if (e != null)
585	<	return (E) e;
585	>	return e;
586		Thread.interrupted();
587		throw new InterruptedException();
588		}
589
530	–	/**
531	–	* @throws InterruptedException {@inheritDoc}
532	–	*/
590		public E poll(long timeout, TimeUnit unit) throws InterruptedException {
591	<	Object e = xfer(null, TIMEOUT, unit.toNanos(timeout));
591	>	E e = xfer(null, TIMEOUT, unit.toNanos(timeout));
592		if (e != null || !Thread.interrupted())
593	<	return (E) e;
593	>	return e;
594		throw new InterruptedException();
595		}
596
#	Line 607 | Line 664 | public class LinkedTransferQueue<E> exte
664		}
665		}
666
667	<
667	>	/**
668	>	* Returns an iterator over the elements in this queue in proper
669	>	* sequence, from head to tail.
670	>	*
671	>	* <p>The returned iterator is a "weakly consistent" iterator that
672	>	* will never throw
673	>	* {@link ConcurrentModificationException ConcurrentModificationException},
674	>	* and guarantees to traverse elements as they existed upon
675	>	* construction of the iterator, and may (but is not guaranteed
676	>	* to) reflect any modifications subsequent to construction.
677	>	*
678	>	* @return an iterator over the elements in this queue in proper sequence
679	>	*/
680		public Iterator<E> iterator() {
681		return new Itr();
682		}
#	Line 622 | Line 691 | public class LinkedTransferQueue<E> exte
691		class Itr implements Iterator<E> {
692		Node<E> next;        // node to return next
693		Node<E> pnext;       // predecessor of next
625	–	Node<E> snext;       // successor of next
694		Node<E> curr;        // last returned node, for remove()
695		Node<E> pcurr;       // predecessor of curr, for remove()
696	<	E nextItem;        // Cache of next item, once committed to in next
696	>	E nextItem;          // Cache of next item, once committed to in next
697
698		Itr() {
699	<	findNext();
699	>	advance();
700		}
701
702		/**
703	<	* Ensures next points to next valid node, or null if none.
703	>	* Moves to next valid node and returns item to return for
704	>	* next(), or null if no such.
705		*/
706	<	void findNext() {
706	>	private E advance() {
707	>	pcurr = pnext;
708	>	curr = next;
709	>	E item = nextItem;
710	>
711		for (;;) {
712	<	Node<E> pred = pnext;
713	<	Node<E> q = next;
714	<	if (pred == null || pred == q) {
642	<	pred = traversalHead();
643	<	q = pred.next;
644	<	}
645	<	if (q == null || !q.isData) {
712	>	pnext = (next == null) ? traversalHead() : next;
713	>	next = pnext.next;
714	>	if (next == pnext) {
715		next = null;
716	<	return;
716	>	continue;  // restart
717		}
718	<	Object x = q.get();
719	<	Node<E> s = q.next;
720	<	if (x != null && q != x && q != s) {
718	>	if (next == null)
719	>	break;
720	>	Object x = next.get();
721	>	if (x != null && x != next) {
722		nextItem = (E) x;
723	<	snext = s;
654	<	pnext = pred;
655	<	next = q;
656	<	return;
723	>	break;
724		}
658	–	pnext = q;
659	–	next = s;
725		}
726	+	return item;
727		}
728
729		public boolean hasNext() {
#	Line 665 | Line 731 | public class LinkedTransferQueue<E> exte
731		}
732
733		public E next() {
734	<	if (next == null) throw new NoSuchElementException();
735	<	pcurr = pnext;
736	<	curr = next;
671	<	pnext = next;
672	<	next = snext;
673	<	E x = nextItem;
674	<	findNext();
675	<	return x;
734	>	if (next == null)
735	>	throw new NoSuchElementException();
736	>	return advance();
737		}
738
739		public void remove() {
#	Line 742 | Line 803 | public class LinkedTransferQueue<E> exte
803		* @return the number of elements in this queue
804		*/
805		public int size() {
806	<	int count = 0;
807	<	Node<E> h = traversalHead();
808	<	for (Node<E> p = h.next; p != null && p.isData; p = p.next) {
809	<	Object x = p.get();
810	<	if (x != null && x != p) {
811	<	if (++count == Integer.MAX_VALUE) // saturated
806	>	for (;;) {
807	>	int count = 0;
808	>	Node<E> pred = traversalHead();
809	>	for (;;) {
810	>	Node<E> q = pred.next;
811	>	if (q == pred) // restart
812		break;
813	+	if (q == null || !q.isData)
814	+	return count;
815	+	Object x = q.get();
816	+	if (x != null && x != q) {
817	+	if (++count == Integer.MAX_VALUE) // saturated
818	+	return count;
819	+	}
820	+	pred = q;
821		}
822		}
754	–	return count;
823		}
824
825		public int getWaitingConsumerCount() {
826	<	int count = 0;
827	<	Node<E> h = traversalHead();
828	<	for (Node<E> p = h.next; p != null && !p.isData; p = p.next) {
829	<	if (p.get() == null) {
830	<	if (++count == Integer.MAX_VALUE)
826	>	// converse of size -- count valid non-data nodes
827	>	for (;;) {
828	>	int count = 0;
829	>	Node<E> pred = traversalHead();
830	>	for (;;) {
831	>	Node<E> q = pred.next;
832	>	if (q == pred) // restart
833		break;
834	+	if (q == null || q.isData)
835	+	return count;
836	+	Object x = q.get();
837	+	if (x == null) {
838	+	if (++count == Integer.MAX_VALUE) // saturated
839	+	return count;
840	+	}
841	+	pred = q;
842		}
843		}
766	–	return count;
767	–	}
768	–
769	–	public int remainingCapacity() {
770	–	return Integer.MAX_VALUE;
844		}
845
846		public boolean remove(Object o) {
#	Line 777 | Line 850 | public class LinkedTransferQueue<E> exte
850		Node<E> pred = traversalHead();
851		for (;;) {
852		Node<E> q = pred.next;
780	–	if (q == null || !q.isData)
781	–	return false;
853		if (q == pred) // restart
854		break;
855	+	if (q == null || !q.isData)
856	+	return false;
857		Object x = q.get();
858		if (x != null && x != q && o.equals(x) &&
859		q.compareAndSet(x, q)) {
#	Line 793 | Line 866 | public class LinkedTransferQueue<E> exte
866		}
867
868		/**
869	+	* Always returns {@code Integer.MAX_VALUE} because a
870	+	* {@code LinkedTransferQueue} is not capacity constrained.
871	+	*
872	+	* @return {@code Integer.MAX_VALUE} (as specified by
873	+	*         {@link BlockingQueue#remainingCapacity()})
874	+	*/
875	+	public int remainingCapacity() {
876	+	return Integer.MAX_VALUE;
877	+	}
878	+
879	+	/**
880		* Save the state to a stream (that is, serialize it).
881		*
882		* @serialData All of the elements (each an {@code E}) in
#	Line 842 | Line 926 | public class LinkedTransferQueue<E> exte
926
927		private static final sun.misc.Unsafe UNSAFE = getUnsafe();
928		private static final long headOffset =
929	<	objectFieldOffset("head", LinkedTransferQueue.class);
929	>	objectFieldOffset(UNSAFE, "head", LinkedTransferQueue.class);
930		private static final long tailOffset =
931	<	objectFieldOffset("tail", LinkedTransferQueue.class);
931	>	objectFieldOffset(UNSAFE, "tail", LinkedTransferQueue.class);
932		private static final long cleanMeOffset =
933	<	objectFieldOffset("cleanMe", LinkedTransferQueue.class);
933	>	objectFieldOffset(UNSAFE, "cleanMe", LinkedTransferQueue.class);
934	>
935
936	<	private static long objectFieldOffset(String field, Class<?> klazz) {
936	>	static long objectFieldOffset(sun.misc.Unsafe UNSAFE,
937	>	String field, Class<?> klazz) {
938		try {
939		return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
940		} catch (NoSuchFieldException e) {

Diff Legend

–	Removed lines
+	Added lines
<	Changed lines
>	Changed lines