src/networks/donkey/donkeyShare.ml

   1 (* Copyright 2001, 2002 b8_bavard, b8_fee_carabine, INRIA *)
   2 (*
   3     This file is part of mldonkey.
   4
   5     mldonkey is free software; you can redistribute it and/or modify
   6     it under the terms of the GNU General Public License as published by
   7     the Free Software Foundation; either version 2 of the License, or
   8     (at your option) any later version.
   9
  10     mldonkey is distributed in the hope that it will be useful,
  11     but WITHOUT ANY WARRANTY; without even the implied warranty of
  12     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13     GNU General Public License for more details.
  14
  15     You should have received a copy of the GNU General Public License
  16     along with mldonkey; if not, write to the Free Software
  17     Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  18 *)
  19
  20 open Int64ops
  21 open CommonGlobals
  22 open Printf2
  23 open Md4
  24 open CommonDownloads
  25 open CommonFile
  26 open CommonShared
  27 open CommonTypes
  28 open Options
  29 open BasicSocket
  30 open TcpBufferedSocket
  31 open DonkeyMftp
  32 open DonkeyImport
  33 open DonkeyProtoCom
  34 open DonkeyTypes
  35 open DonkeyOptions
  36 open CommonOptions
  37 open DonkeyComplexOptions
  38 open DonkeyGlobals
  39
  40 module VB = VerificationBitmap
  41
  42 let must_share_file file codedname has_old_impl =
  43   match file.file_shared with
  44   | Some _ -> ()
  45   | None ->
  46       let full_name = file_disk_name file in
  47       check_magic (as_file file);
  48       let impl = {
  49           impl_shared_update = 1;
  50           impl_shared_fullname = full_name;
  51           impl_shared_codedname = codedname;
  52           impl_shared_size = file_size file;
  53           impl_shared_id = file.file_md4;
  54           impl_shared_num = 0;
  55           impl_shared_uploaded = Int64.zero;
  56           impl_shared_ops = shared_ops;
  57           impl_shared_val = file;
  58           impl_shared_requests = 0;
  59           impl_shared_file = Some (as_file file);
  60           impl_shared_servers = []
  61         } in
  62       file.file_shared <- Some impl;
  63       incr CommonGlobals.nshared_files;
  64       CommonShared.shared_calculate_total_bytes ();
  65       match has_old_impl with
  66         None -> update_shared_num impl
  67       | Some old_impl -> replace_shared old_impl impl
  68
  69 let new_file_to_share sh codedname old_impl =
  70   try
  71 (* How do we compute the total MD4 of the file ? *)
  72
  73     let md4s = sh.sh_md4s in
  74     let md4 = match Array.length md4s with
  75         1 -> md4s.(0)
  76       | 0 -> assert false
  77       | _ -> md4_of_array md4s
  78     in
  79
  80     if !verbose_share then
  81       lprintf_nl "Sharing file with MD4: %s" (Md4.to_string md4);
  82
  83     let user = CommonUserDb.admin_user () in
  84     let file = new_file sh.sh_name FileShared md4 sh.sh_size
  85         "" false user user.user_default_group in
  86     must_share_file file codedname old_impl;
  87     file.file_computed_md4s <- md4s;
  88     add_file_filenames (as_file file) (Filename.basename sh.sh_name);
  89     update_best_name file;
  90 (*  file.file_chunks <- Array.make file.file_nchunks PresentVerified; *)
  91 (*    file.file_absent_chunks <- []; *)
  92 (*    file.file_all_chunks <- String.make file.file_nchunks '1'; *)
  93     (* Should we trust mtimes, or reverify each file.  If we trust
  94      * mtimes, I guess we have to call
  95      * CommonSwarming.set_chunks_verified_bitmap "333..."
  96      * this seems unspeakably ugly, but the alternative is to reverify
  97      * every shared file every hour.
  98      *
  99      * If, however, we could somehow avoid regenerating shared files
 100      * when the directory is scanned, verifying everything on startup
 101      * might be acceptable.
 102      *
 103      * Also, timestamps would be more resilient if we took the maximum
 104      * of mtime and ctime.  (Touch will set ctime to the current time
 105      * regardless of the mtime being set.)
 106      *)
 107     match file.file_swarmer with
 108     | Some s ->
 109         let len = Array.length md4s in
 110         let ver_str = String.make len (VB.state_to_char VB.State_verified) in
 111         CommonSwarming.set_chunks_verified_bitmap s
 112           (VB.of_string ver_str);
 113      (*
 114      CommonSwarming.set_present s [(Int64.zero, file_size file)];
 115      (* If we don't verify now, it will never happen! *)
 116      CommonSwarming.verify_all_blocks s true;
 117      *)
 118         if !verbose_share then
 119           lprintf_nl "verified map of %s = %s"
 120             (codedname) (VB.to_string (CommonSwarming.chunks_verified_bitmap s))
 121     | None ->
 122         if !verbose_share then
 123           lprintf_nl "no swarmer for %s" codedname;
 124         (try
 125           file.file_format <- CommonMultimedia.get_info
 126             (file_disk_name file)
 127         with _ -> ());
 128     (*
 129     (try
 130         DonkeyOvernet.publish_file file
 131       with e ->
 132           lprintf "DonkeyOvernet.publish_file: %s" (Printexc2.to_string e);
 133 lprint_newline ());
 134   *)
 135     if !verbose_share then
 136       lprintf_nl "new_file_to_share: Sharing %s" sh.sh_name;
 137   with e ->
 138       lprintf_nl "Exception %s while sharing %s" (Printexc2.to_string e)
 139       sh.sh_name
 140
 141 let all_shared () =
 142   let shared_files = ref [] in
 143   Hashtbl.iter (fun md4 file ->
 144       match file.file_shared with
 145         None -> ()
 146       | Some _ ->  shared_files := file :: !shared_files
 147   ) files_by_md4;
 148   if !verbose_share then lprintf_nl "scanned shared files, %d files found" (List.length !shared_files);
 149   !shared_files
 150
 151 (* publish shared files to servers, called once per minute *)
 152 let send_new_shared () =
 153 (* sort list to publish least published files first *)
 154   let ( |> ) x f = f x in
 155   let all_shared =
 156     all_shared ()
 157     |> List.map (fun e ->
 158         (match e.file_shared with
 159           Some s -> List.length s.impl_shared_servers
 160          | _ -> 0) , e)
 161     |> List.sort (fun (a,_) (b,_) -> compare a b)
 162     |> List.map snd
 163   in
 164
 165 (* iter through connected servers *)
 166   List.iter (fun s ->
 167
 168 (* publish files only on master servers and do not publish more files than hard limit allows *)
 169     if s.server_master &&
 170       (match s.server_hard_limit with
 171           Some v when (Int64.to_int v) < List.length s.server_sent_shared -> false
 172         | _ -> true) then
 173
 174 (* iter through all shared files and check if the file is already published on the current server
 175    build a list of files_to_send with yet unpublished files *)
 176       begin
 177         let files_to_send = ref [] in
 178         let can_publish f = not (file_is_largefile f && not s.server_has_largefiles) in
 179         List.iter (fun f ->
 180           match f.file_shared with
 181             Some impl ->
 182               if not (List.mem (CommonServer.as_server s.server_server) impl.impl_shared_servers)
 183                 && List.length !files_to_send < !!max_published_files && can_publish f then
 184               files_to_send := f :: !files_to_send
 185               else
 186                 if not (can_publish f) then
 187                   lprintf_nl "Can not publish largefile %s because server %s does not support largefiles"
 188                     (file_best_name f) (string_of_server s)
 189             | _ -> () (* this case never happens *)
 190         ) all_shared;
 191
 192         if !files_to_send <> [] then
 193           begin
 194             if !verbose_share || !verbose then
 195               lprintf_nl "publishing %d new files to %s (holds %d files)"
 196                 (List.length !files_to_send) (string_of_server s)
 197                 (List.length s.server_sent_shared);
 198
 199 (* publish files on server *)
 200             do_if_connected s.server_sock (fun sock ->
 201               server_send_share s.server_has_zlib sock !files_to_send);
 202
 203 (* append new published files to server structure *)
 204             s.server_sent_shared <- !files_to_send @ s.server_sent_shared;
 205
 206 (* iter through published files and append current server *)
 207             List.iter (fun file ->
 208               match file.file_shared with
 209                 Some impl -> impl.impl_shared_servers <-
 210                   impl.impl_shared_servers @ [(CommonServer.as_server s.server_server)]
 211               | _ -> ()) !files_to_send
 212
 213         end
 214       end
 215   ) (logged_in_servers ());
 216
 217 (*
 218 The problem: sh.shared_fd might be closed during the execution of the
 219 thread. Moreover, we don't want to open all the filedescs for all the
 220 files being shared !
 221 *)
 222
 223 exception Wrong_file_size of int64 * int64
 224
 225 let computing = ref false
 226
 227 (*   Compute (at most) one MD4 chunk if needed. *)
 228 let rec check_shared_files () =
 229   let module M = CommonHasher in
 230   if not !computing then
 231     match !shared_files with
 232       [] -> ()
 233     | sh :: files ->
 234         let rec job_creater _ =
 235           try
 236             if not (Sys.file_exists sh.shared_name) then
 237                 raise Not_found;
 238             if Unix32.getsize sh.shared_name <> sh.shared_size then
 239                 raise (Wrong_file_size ((Unix32.getsize sh.shared_name), sh.shared_size));
 240
 241             computing := true;
 242
 243             let end_pos = sh.shared_pos ++ block_size in
 244             let end_pos = min end_pos sh.shared_size in
 245             let len = end_pos -- sh.shared_pos in
 246
 247             if !verbose_md4 then
 248               lprintf_nl "Hashing chunk %d: %Ld-%Ld (%Ld) of %s"
 249                 sh.shared_chunk sh.shared_pos end_pos len sh.shared_name;
 250
 251             M.compute_md4 (Unix32.filename sh.shared_fd) sh.shared_pos len
 252               (fun job ->
 253                 computing := false;
 254                 if job.M.job_error then begin
 255                   lprintf_nl "Error prevent sharing %s" sh.shared_name
 256                 end else
 257                   let new_md4 = job.M.job_result in
 258
 259                   sh.shared_list <- new_md4 :: sh.shared_list;
 260                   sh.shared_pos <- end_pos;
 261                   sh.shared_chunk <- sh.shared_chunk + 1;
 262
 263                   if sh.shared_chunk = get_nchunks sh.shared_size then begin
 264                     let s = {
 265                         sh_name = sh.shared_name;
 266                         sh_size = sh.shared_size;
 267                         sh_md4s = Array.of_list (List.rev sh.shared_list);
 268                         sh_mtime = Unix32.mtime sh.shared_name;
 269                     } in
 270                     lprintf_nl "New shared file %s" sh.shared_name;
 271                     Hashtbl.add shared_files_info
 272                       (s.sh_name, s.sh_size, s.sh_mtime) s;
 273                     known_shared_files =:= s :: !!known_shared_files;
 274                     shared_files := files;
 275                     new_file_to_share s sh.shared_shared.impl_shared_codedname (Some sh.shared_shared);
 276                   end
 277                 else
 278                   job_creater ();
 279                 (* only try back-to-back hashing if hashing is
 280                    handled by a separate thread *)
 281                 if BasicSocket.has_threads () then
 282                   check_shared_files ()
 283             )
 284           with
 285             Wrong_file_size (real,computed) ->
 286             shared_files := files;
 287             lprintf_nl "Computed filesize %Ld does not match physical filesize %Ld, %s not shared"
 288               computed real sh.shared_name
 289           | e ->
 290             shared_files := files;
 291             lprintf_nl "Exception %s prevents sharing of %s"
 292               (Printexc2.to_string e) sh.shared_name
 293         in
 294         job_creater ()
 295
 296 let _ =
 297   network.op_network_share <- (fun fullname codedname size ->
 298       if !verbose_share then
 299         lprintf_nl "op_network_share: Sharing %s" fullname;
 300       try
 301 (*
 302 lprintf "Searching %s" fullname; lprint_newline ();
 303 *)
 304         let mtime = Unix32.mtime fullname in
 305
 306         let s = Hashtbl.find shared_files_info
 307             (fullname, size, mtime) in
 308         (* if s.sh_mtime = mtime && s.sh_size = size then begin *)
 309             if !verbose_share then
 310                 lprintf_nl "donkeyShare: Using old MD4s for %s" fullname;
 311             new_file_to_share s codedname None
 312 (*          end else begin
 313             if !verbose_share then begin
 314                 lprintf "Shared file %s has been modified\n" fullname;
 315               end;
 316             Hashtbl.remove shared_files_info fullname;
 317             known_shared_files =:= List2.removeq s !!known_shared_files;
 318             raise Not_found
 319           end *)
 320       with Not_found ->
 321           if !verbose_share then
 322               lprintf_nl "donkeyShare: No info on %s" fullname;
 323
 324     let found = ref false in
 325     List.iter (fun sh -> if sh.shared_name = fullname then found := true) !shared_files;
 326     if not !found then begin
 327           let rec impl = {
 328               impl_shared_update = 1;
 329               impl_shared_fullname = fullname;
 330               impl_shared_codedname = codedname;
 331               impl_shared_size = size;
 332               impl_shared_num = 0;
 333               impl_shared_uploaded = Int64.zero;
 334               impl_shared_ops = pre_shared_ops;
 335               impl_shared_id = Md4.null;
 336               impl_shared_val = pre_shared;
 337               impl_shared_requests = 0;
 338               impl_shared_file = None;
 339               impl_shared_servers = [];
 340             } and
 341             pre_shared = {
 342               shared_shared = impl;
 343               shared_name = fullname;
 344               shared_size = size;
 345               shared_list = [];
 346               shared_pos = 0L;
 347               shared_chunk = 0;
 348               shared_fd = Unix32.create_ro fullname;
 349             } in
 350           update_shared_num impl;
 351           shared_files := pre_shared :: !shared_files;
 352     end
 353   )
 354
 355 let remember_shared_info file new_name =
 356 (* changed 2.5.24: normalization of file names is done in
 357     CommonShared.new_shared, so it should be done here too. *)
 358   let new_name = Filename2.normalize new_name in
 359 (*  lprintf "******  remember_shared_info for new file %s\n" new_name; *)
 360   if file.file_computed_md4s <> [||] then
 361     try
 362       let mtime =
 363         try
 364           let disk_name = file_disk_name file in
 365           Unix32.mtime disk_name
 366         with _ ->
 367             if !verbose then
 368               lprintf_nl "Share: Trying mtime on new name %s, disk_name %s"
 369           new_name (file_disk_name file);
 370             Unix32.mtime new_name
 371       in
 372
 373       if !verbose_share then
 374           lprintf_nl "Remember %s" new_name;
 375
 376       let size = file_size file in
 377       if not (Hashtbl.mem shared_files_info
 378             (new_name, size, mtime) ) then
 379         let s = {
 380             sh_name = new_name;
 381             sh_size = size;
 382             sh_mtime = mtime;
 383             sh_md4s = file.file_computed_md4s;
 384           } in
 385
 386         known_shared_files =:= s :: !!known_shared_files;
 387         Hashtbl.add shared_files_info (new_name, size, mtime) s
 388     with e ->
 389         lprintf_nl "Exception %s in remember_shared_info"
 390           (Printexc2.to_string e)
 391
 392 let must_share_file file = must_share_file file (file_best_name file) None