peer.cpp

cTorrent advanced 3.3.2。是对 CTorrent 的一个改进版本。这是目前的最新版。

#include "peer.h"

#include <stdlib.h>
#include <string.h>
#include <ctype.h>
#include <sys/time.h>
#include <errno.h>

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

#if !defined(HAVE_CLOCK_GETTIME) || !defined(HAVE_SNPRINTF)
#include "compat.h"
#endif

// Convert a peer ID to a printable string.
int TextPeerID(const 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 is used to let the g_next peer object know if it skipped, as the
   socket could go non-ready if other messages are sent while waiting for bw.
*/
btPeer *btPeer::g_next_up = (btPeer *)0;
btPeer *btPeer::g_next_dn = (btPeer *)0;
unsigned char btPeer::g_defer_up = 0;

btBasic Self;

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)
{
//  CONSOLE.Debug_n("IpEquiv: %s <=> ", inet_ntoa(m_sin.sin_addr));
//  CONSOLE.Debug_n("%s", inet_ntoa(addr.sin_addr));
//  CONSOLE.Debug_n("");
  return (memcmp(&m_sin.sin_addr,&addr.sin_addr,sizeof(struct in_addr)) == 0) ? 
    1 : 0;
}

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

    if( BTCONTENT.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() const
{
  if( BTCONTENT.Seeding() || bitfield.IsEmpty() ) return 0;
  else if( bitfield.IsFull() &&
    BTCONTENT.CheckedPieces() >= BTCONTENT.GetNPieces() ) return 1;
  else{
    BitField tmpBitfield = bitfield;                // what peer has
    tmpBitfield.Except(*BTCONTENT.pBF);             // what I have
    tmpBitfield.Except(*BTCONTENT.pBMasterFilter);  // what I don't want
    tmpBitfield.And(*BTCONTENT.pBChecked);          // what I've checked
    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 = m_next_send_time = 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 = m_last_req_piece = 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_cancel_time = m_latency_timestamp = (time_t)0;
  m_bad_health = 0;
  m_want_again = m_connect = m_retried = 0;
  m_connect_seed = 0;
  m_prefetch_time = (time_t)0;
  m_requested = 0;
  m_prefetch_completion = 0;

  rate_dl.SetSelf(Self.DLRatePtr());
  rate_ul.SetSelf(Self.ULRatePtr());
}

void btPeer::CopyStats(btPeer *peer)
{
  SetDLRate(peer->GetDLRate());
  SetULRate(peer->GetULRate());
  m_unchoke_timestamp = peer->GetLastUnchokeTime();
  m_retried = peer->Retried(); // don't try to reconnect over & over
}

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) CONSOLE.Debug("Choking %p", this);
    if(arg_verbose) CONSOLE.Debug("Choking %p (D=%lluMB@%dK/s)", this,
      (unsigned long long)TotalDL() >> 20, (int)(RateDL() >> 10));
    m_state.local_choked = 1; 
    if( g_next_up == this ) g_next_up = (btPeer *)0;
    if( !reponse_q.IsEmpty()) reponse_q.Empty();
    StopULTimer();
    if( !m_requested && BTCONTENT.IsFull() ){
      // hasn't sent a request since unchoke
      if(arg_verbose) CONSOLE.Debug("%p inactive", this);
      return -1;
    }
    m_requested = 0;
    break;
  case M_UNCHOKE: 
    if( !reponse_q.IsEmpty() ) StartULTimer();
    if( !m_state.local_choked ) return 0;
    m_unchoke_timestamp = now;
//  if(arg_verbose) CONSOLE.Debug("Unchoking %p", this);
    if(arg_verbose) CONSOLE.Debug("Unchoking %p (D=%lluMB@%dK/s)", this,
      (unsigned long long)TotalDL() >> 20, (int)(RateDL() >> 10));
    m_state.local_choked = 0;
    // No data is queued, so rate cannot delay sending.
    m_next_send_time = now;
    break;
  case M_INTERESTED: 
    if( BTCONTENT.Seeding() ) return 0;
    m_standby = 0;
    if( m_state.local_interested ) return 0;
    if(arg_verbose) CONSOLE.Debug("Interested in %p", this);
    m_state.local_interested = 1;
    break;
  case M_NOT_INTERESTED:
    if( !m_state.local_interested ) return 0;
    if(arg_verbose) CONSOLE.Debug("Not interested in %p", this);
    m_state.local_interested = 0; 
    if( !request_q.IsEmpty() ){
      if( CancelRequest() < 0 ) return -1;
      request_q.Empty();
    }
    break;
  default:
    return -1;			// BUG ???
  }
  return stream.Send_State(s);
}

int btPeer::RequestPiece()
{
  size_t idx;
  BitField tmpBitfield, *pfilter;
  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;
  }

  tmpBitfield = bitfield;
  tmpBitfield.Except(*BTCONTENT.pBMasterFilter);
  if( m_last_req_piece < BTCONTENT.GetNPieces() && tmpBitfield.Count() > 1 )
    tmpBitfield.UnSet(m_last_req_piece);
  if( (idx = PENDINGQUEUE.ReAssign(&request_q, tmpBitfield)) <
      BTCONTENT.GetNPieces() ){
    if(arg_verbose)
      CONSOLE.Debug("Assigning #%d to %p from Pending", (int)idx, this);
    if( BTCONTENT.pBMultPeer->IsSet(idx) ) WORLD.CompareRequest(this, idx);
    BTCONTENT.pBMultPeer->Set(idx);
    return SendRequest();
  }

  if( m_cached_idx < BTCONTENT.CheckedPieces() && !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) &&
        (!BTCONTENT.GetFilter() || !BTCONTENT.GetFilter()->IsSet(idx)) &&
        !PENDINGQUEUE.Exist(idx) &&
        !WORLD.AlreadyRequested(idx) ){
      if(arg_verbose) CONSOLE.Debug("Assigning #%d to %p", (int)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)
    pfilter = BTCONTENT.GetFilter();
    do{
      tmpBitfield = bitfield;
      if( pfilter ){
        tmpBitfield.Except(*pfilter);
        pfilter = BTCONTENT.GetNextFilter(pfilter);
      }
    }while( pfilter && tmpBitfield.IsEmpty() );
    if( m_latency < 60 ){
      // Don't dup to very slow/high latency peers.
      if( m_last_req_piece < BTCONTENT.GetNPieces() && tmpBitfield.Count() > 1 )
        tmpBitfield.UnSet(m_last_req_piece);
      idx = WORLD.What_Can_Duplicate(tmpBitfield, this, BTCONTENT.GetNPieces());
      if( idx < BTCONTENT.GetNPieces() ){
        if(arg_verbose) CONSOLE.Debug("Want to dup #%d to %p", (int)idx, this);
        btPeer *peer = WORLD.WhoHas(idx);
        if(peer){
          if(arg_verbose) CONSOLE.Debug("Duping #%d from %p to %p",
            (int)idx, peer, this);
          if( request_q.CopyShuffle(&peer->request_q, idx) < 0 ) return -1;
          WORLD.CompareRequest(this, idx);
          BTCONTENT.pBMultPeer->Set(idx);
          return SendRequest();
        }
      }else if(arg_verbose) CONSOLE.Debug("Nothing to dup to %p", this);
    }
  }

  // Doesn't have a piece that's already in progress--choose another.
  pfilter = BTCONTENT.GetFilter();
  do{
    tmpBitfield = bitfield;
    tmpBitfield.Except(*BTCONTENT.pBF);
    if( pfilter ){
      tmpBitfield.Except(*pfilter);
      pfilter = BTCONTENT.GetNextFilter(pfilter);
    }
    // Don't go after pieces we might already have (but don't know yet)
    tmpBitfield.And(*BTCONTENT.pBChecked);
    // tmpBitfield tells what we need from this peer...
  }while( pfilter && tmpBitfield.IsEmpty() );
  if( m_last_req_piece < BTCONTENT.GetNPieces() && tmpBitfield.Count() > 1 )
    tmpBitfield.UnSet(m_last_req_piece);

  if( tmpBitfield.IsEmpty() ){
    // We don't need to request anything from the peer.
    if( !Need_Remote_Data() )
      return SetLocal(M_NOT_INTERESTED);
    else if( m_last_req_piece < BTCONTENT.GetNPieces() &&
             !BTCONTENT.pBF->IsSet(m_last_req_piece) ){
      // May have excluded the only viable request; allow a retry.
      m_last_req_piece = BTCONTENT.GetNPieces();
      return 0;  // Allow another peer a shot at it first.
    }else{
      if(arg_verbose) CONSOLE.Debug("%p standby", this);
      m_standby = 1;  // nothing to do at the moment
      return 0;
    }
  }

  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.
    int endgame = WORLD.Endgame();
    if( endgame && m_latency < 60 ){
      // OK to duplicate a request, but not to very slow/high latency peers.
//    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) CONSOLE.Debug("Want to dup #%d to %p",
          (int)idx, this);
        btPeer *peer = WORLD.WhoHas(idx);
        if(peer){  // failsafe
          if(arg_verbose) CONSOLE.Debug("Duping #%d from %p to %p",
            (int)idx, peer, this);
          if( request_q.CopyShuffle(&peer->request_q, idx) < 0 ) return -1;
          WORLD.CompareRequest(this, idx);
          BTCONTENT.pBMultPeer->Set(idx);
          return SendRequest();
        }
      }
    }
    btPeer *peer;
    if( request_q.IsEmpty() && (peer = WORLD.Who_Can_Abandon(this)) ){
      // Cancel a request to the slowest peer & request it from this one.
      idx = peer->FindLastCommonRequest(bitfield);
      if(arg_verbose) CONSOLE.Debug("Reassigning #%d from %p to %p",
        (int)idx, peer, this);
      // RequestQueue class "moves" rather than "copies" in assignment!
      if( request_q.Copy(&peer->request_q, idx) < 0 ) return -1;
      WORLD.CompareRequest(this, idx);
      BTCONTENT.pBMultPeer->Set(idx);
      if( endgame ) peer->UnStandby();
      if( peer->CancelPiece(idx) < 0 ) peer->CloseConnection();
      return SendRequest();
    }else if( BTCONTENT.CheckedPieces() >= BTCONTENT.GetNPieces() ){
      if(arg_verbose) CONSOLE.Debug("%p standby", 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) CONSOLE.Debug("Assigning #%d to %p", (int)idx, this);
    return (request_q.CreateWithIdx(idx) < 0) ? -1 : SendRequest();
  }
  return 0;
}