/[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.55 by spiralvoice, Thu Oct 27 21:32:53 2005 UTC revision 1.56 by spiralvoice, Mon Oct 31 18:07:11 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_last_recv : int;      mutable peer_expire : int;
54      mutable peer_last_send : int;      mutable peer_last_send : int;
55        peer_created : int;
56    }    }
57    
58  type search_kind =  type search_kind =
# Line 136  search for files and search for sources. Line 137  search for files and search for sources.
137  | OvernetUnknown21 of peer  | OvernetUnknown21 of peer
138    
139  let print_peer buf p =  let print_peer buf p =
140    Printf.bprintf buf "   { md4 = %s ip = %s port = %d %s kind = %d l_send = %d l_recv = %d}\n"    let hours = (last_time () - p.peer_created) / 3600 in
141      let mins = (last_time () - p.peer_created) / 60 in
142      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"
143      (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
144      (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
145      else "") p.peer_kind p.peer_last_send p.peer_last_recv      else "") p.peer_kind p.peer_last_send p.peer_expire p.peer_created
146        hours mins (if p.peer_expire > last_time () && p.peer_last_send <> 0 then "OK " else "")
147        
148    
149  let message_to_string t =  let message_to_string t =
150    let buf = Buffer.create 100 in    let buf = Buffer.create 100 in
# Line 274  type overnet_search = { Line 279  type overnet_search = {
279      search_md4 : Md4.t;      search_md4 : Md4.t;
280      mutable search_kind : search_for;      mutable search_kind : search_for;
281    
     search_known_peers : (Ip.t * int, peer) Hashtbl.t;  
282      search_waiting_peers : peer Fifo.t array;      search_waiting_peers : peer Fifo.t array;
283      search_asked_peers : peer Fifo.t array;      search_asked_peers : peer Fifo.t array;
284      search_ok_peers : peer Fifo.t array;      search_ok_peers : peer Fifo.t array;
285    
286      mutable search_queries : int;      mutable search_queries : int;
287        (* TODO: search_requests is not used ? *)
288      mutable search_requests : int;      mutable search_requests : int;
289      mutable search_start : int;      mutable search_start : int;
290      mutable search_results : (Md4.t, tag list) Hashtbl.t;      mutable search_results : (Md4.t, tag list) Hashtbl.t;
# Line 326  module Make(Proto: sig Line 331  module Make(Proto: sig
331  *********************************************************************)  *********************************************************************)
332    
333  let max_peers_per_bucket = 20  let max_peers_per_bucket = 20
334  let max_peers_per_prebucket = 100  let max_peers_per_prebucket = 40
 let min_peers_per_block = 16 (* was 2 *)  
 let min_peers_before_connect = 5  
 let max_searches_for_publish = 5  
335  let max_search_queries = 64  let max_search_queries = 64
336  let max_search_requests = 20  let max_search_requests = 20
337  let max_boot_peers = 200  let max_boot_peers = 200
# Line 420  let overnet_options_version = Line 422  let overnet_options_version =
422    
423  *********************************************************************)  *********************************************************************)
424    
425    let dummy_peer =
426      {
427        peer_md4 = Md4.null;
428        peer_ip = Ip.null;
429        peer_port = 0;
430        peer_tcpport = 0;
431        peer_kind = 0;
432        peer_expire = 0;
433        peer_last_send = 0;
434        peer_created = last_time ();
435      }
436    
437  let connected_peers = ref 0  let connected_peers = ref 0
438  let pre_connected_peers = ref 0  let pre_connected_peers = ref 0
439    
# Line 529  are peers that we are not sure of. *) Line 543  are peers that we are not sure of. *)
543  let buckets = Array.init 129 (fun _ -> Fifo.create ())  let buckets = Array.init 129 (fun _ -> Fifo.create ())
544  let prebuckets = Array.init 129 (fun _ -> Fifo.create ())  let prebuckets = Array.init 129 (fun _ -> Fifo.create ())
545    
 let known_peers = Hashtbl.create 127  
546  let to_ping = ref []  let to_ping = ref []
547    module KnownPeers = Weak.Make(struct
548          type t = peer
549          let hash c = Hashtbl.hash (c.peer_ip, c.peer_port)
550          let equal x y = x.peer_port = y.peer_port
551              && x.peer_ip = y.peer_ip
552      end)
553    
554    let known_peers = KnownPeers.create 1023
555    
556  (*  (*
557  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 693  module PublishedFiles = Publish(struct Line 714  module PublishedFiles = Publish(struct
714  let debug_client ip = false  let debug_client ip = false
715  (*  Ip.matches ip !!overnet_debug_clients *)  (*  Ip.matches ip !!overnet_debug_clients *)
716    
717    let checking_kind p =
718    (* according to eMule 0.46c *)
719      if (last_time () - p.peer_last_send) < 10 ||
720         p.peer_kind = 4 then () else begin
721        p.peer_kind <- p.peer_kind + 1;
722        p.peer_expire <- last_time () + 120;
723      end;
724      ()
725      
726    let new_peer_message p =
727    (*  lprintf () "*** Updating time for %s:%d\n" (Ip.to_string p.peer_ip) p.peer_port; *)
728      let calc_peer_kind p =
729        (* according to eMule 0.46c *)
730        let hours = (last_time () - p.peer_created) / 3600 in
731        let lt = last_time () in
732        match hours with
733          0 -> (2, lt + 3600)  (* new kind 2, 1h timeout to ping *)
734        | 1 -> (1, lt + 5400)  (* kind 1, 1 1/2h timeout *)
735        | _ -> (0, lt + 7200)  (* kind 0, 2h timeout *)
736      in
737      let (new_kind,new_expire) = calc_peer_kind p in
738      p.peer_kind <- new_kind;
739      p.peer_expire <- new_expire;
740      ()
741    
742  let udp_send_direct ip port msg =  let udp_send_direct ip port msg =
743    match !udp_sock with    match !udp_sock with
744      None -> ()      None -> ()
# Line 711  let udp_send p msg = Line 757  let udp_send p msg =
757    
758  let bootstrap ip port =  let bootstrap ip port =
759    if !!overnet_update_nodes && is_overnet_ip ip &&    if !!overnet_update_nodes && is_overnet_ip ip &&
760       port <> 0 && not (Hashtbl.mem known_peers (ip,port)) then       port <> 0 && not (KnownPeers.mem known_peers { dummy_peer with peer_port = port ; peer_ip = ip } ) then
761         LimitedList.add unknown_peers (ip,port)         LimitedList.add unknown_peers (ip,port)
762    
763  let new_peer p =  let new_peer p =
764    let ip = p.peer_ip in    let ip = p.peer_ip in
765    if ip <> Ip.localhost && is_overnet_ip ip &&    let port = p.peer_port in
766      p.peer_port <> 0 then  
767      let key = (p.peer_ip, p.peer_port) in    if ip <> Ip.localhost && is_overnet_ip ip && port <> 0 then
768    
769      try      try
770        let pp = Hashtbl.find known_peers key in        let pp = KnownPeers.find known_peers p in
771        if pp.peer_last_recv < p.peer_last_recv then        (* TODO: check for changes in ip:port? Or Hash? *)
         pp.peer_last_recv <- p.peer_last_recv;  
772        pp        pp
773      with _ ->      with _ ->
774        
775          let bucket = bucket_number p.peer_md4 in          let bucket = bucket_number p.peer_md4 in
776    
777    (* Add the peer to the table of known peers *)
778            if bucket <> 128 && p.peer_md4 <> Md4.null then
779              KnownPeers.add known_peers p;
780    
781  (* 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
782  restart. *)  restart. *)
783          if bucket <> 128 then          if bucket <> 128 && p.peer_md4 <> Md4.null then
784              bootstrap p.peer_ip p.peer_port;              bootstrap p.peer_ip p.peer_port;
785    
786  (* bucket = 128 is returned for our very own ID, we dont want it in our buckets  (* bucket = 128 is returned for our very own ID, we dont want it in our buckets
787     Then, enter it in the buckets *)     Then, enter it in the buckets *)
788          if bucket < !n_used_buckets && bucket <> 128 then begin          if bucket < !n_used_buckets && bucket <> 128 &&
789               p.peer_md4 <> Md4.null then begin
790    
791              if Fifo.length prebuckets.(bucket) = max_peers_per_prebucket then              if Fifo.length prebuckets.(bucket) = max_peers_per_prebucket then
792              begin              begin
793                let l = last_time () - 1800 in                let l = last_time () - 1800 in
794                let pp = Fifo.take prebuckets.(bucket) in                let pp = Fifo.take prebuckets.(bucket) in
795                Fifo.put prebuckets.(bucket)                Fifo.put prebuckets.(bucket)
796  (* If we heard of the head of the bucket in the last 30 minutes, we should  (* If the head of the bucket is not dead we should keep it *)
797    keep it. *)                  (if pp.peer_kind < 4 || pp.peer_expire > last_time () then pp else p);
                 (if pp.peer_last_recv > l then pp else p);  
               if pp.peer_last_recv <= l then  
               begin  
 (* Add the peer to the table of known peers *)  
                 Hashtbl.add known_peers key p;  
 (* And remove pp *)  
                 let ppkey = (pp.peer_ip, pp.peer_port) in  
                 Hashtbl.remove known_peers ppkey  
               end  
798              end                end  
799              else begin              else begin
                Hashtbl.add known_peers key p;  
800                 Fifo.put prebuckets.(bucket) p;                 Fifo.put prebuckets.(bucket) p;
801                 incr pre_connected_peers                 incr pre_connected_peers
802              end              end
803          end          end
804          else if !n_used_buckets < 128 && bucket <> 128 then begin          else if !n_used_buckets < 128 && bucket <> 128 &&
805                    p.peer_md4 <> Md4.null then begin
806              Fifo.put prebuckets.(!n_used_buckets) p;              Fifo.put prebuckets.(!n_used_buckets) p;
807              incr pre_connected_peers;              incr pre_connected_peers;
             Hashtbl.add known_peers key p;  
808    
809              while !n_used_buckets < 128 &&              while !n_used_buckets < 128 &&
810                Fifo.length prebuckets.(!n_used_buckets)                Fifo.length prebuckets.(!n_used_buckets)
# Line 770  restart. *) Line 812  restart. *)
812                let b = prebuckets.(!n_used_buckets) in                let b = prebuckets.(!n_used_buckets) in
813                incr n_used_buckets;                incr n_used_buckets;
814                for i = 1 to Fifo.length b do                for i = 1 to Fifo.length b do
815                  let p = Fifo.take b in                  let pp = Fifo.take b in
816                  let bucket = bucket_number p.peer_md4 in                  let bucket = bucket_number p.peer_md4 in
817                  Fifo.put (if bucket >= !n_used_buckets then                  Fifo.put (if bucket >= !n_used_buckets then
818                      prebuckets.(!n_used_buckets) else b) p                      prebuckets.(!n_used_buckets) else b) pp
819                done                done
820              done              done
821          end;          end;
# Line 792  let get_closest_peers md4 nb = Line 834  let get_closest_peers md4 nb =
834        if n > 0 then        if n > 0 then
835          let p = Fifo.take fifo in          let p = Fifo.take fifo in
836          Fifo.put fifo p;          Fifo.put fifo p;
837          if p.peer_last_recv <> 0 then begin          (* kind < 4 so we do not send too much requests and avoid dead contacts not
838               yet removed because of timeouts *)
839            (* TODO: Keep order? Then we need a in_use flag? *)
840            if p.peer_kind < 4 && p.peer_expire > last_time () &&
841               p.peer_last_send <> 0 then begin
842              if !verbose_overnet then begin              if !verbose_overnet then begin
843              lprintf_nl () "Adding good search peer %s:%d"              lprintf_nl () "Adding good search peer %s:%d"
844                (Ip.to_string p.peer_ip) p.peer_port;                (Ip.to_string p.peer_ip) p.peer_port;
# Line 804  let get_closest_peers md4 nb = Line 850  let get_closest_peers md4 nb =
850      in      in
851      iter (min !nb (Fifo.length fifo))      iter (min !nb (Fifo.length fifo))
852    in    in
853    add_list  bucket;    add_list bucket;
854    
855    if !nb > 0 then begin    if !nb > 0 then begin
856    
# Line 834  let my_peer () = Line 880  let my_peer () =
880      peer_port = !!overnet_port;      peer_port = !!overnet_port;
881      peer_kind = 0;      peer_kind = 0;
882      peer_tcpport = !!overnet_tcpport;      peer_tcpport = !!overnet_tcpport;
     peer_last_recv = 0;  
883      peer_last_send = 0;      peer_last_send = 0;
884        peer_expire = 0;
885        peer_created = 0;
886    }    }
887    
888  let get_any_peers  nb =  let get_any_peers  nb =
# Line 846  let get_any_peers  nb = Line 893  let get_any_peers  nb =
893      let fifo = buckets.(bucket) in      let fifo = buckets.(bucket) in
894      let rec iter n =      let rec iter n =
895        if n > 0 then        if n > 0 then
896          (* TODO: keep order *)
897          let p = Fifo.take fifo in          let p = Fifo.take fifo in
898          Fifo.put fifo p;          Fifo.put fifo p;
899          decr nb;          decr nb;
# Line 864  let get_any_peers  nb = Line 912  let get_any_peers  nb =
912    !list    !list
913    
914  let add_search_peer s p =  let add_search_peer s p =
915    if p.peer_ip <> client_ip None then    if p.peer_ip <> Ip.localhost && is_overnet_ip p.peer_ip &&
916    let key = (p.peer_ip, p.peer_port) in       p.peer_port <> 0 then begin
917    if not (Hashtbl.mem s.search_known_peers key) then begin      let nbits = common_bits p.peer_md4 s.search_md4 in
918        Hashtbl.add s.search_known_peers key p;      begin
919        let nbits = common_bits p.peer_md4 s.search_md4 in        try
920        Fifo.put s.search_waiting_peers.(nbits) p;          Fifo.iter (fun pp ->
921              if pp.peer_ip = p.peer_ip &&
922                 pp.peer_port = p.peer_port then
923                raise Exit
924            ) s.search_waiting_peers.(nbits);
925            Fifo.put s.search_waiting_peers.(nbits) p;
926          with Exit -> ()
927      end      end
928      end
929    
930  let create_search kind md4 =  let create_search kind md4 =
931    if !verbose_overnet then lprintf_nl () "create_search";    if !verbose_overnet then lprintf_nl () "create_search";
932    let s = {    let s = {
933        search_md4 = md4;        search_md4 = md4;
934        search_kind = kind;        search_kind = kind;
       search_known_peers = Hashtbl.create 127;  
935        search_queries = 0;        search_queries = 0;
936        search_requests = 0;        search_requests = 0;
937        search_waiting_peers = Array.init 129 (fun _ -> Fifo.create ());        search_waiting_peers = Array.init 129 (fun _ -> Fifo.create ());
# Line 909  let port_of_udp_packet p = Line 963  let port_of_udp_packet p =
963      Unix.ADDR_INET (inet, port) -> port      Unix.ADDR_INET (inet, port) -> port
964    | _ -> assert false    | _ -> assert false
965    
   
 let new_peer_message p =  
 (*  lprintf () "*** Updating time for %s:%d\n" (Ip.to_string p.peer_ip) p.peer_port; *)  
   p.peer_last_recv <- last_time ()  
   
966  let udp_client_handler t p =  let udp_client_handler t p =
967    let other_ip = ip_of_udp_packet p in    let other_ip = ip_of_udp_packet p in
968    let other_port = port_of_udp_packet p in    let other_port = port_of_udp_packet p in
# Line 921  let udp_client_handler t p = Line 970  let udp_client_handler t p =
970      lprintf_nl () "UDP FROM %s:%d type %s"      lprintf_nl () "UDP FROM %s:%d type %s"
971        (Ip.to_string other_ip) other_port        (Ip.to_string other_ip) other_port
972        (message_to_string t);        (message_to_string t);
973      (* Emule uses other_ip:other_port, so do we *)
974      (* Update known_peers by other_port:other_ip *)
975      let sender = new_peer { dummy_peer with peer_port = other_port ; peer_ip = other_ip } in
976      new_peer_message sender;
977    match t with    match t with
978    
979    | OvernetConnect p ->    | OvernetConnect p ->
980        let rec send p =        if is_overnet_ip sender.peer_ip && sender.peer_port <> 0 then
981          new_peer_message p;          let sender = new_peer { p with peer_port = other_port ; peer_ip = other_ip } in
982          let p = new_peer p in          new_peer_message sender;
983            udp_send p (OvernetConnectReply (get_any_peers 20))          udp_send sender (OvernetConnectReply (get_any_peers 20))
984        in         else
985          if is_overnet_ip p.peer_ip && p.peer_port <> 0 then           begin
986            send p             if !verbose_overnet then
987          else               lprintf_nl () "Connect: invalid IP %s:%d received from %s:%d"
           if is_overnet_ip other_ip && other_port <> 0 then  
             begin  
               if !verbose_overnet then  
                 lprintf_nl () "Connect: convert address %s:%d to %s:%d"  
                   (Ip.to_string p.peer_ip) p.peer_port (Ip.to_string other_ip) other_port;  
               p.peer_ip <- other_ip;  
               p.peer_port <- other_port;  
               send p  
             end  
           else  
             begin  
               if !verbose_overnet then  
                 lprintf_nl () "Connect: invalid IP %s:%d received from %s:%d"  
988                    (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;
989                failwith "Message not understood"               failwith "Message not understood"
990              end           end
991    
992    | OvernetConnectReply ps ->    | OvernetConnectReply ps ->
993        UdpSocket.declare_pong other_ip;        UdpSocket.declare_pong other_ip;
# Line 955  let udp_client_handler t p = Line 995  let udp_client_handler t p =
995          match list with          match list with
996            [] -> ()            [] -> ()
997          | [p] ->          | [p] ->
998              new_peer_message p;              let sender = new_peer { p with peer_port = other_port ; peer_ip = other_ip } in
999              if other_port <> p.peer_port || other_ip <> p.peer_ip then              new_peer_message sender
               if !verbose_overnet then  
               lprintf_nl () "Bad IP or port";  
             let p = new_peer p in  
             ()  
1000          | p :: tail ->          | p :: tail ->
1001              let p = new_peer p in              let p = new_peer p in
1002              iter tail              iter tail
# Line 968  let udp_client_handler t p = Line 1004  let udp_client_handler t p =
1004        iter ps;        iter ps;
1005    
1006    | OvernetPublicize p ->    | OvernetPublicize p ->
1007        let rec send p =        let sender = new_peer { p with peer_port = other_port ; peer_ip = other_ip } in
1008          new_peer_message p;        new_peer_message sender;
1009          let p = new_peer p in        if is_overnet_ip sender.peer_ip && sender.peer_port <> 0 then
1010            udp_send p (OvernetPublicized (Some (my_peer ())))          udp_send sender (OvernetPublicized (Some (my_peer ())))
1011        in         else begin
         if is_overnet_ip p.peer_ip && p.peer_port <> 0 then  
           send p  
         else  
           if is_overnet_ip other_ip && other_port <> 0 then  
             begin  
               if !verbose_overnet then  
                 lprintf_nl () "Publicize: convert address %s:%d to %s:%d"  
                   (Ip.to_string p.peer_ip) p.peer_port (Ip.to_string other_ip) other_port;  
               p.peer_ip <- other_ip;  
               p.peer_port <- other_port;  
               send p  
             end  
           else  
             begin  
1012                if !verbose_overnet then                if !verbose_overnet then
1013                  lprintf_nl () "Publicize: invalid IP %s:%d received from %s:%d"                  lprintf_nl () "Publicize: invalid IP %s:%d received from %s:%d"
1014                    (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 997  let udp_client_handler t p = Line 1019  let udp_client_handler t p =
1019        ()        ()
1020    
1021    | OvernetPublicized (Some p) ->    | OvernetPublicized (Some p) ->
       new_peer_message p;  
       let p = new_peer p in  
1022        ()        ()
1023    
1024    | OvernetSearch (nresults, md4, from_who) ->    | OvernetSearch (nresults, md4, from_who) ->
   
1025        let peers = get_closest_peers md4 nresults in        let peers = get_closest_peers md4 nresults in
1026        udp_send_direct other_ip other_port (OvernetSearchReply (md4,peers))        udp_send sender (OvernetSearchReply (md4,peers))
1027    
1028    | OvernetSearchReply (md4, peers) ->    | OvernetSearchReply (md4, peers) ->
1029    
# Line 1012  let udp_client_handler t p = Line 1031  let udp_client_handler t p =
1031        List.iter (fun s ->        List.iter (fun s ->
1032            if s.search_md4 = md4 then begin            if s.search_md4 = md4 then begin
1033                List.iter (add_search_peer s) peers;                List.iter (add_search_peer s) peers;
1034                try                  let nbits = common_bits sender.peer_md4 s.search_md4 in
1035                  let p = Hashtbl.find s.search_known_peers                  Fifo.put s.search_ok_peers.(nbits) sender;
                     (other_ip, other_port) in  
                 new_peer_message p;  
                 let nbits = common_bits p.peer_md4 s.search_md4 in  
                 Fifo.put s.search_ok_peers.(nbits) p;  
               with _ -> ()  
1036              end              end
1037        ) !overnet_searches;        ) !overnet_searches;
1038    
# Line 1034  let udp_client_handler t p = Line 1048  let udp_client_handler t p =
1048            if s.search_md4 = md4 then begin            if s.search_md4 = md4 then begin
1049                s.search_nresults <- s.search_nresults + 1;                s.search_nresults <- s.search_nresults + 1;
1050    
               begin  
                 try  
                   let p = Hashtbl.find s.search_known_peers  
                       (other_ip, other_port) in  
                   new_peer_message p;  
                 with _ -> ()  
               end;  
   
1051                match s.search_kind with                match s.search_kind with
1052                  FileSearch file -> ()                  FileSearch file -> ()
1053    
# Line 1071  let udp_client_handler t p = Line 1077  let udp_client_handler t p =
1077            if s.search_md4 = md4 then begin            if s.search_md4 = md4 then begin
1078                s.search_nresults <- s.search_nresults + 1;                s.search_nresults <- s.search_nresults + 1;
1079    
               begin  
                 try  
                   let p = Hashtbl.find s.search_known_peers  
                       (other_ip, other_port) in  
                   new_peer_message p;  
                 with _ -> ()  
               end;  
   
1080                match s.search_kind with                match s.search_kind with
1081                  FileSearch file ->                  FileSearch file ->
1082                    List.iter (fun p ->                    List.iter (fun p ->
# Line 1118  let udp_client_handler t p = Line 1116  let udp_client_handler t p =
1116              let _, list = List2.cut min list in              let _, list = List2.cut min list in
1117              let list, _ = List2.cut (max - min) list in              let list, _ = List2.cut (max - min) list in
1118              if list <> [] then              if list <> [] then
1119                udp_send_direct other_ip other_port                udp_send sender
1120                  (OvernetSearchSourcesResults (md4, list))                  (OvernetSearchSourcesResults (md4, list))
1121          | Search_for_keyword _ ->          | Search_for_keyword _ ->
1122              let list = PublishedKeywords.get md4 in              let list = PublishedKeywords.get md4 in
1123              let _, list = List2.cut min list in              let _, list = List2.cut min list in
1124              let list, _ = List2.cut (max - min) list in              let list, _ = List2.cut (max - min) list in
1125              if list <> [] then              if list <> [] then
1126                udp_send_direct other_ip other_port                udp_send sender
1127                  (OvernetSearchFilesResults (md4, list))                  (OvernetSearchFilesResults (md4, list))
1128          | _ -> ()          | _ -> ()
1129        end        end
# Line 1143  let udp_client_handler t p = Line 1141  let udp_client_handler t p =
1141         if !verbose_overnet && debug_client other_ip then           if !verbose_overnet && debug_client other_ip then  
1142           lprintf_nl () "GET MY IP (port=%d)\n" other_port;             lprintf_nl () "GET MY IP (port=%d)\n" other_port;  
1143   (* FIXME : should be able to flush the UDP buffer*)     (* FIXME : should be able to flush the UDP buffer*)  
1144         udp_send_direct other_ip other_port (OvernetGetMyIPResult other_ip);           udp_send sender (OvernetGetMyIPResult other_ip);  
1145         udp_send_direct other_ip other_port OvernetGetMyIPDone           udp_send sender OvernetGetMyIPDone  
1146            
1147     | OvernetGetMyIPResult(ip) ->       | OvernetGetMyIPResult(ip) ->  
1148         if !verbose_overnet && debug_client other_ip then           if !verbose_overnet && debug_client other_ip then  
# Line 1160  let udp_client_handler t p = Line 1158  let udp_client_handler t p =
1158            lprintf_nl () "Peer NOT FOUND %s (%s:%d) kind: %d (msg 33)"            lprintf_nl () "Peer NOT FOUND %s (%s:%d) kind: %d (msg 33)"
1159              (Md4.to_string peer.peer_md4) (Ip.to_string peer.peer_ip)              (Md4.to_string peer.peer_md4) (Ip.to_string peer.peer_ip)
1160          peer.peer_port peer.peer_kind;          peer.peer_port peer.peer_kind;
1161          let key = (peer.peer_ip, peer.peer_port) in          let dp = { dummy_peer with peer_port = peer.peer_port ; peer_ip = peer.peer_ip } in
1162          if Hashtbl.mem known_peers key          if KnownPeers.mem known_peers dp
1163          then begin          then begin
1164  (* remove it from the prebuckets and known_peers only *)  (* remove it from the prebuckets and known_peers only *)
1165            try            try
# Line 1172  let udp_client_handler t p = Line 1170  let udp_client_handler t p =
1170                  if p.peer_ip = peer.peer_ip &&                  if p.peer_ip = peer.peer_ip &&
1171                      p.peer_port = peer.peer_port then begin                      p.peer_port = peer.peer_port then begin
1172                    decr pre_connected_peers;                    decr pre_connected_peers;
                   Hashtbl.remove known_peers key;  
1173                  end else Fifo.put b p                  end else Fifo.put b p
1174                done;                done;
1175              done;              done;
# Line 1200  let query_next_peers () = Line 1197  let query_next_peers () =
1197          if  nbits >= 0 then          if  nbits >= 0 then
1198            let len = Fifo.length s.search_waiting_peers.(nbits) in            let len = Fifo.length s.search_waiting_peers.(nbits) in
1199            if len > 0 then            if len > 0 then
1200              let p = Fifo.take  s.search_waiting_peers.(nbits) in              let p = Fifo.take s.search_waiting_peers.(nbits) in
1201    
1202                checking_kind p;
1203              udp_send p (OvernetSearch (nresults, s.search_md4, Some p.peer_md4));              udp_send p (OvernetSearch (nresults, s.search_md4, Some p.peer_md4));
1204              s.search_queries <- s.search_queries + 1;              s.search_queries <- s.search_queries + 1;
1205              Fifo.put  s.search_asked_peers.(nbits) p;              Fifo.put s.search_asked_peers.(nbits) p;
1206    
1207              (if todo > 1 then iter nbits (todo - 1))              (if todo > 1 then iter nbits (todo - 1))
1208            else            else
# Line 1229  let check_current_downloads () = Line 1227  let check_current_downloads () =
1227    
1228  let update_buckets () =  let update_buckets () =
1229    
1230  (* 1. Clean the buckets from too old peers ( last contact > 1 hour ) *)  (* 1. Clean the buckets from too old peers ( by kind and expire ) *)
   
   let overtime = last_time () - 3600 in  
1231    
1232    for i = 0 to !n_used_buckets do    for i = 0 to !n_used_buckets do
1233    
1234      let b = buckets.(i) in      let b = buckets.(i) in
1235      for j = 1 to Fifo.length b do      for j = 1 to Fifo.length b do
1236        let p = Fifo.take b in        let p = Fifo.take b in
1237        if p.peer_last_recv > overtime then Fifo.put b p else        (* bad peers have kind = 4 and did not respond within peer_expire *)
1238          if not (p.peer_kind = 4 && p.peer_expire <= last_time ()) then
1239            Fifo.put b p
1240          else
1241        begin        begin
1242          (* remove the peer also from known_peers *)          decr connected_peers;
1243          let key = (p.peer_ip, p.peer_port) in          if !verbose_overnet then lprintf_nl () "update_bucket1: removing %s:%d" (Ip.to_string p.peer_ip) p.peer_port;
1244          Hashtbl.remove known_peers key;        end;
         decr connected_peers  
       end  
1245      done      done
1246    
1247    done;    done;
1248    
1249  (* 2. Complete buckets with new peers from the prebuckets with  (* 2. Complete buckets with new peers from the prebuckets with
1250  ( last_contact < 1 hour ) *)  ( p.peer_kind < 3 ) *)
1251    
1252    for i = 0 to !n_used_buckets - 1 do    for i = 0 to !n_used_buckets do
1253      let b = buckets.(i) in      let b = buckets.(i) in
1254      if Fifo.length b < max_peers_per_bucket then begin      if Fifo.length b < max_peers_per_bucket then begin
1255    
         try  
1256            let pb = prebuckets.(i) in            let pb = prebuckets.(i) in
1257            for j = 1 to Fifo.length pb do            for j = 1 to Fifo.length pb do
1258    
1259                (* to keep fifo in order travel all peers *)
1260              let p = Fifo.take pb in              let p = Fifo.take pb in
1261              if p.peer_last_recv > overtime &&              (* good peers goto buckets *)
1262                 p.peer_last_recv >= p.peer_last_send then begin              if p.peer_kind < 3 && Fifo.length b < max_peers_per_bucket then begin
1263                  Fifo.put b p;                  Fifo.put b p;
1264                  incr connected_peers;                  incr connected_peers;
1265                  decr pre_connected_peers;                  decr pre_connected_peers;
1266                  if Fifo.length b = max_peers_per_bucket then raise Exit              (* bad peers are removed *)    
1267                end else Fifo.put pb p              end else if p.peer_kind = 4 && p.peer_expire <= last_time () then begin
1268                  decr pre_connected_peers;
1269                  if !verbose_overnet then lprintf_nl () "update_bucket2: removing %s:%d" (Ip.to_string p.peer_ip) p.peer_port;
1270                end else
1271                (* the rest returns in prebuckets *)
1272                Fifo.put pb p
1273            done            done
         with Exit -> ()  
   
1274        end        end
1275    done;    done;
1276    
# Line 1281  from the active buckets and prebuckets * Line 1280  from the active buckets and prebuckets *
1280    let n = ref 0 in    let n = ref 0 in
1281    try    try
1282      let addtol p =      let addtol p =
1283        if p.peer_last_recv <> 0 then begin        if p.peer_kind < 3 then begin
1284          LimitedList.add !!boot_peers (p.peer_ip, p.peer_port);          LimitedList.add !!boot_peers (p.peer_ip, p.peer_port);
1285          incr n;          incr n;
1286          if !n = max_boot_peers then raise Exit          if !n = max_boot_peers then raise Exit
# Line 1299  from the active buckets and prebuckets * Line 1298  from the active buckets and prebuckets *
1298    
1299    ()    ()
1300    
1301    let compute_to_ping () =
1302    (* compute which peers to ping in the next minute *)
1303              to_ping := [];
1304              let n_to_ping = ref 0 in
1305    
1306              let ping_peers b =
1307                Fifo.iter (fun p ->
1308                    if (p.peer_expire <= last_time () && p.peer_kind < 4) ||
1309                        p.peer_last_send = 0 then begin
1310                        to_ping := p :: !to_ping;
1311                        incr n_to_ping;
1312                        if !n_to_ping = 60 then raise Exit
1313                      end
1314                ) b
1315              in
1316              begin
1317                try
1318                  for i = !n_used_buckets downto 8 do
1319                    ping_peers buckets.(i);
1320                    ping_peers prebuckets.(i);
1321                  done;
1322                  for i = min !n_used_buckets 7 downto 0 do
1323                    ping_peers buckets.(i);
1324                  done;
1325                  for i = min !n_used_buckets 7 downto 0 do
1326                    ping_peers prebuckets.(i);
1327                  done;
1328                with Exit -> ()
1329              end;
1330              ()
1331    
1332    
1333  let enable () =  let enable () =
1334    if !!enable_overnet && not !is_enabled then begin    if !!enable_overnet && not !is_enabled then begin
1335        let enabler = is_enabled in        let enabler = is_enabled in
# Line 1319  let enable () = Line 1350  let enable () =
1350              let my_peer = my_peer () in              let my_peer = my_peer () in
1351    
1352  (* ping old peers regularly *)  (* ping old peers regularly *)
1353              begin              let process_to_ping () =
1354                match !to_ping with                begin
1355                  [] -> ()                  match !to_ping with
1356                | p :: tail ->                    [] -> compute_to_ping ();
1357                    | p :: tail ->
1358                      (* do not hammer a peer, we could have send already a search reqeust
1359                         since to_ping is rebuild at least every 60 seconds *)
1360                      if (last_time () - p.peer_last_send) > 60 then begin
1361                        checking_kind p;
1362                        udp_send p (OvernetPublicize my_peer);
1363                      end;
1364                    to_ping := tail;                    to_ping := tail;
1365                    p.peer_last_send <- last_time ();                end;
1366                    udp_send p (OvernetPublicize my_peer);              in
1367              end;              process_to_ping ();
1368    (* Send OvernetConnects and ping more peers *)
1369  (* ping unknown peers *)  (* TODO: How does eMule it? are 50 ok? *)
1370              begin              begin
1371                try                try
1372                    if !connected_peers < 100 then                   if !connected_peers < 50 then
1373                      for i = 1 to 5 do                     for i = 1 to 3 do
1374                         process_to_ping ();
1375                       done;
1376                     (* Do not send too much OvernetConnect, there is no use *)
1377                     if (!connected_peers + !pre_connected_peers) < 30 then
1378                        for i = 1 to 2 do
1379                        let (ip, port) = LimitedList.get unknown_peers in                        let (ip, port) = LimitedList.get unknown_peers in
1380                        udp_send_ping ip port (OvernetConnect my_peer);                        udp_send_ping ip port (OvernetConnect my_peer);
1381                      done                      done
# Line 1359  let enable () = Line 1402  let enable () =
1402                    in                    in
1403  (* 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
1404    search to start asking for results... *)    search to start asking for results... *)
1405                    let min_bucket = mini !n_used_buckets 5 in  (* TODO: what the heck we use !n_used_buckets? we ask every search_ok_peers *)
1406                    for i = 1 to nrequests do                    for i = 1 to nrequests do
1407                      try                      try
1408                        for j = 128 downto min_bucket do                        for j = 128 downto 0 do
1409                          if Fifo.length s.search_ok_peers.(j) > 0 then                          if Fifo.length s.search_ok_peers.(j) > 0 then
1410                            let p = Fifo.take s.search_ok_peers.(j) in                            let p = Fifo.take s.search_ok_peers.(j) in
1411                            udp_send p (                            udp_send p (
# Line 1386  let enable () = Line 1429  let enable () =
1429    
1430            update_buckets ();            update_buckets ();
1431    
1432  (* compute which peers to ping in the next minute *)            compute_to_ping ();
           to_ping := [];  
           let n_to_ping = ref 0 in  
   
           let ping_peers b =  
             let overtime = last_time () - 1800 in  
             Fifo.iter (fun p ->  
                 if p.peer_last_recv < overtime &&  
                   p.peer_last_send < overtime 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;  
1433    
1434  (* 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 *)
1435            let l = last_time () - 300 in            let l = last_time () - 300 in
1436            List.iter ( fun s ->            List.iter ( fun s ->
1437               if s.search_requests < max_search_requests &&               if s.search_requests < max_search_requests &&
1438               s.search_start > l then begin               s.search_start > l then begin
                Hashtbl.clear s.search_known_peers;  
1439                 Array.iter (fun a -> Fifo.clear a) s.search_waiting_peers;                 Array.iter (fun a -> Fifo.clear a) s.search_waiting_peers;
1440                 Array.iter (fun a -> Fifo.clear a) s.search_asked_peers;                 Array.iter (fun a -> Fifo.clear a) s.search_asked_peers;
1441                 Array.iter (fun a -> Fifo.clear a) s.search_ok_peers;                 Array.iter (fun a -> Fifo.clear a) s.search_ok_peers;
# Line 1436  let enable () = Line 1449  let enable () =
1449            ) !overnet_searches;            ) !overnet_searches;
1450        );        );
1451    
 (* Every hour, try a query on our UID to fill our buckets *)  
       add_session_timer enabler 3600. (fun _ ->  
           let s = create_search FillBuckets !!overnet_md4 in  
           ()  
       );  
   
       let s = create_search FillBuckets !!overnet_md4 in  
   
1452        add_session_option_timer enabler overnet_query_peer_period  (fun _ ->        add_session_option_timer enabler overnet_query_peer_period  (fun _ ->
1453            if !!enable_overnet then begin            if !!enable_overnet then begin
1454                query_next_peers ()                query_next_peers ()
# Line 1453  let enable () = Line 1458  let enable () =
1458  (* every 15min for light operations *)  (* every 15min for light operations *)
1459        add_session_timer enabler 900. (fun _ ->        add_session_timer enabler 900. (fun _ ->
1460            if !!enable_overnet then begin            if !!enable_overnet then begin
1461                  let s = create_search FillBuckets !!overnet_md4 in
1462                check_current_downloads ();                check_current_downloads ();
1463              end              end
1464        );        );
1465        add_timer 30. (fun _ ->  
1466            if !!enable_overnet then begin        begin
1467                check_current_downloads ();          let s = create_search FillBuckets !!overnet_md4 in
1468              end          check_current_downloads ();
1469        );        end;
1470          
1471        add_infinite_timer 1800. (fun _ ->        add_infinite_timer 1800. (fun _ ->
1472            if !!enable_overnet then begin            if !!enable_overnet then begin
1473                PublishedKeywords.refresh ();                PublishedKeywords.refresh ();
# Line 1559  let _ = Line 1566  let _ =
1566           let buf = o.conn_buf in           let buf = o.conn_buf in
1567    
1568           Printf.bprintf buf "Peers of known_peers:\n";           Printf.bprintf buf "Peers of known_peers:\n";
1569           Hashtbl.iter (fun key p ->           KnownPeers.iter (fun p ->
1570             let (ip,port) = key in             Printf.bprintf buf "%s:%d:" (Ip.to_string p.peer_ip) p.peer_port;
            Printf.bprintf buf "%s:%d:" (Ip.to_string ip) port;  
1571             print_peer buf p;             print_peer buf p;
1572           ) known_peers;           ) known_peers;
1573           ""           ""
# Line 1753  let _ = Line 1759  let _ =
1759          with _ ->          with _ ->
1760              lprintf_nl () "Unable to send UDP message"; "Unable to send UDP message"              lprintf_nl () "Unable to send UDP message"; "Unable to send UDP message"
1761      ), ":\t\t\t\tsend UDP message (<ip> <port> <msg in hex>)";      ), ":\t\t\t\tsend UDP message (<ip> <port> <msg in hex>)";
1762        
1763      "buckets", Arg_none (fun o ->      "buckets", Arg_none (fun o ->
1764          let buf = o.conn_buf in          let buf = o.conn_buf in
1765          update_buckets ();          update_buckets ();
# Line 1761  let _ = Line 1767  let _ =
1767            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"
1768              !n_used_buckets !connected_peers !pre_connected_peers;              !n_used_buckets !connected_peers !pre_connected_peers;
1769          for i = 0 to !n_used_buckets do          for i = 0 to !n_used_buckets do
1770            if Fifo.length buckets.(i) > 0 ||            Printf.bprintf buf "   bucket[%d] : %d peers (prebucket %d)\n"
1771              Fifo.length prebuckets.(i) > 0 then              i (Fifo.length buckets.(i)) (Fifo.length prebuckets.(i));
               Printf.bprintf buf "   bucket[%d] : %d peers (prebucket %d)\n"  
                 i (Fifo.length buckets.(i)) (Fifo.length prebuckets.(i));  
1772          done;          done;
1773          if o.conn_output = HTML then          if o.conn_output = HTML then
1774            begin            begin
1775              let listtmp = String2.split (Buffer.contents buf) '\n' in              let listtmp = String2.split (Buffer.contents buf) '\n' in
1776              let buckets = ref [] in              let buckets = ref [] in
1777                let i = ref 0 in
1778              List.iter (fun s ->              List.iter (fun s ->
1779                  buckets := !buckets @ [("", "dl-1", s);]                  if !i <> !n_used_buckets + 1 then begin
1780                      buckets := !buckets @ [("", "dl-1", (Printf.sprintf "\\<a href=\\\"submit\\?q=%sdump_bucket\\+%d\\\" target=output\\>[Dump]\\</a\\>%s" command_prefix !i s) );];
1781                      end else begin
1782                      buckets := !buckets @ [("", "dl-1", (Printf.sprintf "\\<a href=\\\"submit\\?q=%sbuckets\\\" target=output\\>[Refresh]\\</a\\>" command_prefix) );]
1783                    end;
1784                    incr i;
1785              ) listtmp;              ) listtmp;
1786              Buffer.reset buf;              Buffer.reset buf;
1787              html_mods_table_one_col buf "ovbucketsTable" "results" ([              html_mods_table_one_col buf "ovbucketsTable" "results" ([
# Line 1829  let forget_search ss = Line 1839  let forget_search ss =
1839      match s.search_kind with      match s.search_kind with
1840        KeywordSearch sss when ss == sss ->        KeywordSearch sss when ss == sss ->
1841          begin          begin
           Hashtbl.clear s.search_known_peers;  
1842            Array.iter (fun a -> Fifo.clear a) s.search_waiting_peers;            Array.iter (fun a -> Fifo.clear a) s.search_waiting_peers;
1843            Array.iter (fun a -> Fifo.clear a) s.search_asked_peers;            Array.iter (fun a -> Fifo.clear a) s.search_asked_peers;
1844            Array.iter (fun a -> Fifo.clear a) s.search_ok_peers;            Array.iter (fun a -> Fifo.clear a) s.search_ok_peers;
# Line 1864  let cancel_recover_file file = Line 1873  let cancel_recover_file file =
1873      match s.search_kind with      match s.search_kind with
1874        FileSearch f when f == file ->        FileSearch f when f == file ->
1875          begin          begin
           Hashtbl.clear s.search_known_peers;  
1876            Array.iter (fun a -> Fifo.clear a) s.search_waiting_peers;            Array.iter (fun a -> Fifo.clear a) s.search_waiting_peers;
1877            Array.iter (fun a -> Fifo.clear a) s.search_asked_peers;            Array.iter (fun a -> Fifo.clear a) s.search_asked_peers;
1878            Array.iter (fun a -> Fifo.clear a) s.search_ok_peers;            Array.iter (fun a -> Fifo.clear a) s.search_ok_peers;
# Line 1933  Define a function to be called when the Line 1941  Define a function to be called when the
1941    
1942        Printf.bprintf buf "  Search hits: %d\n" !search_hits;        Printf.bprintf buf "  Search hits: %d\n" !search_hits;
1943        Printf.bprintf buf "  Source hits: %d\n" !source_hits;        Printf.bprintf buf "  Source hits: %d\n" !source_hits;
1944        Printf.bprintf buf "  Hashtbl.lenght known_peers: %d\n" (Hashtbl.length known_peers);        Printf.bprintf buf "  known_peers stats: %d %d %d %d %d %d\n"
1945            ((fun (n,_,_,_,_,_) -> n)(KnownPeers.stats known_peers))
1946            ((fun (_,n,_,_,_,_) -> n)(KnownPeers.stats known_peers))
1947            ((fun (_,_,n,_,_,_) -> n)(KnownPeers.stats known_peers))
1948            ((fun (_,_,_,n,_,_) -> n)(KnownPeers.stats known_peers))
1949            ((fun (_,_,_,_,n,_) -> n)(KnownPeers.stats known_peers))
1950            ((fun (_,_,_,_,_,n) -> n)(KnownPeers.stats known_peers))
1951          ;
1952    
1953          
1954        Printf.bprintf buf "  to_ping: %d\n" (List.length !to_ping);        Printf.bprintf buf "  to_ping: %d\n" (List.length !to_ping);
       let n_search_known_peers = ref 0 in  
1955        let n_search_waiting_peers = ref 0 in        let n_search_waiting_peers = ref 0 in
1956        let n_search_asked_peers = ref 0 in        let n_search_asked_peers = ref 0 in
1957        let n_search_ok_peers = ref 0 in        let n_search_ok_peers = ref 0 in
# Line 1943  Define a function to be called when the Line 1959  Define a function to be called when the
1959        let n_overnet_searches = ref 0 in        let n_overnet_searches = ref 0 in
1960        List.iter ( fun s ->        List.iter ( fun s ->
1961                for i = 128 downto 0 do                for i = 128 downto 0 do
                n_search_known_peers :=  
                  !n_search_known_peers + (Hashtbl.length s.search_known_peers);  
1962                 n_search_waiting_peers :=                 n_search_waiting_peers :=
1963                   !n_search_waiting_peers + (Fifo.length s.search_waiting_peers.(i));                   !n_search_waiting_peers + (Fifo.length s.search_waiting_peers.(i));
1964                 n_search_asked_peers :=                 n_search_asked_peers :=
# Line 1956  Define a function to be called when the Line 1970  Define a function to be called when the
1970                done;                done;
1971               incr n_overnet_searches               incr n_overnet_searches
1972        ) !overnet_searches;        ) !overnet_searches;
       Printf.bprintf buf "  n_search_known_peers: %d\n" !n_search_known_peers;  
1973        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;
1974        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;
1975        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.55  
changed lines
  Added in v.1.56

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