processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/StreamingMergeSortedGrouper.java
/**
   * Wait for {@link #nextReadIndex} to be moved if necessary and move {@link #curWriteIndex}.
   */
Wait for  #nextReadIndex to be moved if necessary and move  #curWriteIndex.
Wait for #nextReadIndex to be moved if necessary and move #curWriteIndex .
private void increaseWriteIndex()
  {
    final long startAtNs = System.nanoTime();
    final long queryTimeoutAtNs = getQueryTimeoutAtNs(startAtNs);
    final long spinTimeoutAtNs = startAtNs + SPIN_FOR_TIMEOUT_THRESHOLD_NS;
    long timeoutNs = queryTimeoutAtNs - startAtNs;
    long spinTimeoutNs = SPIN_FOR_TIMEOUT_THRESHOLD_NS;
     nextReadIndex is a volatile variable and the changes on it are continuously checked until they are seen in
     the while loop. See the following links.
     * http:docs.oracle.com/javase/specs/jls/se7/html/jls-8.html#jls-8.3.1.4
     * http:docs.oracle.com/javase/specs/jls/se7/html/jls-17.html#jls-17.4.5
     * https:stackoverflow.com/questions/11761552/detailed-semantics-of-volatile-regarding-timeliness-of-visibility
    if (curWriteIndex == maxNumSlots - 1) {
       We additionally check that nextReadIndex is -1 here because the writing thread should wait for the reading
       thread to start reading only when the writing thread tries to overwrite the first slot for the first time.
       The below condition is checked in a while loop instead of using a lock to avoid frequent thread park.
      while ((nextReadIndex == -1 || nextReadIndex == 0) && !Thread.currentThread().isInterrupted()) {
        if (timeoutNs <= 0L) {
          throw new QueryTimeoutException();
        }
         Thread.yield() should not be called from the very beginning
        if (spinTimeoutNs <= 0L) {
          Thread.yield();
        }
        long now = System.nanoTime();
        timeoutNs = queryTimeoutAtNs - now;
        spinTimeoutNs = spinTimeoutAtNs - now;
      }
       Changes on nextReadIndex happens-before changing curWriteIndex.
      curWriteIndex = 0;
    } else {
      final int nextWriteIndex = curWriteIndex + 1;
       The below condition is checked in a while loop instead of using a lock to avoid frequent thread park.
      while ((nextWriteIndex == nextReadIndex) && !Thread.currentThread().isInterrupted()) {
        if (timeoutNs <= 0L) {
          throw new QueryTimeoutException();
        }
         Thread.yield() should not be called from the very beginning
        if (spinTimeoutNs <= 0L) {
          Thread.yield();
        }
        long now = System.nanoTime();
        timeoutNs = queryTimeoutAtNs - now;
        spinTimeoutNs = spinTimeoutAtNs - now;
      }
       Changes on nextReadIndex happens-before changing curWriteIndex.
      curWriteIndex = nextWriteIndex;
    }
  }
private void increaseWriteIndex ( ) { final long startAtNs = System . nanoTime ( ) ; final long queryTimeoutAtNs = getQueryTimeoutAtNs ( startAtNs ) ; final long spinTimeoutAtNs = startAtNs + SPIN_FOR_TIMEOUT_THRESHOLD_NS ; long timeoutNs = queryTimeoutAtNs - startAtNs ; long spinTimeoutNs = SPIN_FOR_TIMEOUT_THRESHOLD_NS ; if ( curWriteIndex == maxNumSlots - 1 ) { while ( ( nextReadIndex == - 1 || nextReadIndex == 0 ) && ! Thread . currentThread ( ) . isInterrupted ( ) ) { if ( timeoutNs <= 0L ) { throw new QueryTimeoutException ( ) ; } if ( spinTimeoutNs <= 0L ) { Thread . yield ( ) ; } long now = System . nanoTime ( ) ; timeoutNs = queryTimeoutAtNs - now ; spinTimeoutNs = spinTimeoutAtNs - now ; } curWriteIndex = 0 ; } else { final int nextWriteIndex = curWriteIndex + 1 ; while ( ( nextWriteIndex == nextReadIndex ) && ! Thread . currentThread ( ) . isInterrupted ( ) ) { if ( timeoutNs <= 0L ) { throw new QueryTimeoutException ( ) ; } if ( spinTimeoutNs <= 0L ) { Thread . yield ( ) ; } long now = System . nanoTime ( ) ; timeoutNs = queryTimeoutAtNs - now ; spinTimeoutNs = spinTimeoutAtNs - now ; } curWriteIndex = nextWriteIndex ; } }
Wait for {@link #nextReadIndex} to be moved if necessary and move {@link #curWriteIndex}.
Wait for { @link #nextReadIndex } to be moved if necessary and move { @link #curWriteIndex } .
// In the below, we check that the array is full and wait for at least one slot to become available. // // nextReadIndex is a volatile variable and the changes on it are continuously checked until they are seen in // the while loop. See the following links. // * http://docs.oracle.com/javase/specs/jls/se7/html/jls-8.html#jls-8.3.1.4 // * http://docs.oracle.com/javase/specs/jls/se7/html/jls-17.html#jls-17.4.5 // * https://stackoverflow.com/questions/11761552/detailed-semantics-of-volatile-regarding-timeliness-of-visibility // We additionally check that nextReadIndex is -1 here because the writing thread should wait for the reading // thread to start reading only when the writing thread tries to overwrite the first slot for the first time. // The below condition is checked in a while loop instead of using a lock to avoid frequent thread park. // Thread.yield() should not be called from the very beginning // Changes on nextReadIndex happens-before changing curWriteIndex. // The below condition is checked in a while loop instead of using a lock to avoid frequent thread park. // Thread.yield() should not be called from the very beginning // Changes on nextReadIndex happens-before changing curWriteIndex.