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 = |
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 |
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; |
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 |
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 |
|
|
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 |
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 -> () |
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. *) |
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 |
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) |
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; |
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; |
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 |
|
|
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 = |
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; |
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 ()); |
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 |
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; |
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 -> |
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; |
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 |
|
|
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 |
|
|
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 |
|
|
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 -> |
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 |
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 |
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 |
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; |
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 |
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 |
|
|
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 |
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 |
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 |
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 ( |
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; |
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 (); |
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 |
"" |
"" |
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 (); |
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" ([ |
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; |
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; |
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 |
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 := |
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; |