directbytebufferpool.java

这是一个基于java编写的torrent的P2P源码

/*
 * Created on Jan 30, 2004
 * Created by Alon Rohter
 * Copyright (C) 2004, 2005, 2006 Alon Rohter, All Rights Reserved.
 * 
 * Copyright (C) 2004, 2005, 2006 Aelitis, All Rights Reserved.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
 *
 * AELITIS, SAS au capital de 46,603.30 euros
 * 8 Allee Lenotre, La Grille Royale, 78600 Le Mesnil le Roi, France.
 */
package org.gudy.azureus2.core3.util;

import java.nio.ByteBuffer;
import java.util.*;
import java.math.*;

import org.gudy.azureus2.core3.config.COConfigurationManager;
import org.gudy.azureus2.core3.disk.DiskManager;
import org.gudy.azureus2.core3.logging.*;

import com.aelitis.azureus.core.diskmanager.cache.*;


/**
 * This class handles allocation of direct ByteBuffers.
 * It always tries to find a free buffer in the buffer pool
 * before creating a new one.
 */
public class 
DirectByteBufferPool 
{

	protected static final boolean 	DEBUG_TRACK_HANDEDOUT 	= AEDiagnostics.TRACE_DBB_POOL_USAGE;
	protected static final boolean 	DEBUG_PRINT_MEM 		= AEDiagnostics.PRINT_DBB_POOL_USAGE;
	protected static final int		DEBUG_PRINT_TIME		= 120*1000;
	
	protected static final boolean 	DEBUG_HANDOUT_SIZES 	= false;

	
	  // There is no point in allocating buffers smaller than 4K,
	  // as direct ByteBuffers are page-aligned to the underlying
	  // system, which is 4096 byte pages under most OS's.
	  // If we want to save memory, we can distribute smaller-than-4K
	  // buffers by using the slice() method to break up a standard buffer
	  // into smaller chunks, but that's more work.
	
	private static final int START_POWER = 12;    // 4096
	private static final int END_POWER   = 25;    // 33554432
  
  		// without an extra bucket here we get lots of wastage with the file cache as typically
  		// 16K data reads result in a buffer slightly bigger than 16K due to protocol header
  		// This means we would bump up to 32K pool entries, hence wasting 16K per 16K entry
  	
	private static final int[]	EXTRA_BUCKETS = { DiskManager.BLOCK_SIZE + 128 };
  
  
	public static final int MAX_SIZE = BigInteger.valueOf(2).pow(END_POWER).intValue();
  
	private static final DirectByteBufferPool pool = new DirectByteBufferPool();
  

	private final Map buffersMap = new LinkedHashMap(END_POWER - START_POWER + 1);
  
	private final Object poolsLock = new Object();

	private static final int	SLICE_END_SIZE				= 2048;	
	private static final int    SLICE_ALLOC_CHUNK_SIZE		= 4096;
  

	private static final short[]		SLICE_ENTRY_SIZES		= { 8, 16, 32, 64, 128, 256, 512, 1024, SLICE_END_SIZE };
	private static final short[]		SLICE_ALLOC_MAXS		= { 256, 256, 128, 64, 64,  64,  64,  64,   64 };
	
	private static final short[]		SLICE_ENTRY_ALLOC_SIZES = new short[SLICE_ENTRY_SIZES.length];
	private static final List[]			slice_entries 			= new List[SLICE_ENTRY_SIZES.length];
	private static final boolean[][]	slice_allocs 			= new boolean[SLICE_ENTRY_SIZES.length][];
	private static final boolean[]		slice_alloc_fails		= new boolean[SLICE_ENTRY_SIZES.length];
	
	static{
		
		int mult = COConfigurationManager.getIntParameter( "memory.slice.limit.multiplier" );
		
		if ( mult > 1 ){
			
			for (int i=0;i<SLICE_ALLOC_MAXS.length;i++){
				
				SLICE_ALLOC_MAXS[i] *= mult;
			}
		}
		
		for (int i=0;i<SLICE_ENTRY_SIZES.length;i++){
			
			SLICE_ENTRY_ALLOC_SIZES[i] = (short)(SLICE_ALLOC_CHUNK_SIZE/SLICE_ENTRY_SIZES[i]);
					
			slice_allocs[i] = new boolean[SLICE_ALLOC_MAXS[i]];
			
			slice_entries[i] = new LinkedList();
		}
	}
	
	private static final long[]			slice_use_count 	= new long[SLICE_ENTRY_SIZES.length];
  
	private final Map handed_out	= new IdentityHashMap();	// for debugging (ByteBuffer has .equals defined on contents, hence IdentityHashMap)
	
	private final Map	size_counts	= new TreeMap();
 
	private static final long COMPACTION_CHECK_PERIOD = 2*60*1000; //2 min
	private static final long MAX_FREE_BYTES = 10*1024*1024; //10 MB
  
	private long bytesIn = 0;
	private long bytesOut = 0;
  
  
	private 
	DirectByteBufferPool() 
	{
	  	
	    //create the buffer pool for each buffer size
	  	
	  	ArrayList	list = new ArrayList();
	  	
	    for (int p=START_POWER; p <= END_POWER; p++) {
	    	
	    	list.add( new Integer(BigInteger.valueOf(2).pow(p).intValue()));
	    }
	    
	    for (int i=0;i<EXTRA_BUCKETS.length;i++){
	    	       
	        list.add( new Integer(EXTRA_BUCKETS[i]));
	    }
	    
	    Integer[]	sizes = new Integer[ list.size() ];
	    list.toArray( sizes );
	    Arrays.sort( sizes);
	    
	    for (int i=0;i<sizes.length;i++){
	    	
	    	ArrayList bufferPool = new ArrayList();
	    	
	    	buffersMap.put(sizes[i], bufferPool);
	    }
	    
	    //initiate periodic timer to check free memory usage
	    SimpleTimer.addPeriodicEvent(
	    	"DirectBB:compact",
	        COMPACTION_CHECK_PERIOD,
	        new TimerEventPerformer() {
	          public void perform( TimerEvent ev ) {
	       
	            compactBuffers();
	          }
	        }
	     );
	    
	    if( DEBUG_PRINT_MEM ) {
	      Timer printer = new Timer("printer");
	      printer.addPeriodicEvent(
	          DEBUG_PRINT_TIME,
	          new TimerEventPerformer() {
	            public void perform( TimerEvent ev ) {
	              System.out.print("DIRECT: given=" +bytesOut/1024/1024+ "MB, returned=" +bytesIn/1024/1024+ "MB, ");
	              
	              long in_use = bytesOut - bytesIn;
	              if( in_use < 1024*1024 ) System.out.print( "in use=" +in_use+ "B, " );
	              else System.out.print( "in use=" +in_use/1024/1024+ "MB, " );
	              
	              long free = bytesFree();
	              if( free < 1024*1024 ) System.out.print( "free=" +free+ "B" );
	              else System.out.print( "free=" +free/1024/1024+ "MB" );
	
	              System.out.println();
	              
	              printInUse( false );
	              
	              long free_mem = Runtime.getRuntime().freeMemory() /1024/1024;
	              long max_mem = Runtime.getRuntime().maxMemory() /1024/1024;
	              long total_mem = Runtime.getRuntime().totalMemory() /1024/1024;
	              System.out.println("HEAP: max=" +max_mem+ "MB, total=" +total_mem+ "MB, free=" +free_mem+ "MB");
	              System.out.println();
	            }
	          }
	      );
	    }
	}

  
  /**
   * Allocate and return a new direct ByteBuffer.
   */
  private ByteBuffer allocateNewBuffer(final int _size) {
    try {
      return ByteBuffer.allocateDirect(_size);
    }
    catch (OutOfMemoryError e) {
       //Debug.out("Running garbage collector...");
       
       clearBufferPools();
       
       runGarbageCollection();

       try {
       		return ByteBuffer.allocateDirect(_size);
       	
       }catch (OutOfMemoryError ex) {
       	
         String msg = "Memory allocation failed: Out of direct memory space.\n"
                    + "To fix: Use the -XX:MaxDirectMemorySize=512m command line option,\n"
                    + "or upgrade your Java JRE to version 1.4.2_05 or 1.5 series or newer.";
       	 Debug.out( msg );
       	 
       	 Logger.log(new LogAlert(LogAlert.UNREPEATABLE, LogAlert.AT_ERROR, msg));
         
         printInUse( true );
         
         throw( ex );
       }
    }
  }

  
  /**
   * Retrieve a buffer from the buffer pool of size at least
   * <b>length</b>, and no larger than <b>DirectByteBufferPool.MAX_SIZE</b>
   */
  public static DirectByteBuffer 
  getBuffer(
  	byte	_allocator,
  	int 	_length) 
  {
    if (_length < 1) {
        Debug.out("requested length [" +_length+ "] < 1");
        return null;
    }

    if (_length > MAX_SIZE) {
        Debug.out("requested length [" +_length+ "] > MAX_SIZE [" +MAX_SIZE+ "]");
        return null;
    }

    return pool.getBufferHelper(_allocator,_length);
  }
  
  
  /**
   * Retrieve an appropriate buffer from the free pool, or
   * create a new one if the pool is empty.
   */
  
  
  	private DirectByteBuffer 
  	getBufferHelper(
		byte	_allocator,
		int 	_length) 
	{
		DirectByteBuffer	res;
		
		if ( _length <= SLICE_END_SIZE ){

			res = getSliceBuffer( _allocator, _length );
			
		}else{
				
			ByteBuffer	buff = null;
			
			Integer reqVal = new Integer(_length);
	    
				//loop through the buffer pools to find a buffer big enough
	    
			Iterator it = buffersMap.keySet().iterator();
	    
			while (it.hasNext()) {
	    	
				Integer keyVal = (Integer)it.next();
	
					//	check if the buffers in this pool are big enough
	      
				if (reqVal.compareTo(keyVal) <= 0) {
	      	
	   
					ArrayList bufferPool = (ArrayList)buffersMap.get(keyVal);
	            
					synchronized ( poolsLock ) { 
	        
						//	make sure we don't remove a buffer when running compaction
						//if there are no free buffers in the pool, create a new one.
						//otherwise use one from the pool
	        	
						if (bufferPool.isEmpty()) {
	          	
							buff = allocateNewBuffer(keyVal.intValue());
	            
						}else{
	          	
							synchronized ( bufferPool ) {
	            	
								buff = (ByteBuffer)bufferPool.remove(bufferPool.size() - 1);
							}
						}
					}
			
					break;
				}
			}
		
			if ( buff == null ){
						      
			    Debug.out("Unable to find an appropriate buffer pool");
			    
			    throw( new RuntimeException( "Unable to find an appropriate buffer pool" ));
			}
			
			res = new DirectByteBuffer( _allocator, buff, this );		   
		}
			
        	// clear doesn't actually zero the data, it just sets pos to 0 etc.
        
		ByteBuffer buff = res.getBufferInternal();
		
        buff.clear();   //scrub the buffer
        
		buff.limit( _length );
		
        bytesOut += buff.capacity();
                                
        if ( DEBUG_PRINT_MEM || DEBUG_TRACK_HANDEDOUT ){
        	
        	synchronized( handed_out ){
        	        	
				if ( DEBUG_HANDOUT_SIZES ){
					
					int	trim_size;
					
					if ( _length < 32 ){
						
						trim_size = 4;
					}else{
						
						trim_size = 16;
					}
					
					int	trim = ((_length+trim_size-1)/trim_size)*trim_size;
					
					Long count = (Long)size_counts.get(new Integer(trim));
					
					if ( count == null ){
						
						size_counts.put( new Integer( trim ), new Long(1));
						
					}else{
						
						size_counts.put( new Integer( trim), new Long( count.longValue() + 1 ));
					}
				}
				
        		if ( handed_out.put( buff, res ) != null ){
          		
        			Debug.out( "buffer handed out twice!!!!");
          		
        			throw( new RuntimeException( "Buffer handed out twice" ));
        		}
        	
				//System.out.println( "[" + handed_out.size() + "] -> " + buff + ", bytesIn = " + bytesIn + ", bytesOut = " + bytesOut );
          	}
        }
        
        // addInUse( dbb.capacity() );
        
        return( res );
    }
  
  
	  /**
	   * Return the given buffer to the appropriate pool.
	   */
	
	protected void 
	returnBuffer(
		DirectByteBuffer ddb ) 
	{
		ByteBuffer	buff = ddb.getBufferInternal();
		
		int	capacity = buff.capacity();

		bytesIn += capacity;
		    
	  	if ( DEBUG_TRACK_HANDEDOUT ){
	  		
	  		synchronized( handed_out ){

	  			if ( handed_out.remove( buff ) == null ){
	  				
	  				Debug.out( "buffer not handed out" );
	  				
	  				throw( new RuntimeException( "Buffer not handed out" ));
	  			}
	  			
	       		// System.out.println( "[" + handed_out.size() + "] <- " + buffer + ", bytesIn = " + bytesIn + ", bytesOut = " + bytesOut );
	  		}
	  	}
	  	
	    // remInUse( buffer.capacity() );
		
		if ( capacity <= SLICE_END_SIZE ){
			
			freeSliceBuffer( ddb );
			
		}else{
		    Integer buffSize = new Integer(capacity);
		    
		    ArrayList bufferPool = (ArrayList)buffersMap.get(buffSize);
		    
		    if (bufferPool != null) {
				
				//no need to sync around 'poolsLock', as adding during compaction is ok
				
		      synchronized ( bufferPool ){
				  
		        bufferPool.add(buff);
		      }
		    }else{
				
		      Debug.out("Invalid buffer given; could not find proper buffer pool");
		    }
		}
	}
  
  
  /**
   * Clears the free buffer pools so that currently
   * unused buffers can be garbage collected.
   */
  private void clearBufferPools() {
    Iterator it = buffersMap.values().iterator();
    while (it.hasNext()) {
        ArrayList bufferPool = (ArrayList)it.next();
        bufferPool.clear();
    }
  }
  
  
  /**
   * Force system garbage collection.
   */
  private void runGarbageCollection() {
    if( DEBUG_PRINT_MEM ) {
      System.out.println( "runGarbageCollection()" );
    }
    System.runFinalization();
    System.gc();
  }
  
  
  /**
   * Checks memory usage of free buffers in buffer pools,
   * and calls the compaction method if necessary.
   */
  private void compactBuffers() {
    long bytesUsed = 0;
    
    synchronized( poolsLock ) {
      
      //count up total bytes used by free buffers
      Iterator it = buffersMap.keySet().iterator();
      while (it.hasNext()) {
        Integer keyVal = (Integer)it.next();
        ArrayList bufferPool = (ArrayList)buffersMap.get(keyVal);
      
        bytesUsed += keyVal.intValue() * bufferPool.size();
      }
      
      //compact buffer pools if they use too much memory
      if (bytesUsed > MAX_FREE_BYTES) {
        compactFreeBuffers(bytesUsed);
      }
      
    }
	
	compactSlices();
  }
 
  
  
  /**
   * Fairly removes free buffers from the pools to limit memory usage.
   */
  private void compactFreeBuffers(final long bytes_used) {
    final int numPools = buffersMap.size();
    long bytesToFree = 0;
    int maxPoolSize = 0;
    
    int[] buffSizes = new int[numPools];
    int[] poolSizes = new int[numPools];
    int[] numToFree = new int[numPools];

    
    //fill size arrays
    int pos = 0;
    Iterator it = buffersMap.keySet().iterator();
    while (it.hasNext()) {
      Integer keyVal = (Integer)it.next();
      ArrayList bufferPool = (ArrayList)buffersMap.get(keyVal);
      
      buffSizes[pos] = keyVal.intValue();
      poolSizes[pos] = bufferPool.size();
      numToFree[pos] = 0;
      
      //find initial max value
      if (poolSizes[pos] > maxPoolSize) maxPoolSize = poolSizes[pos];
      
      pos++;
    }
    
    //calculate the number of buffers to free from each pool
    while( bytesToFree < (bytes_used - MAX_FREE_BYTES) ) {
      for (int i=0; i < numPools; i++) {
        //if the pool size is as large as the current max size
        if (poolSizes[i] == maxPoolSize) {
          //update counts
          numToFree[i]++;
          poolSizes[i]--;