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
7 %%% Created : 19 Jul 2007 by
8 %%% Jesper Louis Andersen <jesper.louis.andersen@gmail.com>
9 %%%-------------------------------------------------------------------
10 -module(etorrent_t_peer_recv
).
12 -behaviour(gen_server
).
14 -include("etorrent_mnesia_table.hrl").
15 -include("etorrent_rate.hrl").
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]).
22 %% gen_server callbacks
23 -export([init
/1, handle_call
/3, handle_cast
/2, handle_info
/2,
24 terminate
/2, code_change
/3]).
26 -record(state
, { remote_peer_id
= none
,
34 local_interested
= false
,
36 remote_request_set
= none
,
44 endgame
= false
, % Are we in endgame mode?
48 file_system_pid
= none
,
49 peer_group_pid
= none
,
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 %%====================================================================
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
,
71 gen_server:start_link(?MODULE
, [LocalPeerId
, InfoHash
,
72 FilesystemPid
, GroupPid
, Id
, Parent
,
75 %%--------------------------------------------------------------------
77 %% Args: Pid ::= pid()
78 %% Description: Gracefully ask the server to stop.
79 %%--------------------------------------------------------------------
81 gen_server:cast(Pid
, stop
).
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
}).
94 %%--------------------------------------------------------------------
95 %% Function: choke(Pid)
96 %% Description: Choke the peer.
97 %%--------------------------------------------------------------------
99 gen_server:cast(Pid
, choke
).
101 %%--------------------------------------------------------------------
102 %% Function: unchoke(Pid)
103 %% Description: Unchoke the peer.
104 %%--------------------------------------------------------------------
106 gen_server:cast(Pid
, unchoke
).
108 %%--------------------------------------------------------------------
109 %% Function: interested(Pid)
110 %% Description: Tell the peer we are interested.
111 %%--------------------------------------------------------------------
113 gen_server:cast(Pid
, interested
).
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
}).
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
}).
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
}).
137 gen_server:cast(Pid
, queue_pieces
).
139 %%====================================================================
140 %% gen_server callbacks
141 %%====================================================================
143 %%--------------------------------------------------------------------
144 %% Function: init(Args) -> {ok, State} |
145 %% {ok, State, Timeout} |
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()),
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
,
162 rate
= etorrent_rate:init(?RATE_FUDGE
),
164 file_system_pid
= FilesystemPid
}}.
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
) ->
177 {reply
, Reply
, State
, 0}.
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
189 case etorrent_peer_communication:initiate_handshake(
191 S#state
.local_peer_id
,
192 S#state
.info_hash
) of
193 {ok
, _ReservedBytes
, PeerId
}
194 when PeerId
== S#state
.local_peer_id
->
196 {ok
, _ReservedBytes
, PeerId
} ->
197 complete_connection_setup(S#state
{ tcp_socket
= Socket
,
198 remote_peer_id
= PeerId
});
205 handle_cast({complete_handshake
, _ReservedBytes
, Socket
, RemotePeerId
}, S
) ->
206 case etorrent_peer_communication:complete_handshake(Socket
,
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
}
213 handle_cast(choke
, S
) ->
214 {atomic
, _
} = etorrent_peer:statechange(self(), local_choking
),
215 etorrent_t_peer_send:choke(S#state
.send_pid
),
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
),
221 {aborted
, _
} -> {stop
, normal
, S
}
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
),
228 handle_cast({endgame_got_chunk
, Chunk
}, S
) ->
229 NS
= handle_endgame_got_chunk(Chunk
, S
),
231 handle_cast(queue_pieces
, S
) ->
232 {ok
, NS
} = try_to_queue_up_pieces(S
),
234 handle_cast(stop
, S
) ->
236 handle_cast(_Msg
, State
) ->
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
248 handle_read_from_socket(S
, Packet
);
253 {error
, etimedout
} ->
255 {error
, timeout
} when S#state
.remote_choked
=:= true
->
257 {error
, timeout
} when S#state
.remote_choked
=:= false
->
258 {ok
, NS
} = try_to_queue_up_pieces(S
),
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
) ->
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
287 _
-> error_logger:info_report([reason_for_termination
, Reason
])
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
) ->
298 %%--------------------------------------------------------------------
299 %%% Internal functions
300 %%--------------------------------------------------------------------
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
) ->
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
),
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
),
324 handle_message({request
, Index
, Offset
, Len
}, S
) ->
325 etorrent_t_peer_send:remote_request(S#state
.send_pid
, Index
, Offset
, Len
),
327 handle_message({cancel
, Index
, Offset
, Len
}, S
) ->
328 etorrent_t_peer_send:cancel(S#state
.send_pid
, Index
, Offset
, Len
),
330 handle_message({have
, PieceNum
}, S
) ->
331 case etorrent_piece:valid(S#state
.torrent_id
, PieceNum
) of
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
));
344 {stop
, {invalid_piece
, S#state
.remote_peer_id
}, S
}
346 handle_message({bitfield
, BitField
}, S
) ->
347 case gb_sets:size(S#state
.piece_set
) of
349 Size
= etorrent_torrent:num_pieces(S#state
.torrent_id
),
351 etorrent_peer_communication:destruct_bitfield(Size
, BitField
),
352 case etorrent_piece:check_interest(S#state
.torrent_id
, PieceSet
) of
354 {ok
, statechange_interested(S#state
{piece_set
= PieceSet
},
357 {ok
, S#state
{piece_set
= PieceSet
}};
359 {stop
, {invalid_piece_2
, S#state
.remote_peer_id
}, S
}
361 N
when is_integer(N
) ->
362 %% This is a bad peer. Kill him!
365 handle_message({piece
, Index
, Offset
, Data
}, S
) ->
366 case handle_got_chunk(Index
, Offset
, Data
, size(Data
), S
) of
368 try_to_queue_up_pieces(NS
)
370 handle_message(Unknown
, S
) ->
371 {stop
, {unknown_message
, Unknown
}, S
}.
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
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
,
388 etorrent_chunk:endgame_remove_chunk(S#state
.send_pid
,
390 {Index
, Offset
, Len
}),
391 S#state
{ remote_request_set
= RS
};
393 %% Not an element in the request queue, ignore
394 etorrent_chunk:endgame_remove_chunk(S#state
.send_pid
,
396 {Index
, Offset
, Len
}),
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
408 ok
= etorrent_fs:write_chunk(S#state
.file_system_pid
,
410 case etorrent_chunk:store_chunk(S#state
.torrent_id
,
415 etorrent_fs:check_piece(S#state
.file_system_pid
,
416 S#state
.peer_group_pid
,
421 %% Tell other peers we got the chunk if in endgame
422 case S#state
.endgame
of
424 case etorrent_chunk:mark_fetched(S#state
.torrent_id
,
425 {Index
, Offset
, Len
}) of
429 etorrent_t_peer_group_mgr:broadcast_got_chunk(
430 S#state
.peer_group_pid
,
431 {Index
, Offset
, Len
})
436 RS
= gb_trees:delete_any({Index
, Offset
, Len
}, S#state
.remote_request_set
),
437 {ok
, S#state
{ remote_request_set
= RS
}};
439 %% Stray piece, we could try to get hold of it but for now we just
440 %% throw it on the floor.
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
) ->
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()}.
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
->
464 try_to_queue_up_pieces(S
) ->
465 case gb_trees:size(S#state
.remote_request_set
) of
466 N
when N
> ?LOW_WATERMARK
->
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(),
476 {ok
, statechange_interested(S
, false
)};
480 queue_items(Items
, S
);
482 queue_items(Items
, S#state
{ endgame
= true
})
486 %%--------------------------------------------------------------------
487 %% Function: queue_items/2
488 %% Args: ChunkList ::= [CompactChunk | ExplicitChunk]
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
}}.
502 queue_items([], _SendPid
, Tree
) -> Tree
;
503 queue_items([{Pn
, Chunks
} | Rest
], SendPid
, Tree
) ->
505 fun ({Offset
, Size
, Ops
}, T
) ->
506 case gb_trees:is_defined({Pn
, Offset
, Size
}, T
) of
510 etorrent_t_peer_send:local_request(SendPid
,
512 gb_trees:enter({Pn
, Offset
, Size
}, Ops
, T
)
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
523 etorrent_t_peer_send:local_request(SendPid
,
525 gb_trees:enter({Pn
, Offset
, Size
}, Ops
, Tree
)
527 queue_items(Rest
, SendPid
, NT
).
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
,
539 S#state
.file_system_pid
,
542 BF
= etorrent_piece:bitfield(S#state
.torrent_id
),
543 etorrent_t_peer_send:bitfield(SendPid
, BF
),
545 {noreply
, S#state
{send_pid
= SendPid
}, 0}.
547 statechange_interested(S
, What
) ->
548 etorrent_t_peer_send:interested(S#state
.send_pid
),
549 S#state
{local_interested
= What
}.
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
559 %%--------------------------------------------------------------------
560 handle_read_from_socket(S
, <<>>) ->
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
,
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
584 handle_read_from_socket(NS#state
{ packet_left
= none
,
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}.