/[mldonkey]/mldonkey/src/networks/fasttrack/fasttrackServers.ml
ViewVC logotype

Diff of /mldonkey/src/networks/fasttrack/fasttrackServers.ml

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

revision 1.22 by spiralvoice, Mon Aug 1 23:53:54 2005 UTC revision 1.23 by spiralvoice, Sun Aug 7 12:57:22 2005 UTC
# Line 44  open FasttrackProtocol Line 44  open FasttrackProtocol
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
# Line 72  let server_parse_after s gconn sock = Line 72  let server_parse_after s gconn sock =
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
# Line 99  let server_parse_netname s gconn sock = Line 99  let server_parse_netname s gconn sock =
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
# Line 129  let server_parse_cipher s gconn sock = Line 129  let server_parse_cipher s gconn sock =
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 ->
# Line 165  let connect_server h = Line 165  let connect_server h =
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;
# Line 178  let connect_server h = Line 178  let connect_server 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 {
# Line 208  let connect_server h = Line 208  let connect_server h =
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;
# Line 232  let connect_server h = Line 232  let connect_server h =
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
# Line 240  let get_file_from_source c file = Line 240  let get_file_from_source c file =
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;
# Line 259  let get_file_from_source c file = Line 259  let get_file_from_source c file =
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
# Line 290  let really_download_file (r : CommonType Line 290  let really_download_file (r : CommonType
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;
# Line 304  let really_download_file (r : CommonType Line 303  let really_download_file (r : CommonType
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 ->
# Line 323  let ask_for_files () = (* called every m Line 322  let ask_for_files () = (* called every m
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
# Line 338  let ask_for_files () = (* called every m Line 337  let ask_for_files () = (* called every m
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);
# Line 352  let _ = Line 351  let _ =
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
# Line 372  let manage_hosts () = Line 371  let manage_hosts () =
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
# Line 386  let rec find_ultrapeer queue = Line 384  let rec find_ultrapeer queue =
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
# Line 409  let try_connect_ultrapeer connect = Line 407  let try_connect_ultrapeer connect =
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
# Line 420  let connect_servers connect = Line 418  let connect_servers connect =
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
# Line 442  let send_query ss = Line 440  let send_query ss =
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 ...." *)
     

Legend:
Removed from v.1.22  
changed lines
  Added in v.1.23

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