peerlist.cpp

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

#include "peerlist.h"  // def.h

#include <sys/types.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) || \
    !defined(HAVE_RANDOM)
#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 = m_downloads = 0;
  m_f_pause = m_f_dlate = m_f_ulate = m_endgame = 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;
  m_dup_req_pieces = 0;
}

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( INVALID_SOCKET != sk && Self.IpEquiv(addr) ){
    if(arg_verbose)
      CONSOLE.Debug("Connection from myself %s", inet_ntoa(addr.sin_addr));
    Tracker.AdjustPeersCount();
    if( INVALID_SOCKET != sk ) CLOSE_SOCKET(sk);
    return -3;
  }

  for( p = m_head; p; p = p->next ){
    if( !PEER_IS_FAILED(p->peer) && p->peer->IpEquiv(addr) ){
      if(arg_verbose) CONSOLE.Debug("Connection from duplicate peer %s",
        inet_ntoa(addr.sin_addr));
      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:  %s",
        inet_ntoa(addr.sin_addr), ntohs(addr.sin_port), strerror(errno));
      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(arg_verbose) CONSOLE.Debug("Connection from %s:%hu (peer %p)",
        inet_ntoa(addr.sin_addr), ntohs(addr.sin_port), peer);
  }

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

  if( cfg_cache_size && !m_f_pause && IsIdle() ){
    int f_idle = 1;
    for( PEERNODE *p = m_head; p; p = p->next ){
      if( p->peer->NeedPrefetch() ){
        if( f_idle || IsIdle() ){
          p->peer->Prefetch(m_unchoke_check_timestamp + m_unchoke_interval);
          time(&now);
          f_idle = 0;
        }else break;
      }
    }
  }

  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, *pp;
  int maxfd = -1;
  SOCKET sk = INVALID_SOCKET;

  m_f_limitu = BandWidthLimitUp(Self.LateUL());
  m_f_limitd = BandWidthLimitDown(Self.LateDL());

 again:
  pp = (PEERNODE*) 0;
  m_seeds_count = m_conn_count = m_downloads = 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( p->peer->Is_Remote_UnChoked() ) m_downloads++;
        }
      }
      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((int)m_f_limitd) )
        FD_SET(sk,rfdp);
      if( !FD_ISSET(sk,wfdp) && p->peer->NeedWrite((int)m_f_limitu) )
        FD_SET(sk,wfdp);

    skip_continue: 
      if( PEER_IS_FAILED(p->peer) ){
        FD_CLR(sk,rfdp);
        FD_CLR(sk,wfdp);
      }
      pp = p;
      p = p->next;
    }
  } // end for
  if( (m_f_limitu && !(m_f_limitu = BandWidthLimitUp(Self.LateUL()))) ||
      (m_f_limitd && !(m_f_limitd = BandWidthLimitDown(Self.LateDL()))) )
    goto again;

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