738 |
let (pending_slots_map : client Intmap.t ref) = ref Intmap.empty |
let (pending_slots_map : client Intmap.t ref) = ref Intmap.empty |
739 |
(* let (pending_slots_fifo : int Fifo.t) = Fifo.create () *) |
(* let (pending_slots_fifo : int Fifo.t) = Fifo.create () *) |
740 |
|
|
741 |
let remaining_bandwidth = ref 0 |
let packet_size = 10240 |
742 |
let total_bandwidth = ref 0 |
|
743 |
let complete_bandwidth = ref 0 |
let streaming_amount = packet_size * 2 |
744 |
let counter = ref 1 |
let streaming_left = ref streaming_amount |
745 |
let sent_bytes = Array.create 10 0 |
let streaming_time = (ref None : float option ref) |
746 |
|
|
747 |
let has_upload = ref 0 |
let has_upload = ref 0 |
748 |
let upload_credit = ref 0 |
let upload_credit = ref 0 |
749 |
|
|
750 |
|
|
751 |
let can_write_len sock len = |
let can_write_len sock len = |
752 |
let bool1 = can_write_len sock len in |
let bool1 = can_write_len sock len in |
753 |
let upload_rate = |
(* changed 2.5.24: Don't put in a socket more than 10 seconds of upload. *) |
754 |
(if !!max_hard_upload_rate = 0 then 10000 else !!max_hard_upload_rate) |
(* ...since the socket shares available bandwidth with others, |
755 |
* 1024 in |
that test has no physical meaning. Should it simply be removed ? *) |
756 |
let bool2 = |
let bool2 = not_buffer_more sock (int_of_float !CommonGlobals.payload_bandwidth * 10) in |
|
( |
|
|
(* lprintf "upload_rate %d -> %d\n" upload_rate |
|
|
(upload_rate * (Fifo.length upload_clients)); *) |
|
|
(* changed 2.5.24 |
|
|
Don't put in a socket more than 10 seconds of upload. |
|
|
*) |
|
|
not_buffer_more sock (upload_rate * 10 (* * (Fifo.length upload_clients) *) )) |
|
|
in |
|
757 |
(* lprintf "can_write_len %b %b\n" bool1 bool2; *) |
(* lprintf "can_write_len %b %b\n" bool1 bool2; *) |
758 |
let b = bool1 && bool2 in |
let b = bool1 && bool2 in |
759 |
(* if not b then |
(* if not b then |
760 |
lprintf "bool1 %b len %d bool2 %b upload %d\n" bool1 len bool2 upload_rate; |
lprintf "bool1 %b len %d bool2 %b upload %d\n" |
761 |
*) |
bool1 len bool2 upload_rate; *) |
762 |
b |
b |
763 |
|
|
764 |
let upload_to_one_client () = |
let next_uploads () = |
|
(* lprintf "upload_to_one_client %d %d\n" (Fifo.length upload_clients) |
|
|
!remaining_bandwidth; |
|
|
Fifo.iter (fun c -> lprintf " client %d\n" (client_num c)) |
|
|
upload_clients; *) |
|
|
if !remaining_bandwidth < 10000 then begin |
|
|
let c = Fifo.take upload_clients in |
|
|
client_can_upload c !remaining_bandwidth |
|
|
end else |
|
|
let per_client = |
|
|
let len = Fifo.length upload_clients in |
|
|
if len * 10000 < !remaining_bandwidth then |
|
|
(* Each client in the Fifo can receive 10000 bytes. |
|
|
Divide the bandwidth between the clients |
|
|
*) |
|
|
(!remaining_bandwidth / 10000 / len) * 10000 |
|
|
else mini 10000 !remaining_bandwidth in |
|
|
let c = Fifo.take upload_clients in |
|
|
client_can_upload c per_client |
|
|
|
|
|
let rec fifo_uploads n = |
|
|
if n>0 && !remaining_bandwidth > 0 then |
|
|
begin |
|
|
upload_to_one_client (); |
|
|
fifo_uploads (n-1) |
|
|
end |
|
765 |
|
|
766 |
let rec next_uploads () = |
let rec next_uploads_aux () = |
767 |
(* lprintf "next_uploads %d\n" !remaining_bandwidth; *) |
let rec next_uploads_round n = |
768 |
let old_remaining_bandwidth = !remaining_bandwidth in |
let upload_to_one_client max_amount = |
769 |
let len = Fifo.length upload_clients in |
let c = Fifo.take upload_clients in |
770 |
fifo_uploads len; |
client_can_upload c max_amount |
771 |
if !remaining_bandwidth < old_remaining_bandwidth then |
(* it's up to client_can_upload to put the client back into the Fifo *) |
772 |
next_uploads () |
in |
773 |
|
|
774 |
|
(* lprintf "next_uploads %d %d\n" |
775 |
|
(Fifo.length upload_clients) !streaming_left; |
776 |
|
Fifo.iter (fun c -> |
777 |
|
lprintf " client %d\n" (client_num c) |
778 |
|
) upload_clients; *) |
779 |
|
if n>0 && |
780 |
|
not (Fifo.empty upload_clients) && |
781 |
|
!streaming_left > 0 then begin |
782 |
|
upload_to_one_client packet_size; |
783 |
|
next_uploads_round (n-1) |
784 |
|
end in |
785 |
|
|
786 |
|
(* stop if no uploader could take anything during the last round *) |
787 |
|
let old_streaming_left = !streaming_left in |
788 |
|
next_uploads_round (Fifo.length upload_clients); |
789 |
|
if !streaming_left < old_streaming_left then |
790 |
|
next_uploads_aux () in |
791 |
|
|
|
let next_uploads () = |
|
|
sent_bytes.(!counter-1) <- sent_bytes.(!counter-1) - !remaining_bandwidth; |
|
792 |
(* |
(* |
793 |
if !verbose_upload then begin |
if !verbose_upload then begin |
794 |
lprintf "Left %d\n" !remaining_bandwidth; |
lprintf "streaming_left %d\n" !streaming_left; |
|
end; *) |
|
|
complete_bandwidth := !complete_bandwidth + !remaining_bandwidth; |
|
|
incr counter; |
|
|
if !counter = 11 then begin |
|
|
counter := 1; |
|
|
total_bandwidth := |
|
|
(if !!max_hard_upload_rate = 0 then 10000 * 1024 |
|
|
else (maxi (!!max_hard_upload_rate - 1) 1) * 1024 ); |
|
|
complete_bandwidth := !total_bandwidth; |
|
|
(* lprintf "Init to %d\n" !total_bandwidth; *) |
|
|
remaining_bandwidth := 0 |
|
|
end; |
|
|
|
|
|
let last_sec = ref 0 in |
|
|
for i = 0 to 9 do |
|
|
last_sec := !last_sec + sent_bytes.(i) |
|
|
done; |
|
|
|
|
|
(* if !verbose_upload then begin |
|
|
lprintf "last sec: %d/%d (left %d)\n" !last_sec !total_bandwidth |
|
|
(!total_bandwidth - !last_sec); |
|
|
(* |
|
|
for i = 0 to 9 do |
|
|
lprintf " last[%d] = %d\n" i sent_bytes.(i) |
|
|
done; *) |
|
|
|
|
795 |
end; *) |
end; *) |
796 |
remaining_bandwidth := mini (mini (mini |
(* buffer empties with time... *) |
797 |
(maxi (!remaining_bandwidth + !total_bandwidth / 10) 10000) |
let new_streaming_time = BasicSocket.current_time () in |
798 |
!total_bandwidth) !complete_bandwidth) |
let deltat = (match !streaming_time with |
799 |
(!total_bandwidth - !last_sec); |
| None -> 0. |
800 |
complete_bandwidth := !complete_bandwidth - !remaining_bandwidth; |
| Some t -> new_streaming_time -. t) in |
801 |
(* lprintf "Remaining %d[%d]\n" !remaining_bandwidth !complete_bandwidth; *) |
streaming_left := !streaming_left + |
802 |
sent_bytes.(!counter-1) <- !remaining_bandwidth; |
(int_of_float (!CommonGlobals.payload_bandwidth *. deltat)); |
803 |
if !remaining_bandwidth > 0 then |
if !streaming_left > streaming_amount then |
804 |
next_uploads () |
streaming_left := streaming_amount; |
805 |
|
streaming_time := Some new_streaming_time; |
806 |
|
next_uploads_aux () |
807 |
|
|
808 |
let reset_upload_timer () = () |
let reset_upload_timer () = () |
809 |
|
|
909 |
|
|
910 |
let slot_bw = 3072 in |
let slot_bw = 3072 in |
911 |
let min_upload_slots = 3 in |
let min_upload_slots = 3 in |
|
(* let estimated_capacity = !!max_hard_upload_rate * 1024 in *) |
|
912 |
let estimated_capacity = detected_uplink_capacity () in |
let estimated_capacity = detected_uplink_capacity () in |
913 |
|
let estimated_capacity = if !!max_hard_upload_rate = 0 then |
914 |
|
estimated_capacity |
915 |
|
else |
916 |
|
(* max_hard_upload_rate lowered manually,... *) |
917 |
|
mini estimated_capacity (!!max_hard_upload_rate * 1024) in |
918 |
if !verbose_upload then |
if !verbose_upload then |
919 |
lprintf_nl "[cUp] usage: %d(%d) capacity: %d" |
lprintf_nl "[cUp] usage: %d(%d) capacity: %d" |
920 |
(short_delay_upload_usage ()) |
(short_delay_upload_usage ()) |
972 |
end |
end |
973 |
|
|
974 |
let consume_bandwidth len = |
let consume_bandwidth len = |
975 |
remaining_bandwidth := !remaining_bandwidth - len |
streaming_left := !streaming_left - len |
|
|
|
|
let remaining_bandwidth () = !remaining_bandwidth |
|
976 |
|
|
977 |
(********************************************************************** |
(********************************************************************** |
978 |
|
|