private void redistributeBuffers ( ) throws IOException { assert Thread . holdsLock ( factoryLock ) ; // All buffers, which are not among the required ones final int numAvailableMemorySegment = totalNumberOfMemorySegments - numTotalRequiredBuffers ; if ( numAvailableMemorySegment == 0 ) { // in this case, we need to redistribute buffers so that every pool gets its minimum for ( LocalBufferPool bufferPool : allBufferPools ) { bufferPool . setNumBuffers ( bufferPool . getNumberOfRequiredMemorySegments ( ) ) ; } return ; } /* * With buffer pools being potentially limited, let's distribute the available memory * segments based on the capacity of each buffer pool, i.e. the maximum number of segments * an unlimited buffer pool can take is numAvailableMemorySegment, for limited buffer pools * it may be less. Based on this and the sum of all these values (totalCapacity), we build * a ratio that we use to distribute the buffers. */ long totalCapacity = 0 ; // long to avoid int overflow for ( LocalBufferPool bufferPool : allBufferPools ) { int excessMax = bufferPool . getMaxNumberOfMemorySegments ( ) - bufferPool . getNumberOfRequiredMemorySegments ( ) ; totalCapacity += Math . min ( numAvailableMemorySegment , excessMax ) ; } // no capacity to receive additional buffers? if ( totalCapacity == 0 ) { return ; // necessary to avoid div by zero when nothing to re-distribute } // since one of the arguments of 'min(a,b)' is a positive int, this is actually // guaranteed to be within the 'int' domain // (we use a checked downCast to handle possible bugs more gracefully). final int memorySegmentsToDistribute = MathUtils . checkedDownCast ( Math . min ( numAvailableMemorySegment , totalCapacity ) ) ; long totalPartsUsed = 0 ; // of totalCapacity int numDistributedMemorySegment = 0 ; for ( LocalBufferPool bufferPool : allBufferPools ) { int excessMax = bufferPool . getMaxNumberOfMemorySegments ( ) - bufferPool . getNumberOfRequiredMemorySegments ( ) ; // shortcut if ( excessMax == 0 ) { continue ; } totalPartsUsed += Math . min ( numAvailableMemorySegment , excessMax ) ; // avoid remaining buffers by looking at the total capacity that should have been // re-distributed up until here // the downcast will always succeed, because both arguments of the subtraction are in the 'int' domain final int mySize = MathUtils . checkedDownCast ( memorySegmentsToDistribute * totalPartsUsed / totalCapacity - numDistributedMemorySegment ) ; numDistributedMemorySegment += mySize ; bufferPool . setNumBuffers ( bufferPool . getNumberOfRequiredMemorySegments ( ) + mySize ) ; } assert ( totalPartsUsed == totalCapacity ) ; assert ( numDistributedMemorySegment == memorySegmentsToDistribute ) ; }
Must be called from synchronized block