/[mldonkey]/mldonkey/src/daemon/common/commonUploads.ml
ViewVC logotype

Diff of /mldonkey/src/daemon/common/commonUploads.ml

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

revision 1.27 by spiralvoice, Mon Jun 20 18:56:44 2005 UTC revision 1.28 by spiralvoice, Thu Jul 28 18:20:51 2005 UTC
# Line 33  open TcpBufferedSocket Line 33  open TcpBufferedSocket
33  open CommonGlobals  open CommonGlobals
34  open CommonOptions  open CommonOptions
35    
36  (*  (*
37  PROBLEMS: most of the time, users won't share their files on all networks.  PROBLEMS: most of the time, users won't share their files on all networks.
38  We should provide a different directory than incoming/, where files  We should provide a different directory than incoming/, where files
39  would be shared, per directory ?  would be shared, per directory ?
# Line 43  and each network would be configured to Line 43  and each network would be configured to
43  directories. Moreover, we should have a different sharing strategy.  directories. Moreover, we should have a different sharing strategy.
44    
45  Default would be: share all files greater than 1 MB in incoming/ on Edonkey.  Default would be: share all files greater than 1 MB in incoming/ on Edonkey.
46      
47  *)  *)
48    
49  (*******************************************************************  (*******************************************************************
50    
     
51                           TYPES                           TYPES
52    
     
53  *******************************************************************)  *******************************************************************)
54    
55  let ed2k_block_size = Int64.of_int 9728000  let ed2k_block_size = Int64.of_int 9728000
56  let tiger_block_size = Int64.of_int (1024 * 1024)  let tiger_block_size = Int64.of_int (1024 * 1024)
57      
58  type shared_file = {  type shared_file = {
59      shared_codedname : string;      shared_codedname : string;
60      shared_info : Store.index;      shared_info : Store.index;
61      shared_fd : Unix32.t;      shared_fd : Unix32.t;
62      shared_format : CommonTypes.format;      shared_format : CommonTypes.format;
63      shared_impl : shared_file shared_impl;      shared_impl : shared_file shared_impl;
64      mutable shared_uids_wanted :      mutable shared_uids_wanted :
65      (file_uid_id * (shared_file -> Uid.t -> unit)) list;      (file_uid_id * (shared_file -> Uid.t -> unit)) list;
66    }    }
67    
# Line 74  and shared_info = { Line 72  and shared_info = {
72      mutable shared_tiger : TigerTree.t array;      mutable shared_tiger : TigerTree.t array;
73      mutable shared_bitprint : Uid.t option;      mutable shared_bitprint : Uid.t option;
74      mutable shared_mtime : float;      mutable shared_mtime : float;
75      mutable shared_uids : Uid.t list;          mutable shared_uids : Uid.t list;
76      mutable shared_id : int;      mutable shared_id : int;
77    }    }
78      
79  and shared_tree =  and shared_tree =
80    {    {
81      shared_dirname : string;      shared_dirname : string;
82      mutable shared_files : shared_file list;      mutable shared_files : shared_file list;
83      mutable shared_dirs : (string * shared_tree) list;      mutable shared_dirs : (string * shared_tree) list;
84    }      }
85    
86  type local_search = {  type local_search = {
87      mutable local_search_results : (shared_file * shared_info) list;      mutable local_search_results : (shared_file * shared_info) list;
88      mutable local_search_query : query;      mutable local_search_query : query;
89    }    }
90      
91  module IndexingSharedFiles = struct  module IndexingSharedFiles = struct
92    
93      let store_name = "shared_store"      let store_name = "shared_store"
94    
95      let search_query s = s.local_search_query      let search_query s = s.local_search_query
96          
97      type search = local_search      type search = local_search
98      type result = shared_info      type result = shared_info
99          
100      let result_names sh = [sh.shared_fullname]      let result_names sh = [sh.shared_fullname]
101      let result_size sh = sh.shared_size      let result_size sh = sh.shared_size
102      let result_uids sh = []      let result_uids sh = []
103      let result_tags sh = []      let result_tags sh = []
104          
105  (* We should probably directly use the Store.index here so that all  (* We should probably directly use the Store.index here so that all
106  shared_infos are stored on disk. *)  shared_infos are stored on disk. *)
107      type stored_result = Store.index      type stored_result = Store.index
108      let result_index r = r      let result_index r = r
109          
110    end    end
111    
112  module IndexedSharedFiles = CommonIndexing.Make(IndexingSharedFiles)  module IndexedSharedFiles = CommonIndexing.Make(IndexingSharedFiles)
113      
     
114  (*************************************************************************)  (*************************************************************************)
115  (*                                                                       *)  (*                                                                       *)
116  (*                         SAVED SHARED FILES                            *)  (*                         SAVED SHARED FILES                            *)
117  (*                                                                       *)  (*                                                                       *)
118  (*************************************************************************)  (*************************************************************************)
119    
     
120  module SharedFileOption = struct  module SharedFileOption = struct
121        
122      let get_value assocs name conv =      let get_value assocs name conv =
123        try conv (List.assoc name assocs)        try conv (List.assoc name assocs)
124        with _ -> failwith (Printf.sprintf "Bad shared file %s" name)        with _ -> failwith (Printf.sprintf "Bad shared file %s" name)
125        
126      let value_to_info v =      let value_to_info v =
127        match v with        match v with
128          Options.Module assocs ->          Options.Module assocs ->
129            let sh_md4s = get_value assocs "md4s"            let sh_md4s = get_value assocs "md4s"
# Line 142  module SharedFileOption = struct Line 138  module SharedFileOption = struct
138                (value_to_list (fun v ->                (value_to_list (fun v ->
139                    Uid.of_string (value_to_string v)))                    Uid.of_string (value_to_string v)))
140            in            in
141              
142            let sh_size = get_value assocs "size"  value_to_int64 in            let sh_size = get_value assocs "size"  value_to_int64 in
143            let sh_name = get_value assocs "name" value_to_filename in            let sh_name = get_value assocs "name" value_to_filename in
144            let sh_mtime = get_value assocs "mtime" value_to_float in            let sh_mtime = get_value assocs "mtime" value_to_float in
# Line 154  module SharedFileOption = struct Line 150  module SharedFileOption = struct
150                | _ -> ()                | _ -> ()
151            ) sh_uids;            ) sh_uids;
152    
153            { shared_fullname = sh_name;            { shared_fullname = sh_name;
154              shared_mtime = sh_mtime;              shared_mtime = sh_mtime;
155              shared_size = sh_size;              shared_size = sh_size;
156              shared_md4s = sh_md4s;              shared_md4s = sh_md4s;
157              shared_tiger = sh_ttr;              shared_tiger = sh_ttr;
158              shared_bitprint = !sh_bitprint;              shared_bitprint = !sh_bitprint;
159              shared_uids = sh_uids;              shared_uids = sh_uids;
160              shared_id = 0;              shared_id = 0;
161            }            }
162              
163        | _ -> failwith "Options: not a shared file info option"        | _ -> failwith "Options: not a shared file info option"
164          
165      let info_to_value info =      let info_to_value info =
166        Options.Module [        Options.Module [
167          "name", filename_to_value info.shared_fullname;          "name", filename_to_value info.shared_fullname;
# Line 175  module SharedFileOption = struct Line 171  module SharedFileOption = struct
171          "ttr", array_to_value TigerTree.hash_to_value info.shared_tiger;          "ttr", array_to_value TigerTree.hash_to_value info.shared_tiger;
172          "uids", list_to_value (fun v ->          "uids", list_to_value (fun v ->
173              string_to_value (Uid.to_string v)) info.shared_uids;              string_to_value (Uid.to_string v)) info.shared_uids;
174        ]            ]
175        
176      let t = define_option_class "SharedFile" value_to_info info_to_value      let t = define_option_class "SharedFile" value_to_info info_to_value
177    end    end
178    
       
179  let shared_ini = create_options_file "shared_files.ini"  let shared_ini = create_options_file "shared_files.ini"
180    
181  let shared_section = file_section shared_ini [] ""  let shared_section = file_section shared_ini [] ""
182    
183  let old_shared_files = define_option shared_section  let old_shared_files = define_option shared_section
184      ["shared_files"] ""      ["shared_files"] ""
185      (list_option SharedFileOption.t) []      (list_option SharedFileOption.t) []
186    
187  let infos_by_name = Hashtbl.create 113  let infos_by_name = Hashtbl.create 113
188      
189    let lprintf_nl () =
190      lprintf "%s[cUp] "
191        (log_time ()); lprintf_nl2
192    
193  let _ =  let _ =
194    set_after_load_hook shared_ini (fun _ ->    set_after_load_hook shared_ini (fun _ ->
195        List.iter (fun info ->        List.iter (fun info ->
# Line 209  let _ = Line 208  let _ =
208    
209  let load () = try Options.load shared_ini with _ -> ()  let load () = try Options.load shared_ini with _ -> ()
210  let save () = Options.save shared_ini  let save () = Options.save shared_ini
211      
212  (*************************************************************************)  (*************************************************************************)
213  (*                                                                       *)  (*                                                                       *)
214  (*                         NETWORK                                       *)  (*                         NETWORK                                       *)
215  (*                                                                       *)  (*                                                                       *)
216  (*************************************************************************)  (*************************************************************************)
217      
218  let network = CommonNetwork.new_network "GS" "Global Shares"  let network = CommonNetwork.new_network "GS" "Global Shares"
219      [ VirtualNetwork ]      [ VirtualNetwork ]
220    
221  let _ =  let _ =
222    network.op_network_connected <- (fun _ -> false);    network.op_network_connected <- (fun _ -> false);
223    network.op_network_is_enabled <- (fun _ -> raise IgnoreNetwork);    network.op_network_is_enabled <- (fun _ -> raise IgnoreNetwork);
224    network.op_network_info <- (fun _ -> raise Not_found);    network.op_network_info <- (fun _ -> raise Not_found);
225    network.op_network_info <- (fun n ->    network.op_network_info <- (fun n ->
226        {        {
227          network_netnum = network.network_num;          network_netnum = network.network_num;
228          network_config_filename = (match network.network_config_file with          network_config_filename = (match network.network_config_file with
229              [] -> "" | opfile :: _ -> options_file_name opfile);              [] -> "" | opfile :: _ -> options_file_name opfile);
# Line 236  let _ = Line 235  let _ =
235          network_connected = 0;          network_connected = 0;
236        });        });
237    network.op_network_connected_servers <- (fun _ -> [])    network.op_network_connected_servers <- (fun _ -> [])
238      
239  let (shared_ops : shared_file CommonShared.shared_ops) =  let (shared_ops : shared_file CommonShared.shared_ops) =
240    CommonShared.new_shared_ops network    CommonShared.new_shared_ops network
241      
     
     
242  let waiting_shared_files = ref []  let waiting_shared_files = ref []
243  let shareds_by_uid = Hashtbl.create 13  let shareds_by_uid = Hashtbl.create 13
244  let shareds_by_id = Hashtbl.create 13  let shareds_by_id = Hashtbl.create 13
# Line 249  let shareds_by_id = Hashtbl.create 13 Line 246  let shareds_by_id = Hashtbl.create 13
246  let add_by_uid uid sh =  let add_by_uid uid sh =
247    let urn = Uid.to_string uid in    let urn = Uid.to_string uid in
248    Hashtbl.add shareds_by_uid uid sh    Hashtbl.add shareds_by_uid uid sh
249      
250  let find_by_uid uid =  let find_by_uid uid =
251    let urn = Uid.to_string uid in    let urn = Uid.to_string uid in
252    Hashtbl.find shareds_by_uid uid    Hashtbl.find shareds_by_uid uid
253      
254      module SharedFilesIndex = IndexedSharedFiles.MakeIndex (struct
255  module SharedFilesIndex = IndexedSharedFiles.MakeIndex (struct        let add_search_result s sh =
       let add_search_result s sh =  
256          let r = Hashtbl.find shareds_by_id sh.shared_id in          let r = Hashtbl.find shareds_by_id sh.shared_id in
257          s.local_search_results <- (r, sh) :: s.local_search_results          s.local_search_results <- (r, sh) :: s.local_search_results
258      end)      end)
259    
     
260  let current_job = ref None  let current_job = ref None
261                  
262  (*******************************************************************  (*******************************************************************
263    
     
264                        DATA STRUCTURES                        DATA STRUCTURES
265    
     
266  *******************************************************************)  *******************************************************************)
267      
268  let shareds_counter = ref 1  let shareds_counter = ref 1
269  let shared_counter = ref (Int64.zero)  let shared_counter = ref (Int64.zero)
270  let shared_files = Hashtbl.create 13  let shared_files = Hashtbl.create 13
271      
272  let new_shared_dir dirname = {  let new_shared_dir dirname = {
273      shared_dirname = dirname;      shared_dirname = dirname;
274      shared_files = [];      shared_files = [];
# Line 286  let shared_tree = new_shared_dir "" Line 279  let shared_tree = new_shared_dir ""
279    
280  (*******************************************************************  (*******************************************************************
281    
     
282                      HASHES COMPUTATION                      HASHES COMPUTATION
283    
     
284  *******************************************************************)  *******************************************************************)
285      
     
286  let md4_of_list md4s =  let md4_of_list md4s =
287    let len = List.length md4s in    let len = List.length md4s in
288    let s = String.create (len * 16) in    let s = String.create (len * 16) in
# Line 324  let rec tiger_of_array array pos block = Line 314  let rec tiger_of_array array pos block =
314    let t = Tiger.string s in    let t = Tiger.string s in
315    let t = TigerTree.direct_of_string (Tiger.direct_to_string t) in    let t = TigerTree.direct_of_string (Tiger.direct_to_string t) in
316    t    t
317      
318  let rec tiger_max_block_size block len =  let rec tiger_max_block_size block len =
319    if block >= len then block    if block >= len then block
320    else tiger_max_block_size (block*2) len    else tiger_max_block_size (block*2) len
321      
322  let tiger_of_array array =  let tiger_of_array array =
323    tiger_of_array array 0 (tiger_max_block_size  1 (Array.length array))    tiger_of_array array 0 (tiger_max_block_size  1 (Array.length array))
324    
# Line 349  let rec tiger_pos2 nblocks = Line 339  let rec tiger_pos2 nblocks =
339    let pos = half + pos in    let pos = half + pos in
340    pos, list    pos, list
341    
342  let tiger_node d1 d2 =  let tiger_node d1 d2 =
343    let s = String.create (1 + Tiger.length * 2) in    let s = String.create (1 + Tiger.length * 2) in
344    s.[0] <- '\001';    s.[0] <- '\001';
345    String.blit (TigerTree.direct_to_string d1) 0 s 1 Tiger.length;    String.blit (TigerTree.direct_to_string d1) 0 s 1 Tiger.length;
# Line 357  let tiger_node d1 d2 = Line 347  let tiger_node d1 d2 =
347    let t = Tiger.string s in    let t = Tiger.string s in
348    let t = TigerTree.direct_of_string (Tiger.direct_to_string t) in    let t = TigerTree.direct_of_string (Tiger.direct_to_string t) in
349    t    t
350      
351  let rec tiger_tree s array pos block =  let rec tiger_tree s array pos block =
352    if block = 1 then    if block = 1 then
353      array.(pos)      array.(pos)
# Line 369  let rec tiger_tree s array pos block = Line 359  let rec tiger_tree s array pos block =
359    let d1 = tiger_tree s array pos (block/2) in    let d1 = tiger_tree s array pos (block/2) in
360    let d2 = tiger_tree s array (pos+block/2) (block/2) in    let d2 = tiger_tree s array (pos+block/2) (block/2) in
361    tiger_node d1 d2    tiger_node d1 d2
362      
363  let rec fill_tiger_tree s list =  let rec fill_tiger_tree s list =
364    match list with    match list with
365      [] -> ()      [] -> ()
# Line 389  let rec fill_tiger_tree s list = Line 379  let rec fill_tiger_tree s list =
379        if acc = 1 then s.(pos+half) <- s.(next_pos+2*half);        if acc = 1 then s.(pos+half) <- s.(next_pos+2*half);
380        fill_tiger_tree s tail        fill_tiger_tree s tail
381    
382  let flatten_tiger_array array =  let flatten_tiger_array array =
383    let len = Array.length array in    let len = Array.length array in
384    let s = String.create ( len * TigerTree.length) in      let s = String.create ( len * TigerTree.length) in
385    for i = 0 to len - 1 do    for i = 0 to len - 1 do
386      String.blit (TigerTree.direct_to_string array.(i)) 0      String.blit (TigerTree.direct_to_string array.(i)) 0
387        s (i * TigerTree.length) TigerTree.length        s (i * TigerTree.length) TigerTree.length
388    done;    done;
389    s    s
390    
391  let unflatten_tiger_array s =  let unflatten_tiger_array s =
392    let len = String.length s / TigerTree.length in    let len = String.length s / TigerTree.length in
393    let array = Array.create len TigerTree.null in      let array = Array.create len TigerTree.null in
394    for i = 0 to len - 1 do    for i = 0 to len - 1 do
395      array.(i) <- TigerTree.direct_of_string      array.(i) <- TigerTree.direct_of_string
396        (String.sub s (i * TigerTree.length) TigerTree.length)        (String.sub s (i * TigerTree.length) TigerTree.length)
397    done;    done;
398    array    array
399      
400  let make_tiger_tree array =  let make_tiger_tree array =
401    let len = Array.length array in    let len = Array.length array in
402    let pos, list = tiger_pos2 len in    let pos, list = tiger_pos2 len in
# Line 416  let make_tiger_tree array = Line 406  let make_tiger_tree array =
406    done;    done;
407    fill_tiger_tree s list;    fill_tiger_tree s list;
408    flatten_tiger_array s    flatten_tiger_array s
409      
410  let build_tiger_tree_file uid ttr =  let build_tiger_tree_file uid ttr =
411    let s = make_tiger_tree ttr in    let s = make_tiger_tree ttr in
412    Unix2.safe_mkdir "ttr";    Unix2.safe_mkdir "ttr";
413    Unix2.can_write_to_directory "ttr";    Unix2.can_write_to_directory "ttr";
414    File.from_string (Filename.concat "ttr" (Uid.to_file_string uid)) s    File.from_string (Filename.concat "ttr" (Uid.to_file_string uid)) s
415      
416  let rec start_job_for sh (wanted_id, handler) =  let rec start_job_for sh (wanted_id, handler) =
417    let info = IndexedSharedFiles.get_result sh.shared_info in    let info = IndexedSharedFiles.get_result sh.shared_info in
418    try    try
419      List.iter (fun id ->      List.iter (fun id ->
420          match wanted_id,Uid.to_uid id with          match wanted_id,Uid.to_uid id with
421            BITPRINT, Bitprint _            BITPRINT, Bitprint _
422          | SHA1, Sha1 _          | SHA1, Sha1 _
423          | ED2K, Ed2k _          | ED2K, Ed2k _
424          | MD5, Md5 _          | MD5, Md5 _
425          | MD5EXT, Md5Ext _          | MD5EXT, Md5Ext _
426          | TIGER, TigerTree _          | TIGER, TigerTree _
427            -> (try handler sh id with _ -> ()); raise Exit            -> (try handler sh id with _ -> ()); raise Exit
428          | _ -> ()          | _ -> ()
429      ) info.shared_uids;      ) info.shared_uids;
430        
431      match wanted_id with      match wanted_id with
432        SHA1 ->        SHA1 ->
433          begin          begin
434            try            try
435              CommonHasher.compute_sha1 (Unix32.filename sh.shared_fd)              CommonHasher.compute_sha1 (Unix32.filename sh.shared_fd)
436              zero info.shared_size (fun job ->              zero info.shared_size (fun job ->
437                  if job.CommonHasher.job_error then begin                  if job.CommonHasher.job_error then begin
438                      lprintf "Error during hashing of %s\n" info.shared_fullname;                      lprintf_nl () "Error during hashing of %s" info.shared_fullname;
439                      current_job := None;                      current_job := None;
440                    end else                    end else
441                    begin                    begin
# Line 454  let rec start_job_for sh (wanted_id, han Line 444  let rec start_job_for sh (wanted_id, han
444                      let uid = Uid.create (Sha1 sha1) in                      let uid = Uid.create (Sha1 sha1) in
445                      info.shared_uids <- uid :: info.shared_uids;                      info.shared_uids <- uid :: info.shared_uids;
446                      IndexedSharedFiles.update_result sh.shared_info info;                      IndexedSharedFiles.update_result sh.shared_info info;
447                        
448                      add_by_uid uid sh;                      add_by_uid uid sh;
449                      start_job_for sh (wanted_id, handler)                        start_job_for sh (wanted_id, handler)
450                    end                    end
451              );              );
452            with e ->            with e ->
453                current_job := None;                current_job := None;
454                raise e                              raise e
455          end              end
456            
457      | BITPRINT ->      | BITPRINT ->
458          let sha1 = ref None in          let sha1 = ref None in
459          let tiger = ref None in          let tiger = ref None in
# Line 480  let rec start_job_for sh (wanted_id, han Line 470  let rec start_job_for sh (wanted_id, han
470                info.shared_uids <- uid :: info.shared_uids;                info.shared_uids <- uid :: info.shared_uids;
471                info.shared_bitprint <- Some uid;                info.shared_bitprint <- Some uid;
472                IndexedSharedFiles.update_result sh.shared_info info;                IndexedSharedFiles.update_result sh.shared_info info;
473                  
474                add_by_uid uid sh;                add_by_uid uid sh;
475                  
476                build_tiger_tree_file uid info.shared_tiger;                build_tiger_tree_file uid info.shared_tiger;
477                  
478                start_job_for sh (wanted_id, handler)                start_job_for sh (wanted_id, handler)
479              
480            | _ -> ()            | _ -> ()
481  (*  (*
482  (* Not enough information to compute the bitprint. Ask for the corresponding  (* Not enough information to compute the bitprint. Ask for the corresponding
# Line 501  computation ??? *) Line 491  computation ??? *)
491  | _ -> ());  | _ -> ());
492    *)    *)
493          end          end
494        
495      | MD5EXT ->      | MD5EXT ->
496          let md5ext =            let md5ext =
497            try            try
498              let fd = Unix32.create_rw info.shared_fullname in              let fd = Unix32.create_rw info.shared_fullname in
499              let file_size = Unix32.getsize64 fd false in              let file_size = Unix32.getsize64 fd false in
# Line 521  computation ??? *) Line 511  computation ??? *)
511          IndexedSharedFiles.update_result sh.shared_info info;          IndexedSharedFiles.update_result sh.shared_info info;
512    
513          add_by_uid uid sh;          add_by_uid uid sh;
514          start_job_for sh (wanted_id, handler)            start_job_for sh (wanted_id, handler)
515        
516      | ED2K ->      | ED2K ->
517          let size = info.shared_size in          let size = info.shared_size in
518          let chunk_size = ed2k_block_size  in          let chunk_size = ed2k_block_size  in
# Line 534  computation ??? *) Line 524  computation ??? *)
524                  pos (min (size -- pos) chunk_size)                  pos (min (size -- pos) chunk_size)
525                (fun job ->                (fun job ->
526                    if job.CommonHasher.job_error then begin                    if job.CommonHasher.job_error then begin
527                        lprintf "Error during hashing of %s\n" info.shared_fullname;                        lprintf_nl () "Error during hashing of %s" info.shared_fullname;
528                        current_job := None;                        current_job := None;
529                      end else begin                      end else begin
530                          
531                        iter (pos ++ chunk_size) (job.CommonHasher.job_result :: hashes)                        iter (pos ++ chunk_size) (job.CommonHasher.job_result :: hashes)
532                      end)                      end)
533              with e ->              with e ->
# Line 553  computation ??? *) Line 543  computation ??? *)
543            IndexedSharedFiles.update_result sh.shared_info info;            IndexedSharedFiles.update_result sh.shared_info info;
544    
545            add_by_uid uid sh;            add_by_uid uid sh;
546            start_job_for sh (wanted_id, handler)                            start_job_for sh (wanted_id, handler)
547          in          in
548          iter zero []          iter zero []
549        
550      | TIGER ->      | TIGER ->
551            
552          if TigerTree.enabled then          if TigerTree.enabled then
553    
554            let size = info.shared_size in            let size = info.shared_size in
# Line 570  computation ??? *) Line 560  computation ??? *)
560                    pos (min (size -- pos) chunk_size)                    pos (min (size -- pos) chunk_size)
561                (fun job ->                (fun job ->
562                      if job.CommonHasher.job_error then begin                      if job.CommonHasher.job_error then begin
563                        lprintf "Error during hashing of %s\n"                        lprintf_nl () "Error during hashing of %s"
564                        info.shared_fullname;                        info.shared_fullname;
565                          current_job := None;                          current_job := None;
566                        end else begin                        end else begin
567                          iter (pos ++ chunk_size)                          iter (pos ++ chunk_size)
568                          (job.CommonHasher.job_result :: hashes)                          (job.CommonHasher.job_result :: hashes)
569                        end)                        end)
570              else              else
# Line 586  computation ??? *) Line 576  computation ??? *)
576              IndexedSharedFiles.update_result sh.shared_info info;              IndexedSharedFiles.update_result sh.shared_info info;
577    
578              add_by_uid uid sh;              add_by_uid uid sh;
579              start_job_for sh (wanted_id, handler)                              start_job_for sh (wanted_id, handler)
580            in            in
581            iter zero []            iter zero []
582              
583      | _ -> raise Exit      | _ -> raise Exit
584        
585    with Exit ->    with Exit ->
586        current_job := None        current_job := None
587    | e -> current_job := None; raise e      | e -> current_job := None; raise e
588      
589  let shared_files_timer _ =  let shared_files_timer _ =
590    match !current_job with    match !current_job with
591    | Some _ -> ()    | Some _ -> ()
# Line 607  let shared_files_timer _ = Line 597  let shared_files_timer _ =
597              [] ->  waiting_shared_files := tail;              [] ->  waiting_shared_files := tail;
598            | uid :: tail ->            | uid :: tail ->
599                if !verbose_share then                if !verbose_share then
600                  lprintf "shared_files_timer: starting job\n";                  lprintf_nl () "shared_files_timer: starting job";
601                sh.shared_uids_wanted <- tail;                sh.shared_uids_wanted <- tail;
602                current_job := Some sh;                current_job := Some sh;
603                start_job_for sh uid                start_job_for sh uid
# Line 618  let ask_for_uid sh uid f = Line 608  let ask_for_uid sh uid f =
608    
609  (*******************************************************************  (*******************************************************************
610    
     
611                        FUNCTIONS                        FUNCTIONS
612    
     
613  *******************************************************************)  *******************************************************************)
614    
615  let rec add_shared_file node sh dir_list =  let rec add_shared_file node sh dir_list =
# Line 652  let new_info full_name size = Line 640  let new_info full_name size =
640          IndexedSharedFiles.remove_result index;          IndexedSharedFiles.remove_result index;
641          raise Not_found;          raise Not_found;
642        end;        end;
643        
644      info.shared_id <- !shareds_counter;      info.shared_id <- !shareds_counter;
645      IndexedSharedFiles.update_result index info;      IndexedSharedFiles.update_result index info;
646      info, index      info, index
# Line 663  let new_info full_name size = Line 651  let new_info full_name size =
651            shared_mtime = mtime;            shared_mtime = mtime;
652            shared_md4s = [||];            shared_md4s = [||];
653            shared_tiger = [||];            shared_tiger = [||];
654            shared_bitprint = None;                      shared_bitprint = None;
655            shared_size = size;            shared_size = size;
656            shared_id = !shareds_counter;            shared_id = !shareds_counter;
657          } in          } in
658        let index =IndexedSharedFiles.add info in        let index =IndexedSharedFiles.add info in
659        Hashtbl.add infos_by_name full_name index;        Hashtbl.add infos_by_name full_name index;
660        info, index        info, index
661          
662  let add_shared full_name codedname size =  let add_shared full_name codedname size =
663    try    try
664      Hashtbl.find shared_files codedname      Hashtbl.find shared_files codedname
665    with Not_found ->    with Not_found ->
666          
667        let fd = Unix32.create_ro full_name in        let fd = Unix32.create_ro full_name in
668    
669        let info, index = new_info full_name size in        let info, index = new_info full_name size in
670          
671        let rec impl = {        let rec impl = {
672            impl_shared_update = 1;            impl_shared_update = 1;
673            impl_shared_fullname = full_name;            impl_shared_fullname = full_name;
# Line 691  let add_shared full_name codedname size Line 679  let add_shared full_name codedname size
679            impl_shared_ops = shared_ops;            impl_shared_ops = shared_ops;
680            impl_shared_val = sh;            impl_shared_val = sh;
681            impl_shared_requests = 0;            impl_shared_requests = 0;
682          }          }
683        and sh = {        and sh = {
684            shared_info = index;            shared_info = index;
685            shared_codedname = codedname;            shared_codedname = codedname;
# Line 700  let add_shared full_name codedname size Line 688  let add_shared full_name codedname size
688            shared_impl = impl;            shared_impl = impl;
689            shared_uids_wanted = [];            shared_uids_wanted = [];
690          } in          } in
691          
692        update_shared_num impl;        update_shared_num impl;
693          
694  (*      lprintf "FILE ADDED: %s\n" codedname;  *)  (*      lprintf "FILE ADDED: %s\n" codedname;  *)
695        Hashtbl.add shared_files codedname sh;        Hashtbl.add shared_files codedname sh;
696        Hashtbl.add shareds_by_id info.shared_id sh;        Hashtbl.add shareds_by_id info.shared_id sh;
697          
698        List.iter (fun uid -> add_by_uid uid sh) info.shared_uids;        List.iter (fun uid -> add_by_uid uid sh) info.shared_uids;
699          
700        SharedFilesIndex.add sh.shared_info;        SharedFilesIndex.add sh.shared_info;
701        add_shared_file shared_tree sh (String2.split codedname '/');        add_shared_file shared_tree sh (String2.split codedname '/');
702        shared_counter := Int64.add !shared_counter size;        shared_counter := Int64.add !shared_counter size;
# Line 719  let iter f = Line 707  let iter f =
707    Hashtbl.iter (fun _ sh ->    Hashtbl.iter (fun _ sh ->
708        f sh        f sh
709    ) shared_files    ) shared_files
710          
711  let query q =  let query q =
712    let s = {    let s = {
713      local_search_query = q;      local_search_query = q;
714      local_search_results = []      local_search_results = []
715      } in      } in
716    SharedFilesIndex.find s;    SharedFilesIndex.find s;
717    s.local_search_results    s.local_search_results
718      
719  let find_by_name name = Hashtbl.find shared_files name  let find_by_name name = Hashtbl.find shared_files name
720      
721  (*let find_by_num num = Hashtbl.find table num *)  (*let find_by_num num = Hashtbl.find table num *)
722    
723  (**********************************************************************  (**********************************************************************
724    
   
725                       UPLOAD SCHEDULER                       UPLOAD SCHEDULER
726    
   
727  ***********************************************************************)  ***********************************************************************)
728    
729  let client_is_connected c = is_connected (client_state c)  let client_is_connected c = is_connected (client_state c)
     
   
 (* Move the uploaders and nu commands to driver *)  
   
   
   
   
   
   
730    
731  let upload_clients = (Fifo.create () : client Fifo.t)  let upload_clients = (Fifo.create () : client Fifo.t)
732    
   
733  let (pending_slots_map : client Intmap.t ref) = ref Intmap.empty  let (pending_slots_map : client Intmap.t ref) = ref Intmap.empty
734  (* let (pending_slots_fifo : int Fifo.t)  = Fifo.create () *)  (* let (pending_slots_fifo : int Fifo.t)  = Fifo.create () *)
735    
736  let remaining_bandwidth = ref 0      let remaining_bandwidth = ref 0
737  let total_bandwidth = ref 0      let total_bandwidth = ref 0
738  let complete_bandwidth = ref 0  let complete_bandwidth = ref 0
739  let counter = ref 1      let counter = ref 1
740  let sent_bytes = Array.create 10 0  let sent_bytes = Array.create 10 0
741  let has_upload = ref 0  let has_upload = ref 0
742  let upload_credit = ref 0  let upload_credit = ref 0
# Line 768  let upload_credit = ref 0 Line 744  let upload_credit = ref 0
744    
745  let can_write_len sock len =  let can_write_len sock len =
746    let bool1 = can_write_len sock len in    let bool1 = can_write_len sock len in
747    let upload_rate =    let upload_rate =
748          (if !!max_hard_upload_rate = 0 then 10000 else !!max_hard_upload_rate)          (if !!max_hard_upload_rate = 0 then 10000 else !!max_hard_upload_rate)
749      * 1024 in      * 1024 in
750    let bool2 =    let bool2 =
751      (      (
752  (*      lprintf "upload_rate %d -> %d\n" upload_rate  (*      lprintf "upload_rate %d -> %d\n" upload_rate
753          (upload_rate * (Fifo.length upload_clients)); *)          (upload_rate * (Fifo.length upload_clients)); *)
754  (* changed 2.5.24  (* changed 2.5.24
755  Don't put in a socket more than 10 seconds of upload.  Don't put in a socket more than 10 seconds of upload.
756    *)    *)
757      not_buffer_more sock (upload_rate * 10  (* * (Fifo.length upload_clients) *) ))      not_buffer_more sock (upload_rate * 10  (* * (Fifo.length upload_clients) *) ))
758    in    in
# Line 796  upload_clients; *) Line 772  upload_clients; *)
772        let c = Fifo.take upload_clients in        let c = Fifo.take upload_clients in
773        client_can_upload c  !remaining_bandwidth        client_can_upload c  !remaining_bandwidth
774      end else      end else
775    let per_client =    let per_client =
776      let len = Fifo.length upload_clients in      let len = Fifo.length upload_clients in
777      if len * 10000 < !remaining_bandwidth then      if len * 10000 < !remaining_bandwidth then
778  (* Each client in the Fifo can receive 10000 bytes.  (* Each client in the Fifo can receive 10000 bytes.
# Line 822  let rec next_uploads () = Line 798  let rec next_uploads () =
798    if !remaining_bandwidth < old_remaining_bandwidth then    if !remaining_bandwidth < old_remaining_bandwidth then
799      next_uploads ()      next_uploads ()
800    
801  let next_uploads () =  let next_uploads () =
802    sent_bytes.(!counter-1) <- sent_bytes.(!counter-1) - !remaining_bandwidth;    sent_bytes.(!counter-1) <- sent_bytes.(!counter-1) - !remaining_bandwidth;
803    (*    (*
804    if !verbose_upload then begin    if !verbose_upload then begin
805        lprintf "Left %d\n" !remaining_bandwidth;        lprintf "Left %d\n" !remaining_bandwidth;
806      end; *)      end; *)
807    complete_bandwidth := !complete_bandwidth + !remaining_bandwidth;    complete_bandwidth := !complete_bandwidth + !remaining_bandwidth;
808    incr counter;    incr counter;
809    if !counter = 11 then begin    if !counter = 11 then begin
810        counter := 1;        counter := 1;
811        total_bandwidth :=        total_bandwidth :=
812        (if !!max_hard_upload_rate = 0 then 10000 * 1024        (if !!max_hard_upload_rate = 0 then 10000 * 1024
813          else (maxi (!!max_hard_upload_rate - 1) 1) * 1024 );          else (maxi (!!max_hard_upload_rate - 1) 1) * 1024 );
814        complete_bandwidth := !total_bandwidth;        complete_bandwidth := !total_bandwidth;
815  (*      lprintf "Init to %d\n" !total_bandwidth;  *)  (*      lprintf "Init to %d\n" !total_bandwidth;  *)
816        remaining_bandwidth := 0                  remaining_bandwidth := 0
817      end;      end;
818      
819    let last_sec = ref 0 in    let last_sec = ref 0 in
820    for i = 0 to 9 do    for i = 0 to 9 do
821      last_sec := !last_sec + sent_bytes.(i)      last_sec := !last_sec + sent_bytes.(i)
# Line 852  let next_uploads () = Line 828  let next_uploads () =
828        for i = 0 to 9 do        for i = 0 to 9 do
829          lprintf "    last[%d] = %d\n" i  sent_bytes.(i)          lprintf "    last[%d] = %d\n" i  sent_bytes.(i)
830        done; *)        done; *)
831          
832      end; *)        end; *)
833    remaining_bandwidth := mini (mini (mini    remaining_bandwidth := mini (mini (mini
834          (maxi (!remaining_bandwidth + !total_bandwidth / 10) 10000)          (maxi (!remaining_bandwidth + !total_bandwidth / 10) 10000)
835        !total_bandwidth) !complete_bandwidth)        !total_bandwidth) !complete_bandwidth)
836    (!total_bandwidth - !last_sec);    (!total_bandwidth - !last_sec);
837    complete_bandwidth := !complete_bandwidth - !remaining_bandwidth;    complete_bandwidth := !complete_bandwidth - !remaining_bandwidth;
838  (*  lprintf "Remaining %d[%d]\n" !remaining_bandwidth !complete_bandwidth;  *)  (*  lprintf "Remaining %d[%d]\n" !remaining_bandwidth !complete_bandwidth;  *)
839    sent_bytes.(!counter-1) <- !remaining_bandwidth;    sent_bytes.(!counter-1) <- !remaining_bandwidth;
840    if !remaining_bandwidth > 0 then    if !remaining_bandwidth > 0 then
841      next_uploads ()      next_uploads ()
842    
843  let reset_upload_timer () = ()  let reset_upload_timer () = ()
844        
845  let reset_upload_timer _ =  let reset_upload_timer _ =
846    reset_upload_timer ()    reset_upload_timer ()
847      
848  let upload_credit_timer _ =  let upload_credit_timer _ =
849    if !has_upload = 0 then    if !has_upload = 0 then
850      (if !upload_credit < 300 then incr upload_credit)      (if !upload_credit < 300 then incr upload_credit)
851    else    else
852      decr has_upload      decr has_upload
853        
854  let ready_for_upload c =  let ready_for_upload c =
855    Fifo.put upload_clients c    Fifo.put upload_clients c
856        
857  let add_pending_slot c =  let add_pending_slot c =
858    if client_has_a_slot c then begin    if client_has_a_slot c then
859        if !verbose_upload then lprintf "Avoided inserting an uploader in pending slots!\n";      if !verbose_upload then lprintf_nl () "Avoided inserting an uploader in pending slots!"
860      end    else
   else  
861    if not (Intmap.mem (client_num c) !pending_slots_map) then    if not (Intmap.mem (client_num c) !pending_slots_map) then
862      begin      begin
863  (* This is useless since it is the goal of the pending_slots_map  (* This is useless since it is the goal of the pending_slots_map
864          else if Fifo.mem pending_slots_fifo (client_num c) then begin          else if Fifo.mem pending_slots_fifo (client_num c) then begin
865          lprintf "Avoided inserting a client twice in pending slots\n";          lprintf "Avoided inserting a client twice in pending slots\n";
866                    
867        end else *)        end else *)
868        pending_slots_map := Intmap.add (client_num c) c !pending_slots_map;        pending_slots_map := Intmap.add (client_num c) c !pending_slots_map;
869      end      end
870        
871  let remove_pending_slot c =  let remove_pending_slot c =
872    if Intmap.mem (client_num c) !pending_slots_map then    if Intmap.mem (client_num c) !pending_slots_map then
873      pending_slots_map := Intmap.remove (client_num c) !pending_slots_map      pending_slots_map := Intmap.remove (client_num c) !pending_slots_map
# Line 906  let rec give_a_slot c = Line 881  let rec give_a_slot c =
881        set_client_has_a_slot c true;        set_client_has_a_slot c true;
882        client_enter_upload_queue c        client_enter_upload_queue c
883      end      end
884        
885  and find_pending_slot () =  and find_pending_slot () =
886    try    try
887      let rec iter () =      let rec iter () =
# Line 922  let add_pending_slot c = Line 897  let add_pending_slot c =
897    let cprio = ref (shared_prio csh) in    let cprio = ref (shared_prio csh) in
898    (* if cdir <> "" then    (* if cdir <> "" then
899      lprintf "Testing cdir %s\n" cdir; *)      lprintf "Testing cdir %s\n" cdir; *)
900    Intmap.iter (fun _ c ->    Intmap.iter (fun _ c ->
901      let sh = client_upload c in      let sh = client_upload c in
902      if shared_dir sh = cdir then decr cprio      if shared_dir sh = cdir then decr cprio
903    ) !CommonClient.uploaders;    ) !CommonClient.uploaders;
# Line 941  let static_refill_upload_slots () = Line 916  let static_refill_upload_slots () =
916    let len = Intmap.length !CommonClient.uploaders in    let len = Intmap.length !CommonClient.uploaders in
917    if len < !!max_upload_slots then find_pending_slot ()    if len < !!max_upload_slots then find_pending_slot ()
918    
919  (* Since dynamic slots allocation is based on feedback, it should not  (* Since dynamic slots allocation is based on feedback, it should not
920   * allocate new slots too fast, since connections need some time to reach   * allocate new slots too fast, since connections need some time to reach
921   * a stable state.   * a stable state.
922   * To compensate for that slow pace, slots are allocated quadratically   * To compensate for that slow pace, slots are allocated quadratically
923   * as long as the link is not saturated.   * as long as the link is not saturated.
924   *)   *)
925    
926  let not_saturated_count = ref 0  let not_saturated_count = ref 0
927  let allocation_cluster = ref 1  let allocation_cluster = ref 1
928      
929  let dynamic_refill_upload_slots () =  let dynamic_refill_upload_slots () =
930    let reset_state () =    let reset_state () =
931      not_saturated_count := 0;      not_saturated_count := 0;
# Line 958  let dynamic_refill_upload_slots () = Line 933  let dynamic_refill_upload_slots () =
933    
934    let open_slots n =    let open_slots n =
935      let i = ref n in      let i = ref n in
936      if !verbose_upload then begin      if !verbose_upload then
937        lprintf "try to allocate %d more slots\n" n;        lprintf_nl () "try to allocate %d more slots" n;
     end;  
938      while !i > 0 do      while !i > 0 do
939        find_pending_slot ();        find_pending_slot ();
940        decr i        decr i
# Line 970  let dynamic_refill_upload_slots () = Line 944  let dynamic_refill_upload_slots () =
944    let min_upload_slots = 3 in    let min_upload_slots = 3 in
945  (*  let estimated_capacity = !!max_hard_upload_rate * 1024 in *)  (*  let estimated_capacity = !!max_hard_upload_rate * 1024 in *)
946    let estimated_capacity = detected_uplink_capacity () in    let estimated_capacity = detected_uplink_capacity () in
947    if !verbose_upload then begin    if !verbose_upload then
948      lprintf "usage: %d(%d) capacity: %d\n"      lprintf_nl () "usage: %d(%d) capacity: %d"
949        (short_delay_upload_usage ())        (short_delay_upload_usage ())
950        (upload_usage ())        (upload_usage ())
951        estimated_capacity;        estimated_capacity;
   end;  
952    let len = Intmap.length !CommonClient.uploaders in    let len = Intmap.length !CommonClient.uploaders in
953    if len < !!max_upload_slots then begin    if len < !!max_upload_slots then begin
954    
955  (* enough free bw for another slot *)  (* enough free bw for another slot *)
956      if short_delay_upload_usage () + slot_bw < estimated_capacity then begin      if short_delay_upload_usage () + slot_bw < estimated_capacity then begin
957        if !verbose_upload then begin        if !verbose_upload then
958          lprintf "uplink not fully used\n";          lprintf_nl () "uplink not fully used";
       end;  
959        incr not_saturated_count        incr not_saturated_count
960      end else reset_state ();      end else reset_state ();
961              
962      if len < min_upload_slots then begin      if len < min_upload_slots then begin
963        if !verbose_upload then begin        if !verbose_upload then
964          lprintf "too few upload slots\n";          lprintf_nl () "too few upload slots";
       end;  
965        open_slots (min_upload_slots - len);        open_slots (min_upload_slots - len);
966        reset_state ()        reset_state ()
967      end else if !not_saturated_count >= 2 then begin      end else if !not_saturated_count >= 2 then begin
# Line 1033  let consume_bandwidth len = Line 1004  let consume_bandwidth len =
1004    remaining_bandwidth := !remaining_bandwidth - len    remaining_bandwidth := !remaining_bandwidth - len
1005    
1006  let remaining_bandwidth () = !remaining_bandwidth  let remaining_bandwidth () = !remaining_bandwidth
     
     
1007    
1008  (**********************************************************************  (**********************************************************************
1009    
   
1010                       DOWNLOAD SCHEDULER                       DOWNLOAD SCHEDULER
1011    
   
1012  ***********************************************************************)  ***********************************************************************)
1013    
1014  let download_credit = ref 0  let download_credit = ref 0
1015  let download_fifo = Fifo.create ()  let download_fifo = Fifo.create ()
1016        
1017  let download_engine () =  let download_engine () =
1018    if not (Fifo.empty download_fifo) then begin    if not (Fifo.empty download_fifo) then begin
1019        let credit = !!max_hard_download_rate in        let credit = !!max_hard_download_rate in
1020        let credit = 2 * (if credit = 0 then 10000 else credit) in        let credit = 2 * (if credit = 0 then 10000 else credit) in
1021        download_credit := !download_credit + credit;        download_credit := !download_credit + credit;
1022        let rec iter () =        let rec iter () =
1023          if !download_credit > 0 && not (Fifo.empty download_fifo) then            if !download_credit > 0 && not (Fifo.empty download_fifo) then
1024            begin            begin
1025              (try              (try
1026                  let (f, len) = Fifo.take download_fifo in                  let (f, len) = Fifo.take download_fifo in
# Line 1066  let download_engine () = Line 1033  let download_engine () =
1033        iter ()        iter ()
1034      end      end
1035    
1036  let queue_download_request f len =    let queue_download_request f len =
1037    if !!max_hard_download_rate = 0 then    if !!max_hard_download_rate = 0 then
1038      f ()      f ()
1039    else    else
1040      Fifo.put download_fifo (f,len)          Fifo.put download_fifo (f,len)
1041    
1042    (* timer started every 1/10 seconds *)    (* timer started every 1/10 seconds *)
1043  let upload_download_timer () =  let upload_download_timer () =
1044    (try download_engine ()    (try download_engine ()
1045      with e ->      with e ->
1046          lprintf "Exception %s in download_engine\n"  (Printexc2.to_string e)          lprintf_nl () "Exception %s in download_engine" (Printexc2.to_string e)
1047    );    );
1048    (try next_uploads ()    (try next_uploads ()
1049    with e ->  lprintf "exc %s in upload\n" (Printexc2.to_string e))    with e -> lprintf_nl () "exc %s in upload" (Printexc2.to_string e))
1050                  
     
1051  let words_of_filename =  let words_of_filename =
1052    let extension_list = [    let extension_list = [
1053        "mp3" ; "avi" ; "jpg" ; "jpeg" ; "txt" ; "mov" ; "mpg" ; "ogm"        "mp3" ; "avi" ; "jpg" ; "jpeg" ; "txt" ; "mov" ; "mpg" ; "ogm"
1054      ]      ]
1055    in          in
1056    let rec remove_short list list2 =    let rec remove_short list list2 =
1057      match list with      match list with
1058        [] -> List.rev list2        [] -> List.rev list2
1059      | s :: list ->      | s :: list ->
1060          if List.mem s extension_list then          if List.mem s extension_list then
1061            remove_short list (s :: list2) else            remove_short list (s :: list2) else
1062            
1063          if String.length s < 5 then (* keywords should had list be 5 bytes *)          if String.length s < 5 then (* keywords should had list be 5 bytes *)
1064            remove_short list list2            remove_short list list2
1065          else          else
1066            remove_short list (s :: list2)            remove_short list (s :: list2)
1067    in    in
1068      
1069    let get_name_keywords file_name =    let get_name_keywords file_name =
1070      match remove_short (String2.stem file_name) [] with      match remove_short (String2.stem file_name) [] with
1071        [] | [_] ->        [] | [_] ->
1072          lprintf "Not enough keywords to recover %s\n" file_name;          lprintf_nl () "Not enough keywords to recover %s" file_name;
1073          [file_name]          [file_name]
1074      | l -> l      | l -> l
1075    in    in
1076    get_name_keywords    get_name_keywords
1077      
1078      open LittleEndian
 open LittleEndian    
1079    
1080  let _ =  let _ =
1081    CommonWeb.add_redirector_info "SHARED" (fun buf ->    CommonWeb.add_redirector_info "SHARED" (fun buf ->
1082        let module S = CommonShared in        let module S = CommonShared in
1083        let total_shared = ref Int64.zero in        let total_shared = ref Int64.zero in
1084        let total_uploaded = ref Int64.zero in        let total_uploaded = ref Int64.zero in
         
1085        S.shared_iter (fun s ->        S.shared_iter (fun s ->
1086            let i = S.as_shared_impl s in            let i = S.as_shared_impl s in
1087            total_uploaded :=            total_uploaded :=
1088            Int64.add !total_uploaded i.S.impl_shared_uploaded;            Int64.add !total_uploaded i.S.impl_shared_uploaded;
1089            total_shared :=            total_shared :=
1090            Int64.add !total_shared i.S.impl_shared_size            Int64.add !total_shared i.S.impl_shared_size
1091        );        );
         
1092        buf_int64 buf !total_shared;        buf_int64 buf !total_shared;
1093        buf_int64 buf !total_uploaded;        buf_int64 buf !total_uploaded;
         
1094    )    )

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