user/ctorrent/peer.cpp

   1 #include "peer.h"
   2
   3 #include <stdlib.h>
   4 #include <string.h>
   5
   6 #include "./btcontent.h"
   7 #include "./msgencode.h"
   8 #include "./peerlist.h"
   9 #include "./btconfig.h"
  10
  11 btBasic Self;
  12
  13 void btBasic::SetIp(struct sockaddr_in addr)
  14 {
  15   memcpy(&m_sin.sin_addr,&addr.sin_addr,sizeof(struct in_addr));
  16 }
  17
  18 void btBasic::SetAddress(struct sockaddr_in addr)
  19 {
  20   memcpy(&m_sin,&addr,sizeof(struct sockaddr_in));
  21 }
  22
  23 int btBasic::IpEquiv(struct sockaddr_in addr)
  24 {
  25   return (memcmp(&m_sin.sin_addr,&addr.sin_addr,sizeof(struct in_addr)) == 0) ?
  26     1 : 0;
  27 }
  28
  29 int btPeer::Need_Local_Data()
  30 {
  31   if( m_state.remote_interested && !bitfield.IsFull()){
  32
  33     if( BTCONTENT.pBF->IsFull() ) return 1; // i am seed
  34
  35     BitField tmpBitfield = *BTCONTENT.pBF;
  36     tmpBitfield.Except(bitfield);
  37     return tmpBitfield.IsEmpty() ? 0 : 1;
  38
  39   }
  40   return 0;
  41 }
  42
  43 int btPeer::Need_Remote_Data()
  44 {
  45   if( BTCONTENT.pBF->IsFull()) return 0;
  46   else if( bitfield.IsFull() ) return 1;
  47   else{
  48     BitField tmpBitfield = bitfield;
  49     tmpBitfield.Except(*BTCONTENT.pBF);
  50     return tmpBitfield.IsEmpty() ? 0 : 1;
  51   }
  52   return 0;
  53 }
  54
  55 btPeer::btPeer()
  56 {
  57   m_f_keepalive = 0;
  58   m_status = P_CONNECTING;
  59   m_unchoke_timestamp = (time_t) 0;
  60   time(&m_last_timestamp);
  61   m_state.remote_choked = m_state.local_choked = 1;
  62   m_state.remote_interested = m_state.local_interested = 0;
  63
  64   m_err_count = 0;
  65   m_cached_idx = BTCONTENT.GetNPieces();
  66 }
  67
  68 int btPeer::SetLocal(unsigned char s)
  69 {
  70   switch(s){
  71   case M_CHOKE:
  72     if( m_state.local_choked ) return 0;
  73     m_state.local_choked = 1;
  74     break;
  75   case M_UNCHOKE:
  76     if( !reponse_q.IsEmpty() ) StartULTimer();
  77     if( !m_state.local_choked ) return 0;
  78     time(&m_unchoke_timestamp);
  79     m_state.local_choked = 0;
  80     break;
  81   case M_INTERESTED:
  82     if( m_state.local_interested ) return 0;
  83     m_state.local_interested = 1;
  84     break;
  85   case M_NOT_INTERESTED:
  86     if( !m_state.local_interested ) return 0;
  87     m_state.local_interested = 0;
  88     break;
  89   default:
  90     return -1;                  // BUG ???
  91   }
  92   return stream.Send_State(s);
  93 }
  94
  95 int btPeer::RequestPiece()
  96 {
  97   size_t idx;
  98
  99   PENDINGQUEUE.ReAssign(&request_q,bitfield);
 100
 101   if( !request_q.IsEmpty() ) return SendRequest();
 102
 103   if( m_cached_idx < BTCONTENT.GetNPieces() ){
 104     idx = m_cached_idx;
 105     m_cached_idx = BTCONTENT.GetNPieces();
 106     if( !BTCONTENT.pBF->IsSet(idx) &&
 107         !PENDINGQUEUE.Exist(idx) &&
 108         !WORLD.AlreadyRequested(idx) ){
 109       return (request_q.CreateWithIdx(idx) < 0) ? -1 : SendRequest();
 110     }
 111   }else{
 112     BitField tmpBitField;
 113     if( bitfield.IsFull() ){
 114       tmpBitField = *BTCONTENT.pBF;
 115       tmpBitField.Invert();
 116     }else{
 117       tmpBitField = bitfield;
 118       tmpBitField.Except(*BTCONTENT.pBF);
 119     }
 120
 121     if( !tmpBitField.IsEmpty() ){
 122       WORLD.CheckBitField(tmpBitField);
 123       if(tmpBitField.IsEmpty()){
 124
 125         btPeer *peer = WORLD.Who_Can_Abandon(this);
 126         if(peer){
 127           peer->StopDLTimer();
 128           request_q = peer->request_q;
 129
 130           if(peer->CancelRequest(request_q.GetHead()) < 0 ||
 131              peer->RequestCheck() < 0){
 132             peer->CloseConnection();
 133           }
 134
 135           return SendRequest();
 136         }
 137
 138       }else{
 139         idx = tmpBitField.Random();
 140         return (request_q.CreateWithIdx(idx) < 0) ? -1 : SendRequest();
 141       }
 142     }
 143   }
 144   return 0;
 145 }
 146
 147 int btPeer::MsgDeliver()
 148 {
 149   size_t r,idx,off,len;
 150
 151   char *msgbuf = stream.in_buffer.BasePointer();
 152
 153   r = ntohl(*(size_t*) msgbuf);
 154
 155   if( 0 == r ){
 156     time(&m_last_timestamp);
 157     if( !m_f_keepalive ) if( stream.Send_Keepalive() < 0 ) return -1;
 158     m_f_keepalive = 0;
 159     return (!m_state.remote_choked && request_q.IsEmpty()) ? RequestCheck() : 0;
 160   }else{
 161     switch(msgbuf[4]){
 162     case M_CHOKE:
 163       if(H_BASE_LEN != r){ return -1;}
 164       m_state.remote_choked = 1;
 165       StopDLTimer();
 166       if( !request_q.IsEmpty()){
 167         PSLICE ps = request_q.GetHead();
 168         PENDINGQUEUE.Pending(&request_q);
 169         if( CancelRequest(ps) < 0) return -1;
 170       }
 171       return 0;
 172     case M_UNCHOKE:
 173       if(H_BASE_LEN != r){return -1;}
 174       m_state.remote_choked = 0;
 175       return RequestCheck();
 176
 177     case M_INTERESTED:
 178       if(H_BASE_LEN != r){return -1;}
 179       m_state.remote_interested = 1;
 180       break;
 181
 182     case M_NOT_INTERESTED:
 183       if(r != H_BASE_LEN){return -1;}
 184
 185       m_state.remote_interested = 0;
 186       StopULTimer();
 187
 188       /* remove peer's reponse queue */
 189       if( !reponse_q.IsEmpty()) reponse_q.Empty();
 190       return 0;
 191     case M_HAVE:
 192       if(H_HAVE_LEN != r){return -1;}
 193
 194       idx = ntohl(*(size_t*) (msgbuf + 5));
 195
 196       if( idx >= BTCONTENT.GetNPieces() || bitfield.IsSet(idx)) return -1;
 197
 198       bitfield.Set(idx);
 199
 200       if( bitfield.IsFull() && BTCONTENT.pBF->IsFull() ){ return -2; }
 201
 202       if( !BTCONTENT.pBF->IsSet(idx) ) m_cached_idx = idx;
 203
 204       return ( !m_state.remote_choked && request_q.IsEmpty() ) ? RequestCheck() : 0;
 205
 206     case M_REQUEST:
 207       if(H_REQUEST_LEN != r || !m_state.remote_interested){ return -1; }
 208
 209       idx = ntohl(*(size_t*)(msgbuf + 5));
 210
 211       if( !BTCONTENT.pBF->IsSet(idx) ) return -1;
 212
 213       off = ntohl(*(size_t*)(msgbuf + 9));
 214       len = ntohl(*(size_t*)(msgbuf + 13));
 215
 216       if( !reponse_q.IsValidRequest(idx, off, len) ) return -1;
 217
 218       return reponse_q.Add(idx, off, len);
 219
 220     case M_PIECE:
 221       if( request_q.IsEmpty() || !m_state.local_interested){
 222         m_err_count++;
 223         return 0;
 224       }
 225       return PieceDeliver(r);
 226
 227     case M_BITFIELD:
 228       if( (r - 1) != bitfield.NBytes() || !bitfield.IsEmpty()) return -1;
 229       bitfield.SetReferBuffer(msgbuf + 5);
 230       if(bitfield.IsFull() && BTCONTENT.pBF->IsFull()) return -2;
 231       return 0;
 232
 233     case M_CANCEL:
 234       if(r != H_CANCEL_LEN || !m_state.remote_interested) return -1;
 235
 236       idx = ntohl(*(size_t*)(msgbuf + 5));
 237       off = ntohl(*(size_t*)(msgbuf + 9));
 238       len = ntohl(*(size_t*)(msgbuf + 13));
 239       if( reponse_q.Remove(idx,off,len) < 0 ){
 240         m_err_count++;
 241         return 0;
 242       }
 243       if( reponse_q.IsEmpty() ) StopULTimer();
 244       return 0;
 245     default:
 246       return -1;                // unknow message type
 247     }
 248   }
 249   return 0;
 250 }
 251
 252 int btPeer::ReponseSlice()
 253 {
 254   size_t len = 0;
 255
 256   reponse_q.Peek((size_t*) 0,(size_t*) 0, &len);
 257
 258   if(len && stream.out_buffer.LeftSize() <= (len + 13 + 3 * 1024))
 259     stream.Flush();
 260
 261   if(len && stream.out_buffer.LeftSize() > (len + 13 + 3 * 1024)){
 262     size_t idx,off;
 263     reponse_q.Pop(&idx,&off,(size_t *) 0);
 264
 265     if(BTCONTENT.ReadSlice(BTCONTENT.global_piece_buffer,idx,off,len) != 0 ){
 266       return -1;
 267     }
 268
 269     Self.DataSended(len);
 270     DataSended(len);
 271     return stream.Send_Piece(idx,off,BTCONTENT.global_piece_buffer,len);
 272   }
 273
 274   return 0;
 275 }
 276
 277 int btPeer::SendRequest()
 278 {
 279   PSLICE ps = request_q.GetHead();
 280   for( ; ps ; ps = ps->next )
 281     if(stream.Send_Request(ps->index,ps->offset,ps->length) < 0){ return -1; }
 282
 283   return stream.Flush();
 284 }
 285
 286 int btPeer::CancelRequest(PSLICE ps)
 287 {
 288   for( ; ps; ps = ps->next){
 289     if(stream.Send_Cancel(ps->index,ps->offset,ps->length) < 0)
 290       return -1;
 291   }
 292   return stream.Flush();
 293 }
 294
 295 int btPeer::ReportComplete(size_t idx)
 296 {
 297   if( BTCONTENT.APieceComplete(idx) ){
 298     WORLD.Tell_World_I_Have(idx);
 299     if( BTCONTENT.pBF->IsFull() ){
 300       ResetDLTimer();
 301       WORLD.CloseAllConnectionToSeed();
 302     }
 303   }else
 304     m_err_count++;
 305   return (P_FAILED == m_status) ? -1 : RequestCheck();
 306 }
 307
 308 int btPeer::PieceDeliver(size_t mlen)
 309 {
 310   size_t idx,off,len;
 311   char *msgbuf = stream.in_buffer.BasePointer();
 312
 313   idx = ntohl(*(size_t*) (msgbuf + 5));
 314   off = ntohl(*(size_t*) (msgbuf + 9));
 315   len = mlen - 9;
 316
 317   if( request_q.Remove(idx,off,len) < 0 ){
 318     m_err_count++;
 319     return 0;
 320   }
 321
 322   if(BTCONTENT.WriteSlice((char*)(msgbuf + 13),idx,off,len) < 0){
 323     return 0;
 324   }
 325
 326   Self.StartDLTimer();
 327   Self.DataRecved(len);
 328   DataRecved(len);
 329
 330   /* if piece download complete. */
 331   return request_q.IsEmpty() ? ReportComplete(idx) : 0;
 332 }
 333
 334 int btPeer::RequestCheck()
 335 {
 336   if( BandWidthLimit() ) return 0;
 337
 338   if( BTCONTENT.pBF->IsFull() ){
 339     if( bitfield.IsFull() ){ return -1; }
 340     return SetLocal(M_NOT_INTERESTED);
 341   }
 342
 343   if( Need_Remote_Data() ){
 344     if(!m_state.local_interested && SetLocal(M_INTERESTED) < 0) return -1;
 345     if(request_q.IsEmpty() && !m_state.remote_choked){
 346       if( RequestPiece() < 0 ) return -1;
 347     }
 348   }
 349
 350   if(!request_q.IsEmpty()) StartDLTimer();
 351   return 0;
 352 }
 353
 354 void btPeer::CloseConnection()
 355 {
 356   if( P_FAILED != m_status ){
 357     m_status = P_FAILED;
 358     stream.Close();
 359   }
 360 }
 361
 362 int btPeer::HandShake()
 363 {
 364   ssize_t r = stream.Feed();
 365   if( r < 0 ) return -1;
 366   else if( r < 68 ){
 367     if(r && memcmp(stream.in_buffer.BasePointer(),BTCONTENT.GetShakeBuffer(),r) != 0) return -1;
 368     return 0;
 369   }
 370
 371   if( memcmp(stream.in_buffer.BasePointer(),BTCONTENT.GetShakeBuffer(),48) != 0 ) return -1;
 372
 373   // ignore peer id verify
 374   if( !BTCONTENT.pBF->IsEmpty()){
 375     char *bf = new char[BTCONTENT.pBF->NBytes()];
 376 #ifndef WINDOWS
 377     if(!bf) return -1;
 378 #endif
 379     BTCONTENT.pBF->WriteToBuffer(bf);
 380     r = stream.Send_Bitfield(bf,BTCONTENT.pBF->NBytes());
 381     delete []bf;
 382   }
 383
 384   if( r >= 0){
 385     if( stream.in_buffer.PickUp(68) < 0 ) return -1;
 386     m_status = P_SUCCESS;
 387   }
 388   return r;
 389 }
 390
 391 int btPeer::Send_ShakeInfo()
 392 {
 393   return stream.Send_Buffer((char*)BTCONTENT.GetShakeBuffer(),68);
 394 }
 395
 396 int btPeer::BandWidthLimit()
 397 {
 398   if( cfg_max_bandwidth <= 0 ) return 0;
 399   return ((Self.RateDL() + Self.RateUL()*2) / 1024 >= cfg_max_bandwidth) ?
 400     1:0;
 401 }
 402
 403 int btPeer::NeedWrite()
 404 {
 405   int yn = 0;
 406   if( stream.out_buffer.Count() || // data need send in buffer.
 407       (!reponse_q.IsEmpty() && CouldReponseSlice() && !BandWidthLimit()) ||
 408       P_CONNECTING == m_status ) // peer is connecting
 409     yn = 1;
 410   return yn;
 411 }
 412
 413 int btPeer::CouldReponseSlice()
 414 {
 415   if(!m_state.local_choked &&
 416      (stream.out_buffer.LeftSize() > reponse_q.GetRequestLen() + 4 * 1024 ))
 417     return 1;
 418   return 0;
 419 }
 420
 421 int btPeer::AreYouOK()
 422 {
 423   m_f_keepalive = 1;
 424   return stream.Send_Keepalive();
 425 }
 426
 427 int btPeer::RecvModule()
 428 {
 429   int f_peer_closed = 0;
 430   ssize_t r;
 431
 432   if ( 64 < m_err_count ) return -1;
 433
 434   r = stream.Feed();
 435
 436   if( r < 0 && r != -2 )
 437     return -1;
 438   else if ( r == -2 )
 439     f_peer_closed = 1;
 440
 441   r = stream.HaveMessage();
 442   for( ; r;){
 443     if( r < 0 ) return -1;
 444     if(MsgDeliver() < 0 || stream.PickMessage() < 0) return -1;
 445     r = stream.HaveMessage();
 446   }
 447   return f_peer_closed ? -1 : 0;
 448 }
 449
 450 int btPeer::SendModule()
 451 {
 452   if( stream.out_buffer.Count() && stream.Flush() < 0) return -1;
 453
 454   if(! reponse_q.IsEmpty() &&  CouldReponseSlice() ) {
 455     StartULTimer();
 456     Self.StartULTimer();
 457   }
 458
 459   for(; !reponse_q.IsEmpty() && CouldReponseSlice(); )
 460     if( ReponseSlice() < 0) return -1;
 461
 462   return 0;
 463 }