/[mldonkey]/mldonkey/src/utils/net/tcpBufferedSocket.ml
ViewVC logotype

Diff of /mldonkey/src/utils/net/tcpBufferedSocket.ml

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

revision 1.27 by spiralvoice, Thu Jul 14 14:02:28 2005 UTC revision 1.28 by spiralvoice, Thu Jul 14 15:34:15 2005 UTC
# Line 21  open Int64ops Line 21  open Int64ops
21  open Printf2  open Printf2
22  open BasicSocket  open BasicSocket
23    
24      
25  let latencies = Hashtbl.create 2131  let latencies = Hashtbl.create 2131
26    
27  let max_opened_connections = ref (fun () -> maxi 20 (MlUnix.max_sockets - 50))  let max_opened_connections = ref (fun () -> maxi 20 (MlUnix.max_sockets - 50))
# Line 76  type buf = { Line 76  type buf = {
76      mutable buf : string;      mutable buf : string;
77      mutable pos : int;      mutable pos : int;
78      mutable len : int;      mutable len : int;
79      mutable max_buf_size:int;      mutable max_buf_size : int;
80      mutable min_buf_size: int      mutable min_buf_size : int
81    }    }
82    
83  type t = {  type t = {
# Line 107  type t = { Line 107  type t = {
107      mutable connecting : bool;      mutable connecting : bool;
108      mutable host : Ip.t;      mutable host : Ip.t;
109      mutable connect_time : float;      mutable connect_time : float;
110        
111      mutable token : token;      mutable token : token;
112    
113      mutable compression : (      mutable compression : (
# Line 132  and bandwidth_controler = { Line 132  and bandwidth_controler = {
132      mutable moved_bytes : int64;      mutable moved_bytes : int64;
133      mutable lost_bytes : int array; (* 3600 samples*)      mutable lost_bytes : int array; (* 3600 samples*)
134      mutable forecast_bytes : int;      mutable forecast_bytes : int;
135        
136      mutable ndone_last_second : int;      mutable ndone_last_second : int;
137    }    }
138    
# Line 181  let schedule_connections () = Line 181  let schedule_connections () =
181    let max_wanted = !max_opened_connections () in    let max_wanted = !max_opened_connections () in
182    let current_connections = !opened_connections in    let current_connections = !opened_connections in
183    let max_connections_per_second = !max_connections_per_second () in    let max_connections_per_second = !max_connections_per_second () in
184      
185    let rec iter todo_managers done_managers =    let rec iter todo_managers done_managers =
186    
187  (*  (*
188      lprintf "todo_managers %d done_managers %d\n"      lprintf "todo_managers %d done_managers %d\n"
189  (List.length todo_managers) (List.length done_managers);  (List.length todo_managers) (List.length done_managers);
190    *)    *)
191      match todo_managers with      match todo_managers with
# Line 231  let unlimited_connection_manager = creat Line 231  let unlimited_connection_manager = creat
231    
232  let reset_connection_scheduler _ =  let reset_connection_scheduler _ =
233    if !verbose_bandwidth > 0 then begin    if !verbose_bandwidth > 0 then begin
234        lprintf "[BW1 %6d] Connections opened this second : %d/%d total %d/%d\n"        lprintf "[BW1 %6d] Connections opened this second : %d/%d total %d/%d\n"
235        (last_time ()) !opened_connections_this_second (!max_connections_per_second ()) !opened_connections (!max_opened_connections ());        (last_time ()) !opened_connections_this_second (!max_connections_per_second ()) !opened_connections (!max_opened_connections ());
             
236      end;      end;
237    List.iter (fun cm ->    List.iter (fun cm ->
238        if !verbose_bandwidth > 0 then begin        if !verbose_bandwidth > 0 then begin
239            if cm.nconnections_last_second > 0 then            if cm.nconnections_last_second > 0 then
240              lprintf "[BW1 %6d]    %s opened %d connections last second\n"              lprintf "[BW1 %6d]    %s opened %d connections last second\n"
241                (last_time ()) cm.cm_name cm.nconnections_last_second;                (last_time ()) cm.cm_name cm.nconnections_last_second;
242            if cm.nwaiting_connections > 0 then            if cm.nwaiting_connections > 0 then
243              lprintf "[BW1 %6d]    %s still waits for %d connections\n"              lprintf "[BW1 %6d]    %s still waits for %d connections\n"
244                (last_time ()) cm.cm_name cm.nwaiting_connections;                (last_time ()) cm.cm_name cm.nwaiting_connections;
245         end;             end;
246        cm.nconnections_last_second <- 0;        cm.nconnections_last_second <- 0;
247    ) !connection_managers;    ) !connection_managers;
248      
249    opened_connections_this_second := 0    opened_connections_this_second := 0
250    
251  let use_token token fd =  let use_token token fd =
# Line 273  let add_connect_latency ip time = Line 272  let add_connect_latency ip time =
272      let delay = int_of_float delayf in      let delay = int_of_float delayf in
273      let delay = if delay > 65000 then 65000 else delay in      let delay = if delay > 65000 then 65000 else delay in
274  (*  (*
275  lprintf "add_connect_latency %s -> %d (%f)\n"  lprintf "add_connect_latency %s -> %d (%f)\n"
276  (Ip.to_string ip) delay delayf;  (Ip.to_string ip) delay delayf;
277    *)    *)
278      try      try
279        let latency, samples = Hashtbl.find latencies ip in        let latency, samples = Hashtbl.find latencies ip in
280        incr samples;        incr samples;
281        if !latency > delay then latency := delay        if !latency > delay then latency := delay
282      with _ ->      with _ ->
283          Hashtbl.add latencies ip (ref delay, ref 1)          Hashtbl.add latencies ip (ref delay, ref 1)
284            
285  let forecast_bytes t nbytes =  let forecast_bytes t nbytes =
286    match t with    match t with
287      None -> ()      None -> ()
288    | Some bc ->          | Some bc ->
289        let nip_packets = 1 + nbytes / !mtu_packet_size in        let nip_packets = 1 + nbytes / !mtu_packet_size in
290        let nbytes = nbytes + nip_packets * !ip_packet_size in        let nbytes = nbytes + nip_packets * !ip_packet_size in
291        let nframes = 1 + nbytes / packet_frame_size in        let nframes = 1 + nbytes / packet_frame_size in
292        bc.forecast_bytes <- bc.forecast_bytes +        bc.forecast_bytes <- bc.forecast_bytes +
293          (nframes * packet_frame_size)          (nframes * packet_frame_size)
294      
295  let register_bytes t nbytes =  let register_bytes t nbytes =
296    match t with    match t with
297      None -> ()      None -> ()
298    | Some bc ->          | Some bc ->
299        let nip_packets = 1 + nbytes / !mtu_packet_size in        let nip_packets = 1 + nbytes / !mtu_packet_size in
300        let nbytes = nbytes + nip_packets * !ip_packet_size in        let nbytes = nbytes + nip_packets * !ip_packet_size in
301        let nframes = 1 + nbytes / packet_frame_size in        let nframes = 1 + nbytes / packet_frame_size in
302        bc.remaining_bytes <- bc.remaining_bytes -        bc.remaining_bytes <- bc.remaining_bytes -
303        (nframes * packet_frame_size)        (nframes * packet_frame_size)
304      
305  let forecast_download t n =  let forecast_download t n =
306    forecast_bytes t.read_control n    forecast_bytes t.read_control n
307    
# Line 334  let best_packet_size nbytes = Line 333  let best_packet_size nbytes =
333  (*************************************************************************)  (*************************************************************************)
334    
335  let copy_read_buffer = ref true  let copy_read_buffer = ref true
336      
337  let big_buffer_len = 65536  let big_buffer_len = 65536
338  let big_buffer = String.create big_buffer_len  let big_buffer = String.create big_buffer_len
339      
340  let min_buffer_read = 2000  let min_buffer_read = 2000
341  let min_read_size = min_buffer_read - 100  let min_read_size = min_buffer_read - 100
342    
# Line 459  let remaining_to_write t =  t.wbuf.len Line 458  let remaining_to_write t =  t.wbuf.len
458  let nread t = t.nread  let nread t = t.nread
459  let nwritten t = t.nwrite  let nwritten t = t.nwrite
460  let can_write t =  t.wbuf.len = 0  let can_write t =  t.wbuf.len = 0
461  let can_write_len t len =  let can_write_len t len =
462    let b = t.wbuf.max_buf_size > t.wbuf.len + len in    let b = t.wbuf.max_buf_size > t.wbuf.len + len in
463  (*  if not b then  (*  if not b then
464      lprintf "can_write_len failed: %d < %d + %d\n"      lprintf "can_write_len failed: %d < %d + %d\n"
# Line 468  let can_write_len t len = Line 467  let can_write_len t len =
467  let not_buffer_more t max =  t.wbuf.len < max  let not_buffer_more t max =  t.wbuf.len < max
468  let can_fill t = t.wbuf.len < (t.wbuf.max_buf_size / 2)  let can_fill t = t.wbuf.len < (t.wbuf.max_buf_size / 2)
469  let get_rtimeout t = get_rtimeout t.sock_in  let get_rtimeout t = get_rtimeout t.sock_in
470  let max_refill t = t.wbuf.max_buf_size - t.wbuf.len  let max_refill t = t.wbuf.max_buf_size - t.wbuf.len
471    
472  let close t s =  let close t s =
473  (*  (*
# Line 507  let shutdown t s = Line 506  let shutdown t s =
506        lprintf "shutdown\n";        lprintf "shutdown\n";
507  end;  end;
508    *)    *)
509    (try    (try
510        BasicSocket.shutdown t.sock_out s;        BasicSocket.shutdown t.sock_out s;
511        if t.sock_in != t.sock_out then        if t.sock_in != t.sock_out then
512          BasicSocket.shutdown t.sock_in s;          BasicSocket.shutdown t.sock_in s;
# Line 543  let write t s pos1 len = Line 542  let write t s pos1 len =
542  (*       lprintf "WRITE [%s]\n" (String.escaped (String.sub s pos1 len)); *)  (*       lprintf "WRITE [%s]\n" (String.escaped (String.sub s pos1 len)); *)
543            let nw = MlUnix.write fd s pos1 len in            let nw = MlUnix.write fd s pos1 len in
544            if !verbose_bandwidth > 1 then begin            if !verbose_bandwidth > 1 then begin
545                lprintf "[BW2 %6d] immediate write %d/%d on %s:%d\n" (last_time ())                lprintf "[BW2 %6d] immediate write %d/%d on %s:%d\n" (last_time ())
546                nw len                nw len
547                  t.name (sock_num t.sock_out);                  t.name (sock_num t.sock_out);
548              end;              end;
549              
550            register_upload t len;            register_upload t len;
551            forecast_download t 0;            forecast_download t 0;
552    
# Line 561  end; *) Line 560  end; *)
560                  bc.moved_bytes <-                  bc.moved_bytes <-
561                    Int64.add bc.moved_bytes (Int64.of_int nw));                    Int64.add bc.moved_bytes (Int64.of_int nw));
562            if t.nwrite = 0 then begin            if t.nwrite = 0 then begin
563  (*              if t.connecting then  (*              if t.connecting then
564                  lprintf "WRITE BEFORE CONNECTION.......\n";                  lprintf "WRITE BEFORE CONNECTION.......\n";
565                lprintf "add_connect_latency at %f\n" (current_time ()); *)                lprintf "add_connect_latency at %f\n" (current_time ()); *)
566                add_connect_latency t.host t.connect_time;                add_connect_latency t.host t.connect_time;
567              end;              end;
568              
569            t.nwrite <- t.nwrite + nw;            t.nwrite <- t.nwrite + nw;
570            if nw = 0 then (close t Closed_by_peer; pos2) else            if nw = 0 then (close t Closed_by_peer; pos2) else
571              pos1 + nw              pos1 + nw
# Line 608  let can_read_handler t sock max_len = Line 607  let can_read_handler t sock max_len =
607        big_buffer, 0, big_buffer_len        big_buffer, 0, big_buffer_len
608      else      else
609      let can_write_in_buffer =      let can_write_in_buffer =
610        if b.buf = "" then        if b.buf = "" then
611          if b.min_buf_size <= min_buffer_read then begin          if b.min_buf_size <= min_buffer_read then begin
612              b.buf <- new_string ();              b.buf <- new_string ();
613              min_buffer_read              min_buffer_read
# Line 674  let can_read_handler t sock max_len = Line 673  let can_read_handler t sock max_len =
673          lprintf "READ LIMITED BY BW CONTROL: %d\n" nread;          lprintf "READ LIMITED BY BW CONTROL: %d\n" nread;
674        end;        end;
675  *)  *)
676        
677      if !verbose_bandwidth > 1 then begin      if !verbose_bandwidth > 1 then begin
678          lprintf "[BW2 %6d] %sread %d/%d/%d on %s:%d\n" (last_time ())          lprintf "[BW2 %6d] %sread %d/%d/%d on %s:%d\n" (last_time ())
679          (if old_len > 0 then "completing " else "") nread can_read max_len          (if old_len > 0 then "completing " else "") nread can_read max_len
680            t.name (sock_num t.sock_in);            t.name (sock_num t.sock_in);
681        end;        end;
682        
683        
684      if !copy_read_buffer then begin      if !copy_read_buffer then begin
685  (*        lprintf "Copying %d bytes\n" nread; *)  (*        lprintf "Copying %d bytes\n" nread; *)
686          buf_add t b big_buffer 0 nread          buf_add t b big_buffer 0 nread
# Line 698  let can_read_handler t sock max_len = Line 697  let can_read_handler t sock max_len =
697          lprintf "    given by buffer: %d\n" buffer_len;          lprintf "    given by buffer: %d\n" buffer_len;
698      end;      end;
699  *)  *)
700        
701      tcp_downloaded_bytes := Int64.add !tcp_downloaded_bytes (Int64.of_int nread);      tcp_downloaded_bytes := Int64.add !tcp_downloaded_bytes (Int64.of_int nread);
702      (match t.read_control with      (match t.read_control with
703          None -> () | Some bc ->          None -> () | Some bc ->
# Line 714  let can_read_handler t sock max_len = Line 713  let can_read_handler t sock max_len =
713        close t Closed_by_peer;        close t Closed_by_peer;
714      end else begin      end else begin
715    
         
716        try        try
717  (*              if t.monitored then  (*              if t.monitored then
718     (lprintf "event handler READ DONE\n"; ); *)     (lprintf "event handler READ DONE\n"; ); *)
# Line 747  let can_write_handler t sock max_len = Line 745  let can_write_handler t sock max_len =
745            try            try
746  (*     lprintf "try write %d/%d\n" max_len t.wbuf.len; *)  (*     lprintf "try write %d/%d\n" max_len t.wbuf.len; *)
747              let fd = fd sock in              let fd = fd sock in
748  (*            lprintf "WRITE [%s]\n" (String.escaped  (*            lprintf "WRITE [%s]\n" (String.escaped
749                (String.sub b.buf b.pos max_len)); *)                (String.sub b.buf b.pos max_len)); *)
750              let nw = MlUnix.write fd b.buf b.pos max_len in              let nw = MlUnix.write fd b.buf b.pos max_len in
751              if !verbose_bandwidth > 1 then begin              if !verbose_bandwidth > 1 then begin
752                  lprintf "[BW2 %6d] postponed %swrite %d/%d/%d on %s:%d\n" (last_time ()) (if max_len < b.len then "partial " else "") nw max_len b.len                  lprintf "[BW2 %6d] postponed %swrite %d/%d/%d on %s:%d\n" (last_time ()) (if max_len < b.len then "partial " else "") nw max_len b.len
753                    t.name (sock_num t.sock_out);                    t.name (sock_num t.sock_out);
754                end;                end;
755                
756  (*            if t.monitored then  (*            if t.monitored then
757  (lprintf "written %d\n" nw; ); *)  (lprintf "written %d\n" nw; ); *)
758          tcp_uploaded_bytes := Int64.add !tcp_uploaded_bytes (Int64.of_int nw);          tcp_uploaded_bytes := Int64.add !tcp_uploaded_bytes (Int64.of_int nw);
# Line 819  let tcp_handler_write t sock = Line 817  let tcp_handler_write t sock =
817                end                end
818            end            end
819    end    end
820      
821  let get_latencies verbose =  let get_latencies verbose =
822    let b = Buffer.create 300 in    let b = Buffer.create 300 in
823    let counter = ref 0 in      let counter = ref 0 in
824    Hashtbl.iter (fun ip (latency, samples) ->    Hashtbl.iter (fun ip (latency, samples) ->
825        incr counter;        incr counter;
826    ) latencies;    ) latencies;
# Line 835  let get_latencies verbose = Line 833  let get_latencies verbose =
833    ) latencies;    ) latencies;
834    Hashtbl.clear latencies;    Hashtbl.clear latencies;
835    Buffer.contents b    Buffer.contents b
836      
837  let tcp_handler t event =  let tcp_handler t event =
838    match event with    match event with
839    | CAN_READ ->    | CAN_READ ->
# Line 862  let tcp_handler t event = Line 860  let tcp_handler t event =
860          if t.nwrite = 0 && t.noproxy && t.connecting then  begin          if t.nwrite = 0 && t.noproxy && t.connecting then  begin
861              t.connecting <- false;              t.connecting <- false;
862              t.event_handler t CONNECTED;              t.event_handler t CONNECTED;
863              t.nwrite = 0              t.nwrite = 0
864            end else true            end else true
865        in        in
866        if can_write then tcp_handler_write t t.sock_out        if can_write then tcp_handler_write t t.sock_out
# Line 931  let reset_bandwidth_controlers _ = Line 929  let reset_bandwidth_controlers _ =
929        bc.remaining_bytes_user bc.total_bytes bc.remaining_bytes;        bc.remaining_bytes_user bc.total_bytes bc.remaining_bytes;
930        bc.remaining_bytes <- bc.total_bytes - bc.forecast_bytes;        bc.remaining_bytes <- bc.total_bytes - bc.forecast_bytes;
931        if !verbose_bandwidth > 0 && bc.ndone_last_second > 0 then        if !verbose_bandwidth > 0 && bc.ndone_last_second > 0 then
932          lprintf "[BW1 %6d] %s read %d/%d last second\n" (last_time ())          lprintf "[BW1 %6d] %s read %d/%d last second\n" (last_time ())
933            bc.bc_name bc.ndone_last_second bc.total_bytes;            bc.bc_name bc.ndone_last_second bc.total_bytes;
934        if !verbose_bandwidth > 0 && bc.forecast_bytes > 0 then        if !verbose_bandwidth > 0 && bc.forecast_bytes > 0 then
935          lprintf "[BW1 %6d] %s forecast read %d bytes for next second\n" (last_time ())          lprintf "[BW1 %6d] %s forecast read %d bytes for next second\n" (last_time ())
936            bc.bc_name bc.forecast_bytes;            bc.bc_name bc.forecast_bytes;
937        bc.forecast_bytes <- 0;        bc.forecast_bytes <- 0;
938        bc.ndone_last_second <- 0;        bc.ndone_last_second <- 0;
# Line 945  let reset_bandwidth_controlers _ = Line 943  let reset_bandwidth_controlers _ =
943        bc.remaining_bytes_user bc.total_bytes bc.remaining_bytes;        bc.remaining_bytes_user bc.total_bytes bc.remaining_bytes;
944        bc.remaining_bytes <- bc.total_bytes - bc.forecast_bytes;        bc.remaining_bytes <- bc.total_bytes - bc.forecast_bytes;
945        if !verbose_bandwidth > 0 && bc.ndone_last_second > 0 then        if !verbose_bandwidth > 0 && bc.ndone_last_second > 0 then
946          lprintf "[BW1 %6d] %s wrote %d/%d last second\n" (last_time ())          lprintf "[BW1 %6d] %s wrote %d/%d last second\n" (last_time ())
947            bc.bc_name bc.ndone_last_second bc.total_bytes;            bc.bc_name bc.ndone_last_second bc.total_bytes;
948        if !verbose_bandwidth > 0 && bc.forecast_bytes > 0 then        if !verbose_bandwidth > 0 && bc.forecast_bytes > 0 then
949          lprintf "[BW1 %6d] %s forecast write %d bytes for next second\n" (last_time ())          lprintf "[BW1 %6d] %s forecast write %d bytes for next second\n" (last_time ())
950            bc.bc_name bc.forecast_bytes;            bc.bc_name bc.forecast_bytes;
951        bc.forecast_bytes <- 0;        bc.forecast_bytes <- 0;
952        bc.ndone_last_second <- 0;        bc.ndone_last_second <- 0;
# Line 1027  let close_after_write t = Line 1025  let close_after_write t =
1025      set_handler t WRITE_DONE (fun t ->      set_handler t WRITE_DONE (fun t ->
1026  (*      lprintf "close_after_write: CLOSE\n";  - log output removed  *)  (*      lprintf "close_after_write: CLOSE\n";  - log output removed  *)
1027          shutdown t Closed_by_user)          shutdown t Closed_by_user)
1028        
1029  exception Http_proxy_error of string  exception Http_proxy_error of string
1030  let http_proxy = ref None  let http_proxy = ref None
1031    
# Line 1051  let set_reader t f = Line 1049  let set_reader t f =
1049                let rstr_pos = 12 (*String.index_from b.buf (rcode_pos+1) ' '*) in                let rstr_pos = 12 (*String.index_from b.buf (rcode_pos+1) ' '*) in
1050                let rstr_end = String.index_from b.buf (rstr_pos+1) '\n' in                let rstr_end = String.index_from b.buf (rstr_pos+1) '\n' in
1051                let rstr = String.sub b.buf (rstr_pos+1) (rstr_end-rstr_pos-1) in                let rstr = String.sub b.buf (rstr_pos+1) (rstr_end-rstr_pos-1) in
1052                lprintf "From proxy for %s: %s %s\n"                lprintf "From proxy for %s: %s %s\n"
1053                  (Ip.to_string sock.host) rcode rstr;                  (Ip.to_string sock.host) rcode rstr;
1054                rcode, rstr, rstr_end                rcode, rstr, rstr_end
1055              with _ ->              with _ ->
# Line 1102  let set_refill t f = Line 1100  let set_refill t f =
1100    
1101  let set_connected t f =  let set_connected t f =
1102    set_handler t CONNECTED f    set_handler t CONNECTED f
1103        
1104  (*************************************************************************)  (*************************************************************************)
1105  (*                                                                       *)  (*                                                                       *)
1106  (*                         Socket Configuration                          *)  (*                         Socket Configuration                          *)
# Line 1123  let set_write_controler t bc = Line 1121  let set_write_controler t bc =
1121    
1122  let set_monitored t b = t.monitored <- b  let set_monitored t b = t.monitored <- b
1123  let monitored t = t.monitored  let monitored t = t.monitored
1124      
1125  let set_rtimeout s t = set_rtimeout s.sock_in t  let set_rtimeout s t = set_rtimeout s.sock_in t
1126  let set_wtimeout s t = set_wtimeout s.sock_out t  let set_wtimeout s t = set_wtimeout s.sock_out t
1127    
# Line 1240  let create_pipe token name fd_in fd_out Line 1238  let create_pipe token name fd_in fd_out
1238        connect_time = 0.;        connect_time = 0.;
1239        compression = None;        compression = None;
1240      } in      } in
1241      
1242    let fname = (fun _ ->    let fname = (fun _ ->
1243          Printf.sprintf "%s (nread: %d nwritten: %d) [U %s,D %s]" name          Printf.sprintf "%s (nread: %d nwritten: %d) [U %s,D %s]" name
1244            t.nread t.nwrite (string_of_bool (t.read_control <> None))            t.nread t.nwrite (string_of_bool (t.read_control <> None))
1245          (string_of_bool (t.write_control <> None));          (string_of_bool (t.write_control <> None));
1246          ;          ;
1247      ) in      ) in
1248      
1249      
1250    let sock_in = BasicSocket.create name fd_in (fun _ event ->    let sock_in = BasicSocket.create name fd_in (fun _ event ->
1251          tcp_handler t event) in          tcp_handler t event) in
1252    if !debug then begin    if !debug then begin
# Line 1261  let create_pipe token name fd_in fd_out Line 1259  let create_pipe token name fd_in fd_out
1259          tcp_handler t event) in          tcp_handler t event) in
1260    set_printer sock_out fname;    set_printer sock_out fname;
1261    set_dump_info sock_out (dump_socket t);    set_dump_info sock_out (dump_socket t);
1262      
1263    t.sock_in <- sock_in;    t.sock_in <- sock_in;
1264    t.sock_out <- sock_out;    t.sock_out <- sock_out;
1265    t    t
# Line 1336  let connect token name host port handler Line 1334  let connect token name host port handler
1334  (* connect to proxy in blocking mode, so we sure, connections established when we send CONNECT *)  (* connect to proxy in blocking mode, so we sure, connections established when we send CONNECT *)
1335          lprintf "via proxy\n";          lprintf "via proxy\n";
1336          Unix.connect s (Unix.ADDR_INET(Ip.to_inet_addr proxy_ip, proxy_port));          Unix.connect s (Unix.ADDR_INET(Ip.to_inet_addr proxy_ip, proxy_port));
1337            
1338          let buf = Buffer.create 200 in          let buf = Buffer.create 200 in
1339          let dotted_host = Unix.string_of_inet_addr host in          let dotted_host = Unix.string_of_inet_addr host in
1340          Printf.bprintf buf "CONNECT %s:%d HTTP/1.1\n" dotted_host port;          Printf.bprintf buf "CONNECT %s:%d HTTP/1.1\n" dotted_host port;
# Line 1351  let connect token name host port handler Line 1349  let connect token name host port handler
1349          ()          ()
1350        end;        end;
1351      let t = create token name s handler in      let t = create token name s handler in
1352        
1353      if !verbose_bandwidth > 1 then begin      if !verbose_bandwidth > 1 then begin
1354          lprintf "[BW2 %6d] connect on %s:%d\n" (last_time ()) t.name (sock_num t.sock_out);          lprintf "[BW2 %6d] connect on %s:%d\n" (last_time ()) t.name (sock_num t.sock_out);
1355        end;        end;
1356        
1357      token.connection_manager.nconnections_last_second <-      token.connection_manager.nconnections_last_second <-
1358        token.connection_manager.nconnections_last_second + 1;        token.connection_manager.nconnections_last_second + 1;
1359        
1360      must_write t.sock_out true;      must_write t.sock_out true;
1361      try      try
1362        t.host <- ip;        t.host <- ip;
# Line 1368  let connect token name host port handler Line 1366  let connect token name host port handler
1366            t.noproxy <- false;            t.noproxy <- false;
1367          end else          end else
1368          Unix.connect s (Unix.ADDR_INET(host,port));          Unix.connect s (Unix.ADDR_INET(host,port));
1369          
1370        t.connecting <- true;        t.connecting <- true;
1371        register_upload t 0;    (* The TCP SYN packet *)        register_upload t 0;    (* The TCP SYN packet *)
1372        forecast_download t 0;  (* The TCP ACK packet *)        forecast_download t 0;  (* The TCP ACK packet *)
# Line 1392  let connect token name host port handler Line 1390  let connect token name host port handler
1390        lprintf "EXCEPTION %s  before connect to host %s:%d\n"        lprintf "EXCEPTION %s  before connect to host %s:%d\n"
1391            (Printexc2.to_string e) (Unix.string_of_inet_addr host) port;            (Printexc2.to_string e) (Unix.string_of_inet_addr host) port;
1392        raise e        raise e
1393          
1394    
1395    
1396  (*************************************************************************)  (*************************************************************************)
# Line 1421  let peer_addr t = Line 1419  let peer_addr t =
1419        match Unix.getpeername fd with        match Unix.getpeername fd with
1420          Unix.ADDR_INET (ip, port) ->          Unix.ADDR_INET (ip, port) ->
1421            let ip = Ip.of_inet_addr ip in            let ip = Ip.of_inet_addr ip in
1422            t.peer_addr <- Some (ip, port);            t.peer_addr <- Some (ip, port);
1423            ip, port            ip, port
1424    | _ -> raise Not_found    | _ -> raise Not_found
1425    
1426  let peer_ip t = fst (peer_addr t)  let peer_ip t = fst (peer_addr t)
1427  let peer_port t = snd (peer_addr t)  let peer_port t = snd (peer_addr t)
1428              
1429            (*            (*
1430  let host t =  let host t =
1431    let fd = fd t.sock_out in    let fd = fd t.sock_out in
# Line 1510  let set_max_input_buffer  t len = Line 1508  let set_max_input_buffer  t len =
1508    
1509  let set_max_output_buffer  t len =  let set_max_output_buffer  t len =
1510    t.wbuf.max_buf_size <- len    t.wbuf.max_buf_size <- len
1511      
1512  (*************************************************************************)  (*************************************************************************)
1513  (*                                                                       *)  (*                                                                       *)
1514  (*                         Compression                                   *)  (*                         Compression                                   *)
# Line 1618  let _ = Line 1616  let _ =
1616        reset_connection_scheduler ();        reset_connection_scheduler ();
1617        deflate_timer ();        deflate_timer ();
1618    );    );
1619      
1620    set_before_select_hook (fun _ ->    set_before_select_hook (fun _ ->
1621        schedule_connections ();        schedule_connections ();
1622        List.iter (fun bc ->        List.iter (fun bc ->
# Line 1636  let _ = Line 1634  let _ =
1634    );    );
1635    
1636  (*  (*
1637  Some ideas:  Some ideas:
1638    
1639  * sort the connections so that we try to read and write as  * sort the connections so that we try to read and write as
1640    much as possible to/from connections on which we already have read/written    much as possible to/from connections on which we already have read/written
1641    a lot.    a lot.
1642  * postpone reads and writes for a few seconds so that we have more to  * postpone reads and writes for a few seconds so that we have more to
1643    read/write.    read/write.
1644    
1645    dhcppc3:~#  ping computer.domain.org -f -s <packet_size> -c 1000    dhcppc3:~#  ping computer.domain.org -f -s <packet_size> -c 1000
1646   where computer.domain.org is a nearby computer on a 100 Mbs link.   where computer.domain.org is a nearby computer on a 100 Mbs link.
1647      
1648  On my cable link, making <packet_size> vary from 1 to 2000 shows:  On my cable link, making <packet_size> vary from 1 to 2000 shows:
1649    * packets > 1450 are 100% lost    * packets > 1450 are 100% lost
1650    * 300 > packets > 1 are 5% lost but exactly the same time (14.5 s)    * 300 > packets > 1 are 5% lost but exactly the same time (14.5 s)
# Line 1660  to an option 'packet_unit' (default 250) Line 1658  to an option 'packet_unit' (default 250)
1658  , pipe 6, ipg/ewma 14.205/0.000 ms  , pipe 6, ipg/ewma 14.205/0.000 ms
1659    
1660    *)    *)
1661      
1662      
1663    set_after_select_hook (fun _ ->    set_after_select_hook (fun _ ->
1664        List.iter (fun bc ->        List.iter (fun bc ->
1665            if bc.remaining_bytes > 0 then begin            if bc.remaining_bytes > 0 then begin
1666                  
1667                bc.connections <- List.sort (fun t1 t2 ->                bc.connections <- List.sort (fun t1 t2 ->
1668                    let w1 = t1.nread in                    let w1 = t1.nread in
1669                    let w2 = t2.nread in                    let w2 = t2.nread in
1670                    compare w1 w2                    compare w1 w2
1671                ) bc.connections;                ) bc.connections;
1672                  
1673                if !verbose_bandwidth > 2 && bc.nconnections > 0 then begin                if !verbose_bandwidth > 2 && bc.nconnections > 0 then begin
1674                    lprintf "[BW3 %6d] %d read-waiting connections for %d allowed\n" (last_time ()) bc.nconnections bc.remaining_bytes;                    lprintf "[BW3 %6d] %d read-waiting connections for %d allowed\n" (last_time ()) bc.nconnections bc.remaining_bytes;
1675                    List.iter (fun t ->                    List.iter (fun t ->
1676                        lprintf "[BW3 %6d]   %s:%d [buffered %d]\n" (last_time ()) t.name (sock_num t.sock_in) ((buf t).len);                        lprintf "[BW3 %6d]   %s:%d [buffered %d]\n" (last_time ()) t.name (sock_num t.sock_in) ((buf t).len);
1677                    ) bc.connections;                    ) bc.connections;
1678                  end;                                end;
1679                                
1680                  
1681                List.iter (fun t ->                List.iter (fun t ->
1682                    if bc.remaining_bytes > 0 then                    if bc.remaining_bytes > 0 then
1683                      let nconnections = maxi bc.nconnections 1 in                      let nconnections = maxi bc.nconnections 1 in
# Line 1689  to an option 'packet_unit' (default 250) Line 1687  to an option 'packet_unit' (default 250)
1687                      (try                      (try
1688  (*                    lprintf "allow to read %d\n" can_read; *)  (*                    lprintf "allow to read %d\n" can_read; *)
1689                          can_read_handler t t.sock_in can_read                          can_read_handler t t.sock_in can_read
1690                        with e ->                        with e ->
1691  (*                          lprintf "Exception %s in can_read_handler %s:%d\n"  (*                          lprintf "Exception %s in can_read_handler %s:%d\n"
1692                              (Printexc2.to_string e)                              (Printexc2.to_string e)
1693                            t.name (sock_num t.sock_in) *) ()                            t.name (sock_num t.sock_in) *) ()
# Line 1699  to an option 'packet_unit' (default 250) Line 1697  to an option 'packet_unit' (default 250)
1697                      bc.nconnections <- bc.nconnections - t.read_power                      bc.nconnections <- bc.nconnections - t.read_power
1698                    else                    else
1699                    if !verbose_bandwidth > 2 then begin                    if !verbose_bandwidth > 2 then begin
1700                        lprintf "[BW3 %6d]    %s:%d could not read\n" (last_time ())                        lprintf "[BW3 %6d]    %s:%d could not read\n" (last_time ())
1701                          t.name (sock_num t.sock_out)                                            t.name (sock_num t.sock_out)
1702                      end                      end
1703                ) bc.connections;                ) bc.connections;
1704  (*              if bc.remaining_bytes > 0 then bc.allow_io := false; *)  (*              if bc.remaining_bytes > 0 then bc.allow_io := false; *)
1705              end;              end;
1706              
1707            if !verbose_bandwidth > 2 then begin            if !verbose_bandwidth > 2 then begin
1708                  
1709                if bc.remaining_bytes > 0 then                if bc.remaining_bytes > 0 then
1710                  lprintf "[BW3 %6d] still %d bytes to read\n" (last_time ()) bc.remaining_bytes;                  lprintf "[BW3 %6d] still %d bytes to read\n" (last_time ()) bc.remaining_bytes;
1711                  
1712                if bc.nconnections > 0 then                if bc.nconnections > 0 then
1713                  lprintf "[BW3 %6d] still %d read-waiting connections after loop\n" (last_time ()) bc.nconnections;                  lprintf "[BW3 %6d] still %d read-waiting connections after loop\n" (last_time ()) bc.nconnections;
1714              end;              end;
1715              
1716              
1717            bc.connections <- [];            bc.connections <- [];
1718            bc.nconnections <- 0;            bc.nconnections <- 0;
1719        ) !read_bandwidth_controlers;        ) !read_bandwidth_controlers;
1720        List.iter (fun bc ->        List.iter (fun bc ->
1721            if bc.remaining_bytes > 0 then begin            if bc.remaining_bytes > 0 then begin
1722                  
1723                bc.connections <- List.sort (fun t1 t2 ->                bc.connections <- List.sort (fun t1 t2 ->
1724                    let w1 = remaining_to_write t1 in                    let w1 = remaining_to_write t1 in
1725                    let w2 = remaining_to_write t2 in                    let w2 = remaining_to_write t2 in
1726                    compare w1 w2                    compare w1 w2
1727                ) bc.connections;                ) bc.connections;
1728                  
1729                if !verbose_bandwidth > 2 && bc.nconnections > 0 then begin                if !verbose_bandwidth > 2 && bc.nconnections > 0 then begin
1730                    lprintf "[BW3 %6d] %d write-waiting connections for %d allowed\n" (last_time ()) bc.nconnections bc.remaining_bytes;                    lprintf "[BW3 %6d] %d write-waiting connections for %d allowed\n" (last_time ()) bc.nconnections bc.remaining_bytes;
1731                    List.iter (fun t ->                    List.iter (fun t ->
# Line 1750  to an option 'packet_unit' (default 250) Line 1748  to an option 'packet_unit' (default 250)
1748                      bc.nconnections <- bc.nconnections - t.write_power;                      bc.nconnections <- bc.nconnections - t.write_power;
1749                    else                    else
1750                    if !verbose_bandwidth > 2 then begin                    if !verbose_bandwidth > 2 then begin
1751                        lprintf "[BW3 %6d]    %s:%d could not write buffered %d bytes\n" (last_time ())                        lprintf "[BW3 %6d]    %s:%d could not write buffered %d bytes\n" (last_time ())
1752                          t.name (sock_num t.sock_out) (remaining_to_write t)                          t.name (sock_num t.sock_out) (remaining_to_write t)
1753                      end                      end
1754                ) bc.connections;                ) bc.connections;
1755  (*          if bc.remaining_bytes > 0 then bc.allow_io := false; *)  (*          if bc.remaining_bytes > 0 then bc.allow_io := false; *)
1756              end;              end;
1757            if !verbose_bandwidth > 2 then begin            if !verbose_bandwidth > 2 then begin
1758                  
1759                if bc.remaining_bytes > 0 then begin                if bc.remaining_bytes > 0 then begin
1760                    lprintf "[BW3 %6d] still %d bytes to write\n" (last_time ()) bc.remaining_bytes;                    lprintf "[BW3 %6d] still %d bytes to write\n" (last_time ()) bc.remaining_bytes;
1761                    List.iter (fun t ->                    List.iter (fun t ->
1762                        let len = remaining_to_write t in                        let len = remaining_to_write t in
1763                        if len > 0 then                        if len > 0 then
1764                          lprintf "[BW3 %6d]    %s:%d could write %d\n" (last_time ())                          lprintf "[BW3 %6d]    %s:%d could write %d\n" (last_time ())
1765                            t.name (sock_num t.sock_out) len                            t.name (sock_num t.sock_out) len
1766                    ) bc.connections                    ) bc.connections
1767                  end;                  end;
1768                  
1769                if bc.nconnections > 0 then                if bc.nconnections > 0 then
1770                  lprintf "[BW3 %6d] still %d write-waiting connections after loop\n" (last_time ()) bc.nconnections;                  lprintf "[BW3 %6d] still %d write-waiting connections after loop\n" (last_time ()) bc.nconnections;
1771              end;              end;
# Line 1775  to an option 'packet_unit' (default 250) Line 1773  to an option 'packet_unit' (default 250)
1773            bc.nconnections <- 0;            bc.nconnections <- 0;
1774        ) !write_bandwidth_controlers        ) !write_bandwidth_controlers
1775    )    )
1776      
1777  let prevent_close t =  let prevent_close t =
1778    prevent_close t.sock_in;      prevent_close t.sock_in;
1779    prevent_close t.sock_out    prevent_close t.sock_out
1780      
1781  let must_write t bool = must_write t.sock_out bool  let must_write t bool = must_write t.sock_out bool
1782  let output_buffered t = t.wbuf.len  let output_buffered t = t.wbuf.len
1783    
# Line 1808  let join_stats titles values = Line 1806  let join_stats titles values =
1806                end;                end;
1807        ) fields_titles fields_values        ) fields_titles fields_values
1808    | _ -> assert false    | _ -> assert false
1809      
1810  let load_stats filename =  let load_stats filename =
1811    try    try
1812      let ic = open_in filename in      let ic = open_in filename in
# Line 1819  let load_stats filename = Line 1817  let load_stats filename =
1817        let values = input_line ic in        let values = input_line ic in
1818        join_stats titles values;        join_stats titles values;
1819        iter_first ic        iter_first ic
1820        
1821      in      in
1822      try      try
1823        iter_first ic        iter_first ic
1824      with      with
1825        End_of_file -> close_in ic        End_of_file -> close_in ic
1826      | e ->      | e ->
1827          lprintf "[BWS] Error %s reading %s\n" (Printexc2.to_string e) filename;          lprintf "[BWS] Error %s reading %s\n" (Printexc2.to_string e) filename;
1828          close_in ic; proc_net_fs := false          close_in ic; proc_net_fs := false
1829    with    with
1830    | e ->    | e ->
1831        lprintf "[BWS] Error %s opening %s\n" (Printexc2.to_string e) filename;        lprintf "[BWS] Error %s opening %s\n" (Printexc2.to_string e) filename;
1832        proc_net_fs := false        proc_net_fs := false
1833          
1834  let proc_net_timer _ =  let proc_net_timer _ =
1835    if !proc_net_fs && !verbose_bandwidth > 0 then begin    if !proc_net_fs && !verbose_bandwidth > 0 then begin
1836        load_stats "/proc/net/netstat";        load_stats "/proc/net/netstat";
1837        load_stats "/proc/net/snmp";        load_stats "/proc/net/snmp";
1838      end      end
1839        
1840      
1841  let _ =  let _ =
1842    Heap.add_memstat "tcpBufferedSocket" (fun level buf ->    Heap.add_memstat "tcpBufferedSocket" (fun level buf ->
1843        Printf.bprintf buf "  %d latencies\n" (Hashtbl.length latencies);        Printf.bprintf buf "  %d latencies\n" (Hashtbl.length latencies);
1844        Printf.bprintf buf "  %d entries in net_stats\n" (Hashtbl.length net_stats);        Printf.bprintf buf "  %d entries in net_stats\n" (Hashtbl.length net_stats);
1845    );    );
1846    add_infinite_timer 1.0 proc_net_timer    add_infinite_timer 1.0 proc_net_timer
1847    
1848  let packet_frame_size = ref 250  let packet_frame_size = ref 250

Legend:
Removed from v.1.27  
changed lines
  Added in v.1.28

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