lib/etorrent-1.0/src/etorrent_t_peer_recv.erl

   1 %%%-------------------------------------------------------------------
   2 %%% File    : etorrent_t_peer_recv.erl
   3 %%% Author  : Jesper Louis Andersen <jesper.louis.andersen@gmail.com>
   4 %%% License : See COPYING
   5 %%% Description : Represents a peers receiving of data
   6 %%%
   7 %%% Created : 19 Jul 2007 by
   8 %%%    Jesper Louis Andersen <jesper.louis.andersen@gmail.com>
   9 %%%-------------------------------------------------------------------
  10 -module(etorrent_t_peer_recv).
  11
  12 -behaviour(gen_server).
  13
  14 -include("etorrent_mnesia_table.hrl").
  15 -include("etorrent_rate.hrl").
  16
  17 %% API
  18 -export([start_link/6, connect/3, choke/1, unchoke/1, interested/1,
  19          have/2, complete_handshake/4, endgame_got_chunk/2,
  20          queue_pieces/1, stop/1]).
  21
  22 %% gen_server callbacks
  23 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
  24          terminate/2, code_change/3]).
  25
  26 -record(state, { remote_peer_id = none,
  27                  local_peer_id = none,
  28                  info_hash = none,
  29
  30                  fast_extension = false, % Peer uses fast extension
  31
  32                  pieces_left,
  33                  seeder = false,
  34                  tcp_socket = none,
  35
  36                  remote_choked = true,
  37
  38                  local_interested = false,
  39
  40                  remote_request_set = none,
  41
  42                  piece_set = unknown,
  43                  piece_request = [],
  44
  45                  packet_left = none,
  46                  packet_iolist = [],
  47
  48                  endgame = false, % Are we in endgame mode?
  49
  50                  parent = none,
  51
  52                  file_system_pid = none,
  53                  send_pid = none,
  54
  55                  rate = none,
  56                  rate_timer = none,
  57                  torrent_id = none}).
  58
  59 -define(DEFAULT_CONNECT_TIMEOUT, 120000). % Default timeout in ms
  60 -define(DEFAULT_CHUNK_SIZE, 16384). % Default size for a chunk. All clients use this.
  61 -define(HIGH_WATERMARK, 15). % How many chunks to queue up to
  62 -define(LOW_WATERMARK, 5).  % Requeue when there are less than this number of pieces in queue
  63 %%====================================================================
  64 %% API
  65 %%====================================================================
  66 %%--------------------------------------------------------------------
  67 %% Function: start_link() -> {ok,Pid} | ignore | {error,Error}
  68 %% Description: Starts the server
  69 %%--------------------------------------------------------------------
  70 start_link(LocalPeerId, InfoHash, FilesystemPid, Id, Parent,
  71           {IP, Port}) ->
  72     gen_server:start_link(?MODULE, [LocalPeerId, InfoHash,
  73                                     FilesystemPid, Id, Parent,
  74                                     {IP, Port}], []).
  75
  76 %%--------------------------------------------------------------------
  77 %% Function: stop/1
  78 %% Args: Pid ::= pid()
  79 %% Description: Gracefully ask the server to stop.
  80 %%--------------------------------------------------------------------
  81 stop(Pid) ->
  82     gen_server:cast(Pid, stop).
  83
  84 %%--------------------------------------------------------------------
  85 %% Function: connect(Pid, IP, Port)
  86 %% Description: Connect to the IP and Portnumber for communication with
  87 %%   the peer. Note we don't handle the connect in the init phase. This is
  88 %%   due to the fact that a connect may take a considerable amount of time.
  89 %%   With this scheme, we spawn off processes, and then make them all attempt
  90 %%   connects in parallel, which is much easier.
  91 %%--------------------------------------------------------------------
  92 connect(Pid, IP, Port) ->
  93     gen_server:cast(Pid, {connect, IP, Port}).
  94
  95 %%--------------------------------------------------------------------
  96 %% Function: choke(Pid)
  97 %% Description: Choke the peer.
  98 %%--------------------------------------------------------------------
  99 choke(Pid) ->
 100     gen_server:cast(Pid, choke).
 101
 102 %%--------------------------------------------------------------------
 103 %% Function: unchoke(Pid)
 104 %% Description: Unchoke the peer.
 105 %%--------------------------------------------------------------------
 106 unchoke(Pid) ->
 107     gen_server:cast(Pid, unchoke).
 108
 109 %%--------------------------------------------------------------------
 110 %% Function: interested(Pid)
 111 %% Description: Tell the peer we are interested.
 112 %%--------------------------------------------------------------------
 113 interested(Pid) ->
 114     gen_server:cast(Pid, interested).
 115
 116 %%--------------------------------------------------------------------
 117 %% Function: have(Pid, PieceNumber)
 118 %% Description: Tell the peer we have just received piece PieceNumber.
 119 %%--------------------------------------------------------------------
 120 have(Pid, PieceNumber) ->
 121     gen_server:cast(Pid, {have, PieceNumber}).
 122
 123 %%--------------------------------------------------------------------
 124 %% Function: endgame_got_chunk(Pid, Index, Offset) -> ok
 125 %% Description: We got the chunk {Index, Offset}, handle it.
 126 %%--------------------------------------------------------------------
 127 endgame_got_chunk(Pid, Chunk) ->
 128     gen_server:cast(Pid, {endgame_got_chunk, Chunk}).
 129
 130 %%--------------------------------------------------------------------
 131 %% Function: complete_handshake(Pid, Capabilities, Socket, PeerId)
 132 %% Description: Complete the handshake initiated by another client.
 133 %%--------------------------------------------------------------------
 134 complete_handshake(Pid, Capabilities, Socket, PeerId) ->
 135     gen_server:cast(Pid, {complete_handshake, Capabilities, Socket, PeerId}).
 136
 137 queue_pieces(Pid) ->
 138     gen_server:cast(Pid, queue_pieces).
 139
 140 %%====================================================================
 141 %% gen_server callbacks
 142 %%====================================================================
 143
 144 %%--------------------------------------------------------------------
 145 %% Function: init(Args) -> {ok, State} |
 146 %%                         {ok, State, Timeout} |
 147 %%                         ignore               |
 148 %%                         {stop, Reason}
 149 %% Description: Initiates the server
 150 %%--------------------------------------------------------------------
 151 init([LocalPeerId, InfoHash, FilesystemPid, Id, Parent, {IP, Port}]) ->
 152     process_flag(trap_exit, true),
 153     {ok, TRef} = timer:send_interval(?RATE_UPDATE, self(), rate_update),
 154     %% TODO: Update the leeching state to seeding when peer finished torrent.
 155     ok = etorrent_peer:new(IP, Port, Id, self(), leeching),
 156     ok = etorrent_choker:monitor(self()),
 157     [T] = etorrent_torrent:select(Id),
 158     {ok, #state{
 159        pieces_left = T#torrent.pieces,
 160        parent = Parent,
 161        local_peer_id = LocalPeerId,
 162        remote_request_set = gb_trees:empty(),
 163        info_hash = InfoHash,
 164        torrent_id = Id,
 165        rate = etorrent_rate:init(?RATE_FUDGE),
 166        rate_timer = TRef,
 167        file_system_pid = FilesystemPid}}.
 168
 169 %%--------------------------------------------------------------------
 170 %% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} |
 171 %%                                      {reply, Reply, State, Timeout} |
 172 %%                                      {noreply, State} |
 173 %%                                      {noreply, State, Timeout} |
 174 %%                                      {stop, Reason, Reply, State} |
 175 %%                                      {stop, Reason, State}
 176 %% Description: Handling call messages
 177 %%--------------------------------------------------------------------
 178 handle_call(_Request, _From, State) ->
 179     Reply = ok,
 180     {reply, Reply, State, 0}.
 181
 182 %%--------------------------------------------------------------------
 183 %% Function: handle_cast(Msg, State) -> {noreply, State} |
 184 %%                                      {noreply, State, Timeout} |
 185 %%                                      {stop, Reason, State}
 186 %% Description: Handling cast messages
 187 %%--------------------------------------------------------------------
 188 handle_cast({connect, IP, Port}, S) ->
 189     case gen_tcp:connect(IP, Port, [binary, {active, false}],
 190                          ?DEFAULT_CONNECT_TIMEOUT) of
 191         {ok, Socket} ->
 192             case etorrent_peer_communication:initiate_handshake(
 193                    Socket,
 194                    S#state.local_peer_id,
 195                    S#state.info_hash) of
 196                 {ok, _Capabilities, PeerId}
 197                   when PeerId == S#state.local_peer_id ->
 198                     {stop, normal, S};
 199                 {ok, Capabilities, PeerId} ->
 200                     FastExtension = lists:member(fast_extension, Capabilities),
 201                     complete_connection_setup(S#state { tcp_socket = Socket,
 202                                                         remote_peer_id = PeerId,
 203                                                         fast_extension = FastExtension});
 204                 {error, _} ->
 205                     {stop, normal, S}
 206             end;
 207         {error, _Reason} ->
 208             {stop, normal, S}
 209     end;
 210 handle_cast({complete_handshake, Capabilities, Socket, RemotePeerId}, S) ->
 211     FastExtension = lists:member(fast_extension, Capabilities),
 212     case etorrent_peer_communication:complete_handshake(Socket,
 213                                                         S#state.info_hash,
 214                                                         S#state.local_peer_id) of
 215         ok -> complete_connection_setup(S#state { tcp_socket = Socket,
 216                                                   remote_peer_id = RemotePeerId,
 217                                                   fast_extension = FastExtension});
 218         {error, stop} -> {stop, normal, S}
 219     end;
 220 handle_cast(choke, S) ->
 221     etorrent_t_peer_send:choke(S#state.send_pid),
 222     {noreply, S, 0};
 223 handle_cast(unchoke, S) ->
 224     etorrent_t_peer_send:unchoke(S#state.send_pid),
 225     {noreply, S, 0};
 226 handle_cast(interested, S) ->
 227     {noreply, statechange_interested(S, true), 0};
 228 handle_cast({have, PN}, S) when S#state.piece_set =:= unknown ->
 229     etorrent_t_peer_send:have(S#state.send_pid, PN),
 230     {noreply, S, 0};
 231 handle_cast({have, PN}, S) ->
 232     case gb_sets:is_element(PN, S#state.piece_set) of
 233         true -> ok;
 234         false -> ok = etorrent_t_peer_send:have(S#state.send_pid, PN)
 235     end,
 236     {noreply, S, 0};
 237 handle_cast({endgame_got_chunk, Chunk}, S) ->
 238     NS = handle_endgame_got_chunk(Chunk, S),
 239     {noreply, NS, 0};
 240 handle_cast(queue_pieces, S) ->
 241     {ok, NS} = try_to_queue_up_pieces(S),
 242     {noreply, NS, 0};
 243 handle_cast(stop, S) ->
 244     {stop, normal, S};
 245 handle_cast(_Msg, State) ->
 246     {noreply, State, 0}.
 247
 248 %%--------------------------------------------------------------------
 249 %% Function: handle_info(Info, State) -> {noreply, State} |
 250 %%                                       {noreply, State, Timeout} |
 251 %%                                       {stop, Reason, State}
 252 %% Description: Handling all non call/cast messages
 253 %%--------------------------------------------------------------------
 254 handle_info(timeout, S) when S#state.tcp_socket =:= none ->
 255     %% Haven't started up yet
 256     {noreply, S, 3000};
 257 handle_info(timeout, S) ->
 258     case gen_tcp:recv(S#state.tcp_socket, 0, 3000) of
 259         {ok, Packet} ->
 260             handle_read_from_socket(S, Packet);
 261         {error, closed} ->
 262             {stop, normal, S};
 263         {error, ebadf} ->
 264             {stop, normal, S};
 265         {error, ehostunreach} ->
 266             {stop, normal, S};
 267         {error, etimedout} ->
 268             {noreply, S, 0};
 269         {error, timeout} when S#state.remote_choked =:= true ->
 270             {noreply, S, 0};
 271         {error, timeout} when S#state.remote_choked =:= false ->
 272             {ok, NS} = try_to_queue_up_pieces(S),
 273             {noreply, NS, 0}
 274     end;
 275 handle_info(rate_update, S) ->
 276     Rate = etorrent_rate:update(S#state.rate, 0),
 277     ok = etorrent_rate_mgr:recv_rate(S#state.torrent_id, self(), Rate#peer_rate.rate, 0),
 278     {noreply, S#state { rate = Rate}, 0};
 279 handle_info(_Info, State) ->
 280     {noreply, State, 0}.
 281
 282 %%--------------------------------------------------------------------
 283 %% Function: terminate(Reason, State) -> void()
 284 %% Description: This function is called by a gen_server when it is about to
 285 %% terminate. It should be the opposite of Module:init/1 and do any necessary
 286 %% cleaning up. When it returns, the gen_server terminates with Reason.
 287 %% The return value is ignored.
 288 %%--------------------------------------------------------------------
 289 terminate(Reason, S) ->
 290     etorrent_peer:delete(self()),
 291     etorrent_counters:release_peer_slot(),
 292     _NS = unqueue_all_pieces(S),
 293     case S#state.tcp_socket of
 294         none ->
 295             ok;
 296         Sock ->
 297             gen_tcp:close(Sock)
 298     end,
 299     case Reason of
 300         normal -> ok;
 301         shutdown -> ok;
 302         _ -> error_logger:info_report([reason_for_termination, Reason])
 303     end,
 304     ok.
 305
 306 %%--------------------------------------------------------------------
 307 %% Func: code_change(OldVsn, State, Extra) -> {ok, NewState}
 308 %% Description: Convert process state when code is changed
 309 %%--------------------------------------------------------------------
 310 code_change(_OldVsn, State, _Extra) ->
 311     {ok, State}.
 312
 313 %%--------------------------------------------------------------------
 314 %%% Internal functions
 315 %%--------------------------------------------------------------------
 316
 317 %%--------------------------------------------------------------------
 318 %% Func: handle_message(Msg, State) -> {ok, NewState} | {stop, Reason, NewState}
 319 %% Description: Process an incoming message Msg from the wire. Return either
 320 %%  {ok, S} if the processing was ok, or {stop, Reason, S} in case of an error.
 321 %%--------------------------------------------------------------------
 322 handle_message(keep_alive, S) ->
 323     {ok, S};
 324 handle_message(choke, S) ->
 325     ok = etorrent_rate_mgr:choke(S#state.torrent_id, self()),
 326     NS = unqueue_all_pieces(S),
 327     {ok, NS#state { remote_choked = true }};
 328 handle_message(unchoke, S) ->
 329     ok = etorrent_rate_mgr:unchoke(S#state.torrent_id, self()),
 330     try_to_queue_up_pieces(S#state{remote_choked = false});
 331 handle_message(interested, S) ->
 332     ok = etorrent_rate_mgr:interested(S#state.torrent_id, self()),
 333     ok = etorrent_t_peer_send:check_choke(S#state.send_pid),
 334     {ok, S};
 335 handle_message(not_interested, S) ->
 336     ok = etorrent_rate_mgr:not_interested(S#state.torrent_id, self()),
 337     ok = etorrent_t_peer_send:check_choke(S#state.send_pid),
 338     {ok, S};
 339 handle_message({request, Index, Offset, Len}, S) ->
 340     etorrent_t_peer_send:remote_request(S#state.send_pid, Index, Offset, Len),
 341     {ok, S};
 342 handle_message({cancel, Index, Offset, Len}, S) ->
 343     etorrent_t_peer_send:cancel(S#state.send_pid, Index, Offset, Len),
 344     {ok, S};
 345 handle_message({have, PieceNum}, S) ->
 346     peer_have(PieceNum, S);
 347 handle_message(have_none, S) when S#state.piece_set =/= unknown ->
 348     {stop, normal, S};
 349 handle_message(have_none, S) when S#state.fast_extension =:= true ->
 350     Size = etorrent_torrent:num_pieces(S#state.torrent_id),
 351     {ok, S#state { piece_set = gb_sets:new(),
 352                    pieces_left = Size,
 353                    seeder = false}};
 354 handle_message(have_all, S) when S#state.piece_set =/= unknown ->
 355     {stop, normal, S};
 356 handle_message(have_all, S) when S#state.fast_extension =:= true ->
 357     Size = etorrent_torrent:num_pieces(S#state.torrent_id),
 358     FullSet = gb_sets:from_list(lists:seq(0, Size - 1)),
 359     {ok, S#state { piece_set = FullSet,
 360                    pieces_left = 0,
 361                    seeder = true }};
 362 handle_message({bitfield, _BF}, S) when S#state.piece_set =/= unknown ->
 363     {ok, {IP, Port}} = inet:peername(S#state.tcp_socket),
 364     etorrent_peer_mgr:enter_bad_peer(IP, Port, S#state.remote_peer_id),
 365     {stop, normal, S};
 366 handle_message({bitfield, BitField}, S) ->
 367     Size = etorrent_torrent:num_pieces(S#state.torrent_id),
 368     {ok, PieceSet} =
 369         etorrent_peer_communication:destruct_bitfield(Size, BitField),
 370     Left = S#state.pieces_left - gb_sets:size(PieceSet),
 371     case Left of
 372         0  -> ok = etorrent_peer:statechange(self(), seeder);
 373         _N -> ok
 374     end,
 375     case etorrent_piece_mgr:check_interest(S#state.torrent_id, PieceSet) of
 376         interested ->
 377             {ok, statechange_interested(S#state {piece_set = PieceSet,
 378                                                  pieces_left = Left,
 379                                                  seeder = Left == 0},
 380                                         true)};
 381         not_interested ->
 382             {ok, S#state{piece_set = PieceSet, pieces_left = Left,
 383                          seeder = Left == 0}};
 384         invalid_piece ->
 385             {stop, {invalid_piece_2, S#state.remote_peer_id}, S}
 386     end;
 387 handle_message({piece, Index, Offset, Data}, S) ->
 388     case handle_got_chunk(Index, Offset, Data, size(Data), S) of
 389         {ok, NS} ->
 390             try_to_queue_up_pieces(NS)
 391     end;
 392 handle_message(Unknown, S) ->
 393     error_logger:info_report([{unknown_message, Unknown}]),
 394     {stop, normal, S}.
 395
 396 %%--------------------------------------------------------------------
 397 %% Func: handle_endgame_got_chunk(Index, Offset, S) -> State
 398 %% Description: Some other peer just downloaded {Index, Offset, Len} so try
 399 %%   not to download it here if we can avoid it.
 400 %%--------------------------------------------------------------------
 401 handle_endgame_got_chunk({Index, Offset, Len}, S) ->
 402     case gb_trees:is_defined({Index, Offset, Len}, S#state.remote_request_set) of
 403         true ->
 404             %% Delete the element from the request set.
 405             RS = gb_trees:delete({Index, Offset, Len}, S#state.remote_request_set),
 406             etorrent_t_peer_send:cancel(S#state.send_pid,
 407                                         Index,
 408                                         Offset,
 409                                         Len),
 410             etorrent_chunk_mgr:endgame_remove_chunk(S#state.send_pid,
 411                                                     S#state.torrent_id,
 412                                                     {Index, Offset, Len}),
 413             S#state { remote_request_set = RS };
 414         false ->
 415             %% Not an element in the request queue, ignore
 416             etorrent_chunk_mgr:endgame_remove_chunk(S#state.send_pid,
 417                                                     S#state.torrent_id,
 418                                                     {Index, Offset, Len}),
 419             S
 420     end.
 421
 422 %%--------------------------------------------------------------------
 423 %% Func: handle_got_chunk(Index, Offset, Data, Len, S) -> {ok, State}
 424 %% Description: We just got some chunk data. Store it in the mnesia DB
 425 %%--------------------------------------------------------------------
 426 handle_got_chunk(Index, Offset, Data, Len, S) ->
 427     case gb_trees:lookup({Index, Offset, Len},
 428                          S#state.remote_request_set) of
 429         {value, Ops} ->
 430             ok = etorrent_fs:write_chunk(S#state.file_system_pid,
 431                                          {Index, Data, Ops}),
 432             case etorrent_chunk_mgr:store_chunk(S#state.torrent_id,
 433                                                 Index,
 434                                                 {Offset, Len},
 435                                                 self()) of
 436                 full ->
 437                     etorrent_fs:check_piece(S#state.file_system_pid, Index);
 438                 ok ->
 439                     ok
 440             end,
 441             %% Tell other peers we got the chunk if in endgame
 442             case S#state.endgame of
 443                 true ->
 444                     case etorrent_chunk_mgr:mark_fetched(S#state.torrent_id,
 445                                                      {Index, Offset, Len}) of
 446                         found ->
 447                             ok;
 448                         assigned ->
 449                             broadcast_got_chunk({Index, Offset, Len}, S#state.torrent_id)
 450                     end;
 451                 false ->
 452                     ok
 453             end,
 454             RS = gb_trees:delete_any({Index, Offset, Len}, S#state.remote_request_set),
 455             {ok, S#state { remote_request_set = RS }};
 456         none ->
 457             %% Stray piece, we could try to get hold of it but for now we just
 458             %%   throw it on the floor.
 459             {ok, S}
 460     end.
 461
 462 %%--------------------------------------------------------------------
 463 %% Function: unqueue_all_pieces/1
 464 %% Description: Unqueue all queued pieces at the other end. We place
 465 %%   the earlier queued items at the end to compensate for quick
 466 %%   choke/unchoke problems and live data.
 467 %%--------------------------------------------------------------------
 468 unqueue_all_pieces(S) ->
 469     %% Put chunks back
 470     ok = etorrent_chunk_mgr:putback_chunks(self()),
 471     %% Tell other peers that there is 0xf00d!
 472     broadcast_queue_pieces(S#state.torrent_id),
 473     %% Clean up the request set.
 474     S#state{remote_request_set = gb_trees:empty()}.
 475
 476 %%--------------------------------------------------------------------
 477 %% Function: try_to_queue_up_requests(state()) -> {ok, state()}
 478 %% Description: Try to queue up requests at the other end.
 479 %%--------------------------------------------------------------------
 480 try_to_queue_up_pieces(S) when S#state.remote_choked == true ->
 481     {ok, S};
 482 try_to_queue_up_pieces(S) ->
 483     case gb_trees:size(S#state.remote_request_set) of
 484         N when N > ?LOW_WATERMARK ->
 485             {ok, S};
 486         %% Optimization: Only replenish pieces modulo some N
 487         N when is_integer(N) ->
 488             PiecesToQueue = ?HIGH_WATERMARK - N,
 489             case etorrent_chunk_mgr:pick_chunks(self(),
 490                                                 S#state.torrent_id,
 491                                                 S#state.piece_set,
 492                                                 PiecesToQueue) of
 493                 not_interested ->
 494                     {ok, statechange_interested(S, false)};
 495                 none_eligible ->
 496                     {ok, S};
 497                 {ok, Items} ->
 498                     queue_items(Items, S);
 499                 {endgame, Items} ->
 500                     queue_items(Items, S#state { endgame = true })
 501             end
 502     end.
 503
 504 %%--------------------------------------------------------------------
 505 %% Function: queue_items/2
 506 %% Args:     ChunkList ::= [CompactChunk | ExplicitChunk]
 507 %%           S         ::= #state
 508 %%           CompactChunk ::= {PieceNumber, ChunkList}
 509 %%           ExplicitChunk ::= {PieceNumber, Offset, Size, Ops}
 510 %%           ChunkList ::= [{Offset, Size, Ops}]
 511 %%           PieceNumber, Offset, Size ::= integer()
 512 %%           Ops ::= file_operations - described elsewhere.
 513 %% Description: Send chunk messages for each chunk we decided to queue.
 514 %%   also add these chunks to the piece request set.
 515 %%--------------------------------------------------------------------
 516 queue_items(ChunkList, S) ->
 517     RSet = queue_items(ChunkList, S#state.send_pid, S#state.remote_request_set),
 518     {ok, S#state { remote_request_set = RSet }}.
 519
 520 queue_items([], _SendPid, Tree) -> Tree;
 521 queue_items([{Pn, Chunks} | Rest], SendPid, Tree) ->
 522     NT = lists:foldl(
 523       fun ({Offset, Size, Ops}, T) ->
 524               case gb_trees:is_defined({Pn, Offset, Size}, T) of
 525                   true ->
 526                       Tree;
 527                   false ->
 528                       etorrent_t_peer_send:local_request(SendPid,
 529                                                          {Pn, Offset, Size}),
 530                       gb_trees:enter({Pn, Offset, Size}, Ops, T)
 531               end
 532       end,
 533       Tree,
 534       Chunks),
 535     queue_items(Rest, SendPid, NT);
 536 queue_items([{Pn, Offset, Size, Ops} | Rest], SendPid, Tree) ->
 537     NT = case gb_trees:is_defined({Pn, Offset, Size}, Tree) of
 538              true ->
 539                  Tree;
 540              false ->
 541                  etorrent_t_peer_send:local_request(SendPid,
 542                                                     {Pn, Offset, Size}),
 543                  gb_trees:enter({Pn, Offset, Size}, Ops, Tree)
 544          end,
 545     queue_items(Rest, SendPid, NT).
 546
 547 %%--------------------------------------------------------------------
 548 %% Function: complete_connection_setup() -> gen_server_reply()}
 549 %% Description: Do the bookkeeping needed to set up the peer:
 550 %%    * enable passive messaging mode on the socket.
 551 %%    * Start the send pid
 552 %%    * Send off the bitfield
 553 %%--------------------------------------------------------------------
 554 complete_connection_setup(S) ->
 555     {ok, SendPid} = etorrent_t_peer_sup:add_sender(S#state.parent,
 556                                                    S#state.tcp_socket,
 557                                                    S#state.file_system_pid,
 558                                                    S#state.torrent_id,
 559                                                    self()),
 560     BF = etorrent_piece_mgr:bitfield(S#state.torrent_id),
 561     etorrent_t_peer_send:bitfield(SendPid, BF),
 562
 563     {noreply, S#state{send_pid = SendPid}, 0}.
 564
 565 statechange_interested(S, What) ->
 566     etorrent_t_peer_send:interested(S#state.send_pid),
 567     S#state{local_interested = What}.
 568
 569
 570 %%--------------------------------------------------------------------
 571 %% Function: handle_read_from_socket(State, Packet) -> NewState
 572 %%             Packet ::= binary()
 573 %%             State  ::= #state()
 574 %% Description: Packet came in. Handle it.
 575 %%--------------------------------------------------------------------
 576 handle_read_from_socket(S, <<>>) ->
 577     {noreply, S, 0};
 578 handle_read_from_socket(S, <<0:32/big-integer, Rest/binary>>) when S#state.packet_left =:= none ->
 579     handle_read_from_socket(S, Rest);
 580 handle_read_from_socket(S, <<Left:32/big-integer, Rest/binary>>) when S#state.packet_left =:= none ->
 581     handle_read_from_socket(S#state { packet_left = Left,
 582                                       packet_iolist = []}, Rest);
 583 handle_read_from_socket(S, Packet) when is_binary(S#state.packet_left) ->
 584     H = S#state.packet_left,
 585     handle_read_from_socket(S#state { packet_left = none },
 586                             <<H/binary, Packet/binary>>);
 587 handle_read_from_socket(S, Packet) when size(Packet) < 4, S#state.packet_left =:= none ->
 588     {noreply, S#state { packet_left = Packet }};
 589 handle_read_from_socket(S, Packet)
 590   when size(Packet) >= S#state.packet_left, is_integer(S#state.packet_left) ->
 591     Left = S#state.packet_left,
 592     <<Data:Left/binary, Rest/binary>> = Packet,
 593     Left = size(Data),
 594     P = iolist_to_binary(lists:reverse([Data | S#state.packet_iolist])),
 595     {Msg, Rate, Amount} = etorrent_peer_communication:recv_message(S#state.rate, P),
 596     case Msg of
 597         {piece, _, _ ,_} ->
 598             ok = etorrent_rate_mgr:recv_rate(S#state.torrent_id,
 599                                              self(), Rate#peer_rate.rate,
 600                                              Amount, last_update);
 601         _Msg ->
 602             ok = etorrent_rate_mgr:recv_rate(S#state.torrent_id,
 603                                              self(), Rate#peer_rate.rate,
 604                                              Amount, normal)
 605     end,
 606     case handle_message(Msg, S#state {rate = Rate}) of
 607         {ok, NS} ->
 608             handle_read_from_socket(NS#state { packet_left = none,
 609                                                packet_iolist = []},
 610                                     Rest);
 611         {stop, Err, NS} ->
 612             {stop, Err, NS}
 613     end;
 614 handle_read_from_socket(S, Packet)
 615   when size(Packet) < S#state.packet_left, is_integer(S#state.packet_left) ->
 616     {noreply, S#state { packet_iolist = [Packet | S#state.packet_iolist],
 617                         packet_left = S#state.packet_left - size(Packet) }, 0}.
 618
 619 broadcast_queue_pieces(TorrentId) ->
 620     Peers = etorrent_peer:all(TorrentId),
 621     lists:foreach(fun (P) ->
 622                           etorrent_t_peer_recv:queue_pieces(P#peer.pid)
 623                   end,
 624                   Peers).
 625
 626
 627 broadcast_got_chunk(Chunk, TorrentId) ->
 628     Peers = etorrent_peer:all(TorrentId),
 629     lists:foreach(fun (Peer) ->
 630                           etorrent_t_peer_recv:endgame_got_chunk(Peer#peer.pid, Chunk)
 631                   end,
 632                   Peers).
 633
 634 peer_have(PN, S) when S#state.piece_set =:= unknown ->
 635     peer_have(PN, S#state {piece_set = gb_sets:new()});
 636 peer_have(PN, S) ->
 637     case etorrent_piece_mgr:valid(S#state.torrent_id, PN) of
 638         true ->
 639             Left = S#state.pieces_left - 1,
 640             case peer_seeds(S#state.torrent_id, Left) of
 641                 ok ->
 642                     PieceSet = gb_sets:add_element(PN, S#state.piece_set),
 643                     NS = S#state{piece_set = PieceSet,
 644                                  pieces_left = Left,
 645                                  seeder = Left == 0},
 646                     case etorrent_piece_mgr:interesting(S#state.torrent_id, PN) of
 647                         true when S#state.local_interested =:= true ->
 648                             try_to_queue_up_pieces(S);
 649                         true when S#state.local_interested =:= false ->
 650                             try_to_queue_up_pieces(statechange_interested(S, true));
 651                         false ->
 652                             {ok, NS}
 653                     end;
 654                 {stop, R} -> {stop, R, S}
 655             end;
 656         false ->
 657             {ok, {IP, Port}} = inet:peername(S#state.tcp_socket),
 658             etorrent_peer_mgr:enter_bad_peer(IP, Port, S#state.remote_peer_id),
 659             {stop, normal, S}
 660     end.
 661
 662 peer_seeds(Id, 0) ->
 663     ok = etorrent_peer:statechange(self(), seeder),
 664     [T] = etorrent_torrent:select(Id),
 665     case T#torrent.state of
 666         seeding -> {stop, normal};
 667         _ -> ok
 668     end;
 669 peer_seeds(_Id, _N) -> ok.
 670