lib/etorrent-1.0/src/etorrent_choker.erl

   1 %%%-------------------------------------------------------------------
   2 %%% File    : etorrent_t_peer_group.erl
   3 %%% Author  : Jesper Louis Andersen <jesper.louis.andersen@gmail.com>
   4 %%% License : See COPYING
   5 %%% Description : Master process for a number of peers.
   6 %%%
   7 %%% Created : 18 Jul 2007 by
   8 %%%      Jesper Louis Andersen <jesper.louis.andersen@gmail.com>
   9 %%%-------------------------------------------------------------------
  10 -module(etorrent_choker).
  11
  12 -behaviour(gen_server).
  13
  14 -include("rate_mgr.hrl").
  15 -include("peer_state.hrl").
  16
  17 -include("etorrent_mnesia_table.hrl").
  18
  19 %% API
  20 -export([start_link/1, perform_rechoke/0, monitor/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, {our_peer_id = none,
  27                 info_hash = none,
  28
  29                 timer_ref = none,
  30                 round = 0,
  31
  32                 optimistic_unchoke_pid = none,
  33                 opt_unchoke_chain = []}).
  34
  35 -record(rechoke_info, {pid :: pid(),
  36                        kind :: 'seeding' | 'leeching',
  37                        state :: 'seeding' | 'leeching' ,
  38                        snubbed :: bool(),
  39                        r_interest_state :: 'interested' | 'not_interested',
  40                        r_choke_state :: 'choked' | 'unchoked' ,
  41                        l_choke :: bool(),
  42                        rate :: float() }).
  43
  44 -define(SERVER, ?MODULE).
  45
  46 -define(ROUND_TIME, 10000).
  47
  48 %%====================================================================
  49 %% API
  50 %%====================================================================
  51 start_link(OurPeerId) ->
  52     gen_server:start_link({local, ?SERVER}, ?MODULE, [OurPeerId], []).
  53
  54 perform_rechoke() ->
  55     gen_server:cast(?SERVER, rechoke).
  56
  57 monitor(Pid) ->
  58     gen_server:call(?SERVER, {monitor, Pid}).
  59
  60 %%====================================================================
  61 %% gen_server callbacks
  62 %%====================================================================
  63
  64 init([OurPeerId]) ->
  65     process_flag(trap_exit, true),
  66     {ok, Tref} = timer:send_interval(?ROUND_TIME, self(), round_tick),
  67     {ok, #state{ our_peer_id = OurPeerId,
  68                      timer_ref = Tref}}.
  69
  70 handle_call({monitor, Pid}, _From, S) ->
  71     _Tref = erlang:monitor(process, Pid),
  72     NewChain = insert_new_peer_into_chain(Pid, S#state.opt_unchoke_chain),
  73     perform_rechoke(),
  74     {reply, ok, S#state { opt_unchoke_chain = NewChain }};
  75 handle_call(Request, _From, State) ->
  76     error_logger:error_report([unknown_peer_group_call, Request]),
  77     Reply = ok,
  78     {reply, Reply, State}.
  79 handle_cast(rechoke, S) ->
  80     rechoke(S),
  81     {noreply, S};
  82 handle_cast(_Msg, State) ->
  83     {noreply, State}.
  84
  85 handle_info(round_tick, S) ->
  86     case S#state.round of
  87         0 ->
  88             {ok, NS} = advance_optimistic_unchoke(S),
  89             rechoke(NS),
  90             {noreply, NS#state { round = 2}};
  91         N when is_integer(N) ->
  92             rechoke(S),
  93             {noreply, S#state{round = S#state.round - 1}}
  94     end;
  95 handle_info({'DOWN', _Ref, process, Pid, Reason}, S)
  96   when (Reason =:= normal) or (Reason =:= shutdown) ->
  97     % The peer shut down normally. Hence we just remove him and start up
  98     %  other peers. Eventually the tracker will re-add him to the peer list
  99
 100     % XXX: We might have to do something else
 101     rechoke(S),
 102
 103     NewChain = lists:delete(Pid, S#state.opt_unchoke_chain),
 104     {noreply, S#state { opt_unchoke_chain = NewChain }};
 105 handle_info({'DOWN', _Ref, process, Pid, _Reason}, S) ->
 106     % The peer shut down unexpectedly re-add him to the queue in the *back*
 107     NS = case etorrent_peer:select(Pid) of
 108              [_Peer] ->
 109                  % XXX: We might have to check that remote is intersted and we were choking
 110                  rechoke(S);
 111              [] -> S
 112          end,
 113
 114     NewChain = lists:delete(Pid, NS#state.opt_unchoke_chain),
 115     {noreply, NS#state{opt_unchoke_chain = NewChain}};
 116 handle_info(Info, State) ->
 117     error_logger:error_report([unknown_info_peer_group, Info]),
 118     {noreply, State}.
 119
 120 terminate(_Reason, _S) ->
 121     ok.
 122
 123 code_change(_OldVsn, State, _Extra) ->
 124     {ok, State}.
 125
 126 %%--------------------------------------------------------------------
 127 %%% Internal functions
 128 %%--------------------------------------------------------------------
 129
 130 %%--------------------------------------------------------------------
 131 %% Function: rechoke(State) -> ok
 132 %% Description: Recalculate the choke/unchoke state of peers
 133 %%--------------------------------------------------------------------
 134 rechoke(S) ->
 135     Peers = build_rechoke_info(S#state.opt_unchoke_chain),
 136     {PreferredDown, PreferredSeed} = split_preferred(Peers),
 137     PreferredSet = prune_preferred_peers(PreferredDown, PreferredSeed),
 138     ToChoke = rechoke_unchoke(Peers, PreferredSet),
 139     rechoke_choke(ToChoke, 0, optimistics(PreferredSet)).
 140
 141 build_rechoke_info(Peers) ->
 142     SeederSet = sets:from_list(seeding_torrents()),
 143     build_rechoke_info(SeederSet, Peers).
 144
 145 build_rechoke_info(_Seeding, []) ->
 146     [];
 147 build_rechoke_info(Seeding, [Pid | Next]) ->
 148     case etorrent_peer:select(Pid) of
 149         [] -> build_rechoke_info(Seeding, Next);
 150         [Peer] ->
 151             Kind = Peer#peer.state,
 152             Snubbed = etorrent_rate_mgr:snubbed(Peer#peer.torrent_id, Pid),
 153             PeerState = etorrent_rate_mgr:select_state(Peer#peer.torrent_id, Pid),
 154             case sets:is_element(Peer#peer.torrent_id, Seeding) of
 155                 true ->
 156                     case etorrent_rate_mgr:fetch_send_rate(
 157                            Peer#peer.torrent_id,
 158                            Pid) of
 159                         none -> build_rechoke_info(Seeding, Next);
 160                         Rate ->
 161                             [#rechoke_info { pid = Pid,
 162                                              kind = Kind,
 163                                              state = seeding,
 164                                              rate = Rate,
 165                                              r_interest_state =
 166                                                PeerState#peer_state.interest_state,
 167                                              r_choke_state =
 168                                                PeerState#peer_state.choke_state,
 169                                              l_choke =
 170                                                PeerState#peer_state.local_choke,
 171                                              snubbed = Snubbed } |
 172                              build_rechoke_info(Seeding, Next)]
 173                     end;
 174                 false ->
 175                     case etorrent_rate_mgr:fetch_recv_rate(
 176                            Peer#peer.torrent_id,
 177                            Pid) of
 178                         none -> build_rechoke_info(Seeding, Next);
 179                         Rate ->
 180                             [#rechoke_info { pid = Pid,
 181                                              kind = Kind,
 182                                              state = leeching,
 183                                              rate = -Rate, % Inverted for later sorting!
 184                                              r_interest_state =
 185                                                PeerState#peer_state.interest_state,
 186                                              r_choke_state =
 187                                                PeerState#peer_state.choke_state,
 188                                              l_choke =
 189                                                PeerState#peer_state.local_choke,
 190                                              snubbed = Snubbed } |
 191                              build_rechoke_info(Seeding, Next)]
 192                     end
 193             end
 194     end.
 195
 196 advance_optimistic_unchoke(S) ->
 197     NewChain = move_cyclic_chain(S#state.opt_unchoke_chain),
 198     case NewChain of
 199         [] ->
 200             {ok, S}; %% No peers yet
 201         [H | _T] ->
 202             etorrent_t_peer_recv:unchoke(H),
 203             {ok, S#state { opt_unchoke_chain = NewChain,
 204                            optimistic_unchoke_pid = H }}
 205     end.
 206
 207 move_cyclic_chain([]) -> [];
 208 move_cyclic_chain(Chain) ->
 209     F = fun (Pid) ->
 210                 case etorrent_peer:select(Pid) of
 211                     [] -> true;
 212                     [P] -> T = etorrent_rate_mgr:select_state(
 213                                  P#peer.torrent_id,
 214                                  Pid),
 215                            not (T#peer_state.interest_state =:= interested
 216                                 andalso T#peer_state.choke_state =:= choked)
 217                 end
 218         end,
 219     {Front, Back} = lists:splitwith(F, Chain),
 220     %% Advance chain
 221     Back ++ Front.
 222
 223 insert_new_peer_into_chain(Pid, Chain) ->
 224     Length = length(Chain),
 225     Index = lists:max([0, crypto:rand_uniform(0, Length)]),
 226     {Front, Back} = lists:split(Index, Chain),
 227     Front ++ [Pid | Back].
 228
 229 upload_slots() ->
 230     case application:get_env(etorrent, max_upload_slots) of
 231         {ok, auto} ->
 232             {ok, Rate} = application:get_env(etorrent, max_upload_rate),
 233             case Rate of
 234                 N when N =<  0 -> 7; %% Educated guess
 235                 N when N  <  9 -> 2;
 236                 N when N  < 15 -> 3;
 237                 N when N  < 42 -> 4;
 238                 N ->
 239                     round(math:sqrt(N * 0.6))
 240             end;
 241         {ok, N} when is_integer(N) ->
 242             N
 243     end.
 244
 245 split_preferred(Peers) ->
 246     {Downs, Leechs} = split_preferred_peers(Peers, [], []),
 247     {lists:keysort(#rechoke_info.rate, Downs),
 248      lists:keysort(#rechoke_info.rate, Leechs)}.
 249
 250 prune_preferred_peers(SDowns, SLeechs) ->
 251     MaxUploads = upload_slots(),
 252     DUploads = lists:max([1, round(MaxUploads * 0.7)]),
 253     SUploads = lists:max([1, round(MaxUploads * 0.3)]),
 254     {SUP2, DUP2} =
 255         case lists:max([0, DUploads - length(SDowns)]) of
 256             0 -> {SUploads, DUploads};
 257             N -> {SUploads + N, DUploads - N}
 258         end,
 259     {SUP3, DUP3} =
 260         case lists:max([0, SUP2 - length(SLeechs)]) of
 261             0 -> {SUP2, DUP2};
 262             K ->
 263                 {SUP2 - K, lists:min([DUP2 + K, length(SDowns)])}
 264         end,
 265     {TSDowns, TSLeechs} = {lists:sublist(SDowns, DUP3),
 266                            lists:sublist(SLeechs, SUP3)},
 267     sets:union(sets:from_list(TSDowns), sets:from_list(TSLeechs)).
 268
 269 rechoke_unchoke([], _PS) ->
 270     [];
 271 rechoke_unchoke([P | Next], PSet) ->
 272     case sets:is_element(P, PSet) of
 273         true ->
 274             etorrent_t_peer_recv:unchoke(P#rechoke_info.pid),
 275             rechoke_unchoke(Next, PSet);
 276         false ->
 277             [P | rechoke_unchoke(Next, PSet)]
 278     end.
 279
 280 optimistics(PSet) ->
 281     MinUp = case application:get_env(etorrent, min_uploads) of
 282                 {ok, N} -> N;
 283                 undefined -> 1
 284             end,
 285     lists:max([MinUp, upload_slots() - sets:size(PSet)]).
 286
 287 rechoke_choke([], _Count, _Optimistics) ->
 288     ok;
 289 rechoke_choke([P | Next], Count, Optimistics) when Count >= Optimistics ->
 290     etorrent_t_peer_recv:choke(P#rechoke_info.pid),
 291     rechoke_choke(Next, Count, Optimistics);
 292 rechoke_choke([P | Next], Count, Optimistics) ->
 293     case P#rechoke_info.kind =:= seeding of
 294         true ->
 295             etorrent_t_peer_recv:choke(P#rechoke_info.pid),
 296             rechoke_choke(Next, Count, Optimistics);
 297         false ->
 298             etorrent_t_peer_recv:unchoke(P#rechoke_info.pid),
 299             case P#rechoke_info.r_interest_state =:= interested of
 300                 true ->
 301                     rechoke_choke(Next, Count+1, Optimistics);
 302                 false ->
 303                     rechoke_choke(Next, Count, Optimistics)
 304             end
 305     end.
 306
 307 split_preferred_peers([], Downs, Leechs) ->
 308     {Downs, Leechs};
 309 split_preferred_peers([P | Next], Downs, Leechs) ->
 310     case P#rechoke_info.kind =:= seeding
 311           orelse P#rechoke_info.r_interest_state =:= not_interested of
 312         true ->
 313             split_preferred_peers(Next, Downs, Leechs);
 314         false when P#rechoke_info.state =:= seeding ->
 315             split_preferred_peers(Next, Downs, [P | Leechs]);
 316         false when P#rechoke_info.snubbed =:= true ->
 317             split_preferred_peers(Next, Downs, Leechs);
 318         false ->
 319             split_preferred_peers(Next, [P | Downs], Leechs)
 320     end.
 321
 322 seeding_torrents() ->
 323     {atomic, Torrents} = etorrent_torrent:all(),
 324     [T#torrent.id || T <- Torrents,
 325                      T#torrent.state =:= seeding].
 326