peer.cpp

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

  int idxfound = 0;
  int retval = 0;

  if( request_q.IsEmpty() ) return 0;

  for(ps = request_q.GetHead(); ps; ps = ps->next){
    if( ps == request_q.NextSend() ) cancel = 0;
    if( idx == ps->index ){
      if( off == ps->offset && len == ps->length ){
        retval = 1;
        request_q.Remove(idx,off,len);
        if(cancel){
          if(arg_verbose) CONSOLE.Debug("Cancelling %d/%d/%d to %p",
            (int)idx, (int)off, (int)len, this);
          if(stream.Send_Cancel(idx,off,len) < 0)
            return -1;
          m_req_out--;
          if( m_req_out > cfg_req_queue_length ){
            if(arg_verbose)
              CONSOLE.Debug("ERROR@3: %p m_req_out underflow, resetting",this);
            m_req_out = 0;
          }
          if( !m_req_out && g_next_dn == this ) g_next_dn = (btPeer *)0;
          m_cancel_time = now;

          // Don't call RequestCheck() here since that could cause the slice
          // we're cancelling to be dup'd from another peer.
        }
        break;
      }
      idxfound = 1;
    }else if( idxfound ) break;
  }
  if( request_q.IsEmpty() ){ 
    StopDLTimer();
    m_standby = 0;
  }

  return retval;
}

size_t btPeer::FindLastCommonRequest(BitField &proposerbf)
{
  PSLICE ps;
  size_t idx, piece;

  idx = piece = BTCONTENT.GetNPieces();
  if( request_q.IsEmpty() ) return piece;
  ps = request_q.GetHead();
  for( ; ps; ps = ps->next ){
    if( ps->index != idx ){
      idx = ps->index;
      if( proposerbf.IsSet(idx) ) piece = idx;
    }
  }
  return piece;
}

int btPeer::ReportComplete(size_t idx, size_t len)
{
  int r;

  if( (r = BTCONTENT.APieceComplete(idx)) > 0 ){
    if(arg_verbose) CONSOLE.Debug("Piece #%d completed", (int)idx);
    PeerError(-1, "Piece completed");
    WORLD.Tell_World_I_Have(idx);
    BTCONTENT.CheckFilter();
    if( BTCONTENT.IsFull() )
      WORLD.CloseAllConnectionToSeed();
  }else if( 0 == r ){  // hash check failed
    // Don't count an error against the peer in initial or endgame mode, since
    // some slices may have come from other peers.
    if( !BTCONTENT.pBMultPeer->IsSet(idx) ){
      // The entire piece came from this peer.
      DataUnRec(BTCONTENT.GetPieceLength(idx) - len);
      if( PeerError(4, "Bad complete") < 0 ) CloseConnection();
      else{
        ResetDLTimer();  // set peer rate=0 so we don't favor for upload
        bitfield.UnSet(idx);  // don't request this piece from this peer again
      }
    }
  }
  // Need to re-download entire piece if check failed, so cleanup in any case.
  m_prefetch_completion = 0;
  if( WORLD.GetDupReqs() && BTCONTENT.pBMultPeer->IsSet(idx) ){
    if( WORLD.CancelPiece(idx) && arg_verbose )
      CONSOLE.Debug("Duplicate request cancelled in piece completion");
  }
  if( PENDINGQUEUE.Delete(idx) && arg_verbose )
    CONSOLE.Debug("Duplicate found in Pending, shouldn't be there");
  BTCONTENT.pBMultPeer->UnSet(idx);
  return r;
}

int btPeer::PieceDeliver(size_t mlen)
{
  size_t idx, off, len;
  char *msgbuf = stream.in_buffer.BasePointer();
  time_t t = (time_t)0;
  int f_accept = 0, f_requested = 0, f_success = 1, f_count = 1, f_want = 1;
  int f_complete = 0, dup = 0;

  idx = get_nl(msgbuf + H_LEN + H_BASE_LEN);
  off = get_nl(msgbuf + H_LEN + H_BASE_LEN + H_INT_LEN);
  len = mlen - H_PIECE_LEN;

  if( !request_q.IsEmpty() ){
    t = request_q.GetReqTime(idx,off,len);
    // Verify whether this is an outstanding request (not for error counting).
    PSLICE ps = request_q.GetHead();
    for( ; ps; ps = ps->next){
      if( ps == request_q.NextSend() ) break;
      if( idx==ps->index && off==ps->offset && len==ps->length ){
        f_requested = 1;
        break;
      }
    }
  }

  // If the slice is outstanding and was cancelled from this peer, accept.
  if( !f_requested && BTCONTENT.pBMultPeer->IsSet(idx) &&
      m_last_timestamp - m_cancel_time <= (m_latency ? (m_latency*2) : 60) &&
      (WORLD.HasSlice(idx, off, len) || PENDINGQUEUE.HasSlice(idx, off, len)) ){
    f_accept = dup = 1;
  }

  Self.StartDLTimer();

  if( f_requested || f_accept ){
    if(arg_verbose) CONSOLE.Debug("Receiving piece %d/%d/%d from %p",
      (int)idx, (int)off, (int)len, this);
    if( !BTCONTENT.pBF->IsSet(idx) &&
        BTCONTENT.WriteSlice(msgbuf + H_LEN + H_PIECE_LEN,idx,off,len) < 0 ){
      CONSOLE.Warning(2, "warn, WriteSlice failed; is filesystem full?");
      f_success = 0;
      // Re-queue the request, unless WriteSlice triggered flush failure
      // (then the request is already in Pending).
      if( f_requested && !BTCONTENT.FlushFailed() ){
        // This removes only the first instance; re-queued request is safe.
        request_q.Remove(idx,off,len);
        m_req_out--;
        if( RequestSlice(idx,off,len) < 0 ){
          // At least it's still queued & will go to Pending at peer close.
          if( f_count ) DataRecved(len);
          return -1;
        }
      }
    }else{  // saved or had the data
      request_q.Remove(idx,off,len);
      if( f_requested ) m_req_out--;
      // Check for & cancel requests for this slice from other peers in initial
      // and endgame modes.
      if( dup || (WORLD.GetDupReqs() && BTCONTENT.pBMultPeer->IsSet(idx)) )
        dup = WORLD.CancelSlice(idx, off, len);
      if( WORLD.GetDupReqs() || BTCONTENT.FlushFailed() )
        dup += PENDINGQUEUE.DeleteSlice(idx, off, len);
    }
  }else{  // not requested--not saved
    if( m_last_timestamp - m_cancel_time > (m_latency ? (m_latency*2) : 60) ){
      char msg[40];
      BTCONTENT.CountUnwantedBlock();
      sprintf(msg, "Unrequested piece %d/%d/%d", (int)idx, (int)off, (int)len);
      if( PeerError(1, msg) < 0 ) return -1;
      ResetDLTimer(); // set peer rate=0 so we don't favor for upload
      f_count = f_want = 0;
    }else{
      if(arg_verbose) CONSOLE.Debug("Unneeded piece %d/%d/%d from %p",
        (int)idx, (int)off, (int)len, this);
      BTCONTENT.CountDupBlock(len);
    }
    f_success = 0;
  }
  if( !m_want_again && f_want ) m_want_again = 1;

  // Determine how many outstanding requests we should maintain, roughly:
  // (request turnaround latency) / (time to transmit one slice)
  if( f_requested ){
    if(t){
      m_latency = (m_last_timestamp <= t) ? 1 : (m_last_timestamp - t);
      if(arg_verbose) CONSOLE.Debug("%p latency is %d sec (receive)",
        this, (int)m_latency);
      m_latency_timestamp = m_last_timestamp;
    }
    size_t rate;
    if( (rate = RateDL()) > len/20 && m_latency_timestamp ){
      // 20==RATE_INTERVAL from rate.cpp.  This is really just a check to see
      // if rate is measurable/usable.
      m_req_send = (size_t)( m_latency / (len / (double)rate) + 1 );
      if( m_req_send < 2 ) m_req_send = 2;
  
      // If latency increases, we will see this as a dlrate decrease.
      if( rate < m_prev_dlrate ) m_req_send++;
      else if( m_last_timestamp - m_latency_timestamp >= 30 &&
          m_req_out == m_req_send - 1 ){
        // Try to force latency measurement every 30 seconds.
        m_req_send--;
        m_latency_timestamp = m_last_timestamp;
      }
      m_prev_dlrate = rate;
    }else if (m_req_send < 5) m_req_send = 5;
  }

  /* if piece download complete. */
  if( f_success && !BTCONTENT.pBF->IsSet(idx) &&
      ( (f_requested && (request_q.IsEmpty() || !request_q.HasIdx(idx))) ||
        (f_accept && !WORLD.WhoHas(idx) && !PENDINGQUEUE.Exist(idx)) ) ){
    // Above WriteSlice may have triggered flush failure.  If data was saved,
    // slice was deleted from Pending.  If piece is incomplete, it's in Pending.
    if( !(BTCONTENT.FlushFailed() && PENDINGQUEUE.Exist(idx)) &&
        !(f_complete = ReportComplete(idx, len)) )
      f_count = 0;
  }

  // Don't count the slice in our DL total if it was unsolicited or bad.
  // (We don't owe the swarm any UL for such data.)
  if( f_count ) DataRecved(len);

  if( !f_complete && dup ) WORLD.CancelOneRequest(idx);

  if( request_q.IsEmpty() ){ 
    StopDLTimer();
    if( f_requested) m_standby = 0;
  }

  return (P_FAILED == m_status) ? -1 :
                                  (m_standby || !f_requested) ? 0 :
                                  RequestCheck();
}

// This is for re-requesting unsuccessful slices.
// Use RequestPiece for normal request queueing.
int btPeer::RequestSlice(size_t idx,size_t off,size_t len)
{
  int r;
  r = request_q.Requeue(idx,off,len);
  if( r < 0 ) return -1;
  else if( r ){
    if(stream.Send_Request(idx,off,len) < 0){ return -1; }
    m_req_out++;
    m_receive_time = now;
  }
  return 0;
}

int btPeer::RequestCheck()
{
  if( BTCONTENT.Seeding() || WORLD.IsPaused() )
    return SetLocal(M_NOT_INTERESTED);

  if( Need_Remote_Data() ){
    if(!m_state.local_interested && SetLocal(M_INTERESTED) < 0) return -1;
    if( !m_state.remote_choked ){
      if( m_req_out > cfg_req_queue_length ){
        if(arg_verbose)
          CONSOLE.Debug("ERROR@4: %p m_req_out underflow, resetting", this);
        m_req_out = 0;
      }
      if( request_q.IsEmpty() && RequestPiece() < 0 ) return -1;
      else if( m_req_out < m_req_send &&
               (m_req_out < 2 || !RateDL() ||
                1 >= (m_req_out+1) * request_q.GetRequestLen() /
                     (double)RateDL() - m_latency)
      // above formula is to try to allow delay between sending batches of reqs
        && SendRequest() < 0 ) return -1;
    }
  }else
    if(m_state.local_interested && SetLocal(M_NOT_INTERESTED) < 0) return -1;
  
  if(!request_q.IsEmpty()) StartDLTimer();
  else StopDLTimer();
  return 0;
}

void btPeer::CloseConnection()
{
  if(arg_verbose) CONSOLE.Debug("%p closed", this);
  if( P_FAILED != m_status ){
    m_status = P_FAILED;
    StopDLTimer();
    StopULTimer();
    stream.Close();
    PutPending();
  }
  if( g_next_up == this ) g_next_up = (btPeer *)0;
  if( g_next_dn == this ) g_next_dn = (btPeer *)0;
}

int btPeer::HandShake()
{
  char txtid[PEER_ID_LEN*2+3];
  ssize_t r = stream.Feed();
  if( r < 0 ){
    if(arg_verbose) CONSOLE.Debug("%p: %s", this,
      (r==-2) ? "remote closed" : strerror(errno));
    return -1;
  }
  if( (r=stream.in_buffer.Count()) < 68 ){
    // If it's not BitTorrent, don't wait around for a complete handshake.
    if( r > 20 ){  // ignore 8 reserved bytes following protocol ID
      if( memcmp(stream.in_buffer.BasePointer()+20,
          BTCONTENT.GetShakeBuffer()+20, (r<28) ? (r-20) : 8) != 0 ){
        if(arg_verbose){
          CONSOLE.Debug_n("");
          CONSOLE.Debug_n("peer %p gave 0x", this);
          for(int i=20; i<r && i<27; i++) CONSOLE.Debug_n("%2.2hx",
            (unsigned short)(unsigned char)(stream.in_buffer.BasePointer()[i]));
          CONSOLE.Debug_n(" as reserved bytes (partial)");
        }
        memcpy(stream.in_buffer.BasePointer()+20, BTCONTENT.GetShakeBuffer()+20,
          (r<28) ? (r-20) : 8);
      }
    }
    if(r && memcmp(stream.in_buffer.BasePointer(),BTCONTENT.GetShakeBuffer(),
        (r<48) ? r : 48) != 0){
      if(arg_verbose){
        CONSOLE.Debug_n("");
        CONSOLE.Debug_n("mine: 0x");
        for(int i=0; i<r && i<48; i++) CONSOLE.Debug_n("%2.2hx",
          (unsigned short)(unsigned char)(BTCONTENT.GetShakeBuffer()[i]));
        CONSOLE.Debug_n("");
        CONSOLE.Debug_n("peer: 0x");
        for(int i=0; i<r && i<48; i++) CONSOLE.Debug_n("%2.2hx",
          (unsigned short)(unsigned char)(stream.in_buffer.BasePointer()[i]));
        if( r>48 ){
          TextPeerID((unsigned char *)(stream.in_buffer.BasePointer()+48),
            txtid);
          CONSOLE.Debug("peer is %s", txtid);
        }
      }
      return -1;
    }
    return 0;
  }

  if( memcmp(stream.in_buffer.BasePointer(), BTCONTENT.GetShakeBuffer(), 68)
        == 0 ){
    if(arg_verbose) CONSOLE.Debug("peer %p is myself", this);
    if( m_connect && !cfg_public_ip ) Self.SetIp(m_sin);
    WORLD.AdjustPeersCount();
    return -1;
  }

  // If the reserved bytes differ, make them the same.
  // If they mean anything important, the handshake is likely to fail anyway.
  if( memcmp(stream.in_buffer.BasePointer()+20, BTCONTENT.GetShakeBuffer()+20,
      8) != 0 ){
    if(arg_verbose){
      CONSOLE.Debug_n("");
      CONSOLE.Debug_n("peer %p gave 0x", this);
      for(int i=20; i<27; i++) CONSOLE.Debug_n("%2.2hx",
        (unsigned short)(unsigned char)(stream.in_buffer.BasePointer()[i]));
      CONSOLE.Debug_n(" as reserved bytes" );
    }