processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/StreamingMergeSortedGrouper.java
/**
       * Wait for {@link StreamingMergeSortedGrouper#curWriteIndex} to be moved if necessary and move
       * {@link StreamingMergeSortedGrouper#nextReadIndex}.
       *
       * @param target the target index {@link StreamingMergeSortedGrouper#nextReadIndex} will move to
       */


@param target the target index  StreamingMergeSortedGrouper#nextReadIndex will move to
@param target the target index StreamingMergeSortedGrouper#nextReadIndex will move to
private void increaseReadIndexTo(int target)
      {
         curWriteIndex 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
        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;
         The below condition is checked in a while loop instead of using a lock to avoid frequent thread park.
        while ((curWriteIndex == -1 || target == curWriteIndex) &&
               !finished && !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 curWriteIndex happens-before changing nextReadIndex.
        nextReadIndex = target;
      }
private void increaseReadIndexTo ( int target ) { 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 ; while ( ( curWriteIndex == - 1 || target == curWriteIndex ) && ! finished && ! Thread . currentThread ( ) . isInterrupted ( ) ) { if ( timeoutNs <= 0L ) { throw new QueryTimeoutException ( ) ; } if ( spinTimeoutNs <= 0L ) { Thread . yield ( ) ; } long now = System . nanoTime ( ) ; timeoutNs = queryTimeoutAtNs - now ; spinTimeoutNs = spinTimeoutAtNs - now ; } nextReadIndex = target ; }
Wait for {@link StreamingMergeSortedGrouper#curWriteIndex} to be moved if necessary and move
 {@link StreamingMergeSortedGrouper#nextReadIndex}.
Wait for { @link StreamingMergeSortedGrouper#curWriteIndex } to be moved if necessary and move { @link StreamingMergeSortedGrouper#nextReadIndex } .
// Check that the array is empty and wait for at least one slot to become available. // // curWriteIndex 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 // 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 curWriteIndex happens-before changing nextReadIndex.