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