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