public Iterable < V > sorted ( Iterator < V > input ) { ExecutorService executor = new ThreadPoolExecutor ( this . numThreads , this . numThreads , 1000L , TimeUnit . MILLISECONDS , new SynchronousQueue < Runnable > ( ) , new CallerRunsPolicy ( ) ) ; final AtomicInteger count = new AtomicInteger ( 0 ) ; final List < File > tempFiles = Collections . synchronizedList ( new ArrayList < File > ( ) ) ; while ( input . hasNext ( ) ) { final int segmentId = count . getAndIncrement ( ) ; final long segmentStartMs = System . currentTimeMillis ( ) ; logger . info ( "Segment " + segmentId + ": filling sort buffer for segment..." ) ; @ SuppressWarnings ( "unchecked" ) final V [ ] buffer = ( V [ ] ) new Object [ internalSortSize ] ; int segmentSizeIter = 0 ; for ( ; segmentSizeIter < internalSortSize && input . hasNext ( ) ; segmentSizeIter ++ ) buffer [ segmentSizeIter ] = input . next ( ) ; final int segmentSize = segmentSizeIter ; logger . info ( "Segment " + segmentId + ": sort buffer filled...adding to sort queue." ) ; // sort and write out asynchronously executor . execute ( new Runnable ( ) { public void run ( ) { logger . info ( "Segment " + segmentId + ": sorting buffer." ) ; long start = System . currentTimeMillis ( ) ; Arrays . sort ( buffer , 0 , segmentSize , comparator ) ; long elapsed = System . currentTimeMillis ( ) - start ; logger . info ( "Segment " + segmentId + ": sort completed in " + elapsed + " ms, writing to temp file." ) ; // write out values to a temp file try { File tempFile = File . createTempFile ( "segment-" , ".dat" , tempDir ) ; tempFile . deleteOnExit ( ) ; tempFiles . add ( tempFile ) ; OutputStream os = new BufferedOutputStream ( new FileOutputStream ( tempFile ) , bufferSize ) ; if ( gzip ) os = new GZIPOutputStream ( os ) ; DataOutputStream output = new DataOutputStream ( os ) ; for ( int i = 0 ; i < segmentSize ; i ++ ) writeValue ( output , buffer [ i ] ) ; output . close ( ) ; } catch ( IOException e ) { throw new VoldemortException ( e ) ; } long segmentElapsed = System . currentTimeMillis ( ) - segmentStartMs ; logger . info ( "Segment " + segmentId + ": completed processing of segment in " + segmentElapsed + " ms." ) ; } } ) ; } // wait for all sorting to complete executor . shutdown ( ) ; try { executor . awaitTermination ( Long . MAX_VALUE , TimeUnit . MILLISECONDS ) ; // create iterator over sorted values return new DefaultIterable < V > ( new ExternalSorterIterator ( tempFiles , bufferSize / tempFiles . size ( ) ) ) ; } catch ( InterruptedException e ) { throw new RuntimeException ( e ) ; } }
Produce an iterator over the input values in sorted order . Sorting will occur in the fixed space configured in the constructor data will be dumped to disk as necessary .