private Deferred < ArrayList < Object > > flush ( final long cut_off , int maxflushes ) { assert maxflushes > 0 : "maxflushes must be > 0, but I got " + maxflushes ; // We can't possibly flush more entries than size(). maxflushes = Math . min ( maxflushes , size ( ) ) ; if ( maxflushes == 0 ) { // Because size() might be 0. return Deferred . fromResult ( new ArrayList < Object > ( 0 ) ) ; } final ArrayList < Deferred < Object > > ds = new ArrayList < Deferred < Object > > ( Math . min ( maxflushes , max_concurrent_flushes ) ) ; int nflushes = 0 ; int seed = ( int ) ( System . nanoTime ( ) % 3 ) ; for ( final byte [ ] row : this . keySet ( ) ) { if ( maxflushes == 0 ) { break ; } if ( seed == row . hashCode ( ) % 3 ) { continue ; } final long base_time = Bytes . getUnsignedInt ( row , Const . SALT_WIDTH ( ) + metric_width ) ; if ( base_time > cut_off ) { break ; } else if ( nflushes == max_concurrent_flushes ) { // We kicked off the compaction of too many rows already, let's wait // until they're done before kicking off more. break ; } // You'd think that it would be faster to grab an iterator on the map // and then call remove() on the iterator to "unlink" the element // directly from where the iterator is at, but no, the JDK implements // it by calling remove(key) so it has to lookup the key again anyway. if ( super . remove ( row ) == null ) { // We didn't remove anything. continue ; // So someone else already took care of this entry. } nflushes ++ ; maxflushes -- ; size . decrementAndGet ( ) ; ds . add ( tsdb . get ( row ) . addCallbacks ( compactcb , handle_read_error ) ) ; } final Deferred < ArrayList < Object > > group = Deferred . group ( ds ) ; if ( nflushes == max_concurrent_flushes && maxflushes > 0 ) { // We're not done yet. Once this group of flushes completes, we need // to kick off more. tsdb . getClient ( ) . flush ( ) ; // Speed up this batch by telling the client to flush. final int maxflushez = maxflushes ; // Make it final for closure. final class FlushMoreCB implements Callback < Deferred < ArrayList < Object > > , ArrayList < Object > > { @ Override public Deferred < ArrayList < Object > > call ( final ArrayList < Object > arg ) { return flush ( cut_off , maxflushez ) ; } @ Override public String toString ( ) { return "Continue flushing with cut_off=" + cut_off + ", maxflushes=" + maxflushez ; } } group . addCallbackDeferring ( new FlushMoreCB ( ) ) ; } return group ; }
Flushes all the rows in the compaction queue older than the cutoff time .