slumberdb/slumberdb-service-model/src/main/java/org/boon/slumberdb/stores/BaseDataStore.java
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();

            }
        }

    }
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();

            }
        }

    }
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.

@throws InterruptedException
Called from worker thread .