diskmanagerimpl.java

一个基于JAVA的多torrent下载程序

/*
 * Azureus - a Java Bittorrent client
 *
 * This program is free software; you can redistribute it and/or modify
 * the Free Software Foundation; either version 2 of the License.
 *
 * 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 ( see the LICENSE file ).
 *
 * 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
 * 
 * Created on Oct 18, 2003
 * Created by Paul Gardner
 * Modified Apr 13, 2004 by Alon Rohter
 * Copyright (C) 2004, 2005, 2006 Aelitis, All Rights Reserved.
 * 
 */

package org.gudy.azureus2.core3.disk.impl;

import java.io.*;
import java.util.*;

import org.gudy.azureus2.core3.config.*;
import org.gudy.azureus2.core3.disk.*;
import org.gudy.azureus2.core3.disk.impl.access.*;
import org.gudy.azureus2.core3.disk.impl.piecemapper.*;
import org.gudy.azureus2.core3.disk.impl.resume.RDResumeHandler;
import org.gudy.azureus2.core3.download.*;
import org.gudy.azureus2.core3.internat.*;
import org.gudy.azureus2.core3.logging.*;
import org.gudy.azureus2.core3.torrent.*;
import org.gudy.azureus2.core3.util.*;
import org.gudy.azureus2.platform.*;
import org.gudy.azureus2.plugins.platform.PlatformManagerException;

import com.aelitis.azureus.core.diskmanager.access.*;
import com.aelitis.azureus.core.diskmanager.cache.*;
import com.aelitis.azureus.core.diskmanager.file.FMFileManagerFactory;


/**
 * 
 * The disk Wrapper.
 * 
 * @author Tdv_VgA
 * @author MjrTom
 *			2005/Oct/08: new piece-picking support changes
 *			2006/Jan/02: refactoring piece picking related code
 *
 */

public class 
DiskManagerImpl
	extends LogRelation
	implements DiskManagerHelper, DMReaderAdapter
{  
	private static final LogIDs LOGID = LogIDs.DISK;
	
	private static DiskAccessController	disk_access_controller;
	
	static {
		int	max_read_threads 		= COConfigurationManager.getIntParameter( "diskmanager.perf.read.maxthreads" );
		int	max_read_mb 			= COConfigurationManager.getIntParameter( "diskmanager.perf.read.maxmb" );
		int	max_write_threads 		= COConfigurationManager.getIntParameter( "diskmanager.perf.write.maxthreads" );
		int	max_write_mb 			= COConfigurationManager.getIntParameter( "diskmanager.perf.write.maxmb" );
		
		disk_access_controller = 
			DiskAccessControllerFactory.create(
					max_read_threads, max_read_mb,
					max_write_threads, max_write_mb );
		
		if (Logger.isEnabled()){
			Logger.log(
					new LogEvent( 
							LOGID, 
							"Disk access controller params: " +
								max_read_threads + "/" + max_read_mb + "/" + max_write_threads + "/" + max_write_mb ));
			
		}
	}

	private static DiskManagerRecheckScheduler		recheck_scheduler 		= new DiskManagerRecheckScheduler();
	private static DiskManagerAllocationScheduler	allocation_scheduler 	= new DiskManagerAllocationScheduler();
	
	private boolean	used	= false;
	
	private boolean started = false;
	private AESemaphore	started_sem	= new AESemaphore( "DiskManager::started" );
	private boolean	starting;
	private boolean	stopping;
	
	
	private int state_set_via_method;
	protected String errorMessage = "";

	private int pieceLength;
	private int lastPieceLength;

	private int			nbPieces;		// total # pieces in this torrent
	private long		totalLength;	// total # bytes in this torrent
	private int			percentDone; 
	private long		allocated;
	private long		remaining;

    
	private	TOTorrent		torrent;


	private DMReader				reader;
	private DMChecker				checker;
	private DMWriter				writer;
	
	private RDResumeHandler			resume_handler;
	private DMPieceMapper			piece_mapper;
		
	private DiskManagerPieceImpl[]	pieces;
	private DMPieceList[]			pieceMap;

	private DiskManagerFileInfoImpl[] 	files;
    protected DownloadManager 		download_manager;

	private boolean alreadyMoved = false;

	private boolean				skipped_file_set_changed =true;	// go over them once when starting
	private long				skipped_file_set_size;
	private long				skipped_but_downloaded;
	
	
		// DiskManager listeners
	
	private static final int LDT_STATECHANGED			= 1;
	private static final int LDT_PRIOCHANGED			= 2;
	private static final int LDT_PIECE_DONE_CHANGED		= 3;
	private static final int LDT_ACCESS_MODE_CHANGED	= 4;
	
	protected static ListenerManager	listeners_aggregator 	= ListenerManager.createAsyncManager(
			"DiskM:ListenAggregatorDispatcher",
			new ListenerManagerDispatcher()
			{
				public void
				dispatch(
					Object		_listener,
					int			type,
					Object		value )
				{
					DiskManagerListener	listener = (DiskManagerListener)_listener;
					
					if (type == LDT_STATECHANGED){
						
						int params[] = (int[])value;
						
  						listener.stateChanged(params[0], params[1]);
  						
					}else if (type == LDT_PRIOCHANGED) {
						
					    listener.filePriorityChanged((DiskManagerFileInfo)value);
					    
					}else if (type == LDT_PIECE_DONE_CHANGED) {
						
					    listener.pieceDoneChanged((DiskManagerPiece)value);
					    
					}else if (type == LDT_ACCESS_MODE_CHANGED) {
						
						Object[]	o = (Object[])value;
						
					    listener.fileAccessModeChanged( 
					    	(DiskManagerFileInfo)o[0],
					    	((Integer)o[1]).intValue(),
					    	((Integer)o[2]).intValue());
					}
				}
			});		
	
	private ListenerManager	listeners 	= ListenerManager.createManager(
			"DiskM:ListenDispatcher",
			new ListenerManagerDispatcher()
			{
				public void
				dispatch(
					Object		listener,
					int			type,
					Object		value )
				{
					listeners_aggregator.dispatch( listener, type, value );
				}
			});	
	
	private AEMonitor	start_stop_mon	= new AEMonitor( "DiskManager:startStop" );
	private AEMonitor	file_piece_mon	= new AEMonitor( "DiskManager:filePiece" );
	
	
	protected static int		max_read_block_size;
	
	static{    	
	    	
		ParameterListener param_listener = new ParameterListener() {
	    	    public void 
				parameterChanged( 
					String  str ) 
	    	    { 	      
					max_read_block_size	= COConfigurationManager.getIntParameter( "BT Request Max Block Size" );
	    	    }
		};

	 	COConfigurationManager.addAndFireParameterListener( "BT Request Max Block Size", param_listener);
	}
	   
	public 
	DiskManagerImpl(
		TOTorrent			_torrent, 
		DownloadManager 	_dmanager) 
	{
	    torrent 			= _torrent;
	    download_manager 	= _dmanager;
	 
	    pieces		= new DiskManagerPieceImpl[0];	// in case things go wrong later
	    
	    setState( INITIALIZING );
	    
	    percentDone = 0;
	    
		if ( torrent == null ){
			
			errorMessage	 = "Torrent not available";
			
			setState( FAULTY );
			
			return;
		}
   
		LocaleUtilDecoder	locale_decoder = null;
		
		try{
			locale_decoder = LocaleUtil.getSingleton().getTorrentEncoding( torrent );
			
		}catch( TOTorrentException e ){
			
			Debug.printStackTrace( e );
			
			errorMessage = TorrentUtils.exceptionToText(e);
			
			setState( FAULTY );
			
			return;
			
		}catch( Throwable e ){
			
			Debug.printStackTrace( e );
			
			errorMessage = "Initialisation failed - " + Debug.getNestedExceptionMessage(e);
			
			setState( FAULTY );
			
			return;
		}
		
		piece_mapper	= DMPieceMapperFactory.create( torrent );
		
		try{
			piece_mapper.construct( locale_decoder, download_manager.getAbsoluteSaveLocation().getName());
			
		}catch( Throwable e ){
			
			Debug.printStackTrace( e );

			errorMessage = "Failed to build piece map - " + Debug.getNestedExceptionMessage(e);
			
			setState( FAULTY );
			
			return;
		}

		totalLength	= piece_mapper.getTotalLength();
		remaining 	= totalLength;

		nbPieces 	= torrent.getNumberOfPieces();
		
		pieceLength		= (int)torrent.getPieceLength();
		lastPieceLength	= piece_mapper.getLastPieceLength();
		
		pieces		= new DiskManagerPieceImpl[nbPieces];
		for (int i =0; i <nbPieces; i++)
		{
			pieces[i] =new DiskManagerPieceImpl(this, i);
		}

		reader			= DMAccessFactory.createReader(this);
		
		checker			= DMAccessFactory.createChecker(this);
		
		writer			= DMAccessFactory.createWriter(this);
		
		resume_handler	= new RDResumeHandler( this, checker );
	
	}

	public void 
	start() 
	{		
		try{
			start_stop_mon.enter();
	
			if ( used ){
				
				Debug.out( "DiskManager reuse not supported!!!!" );
			}
			
			used	= true;
			
			if ( getState() == FAULTY ){
				
				Debug.out( "starting a faulty disk manager");
				
				return;
			}
			
			started 	= true;
			starting	= true;
			
		    Thread init = 
		    	new AEThread("DiskManager:start") 
				{
					public void 
					runSupport() 
					{
						try{
							startSupport();
							
						}catch( Throwable e ){
							
							errorMessage = Debug.getNestedExceptionMessage(e) + " (start)";
							
							setState( FAULTY );

						}finally{
														
							started_sem.release();
						}
						
						boolean	stop_required;
						
						try{
							start_stop_mon.enter();
						
							stop_required = DiskManagerImpl.this.getState() == DiskManager.FAULTY || stopping;
							
							starting	= false;
							
						}finally{
							
							start_stop_mon.exit();
						}
						
						if ( stop_required ){
						
							DiskManagerImpl.this.stop();
						}
					}
				};
				
			init.setPriority(Thread.MIN_PRIORITY);
			
			init.start();
			
		}finally{
			
			start_stop_mon.exit();
		}
	}

	private void 
	startSupport() 
	{		
			//if the data file is already in the completed files dir, we want to use it
		
		boolean moveWhenDone = COConfigurationManager.getBooleanParameter("Move Completed When Done", false);
		
		String moveToDir = COConfigurationManager.getStringParameter("Completed Files Directory", "");
   
		if ( moveWhenDone && moveToDir.length() > 0 && download_manager.isPersistent()){
		
				//if the data file already resides in the completed files dir
					
			if ( filesExist( moveToDir )){
				
				alreadyMoved = true;
		
				download_manager.setTorrentSaveDir( moveToDir );
			}
		}

		reader.start();
		
		checker.start();
		
		writer.start();
				
			//allocate / check every file

		int newFiles = allocateFiles();
      
		if ( getState() == FAULTY ){
			
				// bail out if broken in the meantime
				// state will be "faulty" if the allocation process is interrupted by a stop
			
			return;
		}
    
        pieceMap = piece_mapper.getPieceMap();

		constructFilesPieces();
		
		if ( getState() == FAULTY  ){
			
				// bail out if broken in the meantime

			return;
		}
		
		resume_handler.start();
		  
		if ( newFiles == 0 ){
			
			resume_handler.checkAllPieces(false);
			
		}else if ( newFiles != files.length ){
			
				//	if not a fresh torrent, check pieces ignoring fast resume data
			
			resume_handler.checkAllPieces(true);
		}
		
		if ( getState() == FAULTY  ){
			
			return;
		}
		
			// in all the above cases we want to continue to here if we have been "stopped" as
			// other components require that we end up either FAULTY or READY
		
			//3.Change State   
		
		setState( READY );
	}

	public void 
	stop() 
	{	
		try{
			start_stop_mon.enter();
		
			if ( !started ){
			
				return;
			}
			
				// we need to be careful if we're still starting up as this may be
				// a re-entrant "stop" caused by a faulty state being reported during
				// startup. Defer the actual stop until starting is complete
			
			if ( starting ){
				
				stopping	= true;

					// we can however safely stop things at this point - this is important
					// to interrupt an alloc/recheck process that might be holding up the start
					// operation
				
			   	checker.stop();
		    	
			   	writer.stop();
			   	
				reader.stop();
				
				resume_handler.stop();

				return;
			}
			
			started		= false;
			
			stopping	= false;
			
		}finally{
			
			start_stop_mon.exit();
		}
		
		started_sem.reserve();
		
		boolean	checking = checker.getCompleteRecheckStatus() != -1;
		
    	checker.stop();
    	
    	writer.stop();
    	
		reader.stop();
		
		resume_handler.stop();
		if ( files != null ){
			
			for (int i = 0; i < files.length; i++){
				
				try{
					if (files[i] != null) {
						
						files[i].getCacheFile().close();
					}
				}catch ( Throwable e ){
					
					setFailed( "File close fails: " + Debug.getNestedExceptionMessage(e));