diskmanagerchannelimpl.java

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

/*
 * Created on 29-Mar-2006
 * Created by Paul Gardner
 * Copyright (C) 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.pluginsimpl.local.disk;

import java.util.*;

import org.gudy.azureus2.core3.disk.DiskManagerFileInfoListener;
import org.gudy.azureus2.core3.download.DownloadManagerPeerListener;
import org.gudy.azureus2.core3.peer.PEPeer;
import org.gudy.azureus2.core3.peer.PEPeerManager;
import org.gudy.azureus2.core3.peer.PEPiece;
import org.gudy.azureus2.core3.torrent.TOTorrent;
import org.gudy.azureus2.core3.torrent.TOTorrentFile;
import org.gudy.azureus2.core3.util.AESemaphore;
import org.gudy.azureus2.core3.util.Average;
import org.gudy.azureus2.core3.util.Debug;
import org.gudy.azureus2.core3.util.DirectByteBuffer;
import org.gudy.azureus2.core3.util.SystemTime;
import org.gudy.azureus2.plugins.disk.DiskManagerChannel;
import org.gudy.azureus2.plugins.disk.DiskManagerEvent;
import org.gudy.azureus2.plugins.disk.DiskManagerListener;
import org.gudy.azureus2.plugins.disk.DiskManagerRequest;
import org.gudy.azureus2.plugins.download.Download;
import org.gudy.azureus2.plugins.utils.PooledByteBuffer;
import org.gudy.azureus2.pluginsimpl.local.download.DownloadImpl;
import org.gudy.azureus2.pluginsimpl.local.utils.PooledByteBufferImpl;

import com.aelitis.azureus.core.peermanager.piecepicker.PiecePicker;
import com.aelitis.azureus.core.peermanager.piecepicker.PieceRTAProvider;

public class 
DiskManagerChannelImpl 
	implements DiskManagerChannel, DiskManagerFileInfoListener, DownloadManagerPeerListener, PieceRTAProvider
{
	private static final int		BUFFER_SECS 			= 30;
	private static final int		MIN_PIECES_TO_BUFFER	= 3;
	
	private static final boolean	TRACE = false;
	
	private static final int COMPACT_DELAY	= 32;
	
	private static final int MAX_READ_CHUNK	= 64*1024;
	
	private static final Comparator comparator = new
		Comparator()
		{
			public int 
		   	compare(
		   		Object _o1, 
				Object _o2)
			{
				dataEntry	o1 = (dataEntry)_o1;
				dataEntry	o2 = (dataEntry)_o2;
				
				long	offset1 = o1.getOffset();
				long	length1	= o1.getLength();
				
				long	offset2 = o2.getOffset();
				long	length2	= o2.getLength();
			
		   	
				long	res;
				
				if ( offset1 == offset2 ){
					
					res = length1 - length2;
					
				}else{
					
					res = offset1 - offset2;
				}
				
				if ( res == 0 ){
					return(0);
				}else if ( res < 0 ){
					return(-1);
				}else{
					return(1);
				}
			}
		};
		
	private static final String	channel_key = "DiskManagerChannel";
	private static int	channel_id_next;
	
	private DownloadImpl	download;
	
	private org.gudy.azureus2.pluginsimpl.local.disk.DiskManagerFileInfoImpl	plugin_file;
	private org.gudy.azureus2.core3.disk.DiskManagerFileInfo					core_file;
	
	private Set	data_written = new TreeSet( comparator );
	
	private int compact_delay	= COMPACT_DELAY;
	
	private List	waiters	= new ArrayList();

	private long	file_offset_in_torrent;
	private long	piece_size;
		
	private Average	byte_rate = Average.getInstance( 1000, 20 );
	
	private long	current_position;
	private request	current_request;
	
	private PEPeerManager	peer_manager;
	
	private long[]	rtas;
	
	private int		channel_id;
	
	protected
	DiskManagerChannelImpl(
		DownloadImpl															_download,
		org.gudy.azureus2.pluginsimpl.local.disk.DiskManagerFileInfoImpl		_plugin_file )
	{
		download		= _download;
		plugin_file		= _plugin_file;
		
		core_file		= plugin_file.getCore();
		
		synchronized( DiskManagerChannelImpl.class ){
			
			channel_id = channel_id_next++;
		}
				
		TOTorrentFile	tf = core_file.getTorrentFile();
		
		TOTorrent 	torrent = tf.getTorrent();
		
		TOTorrentFile[]	tfs = torrent.getFiles();

		rtas	= new long[torrent.getNumberOfPieces()];
		
		core_file.getDownloadManager().addPeerListener(this);
			
		for (int i=0;i<core_file.getIndex();i++){
				
			file_offset_in_torrent += tfs[i].getLength();
		}
			
		piece_size	= tf.getTorrent().getPieceLength();
		
		core_file.addListener( this );
	}
	
	public DiskManagerRequest
	createRequest()
	{
		if ( core_file.getDownloaded() != core_file.getLength()){
			
			if ( core_file.isSkipped()){
				
				core_file.setSkipped( false );
			}
			
			boolean	force_start = download.isForceStart();
			
			if ( !force_start ){
				
				synchronized( download ){
					
					Map	dl_state = (Map)download.getDownload().getData( channel_key );
					
					if ( dl_state == null ){
						
						dl_state = new HashMap();
						
						download.getDownload().setData( channel_key, dl_state );
					}
					
					dl_state.put( ""+channel_id, "" );
				}
				
				download.setForceStart( true );
			}
		}
		
		current_request = new request();
		
		return( current_request );
	}
	
	public void
	dataWritten(
		long	offset,
		long	length )
	{
		if ( TRACE ){
			System.out.println( "data written:" + offset + "/" + length );
		}
		
		dataEntry	entry = new dataEntry( offset, length );
		
		synchronized( data_written ){
			
			data_written.add( entry );
			
			compact_delay--;
			
			if ( compact_delay == 0 ){
				
				compact_delay	= COMPACT_DELAY;
				
				Iterator	it = data_written.iterator();
				
				dataEntry	prev_e	= null;
				
				while( it.hasNext()){
					
					dataEntry	this_e = (dataEntry)it.next();
					
					if ( prev_e == null ){
						
						prev_e = this_e;
						
					}else{
						
						long	prev_offset = prev_e.getOffset();
						long	prev_length	= prev_e.getLength();
						long	this_offset = this_e.getOffset();
						long	this_length	= this_e.getLength();
						
						if ( this_offset <= prev_offset + prev_length ){
							
							if ( TRACE ){	
								System.out.println( "merging: " + prev_e.getString()  + "/" + this_e.getString());
							}
							
							it.remove();
							
							prev_e.setLength( Math.max( prev_offset + prev_length, this_offset + this_length ) - prev_offset );
						
							if ( TRACE ){	
								System.out.println( "    -> " + prev_e.getString());
							}

						}else{
							
							prev_e = this_e;
						}
					}
				}
			}
		
			for (int i=0;i<waiters.size();i++){
					
				((AESemaphore)waiters.get(i)).release();
			}
		}
	}
	
	public void
	dataChecked(
		long	offset,
		long	length )
	{
		// System.out.println( "data checked:" + offset + "/" + length );
	}
	
	public void
	peerManagerAdded(
		PEPeerManager	manager )
	{
		peer_manager = manager;
		
		manager.getPiecePicker().addRTAProvider( this );
	}
	
	public void
	peerManagerRemoved(
		PEPeerManager	manager )
	{
		peer_manager = null;
		
		manager.getPiecePicker().removeRTAProvider( this );
	}
	
	public void
	peerAdded(
		PEPeer 	peer )
	{
	}
		
	public void
	peerRemoved(
		PEPeer	peer )
	{
	}
		
	public void
	pieceAdded(
		PEPiece 	piece )
	{
	}
		
	public void
	pieceRemoved(
		PEPiece		piece )
	{
	}
	       	
   	public long[]
   	updateRTAs(
   		PiecePicker		picker )
   	{
   		long	overall_pos = current_position + file_offset_in_torrent;
   		
   		int	first_piece = (int)( overall_pos / piece_size );
   		
   		long	rate = byte_rate.getAverage();
   		
   		long	buffer_bytes = BUFFER_SECS * rate;
   		
   		int	pieces_to_buffer = (int)( buffer_bytes / piece_size );
   		
   		if ( pieces_to_buffer < 1 ){
   			
   			pieces_to_buffer	= 1;
   		}
   		
   		int	millis_per_piece = BUFFER_SECS*1000/pieces_to_buffer; 

   		if ( pieces_to_buffer < MIN_PIECES_TO_BUFFER ){
   			
   			pieces_to_buffer = MIN_PIECES_TO_BUFFER;
   		}
   		   		
   		// System.out.println( "rate = " + rate + ", buffer_bytes = " + buffer_bytes + ", pieces = " + pieces_to_buffer + ", millis_per_piece = " + millis_per_piece );
   		
   		Arrays.fill( rtas, 0 );
   		 
   		long	now = SystemTime.getCurrentTime();
   		
   		for (int i=first_piece;i<first_piece+pieces_to_buffer&&i<rtas.length;i++){
   			
   			rtas[i]	= now + (( i - first_piece ) * millis_per_piece );
   		}
   		
   		return( rtas );
   	}
   	
	public long
	getCurrentPosition()
	{
		return( current_position );
	}
	
	public long
	getBlockingPosition()
	{
		request r = current_request;
		
		if ( r == null ){
			
			return( current_position );
		}
		
		return( current_position + r.getAvailableBytes());
	}
	
	public void
	destroy()
	{
		core_file.removeListener( this );
		
		core_file.getDownloadManager().removePeerListener(this);
		
		if ( peer_manager != null ){
			
			peer_manager.getPiecePicker().removeRTAProvider( this );