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     case etorrent_peer:select(Pid) of
 108         [_Peer] -> ok = rechoke(S);
 109              [] -> ok
 110     end,
 111
 112     NewChain = lists:delete(Pid, S#state.opt_unchoke_chain),
 113     {noreply, S#state{opt_unchoke_chain = NewChain}};
 114 handle_info(Info, State) ->
 115     error_logger:error_report([unknown_info_peer_group, Info]),
 116     {noreply, State}.
 117
 118 terminate(_Reason, _S) ->
 119     ok.
 120
 121 code_change(_OldVsn, State, _Extra) ->
 122     {ok, State}.
 123
 124 %%--------------------------------------------------------------------
 125 %%% Internal functions
 126 %%--------------------------------------------------------------------
 127
 128 %%--------------------------------------------------------------------
 129 %% Function: rechoke(State) -> ok
 130 %% Description: Recalculate the choke/unchoke state of peers
 131 %%--------------------------------------------------------------------
 132 rechoke(S) ->
 133     Peers = build_rechoke_info(S#state.opt_unchoke_chain),
 134     {PreferredDown, PreferredSeed} = split_preferred(Peers),
 135     PreferredSet = prune_preferred_peers(PreferredDown, PreferredSeed),
 136     ToChoke = rechoke_unchoke(Peers, PreferredSet),
 137     rechoke_choke(ToChoke, 0, optimistics(PreferredSet)).
 138
 139 build_rechoke_info(Peers) ->
 140     SeederSet = sets:from_list(seeding_torrents()),
 141     build_rechoke_info(SeederSet, Peers).
 142
 143 build_rechoke_info(_Seeding, []) ->
 144     [];
 145 build_rechoke_info(Seeding, [Pid | Next]) ->
 146     case etorrent_peer:select(Pid) of
 147         [] -> build_rechoke_info(Seeding, Next);
 148         [Peer] ->
 149             Kind = Peer#peer.state,
 150             Snubbed = etorrent_rate_mgr:snubbed(Peer#peer.torrent_id, Pid),
 151             PeerState = etorrent_rate_mgr:select_state(Peer#peer.torrent_id, Pid),
 152             case sets:is_element(Peer#peer.torrent_id, Seeding) of
 153                 true ->
 154                     case etorrent_rate_mgr:fetch_send_rate(
 155                            Peer#peer.torrent_id,
 156                            Pid) of
 157                         none -> build_rechoke_info(Seeding, Next);
 158                         Rate ->
 159                             [#rechoke_info { pid = Pid,
 160                                              kind = Kind,
 161                                              state = seeding,
 162                                              rate = Rate,
 163                                              r_interest_state =
 164                                                PeerState#peer_state.interest_state,
 165                                              r_choke_state =
 166                                                PeerState#peer_state.choke_state,
 167                                              l_choke =
 168                                                PeerState#peer_state.local_choke,
 169                                              snubbed = Snubbed } |
 170                              build_rechoke_info(Seeding, Next)]
 171                     end;
 172                 false ->
 173                     case etorrent_rate_mgr:fetch_recv_rate(
 174                            Peer#peer.torrent_id,
 175                            Pid) of
 176                         none -> build_rechoke_info(Seeding, Next);
 177                         Rate ->
 178                             [#rechoke_info { pid = Pid,
 179                                              kind = Kind,
 180                                              state = leeching,
 181                                              rate = -Rate, % Inverted for later sorting!
 182                                              r_interest_state =
 183                                                PeerState#peer_state.interest_state,
 184                                              r_choke_state =
 185                                                PeerState#peer_state.choke_state,
 186                                              l_choke =
 187                                                PeerState#peer_state.local_choke,
 188                                              snubbed = Snubbed } |
 189                              build_rechoke_info(Seeding, Next)]
 190                     end
 191             end
 192     end.
 193
 194 advance_optimistic_unchoke(S) ->
 195     NewChain = move_cyclic_chain(S#state.opt_unchoke_chain),
 196     case NewChain of
 197         [] ->
 198             {ok, S}; %% No peers yet
 199         [H | _T] ->
 200             etorrent_t_peer_recv:unchoke(H),
 201             {ok, S#state { opt_unchoke_chain = NewChain,
 202                            optimistic_unchoke_pid = H }}
 203     end.
 204
 205 move_cyclic_chain([]) -> [];
 206 move_cyclic_chain(Chain) ->
 207     F = fun (Pid) ->
 208                 case etorrent_peer:select(Pid) of
 209                     [] -> true;
 210                     [P] -> T = etorrent_rate_mgr:select_state(
 211                                  P#peer.torrent_id,
 212                                  Pid),
 213                            not (T#peer_state.interest_state =:= interested
 214                                 andalso T#peer_state.choke_state =:= choked)
 215                 end
 216         end,
 217     {Front, Back} = lists:splitwith(F, Chain),
 218     %% Advance chain
 219     Back ++ Front.
 220
 221 insert_new_peer_into_chain(Pid, Chain) ->
 222     Length = length(Chain),
 223     Index = lists:max([0, crypto:rand_uniform(0, Length)]),
 224     {Front, Back} = lists:split(Index, Chain),
 225     Front ++ [Pid | Back].
 226
 227 upload_slots() ->
 228     case application:get_env(etorrent, max_upload_slots) of
 229         {ok, auto} ->
 230             {ok, Rate} = application:get_env(etorrent, max_upload_rate),
 231             case Rate of
 232                 N when N =<  0 -> 7; %% Educated guess
 233                 N when N  <  9 -> 2;
 234                 N when N  < 15 -> 3;
 235                 N when N  < 42 -> 4;
 236                 N ->
 237                     round(math:sqrt(N * 0.6))
 238             end;
 239         {ok, N} when is_integer(N) ->
 240             N
 241     end.
 242
 243 split_preferred(Peers) ->
 244     {Downs, Leechs} = split_preferred_peers(Peers, [], []),
 245     {lists:keysort(#rechoke_info.rate, Downs),
 246      lists:keysort(#rechoke_info.rate, Leechs)}.
 247
 248 prune_preferred_peers(SDowns, SLeechs) ->
 249     MaxUploads = upload_slots(),
 250     DUploads = lists:max([1, round(MaxUploads * 0.7)]),
 251     SUploads = lists:max([1, round(MaxUploads * 0.3)]),
 252     {SUP2, DUP2} =
 253         case lists:max([0, DUploads - length(SDowns)]) of
 254             0 -> {SUploads, DUploads};
 255             N -> {SUploads + N, DUploads - N}
 256         end,
 257     {SUP3, DUP3} =
 258         case lists:max([0, SUP2 - length(SLeechs)]) of
 259             0 -> {SUP2, DUP2};
 260             K ->
 261                 {SUP2 - K, lists:min([DUP2 + K, length(SDowns)])}
 262         end,
 263     {TSDowns, TSLeechs} = {lists:sublist(SDowns, DUP3),
 264                            lists:sublist(SLeechs, SUP3)},
 265     sets:union(sets:from_list(TSDowns), sets:from_list(TSLeechs)).
 266
 267 rechoke_unchoke([], _PS) ->
 268     [];
 269 rechoke_unchoke([P | Next], PSet) ->
 270     case sets:is_element(P, PSet) of
 271         true ->
 272             etorrent_t_peer_recv:unchoke(P#rechoke_info.pid),
 273             rechoke_unchoke(Next, PSet);
 274         false ->
 275             [P | rechoke_unchoke(Next, PSet)]
 276     end.
 277
 278 optimistics(PSet) ->
 279     MinUp = case application:get_env(etorrent, min_uploads) of
 280                 {ok, N} -> N;
 281                 undefined -> 1
 282             end,
 283     lists:max([MinUp, upload_slots() - sets:size(PSet)]).
 284
 285 rechoke_choke([], _Count, _Optimistics) ->
 286     ok;
 287 rechoke_choke([P | Next], Count, Optimistics) when Count >= Optimistics ->
 288     etorrent_t_peer_recv:choke(P#rechoke_info.pid),
 289     rechoke_choke(Next, Count, Optimistics);
 290 rechoke_choke([P | Next], Count, Optimistics) ->
 291     case P#rechoke_info.kind =:= seeding of
 292         true ->
 293             etorrent_t_peer_recv:choke(P#rechoke_info.pid),
 294             rechoke_choke(Next, Count, Optimistics);
 295         false ->
 296             etorrent_t_peer_recv:unchoke(P#rechoke_info.pid),
 297             case P#rechoke_info.r_interest_state =:= interested of
 298                 true ->
 299                     rechoke_choke(Next, Count+1, Optimistics);
 300                 false ->
 301                     rechoke_choke(Next, Count, Optimistics)
 302             end
 303     end.
 304
 305 split_preferred_peers([], Downs, Leechs) ->
 306     {Downs, Leechs};
 307 split_preferred_peers([P | Next], Downs, Leechs) ->
 308     case P#rechoke_info.kind =:= seeding
 309           orelse P#rechoke_info.r_interest_state =:= not_interested of
 310         true ->
 311             split_preferred_peers(Next, Downs, Leechs);
 312         false when P#rechoke_info.state =:= seeding ->
 313             split_preferred_peers(Next, Downs, [P | Leechs]);
 314         false when P#rechoke_info.snubbed =:= true ->
 315             split_preferred_peers(Next, Downs, Leechs);
 316         false ->
 317             split_preferred_peers(Next, [P | Downs], Leechs)
 318     end.
 319
 320 seeding_torrents() ->
 321     {atomic, Torrents} = etorrent_torrent:all(),
 322     [T#torrent.id || T <- Torrents,
 323                      T#torrent.state =:= seeding].
 324