/[mldonkey]/mldonkey/src/networks/donkey/donkeyOvernet.ml
ViewVC logotype

Diff of /mldonkey/src/networks/donkey/donkeyOvernet.ml

Parent Directory Parent Directory | Revision Log Revision Log | View Patch Patch

revision 1.54 by spiralvoice, Tue Oct 25 10:30:58 2005 UTC revision 1.55 by spiralvoice, Thu Oct 27 21:32:53 2005 UTC
# Line 50  type peer = Line 50  type peer =
50      mutable peer_port : int;      mutable peer_port : int;
51      mutable peer_tcpport : int;      mutable peer_tcpport : int;
52      mutable peer_kind : int;      mutable peer_kind : int;
53      mutable peer_expire : int;      mutable peer_last_recv : int;
54      mutable peer_last_send : int;      mutable peer_last_send : int;
     peer_created : int;  
55    }    }
56    
57  type search_kind =  type search_kind =
# Line 137  search for files and search for sources. Line 136  search for files and search for sources.
136  | OvernetUnknown21 of peer  | OvernetUnknown21 of peer
137    
138  let print_peer buf p =  let print_peer buf p =
139    let hours = (last_time () - p.peer_created) / 3600 in    Printf.bprintf buf "   { md4 = %s ip = %s port = %d %s kind = %d l_send = %d l_recv = %d}\n"
   let mins = (last_time () - p.peer_created) / 60 in  
   Printf.bprintf buf "   { md4 = %s ip = %s port = %d %s kind = %d l_send = %d expire = %d created = %d h = %d mins = %d %s}\n"  
140      (Md4.to_string p.peer_md4) (Ip.to_string p.peer_ip) p.peer_port      (Md4.to_string p.peer_md4) (Ip.to_string p.peer_ip) p.peer_port
141      (if p.peer_tcpport <> 0 then Printf.sprintf "tcp = %d" p.peer_tcpport      (if p.peer_tcpport <> 0 then Printf.sprintf "tcp = %d" p.peer_tcpport
142      else "") p.peer_kind p.peer_last_send p.peer_expire p.peer_created      else "") p.peer_kind p.peer_last_send p.peer_last_recv
     hours mins (if p.peer_expire > last_time () && p.peer_last_send <> 0 then "OK " else "")  
       
143    
144  let message_to_string t =  let message_to_string t =
145    let buf = Buffer.create 100 in    let buf = Buffer.create 100 in
# Line 279  type overnet_search = { Line 274  type overnet_search = {
274      search_md4 : Md4.t;      search_md4 : Md4.t;
275      mutable search_kind : search_for;      mutable search_kind : search_for;
276    
277        search_known_peers : (Ip.t * int, peer) Hashtbl.t;
278      search_waiting_peers : peer Fifo.t array;      search_waiting_peers : peer Fifo.t array;
279      search_asked_peers : peer Fifo.t array;      search_asked_peers : peer Fifo.t array;
280      search_ok_peers : peer Fifo.t array;      search_ok_peers : peer Fifo.t array;
281    
282      mutable search_queries : int;      mutable search_queries : int;
     (* TODO: search_requests is not used ? *)  
283      mutable search_requests : int;      mutable search_requests : int;
284      mutable search_start : int;      mutable search_start : int;
285      mutable search_results : (Md4.t, tag list) Hashtbl.t;      mutable search_results : (Md4.t, tag list) Hashtbl.t;
# Line 331  module Make(Proto: sig Line 326  module Make(Proto: sig
326  *********************************************************************)  *********************************************************************)
327    
328  let max_peers_per_bucket = 20  let max_peers_per_bucket = 20
329  let max_peers_per_prebucket = 40  let max_peers_per_prebucket = 100
330    let min_peers_per_block = 16 (* was 2 *)
331    let min_peers_before_connect = 5
332    let max_searches_for_publish = 5
333  let max_search_queries = 64  let max_search_queries = 64
334  let max_search_requests = 20  let max_search_requests = 20
335  let max_boot_peers = 200  let max_boot_peers = 200
# Line 422  let overnet_options_version = Line 420  let overnet_options_version =
420    
421  *********************************************************************)  *********************************************************************)
422    
 let dummy_peer =  
   {  
     peer_md4 = Md4.null;  
     peer_ip = Ip.null;  
     peer_port = 0;  
     peer_tcpport = 0;  
     peer_kind = 0;  
     peer_expire = 0;  
     peer_last_send = 0;  
     peer_created = last_time ();  
   }  
   
423  let connected_peers = ref 0  let connected_peers = ref 0
424  let pre_connected_peers = ref 0  let pre_connected_peers = ref 0
425    
# Line 543  are peers that we are not sure of. *) Line 529  are peers that we are not sure of. *)
529  let buckets = Array.init 129 (fun _ -> Fifo.create ())  let buckets = Array.init 129 (fun _ -> Fifo.create ())
530  let prebuckets = Array.init 129 (fun _ -> Fifo.create ())  let prebuckets = Array.init 129 (fun _ -> Fifo.create ())
531    
532    let known_peers = Hashtbl.create 127
533  let to_ping = ref []  let to_ping = ref []
 module KnownPeers = Weak.Make(struct  
       type t = peer  
       let hash c = Hashtbl.hash (c.peer_ip, c.peer_port)  
       let equal x y = x.peer_port = y.peer_port  
           && x.peer_ip = y.peer_ip  
   end)  
   
 let known_peers = KnownPeers.create 1023  
534    
535  (*  (*
536  We keep the data in buckets depending on the number of bits they have  We keep the data in buckets depending on the number of bits they have
# Line 714  module PublishedFiles = Publish(struct Line 693  module PublishedFiles = Publish(struct
693  let debug_client ip = false  let debug_client ip = false
694  (*  Ip.matches ip !!overnet_debug_clients *)  (*  Ip.matches ip !!overnet_debug_clients *)
695    
 let checking_kind p =  
 (* according to eMule 0.46c *)  
   if (last_time () - p.peer_last_send) < 10 ||  
      p.peer_kind = 4 then () else begin  
     p.peer_kind <- p.peer_kind + 1;  
     p.peer_expire <- last_time () + 120;  
   end;  
   ()  
     
 let new_peer_message p =  
 (*  lprintf () "*** Updating time for %s:%d\n" (Ip.to_string p.peer_ip) p.peer_port; *)  
   let calc_peer_kind p =  
     (* according to eMule 0.46c *)  
     let hours = (last_time () - p.peer_created) / 3600 in  
     let lt = last_time () in  
     match hours with  
       0 -> (2, lt + 3600)  (* new kind 2, 1h timeout to ping *)  
     | 1 -> (1, lt + 5400)  (* kind 1, 1 1/2h timeout *)  
     | _ -> (0, lt + 7200)  (* kind 0, 2h timeout *)  
   in  
   let (new_kind,new_expire) = calc_peer_kind p in  
   p.peer_kind <- new_kind;  
   p.peer_expire <- new_expire;  
   ()  
   
696  let udp_send_direct ip port msg =  let udp_send_direct ip port msg =
697    match !udp_sock with    match !udp_sock with
698      None -> ()      None -> ()
# Line 757  let udp_send p msg = Line 711  let udp_send p msg =
711    
712  let bootstrap ip port =  let bootstrap ip port =
713    if !!overnet_update_nodes && is_overnet_ip ip &&    if !!overnet_update_nodes && is_overnet_ip ip &&
714       port <> 0 && not (KnownPeers.mem known_peers { dummy_peer with peer_port = port ; peer_ip = ip } ) then       port <> 0 && not (Hashtbl.mem known_peers (ip,port)) then
715         LimitedList.add unknown_peers (ip,port)         LimitedList.add unknown_peers (ip,port)
716    
717  let new_peer p =  let new_peer p =
718    let ip = p.peer_ip in    let ip = p.peer_ip in
719    let port = p.peer_port in    if ip <> Ip.localhost && is_overnet_ip ip &&
720        p.peer_port <> 0 then
721    if ip <> Ip.localhost && is_overnet_ip ip && port <> 0 then      let key = (p.peer_ip, p.peer_port) in
   
   
722      try      try
723        let pp = KnownPeers.find known_peers p in        let pp = Hashtbl.find known_peers key in
724        (* TODO: check for changes in ip:port? Or Hash? *)        if pp.peer_last_recv < p.peer_last_recv then
725            pp.peer_last_recv <- p.peer_last_recv;
726        pp        pp
727      with _ ->      with _ ->
728    
 (* Add the peer to the table of known peers *)  
         KnownPeers.add known_peers p;  
   
729          let bucket = bucket_number p.peer_md4 in          let bucket = bucket_number p.peer_md4 in
730  (* First, enter the peer in the boot_peers to be able to use at next  (* First, enter the peer in the boot_peers to be able to use at next
731  restart. *)  restart. *)
# Line 791  restart. *) Line 741  restart. *)
741                let l = last_time () - 1800 in                let l = last_time () - 1800 in
742                let pp = Fifo.take prebuckets.(bucket) in                let pp = Fifo.take prebuckets.(bucket) in
743                Fifo.put prebuckets.(bucket)                Fifo.put prebuckets.(bucket)
744  (* If the head of the bucket is not dead we should keep it *)  (* If we heard of the head of the bucket in the last 30 minutes, we should
745                  (if pp.peer_kind < 4 || pp.peer_expire > last_time () then pp else p);    keep it. *)
746                    (if pp.peer_last_recv > l then pp else p);
747                  if pp.peer_last_recv <= l then
748                  begin
749    (* Add the peer to the table of known peers *)
750                    Hashtbl.add known_peers key p;
751    (* And remove pp *)
752                    let ppkey = (pp.peer_ip, pp.peer_port) in
753                    Hashtbl.remove known_peers ppkey
754                  end
755              end                end  
756              else begin              else begin
757                   Hashtbl.add known_peers key p;
758                 Fifo.put prebuckets.(bucket) p;                 Fifo.put prebuckets.(bucket) p;
759                 incr pre_connected_peers                 incr pre_connected_peers
760              end              end
# Line 802  restart. *) Line 762  restart. *)
762          else if !n_used_buckets < 128 && bucket <> 128 then begin          else if !n_used_buckets < 128 && bucket <> 128 then begin
763              Fifo.put prebuckets.(!n_used_buckets) p;              Fifo.put prebuckets.(!n_used_buckets) p;
764              incr pre_connected_peers;              incr pre_connected_peers;
765                Hashtbl.add known_peers key p;
766    
767              while !n_used_buckets < 128 &&              while !n_used_buckets < 128 &&
768                Fifo.length prebuckets.(!n_used_buckets)                Fifo.length prebuckets.(!n_used_buckets)
# Line 809  restart. *) Line 770  restart. *)
770                let b = prebuckets.(!n_used_buckets) in                let b = prebuckets.(!n_used_buckets) in
771                incr n_used_buckets;                incr n_used_buckets;
772                for i = 1 to Fifo.length b do                for i = 1 to Fifo.length b do
773                  let pp = Fifo.take b in                  let p = Fifo.take b in
774                  let bucket = bucket_number p.peer_md4 in                  let bucket = bucket_number p.peer_md4 in
775                  Fifo.put (if bucket >= !n_used_buckets then                  Fifo.put (if bucket >= !n_used_buckets then
776                      prebuckets.(!n_used_buckets) else b) pp                      prebuckets.(!n_used_buckets) else b) p
777                done                done
778              done              done
779          end;          end;
# Line 831  let get_closest_peers md4 nb = Line 792  let get_closest_peers md4 nb =
792        if n > 0 then        if n > 0 then
793          let p = Fifo.take fifo in          let p = Fifo.take fifo in
794          Fifo.put fifo p;          Fifo.put fifo p;
795          (* kind < 4 so we do not send too much requests and avoid dead contacts not          if p.peer_last_recv <> 0 then begin
            yet removed because of timeouts *)  
         (* TODO: Keep order? Then we need a in_use flag? *)  
         if p.peer_kind < 4 && p.peer_expire > last_time () &&  
            p.peer_last_send <> 0 then begin  
796              if !verbose_overnet then begin              if !verbose_overnet then begin
797              lprintf_nl () "Adding good search peer %s:%d"              lprintf_nl () "Adding good search peer %s:%d"
798                (Ip.to_string p.peer_ip) p.peer_port;                (Ip.to_string p.peer_ip) p.peer_port;
# Line 847  let get_closest_peers md4 nb = Line 804  let get_closest_peers md4 nb =
804      in      in
805      iter (min !nb (Fifo.length fifo))      iter (min !nb (Fifo.length fifo))
806    in    in
807    add_list bucket;    add_list  bucket;
808    
809    if !nb > 0 then begin    if !nb > 0 then begin
810    
# Line 877  let my_peer () = Line 834  let my_peer () =
834      peer_port = !!overnet_port;      peer_port = !!overnet_port;
835      peer_kind = 0;      peer_kind = 0;
836      peer_tcpport = !!overnet_tcpport;      peer_tcpport = !!overnet_tcpport;
837        peer_last_recv = 0;
838      peer_last_send = 0;      peer_last_send = 0;
     peer_expire = 0;  
     peer_created = 0;  
839    }    }
840    
841  let get_any_peers  nb =  let get_any_peers  nb =
# Line 890  let get_any_peers  nb = Line 846  let get_any_peers  nb =
846      let fifo = buckets.(bucket) in      let fifo = buckets.(bucket) in
847      let rec iter n =      let rec iter n =
848        if n > 0 then        if n > 0 then
       (* TODO: keep order *)  
849          let p = Fifo.take fifo in          let p = Fifo.take fifo in
850          Fifo.put fifo p;          Fifo.put fifo p;
851          decr nb;          decr nb;
# Line 909  let get_any_peers  nb = Line 864  let get_any_peers  nb =
864    !list    !list
865    
866  let add_search_peer s p =  let add_search_peer s p =
867    if p.peer_ip <> Ip.localhost && is_overnet_ip p.peer_ip &&    if p.peer_ip <> client_ip None then
868       p.peer_port <> 0 then begin    let key = (p.peer_ip, p.peer_port) in
869      let nbits = common_bits p.peer_md4 s.search_md4 in    if not (Hashtbl.mem s.search_known_peers key) then begin
870      begin        Hashtbl.add s.search_known_peers key p;
871        try        let nbits = common_bits p.peer_md4 s.search_md4 in
872          Fifo.iter (fun pp ->        Fifo.put s.search_waiting_peers.(nbits) p;
           if pp.peer_ip = p.peer_ip &&  
              pp.peer_port = p.peer_port then  
             raise Exit  
         ) s.search_waiting_peers.(nbits);  
         Fifo.put s.search_waiting_peers.(nbits) p;  
       with Exit -> ()  
873      end      end
   end  
874    
875  let create_search kind md4 =  let create_search kind md4 =
876    if !verbose_overnet then lprintf_nl () "create_search";    if !verbose_overnet then lprintf_nl () "create_search";
877    let s = {    let s = {
878        search_md4 = md4;        search_md4 = md4;
879        search_kind = kind;        search_kind = kind;
880          search_known_peers = Hashtbl.create 127;
881        search_queries = 0;        search_queries = 0;
882        search_requests = 0;        search_requests = 0;
883        search_waiting_peers = Array.init 129 (fun _ -> Fifo.create ());        search_waiting_peers = Array.init 129 (fun _ -> Fifo.create ());
# Line 960  let port_of_udp_packet p = Line 909  let port_of_udp_packet p =
909      Unix.ADDR_INET (inet, port) -> port      Unix.ADDR_INET (inet, port) -> port
910    | _ -> assert false    | _ -> assert false
911    
912    
913    let new_peer_message p =
914    (*  lprintf () "*** Updating time for %s:%d\n" (Ip.to_string p.peer_ip) p.peer_port; *)
915      p.peer_last_recv <- last_time ()
916    
917  let udp_client_handler t p =  let udp_client_handler t p =
918    let other_ip = ip_of_udp_packet p in    let other_ip = ip_of_udp_packet p in
919    let other_port = port_of_udp_packet p in    let other_port = port_of_udp_packet p in
# Line 967  let udp_client_handler t p = Line 921  let udp_client_handler t p =
921      lprintf_nl () "UDP FROM %s:%d type %s"      lprintf_nl () "UDP FROM %s:%d type %s"
922        (Ip.to_string other_ip) other_port        (Ip.to_string other_ip) other_port
923        (message_to_string t);        (message_to_string t);
   (* Emule uses other_ip:other_port, so do we *)  
   (* Update known_peers by other_port:other_ip *)  
   let sender = { dummy_peer with peer_port = other_port ; peer_ip = other_ip } in  
   begin  
     try  
       let sender = KnownPeers.find known_peers sender in  
       (* TODO: check for changes in ip:port? Or Hash? *)  
       new_peer_message sender;  
     with _ -> ();  
   end;  
924    match t with    match t with
925    
926    | OvernetConnect p ->    | OvernetConnect p ->
927        if is_overnet_ip sender.peer_ip && sender.peer_port <> 0 then        let rec send p =
928          udp_send sender (OvernetConnectReply (get_any_peers 20))          new_peer_message p;
929         else          let p = new_peer p in
930           begin            udp_send p (OvernetConnectReply (get_any_peers 20))
931             if !verbose_overnet then        in
932               lprintf_nl () "Connect: invalid IP %s:%d received from %s:%d"          if is_overnet_ip p.peer_ip && p.peer_port <> 0 then
933              send p
934            else
935              if is_overnet_ip other_ip && other_port <> 0 then
936                begin
937                  if !verbose_overnet then
938                    lprintf_nl () "Connect: convert address %s:%d to %s:%d"
939                      (Ip.to_string p.peer_ip) p.peer_port (Ip.to_string other_ip) other_port;
940                  p.peer_ip <- other_ip;
941                  p.peer_port <- other_port;
942                  send p
943                end
944              else
945                begin
946                  if !verbose_overnet then
947                    lprintf_nl () "Connect: invalid IP %s:%d received from %s:%d"
948                    (Ip.to_string p.peer_ip) p.peer_port (Ip.to_string other_ip) other_port;                    (Ip.to_string p.peer_ip) p.peer_port (Ip.to_string other_ip) other_port;
949               failwith "Message not understood"                failwith "Message not understood"
950           end              end
951    
952    | OvernetConnectReply ps ->    | OvernetConnectReply ps ->
953        UdpSocket.declare_pong other_ip;        UdpSocket.declare_pong other_ip;
# Line 996  let udp_client_handler t p = Line 955  let udp_client_handler t p =
955          match list with          match list with
956            [] -> ()            [] -> ()
957          | [p] ->          | [p] ->
958                new_peer_message p;
959                if other_port <> p.peer_port || other_ip <> p.peer_ip then
960                  if !verbose_overnet then
961                  lprintf_nl () "Bad IP or port";
962              let p = new_peer p in              let p = new_peer p in
963              ()              ()
964          | p :: tail ->          | p :: tail ->
# Line 1005  let udp_client_handler t p = Line 968  let udp_client_handler t p =
968        iter ps;        iter ps;
969    
970    | OvernetPublicize p ->    | OvernetPublicize p ->
971        if is_overnet_ip sender.peer_ip && sender.peer_port <> 0 then        let rec send p =
972          udp_send sender (OvernetPublicized (Some (my_peer ())))          new_peer_message p;
973         else begin          let p = new_peer p in
974              udp_send p (OvernetPublicized (Some (my_peer ())))
975          in
976            if is_overnet_ip p.peer_ip && p.peer_port <> 0 then
977              send p
978            else
979              if is_overnet_ip other_ip && other_port <> 0 then
980                begin
981                  if !verbose_overnet then
982                    lprintf_nl () "Publicize: convert address %s:%d to %s:%d"
983                      (Ip.to_string p.peer_ip) p.peer_port (Ip.to_string other_ip) other_port;
984                  p.peer_ip <- other_ip;
985                  p.peer_port <- other_port;
986                  send p
987                end
988              else
989                begin
990                if !verbose_overnet then                if !verbose_overnet then
991                  lprintf_nl () "Publicize: invalid IP %s:%d received from %s:%d"                  lprintf_nl () "Publicize: invalid IP %s:%d received from %s:%d"
992                    (Ip.to_string p.peer_ip) p.peer_port (Ip.to_string other_ip) other_port;                    (Ip.to_string p.peer_ip) p.peer_port (Ip.to_string other_ip) other_port;
# Line 1018  let udp_client_handler t p = Line 997  let udp_client_handler t p =
997        ()        ()
998    
999    | OvernetPublicized (Some p) ->    | OvernetPublicized (Some p) ->
1000          new_peer_message p;
1001          let p = new_peer p in
1002        ()        ()
1003    
1004    | OvernetSearch (nresults, md4, from_who) ->    | OvernetSearch (nresults, md4, from_who) ->
1005    
1006        let peers = get_closest_peers md4 nresults in        let peers = get_closest_peers md4 nresults in
1007        udp_send sender (OvernetSearchReply (md4,peers))        udp_send_direct other_ip other_port (OvernetSearchReply (md4,peers))
1008    
1009    | OvernetSearchReply (md4, peers) ->    | OvernetSearchReply (md4, peers) ->
1010    
# Line 1030  let udp_client_handler t p = Line 1012  let udp_client_handler t p =
1012        List.iter (fun s ->        List.iter (fun s ->
1013            if s.search_md4 = md4 then begin            if s.search_md4 = md4 then begin
1014                List.iter (add_search_peer s) peers;                List.iter (add_search_peer s) peers;
1015                  let nbits = common_bits sender.peer_md4 s.search_md4 in                try
1016                  Fifo.put s.search_ok_peers.(nbits) sender;                  let p = Hashtbl.find s.search_known_peers
1017                        (other_ip, other_port) in
1018                    new_peer_message p;
1019                    let nbits = common_bits p.peer_md4 s.search_md4 in
1020                    Fifo.put s.search_ok_peers.(nbits) p;
1021                  with _ -> ()
1022              end              end
1023        ) !overnet_searches;        ) !overnet_searches;
1024    
# Line 1047  let udp_client_handler t p = Line 1034  let udp_client_handler t p =
1034            if s.search_md4 = md4 then begin            if s.search_md4 = md4 then begin
1035                s.search_nresults <- s.search_nresults + 1;                s.search_nresults <- s.search_nresults + 1;
1036    
1037                  begin
1038                    try
1039                      let p = Hashtbl.find s.search_known_peers
1040                          (other_ip, other_port) in
1041                      new_peer_message p;
1042                    with _ -> ()
1043                  end;
1044    
1045                match s.search_kind with                match s.search_kind with
1046                  FileSearch file -> ()                  FileSearch file -> ()
1047    
# Line 1076  let udp_client_handler t p = Line 1071  let udp_client_handler t p =
1071            if s.search_md4 = md4 then begin            if s.search_md4 = md4 then begin
1072                s.search_nresults <- s.search_nresults + 1;                s.search_nresults <- s.search_nresults + 1;
1073    
1074                  begin
1075                    try
1076                      let p = Hashtbl.find s.search_known_peers
1077                          (other_ip, other_port) in
1078                      new_peer_message p;
1079                    with _ -> ()
1080                  end;
1081    
1082                match s.search_kind with                match s.search_kind with
1083                  FileSearch file ->                  FileSearch file ->
1084                    List.iter (fun p ->                    List.iter (fun p ->
# Line 1115  let udp_client_handler t p = Line 1118  let udp_client_handler t p =
1118              let _, list = List2.cut min list in              let _, list = List2.cut min list in
1119              let list, _ = List2.cut (max - min) list in              let list, _ = List2.cut (max - min) list in
1120              if list <> [] then              if list <> [] then
1121                udp_send sender                udp_send_direct other_ip other_port
1122                  (OvernetSearchSourcesResults (md4, list))                  (OvernetSearchSourcesResults (md4, list))
1123          | Search_for_keyword _ ->          | Search_for_keyword _ ->
1124              let list = PublishedKeywords.get md4 in              let list = PublishedKeywords.get md4 in
1125              let _, list = List2.cut min list in              let _, list = List2.cut min list in
1126              let list, _ = List2.cut (max - min) list in              let list, _ = List2.cut (max - min) list in
1127              if list <> [] then              if list <> [] then
1128                udp_send sender                udp_send_direct other_ip other_port
1129                  (OvernetSearchFilesResults (md4, list))                  (OvernetSearchFilesResults (md4, list))
1130          | _ -> ()          | _ -> ()
1131        end        end
# Line 1140  let udp_client_handler t p = Line 1143  let udp_client_handler t p =
1143         if !verbose_overnet && debug_client other_ip then           if !verbose_overnet && debug_client other_ip then  
1144           lprintf_nl () "GET MY IP (port=%d)\n" other_port;             lprintf_nl () "GET MY IP (port=%d)\n" other_port;  
1145   (* FIXME : should be able to flush the UDP buffer*)     (* FIXME : should be able to flush the UDP buffer*)  
1146         udp_send sender (OvernetGetMyIPResult other_ip);           udp_send_direct other_ip other_port (OvernetGetMyIPResult other_ip);  
1147         udp_send sender OvernetGetMyIPDone           udp_send_direct other_ip other_port OvernetGetMyIPDone  
1148            
1149     | OvernetGetMyIPResult(ip) ->       | OvernetGetMyIPResult(ip) ->  
1150         if !verbose_overnet && debug_client other_ip then           if !verbose_overnet && debug_client other_ip then  
# Line 1157  let udp_client_handler t p = Line 1160  let udp_client_handler t p =
1160            lprintf_nl () "Peer NOT FOUND %s (%s:%d) kind: %d (msg 33)"            lprintf_nl () "Peer NOT FOUND %s (%s:%d) kind: %d (msg 33)"
1161              (Md4.to_string peer.peer_md4) (Ip.to_string peer.peer_ip)              (Md4.to_string peer.peer_md4) (Ip.to_string peer.peer_ip)
1162          peer.peer_port peer.peer_kind;          peer.peer_port peer.peer_kind;
1163          let dp = { dummy_peer with peer_port = peer.peer_port ; peer_ip = peer.peer_ip } in          let key = (peer.peer_ip, peer.peer_port) in
1164          if KnownPeers.mem known_peers dp          if Hashtbl.mem known_peers key
1165          then begin          then begin
1166  (* remove it from the prebuckets and known_peers only *)  (* remove it from the prebuckets and known_peers only *)
1167            try            try
# Line 1169  let udp_client_handler t p = Line 1172  let udp_client_handler t p =
1172                  if p.peer_ip = peer.peer_ip &&                  if p.peer_ip = peer.peer_ip &&
1173                      p.peer_port = peer.peer_port then begin                      p.peer_port = peer.peer_port then begin
1174                    decr pre_connected_peers;                    decr pre_connected_peers;
1175                      Hashtbl.remove known_peers key;
1176                  end else Fifo.put b p                  end else Fifo.put b p
1177                done;                done;
1178              done;              done;
# Line 1196  let query_next_peers () = Line 1200  let query_next_peers () =
1200          if  nbits >= 0 then          if  nbits >= 0 then
1201            let len = Fifo.length s.search_waiting_peers.(nbits) in            let len = Fifo.length s.search_waiting_peers.(nbits) in
1202            if len > 0 then            if len > 0 then
1203              let p = Fifo.take s.search_waiting_peers.(nbits) in              let p = Fifo.take  s.search_waiting_peers.(nbits) in
1204    
             checking_kind p;  
1205              udp_send p (OvernetSearch (nresults, s.search_md4, Some p.peer_md4));              udp_send p (OvernetSearch (nresults, s.search_md4, Some p.peer_md4));
1206              s.search_queries <- s.search_queries + 1;              s.search_queries <- s.search_queries + 1;
1207              Fifo.put s.search_asked_peers.(nbits) p;              Fifo.put  s.search_asked_peers.(nbits) p;
1208    
1209              (if todo > 1 then iter nbits (todo - 1))              (if todo > 1 then iter nbits (todo - 1))
1210            else            else
# Line 1226  let check_current_downloads () = Line 1229  let check_current_downloads () =
1229    
1230  let update_buckets () =  let update_buckets () =
1231    
1232  (* 1. Clean the buckets from too old peers ( by kind and expire ) *)  (* 1. Clean the buckets from too old peers ( last contact > 1 hour ) *)
1233    
1234      let overtime = last_time () - 3600 in
1235    
1236    for i = 0 to !n_used_buckets do    for i = 0 to !n_used_buckets do
1237    
1238      let b = buckets.(i) in      let b = buckets.(i) in
1239      for j = 1 to Fifo.length b do      for j = 1 to Fifo.length b do
1240        let p = Fifo.take b in        let p = Fifo.take b in
1241        (* bad peers have kind = 4 and did not respond within peer_expire *)        if p.peer_last_recv > overtime then Fifo.put b p else
       if not (p.peer_kind = 4 && p.peer_expire <= last_time ()) then  
         Fifo.put b p  
       else  
1242        begin        begin
1243          decr connected_peers;          (* remove the peer also from known_peers *)
1244          if !verbose_overnet then lprintf_nl () "update_bucket1: removing %s:%d" (Ip.to_string p.peer_ip) p.peer_port;          let key = (p.peer_ip, p.peer_port) in
1245        end;          Hashtbl.remove known_peers key;
1246            decr connected_peers
1247          end
1248      done      done
1249    
1250    done;    done;
1251    
1252  (* 2. Complete buckets with new peers from the prebuckets with  (* 2. Complete buckets with new peers from the prebuckets with
1253  ( p.peer_kind < 3 ) *)  ( last_contact < 1 hour ) *)
1254    
1255    for i = 0 to !n_used_buckets do    for i = 0 to !n_used_buckets - 1 do
1256      let b = buckets.(i) in      let b = buckets.(i) in
1257      if Fifo.length b < max_peers_per_bucket then begin      if Fifo.length b < max_peers_per_bucket then begin
1258    
1259            try
1260            let pb = prebuckets.(i) in            let pb = prebuckets.(i) in
1261            for j = 1 to Fifo.length pb do            for j = 1 to Fifo.length pb do
1262    
             (* to keep fifo in order travel all peers *)  
1263              let p = Fifo.take pb in              let p = Fifo.take pb in
1264              (* good peers goto buckets *)              if p.peer_last_recv > overtime &&
1265              if p.peer_kind < 3 && Fifo.length b < max_peers_per_bucket then begin                 p.peer_last_recv >= p.peer_last_send then begin
1266                  Fifo.put b p;                  Fifo.put b p;
1267                  incr connected_peers;                  incr connected_peers;
1268                  decr pre_connected_peers;                  decr pre_connected_peers;
1269              (* bad peers are removed *)                      if Fifo.length b = max_peers_per_bucket then raise Exit
1270              end else if p.peer_kind = 4 && p.peer_expire <= last_time () then begin                end else Fifo.put pb p
1271                decr pre_connected_peers;  
               if !verbose_overnet then lprintf_nl () "update_bucket2: removing %s:%d" (Ip.to_string p.peer_ip) p.peer_port;  
             end else  
             (* the rest returns in prebuckets *)  
             Fifo.put pb p  
1272            done            done
1273            with Exit -> ()
1274    
1275        end        end
1276    done;    done;
1277    
# Line 1279  from the active buckets and prebuckets * Line 1281  from the active buckets and prebuckets *
1281    let n = ref 0 in    let n = ref 0 in
1282    try    try
1283      let addtol p =      let addtol p =
1284        if p.peer_kind < 3 then begin        if p.peer_last_recv <> 0 then begin
1285          LimitedList.add !!boot_peers (p.peer_ip, p.peer_port);          LimitedList.add !!boot_peers (p.peer_ip, p.peer_port);
1286          incr n;          incr n;
1287          if !n = max_boot_peers then raise Exit          if !n = max_boot_peers then raise Exit
# Line 1297  from the active buckets and prebuckets * Line 1299  from the active buckets and prebuckets *
1299    
1300    ()    ()
1301    
 let compute_to_ping () =  
 (* compute which peers to ping in the next minute *)  
           to_ping := [];  
           let n_to_ping = ref 0 in  
   
           let ping_peers b =  
             Fifo.iter (fun p ->  
                 if (p.peer_expire <= last_time () && p.peer_kind < 4) ||  
                     p.peer_last_send = 0 then begin  
                     to_ping := p :: !to_ping;  
                     incr n_to_ping;  
                     if !n_to_ping = 60 then raise Exit  
                   end  
             ) b  
           in  
           begin  
             try  
               for i = !n_used_buckets downto 8 do  
                 ping_peers buckets.(i);  
                 ping_peers prebuckets.(i);  
               done;  
               for i = min !n_used_buckets 7 downto 0 do  
                 ping_peers buckets.(i);  
               done;  
               for i = min !n_used_buckets 7 downto 0 do  
                 ping_peers prebuckets.(i);  
               done;  
             with Exit -> ()  
           end;  
           ()  
   
   
1302  let enable () =  let enable () =
1303    if !!enable_overnet && not !is_enabled then begin    if !!enable_overnet && not !is_enabled then begin
1304        let enabler = is_enabled in        let enabler = is_enabled in
# Line 1349  let enable () = Line 1319  let enable () =
1319              let my_peer = my_peer () in              let my_peer = my_peer () in
1320    
1321  (* ping old peers regularly *)  (* ping old peers regularly *)
1322              let process_to_ping () =              begin
1323                begin                match !to_ping with
1324                  match !to_ping with                  [] -> ()
1325                    [] -> compute_to_ping ();                | p :: tail ->
                 | p :: tail ->  
                   (* do not hammer a peer, we could have send already a search reqeust  
                      since to_ping is rebuild at least every 60 seconds *)  
                   if (last_time () - p.peer_last_send) > 60 then begin  
                     checking_kind p;  
                     udp_send p (OvernetPublicize my_peer);  
                   end;  
1326                    to_ping := tail;                    to_ping := tail;
1327                end;                    p.peer_last_send <- last_time ();
1328              in                    udp_send p (OvernetPublicize my_peer);
1329              process_to_ping ();              end;
1330  (* Send OvernetConnects and ping more peers *)  
1331  (* TODO: How does eMule it? are 50 ok? *)  (* ping unknown peers *)
1332              begin              begin
1333                try                try
1334                   if !connected_peers < 50 then                    if !connected_peers < 100 then
1335                     for i = 1 to 3 do                      for i = 1 to 5 do
                      process_to_ping ();  
                    done;  
                  (* Do not send too much OvernetConnect, there is no use *)  
                  if (!connected_peers + !pre_connected_peers) < 30 then  
                     for i = 1 to 2 do  
1336                        let (ip, port) = LimitedList.get unknown_peers in                        let (ip, port) = LimitedList.get unknown_peers in
1337                        udp_send_ping ip port (OvernetConnect my_peer);                        udp_send_ping ip port (OvernetConnect my_peer);
1338                      done                      done
# Line 1401  let enable () = Line 1359  let enable () =
1359                    in                    in
1360  (* If we use fewer than 5 buckets, we shouldn't wait for 5 buckets in the  (* If we use fewer than 5 buckets, we shouldn't wait for 5 buckets in the
1361    search to start asking for results... *)    search to start asking for results... *)
1362  (* TODO: what the heck we use !n_used_buckets? we ask every search_ok_peers *)                    let min_bucket = mini !n_used_buckets 5 in
1363                    for i = 1 to nrequests do                    for i = 1 to nrequests do
1364                      try                      try
1365                        for j = 128 downto 0 do                        for j = 128 downto min_bucket do
1366                          if Fifo.length s.search_ok_peers.(j) > 0 then                          if Fifo.length s.search_ok_peers.(j) > 0 then
1367                            let p = Fifo.take s.search_ok_peers.(j) in                            let p = Fifo.take s.search_ok_peers.(j) in
1368                            udp_send p (                            udp_send p (
# Line 1428  let enable () = Line 1386  let enable () =
1386    
1387            update_buckets ();            update_buckets ();
1388    
1389            compute_to_ping ();  (* compute which peers to ping in the next minute *)
1390              to_ping := [];
1391              let n_to_ping = ref 0 in
1392    
1393              let ping_peers b =
1394                let overtime = last_time () - 1800 in
1395                Fifo.iter (fun p ->
1396                    if p.peer_last_recv < overtime &&
1397                      p.peer_last_send < overtime then begin
1398                        to_ping := p :: !to_ping;
1399                        incr n_to_ping;
1400                        if !n_to_ping = 60 then raise Exit
1401                      end
1402                ) b
1403              in
1404    
1405              begin
1406                try
1407                  for i = !n_used_buckets downto 8 do
1408                    ping_peers buckets.(i);
1409                    ping_peers prebuckets.(i);
1410                  done;
1411                  for i = min !n_used_buckets 7 downto 0 do
1412                    ping_peers buckets.(i);
1413                  done;
1414                  for i = min !n_used_buckets 7 downto 0 do
1415                    ping_peers prebuckets.(i);
1416                  done;
1417                with Exit -> ()
1418              end;
1419    
1420  (* reset the Hashtbls and Fifos for searches that are older than 5 minutes *)  (* reset the Hashtbls and Fifos for searches that are older than 5 minutes *)
1421            let l = last_time () - 300 in            let l = last_time () - 300 in
1422            List.iter ( fun s ->            List.iter ( fun s ->
1423               if s.search_requests < max_search_requests &&               if s.search_requests < max_search_requests &&
1424               s.search_start > l then begin               s.search_start > l then begin
1425                   Hashtbl.clear s.search_known_peers;
1426                 Array.iter (fun a -> Fifo.clear a) s.search_waiting_peers;                 Array.iter (fun a -> Fifo.clear a) s.search_waiting_peers;
1427                 Array.iter (fun a -> Fifo.clear a) s.search_asked_peers;                 Array.iter (fun a -> Fifo.clear a) s.search_asked_peers;
1428                 Array.iter (fun a -> Fifo.clear a) s.search_ok_peers;                 Array.iter (fun a -> Fifo.clear a) s.search_ok_peers;
# Line 1468  let enable () = Line 1456  let enable () =
1456                check_current_downloads ();                check_current_downloads ();
1457              end              end
1458        );        );
1459          add_timer 30. (fun _ ->
1460              if !!enable_overnet then begin
1461                  check_current_downloads ();
1462                end
1463          );
1464        add_infinite_timer 1800. (fun _ ->        add_infinite_timer 1800. (fun _ ->
1465            if !!enable_overnet then begin            if !!enable_overnet then begin
1466                PublishedKeywords.refresh ();                PublishedKeywords.refresh ();
# Line 1566  let _ = Line 1559  let _ =
1559           let buf = o.conn_buf in           let buf = o.conn_buf in
1560    
1561           Printf.bprintf buf "Peers of known_peers:\n";           Printf.bprintf buf "Peers of known_peers:\n";
1562           KnownPeers.iter (fun p ->           Hashtbl.iter (fun key p ->
1563             Printf.bprintf buf "%s:%d:" (Ip.to_string p.peer_ip) p.peer_port;             let (ip,port) = key in
1564               Printf.bprintf buf "%s:%d:" (Ip.to_string ip) port;
1565             print_peer buf p;             print_peer buf p;
1566           ) known_peers;           ) known_peers;
1567           ""           ""
# Line 1759  let _ = Line 1753  let _ =
1753          with _ ->          with _ ->
1754              lprintf_nl () "Unable to send UDP message"; "Unable to send UDP message"              lprintf_nl () "Unable to send UDP message"; "Unable to send UDP message"
1755      ), ":\t\t\t\tsend UDP message (<ip> <port> <msg in hex>)";      ), ":\t\t\t\tsend UDP message (<ip> <port> <msg in hex>)";
1756        
1757      "buckets", Arg_none (fun o ->      "buckets", Arg_none (fun o ->
1758          let buf = o.conn_buf in          let buf = o.conn_buf in
1759          update_buckets ();          update_buckets ();
# Line 1767  let _ = Line 1761  let _ =
1761            Printf.bprintf buf "Number of used buckets %d with %d peers (prebucket: %d peers)\n"            Printf.bprintf buf "Number of used buckets %d with %d peers (prebucket: %d peers)\n"
1762              !n_used_buckets !connected_peers !pre_connected_peers;              !n_used_buckets !connected_peers !pre_connected_peers;
1763          for i = 0 to !n_used_buckets do          for i = 0 to !n_used_buckets do
1764            Printf.bprintf buf "   bucket[%d] : %d peers (prebucket %d)\n"            if Fifo.length buckets.(i) > 0 ||
1765              i (Fifo.length buckets.(i)) (Fifo.length prebuckets.(i));              Fifo.length prebuckets.(i) > 0 then
1766                  Printf.bprintf buf "   bucket[%d] : %d peers (prebucket %d)\n"
1767                    i (Fifo.length buckets.(i)) (Fifo.length prebuckets.(i));
1768          done;          done;
1769          if o.conn_output = HTML then          if o.conn_output = HTML then
1770            begin            begin
1771              let listtmp = String2.split (Buffer.contents buf) '\n' in              let listtmp = String2.split (Buffer.contents buf) '\n' in
1772              let buckets = ref [] in              let buckets = ref [] in
             let i = ref 0 in  
1773              List.iter (fun s ->              List.iter (fun s ->
1774                  if !i <> !n_used_buckets + 1 then begin                  buckets := !buckets @ [("", "dl-1", s);]
                   buckets := !buckets @ [("", "dl-1", (Printf.sprintf "\\<a href=\\\"submit\\?q=%sdump_bucket\\+%d\\\" target=output\\>[Dump]\\</a\\>%s" command_prefix !i s) );];  
                   end else begin  
                   buckets := !buckets @ [("", "dl-1", (Printf.sprintf "\\<a href=\\\"submit\\?q=%sbuckets\\\" target=output\\>[Refresh]\\</a\\>" command_prefix) );]  
                 end;  
                 incr i;  
1775              ) listtmp;              ) listtmp;
1776              Buffer.reset buf;              Buffer.reset buf;
1777              html_mods_table_one_col buf "ovbucketsTable" "results" ([              html_mods_table_one_col buf "ovbucketsTable" "results" ([
# Line 1839  let forget_search ss = Line 1829  let forget_search ss =
1829      match s.search_kind with      match s.search_kind with
1830        KeywordSearch sss when ss == sss ->        KeywordSearch sss when ss == sss ->
1831          begin          begin
1832              Hashtbl.clear s.search_known_peers;
1833            Array.iter (fun a -> Fifo.clear a) s.search_waiting_peers;            Array.iter (fun a -> Fifo.clear a) s.search_waiting_peers;
1834            Array.iter (fun a -> Fifo.clear a) s.search_asked_peers;            Array.iter (fun a -> Fifo.clear a) s.search_asked_peers;
1835            Array.iter (fun a -> Fifo.clear a) s.search_ok_peers;            Array.iter (fun a -> Fifo.clear a) s.search_ok_peers;
# Line 1873  let cancel_recover_file file = Line 1864  let cancel_recover_file file =
1864      match s.search_kind with      match s.search_kind with
1865        FileSearch f when f == file ->        FileSearch f when f == file ->
1866          begin          begin
1867              Hashtbl.clear s.search_known_peers;
1868            Array.iter (fun a -> Fifo.clear a) s.search_waiting_peers;            Array.iter (fun a -> Fifo.clear a) s.search_waiting_peers;
1869            Array.iter (fun a -> Fifo.clear a) s.search_asked_peers;            Array.iter (fun a -> Fifo.clear a) s.search_asked_peers;
1870            Array.iter (fun a -> Fifo.clear a) s.search_ok_peers;            Array.iter (fun a -> Fifo.clear a) s.search_ok_peers;
# Line 1941  Define a function to be called when the Line 1933  Define a function to be called when the
1933    
1934        Printf.bprintf buf "  Search hits: %d\n" !search_hits;        Printf.bprintf buf "  Search hits: %d\n" !search_hits;
1935        Printf.bprintf buf "  Source hits: %d\n" !source_hits;        Printf.bprintf buf "  Source hits: %d\n" !source_hits;
1936        Printf.bprintf buf "  known_peers stats: %d %d %d %d %d %d\n"        Printf.bprintf buf "  Hashtbl.lenght known_peers: %d\n" (Hashtbl.length known_peers);
         ((fun (n,_,_,_,_,_) -> n)(KnownPeers.stats known_peers))  
         ((fun (_,n,_,_,_,_) -> n)(KnownPeers.stats known_peers))  
         ((fun (_,_,n,_,_,_) -> n)(KnownPeers.stats known_peers))  
         ((fun (_,_,_,n,_,_) -> n)(KnownPeers.stats known_peers))  
         ((fun (_,_,_,_,n,_) -> n)(KnownPeers.stats known_peers))  
         ((fun (_,_,_,_,_,n) -> n)(KnownPeers.stats known_peers))  
       ;  
   
         
1937        Printf.bprintf buf "  to_ping: %d\n" (List.length !to_ping);        Printf.bprintf buf "  to_ping: %d\n" (List.length !to_ping);
1938          let n_search_known_peers = ref 0 in
1939        let n_search_waiting_peers = ref 0 in        let n_search_waiting_peers = ref 0 in
1940        let n_search_asked_peers = ref 0 in        let n_search_asked_peers = ref 0 in
1941        let n_search_ok_peers = ref 0 in        let n_search_ok_peers = ref 0 in
# Line 1959  Define a function to be called when the Line 1943  Define a function to be called when the
1943        let n_overnet_searches = ref 0 in        let n_overnet_searches = ref 0 in
1944        List.iter ( fun s ->        List.iter ( fun s ->
1945                for i = 128 downto 0 do                for i = 128 downto 0 do
1946                   n_search_known_peers :=
1947                     !n_search_known_peers + (Hashtbl.length s.search_known_peers);
1948                 n_search_waiting_peers :=                 n_search_waiting_peers :=
1949                   !n_search_waiting_peers + (Fifo.length s.search_waiting_peers.(i));                   !n_search_waiting_peers + (Fifo.length s.search_waiting_peers.(i));
1950                 n_search_asked_peers :=                 n_search_asked_peers :=
# Line 1970  Define a function to be called when the Line 1956  Define a function to be called when the
1956                done;                done;
1957               incr n_overnet_searches               incr n_overnet_searches
1958        ) !overnet_searches;        ) !overnet_searches;
1959          Printf.bprintf buf "  n_search_known_peers: %d\n" !n_search_known_peers;
1960        Printf.bprintf buf "  n_search_waiting_peers: %d\n" !n_search_waiting_peers;        Printf.bprintf buf "  n_search_waiting_peers: %d\n" !n_search_waiting_peers;
1961        Printf.bprintf buf "  n_search_asked_peers: %d\n" !n_search_asked_peers;        Printf.bprintf buf "  n_search_asked_peers: %d\n" !n_search_asked_peers;
1962        Printf.bprintf buf "  n_search_ok_peers: %d\n" !n_search_ok_peers;        Printf.bprintf buf "  n_search_ok_peers: %d\n" !n_search_ok_peers;

Legend:
Removed from v.1.54  
changed lines
  Added in v.1.55

savannah-hackers-public@gnu.org
ViewVC Help
Powered by ViewVC 1.1.26