peerlist.cpp

在嵌入linux-arm上运行的bt下载应用程序

#include <sys/types.h>

#include "peerlist.h"

#include <stdlib.h>

#include <stdio.h>
#include <errno.h>

#include <string.h>

#include "btconfig.h"
#include "connect_nonb.h"
#include "setnonblock.h"
#include "btcontent.h"
#include "msgencode.h"

#include "iplist.h"
#include "tracker.h"
#include "ctcs.h"
#include "bttime.h"


#define MAX_UNCHOKE 3
#define UNCHOKE_INTERVAL 10

#define OPT_INTERVAL 30

#define KEEPALIVE_INTERVAL 117

#define LISTEN_PORT_MAX 2706
#define LISTEN_PORT_MIN 2106

#define PEER_IS_SUCCESS(peer) (P_SUCCESS == (peer)->GetStatus())
#define PEER_IS_FAILED(peer) (P_FAILED == (peer)->GetStatus())
#define NEED_MORE_PEERS() (m_peers_count < cfg_max_peers)

const char LIVE_CHAR[4] = {'-', '\\','|','/'};

PeerList WORLD;

PeerList::PeerList()
{
  m_unchoke_check_timestamp =
    m_keepalive_check_timestamp =
    m_opt_timestamp = time((time_t*) 0);

  m_head = (PEERNODE*) 0;
  m_listen_sock = INVALID_SOCKET;
  m_peers_count = m_seeds_count = 0;
  m_live_idx = 0;
}

PeerList::~PeerList()
{
  PEERNODE *p,*pnext;
  for(p = m_head; p ; ){
    pnext = p->next;
    delete p->peer;
    delete p;
    p = pnext;
  }
}

int PeerList::IsEmpty() const
{
  return m_peers_count ? 0 : 1;
}

void PeerList::CloseAll()
{
  PEERNODE *p;
  for(p = m_head; p;){
    m_head = p->next;
    delete (p->peer);
    delete p;
    p = m_head;
  }
}

int PeerList::NewPeer(struct sockaddr_in addr, SOCKET sk)
{
  PEERNODE *p;
  btPeer *peer = (btPeer*) 0;
  int r;

  if( m_peers_count >= cfg_max_peers ){
    if( INVALID_SOCKET != sk ) CLOSE_SOCKET(sk);
    return -4;
  }
  
  if( Self.IpEquiv(addr) ){ 
    if(INVALID_SOCKET != sk) CLOSE_SOCKET(sk); return -3;} // myself

  for(p = m_head; p; p = p->next){
    if(PEER_IS_FAILED(p->peer)) continue;
    if( p->peer->IpEquiv(addr)){  // already exist.
      if( INVALID_SOCKET != sk) CLOSE_SOCKET(sk); 
      return -3;
    }
  }
  
  if( INVALID_SOCKET == sk ){
    if( INVALID_SOCKET == (sk = socket(AF_INET,SOCK_STREAM,0)) ) return -1;
    
    if( setfd_nonblock(sk) < 0) goto err;

    if(arg_verbose) fprintf(stderr, "Connecting to %s:%hu\n",
        inet_ntoa(addr.sin_addr), ntohs(addr.sin_port));
    if( -1 == (r = connect_nonb(sk,(struct sockaddr*)&addr)) ) return -1;

    peer = new btPeer;

#ifndef WINDOWS
    if( !peer ) goto err;
#endif

    peer->SetAddress(addr);
    peer->stream.SetSocket(sk);
    peer->SetStatus( (-2 == r) ? P_CONNECTING : P_HANDSHAKE );

  }else{
    if( setfd_nonblock(sk) < 0) goto err;

    peer = new btPeer;

#ifndef WINDOWS
    if( !peer ) goto err;
#endif

    peer->SetAddress(addr);
    peer->stream.SetSocket(sk);
    peer->SetStatus(P_HANDSHAKE);
  }

  if( P_HANDSHAKE == peer->GetStatus() )
    if( peer->Send_ShakeInfo() != 0 ) { delete peer; return -1; }

  p = new PEERNODE;
#ifndef WINDOWS
  if( !p ){ delete peer; return -1;}
#endif
  m_peers_count++;

  p->peer = peer;
  
  p->next = m_head;
  m_head = p;

  return 0;
 err:
  CLOSE_SOCKET(sk);
  return -1;
}

int PeerList::FillFDSET(const time_t *pnow,fd_set *rfdp,fd_set *wfdp)
{
  PEERNODE *p;
  PEERNODE *pp = (PEERNODE*) 0;
  int f_keepalive_check = 0;
  int f_unchoke_check = 0;
  int maxfd = -1;
  int i = 0;
  SOCKET sk = INVALID_SOCKET;
  struct sockaddr_in addr;
  btPeer * UNCHOKER[MAX_UNCHOKE + 1];
  
  if( !Tracker.IsPaused() && !Tracker.IsQuitting() )
    for( ;NEED_MORE_PEERS() && !IPQUEUE.IsEmpty(); ){
      if(IPQUEUE.Pop(&addr) < 0) break;
      if(NewPeer(addr,INVALID_SOCKET) == -4) break;
    }

  Self.SetCurrentRates();

  // show status line.
  if( m_pre_dlrate.TimeUsed(pnow) ){
    char partial[30] = "";

    char *runtime_msg = new char[256];
    char *tmp_msg = runtime_msg;
    memset(runtime_msg, 0, 256);

    if(arg_file_to_download){
      BitField tmpBitField =  *BTCONTENT.pBF;
      tmpBitField.Except(*BTCONTENT.pBFilter);
      sprintf( partial, "P:%u/%u ", 
        tmpBitField.Count(),
        BTCONTENT.getFilePieces(arg_file_to_download) );
    }

    sprintf(runtime_msg, "\r                                                                               ");
    sprintf(runtime_msg, "\r%c %u/%u/%u [%u/%u/%u] %lluMB,%lluMB | %u,%uK/s | %u,%uK E:%u,%u %s%s ",
           LIVE_CHAR[m_live_idx],

           m_seeds_count,
           m_peers_count - m_seeds_count,
           Tracker.GetPeersCount(),

           BTCONTENT.pBF->Count(),
           BTCONTENT.pBF->NBits(),
           Pieces_I_Can_Get(),

           Self.TotalDL() >> 20, Self.TotalUL() >> 20,

           Self.RateDL() >> 10, Self.RateUL() >> 10,

           m_pre_dlrate.RateMeasure(Self.GetDLRate()) >> 10,
           m_pre_ulrate.RateMeasure(Self.GetULRate()) >> 10,

           Tracker.GetRefuseClick(),
           Tracker.GetOkClick(),

           partial,

           (Tracker.GetStatus()==T_CONNECTING) ? "Connecting" :
             ( (Tracker.GetStatus()==T_READY) ? "Connected" :
                 (Tracker.IsPaused() ?
                  ((Tracker.GetStatus()==T_FINISHED) ? "Paused" : "Pausing") :
                  (Tracker.IsQuitting() ? "Quitting" : "")) )
    );
    if (arg_runtime_status_fd>0) {
      write(arg_runtime_status_fd, runtime_msg, 256);
      // rewind to offset 0.
      lseek(arg_runtime_status_fd, 0, SEEK_SET);
      //sprintf(tmp_msg, "%s\n\n", runtime_msg);
      //write(arg_runtime_status_fd, tmp_msg, strlen(tmp_msg));
    } else {
      fprintf(stdout, "%s",runtime_msg);
      fflush(stdout);
    }
    delete []runtime_msg;

    m_pre_dlrate = Self.GetDLRate();
    m_pre_ulrate = Self.GetULRate();
    m_live_idx++;

    if(arg_ctcs) CTCS.Report_Status(
           m_seeds_count, m_peers_count - m_seeds_count,
           BTCONTENT.pBF->Count(), BTCONTENT.pBF->NBits(), Pieces_I_Can_Get(),
           Self.RateDL(), Self.RateUL(),
           Self.TotalDL(), Self.TotalUL(),
           cfg_max_bandwidth_down, cfg_max_bandwidth_up);
  }
    
  if(KEEPALIVE_INTERVAL <= (*pnow - m_keepalive_check_timestamp)){
    m_keepalive_check_timestamp = *pnow;
    f_keepalive_check = 1;
  }

  if(UNCHOKE_INTERVAL <= (*pnow - m_unchoke_check_timestamp)){
    
    m_unchoke_check_timestamp = *pnow;
    f_unchoke_check = 1;
    
  }

  if( f_unchoke_check ) {
    memset(UNCHOKER, 0, (MAX_UNCHOKE + 1) * sizeof(btPeer*));
    if (OPT_INTERVAL <= *pnow - m_opt_timestamp) m_opt_timestamp = 0;
  }

  m_seeds_count = 0;
  for(p = m_head; p;){
    if( PEER_IS_FAILED(p->peer)){
      if( pp ) pp->next = p->next; else m_head = p->next;
      delete p->peer;
      delete p;
      m_peers_count--;
      if( pp ) p = pp->next; else p = m_head;
      continue;
    }else{
      if (p->peer->bitfield.IsFull()) m_seeds_count++;
      if( f_keepalive_check ){

        if(3 * KEEPALIVE_INTERVAL <= (*pnow - p->peer->GetLastTimestamp())){
          if(arg_verbose) fprintf(stderr, "close: keepalive expired\n");
          p->peer->CloseConnection();
          goto skip_continue;
        }
        
        if(PEER_IS_SUCCESS(p->peer) && 
           KEEPALIVE_INTERVAL <= (*pnow - p->peer->GetLastTimestamp()) &&
           p->peer->AreYouOK() < 0){
          if(arg_verbose) fprintf(stderr, "close: keepalive death\n");
          p->peer->CloseConnection();
          goto skip_continue;
        }
      }

      if( f_unchoke_check && PEER_IS_SUCCESS(p->peer) ){

        if( p->peer->Is_Remote_Interested() && p->peer->Need_Local_Data() )
            UnChokeCheck(p->peer, UNCHOKER);
        else if(p->peer->SetLocal(M_CHOKE) < 0){
          if(arg_verbose) fprintf(stderr, "close: Can't choke peer\n");
          p->peer->CloseConnection();
          goto skip_continue;
        }
      }

      sk = p->peer->stream.GetSocket();
      if(maxfd < sk) maxfd = sk;
      if( p->peer->NeedRead() ) FD_SET(sk,rfdp);

      if( p->peer->NeedWrite() ) FD_SET(sk,wfdp);
    skip_continue: 
      pp = p;
      p = p->next;
    }
  } // end for
  

  if( INVALID_SOCKET != m_listen_sock && m_peers_count < cfg_max_peers){
    FD_SET(m_listen_sock, rfdp);
    if( maxfd < m_listen_sock ) maxfd = m_listen_sock;
  }

  if( f_unchoke_check ){
    if(arg_verbose) fprintf(stderr, "\nUnchoker ");
    if (!m_opt_timestamp){
      if(arg_verbose) fprintf(stderr, "(opt) ");
      m_opt_timestamp = *pnow;
    }
    for( i = 0; i < MAX_UNCHOKE + 1; i++){

      if( (btPeer*) 0 == UNCHOKER[i]) break;

      if( PEER_IS_FAILED(UNCHOKER[i]) ) continue;

      if(arg_verbose){
        fprintf(stderr, "D=%lluMB@%uK/s:U=%lluMB ",
          UNCHOKER[i]->TotalDL() >> 20, UNCHOKER[i]->RateDL() >> 10,
          UNCHOKER[i]->TotalUL() >> 20);
        if( UNCHOKER[i]->bitfield.IsEmpty() ) fprintf(stderr, "(empty) ");
      }
      if( UNCHOKER[i]->SetLocal(M_UNCHOKE) < 0){
        if(arg_verbose) fprintf(stderr, "close: Can't unchoke peer\n");
        UNCHOKER[i]->CloseConnection();
        continue;
      }

      sk = UNCHOKER[i]->stream.GetSocket();

      if(!FD_ISSET(sk,wfdp) && UNCHOKER[i]->NeedWrite()){
        FD_SET(sk,wfdp);
        if( maxfd < sk) maxfd = sk;
      }
    } // end for
    if(arg_verbose) fprintf(stderr, "\n");
  }

  Self.ClearCurrentRates();
  
  return maxfd;
}

btPeer* PeerList::Who_Can_Abandon(btPeer *proposer)
{
  PEERNODE *p;
  btPeer *peer = (btPeer*) 0;
  for(p = m_head; p; p = p->next){
    if(!PEER_IS_SUCCESS(p->peer) || p->peer == proposer ||
       p->peer->request_q.IsEmpty() ) continue;

    if(proposer->bitfield.IsSet(p->peer->request_q.GetRequestIdx())){
      if(!peer){
        if( p->peer->RateDL() < proposer->RateDL() ) peer = p->peer;
      }else{
        if( p->peer->RateDL() < peer->RateDL() ) peer = p->peer;
      }
    }
  }//end for
  return peer;
}

// This takes an index parameter to facilitate modification of the function to
// allow targeting of a specific piece.  It's currently only used as a flag to
// specify endgame or initial-piece mode though.
size_t PeerList::What_Can_Duplicate(BitField &bf, btPeer *proposer, size_t idx)
{
  PEERNODE *p;
  btPeer *peer = (btPeer*) 0;
  int endgame;
  PSLICE ps;
  size_t piece, piece1, qsize, mark, bench;
  unsigned long rndbits;
  int r=0;

  endgame = idx < BTCONTENT.GetNPieces();	// else initial-piece mode
  if(endgame) mark = 0;
  else mark = cfg_req_queue_length;
  bench = BTCONTENT.GetNPieces();

  // In initial mode, only dup a piece with trade value.
  // In endgame mode, dup any if there are no pieces with trade value.
  FindValuedPieces(bf, proposer, !endgame);
  if( bf.IsEmpty() ){
    if(endgame) bf = proposer->bitfield;
    else return BTCONTENT.GetNPieces();
  }

  /* In endgame mode, select from peers with the longest request queue.
     In initial mode, select from peers with the shortest non-empty request
     queue.  */
  for(p = m_head; p; p = p->next){
    if( !PEER_IS_SUCCESS(p->peer) || p->peer == proposer ) continue;

    if( p->peer->request_q.IsEmpty() ) continue;

    piece = BTCONTENT.GetNPieces();
    ps = p->peer->request_q.GetHead();
    for( ; ps; ps = ps->next){
      if( piece == ps->index || bench == ps->index || piece1 == ps->index ||
          !bf.IsSet(ps->index) || proposer->request_q.HasIdx(ps->index) )
        continue;
      piece = ps->index;

      qsize = p->peer->request_q.Qlen(piece);
      if( (endgame && qsize > mark) ||
          (!endgame && (qsize < mark || !peer)) ){
        mark = qsize;
        peer = p->peer;
        piece1 = piece;
      }else if( qsize == mark ){
        if( !r-- ){
          rndbits = random();
          r = 30;
        }
        if( bench != piece && (rndbits>>=1)&01 ){
          bench = piece1;
          peer = p->peer;
          piece1 = piece;
        }
      }
    }
  }
  return peer ? piece1 : BTCONTENT.GetNPieces();
}

void PeerList::FindValuedPieces(BitField &bf, btPeer *proposer, int initial)
{
  PEERNODE *p;
  BitField bf_all_have = bf, bf_int_have = bf,
    bf_others_have, bf_only_he_has = bf, bf_prefer;

  for(p = m_head; p; p = p->next){
    if( !PEER_IS_SUCCESS(p->peer) || p->peer == proposer ) continue;
    if( p->peer->Need_Remote_Data() )
      bf_int_have.And(p->peer->bitfield);
    bf_all_have.And(p->peer->bitfield);
    if( !initial && !p->peer->bitfield.IsFull() )
      bf_only_he_has.Except(p->peer->bitfield);
    else bf_others_have.Comb(p->peer->bitfield);
  }
  /* bf_all_have is now pertinent pieces that all peers have
     bf_int_have is pertinent pieces that all peers in which I'm interested have
     We prefer to get pieces that those peers need, if we can.  Otherwise go
     for pieces that any peer needs in hopes of future reciprocation. */
  if( !bf_int_have.IsFull() )
    bf_all_have = bf_int_have;
  bf_all_have.Invert();
  bf.And(bf_all_have); // bf is now pertinent pieces that not everyone has

  bf_prefer = initial ? bf_others_have : bf_only_he_has;