btsession.java

p2p仿真

/*
 * @(#)BTSession.java	ver 1.2  6/20/2005
 *
 * Copyright 2005 Weishuai Yang ([e-mail]). 
 * All rights reserved.
 * 
 */


package gps.protocol.BT;

import gps.Simulator;
import gps.event.SimEvent;
import gps.event.SimEventHandler;
import gps.event.SimEventScheduler;
import gps.gui.ProtocolPanel;
import gps.protocol.BT.algorithm.BTAlgorithmPieceSelection;
import gps.protocol.BT.algorithm.BTAlgorithmReChoking;
import gps.protocol.BT.param.BTGetRequest;
import gps.protocol.BT.param.BTPeerMessage;
import gps.protocol.BT.param.BTTorrent;
import gps.protocol.BT.param.BTTrackerResponse;
import gps.util.LogFormatter;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.logging.Logger;



/**
 * BT Session corresponding to a downloading or uploading session for a document.
 * 
 * @author  Weishuai Yang
 * @version 1.2,  6/20/2005
 */
public class BTSession implements SimEventHandler {

    /**
     * peer candidate list length
     */
	public static final int PEER_CANDIDATE_LIST_LENGTH=50;
	/**
	 * max allowed connection num
	 */
	public static final int MAX_ALLOWED_CONNECTION_NUM=60;
	
	/**
	 * session status - working
	 */
	public static final int WORKING=0;
	/**
	 * session status - completed
	 */
	public static final int COMPLETED=1;
	/**
	 * session status - stopped
	 */
	public static final int STOPPED=2;
	/**
	 * session status - failed
	 */
	public static final int FAILED=3;


	/**
	 * session status
	 */
	public int mStatus=WORKING;
	/**
	 * if closed, set up the mClosed flag, and remove from hashmap
	 */
	public boolean mRemoved=false;


	/**
	 * singleton reference to event scheduler
	 */
	private static SimEventScheduler mScheduler=SimEventScheduler.getInstance();
	
	/**
	 * singleton reference to debug log
	 */
	private static Logger mDebugLog = Logger.getLogger("Debug");
	/**
	 * singleton reference to trace log
	 */
	private static Logger mTraceLog = Logger.getLogger("Trace");


	/**
	 * torrent file including the document's meta information
	 */
	private BTTorrent mTorrent=null;
	
	/**
	 * the downloading/uploading document
	 */
	private BTDocument mDocument=null;

	
	/**
	 * reference back to the hosting peer
	 */
	private BTPeer mAgent=null;
	
	/**
	 * connections to other peers shareing the same document
	 */
	private LinkedHashMap mConnections=null;
	
	/**
	 * mKnowledge is the knowledge base of which piece is on which peer, only includes pieces this agent doesn't have
	 */
	private LinkedHashSet mKnowledge[];
	
	/**
	 * peer candidate list
	 */
	private LinkedList mPeerCandidateList=null;
	/**
	 * unchoked list
	 */
	private LinkedHashSet mUnchokedList=null;
	
	/**
	 * the amount downloaded between snubbing detection
	 */
	private double mDownloadBetweenSnubbingDetection=0;
	
	/**
	 * next tracker announencement event
	 */
	private BTEvent mNextTrackerAnnouncement=null;
	
	/**
	 * rechoking algorithm
	 */
	private BTAlgorithmReChoking mBTARC=new BTAlgorithmReChoking(this);
	
	/**
	 * piece selection algorithm
	 */
	private BTAlgorithmPieceSelection mPieceSelection = null;
	
	
	
	/**
	 * constructs a bt session from torrent
	 * @param t torrent file
	 * @param agent peer
	 */
	public BTSession(BTTorrent t, BTPeer agent){
		mTorrent=t;
		mAgent=agent;
		//possible for a resume
		mDocument=mAgent.getBTDocument(t.mDocHashKey);
		if(mDocument==null){
			mDocument=new BTDocument(t.mPieceLength, t.mLength, t.mDocHashKey);
			mAgent.ownDocument(mDocument);
		}
		//by default, this new document is not whole
		mConnections=new LinkedHashMap();
		mKnowledge=new LinkedHashSet[mDocument.getPieceNum()];
		for(int i=0; i<mDocument.getPieceNum();i++){
			mKnowledge[i]=new LinkedHashSet();
		}
		mPeerCandidateList=new LinkedList();
		mUnchokedList=new LinkedHashSet();
		
		mPieceSelection=new BTAlgorithmPieceSelection(mDocument.getID()*1000+getAgent().getID());
		mBTARC.run();
	}
	/**
	 * constructs a bt session from document hash key
	 * @param key document hash key
	 * @param agent peer
	 */
	public BTSession(String key, BTPeer agent){
		//this is a seed session, however, the document may be partial
	    mAgent=agent;
		//possible for a resume
		mDocument=mAgent.getBTDocument(key);
		//this document must exist
		mConnections=new LinkedHashMap();
		mKnowledge=new LinkedHashSet[mDocument.getPieceNum()];
		for(int i=0; i<mDocument.getPieceNum();i++){
			mKnowledge[i]=new LinkedHashSet();
		}
		mPeerCandidateList=new LinkedList();
		mUnchokedList=new LinkedHashSet();
		mPieceSelection=new BTAlgorithmPieceSelection(mDocument.getID()*1000+getAgent().getID());
		mBTARC.run();
	}


	/**
	 * gets the Agent object on which this session is running
	 * @return agent
	 */
	 
	public BTPeer getAgent(){
	    return mAgent;
	}
	
	
	/**
	 * set torrent for existing session
	 * @param t torrent
	 */
	public void setTorrent(BTTorrent t){
		mTorrent=t;
		if(t.mDocHashKey!=mDocument.getKey()){
			mDebugLog.severe("set torrent to a session with different key!");
			mDocument=mAgent.getBTDocument(t.mDocHashKey);
			if(mDocument==null)
				mDocument=new BTDocument(t.mPieceLength, t.mLength, t.mDocHashKey);
			//by default, this new document is not whole
			/*
			mConnections=new LinkedHashMap();
			mKnowledge=new LinkedHashSet[mDocument.getPieceNum()];
			for(int i=0; i<mDocument.getPieceNum();i++){
				mKnowledge[i]=new LinkedHashSet();
			}
			*/
		}
	}
	/**
	 * gets document that the session is downloading/uploading
	 * @return document object
	 */
	public BTDocument getDocument(){
		return mDocument;
	}
	
	/**
	 * gets connections
	 * @return hash map of connections
	 */
	public LinkedHashMap getConnections(){
		return mConnections;
	}
	/**
	 * gets connection to a peer
	 * @param p peer as an index
	 * @return connection
	 */
	public BTSocket getConnection(BTPeer p){
		return (BTSocket)mConnections.get(p);
	}
	/**
	 * gets candidate list
	 * @return candidate list
	 */
	public LinkedList getPeerCandidateList(){
		return mPeerCandidateList;
	}
	
	/**
	 * gets unchoked list
	 * @return unchoked list
	 */
	public LinkedHashSet getUnchokedList(){
		return mUnchokedList;
	}
	/**
	 * gets download amount between snubbing detection
	 * @return download amount
	 */
	public double getDownloadBetweenSnubbingDetection(){
		return mDownloadBetweenSnubbingDetection;
	}
	/**
	 * sets download amount between snubbing detection
	 * @param d new download amount
	 */
	public void setDownloadBetweenSnubbingDetection(double d){
		mDownloadBetweenSnubbingDetection=d;
	}
	/**
	 * process event sent to this session
	 * @param e the event object with some parameters in it
	 * @return true if already handled
	 */
	public boolean handle(SimEvent e) {
		switch(e.getType()){
			case BTEvent.SESSION_TRACKER_RESPONSE: 
				handleTrackerResponse((BTEvent)e);
				return true;
			case BTEvent.SESSION_ANNOUNCEMENT: 
				announceTracker();
				return true;
			case BTEvent.PEER_MESSAGE: 
				handlePeerMessage((BTEvent)e);
				return true;
			//connectin timeout event comes from itself
			case BTEvent.CONNECTION_TIMEOUT: 
				handleConnectionTimeout((BTEvent)e);
				return true;
			default:
				mDebugLog.warning("received event "+e+" unprocessed");
		}
		return false;
	}	
	
	/**
	 * process tracker response evnet
	 * @param e the event object
	 */
	public void handleTrackerResponse(BTEvent e){
	    
	    BTTrackerResponse bttr=(BTTrackerResponse)e.getAddParam();
		if(bttr.mFailure){
			mDebugLog.warning("received failure from tracker");
			return;
		}
		BTTracker tracker=(BTTracker)e.getParam();
		String docHaskKey=bttr.mDocHashKey;
		ArrayList peerList=new ArrayList(bttr.mPeers);
		if(peerList.contains(mAgent)){
			peerList.remove(mAgent);
		}	

		//add all the peers to mPeerList
		for(int i=0;i<peerList.size();i++){
			BTPeer p=(BTPeer)peerList.get(i);
			if(!mPeerCandidateList.contains(p))
				mPeerCandidateList.addLast(p);
		}

		if(mPeerCandidateList.contains(mAgent)){
		    mPeerCandidateList.remove(mAgent);
		}	
		
		if(mPeerCandidateList.size()+peerList.size()>PEER_CANDIDATE_LIST_LENGTH){
			while(mPeerCandidateList.size()>=PEER_CANDIDATE_LIST_LENGTH)
		    	mPeerCandidateList.removeFirst();
		}
		//select some nodes to connect, and those selected are removed from the mPeerCandidateList;
		
		ArrayList toConn=getAgent().mPeerSelection.selectPeers(mPeerCandidateList);

		for(int i=0;i<toConn.size();i++){
			//connect to unconnected peers, initially choked and interested
			if(!mConnections.containsKey((BTPeer)toConn.get(i)))
		    	connectTo((BTPeer)toConn.get(i), true);
		    //if already connected, do nothing
		}	
		
		if(bttr.mInterval!=0){
			//schedule furture announcement without parameter
			BTEvent bte=new BTEvent(mScheduler.getCurrent()+bttr.mInterval, BTEvent.SESSION_ANNOUNCEMENT, this, null);
			mScheduler.enqueue(bte);
			mNextTrackerAnnouncement=bte;
		}

	}

	/**
	 * handles all the peer message from peers except for hand shaking.
	 * hand shaking is handled by btpeer. 
	 *
	 * @param e the event object
	 */
	public void handlePeerMessage(BTEvent e){
		
		BTSession bts=(BTSession)e.getParam();

		
		BTPeerMessage btpm=(BTPeerMessage)e.getAddParam();
		
		/*
		mDebugLog.info(""+mScheduler.getCurrent()+" "+this+"receiving peer message from " + 
						bts.getAgent()+" for "+btpm.mDocHashKey+" Type: "+
						btpm.messageToString());
		*/		
		
		if(btpm.mDocHashKey!=mDocument.getKey()){
			mDebugLog.warning(""+mScheduler.getCurrent()+" "+this+"receiving peer message from " + 
								bts.getAgent()+" with "+btpm.mDocHashKey+" Type: "+
								btpm.messageToString()+", but docHashKey for this session is "+
								mDocument.getKey());
		}

		//only BITFIELD message can have no existing connection
		if(btpm.mType!=BTPeerMessage.BITFIELD){
		    if(!mConnections.containsKey(bts.getAgent())){
		        mDebugLog.warning(mAgent+": no existing connectin when receiving a peer message from"+bts.getAgent());
		        return;
		    }
		    else{
		        //only if it's not bitfiled, extend the timeout time
				BTSocket btc=(BTSocket)mConnections.get(bts.getAgent());
				if(btc.mTimeout.getTimeStamp()<mScheduler.getCurrent()+BTSocket.CONNECTION_TIMEOUT){
					mScheduler.cancel(btc.mTimeout);
					btc.mTimeout.setTimeStamp(mScheduler.getCurrent()+BTSocket.CONNECTION_TIMEOUT);
					mScheduler.enqueue(btc.mTimeout);
					//mDebugLog.info(mScheduler.getCurrent()+": "+mAgent+", received "+btpm+", connection timeout updated1 to " +(mScheduler.getCurrent()+BTSocket.CONNECTION_TIMEOUT)+" for connection to "+bts.getAgent());
				}
		    }
		} 

		switch(btpm.mType){
			
			//keep alive is ignored currently
			case BTPeerMessage.KEEP_ALIVE:  return;
			
			//receive bitfield, update knowledge and set up connection
			case BTPeerMessage.BITFIELD:
			    handlePMBitField(e);
				return;