peer.cpp

ctorrent源码

#include "peer.h"

#include <stdlib.h>
#include <string.h>
#include <ctype.h>

#include "btstream.h"
#include "./btcontent.h"
#include "./msgencode.h"
#include "./peerlist.h"
#include "./btconfig.h"
#include "bttime.h"

size_t get_nl(char *sfrom)
{
  unsigned char *from = (unsigned char *)sfrom;
  size_t t;
  t = (*from++) << 24;
  t |= (*from++) << 16;
  t |= (*from++) << 8;
  t |= *from;
  return t;
}

void set_nl(char *sto, size_t from)
{
  unsigned char *to = (unsigned char *)sto;
  *to++ = (from >> 24) & 0xff;
  *to++ = (from >> 16) & 0xff;
  *to++ = (from >> 8) & 0xff;
  *to = from & 0xff;
}

// Convert a peer ID to a printable string.
int TextPeerID(unsigned char *peerid, char *txtid)
{
  int i, j;

  for(i=j=0; i < PEER_ID_LEN; i++){
    if( i==j && isprint(peerid[i]) && !isspace(peerid[i]) )
      txtid[j++] = peerid[i];
    else{
      if(i==j){ sprintf(txtid+j, "0x"); j+=2; }
      snprintf(txtid+j, 3, "%.2X", (int)(peerid[i]));
      j += 2;
    }
  }
  txtid[j] = '\0';

  return 0;
}


/* g_next_up is used to rotate uploading.  If we have the opportunity to
   upload to a peer but skip it due to bw limiting, the var is set to point to
   that peer and it will be given priority at the next opportunity.
   g_next_dn is similar, but for downloading.
   g_defer_up/dn is used to let the g_next peer object know if it skipped.
*/
btPeer *g_next_up = (btPeer *)0;
btPeer *g_next_dn = (btPeer *)0;
unsigned char g_defer_up = 0;
unsigned char g_defer_dn = 0;
btBasic Self;

void btBasic::SetCurrentRates()
{
  m_current_dl = rate_dl.RateMeasure();
  m_current_ul = rate_ul.RateMeasure();
  m_use_current = 1;
}

void btBasic::SetIp(struct sockaddr_in addr)
{
  memcpy(&m_sin.sin_addr,&addr.sin_addr,sizeof(struct in_addr));
}

void btBasic::SetAddress(struct sockaddr_in addr)
{
  memcpy(&m_sin,&addr,sizeof(struct sockaddr_in));
}

int btBasic::IpEquiv(struct sockaddr_in addr)
{
//	fprintf(stdout,"IpEquiv: %s <=> ", inet_ntoa(m_sin.sin_addr));
//	fprintf(stdout,"%s\n", inet_ntoa(addr.sin_addr));
        return (memcmp(&m_sin.sin_addr,&addr.sin_addr,sizeof(struct in_addr)) == 0) ? 
    1 : 0;
}

int btPeer::Need_Local_Data()
{
  if( m_state.remote_interested && !bitfield.IsFull()){

    if( BTCONTENT.pBF->IsFull() ) return 1; // i am seed

    BitField tmpBitfield = *BTCONTENT.pBF;
    tmpBitfield.Except(bitfield);
    return tmpBitfield.IsEmpty() ? 0 : 1;

  }
  return 0;
}

int btPeer::Need_Remote_Data()
{

  if( BTCONTENT.pBF->IsFull()) return 0;
  else if( bitfield.IsFull() ) return 1;
  else{
    BitField tmpBitfield = bitfield;
    tmpBitfield.Except(*BTCONTENT.pBF);
    tmpBitfield.Except(*BTCONTENT.pBFilter);
    return tmpBitfield.IsEmpty() ? 0 : 1;
  }
  return 0;
}

btPeer::btPeer()
{
  m_f_keepalive = 0;
  m_status = P_CONNECTING;
  m_unchoke_timestamp = (time_t) 0;
  m_last_timestamp = now;
  m_state.remote_choked = m_state.local_choked = 1;
  m_state.remote_interested = m_state.local_interested = 0;

  m_err_count = 0;
  m_cached_idx = BTCONTENT.GetNPieces();
  m_standby = 0;
  m_req_send = 5;
  m_req_out = 0;
  m_latency = 0;
  m_prev_dlrate = 0;
  m_health_time = m_receive_time = m_choketime = m_last_timestamp;
  m_bad_health = 0;
}

int btPeer::SetLocal(unsigned char s)
{
  switch(s){
  case M_CHOKE:
    if( m_state.local_choked ) return 0;
    m_unchoke_timestamp = now;
//  if(arg_verbose) fprintf(stderr, "Choking %p\n", this);
    if(arg_verbose) fprintf(stderr, "Choking %p (D=%lluMB@%uK/s)\n", this,
      TotalDL() >> 20, RateDL() >> 10);
    m_state.local_choked = 1; 
    if( g_next_up == this ) g_next_up = (btPeer *)0;
    if( !reponse_q.IsEmpty()) reponse_q.Empty();
    break;
  case M_UNCHOKE: 
    if( !reponse_q.IsEmpty() ) StartULTimer();
    if( !m_state.local_choked ) return 0;
    m_unchoke_timestamp = now;
//  if(arg_verbose) fprintf(stderr, "Unchoking %p\n", this);
    if(arg_verbose) fprintf(stderr, "Unchoking %p (D=%lluMB@%uK/s)\n", this,
      TotalDL() >> 20, RateDL() >> 10);
    m_state.local_choked = 0;
    break;
  case M_INTERESTED: 
    m_standby = 0;
    if( m_state.local_interested ) return 0;
    if(arg_verbose) fprintf(stderr, "Interested in %p\n", this);
    m_state.local_interested = 1;
    break;
  case M_NOT_INTERESTED:
    if( !m_state.local_interested ) return 0;
    if(arg_verbose) fprintf(stderr, "Not interested in %p\n", this);
    m_state.local_interested = 0; 
    if( !request_q.IsEmpty() ){
      CancelRequest(request_q.GetHead());
      request_q.Empty();
    }
    break;
  default:
    return -1;			// BUG ???
  }
  return stream.Send_State(s);
}

int btPeer::RequestPiece()
{
  size_t idx;
  int endgame = 0;

  size_t qsize = request_q.Qsize();
  size_t psize = BTCONTENT.GetPieceLength() / cfg_req_slice_size;

  // See if there's room in the queue for a new piece.
  // Also, don't queue another piece if we still have a full piece queued.
  if( cfg_req_queue_length - qsize < psize || qsize >= psize ){
    m_req_send = m_req_out;   // don't come back until you receive something.
    return 0;
  }

  if( PENDINGQUEUE.ReAssign(&request_q,bitfield) ){
    if(arg_verbose) fprintf(stderr, "Assigning to %p from Pending\n", this);
    return SendRequest();
  }

  if( m_cached_idx < BTCONTENT.GetNPieces() && !BTCONTENT.pBF->IsEmpty() ){
    // A HAVE msg already selected what we want from this peer
    // but ignore it in initial-piece mode.
    idx = m_cached_idx;
    m_cached_idx = BTCONTENT.GetNPieces();
    if( !BTCONTENT.pBF->IsSet(idx) &&
        !PENDINGQUEUE.Exist(idx) &&
        !WORLD.AlreadyRequested(idx) ){
      if(arg_verbose) fprintf(stderr, "Assigning #%u to %p\n", idx, this);
      return (request_q.CreateWithIdx(idx) < 0) ? -1 : SendRequest();
    }
  }	// If we didn't want the cached piece, select another.
  if( BTCONTENT.pBF->IsEmpty() ){
    // If we don't have a complete piece yet, try to get one that's already
    // in progress.  (Initial-piece mode)
    BitField tmpBitField = bitfield;
    idx = WORLD.What_Can_Duplicate(tmpBitField, this, BTCONTENT.GetNPieces());
    if( idx < BTCONTENT.GetNPieces() ){
      if(arg_verbose) fprintf(stderr, "Want to dup #%u to %p\n", idx, this);
      btPeer *peer = WORLD.WhoHas(idx);
      if(peer){
        if(arg_verbose) fprintf( stderr, "Duping: %p to %p (#%u)\n",
          peer, this, idx );
        return (request_q.CopyShuffle(&peer->request_q, idx) < 0) ?
          -1 : SendRequest();
      }
    }else if(arg_verbose) fprintf(stderr, "Nothing to dup to %p\n", this);
  }	// Doesn't have a piece that's already in progress--choose another.
    BitField tmpBitField;
    if( bitfield.IsFull() ){
      // peer is a seed
      tmpBitField = *BTCONTENT.pBF;
      tmpBitField.Invert();
    }else{
      tmpBitField = bitfield;
      tmpBitField.Except(*BTCONTENT.pBF);
    }
    // The filter tells what we don't want.
    tmpBitField.Except(*BTCONTENT.pBFilter);
    // tmpBitField tells what we need from this peer...

    if( !tmpBitField.IsEmpty() ){
      BitField tmpBitField2 = tmpBitField;
      WORLD.CheckBitField(tmpBitField2);
      // [tmpBitField2]... that we haven't requested from anyone.
      if(tmpBitField2.IsEmpty()){
        // Everything this peer has that I want, I've already requested.
        if( arg_file_to_download ){
          BitField afdBitField =  *BTCONTENT.pBF;
          afdBitField.Except(*BTCONTENT.pBFilter);
          endgame = ( BTCONTENT.getFilePieces(arg_file_to_download)
                      - afdBitField.Count() ) < WORLD.TotalPeers();
        }else
          endgame = ( WORLD.Pieces_I_Can_Get() - BTCONTENT.pBF->Count() )
                      < WORLD.TotalPeers();
        if(endgame){	// OK to duplicate a request.
//        idx = tmpBitField.Random();
          idx = 0;	// flag for Who_Can_Duplicate()
          BitField tmpBitField3 = tmpBitField2;
          idx = WORLD.What_Can_Duplicate(tmpBitField3, this, idx);
          if( idx < BTCONTENT.GetNPieces() ){
            if(arg_verbose) fprintf(stderr,"Want to dup #%u to %p\n",idx,this);
            btPeer *peer = WORLD.WhoHas(idx);
            if(peer){
              if(arg_verbose) fprintf( stderr, "Duping: %p to %p (#%u)\n",
                peer, this, idx );
              return (request_q.CopyShuffle(&peer->request_q, idx) < 0) ?
                -1 : SendRequest();
            }
          }else if(arg_verbose) fprintf(stderr, "Nothing to dup to %p\n",this);
        }else{	// not endgame mode
          btPeer *peer = WORLD.Who_Can_Abandon(this); // slowest choice
          if(peer){
            // Cancel a request to the slowest peer & request it from this one.
            if(arg_verbose) fprintf( stderr, "Reassigning %p to %p (#%u)\n",
              peer, this, peer->request_q.GetRequestIdx() );
            // RequestQueue class "moves" rather than "copies" in assignment!
            if( request_q.Copy(&peer->request_q) < 0 ) return -1;
            if(peer->CancelPiece() < 0 || peer->RequestCheck() < 0)
              peer->CloseConnection();
            return SendRequest();
          }else{
            if(arg_verbose) fprintf(stderr, "%p standby\n", this);
            m_standby = 1;	// nothing to do at the moment
          }
        }
      }else{
        // Request something that we haven't requested yet (most common case).
        // Try to make it something that has good trade value.
        BitField tmpBitField3 = tmpBitField2;
        WORLD.FindValuedPieces(tmpBitField3, this, BTCONTENT.pBF->IsEmpty());
        if( tmpBitField3.IsEmpty() ) tmpBitField3 = tmpBitField2;
        idx = tmpBitField3.Random();
        if(arg_verbose) fprintf(stderr, "Assigning #%u to %p\n", idx, this);
        return (request_q.CreateWithIdx(idx) < 0) ? -1 : SendRequest();
      }
    }else{
      // We don't need anything from the peer.  How'd we get here?
      return SetLocal(M_NOT_INTERESTED);
    }
  return 0;
}

int btPeer::MsgDeliver()
{
  size_t r,idx,off,len;
  int retval = 0;

  char *msgbuf = stream.in_buffer.BasePointer();

  r = get_nl(msgbuf);

  // Don't require keepalives if we're receiving other messages.
  m_last_timestamp = now;
  if( 0 == r ){
    if( !m_f_keepalive ) if( stream.Send_Keepalive() < 0 ) return -1;
    m_f_keepalive = 0;
    return 0;
  }else{
    switch(msgbuf[4]){
    case M_CHOKE:
      if(H_BASE_LEN != r){ return -1;}
      if(arg_verbose) fprintf(stderr, "%p choked me\n", this);
      if( m_lastmsg == M_UNCHOKE && m_last_timestamp <= m_choketime+1 ){
        m_err_count+=2;
        if(arg_verbose) fprintf(stderr,"err: %p (%d) Choke oscillation\n",
          this, m_err_count);
      }
      m_choketime = m_last_timestamp;
      m_state.remote_choked = 1;
      StopDLTimer();
      if( g_next_dn == this ) g_next_dn = (btPeer *)0;
      if( !request_q.IsEmpty()){
        m_req_out = 0;
        PENDINGQUEUE.Pending(&request_q);
      }
      break;

    case M_UNCHOKE:
      if(H_BASE_LEN != r){return -1;}
      if(arg_verbose) fprintf(stderr, "%p unchoked me\n", this);
      if( m_lastmsg == M_CHOKE && m_last_timestamp <= m_choketime+1 ){
        m_err_count+=2;
        if(arg_verbose) fprintf(stderr,"err: %p (%d) Choke oscillation\n",
          this, m_err_count);
      }
      m_choketime = m_last_timestamp;
      m_state.remote_choked = 0;
      retval = RequestCheck();
      break;

    case M_INTERESTED:
      if(H_BASE_LEN != r){return -1;}
      if(arg_verbose) fprintf(stderr, "%p is interested\n", this);
      m_state.remote_interested = 1;
      break;