private void processReadQueue ( ) throws InterruptedException { ReadStatus readStatus = new ReadStatus ( ) ; while ( true ) { DataStoreRequest request = readOperationsQueue . poll ( dataStoreConfig . pollTimeoutMS ( ) , TimeUnit . MILLISECONDS ) ; while ( request != null ) { readStatus . tracker . addCall ( request , outputDataQueue ) ; readOperationsBatch . add ( request ) ; if ( readOperationsBatch . size ( ) > dataStoreConfig . processQueueMaxBatchSize ( ) ) { break ; } request = readOperationsQueue . poll ( ) ; } if ( readOperationsBatch . size ( ) > 0 ) { try { recievedReadBatch ( new ArrayList <> ( readOperationsBatch ) ) ; } finally { readOperationsBatch . clear ( ) ; } } else { flushReadsIfNeeded ( ) ; } if ( readStatus . readBatchSize . size ( ) > 1_000 ) { StatCount count ; final long now = Timer . timer ( ) . time ( ) ; count = new StatCount ( now , DataStoreSource . SERVER , Action . GET_STATS , "Thread TIME USER BaseDataStore " + Thread . currentThread ( ) . getName ( ) , Sys . threadUserTime ( ) ) ; this . outputDataQueue . put ( count ) ; count = new StatCount ( now , DataStoreSource . SERVER , Action . GET_STATS , "Thread TIME CPU BaseDataStore " + Thread . currentThread ( ) . getName ( ) , Sys . threadCPUTime ( ) ) ; this . outputDataQueue . put ( count ) ; count = new StatCount ( now , source , Action . GET , "BaseDataStore readStatus.readBatchSize.max" , readStatus . readBatchSize . max ( ) ) ; outputDataQueue . put ( count ) ; count = new StatCount ( now , source , Action . GET , "BaseDataStore readStatus.readBatchSize.min" , readStatus . readBatchSize . min ( ) ) ; outputDataQueue . put ( count ) ; count = new StatCount ( now , source , Action . GET , "BaseDataStore readStatus.readBatchSize.median" , readStatus . readBatchSize . median ( ) ) ; outputDataQueue . put ( count ) ; count = new StatCount ( now , source , Action . GET , "BaseDataStore readStatus.readBatchSize.mean" , readStatus . readBatchSize . mean ( ) ) ; outputDataQueue . put ( count ) ; count = new StatCount ( now , source , Action . GET , "BaseDataStore readStatus.readBatchSize.standardDeviation" , readStatus . readBatchSize . standardDeviation ( ) ) ; outputDataQueue . put ( count ) ; count = new StatCount ( now , source , Action . GET , "BaseDataStore readStatus.readBatchSize.variance" , readStatus . readBatchSize . variance ( ) ) ; outputDataQueue . put ( count ) ; readStatus . readBatchSize . clear ( ) ; } } }
Called from worker thread .