public final void checkSpillLimits ( ) { long currentTotal ; long currentSize ; synchronized ( this ) { currentTotal = _countTotal ; currentSize = _countTotalBytes ; } if ( ! _spilling ) { // We are not currently spilling so we need to calculate the moving average // for the number of items on our stream and then check the moving average // and the total size of the items on the stream against our configured limits. // moving total is accumulated value over last 20 calculations minus the // immediately previous value. This saves both a division and storing the // last average. To get the current running total we add the current total // to the _movingTotal _movingTotal = _movingTotal + currentTotal ; // we calculate the moving average by dividing moving total by number of // cycles in the moving window. long movingAverage = _movingTotal / MOVING_AVERAGE_LENGTH ; // we then diminish the moving total by the moving average leaving it free for next time. _movingTotal = _movingTotal - movingAverage ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( this , tc , "Stream is NOT SPILLING" ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( this , tc , "Current size :" + currentSize ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( this , tc , "Current total :" + currentTotal ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( this , tc , "Moving average:" + movingAverage ) ; if ( movingAverage >= _movingAverageHighLimit || // There are too many items on the stream currentSize >= _totalSizeHighLimit ) // The size of items on the stream is too large { _spilling = true ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( this , tc , "Stream has STARTED SPILLING" ) ; } } else { // We are spilling so we just need to check against our configured limits and // if we are below them for both number of items and total size of items then // we can stop spilling. If we do stop spilling we also need to reset our moving // average counter. if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( this , tc , "Stream is SPILLING" ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( this , tc , "Current size :" + currentSize ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( this , tc , "Current total :" + currentTotal ) ; if ( currentTotal <= _movingAverageLowLimit && // There are only a few items on the stream currentSize <= _totalSizeLowLimit ) // AND The items on the stream are small { _spilling = false ; _movingTotal = _movingAverageLowLimit * ( MOVING_AVERAGE_LENGTH - 1 ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( this , tc , "Stream has STOPPED SPILLING" ) ; } } }
Instead of just triggering spilling when we have a certain number of Items on a stream we now have the ability to trigger spilling if the total size of the Items on a stream goes over a pre - defined limit . This should allow us to control the memory usage of a stream in a more intuitive fashion .