-
Notifications
You must be signed in to change notification settings - Fork 669
/
Copy pathSynchronousQueue.java
1667 lines (1442 loc) · 70.2 KB
/
SynchronousQueue.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/*
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License version 2 only, as
* published by the Free Software Foundation. Oracle designates this
* particular file as subject to the "Classpath" exception as provided
* by Oracle in the LICENSE file that accompanied this code.
*
* This code is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
* version 2 for more details (a copy is included in the LICENSE file that
* accompanied this code).
*
* You should have received a copy of the GNU General Public License version
* 2 along with this work; if not, write to the Free Software Foundation,
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
* or visit www.oracle.com if you need additional information or have any
* questions.
*/
/*
* This file is available under and governed by the GNU General Public
* License version 2 only, as published by the Free Software Foundation.
* However, the following notice accompanied the original version of this
* file:
*
* Written by Doug Lea, Bill Scherer, and Michael Scott with
* assistance from members of JCP JSR-166 Expert Group and released to
* the public domain, as explained at
* http://creativecommons.org/publicdomain/zero/1.0/
*/
package java.util.concurrent;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.util.AbstractQueue;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.Objects;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.concurrent.locks.LockSupport;
import java.util.concurrent.locks.ReentrantLock;
/**
* A {@linkplain BlockingQueue blocking queue} in which each insert
* operation must wait for a corresponding remove operation by another
* thread, and vice versa. A synchronous queue does not have any
* internal capacity, not even a capacity of one. You cannot
* {@code peek} at a synchronous queue because an element is only
* present when you try to remove it; you cannot insert an element
* (using any method) unless another thread is trying to remove it;
* you cannot iterate as there is nothing to iterate. The
* <em>head</em> of the queue is the element that the first queued
* inserting thread is trying to add to the queue; if there is no such
* queued thread then no element is available for removal and
* {@code poll()} will return {@code null}. For purposes of other
* {@code Collection} methods (for example {@code contains}), a
* {@code SynchronousQueue} acts as an empty collection. This queue
* does not permit {@code null} elements.
*
* <p>Synchronous queues are similar to rendezvous channels used in
* CSP and Ada. They are well suited for handoff designs, in which an
* object running in one thread must sync up with an object running
* in another thread in order to hand it some information, event, or
* task.
*
* <p>This class supports an optional fairness policy for ordering
* waiting producer and consumer threads. By default, this ordering
* is not guaranteed. However, a queue constructed with fairness set
* to {@code true} grants threads access in FIFO order.
*
* <p>This class and its iterator implement all of the <em>optional</em>
* methods of the {@link Collection} and {@link Iterator} interfaces.
*
* <p>This class is a member of the
* <a href="{@docRoot}/java.base/java/util/package-summary.html#CollectionsFramework">
* Java Collections Framework</a>.
*
* @since 1.5
* @author Doug Lea and Bill Scherer and Michael Scott
* @param <E> the type of elements held in this queue
*/
/*
* 链式有界(不能无限存储)单向阻塞队列,线程安全(CAS)
*
* SynchronousQueue的侧重点是存储【操作】,换个角度讲是【交换】数据,而不是存储【数据】,
* 【同类操作】到达时,它们会被存储在该队列中,
* 【互补操作】到来时,它们之间先是“传递”/“交换”数据,随后,两个【操作】互相抵消。
*
* 使用SynchronousQueue的时候,如果一个线程的put()操作找不到互补的take()操作时,它将陷入阻塞,
* 这一点上与LinkedTransferQueue不同。
*/
public class SynchronousQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, Serializable {
/*
* This class implements extensions of the dual stack and dual
* queue algorithms described in "Nonblocking Concurrent Objects
* with Condition Synchronization", by W. N. Scherer III and
* M. L. Scott. 18th Annual Conf. on Distributed Computing,
* Oct. 2004 (see also
* http://www.cs.rochester.edu/u/scott/synchronization/pseudocode/duals.html).
* The (Lifo) stack is used for non-fair mode, and the (Fifo)
* queue for fair mode. The performance of the two is generally
* similar. Fifo usually supports higher throughput under
* contention but Lifo maintains higher thread locality in common
* applications.
*
* A dual queue (and similarly stack) is one that at any given
* time either holds "data" -- items provided by put operations,
* or "requests" -- slots representing take operations, or is
* empty. A call to "fulfill" (i.e., a call requesting an item
* from a queue holding data or vice versa) dequeues a
* complementary node. The most interesting feature of these
* queues is that any operation can figure out which mode the
* queue is in, and act accordingly without needing locks.
*
* Both the queue and stack extend abstract class Transferer
* defining the single method transfer that does a put or a
* take. These are unified into a single method because in dual
* data structures, the put and take operations are symmetrical,
* so nearly all code can be combined. The resulting transfer
* methods are on the long side, but are easier to follow than
* they would be if broken up into nearly-duplicated parts.
*
* The queue and stack data structures share many conceptual
* similarities but very few concrete details. For simplicity,
* they are kept distinct so that they can later evolve
* separately.
*
* The algorithms here differ from the versions in the above paper
* in extending them for use in synchronous queues, as well as
* dealing with cancellation. The main differences include:
*
* 1. The original algorithms used bit-marked pointers, but
* the ones here use mode bits in nodes, leading to a number
* of further adaptations.
* 2. SynchronousQueues must block threads waiting to become
* fulfilled.
* 3. Support for cancellation via timeout and interrupts,
* including cleaning out cancelled nodes/threads
* from lists to avoid garbage retention and memory depletion.
*
* Blocking is mainly accomplished using LockSupport park/unpark,
* except that nodes that appear to be the next ones to become
* fulfilled first spin a bit (on multiprocessors only). On very
* busy synchronous queues, spinning can dramatically improve
* throughput. And on less busy ones, the amount of spinning is
* small enough not to be noticeable.
*
* Cleaning is done in different ways in queues vs stacks. For
* queues, we can almost always remove a node immediately in O(1)
* time (modulo retries for consistency checks) when it is
* cancelled. But if it may be pinned as the current tail, it must
* wait until some subsequent cancellation. For stacks, we need a
* potentially O(n) traversal to be sure that we can remove the
* node, but this can run concurrently with other threads
* accessing the stack.
*
* While garbage collection takes care of most node reclamation
* issues that otherwise complicate nonblocking algorithms, care
* is taken to "forget" references to data, other nodes, and
* threads that might be held on to long-term by blocked
* threads. In cases where setting to null would otherwise
* conflict with main algorithms, this is done by changing a
* node's link to now point to the node itself. This doesn't arise
* much for Stack nodes (because blocked threads do not hang on to
* old head pointers), but references in Queue nodes must be
* aggressively forgotten to avoid reachability of everything any
* node has ever referred to since arrival.
*/
/**
* The number of times to spin before blocking in timed waits.
* The value is empirically derived -- it works well across a
* variety of processors and OSes. Empirically, the best value
* seems not to vary with number of CPUs (beyond 2) so is just
* a constant.
*/
// 带有超时标记时的自旋次数
static final int MAX_TIMED_SPINS = (Runtime.getRuntime().availableProcessors()<2) ? 0 : 32;
/**
* The number of times to spin before blocking in untimed waits.
* This is greater than timed value because untimed waits spin
* faster since they don't need to check times on each spin.
*/
// 不带有超时标记时的自旋次数
static final int MAX_UNTIMED_SPINS = MAX_TIMED_SPINS * 16;
/**
* The number of nanoseconds for which it is faster to spin rather than to use timed park. A rough estimate suffices.
*/
// 进入阻塞的最小时间阙值,当剩余阻塞时间小于这个值时,不再进入阻塞,而是使用自旋
static final long SPIN_FOR_TIMEOUT_THRESHOLD = 1000L;
/**
* The transferer. Set only in constructor, but cannot be declared
* as final without further complicating serialization. Since
* this is accessed only at most once per public method, there
* isn't a noticeable performance penalty for using volatile
* instead of final here.
*/
// 传递者
private transient volatile Transferer<E> transferer;
static {
// Reduce the risk of rare disastrous classloading in first call to
// LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773
Class<?> ensureLoaded = LockSupport.class;
}
/*▼ 构造器 ████████████████████████████████████████████████████████████████████████████████┓ */
/**
* Creates a {@code SynchronousQueue} with nonfair access policy.
*/
// 默认构造"非公平模式"的同步阻塞队列
public SynchronousQueue() {
this(false);
}
/**
* Creates a {@code SynchronousQueue} with the specified fairness policy.
*
* @param fair if true, waiting threads contend in FIFO order for
* access; otherwise the order is unspecified.
*/
// 构造同步阻塞队列,可在参数中设置"公平模式"或"非公平模式"
public SynchronousQueue(boolean fair) {
transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
}
/*▲ 构造器 ████████████████████████████████████████████████████████████████████████████████┛ */
/*▼ 入队 ████████████████████████████████████████████████████████████████████████████████┓ */
/**
* Inserts the specified element into this queue, if another thread is
* waiting to receive it.
*
* @param e the element to add
*
* @return {@code true} if the element was added to this queue, else
* {@code false}
*
* @throws NullPointerException if the specified element is null
*/
// 入队,线程安全,没有互补操作/结点时不阻塞,直接返回false
public boolean offer(E e) {
if(e == null) {
throw new NullPointerException();
}
// 需要等待超时,但超时时间为0
return transferer.transfer(e, true, 0) != null;
}
/**
* Inserts the specified element into this queue, waiting if necessary
* up to the specified wait time for another thread to receive it.
*
* @return {@code true} if successful, or {@code false} if the
* specified waiting time elapses before a consumer appears
*
* @throws InterruptedException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/
// 入队,线程安全,没有互补操作/结点时阻塞一段时间,如果在指定的时间内没有成功传递元素,则返回false
public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
if(e == null) {
throw new NullPointerException();
}
// 需要等待超时
if(transferer.transfer(e, true, unit.toNanos(timeout)) != null) {
return true;
}
if(!Thread.interrupted()) {
return false;
}
throw new InterruptedException();
}
/**
* Adds the specified element to this queue,
* waiting if necessary for another thread to receive it.
*
* @throws InterruptedException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/
// 入队,线程安全,没有互补操作/结点时,线程被阻塞
public void put(E e) throws InterruptedException {
if(e == null) {
throw new NullPointerException();
}
// 无需等待超时
if(transferer.transfer(e, false, 0) == null) {
// 清除线程的中断状态
Thread.interrupted();
throw new InterruptedException();
}
}
/*▲ 入队 ████████████████████████████████████████████████████████████████████████████████┛ */
/*▼ 出队 ████████████████████████████████████████████████████████████████████████████████┓ */
/**
* Retrieves and removes the head of this queue, if another thread
* is currently making an element available.
*
* @return the head of this queue, or {@code null} if no
* element is available
*/
// 出队,线程安全,没有互补操作/结点时不阻塞,直接返回null
public E poll() {
// 需要等待超时,但超时时间为0
return transferer.transfer(null, true, 0);
}
/**
* Retrieves and removes the head of this queue, waiting
* if necessary up to the specified wait time, for another thread
* to insert it.
*
* @return the head of this queue, or {@code null} if the
* specified waiting time elapses before an element is present
*
* @throws InterruptedException {@inheritDoc}
*/
// 出队,线程安全,没有互补操作/结点时阻塞一段时间,如果在指定的时间内没有成功传递元素,则返回null
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
// 需要等待超时
E e = transferer.transfer(null, true, unit.toNanos(timeout));
if(e != null || !Thread.interrupted()) {
return e;
}
throw new InterruptedException();
}
/**
* Retrieves and removes the head of this queue, waiting if necessary
* for another thread to insert it.
*
* @return the head of this queue
*
* @throws InterruptedException {@inheritDoc}
*/
// 出队,线程安全,没有互补操作/结点时,线程被阻塞
public E take() throws InterruptedException {
// 无需等待超时
E e = transferer.transfer(null, false, 0);
if(e != null) {
return e;
}
Thread.interrupted();
throw new InterruptedException();
}
/**
* Always returns {@code false}.
* A {@code SynchronousQueue} has no internal capacity.
*
* @param o the element to remove
*
* @return {@code false}
*/
// 总是返回false,该结构不用于存储数据,所以没有可被移除的数据
public boolean remove(Object o) {
return false;
}
/**
* Always returns {@code false}.
* A {@code SynchronousQueue} has no internal capacity.
*
* @param c the collection
*
* @return {@code false}
*/
// 总是返回false,该结构不用于存储数据,所以没有可被移除的数据
public boolean removeAll(Collection<?> c) {
return false;
}
/**
* Always returns {@code false}.
* A {@code SynchronousQueue} has no internal capacity.
*
* @param c the collection
*
* @return {@code false}
*/
// 总是返回false,该结构不用于存储数据,所以没有可保留的数据
public boolean retainAll(Collection<?> c) {
return false;
}
/**
* Does nothing.
* A {@code SynchronousQueue} has no internal capacity.
*/
// 清空,这里是空实现,该结构不用于存储数据,所以无需清理
public void clear() {
}
/**
* @throws UnsupportedOperationException {@inheritDoc}
* @throws ClassCastException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
* @throws IllegalArgumentException {@inheritDoc}
*/
// 取出之前所有阻塞的"存"操作中的数据,存入给定的容器
public int drainTo(Collection<? super E> c) {
Objects.requireNonNull(c);
if(c == this) {
throw new IllegalArgumentException();
}
int n = 0;
for(E e; (e = poll()) != null; n++) {
c.add(e);
}
return n;
}
/**
* @throws UnsupportedOperationException {@inheritDoc}
* @throws ClassCastException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
* @throws IllegalArgumentException {@inheritDoc}
*/
// 取出之前阻塞的"存"操作中的数据,存入给定的容器,最多取maxElements个数据
public int drainTo(Collection<? super E> c, int maxElements) {
Objects.requireNonNull(c);
if(c == this) {
throw new IllegalArgumentException();
}
int n = 0;
for(E e; n<maxElements && (e = poll()) != null; n++) {
c.add(e);
}
return n;
}
/*▲ 出队 ████████████████████████████████████████████████████████████████████████████████┛ */
/*▼ 取值 ████████████████████████████████████████████████████████████████████████████████┓ */
/**
* Always returns {@code null}.
* A {@code SynchronousQueue} does not return elements
* unless actively waited on.
*
* @return {@code null}
*/
// 总是返回null,代表该结构不用于保存数据
public E peek() {
return null;
}
/*▲ 取值 ████████████████████████████████████████████████████████████████████████████████┛ */
/*▼ 包含查询 ████████████████████████████████████████████████████████████████████████████████┓ */
/**
* Always returns {@code false}.
* A {@code SynchronousQueue} has no internal capacity.
*
* @param o the element
*
* @return {@code false}
*/
// 总是返回false,代表该结构不用于保存数据
public boolean contains(Object o) {
return false;
}
/**
* Returns {@code false} unless the given collection is empty.
* A {@code SynchronousQueue} has no internal capacity.
*
* @param c the collection
*
* @return {@code false} unless given collection is empty
*/
// 返回值表示给定的容器是否为null,与同步阻塞队列自身的结构无关
public boolean containsAll(Collection<?> c) {
return c.isEmpty();
}
/*▲ 包含查询 ████████████████████████████████████████████████████████████████████████████████┛ */
/*▼ 视图 ████████████████████████████████████████████████████████████████████████████████┓ */
/**
* Returns a zero-length array.
*
* @return a zero-length array
*/
// 返回一个无容量的数组
public Object[] toArray() {
return new Object[0];
}
/**
* Sets the zeroth element of the specified array to {@code null}
* (if the array has non-zero length) and returns it.
*
* @param a the array
*
* @return the specified array
*
* @throws NullPointerException if the specified array is null
*/
// 返回只包含一个null元素的数组
public <T> T[] toArray(T[] a) {
if(a.length>0) {
a[0] = null;
}
return a;
}
/*▲ 视图 ████████████████████████████████████████████████████████████████████████████████┛ */
/*▼ 迭代 ████████████████████████████████████████████████████████████████████████████████┓ */
/**
* Returns an empty iterator in which {@code hasNext} always returns
* {@code false}.
*
* @return an empty iterator
*/
// 返回一个空的迭代器
public Iterator<E> iterator() {
return Collections.emptyIterator();
}
/**
* Returns an empty spliterator in which calls to
* {@link Spliterator#trySplit() trySplit} always return {@code null}.
*
* @return an empty spliterator
*
* @since 1.8
*/
// 返回一个空的迭代器
public Spliterator<E> spliterator() {
return Spliterators.emptySpliterator();
}
/*▲ 迭代 ████████████████████████████████████████████████████████████████████████████████┛ */
/*▼ 杂项 ████████████████████████████████████████████████████████████████████████████████┓ */
/**
* Always returns zero.
* A {@code SynchronousQueue} has no internal capacity.
*
* @return zero
*/
// 总是返回0,因为该结构不用于保存数据
public int size() {
return 0;
}
/**
* Always returns zero.
* A {@code SynchronousQueue} has no internal capacity.
*
* @return zero
*/
// 总是返回0,因为该结构不用于保存数据
public int remainingCapacity() {
return 0;
}
/**
* Always returns {@code true}.
* A {@code SynchronousQueue} has no internal capacity.
*
* @return {@code true}
*/
// 总是返回true,因为该结构不用于保存数据
public boolean isEmpty() {
return true;
}
/*▲ 杂项 ████████████████████████████████████████████████████████████████████████████████┛ */
/*▼ 序列化 ████████████████████████████████████████████████████████████████████████████████┓ */
/*
* To cope with serialization strategy in the 1.5 version of
* SynchronousQueue, we declare some unused classes and fields
* that exist solely to enable serializability across versions.
* These fields are never used, so are initialized only if this
* object is ever serialized or deserialized.
*/
private static final long serialVersionUID = -3223113410248163686L;
private ReentrantLock qlock;
private WaitQueue waitingProducers;
private WaitQueue waitingConsumers;
/**
* Saves this queue to a stream (that is, serializes it).
*
* @param s the stream
*
* @throws java.io.IOException if an I/O error occurs
*/
private void writeObject(ObjectOutputStream s) throws java.io.IOException {
boolean fair = transferer instanceof TransferQueue;
if(fair) {
qlock = new ReentrantLock(true);
waitingProducers = new FifoWaitQueue();
waitingConsumers = new FifoWaitQueue();
} else {
qlock = new ReentrantLock();
waitingProducers = new LifoWaitQueue();
waitingConsumers = new LifoWaitQueue();
}
s.defaultWriteObject();
}
/**
* Reconstitutes this queue from a stream (that is, deserializes it).
*
* @param s the stream
*
* @throws ClassNotFoundException if the class of a serialized object
* could not be found
* @throws java.io.IOException if an I/O error occurs
*/
private void readObject(ObjectInputStream s) throws java.io.IOException, ClassNotFoundException {
s.defaultReadObject();
if(waitingProducers instanceof FifoWaitQueue)
transferer = new TransferQueue<E>();
else
transferer = new TransferStack<E>();
}
@SuppressWarnings("serial")
static class WaitQueue implements Serializable {
}
static class LifoWaitQueue extends WaitQueue {
private static final long serialVersionUID = -3633113410248163686L;
}
static class FifoWaitQueue extends WaitQueue {
private static final long serialVersionUID = -3623113410248163686L;
}
/*▲ 序列化 ████████████████████████████████████████████████████████████████████████████████┛ */
/**
* Always returns {@code "[]"}.
*
* @return {@code "[]"}
*/
public String toString() {
return "[]";
}
/**
* Shared internal API for dual stacks and queues.
*/
// 数据"传递"接口,用于不同类型操作之间传递数据
abstract static class Transferer<E> {
/**
* Performs a put or take.
*
* @param e if non-null, the item to be handed to a consumer;
* if null, requests that transfer return an item
* offered by producer.
* @param timed if this operation should timeout
* @param nanos the timeout, in nanoseconds
* @return if non-null, the item provided or received; if null,
* the operation failed due to timeout or interrupt --
* the caller can distinguish which of these occurred
* by checking Thread.interrupted.
*/
// "传递"数据,匹配/抵消操作
abstract E transfer(E e, boolean timed, long nanos);
}
/** Dual Queue */
/*
* "公平模式"的同步阻塞队列的实现
*
* 使用链队来模拟,这意味着先来的操作先被执行
* 注:该结构会在队头之前设置一个空结点,以标记队列为null
*/
static final class TransferQueue<E> extends Transferer<E> {
/*
* This extends Scherer-Scott dual queue algorithm, differing,
* among other ways, by using modes within nodes rather than
* marked pointers. The algorithm is a little simpler than
* that for stacks because fulfillers do not need explicit
* nodes, and matching is done by CAS'ing QNode.item field
* from non-null to null (for put) or vice versa (for take).
*/
/** Head of queue */
// 队头
transient volatile QNode head;
/** Tail of queue */
// 队尾
transient volatile QNode tail;
/**
* Reference to a cancelled node that might not yet have been
* unlinked from queue because it was the last inserted node
* when it was cancelled.
*/
transient volatile QNode cleanMe;
// VarHandle mechanics
private static final VarHandle QHEAD;
private static final VarHandle QTAIL;
private static final VarHandle QCLEANME;
static {
try {
MethodHandles.Lookup l = MethodHandles.lookup();
QHEAD = l.findVarHandle(TransferQueue.class, "head", QNode.class);
QTAIL = l.findVarHandle(TransferQueue.class, "tail", QNode.class);
QCLEANME = l.findVarHandle(TransferQueue.class, "cleanMe", QNode.class);
} catch(ReflectiveOperationException e) {
throw new ExceptionInInitializerError(e);
}
}
TransferQueue() {
QNode h = new QNode(null, false); // initialize to dummy node.
head = h;
tail = h;
}
/**
* Puts or takes an item.
*/
// "传递"数据,匹配/抵消操作
@SuppressWarnings("unchecked")
E transfer(E e, boolean timed, long nanos) {
/* Basic algorithm is to loop trying to take either of two actions:
*
* 1. If queue apparently empty or holding same-mode nodes,
* try to add node to queue of waiters, wait to be
* fulfilled (or cancelled) and return matching item.
*
* 2. If queue apparently contains waiting items, and this
* call is of complementary mode, try to fulfill by CAS'ing
* item field of waiting node and dequeuing it, and then
* returning matching item.
*
* In each case, along the way, check for and try to help
* advance head and tail on behalf of other stalled/slow
* threads.
*
* The loop starts off with a null check guarding against
* seeing uninitialized head or tail values. This never
* happens in current SynchronousQueue, but could if
* callers held non-volatile/final ref to the
* transferer. The check is here anyway because it places
* null checks at top of loop, which is usually faster
* than having them implicitly interspersed.
*/
QNode s = null; // constructed/reused as needed
// 当前结点是否为数据结点。当遇到"存"操作时,该结点成为数据结点
boolean isData = (e != null);
for(; ; ) {
// 每次进入循环都要重置h和t
QNode t = tail;
QNode h = head;
// saw uninitialized value
if(t == null || h == null) {
// spin
continue;
}
// 队列为空,或者队尾结点与当前结点的数据模式一致
if(h == t || t.isData == isData) { // empty or same-mode
// 指向队尾之后的结点
QNode tn = t.next;
// 队尾已被别的线程修改
if(t != tail) { // inconsistent read
continue;
}
// 队尾处于滞后状态,则先更新队尾
if(tn != null) { // lagging tail
// 原子地将队尾更新为tn
advanceTail(t, tn);
continue;
}
// 需要等待,但超时了,不必再等
if(timed && nanos<=0L) { // can't wait
return null;
}
// 首次到达这里,构建新结点
if(s == null) {
s = new QNode(e, isData);
}
// 队尾没有滞后时,原子地更新队尾的next域为s
if(!t.casNext(null, s)) { // failed to link in
continue;
}
// 原子地将队尾更新为s
advanceTail(t, s); // swing tail and wait
// 尝试阻塞当前线程(操作),直到匹配的操作到来后唤醒它,返回数据域
Object x = awaitFulfill(s, e, timed, nanos);
// 如果结点已经被标记为取消
if(x == s) { // wait was cancelled
clean(t, s);
return null;
}
// 如果s.next!=s,即s仍在队列中
if(!s.isOffList()) { // not already unlinked
advanceHead(t, s); // unlink if head
// 遗忘数据域,跟取消的效果一样
if(x != null) { // and forget fields
s.item = s;
}
s.waiter = null;
}
return (x != null) ? (E) x : e;
} else { // complementary-mode
/*
* 至此:h!=t && t.isData!=isData
* 说明队列不为null,且队尾结点与当前结点的数据模式不一致
* 这时候该进行匹配操作了(进入互补模式)
*/
// 取出队头结点
QNode m = h.next; // node to fulfill
// 确保队列不为null,且头尾结点未被修改
if(t != tail || m == null || h != head) {
continue; // inconsistent read
}
// 取出队头数据(这里可能出现线程争用)
Object x = m.item;
if(isData == (x != null) || // m already fulfilled(该结点已被其它线程处理了)
x == m || // m cancelled(该结点已经被取消了)
!m.casItem(x, e)) { // lost CAS(更新数据域,以便awaitFulfill()方法被唤醒后退出)
/* 线程争用失败的线程到这里 */
/*
* 至此,说明队头结点已经被处理/取消,则可以跳过它了
* (也可能是线程争用失败了,此时起到加速作用)
*/
advanceHead(h, m); // dequeue and retry
continue;
}
/* 线程争用成功的线程到这里 */
// 原子地将队头更新为m
advanceHead(h, m); // successfully fulfilled
// 唤醒阻塞的操作/结点/线程
LockSupport.unpark(m.waiter);
// 返回数据
return (x != null) ? (E) x : e;
}
}
}
/**
* Spins/blocks until node s is fulfilled.
*
* @param s the waiting node
* @param e the comparison value for checking match
* @param timed true if timed wait
* @param nanos timeout value
*
* @return matched item, or s if cancelled
*/
// 尝试阻塞当前线程(操作),直到匹配的操作到来后唤醒它,返回数据域
Object awaitFulfill(QNode s, E e, boolean timed, long nanos) {
/* Same idea as TransferStack.awaitFulfill */
// 计算截止时间
final long deadline = timed ? System.nanoTime() + nanos : 0L;
Thread w = Thread.currentThread();
// 设置自旋次数
int spins = (head.next == s) ? (timed ? MAX_TIMED_SPINS : MAX_UNTIMED_SPINS) : 0;
for(; ; ) {
// 如果当前线程带有中断标记,则取消操作
if(w.isInterrupted()) {
s.tryCancel(e);
}
// 操作已经被取消,或者该结点已被处理(数据域会发生变化)
Object x = s.item;
if(x != e) {
return x;
}
// 如果需要等待
if(timed) {
// 计算等待时长
nanos = deadline - System.nanoTime();
// 如果等待超时,则取消该操作
if(nanos<=0L) {
s.tryCancel(e);
continue;
}
}
// 如果需要自旋,自旋次数减一
if(spins>0) {
--spins;
// 如果需要自旋,则进入"忙等待"状态
Thread.onSpinWait();
} else if(s.waiter == null) {
// 记录将要被阻塞线程
s.waiter = w;
} else if(!timed) {
// 阻塞当前线程,并用当前对象本身作为阻塞标记
LockSupport.park(this);
} else if(nanos>SPIN_FOR_TIMEOUT_THRESHOLD) {
// 带超时的阻塞
LockSupport.parkNanos(this, nanos);
}
}
}