[Mldonkey-commits] mldonkey distrib/ChangeLog src/networks/bittorr...

[mldonkey-commits]

CVSROOT:        /sources/mldonkey
Module name:    mldonkey
Changes by:     spiralvoice <spiralvoice>       11/03/20 17:59:46

Modified files:
        distrib        : ChangeLog 
        src/networks/bittorrent: bTGlobals.ml bTOptions.ml bT_DHT.ml 
                                 kademlia.ml 

Log message:
        patch #7506

CVSWeb URLs:
http://cvs.savannah.gnu.org/viewcvs/mldonkey/distrib/ChangeLog?cvsroot=mldonkey&r1=1.1524&r2=1.1525
http://cvs.savannah.gnu.org/viewcvs/mldonkey/src/networks/bittorrent/bTGlobals.ml?cvsroot=mldonkey&r1=1.89&r2=1.90
http://cvs.savannah.gnu.org/viewcvs/mldonkey/src/networks/bittorrent/bTOptions.ml?cvsroot=mldonkey&r1=1.44&r2=1.45
http://cvs.savannah.gnu.org/viewcvs/mldonkey/src/networks/bittorrent/bT_DHT.ml?cvsroot=mldonkey&r1=1.3&r2=1.4
http://cvs.savannah.gnu.org/viewcvs/mldonkey/src/networks/bittorrent/kademlia.ml?cvsroot=mldonkey&r1=1.2&r2=1.3

Patches:
Index: distrib/ChangeLog
===================================================================
RCS file: /sources/mldonkey/mldonkey/distrib/ChangeLog,v
retrieving revision 1.1524
retrieving revision 1.1525
diff -u -b -r1.1524 -r1.1525
--- distrib/ChangeLog   14 Mar 2011 18:57:25 -0000      1.1524
+++ distrib/ChangeLog   20 Mar 2011 17:59:46 -0000      1.1525
@@ -14,6 +14,9 @@
 ChangeLog
 =========
 
+2011/03/20
+7506: BT/DHT: show stats, choose random port (ygrek)
+
 2011/03/14
 7501: Options: Disable Overnet by default and remove contact.dat from web_infos
 

Index: src/networks/bittorrent/bTGlobals.ml
===================================================================
RCS file: /sources/mldonkey/mldonkey/src/networks/bittorrent/bTGlobals.ml,v
retrieving revision 1.89
retrieving revision 1.90
diff -u -b -r1.89 -r1.90
--- src/networks/bittorrent/bTGlobals.ml        10 Mar 2011 19:23:54 -0000      
1.89
+++ src/networks/bittorrent/bTGlobals.ml        20 Mar 2011 17:59:46 -0000      
1.90
@@ -925,7 +925,7 @@
 
 **************************************************************)
 
-let _ =
+let () =
   Heap.add_memstat "BittorrentGlobals" (fun level buf ->
      Printf.bprintf buf "Number of old files: %d\n" (List.length !!old_files);
      let downloads = ref 0 in
@@ -940,3 +940,28 @@
      Printf.bprintf buf "files_by_uid: %d\n" (Hashtbl.length files_by_uid);
      Printf.bprintf buf "ft_by_num: %d\n" (Hashtbl.length ft_by_num);
   )
+
+open BT_DHT
+
+let () =
+  Heap.add_memstat "BittorrentDHT" (fun _level buf ->
+    match !bt_dht with
+    | None -> ()
+    | Some dht ->
+    let (buckets,nodes,keys,peers) = stat dht in
+    Printf.bprintf buf "Routing : %d nodes in %d buckets\n" nodes buckets;
+    Printf.bprintf buf "Storage : %d keys with %d peers\n" keys peers;
+    List.iter (fun s -> Printf.bprintf buf "%s\n" s) (rpc_stats dht);
+    let queries = 
["PING",`Ping;"FIND_NODE",`FindNode;"GET_PEERS",`GetPeers;"ANNOUNCE",`Announce] 
in
+    Printf.bprintf buf "Outgoing queries : ok/error/timeout\n";
+    List.iter begin fun (name,qt) ->
+      let get k = try Hashtbl.find dht.M.stats (qt,`Out k) with Not_found -> 0 
in
+      Printf.bprintf buf "%s: %d/%d/%d\n" name (get `Answer) (get `Error) (get 
`Timeout);
+      end queries;
+    Printf.bprintf buf "Incoming queries\n";
+    List.iter begin fun (name,qt) ->
+      let get () = try Hashtbl.find dht.M.stats (qt,`In) with Not_found -> 0 in
+      Printf.bprintf buf "%s: %d\n" name (get ())
+      end queries
+  )
+

Index: src/networks/bittorrent/bTOptions.ml
===================================================================
RCS file: /sources/mldonkey/mldonkey/src/networks/bittorrent/bTOptions.ml,v
retrieving revision 1.44
retrieving revision 1.45
diff -u -b -r1.44 -r1.45
--- src/networks/bittorrent/bTOptions.ml        23 Jan 2011 15:20:26 -0000      
1.44
+++ src/networks/bittorrent/bTOptions.ml        20 Mar 2011 17:59:46 -0000      
1.45
@@ -165,8 +165,8 @@
   else !!user_agent
 
 let dht_port = define_option bittorrent_section ["dht_port"]
-  "The UDP port to bind the DHT node to (0 to disable)"
-  port_option 12345
+  "DHT port (UDP, set 0 to disable)"
+  port_option (2000 + Random.int 20000)
 
 let use_trackers = define_option bittorrent_section ["use_trackers"]
   "Send announces to trackers"

Index: src/networks/bittorrent/bT_DHT.ml
===================================================================
RCS file: /sources/mldonkey/mldonkey/src/networks/bittorrent/bT_DHT.ml,v
retrieving revision 1.3
retrieving revision 1.4
diff -u -b -r1.3 -r1.4
--- src/networks/bittorrent/bT_DHT.ml   10 Mar 2011 19:26:22 -0000      1.3
+++ src/networks/bittorrent/bT_DHT.ml   20 Mar 2011 17:59:46 -0000      1.4
@@ -56,6 +56,8 @@
 
 end
 
+let stats_add h k n = Hashtbl.replace h k (n + try Hashtbl.find h k with 
Not_found -> 0)
+
 module KRPC = struct
 
 type dict = (string * Bencode.value) list
@@ -120,12 +122,26 @@
 
 module A = Assoc2
 
-let send sock (ip,port as addr) txnmsg =
+let send sock stats (ip,port as addr) txnmsg =
   let s = encode txnmsg in
   if !debug then lprintf_nl "KRPC to %s : %S" (show_addr addr) s;
+  stats_add stats `Sent 1;
+  stats_add stats `SentBytes (String.length s);
   write sock false s ip port
 
-type t = UdpSocket.t * (addr, string, (addr -> dict -> unit) * (unit -> unit) 
* int) A.t
+type stats_key = [ `Timeout | `Sent | `SentBytes | `Recv | `RecvBytes | 
`Decoded | `Handled | `NoTxn ]
+type t =
+  UdpSocket.t *
+  (stats_key, int) Hashtbl.t * 
+  (addr, string, (addr -> dict -> unit) * ([`Error|`Timeout]-> unit) * int) A.t
+let show_stats t =
+  let get k = try Hashtbl.find t k with Not_found -> 0 in
+  [
+    sprintf "rpc recv %d pkts (%d bytes)" (get `Recv) (get `RecvBytes);
+    sprintf "rpc sent %d pkts (%d bytes)" (get `Sent) (get `SentBytes);
+    sprintf "rpc decoded %d, handled %d" (get `Decoded) (get `Handled);
+    sprintf "rpc timeouted %d, orphan %d" (get `Timeout) (get `NoTxn);
+  ]
 
 let create port enabler bw_control answer : t =
   let socket = create Unix.inet_addr_any port (fun sock event ->
@@ -141,15 +157,17 @@
   set_wtimeout (sock socket) 5.;
   set_rtimeout (sock socket) 5.;
   let h = A.create () in
+  let stats = Hashtbl.create 10 in
   let timeout h =
     let now = last_time () in
     let bad = ref [] in
     let total = ref 0 in
     A.iter h (fun addr txn (_,kerr,t) -> incr total; if t < now then bad := 
(addr,txn,kerr) :: !bad);
     if !debug then lprintf_nl "timeouted %d of %d DHT queries" (List.length 
!bad) !total;
+    stats_add stats `Timeout (List.length !bad);
     List.iter (fun (addr,txn,kerr) ->
       A.remove h addr txn;
-      try kerr () with exn -> if !debug then lprintf_nl ~exn "timeout for %s" 
(show_addr addr)) !bad;
+      try kerr `Timeout with exn -> if !debug then lprintf_nl ~exn "timeout 
for %s" (show_addr addr)) !bad;
   in
   BasicSocket.add_session_timer enabler 5. (fun () -> timeout h);
   let handle addr (txn,ver,msg) =
@@ -159,15 +177,19 @@
     | Error (code,msg) ->
         if !verb then lprintf_nl "error received from %s%s : %Ld %S" 
(show_addr addr) !!version code msg;
         begin match A.find h addr txn with
-        | None -> if !verb then lprintf_nl "no txn %S for %s%s" txn (show_addr 
addr) !!version
-        | Some (_, kerr, _) -> A.remove h addr txn; kerr ()
+        | None ->
+          stats_add stats `NoTxn 1;
+          if !verb then lprintf_nl "no txn %S for %s%s" txn (show_addr addr) 
!!version
+        | Some (_, kerr, _) -> A.remove h addr txn; kerr `Error
         end
     | Query (name,args) ->
         let ret = answer addr name args in
-        send socket addr (txn, ret)
+        send socket stats addr (txn, ret)
     | Response ret ->
         match A.find h addr txn with
-        | None -> if !verb then lprintf_nl "no txn %S for %s%s" txn (show_addr 
addr) !!version
+        | None ->
+          stats_add stats `NoTxn 1;
+          if !verb then lprintf_nl "no txn %S for %s%s" txn (show_addr addr) 
!!version
         | Some (k,_,_) -> A.remove h addr txn; k addr ret
   in
   let handle p =
@@ -177,13 +199,17 @@
       let addr = (Ip.of_inet_addr inet_addr, port) in
       let ret = ref None in
       try
+        stats_add stats `RecvBytes (String.length p.udp_content);
+        stats_add stats `Recv 1;
         let r = decode_exn p.udp_content in
+        stats_add stats `Decoded 1;
         ret := Some r;
-        handle addr r
+        handle addr r;
+        stats_add stats `Handled 1;
       with exn ->
         let version = match !ret with Some (_,Some s,_) -> sprintf " client 
%S" s | _ -> "" in
         if !verb then lprintf_nl ~exn "handle packet from %s%s : %S" 
(show_addr addr) version p.udp_content;
-        let error txn code str = send socket addr (txn,(Error (Int64.of_int 
code,str))) in
+        let error txn code str = send socket stats addr (txn,(Error 
(Int64.of_int code,str))) in
         match exn,!ret with
         | Malformed_packet x, Some (txn, _, _)
         | Protocol_error ("",x), Some(txn, _, _) | Protocol_error (txn,x), _ 
-> error txn 203 x
@@ -192,15 +218,15 @@
         | _ -> ()
   in
   udp_set_reader socket handle;
-  (socket,h)
+  (socket,stats,h)
 
-let shutdown (socket,h) =
+let shutdown (socket,_,h) =
   close socket Closed_by_user;
   A.iter h (fun addr _ (_,kerr,_) ->
-    try kerr () with exn -> if !verb then lprintf_nl ~exn "shutdown for %s" 
(show_addr addr));
+    try kerr `Timeout with exn -> if !verb then lprintf_nl ~exn "shutdown for 
%s" (show_addr addr));
   A.clear h
 
-let write (socket,h) msg addr k ~kerr =
+let write (socket,stats,h) msg addr k ~kerr =
   let tt = Assoc2.find_all h addr in
   let rec loop () = (* choose txn FIXME *)
     let txn = string_of_int (Random.int 1_000_000) in
@@ -210,7 +236,7 @@
   in
   let txn = loop () in
   Assoc2.add h addr txn (k,kerr,last_time () + dht_query_timeout);
-  send socket addr (txn,msg)
+  send socket stats addr (txn,msg)
 
 end (* KRPC *)
 
@@ -350,21 +376,33 @@
 
 module M = struct
 
+type query_type = [ `Ping | `FindNode | `GetPeers | `Announce ]
+type answer_type = [ `Answer | `Error | `Timeout ]
+
 type t = {
   rt : Kademlia.table; (* routing table *)
   rpc : KRPC.t; (* KRPC protocol socket *)
   dht_port : int; (* port *)
   torrents : (H.t, int Peers.t) Hashtbl.t; (* torrents announced by other 
peers *)
   enabler : bool ref; (* timers' enabler *)
+  stats : (query_type * [ `In | `Out of answer_type ], int) Hashtbl.t; (* 
statistics *)
 }
 
+let query_type_of_query = function
+| Ping -> `Ping
+| FindNode _ -> `FindNode
+| GetPeers _ -> `GetPeers
+| Announce _ -> `Announce
+
 let dht_query t addr q k ~kerr =
   if !debug then lprintf_nl "DHT query to %s : %s" (show_addr addr) 
(show_query q);
+  let qt = query_type_of_query q in
   KRPC.write t.rpc (make_query t.rt.self q) addr begin fun addr dict ->
-    let (id,r) = try parse_response_exn q dict with exn -> kerr (); raise exn 
in
+    let (id,r) = try parse_response_exn q dict with exn -> stats_add t.stats 
(qt, `Out `Error) 1; kerr (); raise exn in
     if !debug then lprintf_nl "DHT response from %s (%s) : %s" (show_addr 
addr) (show_id id) (show_response r);
+    stats_add t.stats (qt, `Out `Answer) 1;
     k (id,addr) r
-  end ~kerr
+  end ~kerr:(fun reason -> stats_add t.stats (qt, `Out (reason:>answer_type)) 
1; kerr ())
 
 let ping t addr k = dht_query t addr Ping begin fun node r ->
   match r with Ack -> k (Some node)
@@ -407,7 +445,8 @@
   let rpc = KRPC.create dht_port enabler bw_control answer in
   let torrents = Hashtbl.create 8 in
   manage_timeouts enabler torrents;
-  { rt = rt; rpc = rpc; torrents = torrents; dht_port = dht_port; enabler = 
enabler; }
+  { rt = rt; rpc = rpc; torrents = torrents; dht_port = dht_port; enabler = 
enabler;
+    stats = Hashtbl.create 10; }
 
 let shutdown dht =
   dht.enabler := false;
@@ -579,6 +618,12 @@
   torrents
 
 let show dht = show_table dht.M.rt; show_torrents dht.M.torrents
+let stat dht =
+  buckets dht.M.rt,
+  size dht.M.rt,
+  Hashtbl.length dht.M.torrents,
+  Hashtbl.fold (fun _ peers acc -> acc + Peers.fold (fun _ _ acc -> acc + 1) 
peers 0) dht.M.torrents 0
+let rpc_stats dht = let (_,st,_) = dht.M.rpc in KRPC.show_stats st
 
 let bootstrap dht host addr k =
   M.ping dht addr begin function
@@ -642,6 +687,7 @@
     let node = (id,addr) in
     if !debug then lprintf_nl "DHT query from %s : %s" (show_node node) 
(show_query q);
     update !!dht Good id addr;
+    stats_add (!!dht).M.stats (M.query_type_of_query q, `In) 1;
     let response =
       match q with
       | Ping -> Ack

Index: src/networks/bittorrent/kademlia.ml
===================================================================
RCS file: /sources/mldonkey/mldonkey/src/networks/bittorrent/kademlia.ml,v
retrieving revision 1.2
retrieving revision 1.3
diff -u -b -r1.2 -r1.3
--- src/networks/bittorrent/kademlia.ml 10 Mar 2011 19:26:22 -0000      1.2
+++ src/networks/bittorrent/kademlia.ml 20 Mar 2011 17:59:46 -0000      1.3
@@ -352,6 +352,7 @@
   | L b -> f acc b
 
 let size t = fold (fun acc b -> acc + Array.length b.nodes) 0 t.root
+let buckets t = fold (fun acc b -> acc + 1) 0 t.root
 
 (*
 module NoNetwork : Network = struct