peerlist.cpp

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

  size_t idx;
  BitField rqbf, dupbf;

  for( p = m_head; p; p = p->next ){
    if( !PEER_IS_SUCCESS(p->peer) || p->peer->request_q.IsEmpty() ) continue;
    ps = p->peer->request_q.GetHead();
    idx = BTCONTENT.GetNPieces();
    for( ; ps; ps = ps->next ){
      if( idx == ps->index ) continue;
      idx = ps->index;
      if( rqbf.IsSet(idx) ) dupbf.Set(idx);
      else{
        rqbf.Set(idx);
        if( PENDINGQUEUE.Exist(idx) ) dupbf.Set(idx);
      }
    }
  }
  m_dup_req_pieces = dupbf.Count();
  CONSOLE.Debug("recalc: %d dup req pieces", (int)m_dup_req_pieces);
}

void PeerList::Tell_World_I_Have(size_t idx)
{
  PEERNODE *p;
  int f_seed = 0;

  if ( BTCONTENT.Seeding() ) f_seed = 1;

  for( p = m_head; p; p = p->next ){
    if( !PEER_IS_SUCCESS(p->peer) ) continue;

    // Don't send HAVE to seeders, except for our first piece.
    if( (!p->peer->bitfield.IsFull() || 1==BTCONTENT.pBF->Count()) &&
        p->peer->stream.Send_Have(idx) < 0) 
      p->peer->CloseConnection();

    else if( f_seed ){
      // request queue is emptied by setting not-interested state
      if( p->peer->SetLocal(M_NOT_INTERESTED) < 0 ){
        if(arg_verbose)
          CONSOLE.Debug("close: Can't set self not interested (T_W_I_H)");
        p->peer->CloseConnection();
      }
    }
  } // end for
}

int PeerList::Accepter()
{
  SOCKET newsk;
  socklen_t addrlen;
  struct sockaddr_in addr;
  addrlen = sizeof(struct sockaddr_in);
  newsk = accept(m_listen_sock,(struct sockaddr*) &addr,&addrlen);
//  CONSOLE.Debug("incoming! %s:%hu",
//    inet_ntoa(addr.sin_addr), ntohs(addr.sin_port));

  if( INVALID_SOCKET == newsk ) return -1;

  if( AF_INET != addr.sin_family || addrlen != sizeof(struct sockaddr_in) ){
    CLOSE_SOCKET(newsk);
    return -1;
  }

  if( Tracker.IsQuitting() ){
    CLOSE_SOCKET(newsk);
    return -1;
  }

  return NewPeer(addr,newsk);
}

int PeerList::Initial_ListenPort()
{
  int r = 0;
  struct sockaddr_in lis_addr;
  memset(&lis_addr,0, sizeof(sockaddr_in));
  lis_addr.sin_family = AF_INET;
  lis_addr.sin_addr.s_addr = INADDR_ANY;
  strcpy(m_listen, "n/a");

  m_listen_sock = socket(AF_INET,SOCK_STREAM,0);

  if( INVALID_SOCKET == m_listen_sock ) return -1;

  if ( cfg_listen_ip != 0 )
    lis_addr.sin_addr.s_addr = cfg_listen_ip;

  if(cfg_listen_port){
    int opt = 1;
    setsockopt(m_listen_sock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
    lis_addr.sin_port = htons(cfg_listen_port);
    if( bind(m_listen_sock, (struct sockaddr*)&lis_addr,
        sizeof(struct sockaddr_in)) == 0 ) 
      r = 1;
    else{
      opt = 0;
      setsockopt(m_listen_sock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
      CONSOLE.Warning(2, "warn, couldn't bind on specified port %d:  %s",
        cfg_listen_port, strerror(errno));
    }
  }

  if( !r && (!cfg_listen_port || cfg_listen_port > 1025) ){
    r = -1;
    if(cfg_listen_port){
      cfg_min_listen_port = cfg_listen_port -
                            (cfg_max_listen_port - cfg_min_listen_port);
      if( cfg_min_listen_port < 1025 ) cfg_min_listen_port = 1025;
      cfg_max_listen_port = cfg_listen_port;
    }
    cfg_listen_port = cfg_max_listen_port;
    for( ; r != 0; ){
      lis_addr.sin_port = htons(cfg_listen_port);
      r = bind(m_listen_sock, (struct sockaddr*)&lis_addr,
        sizeof(struct sockaddr_in));
      if(r != 0){
        cfg_listen_port--;
        if(cfg_listen_port < cfg_min_listen_port){
          CLOSE_SOCKET(m_listen_sock);
          CONSOLE.Warning(1, "error, couldn't bind port from %d to %d:  %s",
            cfg_min_listen_port, cfg_max_listen_port, strerror(errno));
          return -1;
        }
      }
    } /* end for(; r != 0;) */
  }

  if(listen(m_listen_sock,5) == -1){
    CLOSE_SOCKET(m_listen_sock);
    CONSOLE.Warning(1, "error, couldn't listen on port %d: %s",
      cfg_listen_port,strerror(errno));
    return -1;
  }

  if( setfd_nonblock(m_listen_sock) < 0){
    CLOSE_SOCKET(m_listen_sock);
    CONSOLE.Warning(1, "error, couldn't set socket to nonblock mode.");
    return -1;
  }

  snprintf(m_listen, sizeof(m_listen), "%s:%d",
    inet_ntoa(lis_addr.sin_addr), ntohs(lis_addr.sin_port));
  CONSOLE.Print("Listening on %s", m_listen);

  return 0;
}

size_t PeerList::Pieces_I_Can_Get() const
{
  BitField tmpBitField;
  return Pieces_I_Can_Get(&tmpBitField);
}

size_t PeerList::Pieces_I_Can_Get(BitField *ptmpBitField) const
{
  if( m_seeds_count > 0 || BTCONTENT.IsFull() )
    ptmpBitField->SetAll();
  else{
    PEERNODE *p;

    *ptmpBitField = *BTCONTENT.pBF;

    for( p = m_head; p && !ptmpBitField->IsFull(); p = p->next ){
      if( PEER_IS_SUCCESS(p->peer) )
        ptmpBitField->Comb(p->peer->bitfield);
    }
  }
  return ptmpBitField->Count();
}

int PeerList::AlreadyRequested(size_t idx) const
{
  PEERNODE *p;
  for( p = m_head; p; p = p->next ){
    if( !PEER_IS_SUCCESS(p->peer) || p->peer->request_q.IsEmpty()) continue;
    if( p->peer->request_q.HasIdx(idx) ) return 1;
  }
  return 0;
}

void PeerList::CheckBitField(BitField &bf)
{
  PEERNODE *p;
  PSLICE ps;
  size_t idx;
  for( p = m_head; p ; p = p->next ){
    if( !PEER_IS_SUCCESS(p->peer) || p->peer->request_q.IsEmpty()) continue;
    ps = p->peer->request_q.GetHead();
    idx = BTCONTENT.GetNPieces();
    for( ; ps; ps = ps->next ){
      if( ps->index != idx ){
        bf.UnSet(ps->index);
        idx = ps->index;
      }
    }
  }
}

void PeerList::PrintOut() const
{
  PEERNODE *p = m_head;
  struct sockaddr_in sin;
  CONSOLE.Print("PEER LIST");
  for( ; p ; p = p->next ){
        if(PEER_IS_FAILED(p->peer)) continue;
        p->peer->dump();
  }
}

void PeerList::AnyPeerReady(fd_set *rfdp, fd_set *wfdp, int *nready,
  fd_set *rfdnextp, fd_set *wfdnextp)
{
  PEERNODE *p;
  btPeer *peer;
  SOCKET sk;
  int need_check_send = 0;

  if( FD_ISSET(m_listen_sock, rfdp) ){
    (*nready)--;
    if( !Self.OntimeDL() && !Self.OntimeUL() ){
      FD_CLR(m_listen_sock,rfdnextp);
      Accepter();
    }
  }

  for( p = m_head; p && (*nready || need_check_send) ; p = p->next ){
    if( PEER_IS_FAILED(p->peer) ) continue;

    peer = p->peer;
    sk = peer->stream.GetSocket();

    if( P_SUCCESS == peer->GetStatus() ){
      if( FD_ISSET(sk,rfdp) ){
        (*nready)--;
        if( !Self.OntimeUL() ){
          FD_CLR(sk,rfdnextp);
          if( peer->RecvModule() < 0 ){
            if(arg_verbose) CONSOLE.Debug("close: receive");
            peer->CloseConnection();
          }
        }
      }
      if( !Self.OntimeDL() && !Self.OntimeUL() &&
          P_SUCCESS == peer->GetStatus() && peer->HealthCheck() < 0 ){
        if(arg_verbose) CONSOLE.Debug("close: unhealthy");
        peer->CloseConnection();
      }
      if( PEER_IS_FAILED(peer) ){
        if( FD_ISSET(sk,wfdp) ) (*nready)--;
        FD_CLR(sk,wfdnextp);
      }
    }
    if( P_SUCCESS == peer->GetStatus() ){
      if( FD_ISSET(sk,wfdp) ){
        (*nready)--;
        if( !Self.OntimeDL() ){
          FD_CLR(sk,wfdnextp);
          if( peer->SendModule() < 0 ){
            if(arg_verbose) CONSOLE.Debug("close: send");
            peer->CloseConnection();
            FD_CLR(sk,rfdnextp);
          }
          need_check_send = 1;
        }
      }else if( !Self.OntimeDL() )
        need_check_send = (peer->CheckSendStatus() && need_check_send);
    }
    else if( P_HANDSHAKE == peer->GetStatus() ){
      if( FD_ISSET(sk,rfdp) ){
        (*nready)--;
        if( !Self.OntimeDL() && !Self.OntimeUL() ){
          FD_CLR(sk,rfdnextp);
          if( peer->HandShake() < 0 ){
            if(arg_verbose) CONSOLE.Debug("close: bad handshake");
            peer->CloseConnection();
            FD_CLR(sk,wfdnextp);
          }
        }
      }
      if( FD_ISSET(sk,wfdp) ){
        (*nready)--;
        if( !Self.OntimeDL() && !Self.OntimeUL() ){
          FD_CLR(sk,wfdnextp);
          if( peer->SendModule() < 0 ){
            if(arg_verbose) CONSOLE.Debug("close: send handshake");
            peer->CloseConnection();
            FD_CLR(sk,rfdnextp);
          }
        }
      }
    }
    else if( P_CONNECTING == peer->GetStatus() ){
      if( FD_ISSET(sk,wfdp) ){
        (*nready)--; 
        if( !Self.OntimeDL() && !Self.OntimeUL() ){
          FD_CLR(sk,wfdnextp);
          if( peer->Send_ShakeInfo() < 0 ){
            if(arg_verbose) CONSOLE.Debug("close: Sending handshake");
            peer->CloseConnection();
            FD_CLR(sk,rfdnextp);
          }else peer->SetStatus(P_HANDSHAKE);
        }
        if( FD_ISSET(sk,rfdp) ) (*nready)--; 
      }else if( FD_ISSET(sk,rfdp) ){  // connect failed.
        (*nready)--; 
        if( !Self.OntimeDL() && !Self.OntimeUL() ){
          FD_CLR(sk,rfdnextp);
          if(arg_verbose) CONSOLE.Debug("close: connect failed");
          peer->CloseConnection();
          FD_CLR(sk,wfdnextp);
        }
      }
    }
  }// end for

  if( !m_ul_limited && !BandWidthLimitUp() ) m_missed_count++;
}

void PeerList::CloseAllConnectionToSeed()
{
  PEERNODE *p = m_head;
  for( ; p; p = p->next ){
    if( p->peer->bitfield.IsFull() ||
        /* Drop peers who remain uninterested, but keep recent connections.
           Peers who connected recently will resolve by bitfield exchange. */
        (PEER_IS_SUCCESS(p->peer) && !p->peer->Is_Remote_Interested() &&
          BTCONTENT.GetSeedTime() - now >= 300 &&
          !p->peer->ConnectedWhileSeed()) ){
      p->peer->DontWantAgain();
      if(arg_verbose) CONSOLE.Debug("close: seed<->seed");
      p->peer->CloseConnection();
    }
    else p->peer->stream.in_buffer.SetSize(BUF_DEF_SIZ);
  }
}

int PeerList::UnChokeCheck(btPeer* peer, btPeer *peer_array[])
{
  int i = 0;
  int cancel_idx = 0;
  btPeer *loster = (btPeer*) 0;
  int f_seed = BTCONTENT.Seeding();
  int no_opt = 0;
  unsigned long rndbits;
  int r=0;
  int retval = 0;

  if(m_opt_timestamp) no_opt = 1;
  if(f_seed) no_opt = 1 - no_opt;

// Find my 3 or 4 fastest peers.
// The m_max_unchoke+1 (4th) slot is for the optimistic unchoke when it happens.

  // Find a slot for the candidate--the slowest peer, or an available slot.
  for( cancel_idx = i = 0; i < m_max_unchoke + no_opt; i++ ){
    if((btPeer*) 0 == peer_array[i] ||
        PEER_IS_FAILED(peer_array[i]) ){	// 有空位
      cancel_idx = i; 
      break;
    }else{
      if(cancel_idx == i) continue;

      if(f_seed){
        // compare time unchoked
        if( (!peer_array[i]->Is_Local_UnChoked() &&
            (peer_array[cancel_idx]->Is_Local_UnChoked() ||
              peer_array[cancel_idx]->GetLastUnchokeTime() <
                peer_array[i]->GetLastUnchokeTime())) ||
            (peer_array[i]->Is_Local_UnChoked() &&
             peer_array[cancel_idx]->Is_Local_UnChoked() &&
             peer_array[i]->GetLastUnchokeTime() <
               peer_array[cancel_idx]->GetLastUnchokeTime()) )
          cancel_idx = i;
      }else{
        // compare download rate.
        if( peer_array[cancel_idx]->RateDL() > peer_array[i]->RateDL()
          //if equal, reciprocate to the peer we've sent less to, proportionally
          ||(peer_array[cancel_idx]->RateDL() == peer_array[i]->RateDL()
            && peer_array[cancel_idx]->TotalUL()
                / (peer_array[cancel_idx]->TotalDL()+.001)
              < peer_array[i]->TotalUL() / (peer_array[i]->TotalDL()+.001)) )
          cancel_idx = i;
      }
    }
  } // end for

  if( (btPeer*) 0 != peer_array[cancel_idx] &&
      PEER_IS_SUCCESS(peer_array[cancel_idx]) ){
    if(f_seed){
      if( (!peer_array[cancel_idx]->Is_Local_UnChoked() &&
           (peer->Is_Local_UnChoked() ||
             peer->GetLastUnchokeTime() <
               peer_array[cancel_idx]->GetLastUnchokeTime())) ||
           (peer_array[cancel_idx]->Is_Local_UnChoked() &&
            peer->Is_Local_UnChoked() &&
            peer_array[cancel_idx]->GetLastUnchokeTime() <
              peer->GetLastUnchokeTime()) ){
        loster = peer_array[cancel_idx];
        peer_array[cancel_idx] = peer;
      }else
        loster = peer;