peerlist.cpp

最经典的bittorrent协议的实现的源码

#include <sys/types.h>

#include "peerlist.h"

#include <stdlib.h>

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

#include <string.h>

#include <sys/time.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"
#include "console.h"

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

#define MIN_UNCHOKES 3
#define MIN_OPT_CYCLE 3
#define MIN_UNCHOKE_INTERVAL 10

#define KEEPALIVE_INTERVAL 117

#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 =
    m_interval_timestamp = time((time_t*) 0);
  m_unchoke_interval = MIN_UNCHOKE_INTERVAL;
  m_opt_interval = MIN_OPT_CYCLE * MIN_UNCHOKE_INTERVAL;

  m_head = m_dead = (PEERNODE*) 0;
  m_listen_sock = INVALID_SOCKET;
  m_peers_count = m_seeds_count = m_conn_count = 0;
  m_f_pause = 0;
  m_max_unchoke = MIN_UNCHOKES;
  m_defer_count = m_missed_count = 0;
  m_upload_count = m_up_opt_count = 0;
  m_prev_limit_up = cfg_max_bandwidth_up;
}

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

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, *pp, *pnext;
  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) ){  // myself
    if( INVALID_SOCKET != sk ) CLOSE_SOCKET(sk);
    return -3;
  }

  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;
    }
  }

  // See if we've had this peer before, and maintain its stats.
  // Do it here instead of later to insure we purge old entries periodically.
  pp = (PEERNODE *)0;
  for( p = m_dead; p; p = pnext ){
    if( p->peer->IpEquiv(addr) ) break;
    else{
      pnext = p->next;
      if( p->peer->GetLastTimestamp() + 2 * Tracker.GetInterval() < now ){
        delete p->peer;
        if( pp ) pp->next = p->next;
        else m_dead = p->next;
        delete p;
      }else pp = p;
    }
  }

  if( INVALID_SOCKET == sk ){
    if( INVALID_SOCKET == (sk = socket(AF_INET,SOCK_STREAM,0)) ) return -1;
    
    if( setfd_nonblock(sk) < 0) goto err;

    if( -1 == (r = connect_nonb(sk,(struct sockaddr*)&addr)) ){
      if(arg_verbose) CONSOLE.Debug("Connect to peer at %s:%hu failed",
        inet_ntoa(addr.sin_addr), ntohs(addr.sin_port));
      return -1;
    }

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

    peer->SetConnect();
    peer->SetAddress(addr);
    peer->stream.SetSocket(sk);
    peer->SetStatus( (-2 == r) ? P_CONNECTING : P_HANDSHAKE );
    if(arg_verbose) CONSOLE.Debug("Connecting to %s:%hu (peer %p)",
        inet_ntoa(addr.sin_addr), ntohs(addr.sin_port), peer);

  }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( !BTCONTENT.Seeding() &&
      peer->stream.in_buffer.SetSize(BUF_DEF_SIZ + cfg_req_slice_size) < 0 )
    goto err;

  if( P_HANDSHAKE == peer->GetStatus() )
    if( peer->Send_ShakeInfo() != 0 ) goto err;

  if( p ){   // resurrected! (reconnected with an old peer)
    if( pp ) pp->next = p->next;
    else m_dead = p->next;
    peer->CopyStats(p->peer);
    delete p->peer;
  }else{
    p = new PEERNODE;
#ifndef WINDOWS
    if( !p ) goto err;
#endif
  }

  m_peers_count++;
  p->peer = peer;
  p->next = m_head;
  m_head = p;
  return 0;

 err:
  if( peer ) delete peer;
  if( INVALID_SOCKET != sk ) CLOSE_SOCKET(sk);
  return -1;
}

int PeerList::IntervalCheck(fd_set *rfdp, fd_set *wfdp)
{
  int f_keepalive_check = 0;
  int f_unchoke_check = 0;
  int i = 0;
  btPeer **UNCHOKER;

  // No pause check here--stay ready by continuing to acquire peers.
  if( !Tracker.IsQuitting() ){
    struct sockaddr_in addr;
    for( ; NEED_MORE_PEERS() && !IPQUEUE.IsEmpty(); ){
      if(IPQUEUE.Pop(&addr) < 0) break;
      if(NewPeer(addr,INVALID_SOCKET) == -4) break;
    }
  }

  m_ul_limited = BandWidthLimitUp(Self.LateUL());

  // After seeding a while, disconnect uninterested peers & shrink in_buffers.
  if( now - BTCONTENT.GetSeedTime() <= 301 &&
      now - BTCONTENT.GetSeedTime() >= 300 )
    CloseAllConnectionToSeed();
    
  if( KEEPALIVE_INTERVAL <= now - m_keepalive_check_timestamp ){
    m_keepalive_check_timestamp = now;
    f_keepalive_check = 1;
  }

  if( m_unchoke_interval <= now - m_unchoke_check_timestamp && m_head &&
      !m_f_pause ){
    f_unchoke_check = 1;

    if( m_missed_count > m_upload_count && cfg_max_bandwidth_up ){
      size_t unchokes = GetUnchoked();  // already adds one (opt)
      if( unchokes < MIN_UNCHOKES ) m_max_unchoke = MIN_UNCHOKES;
      else{
        m_max_unchoke = unchokes;
        if(arg_verbose)
          CONSOLE.Debug("max unchokes up to %d", (int)m_max_unchoke);
      }
    }else if(arg_verbose) CONSOLE.Debug("UL missed %d sending %d",
      (int)m_missed_count, (int)m_upload_count);
    m_up_opt_count += m_upload_count;
    m_missed_count = m_upload_count = 0;

    if( m_opt_interval && m_opt_interval <= now - m_opt_timestamp ){
      m_opt_timestamp = 0;
      if( m_defer_count > m_up_opt_count &&
          m_max_unchoke > MIN_UNCHOKES && cfg_max_bandwidth_up ){
        m_max_unchoke--;
        if(arg_verbose)
          CONSOLE.Debug("max unchokes down to %d", (int)m_max_unchoke);
      }else if(arg_verbose) CONSOLE.Debug("UL deferred %d sending %d",
        (int)m_defer_count, (int)m_up_opt_count);
      m_defer_count = m_up_opt_count = 0;
    }

    if( 0==cfg_max_bandwidth_up ) m_max_unchoke = MIN_UNCHOKES;

    UNCHOKER = new btPeer *[m_max_unchoke + 1];
    if( UNCHOKER ) memset(UNCHOKER, 0, (m_max_unchoke + 1) * sizeof(btPeer*));
    else CONSOLE.Warning(1, "warn, failed to allocate unchoke array.");

    SetUnchokeIntervals();
  }else{  // no unchoke check
    if( now < m_unchoke_check_timestamp ) m_unchoke_check_timestamp = now;
    if( MIN_UNCHOKE_INTERVAL <= now - m_interval_timestamp ){
      m_interval_timestamp = now;
      // If up bw limit has changed enough, recompute the intervals.
      // This is primarily to prevent a low limit from delaying an unchoke for
      // a long time even after the limit has been increased.
      if( !BandWidthLimitUp() ||
          ( m_prev_limit_up &&
            abs((int)cfg_max_bandwidth_up - (int)m_prev_limit_up) /
                (double)m_prev_limit_up  >
              1 / (double)m_unchoke_interval &&
            ( cfg_max_bandwidth_up < cfg_req_slice_size * (MIN_OPT_CYCLE-1) /
                                     (MIN_UNCHOKE_INTERVAL * MIN_OPT_CYCLE) ||
              m_prev_limit_up < cfg_req_slice_size * (MIN_OPT_CYCLE-1) /
                                (MIN_UNCHOKE_INTERVAL * MIN_OPT_CYCLE) ) ) ){
        SetUnchokeIntervals();
      }
    }else if( now < m_interval_timestamp ) m_interval_timestamp = now;
  }
  return FillFDSet(rfdp, wfdp, f_keepalive_check, f_unchoke_check, UNCHOKER);
}

int PeerList::FillFDSet(fd_set *rfdp, fd_set *wfdp, int f_keepalive_check,
  int f_unchoke_check, btPeer **UNCHOKER)
{
  PEERNODE *p;
  PEERNODE *pp = (PEERNODE*) 0;
  int maxfd = -1;
  SOCKET sk = INVALID_SOCKET;

  m_seeds_count = 0;
  m_conn_count = 0;
  size_t interested_count = 0;
  for( p = m_head; p; ){
    sk = p->peer->stream.GetSocket();
    if( PEER_IS_FAILED(p->peer) ){
      if( sk != INVALID_SOCKET ){
        FD_CLR(sk,rfdp);
        FD_CLR(sk,wfdp);
      }
      if( p->peer->CanReconnect() ){ // connect to this peer again
        if(arg_verbose) CONSOLE.Debug("Adding %p for reconnect", p->peer);
        p->peer->Retry();
        struct sockaddr_in addr;
        p->peer->GetAddress(&addr);
        IPQUEUE.Add(&addr);
      }
      if( pp ) pp->next = p->next; else m_head = p->next;
      if( p->peer->TotalDL() || p->peer->TotalUL() ){  // keep stats
        p->peer->SetLastTimestamp();
        p->next = m_dead;
        m_dead = p;
      }else{
        delete p->peer;
        delete p;
      }
      m_peers_count--;
      if( pp ) p = pp->next; else p = m_head;
      continue;
    }else{
      if( !PEER_IS_SUCCESS(p->peer) ) m_conn_count++;
      else{
        if( p->peer->bitfield.IsFull() ) m_seeds_count++;
        if( p->peer->Is_Local_Interested() ) interested_count++;
      }
      if( f_keepalive_check ){
        if( 3 * KEEPALIVE_INTERVAL <= now - p->peer->GetLastTimestamp() ){
          if(arg_verbose) CONSOLE.Debug("close: keepalive expired");
          p->peer->CloseConnection();
          goto skip_continue;
        }
        if( PEER_IS_SUCCESS(p->peer) && 
            KEEPALIVE_INTERVAL <= now - p->peer->GetLastTimestamp() &&
            p->peer->AreYouOK() < 0 ){
          if(arg_verbose) CONSOLE.Debug("close: keepalive death");
          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() ){
          if( UNCHOKER && UnChokeCheck(p->peer, UNCHOKER) < 0 )
            goto skip_continue;
        }else if(p->peer->SetLocal(M_CHOKE) < 0){
          if(arg_verbose) CONSOLE.Debug("close: Can't choke peer");
          p->peer->CloseConnection();
          goto skip_continue;
        }
      }

      if( PEER_IS_FAILED(p->peer) ) goto skip_continue;  // failsafe
      if(maxfd < sk) maxfd = sk;
      if( !FD_ISSET(sk,rfdp) && p->peer->NeedRead() ) FD_SET(sk,rfdp);
      if( !FD_ISSET(sk,wfdp) && p->peer->NeedWrite() ) FD_SET(sk,wfdp);

      if( p->peer->Is_Local_UnChoked() && cfg_cache_size && !m_f_pause &&
          IsIdle() )
        p->peer->Prefetch(m_unchoke_check_timestamp + m_unchoke_interval);

    skip_continue: 
      if( PEER_IS_FAILED(p->peer) ){
        FD_CLR(sk,rfdp);
        FD_CLR(sk,wfdp);
      }
      pp = p;
      p = p->next;
    }
  } // end for

  if( 0==interested_count ) Self.StopDLTimer();

  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 && UNCHOKER ){
    m_unchoke_check_timestamp = now;  // time of the last unchoke check
    if (!m_opt_timestamp) m_opt_timestamp = now;

    if( !UNCHOKER[0] ) Self.StopULTimer();

    for( int i = 0; i < m_max_unchoke + 1; i++ ){
      if( !UNCHOKER[i] ) break;

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

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

      if( UNCHOKER[i]->SetLocal(M_UNCHOKE) < 0 ){
        if(arg_verbose) CONSOLE.Debug("close: Can't unchoke peer");
        UNCHOKER[i]->CloseConnection();
        FD_CLR(sk,rfdp);
        FD_CLR(sk,wfdp);
        continue;
      }

      if( !FD_ISSET(sk,wfdp) && UNCHOKER[i]->NeedWrite() ){
        FD_SET(sk,wfdp);
        if( maxfd < sk) maxfd = sk;
      }
    } // end for
    delete []UNCHOKER;
  }

  return maxfd;
}

void PeerList::SetUnchokeIntervals()
{
  time_t old_unchoke_int = m_unchoke_interval, old_opt_int = m_opt_interval;

  // Unchoke peers long enough to have a chance at getting some data.
  if( BandWidthLimitUp() && BTCONTENT.Seeding() ){
    int optx = (int)( 1 / (1 - (double)MIN_UNCHOKE_INTERVAL *
                               cfg_max_bandwidth_up / cfg_req_slice_size) );
    if( optx < 0 ) optx = 0;
    if( optx < MIN_OPT_CYCLE ){
      optx = MIN_OPT_CYCLE;
      double interval = cfg_req_slice_size /
           (cfg_max_bandwidth_up * MIN_OPT_CYCLE / (double)(MIN_OPT_CYCLE-1));
      m_unchoke_interval = (size_t)interval;
      if( interval - (int)interval > 0 ) m_unchoke_interval++;
      if( m_unchoke_interval < MIN_UNCHOKE_INTERVAL )
        m_unchoke_interval = MIN_UNCHOKE_INTERVAL;
    }else{
      // Allow each peer at least 60 seconds unchoked.
      m_unchoke_interval = MIN_UNCHOKE_INTERVAL;
      if( m_max_unchoke+1 < 60 / m_unchoke_interval ){
        int maxopt = (int)( 1 / (1 - (double)(m_max_unchoke+1) *
                                     m_unchoke_interval / 60) );
        if( maxopt > MIN_OPT_CYCLE && optx > maxopt ) optx = maxopt;
      }
      if( optx > m_max_unchoke+2 ) optx = m_max_unchoke+2;
    }
    m_opt_interval = optx * m_unchoke_interval;
  }else if( BandWidthLimitUp() && !BTCONTENT.Seeding() ){
    // Need to be able to upload a slice per interval.
    double interval = cfg_req_slice_size / (double)cfg_max_bandwidth_up;
    m_unchoke_interval = (size_t)interval;
    if( interval - (int)interval > 0 ) m_unchoke_interval++;
    if( m_unchoke_interval < MIN_UNCHOKE_INTERVAL )
      m_unchoke_interval = MIN_UNCHOKE_INTERVAL;