44 |
open FasttrackComplexOptions |
open FasttrackComplexOptions |
45 |
open FasttrackProto |
open FasttrackProto |
46 |
|
|
47 |
let bootstrap_from_imesh = define_option fasttrack_section |
let bootstrap_from_imesh = define_option fasttrack_section |
48 |
["bootstrap_from_imesh"] |
["bootstrap_from_imesh"] |
49 |
"(only for development tests)" |
"(only for development tests)" |
50 |
bool_option true |
bool_option true |
51 |
|
|
52 |
let server_parse_after s gconn sock = |
let server_parse_after s gconn sock = |
53 |
try |
try |
54 |
match s.server_ciphers with |
match s.server_ciphers with |
55 |
None -> assert false |
None -> assert false |
72 |
iter () |
iter () |
73 |
in |
in |
74 |
iter () |
iter () |
75 |
with e -> |
with e -> |
76 |
lprintf "Exception %s in server_parse_after\n" |
lprintf "Exception %s in server_parse_after\n" |
77 |
(Printexc2.to_string e); |
(Printexc2.to_string e); |
78 |
close sock (Closed_for_error "Reply not understood") |
close sock (Closed_for_error "Reply not understood") |
79 |
|
|
80 |
let server_connection_hook = ref (None : (server -> unit) option) |
let server_connection_hook = ref (None : (server -> unit) option) |
81 |
|
|
82 |
let greet_supernode s = |
let greet_supernode s = |
83 |
(match !server_connection_hook with |
(match !server_connection_hook with |
84 |
None -> |
None -> |
85 |
server_send s TcpMessages.DirectPacket ( |
server_send s TcpMessages.DirectPacket ( |
86 |
TcpMessages.NodeInfoReq ( |
TcpMessages.NodeInfoReq ( |
87 |
client_ip s.server_sock, |
client_ip s.server_sock, |
88 |
!!client_port, |
!!client_port, |
89 |
default_bandwidth, |
default_bandwidth, |
90 |
client_name ())) |
client_name ())) |
91 |
| Some f -> f s) |
| Some f -> f s) |
92 |
(* ; server_send_ping s *) |
(* ; server_send_ping s *) |
93 |
|
|
94 |
let server_parse_netname s gconn sock = |
let server_parse_netname s gconn sock = |
95 |
let b = TcpBufferedSocket.buf sock in |
let b = TcpBufferedSocket.buf sock in |
96 |
let len = b.len in |
let len = b.len in |
97 |
let start_pos = b.pos in |
let start_pos = b.pos in |
99 |
let buf = b.buf in |
let buf = b.buf in |
100 |
let net = String.sub buf start_pos len in |
let net = String.sub buf start_pos len in |
101 |
if !verbose_msg_raw then |
if !verbose_msg_raw then |
102 |
lprintf "net:[%s]\n" (String.escaped net); |
lprintf "net:[%s]\n" (String.escaped net); |
103 |
let rec iter pos = |
let rec iter pos = |
104 |
if pos < end_pos then |
if pos < end_pos then |
105 |
if buf.[pos] = '\000' then begin |
if buf.[pos] = '\000' then begin |
106 |
let netname = String.sub buf start_pos (pos-start_pos) in |
let netname = String.sub buf start_pos (pos-start_pos) in |
107 |
if !verbose_msg_raw then |
if !verbose_msg_raw then |
108 |
lprintf "netname: [%s]\n" (String.escaped netname); |
lprintf "netname: [%s]\n" (String.escaped netname); |
109 |
buf_used b (pos-start_pos+1); |
buf_used b (pos-start_pos+1); |
110 |
match s.server_ciphers with |
match s.server_ciphers with |
111 |
None -> assert false |
None -> assert false |
112 |
| Some ciphers -> |
| Some ciphers -> |
113 |
gconn.gconn_handler <- |
gconn.gconn_handler <- |
114 |
CipherReader (ciphers.in_cipher, server_parse_after s); |
CipherReader (ciphers.in_cipher, server_parse_after s); |
115 |
greet_supernode s |
greet_supernode s |
116 |
end else |
end else |
117 |
iter (pos+1) |
iter (pos+1) |
118 |
in |
in |
119 |
iter start_pos |
iter start_pos |
120 |
|
|
121 |
let server_parse_cipher s gconn sock = |
let server_parse_cipher s gconn sock = |
122 |
H.connected s.server_host; |
H.connected s.server_host; |
123 |
let b = TcpBufferedSocket.buf sock in |
let b = TcpBufferedSocket.buf sock in |
124 |
if b.len >= 8 then |
if b.len >= 8 then |
129 |
lprintf "Cipher received from server\n"; |
lprintf "Cipher received from server\n"; |
130 |
get_cipher_from_packet b.buf b.pos ciphers.in_cipher; |
get_cipher_from_packet b.buf b.pos ciphers.in_cipher; |
131 |
init_cipher ciphers.in_cipher; |
init_cipher ciphers.in_cipher; |
132 |
|
|
133 |
xor_ciphers ciphers.out_cipher ciphers.in_cipher; |
xor_ciphers ciphers.out_cipher ciphers.in_cipher; |
134 |
init_cipher ciphers.out_cipher; |
init_cipher ciphers.out_cipher; |
135 |
|
|
136 |
buf_used b 8; |
buf_used b 8; |
137 |
server_crypt_and_send s ciphers.out_cipher (network_name ^ "\000"); |
server_crypt_and_send s ciphers.out_cipher (network_name ^ "\000"); |
138 |
gconn.gconn_handler <- CipherReader (ciphers.in_cipher, server_parse_netname s); |
gconn.gconn_handler <- CipherReader (ciphers.in_cipher, server_parse_netname s); |
139 |
if !verbose_msg_raw then |
if !verbose_msg_raw then |
140 |
lprintf "waiting for netname\n" |
lprintf "waiting for netname\n" |
141 |
|
|
142 |
let client_cipher_seed () = |
let client_cipher_seed () = |
143 |
(* Int32.of_int (Random.int max_int) *) |
(* Int32.of_int (Random.int max_int) *) |
144 |
Int32.of_string "0x0fACB1238" |
Int32.of_string "0x0fACB1238" |
145 |
|
|
146 |
let connection_header_hook = ref None |
let connection_header_hook = ref None |
147 |
|
|
148 |
let connect_server h = |
let connect_server h = |
149 |
let s = match h.host_server with |
let s = match h.host_server with |
150 |
None -> |
None -> |
151 |
let s = new_server h.host_addr h.host_port in |
let s = new_server h.host_addr h.host_port in |
152 |
h.host_server <- Some s; |
h.host_server <- Some s; |
153 |
s |
s |
154 |
| Some s -> s |
| Some s -> s |
155 |
in |
in |
156 |
match s.server_sock with |
match s.server_sock with |
157 |
| NoConnection -> |
| NoConnection -> |
158 |
incr nservers; |
incr nservers; |
159 |
let token = |
let token = |
160 |
add_pending_connection connection_manager (fun token -> |
add_pending_connection connection_manager (fun token -> |
165 |
failwith "Invalid IP for server\n"; |
failwith "Invalid IP for server\n"; |
166 |
let port = s.server_host.host_port in |
let port = s.server_host.host_port in |
167 |
if !verbose_msg_servers then |
if !verbose_msg_servers then |
168 |
lprintf "CONNECT TO %s:%d\n" |
lprintf "CONNECT TO %s:%d\n" |
169 |
(Ip.string_of_addr h.host_addr) port; |
(Ip.string_of_addr h.host_addr) port; |
170 |
H.set_request h Tcp_Connect; |
H.set_request h Tcp_Connect; |
171 |
H.try_connect h; |
H.try_connect h; |
178 |
let ip = Ip.to_inet_addr ip in |
let ip = Ip.to_inet_addr ip in |
179 |
let sock = connect token "fasttrack to server" |
let sock = connect token "fasttrack to server" |
180 |
ip port |
ip port |
181 |
(fun sock event -> |
(fun sock event -> |
182 |
match event with |
match event with |
183 |
BASIC_EVENT (RTIMEOUT|LTIMEOUT) -> |
BASIC_EVENT (RTIMEOUT|LTIMEOUT) -> |
184 |
(* lprintf "RTIMEOUT\n"; *) |
(* lprintf "RTIMEOUT\n"; *) |
185 |
disconnect_from_server nservers s Closed_for_timeout |
disconnect_from_server nservers s Closed_for_timeout |
186 |
| _ -> () |
| _ -> () |
187 |
) in |
) in |
188 |
TcpBufferedSocket.set_read_controler sock download_control; |
TcpBufferedSocket.set_read_controler sock download_control; |
189 |
TcpBufferedSocket.set_write_controler sock upload_control; |
TcpBufferedSocket.set_write_controler sock upload_control; |
190 |
|
|
191 |
set_server_state s Connecting; |
set_server_state s Connecting; |
192 |
s.server_sock <- Connection sock; |
s.server_sock <- Connection sock; |
193 |
incr nservers; |
incr nservers; |
194 |
set_fasttrack_sock sock !verbose_msg_servers |
set_fasttrack_sock sock !verbose_msg_servers |
195 |
(Reader (server_parse_cipher s) |
(Reader (server_parse_cipher s) |
196 |
); |
); |
197 |
set_closer sock (fun _ error -> |
set_closer sock (fun _ error -> |
198 |
(* lprintf "CLOSER %s\n" error; *) |
(* lprintf "CLOSER %s\n" error; *) |
199 |
disconnect_from_server nservers s error); |
disconnect_from_server nservers s error); |
200 |
set_rtimeout sock !!server_connection_timeout; |
set_rtimeout sock !!server_connection_timeout; |
201 |
|
|
202 |
let in_cipher = create_cipher () in |
let in_cipher = create_cipher () in |
203 |
let out_cipher = create_cipher () in |
let out_cipher = create_cipher () in |
204 |
s.server_ciphers <- Some { |
s.server_ciphers <- Some { |
208 |
out_xinu = Int64.of_int 0x51; |
out_xinu = Int64.of_int 0x51; |
209 |
}; |
}; |
210 |
set_cipher out_cipher (client_cipher_seed ()) 0x29; |
set_cipher out_cipher (client_cipher_seed ()) 0x29; |
211 |
|
|
212 |
let s = String.create 12 in |
let s = String.create 12 in |
213 |
|
|
214 |
(match !connection_header_hook with |
(match !connection_header_hook with |
215 |
None -> |
None -> |
216 |
s.[0] <- '\250'; |
s.[0] <- '\250'; |
217 |
s.[1] <- '\000'; |
s.[1] <- '\000'; |
218 |
s.[2] <- '\182'; |
s.[2] <- '\182'; |
219 |
s.[3] <- '\043'; |
s.[3] <- '\043'; |
220 |
| Some f -> f s); |
| Some f -> f s); |
221 |
|
|
222 |
cipher_packet_set out_cipher s 4; |
cipher_packet_set out_cipher s 4; |
223 |
|
|
224 |
if !verbose_msg_raw then begin |
if !verbose_msg_raw then begin |
225 |
lprintf "SENDING %s\n" (String.escaped s); |
lprintf "SENDING %s\n" (String.escaped s); |
226 |
AnyEndian.dump s; |
AnyEndian.dump s; |
232 |
in |
in |
233 |
s.server_sock <- ConnectionWaiting token; |
s.server_sock <- ConnectionWaiting token; |
234 |
| _ -> () |
| _ -> () |
235 |
|
|
236 |
let get_file_from_source c file = |
let get_file_from_source c file = |
237 |
try |
try |
238 |
if connection_can_try c.client_connection_control then begin |
if connection_can_try c.client_connection_control then begin |
240 |
match c.client_user.user_kind with |
match c.client_user.user_kind with |
241 |
Indirect_location ("", uid) -> |
Indirect_location ("", uid) -> |
242 |
(* |
(* |
243 |
lprintf "++++++ ASKING FOR PUSH +++++++++\n"; |
lprintf "++++++ ASKING FOR PUSH +++++++++\n"; |
244 |
|
|
245 |
(* do as if connection failed. If it connects, connection will be set to OK *) |
(* do as if connection failed. If it connects, connection will be set to OK *) |
246 |
connection_failed c.client_connection_control; |
connection_failed c.client_connection_control; |
259 |
end |
end |
260 |
with e -> |
with e -> |
261 |
lprintf "get_file_from_source: exception %s\n" (Printexc2.to_string e) |
lprintf "get_file_from_source: exception %s\n" (Printexc2.to_string e) |
262 |
|
|
263 |
let exit = Exit |
let exit = Exit |
264 |
|
|
265 |
let disconnect_server s r = |
let disconnect_server s r = |
266 |
match s.server_sock with |
match s.server_sock with |
267 |
| Connection sock -> close sock r |
| Connection sock -> close sock r |
268 |
| ConnectionWaiting token -> |
| ConnectionWaiting token -> |
269 |
cancel_token token; |
cancel_token token; |
270 |
s.server_sock <- NoConnection; |
s.server_sock <- NoConnection; |
271 |
free_ciphers s |
free_ciphers s |
272 |
|
|
273 |
| _ -> () |
| _ -> () |
274 |
|
|
275 |
let really_recover_file file = |
let really_recover_file file = |
276 |
List.iter (fun s -> |
List.iter (fun s -> |
277 |
List.iter (fun ss -> |
List.iter (fun ss -> |
278 |
if not (Fifo.mem s.server_searches ss) then |
if not (Fifo.mem s.server_searches ss) then |
279 |
Fifo.put s.server_searches ss |
Fifo.put s.server_searches ss |
280 |
) file.file_searches |
) file.file_searches |
281 |
) !connected_servers |
) !connected_servers |
282 |
|
|
283 |
let really_download_file (r : CommonTypes.result_info) = |
let really_download_file (r : CommonTypes.result_info) = |
284 |
let rec iter uids = |
let rec iter uids = |
285 |
match uids with |
match uids with |
286 |
uid :: tail -> |
uid :: tail -> |
287 |
(match Uid.to_uid uid with |
(match Uid.to_uid uid with |
288 |
Md5Ext hash -> hash, Uid.to_file_string uid |
Md5Ext hash -> hash, Uid.to_file_string uid |
290 |
| [] -> raise IgnoreNetwork |
| [] -> raise IgnoreNetwork |
291 |
in |
in |
292 |
let hash,file_temp = iter r.result_uids in |
let hash,file_temp = iter r.result_uids in |
293 |
|
|
294 |
|
let file = new_file file_temp (List.hd r.result_names) |
|
let file = new_file file_temp (List.hd r.result_names) |
|
295 |
r.result_size [Uid.create (Md5Ext hash)] in |
r.result_size [Uid.create (Md5Ext hash)] in |
296 |
if !verbose then |
if !verbose then |
297 |
lprintf "DOWNLOAD FILE %s\n" file.file_name; |
lprintf "DOWNLOAD FILE %s\n" file.file_name; |
303 |
List.iter (fun (user, _) -> |
List.iter (fun (user, _) -> |
304 |
let c = new_client user.user_kind in |
let c = new_client user.user_kind in |
305 |
add_download file c (); |
add_download file c (); |
306 |
get_file_from_source c file; |
get_file_from_source c file; |
307 |
) !sources; |
) !sources; |
308 |
end; |
end; |
309 |
file |
file |
310 |
|
|
311 |
let ask_for_files () = (* called every minute *) |
let ask_for_files () = (* called every minute *) |
312 |
List.iter (fun file -> |
List.iter (fun file -> |
313 |
List.iter (fun c -> |
List.iter (fun c -> |
322 |
FileUidSearch (file,file_hash) -> |
FileUidSearch (file,file_hash) -> |
323 |
if file_state file = FileDownloading then |
if file_state file = FileDownloading then |
324 |
server_send s M.DirectPacket |
server_send s M.DirectPacket |
325 |
(M.SearchReq |
(M.SearchReq |
326 |
(32, ss.search_uid, M.QueryLocationReq file_hash)) |
(32, ss.search_uid, M.QueryLocationReq file_hash)) |
327 |
| UserSearch (_, words, (realm, tags)) -> |
| UserSearch (_, words, (realm, tags)) -> |
328 |
|
|
329 |
let realm = |
let realm = |
330 |
match realm with |
match realm with |
331 |
"audio" -> 0x21 |
"audio" -> 0x21 |
332 |
| "video" -> 0x22 |
| "video" -> 0x22 |
337 |
in |
in |
338 |
|
|
339 |
server_send s M.DirectPacket |
server_send s M.DirectPacket |
340 |
(M.SearchReq |
(M.SearchReq |
341 |
(32, ss.search_uid, M.QueryFilesReq |
(32, ss.search_uid, M.QueryFilesReq |
342 |
(words, realm, tags))) |
(words, realm, tags))) |
343 |
with _ -> () |
with _ -> () |
344 |
) !connected_servers; |
) !connected_servers; |
345 |
() |
() |
346 |
|
|
347 |
let _ = |
let _ = |
348 |
server_ops.op_server_disconnect <- (fun s -> |
server_ops.op_server_disconnect <- (fun s -> |
349 |
disconnect_server s Closed_by_user); |
disconnect_server s Closed_by_user); |
351 |
disconnect_server s Closed_by_user |
disconnect_server s Closed_by_user |
352 |
) |
) |
353 |
|
|
354 |
let nranges file = |
let nranges file = |
355 |
Int64.to_int (Int64.div (file_size file) |
Int64.to_int (Int64.div (file_size file) |
356 |
min_range_size) + 5 |
min_range_size) + 5 |
357 |
|
|
358 |
let manage_hosts () = |
let manage_hosts () = |
359 |
H.manage_hosts (); |
H.manage_hosts (); |
360 |
List.iter (fun file -> |
List.iter (fun file -> |
361 |
if file_state file = FileDownloading then |
if file_state file = FileDownloading then |
362 |
try |
try |
363 |
(* For each file, we allow only (nranges+5) simultaneous communications, |
(* For each file, we allow only (nranges+5) simultaneous communications, |
364 |
to prevent too many clients from saturing the line for only one file. *) |
to prevent too many clients from saturing the line for only one file. *) |
365 |
let max_nconnected_clients = nranges file in |
let max_nconnected_clients = nranges file in |
366 |
while file.file_nconnected_clients < max_nconnected_clients do |
while file.file_nconnected_clients < max_nconnected_clients do |
371 |
with _ -> () |
with _ -> () |
372 |
) !current_files |
) !current_files |
373 |
|
|
|
|
|
374 |
let rec find_ultrapeer queue = |
let rec find_ultrapeer queue = |
375 |
let (next,h) = Queue.head queue in |
let (next,h) = Queue.head queue in |
376 |
try |
try |
384 |
|
|
385 |
let ft_boot () = |
let ft_boot () = |
386 |
let imesh_ip = Ip.addr_of_string "fm2.imesh.com" in |
let imesh_ip = Ip.addr_of_string "fm2.imesh.com" in |
387 |
let (h : host) = |
let (h : host) = |
388 |
if !verbose then lprintf_nl "FT: Bootstrapping from Imesh %s" (Ip.string_of_addr imesh_ip); |
if !verbose then lprintf_nl "FT: Bootstrapping from Imesh %s" (Ip.string_of_addr imesh_ip); |
389 |
H.new_host imesh_ip 1214 IndexServer |
H.new_host imesh_ip 1214 IndexServer |
390 |
in |
in |
391 |
connect_server h |
connect_server h |
392 |
|
|
393 |
let try_connect_ultrapeer connect = |
let try_connect_ultrapeer connect = |
394 |
(* lprintf "try_connect_ultrapeer....\n"; *) |
(* lprintf "try_connect_ultrapeer....\n"; *) |
395 |
let h = |
let h = |
396 |
try |
try |
397 |
find_ultrapeer ultrapeers_waiting_queue |
find_ultrapeer ultrapeers_waiting_queue |
398 |
with _ -> |
with _ -> |
399 |
(* lprintf "not in ultrapeers_waiting_queue\n"; *) |
(* lprintf "not in ultrapeers_waiting_queue\n"; *) |
400 |
if !!bootstrap_from_imesh then |
if !!bootstrap_from_imesh then |
401 |
let imesh_ip = Ip.addr_of_string "fm2.imesh.com" in |
let imesh_ip = Ip.addr_of_string "fm2.imesh.com" in |
402 |
let (h : host) = |
let (h : host) = |
403 |
if !verbose then lprintf_nl "FT: Bootstrapping from Imesh %s" (Ip.string_of_addr imesh_ip); |
if !verbose then lprintf_nl "FT: Bootstrapping from Imesh %s" (Ip.string_of_addr imesh_ip); |
404 |
H.new_host (Ip.addr_of_string "fm2.imesh.com") 1214 IndexServer in |
H.new_host (Ip.addr_of_string "fm2.imesh.com") 1214 IndexServer in |
405 |
find_ultrapeer peers_waiting_queue |
find_ultrapeer peers_waiting_queue |
407 |
in |
in |
408 |
(* lprintf "contacting..\n"; *) |
(* lprintf "contacting..\n"; *) |
409 |
connect h |
connect h |
410 |
|
|
411 |
let connect_servers connect = |
let connect_servers connect = |
412 |
(* lprintf "connect_servers %d %d\n" !nservers !!max_ultrapeers; *) |
(* lprintf "connect_servers %d %d\n" !nservers !!max_ultrapeers; *) |
413 |
(if !!max_ultrapeers > List.length !connected_servers then |
(if !!max_ultrapeers > List.length !connected_servers then |
418 |
try_connect_ultrapeer connect |
try_connect_ultrapeer connect |
419 |
done |
done |
420 |
with _ -> ()) |
with _ -> ()) |
421 |
|
|
422 |
(* Looks like there is no ping to send in Fasttrack? *) |
(* Looks like there is no ping to send in Fasttrack? *) |
423 |
let send_pings () = () |
let send_pings () = () |
424 |
|
|
425 |
let send_query ss = |
let send_query ss = |
426 |
let s_uid = ss.search_uid in |
let s_uid = ss.search_uid in |
427 |
let module M = TcpMessages in |
let module M = TcpMessages in |
440 |
M.QueryFilesReq (words, realm, tags) |
M.QueryFilesReq (words, realm, tags) |
441 |
| FileUidSearch (_, file_hash) -> |
| FileUidSearch (_, file_hash) -> |
442 |
M.QueryLocationReq file_hash |
M.QueryLocationReq file_hash |
443 |
in |
in |
444 |
List.iter (fun s -> |
List.iter (fun s -> |
445 |
FasttrackProto.server_send s M.DirectPacket ( |
FasttrackProto.server_send s M.DirectPacket ( |
446 |
M.SearchReq (32, s_uid, t))) !connected_servers |
M.SearchReq (32, s_uid, t))) !connected_servers |
447 |
|
|
448 |
(* |
(* |
449 |
TODO: |
TODO: |
450 |
|
|
451 |
push request: we send a push to the server when we cannot connect to |
push request: we send a push to the server when we cannot connect to |
452 |
a particular client. The client by connecting to us with a |
a particular client. The client by connecting to us with a |
453 |
"GIVE <push_id>\r\n" request, to which we can reply by a "GET ...." *) |
"GIVE <push_id>\r\n" request, to which we can reply by a "GET ...." *) |
|
|
|