1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178
|
(* Unison file synchronizer: src/transport.ml *)
(* Copyright 1999-2007 (see COPYING for details) *)
open Common
open Lwt
let debug = Trace.debug "transport"
(*****************************************************************************)
(* MAIN FUNCTIONS *)
(*****************************************************************************)
let fileSize uiFrom uiTo =
match uiFrom, uiTo with
_, Updates (File (props, ContentsUpdated (_, _, ress)), _) ->
(Props.length props, Osx.ressLength ress)
| Updates (_, Previous (`FILE, props, _, ress)),
(NoUpdates | Updates (File (_, ContentsSame), _)) ->
(Props.length props, Osx.ressLength ress)
| _ ->
assert false
let maxthreads =
Prefs.createInt "maxthreads" 20
"maximum number of simultaneous file transfers"
("This preference controls how much concurrency is allowed during"
^ " the transport phase. Normally, it should be set reasonably high "
^ "(default is 20) to maximize performance, but when Unison is used "
^ "over a low-bandwidth link it may be helpful to set it lower (e.g. "
^ "to 1) so that Unison doesn't soak up all the available bandwidth."
)
let actionReg = Lwt_util.make_region (Prefs.read maxthreads)
(* Logging for a thread: write a message before and a message after the
execution of the thread. *)
let logLwt (msgBegin: string)
(t: unit -> 'a Lwt.t)
(fMsgEnd: 'a -> string)
: 'a Lwt.t =
Trace.log msgBegin;
Lwt.bind (t ()) (fun v ->
Trace.log (fMsgEnd v);
Lwt.return v)
(* [logLwtNumbered desc t] provides convenient logging for a thread given a
description [desc] of the thread [t ()], generate pair of messages of the
following form in the log:
*
[BGN] <desc>
...
[END] <desc>
**)
let rLogCounter = ref 0
let logLwtNumbered (lwtDescription: string) (lwtShortDescription: string)
(t: unit -> 'a Lwt.t): 'a Lwt.t =
let _ = (rLogCounter := (!rLogCounter) + 1; !rLogCounter) in
let lwtDescription = Util.replacesubstring lwtDescription "\n " "" in
logLwt (Printf.sprintf "[BGN] %s\n" lwtDescription) t
(fun _ ->
Printf.sprintf "[END] %s\n" lwtShortDescription)
let stashCurrentVersionOnRoot: Common.root -> Path.t -> unit Lwt.t =
Remote.registerRootCmd
"stashCurrentVersion"
(fun (fspath, path) ->
Lwt.return (Stasher.stashCurrentVersion fspath (Update.translatePathLocal fspath path) None))
let stashCurrentVersions fromRoot toRoot path =
stashCurrentVersionOnRoot fromRoot path >>= (fun()->
stashCurrentVersionOnRoot toRoot path)
let doAction (fromRoot,toRoot) path fromContents toContents id =
Lwt_util.resize_region actionReg (Prefs.read maxthreads);
Lwt_util.run_in_region actionReg 1 (fun () ->
if not !Trace.sendLogMsgsToStderr then
Trace.statusDetail (Path.toString path);
Remote.Thread.unwindProtect (fun () ->
match fromContents, toContents with
(`ABSENT, _, _, _), (_, _, _, uiTo) ->
logLwtNumbered
("Deleting " ^ Path.toString path ^
"\n from "^ root2string toRoot)
("Deleting " ^ Path.toString path)
(fun () -> Files.delete fromRoot path toRoot path uiTo)
(* No need to transfer the whole directory/file if there were only
property modifications on one side. (And actually, it would be
incorrect to transfer a directory in this case.) *)
| (_, (`Unchanged | `PropsChanged), fromProps, uiFrom),
(_, (`Unchanged | `PropsChanged), toProps, uiTo) ->
logLwtNumbered
("Copying properties for " ^ Path.toString path
^ "\n from " ^ root2string fromRoot ^ "\n to " ^
root2string toRoot)
("Copying properties for " ^ Path.toString path)
(fun () ->
Files.setProp
fromRoot path toRoot path fromProps toProps uiFrom uiTo)
| (`FILE, _, _, uiFrom), (`FILE, _, _, uiTo) ->
logLwtNumbered
("Updating file " ^ Path.toString path ^ "\n from " ^
root2string fromRoot ^ "\n to " ^
root2string toRoot)
("Updating file " ^ Path.toString path)
(fun () ->
Files.copy (`Update (fileSize uiFrom uiTo))
fromRoot path uiFrom toRoot path uiTo id >>= (fun()->
stashCurrentVersions fromRoot toRoot path))
| (_, _, _, uiFrom), (_, _, _, uiTo) ->
logLwtNumbered
("Copying " ^ Path.toString path ^ "\n from " ^
root2string fromRoot ^ "\n to " ^
root2string toRoot)
("Copying " ^ Path.toString path)
(fun () ->
Files.copy `Copy
fromRoot path uiFrom toRoot path uiTo id >>= (fun()->
stashCurrentVersions fromRoot toRoot path)))
(fun e -> Trace.log
(Printf.sprintf
"Failed: %s\n" (Util.printException e));
return ()))
let propagate root1 root2 reconItem id showMergeFn =
let path = reconItem.path in
match reconItem.replicas with
Problem p ->
Trace.log (Printf.sprintf "[ERROR] Skipping %s\n %s\n"
(Path.toString path) p);
return ()
| Different(rc1,rc2,dir,_) ->
match !dir with
Conflict ->
Trace.log (Printf.sprintf "[CONFLICT] Skipping %s\n"
(Path.toString path));
return ()
| Replica1ToReplica2 ->
doAction (root1, root2) path rc1 rc2 id
| Replica2ToReplica1 ->
doAction (root2, root1) path rc2 rc1 id
| Merge ->
begin match (rc1,rc2) with
(`FILE, _, _, ui1), (`FILE, _, _, ui2) ->
Files.merge root1 root2 path id ui1 ui2 showMergeFn;
return ()
| _ -> raise (Util.Transient "Can only merge two existing files")
end
let transportItem reconItem id showMergeFn =
let (root1,root2) = Globals.roots() in
propagate root1 root2 reconItem id showMergeFn
(* ---------------------------------------------------------------------- *)
let logStart () =
Abort.reset ();
let tm = Util.localtime (Util.time()) in
let m =
Printf.sprintf
"%s%s started propagating changes at %02d:%02d:%02d on %02d %s %04d\n"
(if Prefs.read Trace.terse || Prefs.read Globals.batch then "" else "\n\n")
(String.uppercase Uutil.myNameAndVersion)
tm.Unix.tm_hour tm.Unix.tm_min tm.Unix.tm_sec
tm.Unix.tm_mday (Util.monthname tm.Unix.tm_mon)
(tm.Unix.tm_year+1900) in
Trace.logverbose m
let logFinish () =
let tm = Util.localtime (Util.time()) in
let m =
Printf.sprintf
"%s finished propagating changes at %02d:%02d:%02d on %02d %s %04d\n%s"
(String.uppercase Uutil.myNameAndVersion)
tm.Unix.tm_hour tm.Unix.tm_min tm.Unix.tm_sec
tm.Unix.tm_mday (Util.monthname tm.Unix.tm_mon)
(tm.Unix.tm_year+1900)
(if Prefs.read Trace.terse || Prefs.read Globals.batch then "" else "\n\n") in
Trace.logverbose m
|