@@ -287,17 +287,17 @@ private static final class SpscBlockingQueue<T> implements BlockingQueue<T> {
287
287
getLong ("io.servicetalk.concurrent.internal.blockingIterableYieldNs" , 1024 );
288
288
289
289
@ SuppressWarnings ("rawtypes" )
290
- private static final AtomicLongFieldUpdater <SpscBlockingQueue > pcIndexUpdater =
291
- AtomicLongFieldUpdater .newUpdater (SpscBlockingQueue .class , "pcIndex " );
290
+ private static final AtomicLongFieldUpdater <SpscBlockingQueue > producerConsumerIndexUpdater =
291
+ AtomicLongFieldUpdater .newUpdater (SpscBlockingQueue .class , "producerConsumerIndex " );
292
292
private final Queue <T > spscQueue ;
293
293
@ Nullable
294
294
private Thread consumerThread ;
295
295
/**
296
- * high 32 bits == producer index (see {@link #pIndex (long)})
297
- * low 32 bits == consumer index (see {@link #cIndex (long)}}
298
- * @see #computePCIndex (int, int)
296
+ * high 32 bits == producer index (see {@link #producerIndex (long)})
297
+ * low 32 bits == consumer index (see {@link #consumerIndex (long)}}
298
+ * @see #combineIndexes (int, int)
299
299
*/
300
- private volatile long pcIndex ;
300
+ private volatile long producerConsumerIndex ;
301
301
302
302
SpscBlockingQueue (Queue <T > spscQueue ) {
303
303
this .spscQueue = requireNonNull (spscQueue );
@@ -513,10 +513,11 @@ public String toString() {
513
513
514
514
private void producerSignalAdded () {
515
515
for (;;) {
516
- final long currIndex = pcIndex ;
517
- final int producer = pIndex (currIndex );
518
- final int consumer = cIndex (currIndex );
519
- if (pcIndexUpdater .compareAndSet (this , currIndex , computePCIndex (producer + 1 , consumer ))) {
516
+ final long currIndex = producerConsumerIndex ;
517
+ final int producer = producerIndex (currIndex );
518
+ final int consumer = consumerIndex (currIndex );
519
+ if (producerConsumerIndexUpdater .compareAndSet (this , currIndex ,
520
+ combineIndexes (producer + 1 , consumer ))) {
520
521
if (producer - consumer <= 0 && consumerThread != null ) {
521
522
final Thread wakeThread = consumerThread ;
522
523
consumerThread = null ;
@@ -530,20 +531,22 @@ private void producerSignalAdded() {
530
531
private T take0 (BiLongFunction <TimeUnit , T > taker , long timeout , TimeUnit unit ) throws InterruptedException {
531
532
final Thread currentThread = Thread .currentThread ();
532
533
for (;;) {
533
- long currIndex = pcIndex ;
534
- final int producer = pIndex (currIndex );
535
- final int consumer = cIndex (currIndex );
534
+ long currIndex = producerConsumerIndex ;
535
+ final int producer = producerIndex (currIndex );
536
+ final int consumer = consumerIndex (currIndex );
536
537
if (producer == consumer ) {
537
538
// Set consumerThread before pcIndex, to establish happens-before with producer thread.
538
539
consumerThread = currentThread ;
539
- if (pcIndexUpdater .compareAndSet (this , currIndex , computePCIndex (producer , consumer + 1 ))) {
540
+ if (producerConsumerIndexUpdater .compareAndSet (this , currIndex ,
541
+ combineIndexes (producer , consumer + 1 ))) {
540
542
return taker .apply (timeout , unit );
541
543
}
542
544
} else {
543
545
final T item = spscQueue .poll ();
544
546
if (item != null ) {
545
- while (!pcIndexUpdater .compareAndSet (this , currIndex , computePCIndex (producer , consumer + 1 ))) {
546
- currIndex = pcIndex ;
547
+ while (!producerConsumerIndexUpdater .compareAndSet (this , currIndex ,
548
+ combineIndexes (producer , consumer + 1 ))) {
549
+ currIndex = producerConsumerIndex ;
547
550
}
548
551
return item ;
549
552
}
@@ -555,10 +558,11 @@ private T take0(BiLongFunction<TimeUnit, T> taker, long timeout, TimeUnit unit)
555
558
556
559
private void consumerSignalRemoved (final int i ) {
557
560
for (;;) {
558
- final long currIndex = pcIndex ;
559
- final int producer = pIndex (currIndex );
560
- final int consumer = cIndex (currIndex );
561
- if (pcIndexUpdater .compareAndSet (this , currIndex , computePCIndex (producer , consumer + i ))) {
561
+ final long currIndex = producerConsumerIndex ;
562
+ final int producer = producerIndex (currIndex );
563
+ final int consumer = consumerIndex (currIndex );
564
+ if (producerConsumerIndexUpdater .compareAndSet (this , currIndex ,
565
+ combineIndexes (producer , consumer + i ))) {
562
566
break ;
563
567
}
564
568
}
@@ -616,16 +620,16 @@ private static void checkInterrupted() throws InterruptedException {
616
620
}
617
621
}
618
622
619
- private static long computePCIndex (int producer , int consumer ) {
623
+ private static long combineIndexes (int producer , int consumer ) {
620
624
return ((long ) producer << 32 ) | consumer ;
621
625
}
622
626
623
- private static int cIndex (long pcIndex ) {
624
- return (int ) pcIndex ;
627
+ private static int consumerIndex (long producerConsumerIndex ) {
628
+ return (int ) producerConsumerIndex ;
625
629
}
626
630
627
- private static int pIndex (long pcIndex ) {
628
- return (int ) (pcIndex >>> 32 );
631
+ private static int producerIndex (long producerConsumerIndex ) {
632
+ return (int ) (producerConsumerIndex >>> 32 );
629
633
}
630
634
631
635
private interface BiLongFunction <T , R > {
0 commit comments