[Unison-hackers] [unison-svn] r350 - trunk/src
vouillon@seas.upenn.edu
vouillon at seas.upenn.edu
Tue Jun 9 11:41:38 EDT 2009
Author: vouillon
Date: 2009-06-09 11:41:29 -0400 (Tue, 09 Jun 2009)
New Revision: 350
Modified:
trunk/src/.depend
trunk/src/RECENTNEWS
trunk/src/abort.ml
trunk/src/abort.mli
trunk/src/copy.ml
trunk/src/copy.mli
trunk/src/files.ml
trunk/src/mkProjectInfo.ml
trunk/src/remote.ml
trunk/src/remote.mli
trunk/src/transfer.ml
trunk/src/transfer.mli
trunk/src/transport.ml
trunk/src/update.ml
trunk/src/update.mli
Log:
* Bumped minor version: many incompatible protocol changes
* Message lengths are checksummed to guard against protocol corruption
(no more [Invalid_argument "String.create"])
* Experimental streaming protocol for transferring file contents (can
be disabled by setting the directive "stream" to false): file
contents is transfered asynchronously (without waiting for a response
from the destination after each chunk sent) rather than using the
synchronous RPC mechanism. As a consequence:
- Unison now transfers the contents of a single file at a time
(Unison used to transfer several contents simultaneously in order
to hide the connection latency.)
- the transfer of large files uses the full available bandwidth
and is not slowed done due to the connection latency anymore
- we get performance improvement for small files as well by
scheduling many files simultaneously (as scheduling a file for
transfer consume little ressource: it does not mean allocating a
large buffer anymore)
* Improvement to the code for resuming directory transfers:
- if a file was not correctly transferred (or the source has been
modified since, with unchanged size), Unison performs a new
transfer rather than failing
- spurious files are deleted (this can happen if a file is deleted
on the source replica before resuming the transfer; not deleting
the file would result in it reappearing on the target replica)
* More file transfer implementation cleanup. In particular, the
"paranoid check" (checking whether the file has been correctly
transferred) is moved to copy.ml. This way, one can avoid computing
a file fingerprint twice when a file is already transferred, and
when using the "transfer by copying" optimization.
The check of the source file is also moved to copy.ml, so if the
paranoid check fails, we can tell whether this is due to the source
file being modified.
Modified: trunk/src/.depend
===================================================================
--- trunk/src/.depend 2009-06-09 08:53:55 UTC (rev 349)
+++ trunk/src/.depend 2009-06-09 15:41:29 UTC (rev 350)
@@ -1,4 +1,4 @@
-abort.cmi: uutil.cmi
+abort.cmi: uutil.cmi lwt/lwt.cmi
bytearray.cmi:
case.cmi: ubase/prefs.cmi
checksum.cmi:
@@ -6,7 +6,7 @@
common.cmi: uutil.cmi props.cmi path.cmi osx.cmi os.cmi name.cmi fspath.cmi \
fileinfo.cmi
copy.cmi: uutil.cmi props.cmi path.cmi osx.cmi os.cmi lwt/lwt.cmi fspath.cmi \
- common.cmi
+ fileinfo.cmi common.cmi
external.cmi:
fileinfo.cmi: system.cmi props.cmi path.cmi osx.cmi fspath.cmi
files.cmi: uutil.cmi system.cmi props.cmi path.cmi lwt/lwt_util.cmi \
@@ -46,9 +46,9 @@
uutil.cmi:
xferhint.cmi: ubase/prefs.cmi path.cmi os.cmi fspath.cmi
abort.cmo: uutil.cmi ubase/util.cmi ubase/trace.cmi ubase/safelist.cmi \
- abort.cmi
+ lwt/lwt_util.cmi lwt/lwt.cmi abort.cmi
abort.cmx: uutil.cmx ubase/util.cmx ubase/trace.cmx ubase/safelist.cmx \
- abort.cmi
+ lwt/lwt_util.cmx lwt/lwt.cmx abort.cmi
bytearray.cmo: bytearray.cmi
bytearray.cmx: bytearray.cmi
case.cmo: ubase/util.cmi unicode.cmi ubase/prefs.cmi case.cmi
@@ -154,11 +154,13 @@
ubase/safelist.cmx props.cmx ubase/prefs.cmx pred.cmx path.cmx name.cmx \
globals.cmx fileinfo.cmx common.cmx recon.cmi
remote.cmo: uutil.cmi ubase/util.cmi ubase/trace.cmi terminal.cmi system.cmi \
- ubase/safelist.cmi ubase/prefs.cmi os.cmi lwt/lwt_unix.cmi lwt/lwt.cmi \
- fspath.cmi fs.cmi common.cmi clroot.cmi case.cmi bytearray.cmi remote.cmi
+ ubase/safelist.cmi ubase/prefs.cmi os.cmi lwt/lwt_util.cmi \
+ lwt/lwt_unix.cmi lwt/lwt.cmi fspath.cmi fs.cmi common.cmi clroot.cmi \
+ case.cmi bytearray.cmi remote.cmi
remote.cmx: uutil.cmx ubase/util.cmx ubase/trace.cmx terminal.cmx system.cmx \
- ubase/safelist.cmx ubase/prefs.cmx os.cmx lwt/lwt_unix.cmx lwt/lwt.cmx \
- fspath.cmx fs.cmx common.cmx clroot.cmx case.cmx bytearray.cmx remote.cmi
+ ubase/safelist.cmx ubase/prefs.cmx os.cmx lwt/lwt_util.cmx \
+ lwt/lwt_unix.cmx lwt/lwt.cmx fspath.cmx fs.cmx common.cmx clroot.cmx \
+ case.cmx bytearray.cmx remote.cmi
sortri.cmo: ubase/util.cmi ubase/safelist.cmi ubase/prefs.cmi pred.cmi \
path.cmi common.cmi sortri.cmi
sortri.cmx: ubase/util.cmx ubase/safelist.cmx ubase/prefs.cmx pred.cmx \
@@ -269,14 +271,14 @@
system.cmi stasher.cmi ubase/safelist.cmi remote.cmi props.cmi \
ubase/prefs.cmi pred.cmi path.cmi osx.cmi os.cmi name.cmi ubase/myMap.cmi \
lwt/lwt_unix.cmi lwt/lwt.cmi lock.cmi globals.cmi fspath.cmi fs.cmi \
- fingerprint.cmi fileinfo.cmi external.cmi common.cmi update.cmi
+ fingerprint.cmi fileinfo.cmi common.cmi case.cmi update.cmi
update.cmx: xferhint.cmx uutil.cmx ubase/util.cmx tree.cmx ubase/trace.cmx \
system.cmx stasher.cmx ubase/safelist.cmx remote.cmx props.cmx \
ubase/prefs.cmx pred.cmx path.cmx osx.cmx os.cmx name.cmx ubase/myMap.cmx \
lwt/lwt_unix.cmx lwt/lwt.cmx lock.cmx globals.cmx fspath.cmx fs.cmx \
- fingerprint.cmx fileinfo.cmx external.cmx common.cmx update.cmi
-uutil.cmo: ubase/util.cmi ubase/projectInfo.cmo uutil.cmi
-uutil.cmx: ubase/util.cmx ubase/projectInfo.cmx uutil.cmi
+ fingerprint.cmx fileinfo.cmx common.cmx case.cmx update.cmi
+uutil.cmo: ubase/util.cmi ubase/trace.cmi ubase/projectInfo.cmo uutil.cmi
+uutil.cmx: ubase/util.cmx ubase/trace.cmx ubase/projectInfo.cmx uutil.cmi
xferhint.cmo: ubase/util.cmi ubase/trace.cmi ubase/prefs.cmi path.cmi os.cmi \
fspath.cmi xferhint.cmi
xferhint.cmx: ubase/util.cmx ubase/trace.cmx ubase/prefs.cmx path.cmx os.cmx \
Modified: trunk/src/RECENTNEWS
===================================================================
--- trunk/src/RECENTNEWS 2009-06-09 08:53:55 UTC (rev 349)
+++ trunk/src/RECENTNEWS 2009-06-09 15:41:29 UTC (rev 350)
@@ -1,3 +1,43 @@
+CHANGES FROM VERSION 2.35.-17
+
+* Bumped minor version: many incompatible protocol changes
+
+* Message lengths are checksummed to guard against protocol corruption
+ (no more [Invalid_argument "String.create"])
+
+* Experimental streaming protocol for transferring file contents (can
+ be disabled by setting the directive "stream" to false): file
+ contents is transfered asynchronously (without waiting for a response
+ from the destination after each chunk sent) rather than using the
+ synchronous RPC mechanism. As a consequence:
+ - Unison now transfers the contents of a single file at a time
+ (Unison used to transfer several contents simultaneously in order
+ to hide the connection latency.)
+ - the transfer of large files uses the full available bandwidth
+ and is not slowed done due to the connection latency anymore
+ - we get performance improvement for small files as well by
+ scheduling many files simultaneously (as scheduling a file for
+ transfer consume little ressource: it does not mean allocating a
+ large buffer anymore)
+
+* Improvement to the code for resuming directory transfers:
+ - if a file was not correctly transferred (or the source has been
+ modified since, with unchanged size), Unison performs a new
+ transfer rather than failing
+ - spurious files are deleted (this can happen if a file is deleted
+ on the source replica before resuming the transfer; not deleting
+ the file would result in it reappearing on the target replica)
+
+* More file transfer implementation cleanup. In particular, the
+ "paranoid check" (checking whether the file has been correctly
+ transferred) is moved to copy.ml. This way, one can avoid computing
+ a file fingerprint twice when a file is already transferred, and
+ when using the "transfer by copying" optimization.
+ The check of the source file is also moved to copy.ml, so if the
+ paranoid check fails, we can tell whether this is due to the source
+ file being modified.
+
+-------------------------------
CHANGES FROM VERSION 2.34.0
* Improvement to the code for resuming directory transfers:
Modified: trunk/src/abort.ml
===================================================================
--- trunk/src/abort.ml 2009-06-09 08:53:55 UTC (rev 349)
+++ trunk/src/abort.ml 2009-06-09 15:41:29 UTC (rev 350)
@@ -44,3 +44,26 @@
end
let testException e = e = Util.Transient "Aborted"
+
+let (>>=) = Lwt.bind
+
+let mergeErrors id e runningThreads =
+ if not (testException e) then file id;
+ match e with
+ Util.Transient _ ->
+ let e = ref e in
+ Lwt_util.iter
+ (fun act ->
+ Lwt.catch
+ (fun () -> act >>= fun _ -> Lwt.return ())
+ (fun e' ->
+ match e' with
+ Util.Transient _ ->
+ if testException !e then e := e';
+ Lwt.return ()
+ | _ ->
+ Lwt.fail e'))
+ runningThreads >>= fun () ->
+ Lwt.fail !e
+ | _ ->
+ Lwt.fail e
Modified: trunk/src/abort.mli
===================================================================
--- trunk/src/abort.mli 2009-06-09 08:53:55 UTC (rev 349)
+++ trunk/src/abort.mli 2009-06-09 15:41:29 UTC (rev 350)
@@ -13,3 +13,8 @@
(* Test whether the exeption is an abort exception. *)
val testException : exn -> bool
+
+(* When one thread has failed (in a non-fatal way), this function will
+ abort the current transfer and wait for all other threads in the
+ list to terminate before continuing *)
+val mergeErrors : Uutil.File.t -> exn -> 'a Lwt.t list -> 'b Lwt.t
Modified: trunk/src/copy.ml
===================================================================
--- trunk/src/copy.ml 2009-06-09 08:53:55 UTC (rev 349)
+++ trunk/src/copy.ml 2009-06-09 15:41:29 UTC (rev 350)
@@ -42,17 +42,102 @@
(****)
+(* Check whether the source file has been modified during synchronization *)
+let checkContentsChangeLocal
+ fspathFrom pathFrom archDesc archDig archStamp archRess paranoid =
+ let info = Fileinfo.get true fspathFrom pathFrom in
+ let clearlyModified =
+ info.Fileinfo.typ <> `FILE
+ || Props.length info.Fileinfo.desc <> Props.length archDesc
+ || Osx.ressLength info.Fileinfo.osX.Osx.ressInfo <>
+ Osx.ressLength archRess
+ in
+ let dataClearlyUnchanged =
+ not clearlyModified
+ && Props.same_time info.Fileinfo.desc archDesc
+(*FIX: should export from update.ml?
+ && not (excelFile path)
+*)
+ && match archStamp with
+ Some (Fileinfo.InodeStamp inode) -> info.Fileinfo.inode = inode
+ | Some (Fileinfo.CtimeStamp ctime) -> true
+ | None -> false
+ in
+ let ressClearlyUnchanged =
+ not clearlyModified
+ && Osx.ressUnchanged archRess info.Fileinfo.osX.Osx.ressInfo
+ None dataClearlyUnchanged
+ in
+ if dataClearlyUnchanged && ressClearlyUnchanged then begin
+ if paranoid then begin
+ let newDig = Os.fingerprint fspathFrom pathFrom info in
+ if archDig <> newDig then
+ raise (Util.Transient (Printf.sprintf
+ "The source file %s\n\
+ has been modified but the fast update detection mechanism\n\
+ failed to detect it. Try running once with the fastcheck\n\
+ option set to 'no'."
+ (Fspath.toPrintString (Fspath.concat fspathFrom pathFrom))))
+ end
+ end else if
+ clearlyModified
+ || archDig <> Os.fingerprint fspathFrom pathFrom info
+ then
+ raise (Util.Transient (Printf.sprintf
+ "The source file %s\nhas been modified during synchronization. \
+ Transfer aborted."
+ (Fspath.toPrintString (Fspath.concat fspathFrom pathFrom))))
+
+let checkContentsChangeOnHost =
+ Remote.registerRootCmd
+ "checkContentsChange"
+ (fun (fspathFrom,
+ (pathFrom, archDesc, archDig, archStamp, archRess, paranoid)) ->
+ checkContentsChangeLocal
+ fspathFrom pathFrom archDesc archDig archStamp archRess paranoid;
+ Lwt.return ())
+
+let checkContentsChange
+ root pathFrom archDesc archDig archStamp archRess paranoid =
+ checkContentsChangeOnHost
+ root (pathFrom, archDesc, archDig, archStamp, archRess, paranoid)
+
+(****)
+
let fileIsTransferred fspathTo pathTo desc fp ress =
let info = Fileinfo.get false fspathTo pathTo in
(info,
- info.Fileinfo.typ = `FILE &&
+ info.Fileinfo.typ = `FILE
+ &&
Props.length info.Fileinfo.desc = Props.length desc
- && Osx.ressLength info.Fileinfo.osX.Osx.ressInfo =
- Osx.ressLength ress
- &&
+ &&
+ Osx.ressLength info.Fileinfo.osX.Osx.ressInfo =
+ Osx.ressLength ress
+ &&
let fp' = Os.fingerprint fspathTo pathTo info in
fp' = fp)
+type transferStatus =
+ Success of Fileinfo.t
+ | Failure of string
+
+(* Paranoid check: recompute the transferred file's digest to match it
+ with the archive's *)
+let paranoidCheck fspathTo pathTo desc fp ress =
+ let info = Fileinfo.get false fspathTo pathTo in
+ let fp' = Os.fingerprint fspathTo pathTo info in
+ if fp' <> fp then begin
+ let savepath = Path.addSuffixToFinalName pathTo "-bad" in
+ Os.rename "save temp" fspathTo pathTo fspathTo savepath;
+ Lwt.return (Failure (Printf.sprintf
+ "The file %s was incorrectly transferred (fingerprint mismatch in %s) \
+ -- temp file saved as %s"
+ (Path.toString pathTo)
+ (Os.reasonForFingerprintMismatch fp fp')
+ (Path.toString savepath)))
+ end else
+ Lwt.return (Success info)
+
(****)
let removeOldTempFile fspathTo pathTo =
@@ -156,14 +241,13 @@
else Trace.log s
let tryCopyMovedFile fspathTo pathTo realPathTo update desc fp ress id =
- Prefs.read Xferhint.xferbycopying
- &&
+ if not (Prefs.read Xferhint.xferbycopying) then None else
Util.convertUnixErrorsToTransient "tryCopyMovedFile" (fun() ->
debug (fun () -> Util.msg "tryCopyMovedFile: -> %s /%s/\n"
(Path.toString pathTo) (Os.fullfingerprint_to_string fp));
match Xferhint.lookup fp with
None ->
- false
+ None
| Some (candidateFspath, candidatePath) ->
loggit (Printf.sprintf
"Shortcut: copying %s from local file %s\n"
@@ -184,7 +268,7 @@
if isTransferred then begin
debug (fun () -> Util.msg "tryCopyMoveFile: success.\n");
Xferhint.insertEntry (fspathTo, pathTo) fp;
- true
+ Some info
end else begin
debug (fun () ->
Util.msg "tryCopyMoveFile: candidate file modified!");
@@ -193,14 +277,14 @@
loggit (Printf.sprintf
"Shortcut didn't work because %s was modified\n"
(Path.toString candidatePath));
- false
+ None
end
end else begin
loggit (Printf.sprintf
"Shortcut didn't work because %s disappeared!\n"
(Path.toString candidatePath));
Xferhint.deleteEntry (candidateFspath, candidatePath);
- false
+ None
end
with
Util.Transient s ->
@@ -211,7 +295,7 @@
loggit (Printf.sprintf
"Local copy of %s failed\n"
(Path.toString candidatePath));
- false)
+ None)
(****)
@@ -236,8 +320,7 @@
Util.convertUnixErrorsToTransient
"processing a transfer instruction"
(fun () ->
- ignore (Remote.MsgIdMap.find file_id !decompressor ti));
- Lwt.return ()
+ ignore (Remote.MsgIdMap.find file_id !decompressor ti))
let marshalTransferInstruction =
(fun (file_id, (data, pos, len)) rem ->
@@ -247,17 +330,17 @@
let len = Bytearray.length buf - pos - Remote.intSize in
(Remote.decodeInt buf pos, (buf, pos + Remote.intSize, len)))
-let processTransferInstructionRemotely =
- Remote.registerSpecialServerCmd
+let streamTransferInstruction =
+ Remote.registerStreamCmd
"processTransferInstruction" marshalTransferInstruction
- Remote.defaultMarshalingFunctions processTransferInstruction
+ processTransferInstruction
-let blockInfos = ref Remote.MsgIdMap.empty
-
let compress conn
(biOpt, fspathFrom, pathFrom, fileKind, sizeFrom, id, file_id) =
Util.convertUnixErrorsToTransient "rsync sender"
(fun () ->
+ streamTransferInstruction conn
+ (fun processTransferInstructionRemotely ->
let infd = openFileIn fspathFrom pathFrom fileKind in
lwt_protect
(fun () ->
@@ -266,47 +349,22 @@
Uutil.showProgress id (Uutil.Filesize.ofInt count) "r" in
let compr =
match biOpt with
- None -> Transfer.send infd sizeFrom showProgress
- | Some bi -> let remBi =
- try
- Remote.MsgIdMap.find file_id !blockInfos
- with Not_found ->
- []
- in
- let bi = bi :: remBi in
- blockInfos :=
- Remote.MsgIdMap.remove file_id !blockInfos;
- Transfer.Rsync.rsyncCompress
- bi infd sizeFrom showProgress
+ None ->
+ Transfer.send infd sizeFrom showProgress
+ | Some bi ->
+ Transfer.Rsync.rsyncCompress
+ bi infd sizeFrom showProgress
in
compr
- (fun ti ->
- processTransferInstructionRemotely conn (file_id, ti))
+ (fun ti -> processTransferInstructionRemotely (file_id, ti))
>>= fun () ->
close_in infd;
Lwt.return ())
(fun () ->
- close_in_noerr infd))
+ close_in_noerr infd)))
let compressRemotely = Remote.registerServerCmd "compress" compress
-let receiveRemBiLocally _ (file_id, bi) =
- let bil =
- try
- Remote.MsgIdMap.find file_id !blockInfos
- with Not_found ->
- []
- in
- blockInfos := Remote.MsgIdMap.add file_id (bi :: bil) !blockInfos;
- Lwt.return ()
-
-let receiveRemBi = Remote.registerServerCmd "receiveRemBi" receiveRemBiLocally
-let rec sendRemBi conn file_id remBi =
- match remBi with
- [] -> Lwt.return ()
- | x :: r -> sendRemBi conn file_id r >>= (fun () ->
- receiveRemBi conn (file_id, x))
-
let close_all infd outfd =
Util.convertUnixErrorsToTransient
"closing files"
@@ -372,7 +430,7 @@
(fun () -> close_in_noerr ifd)
in
infd := Some ifd;
- (bi,
+ (Some bi,
(* Rsync decompressor *)
fun ti ->
let fd =
@@ -383,7 +441,7 @@
in
if eof then begin close_out fd; outfd := None end))
| _ ->
- ([],
+ (None,
(* Simple generic decompressor *)
fun ti ->
let fd = destinationFd fspathTo pathTo fileKind srcFileSize outfd in
@@ -395,14 +453,8 @@
(fun () ->
decompressor := Remote.MsgIdMap.add file_id decompr !decompressor;
Uutil.showProgress id Uutil.Filesize.zero "f";
- let (firstBi, remBi) =
- match bi with
- [] -> (None, [])
- | firstBi :: remBi -> (Some firstBi, remBi)
- in
- sendRemBi connFrom file_id remBi >>= fun () ->
compressRemotely connFrom
- (firstBi, fspathFrom, pathFrom, fileKind, srcFileSize, id, file_id)
+ (bi, fspathFrom, pathFrom, fileKind, srcFileSize, id, file_id)
>>= fun () ->
decompressor :=
Remote.MsgIdMap.remove file_id !decompressor; (* For GC *)
@@ -416,18 +468,6 @@
(****)
-let fileSize (fspath, path) =
- Util.convertUnixErrorsToTransient
- "getting file size"
- (fun () ->
- Lwt.return
- (Props.length (Fileinfo.get false fspath path).Fileinfo.desc))
-
-let fileSizeOnHost =
- Remote.registerServerCmd "fileSize" (fun _ -> fileSize)
-
-(****)
-
let transferRessourceForkAndSetFileinfo
connFrom fspathFrom pathFrom fspathTo pathTo realPathTo
update desc fp ress id =
@@ -441,30 +481,20 @@
Lwt.return ()
end >>= fun () ->
setFileinfo fspathTo pathTo realPathTo update desc;
- Lwt.return ()
+ paranoidCheck fspathTo pathTo desc fp ress
-(* The ressOnly flag tells reallyTransferFile to skip transferring
- the data fork (which has already been taken care of by some external
- utility) and just transfer the resource fork (which external utilities
- are not necessarily good at). *)
let reallyTransferFile
connFrom fspathFrom pathFrom fspathTo pathTo realPathTo
- update desc fp ress ressOnly id =
- debug (fun() -> Util.msg "reallyTransferFile(%s,%s) -> (%s,%s,%s,%s)%s\n"
+ update desc fp ress id =
+ debug (fun() -> Util.msg "reallyTransferFile(%s,%s) -> (%s,%s,%s,%s)\n"
(Fspath.toDebugString fspathFrom) (Path.toString pathFrom)
(Fspath.toDebugString fspathTo) (Path.toString pathTo)
- (Path.toString realPathTo) (Props.toString desc)
- (if ressOnly then " (ONLY RESOURCE FORK)" else ""));
- (if ressOnly then
- (* Skip data fork *)
- Lwt.return ()
- else begin
+ (Path.toString realPathTo) (Props.toString desc));
removeOldTempFile fspathTo pathTo;
(* Data fork *)
transferFileContents
connFrom fspathFrom pathFrom fspathTo pathTo realPathTo update
- `DATA (Props.length desc) id
- end) >>= fun () ->
+ `DATA (Props.length desc) id >>= fun () ->
transferRessourceForkAndSetFileinfo
connFrom fspathFrom pathFrom fspathTo pathTo realPathTo
update desc fp ress id
@@ -514,41 +544,6 @@
^ "added if the value of {\\tt copyprog} contains the string "
^ "{\\tt rsync}.")
-let tryCopyMovedFileLocal connFrom
- (fspathTo, pathTo, realPathTo, update, desc, fp, ress, id) =
- Lwt.return (tryCopyMovedFile fspathTo pathTo realPathTo update desc fp ress id)
-let tryCopyMovedFileOnRoot =
- Remote.registerRootCmdWithConnection "tryCopyMovedFile" tryCopyMovedFileLocal
-
-let setFileinfoLocal connFrom (fspathTo, pathTo, desc) =
- Lwt.return (Fileinfo.set fspathTo pathTo (`Set Props.fileDefault) desc)
-let setFileinfoOnRoot =
- Remote.registerRootCmdWithConnection "setFileinfo" setFileinfoLocal
-
-let targetExists checkSize fspathTo pathTo =
- let info = Fileinfo.get false fspathTo pathTo in
- info.Fileinfo.typ = `FILE
- && (match checkSize with
- `MakeWriteableAndCheckNonempty ->
- let perms = Props.perms info.Fileinfo.desc in
- let perms' = perms lor 0o600 in
- Util.convertUnixErrorsToTransient
- "making target writable"
- (fun () -> Fs.chmod (Fspath.concat fspathTo pathTo) perms');
- Props.length info.Fileinfo.desc > Uutil.Filesize.zero
- | `CheckDataSize desc ->
- Props.length info.Fileinfo.desc = Props.length desc
- | `CheckSize (desc,ress) ->
- Props.length info.Fileinfo.desc = Props.length desc
- && Osx.ressLength info.Fileinfo.osX.Osx.ressInfo =
- Osx.ressLength ress)
-
-let targetExistsLocal connFrom (checkSize, fspathTo, pathTo) =
- Lwt.return (targetExists checkSize fspathTo pathTo)
-let targetExistsOnRoot =
- Remote.registerRootCmdWithConnection
- "targetExists" targetExistsLocal
-
let formatConnectionInfo root =
match root with
Common.Local, _ -> ""
@@ -579,59 +574,116 @@
(Int64.of_int (Prefs.read copythreshold)))
&& update = `Copy
+let prepareExternalTransfer fspathTo pathTo =
+ let info = Fileinfo.get false fspathTo pathTo in
+ match info.Fileinfo.typ with
+ `FILE when Props.length info.Fileinfo.desc > Uutil.Filesize.zero ->
+ let perms = Props.perms info.Fileinfo.desc in
+ let perms' = perms lor 0o600 in
+ begin try
+ Fs.chmod (Fspath.concat fspathTo pathTo) perms'
+ with Unix.Unix_error _ -> () end;
+ true
+ | `ABSENT ->
+ false
+ | _ ->
+ debug (fun() -> Util.msg "Removing old temp file %s / %s\n"
+ (Fspath.toDebugString fspathTo) (Path.toString pathTo));
+ Os.delete fspathTo pathTo;
+ false
+
+let finishExternalTransferLocal connFrom
+ (fspathFrom, pathFrom, fspathTo, pathTo, realPathTo,
+ update, desc, fp, ress, id) =
+ let info = Fileinfo.get false fspathTo pathTo in
+ if
+ info.Fileinfo.typ <> `FILE ||
+ Props.length info.Fileinfo.desc <> Props.length desc
+ then
+ raise (Util.Transient (Printf.sprintf
+ "External copy program did not create target file (or bad length): %s"
+ (Path.toString pathTo)));
+ transferRessourceForkAndSetFileinfo
+ connFrom fspathFrom pathFrom fspathTo pathTo realPathTo
+ update desc fp ress id
+
+let finishExternalTransferOnRoot =
+ Remote.registerRootCmdWithConnection
+ "finishExternalTransfer" finishExternalTransferLocal
+
let transferFileUsingExternalCopyprog
rootFrom pathFrom rootTo fspathTo pathTo realPathTo
- update desc fp ress id =
- tryCopyMovedFileOnRoot rootTo rootFrom
- (fspathTo, pathTo, realPathTo, update, desc, fp, ress, id)
- >>= (fun b ->
- if b then Lwt.return ()
- else begin
- Uutil.showProgress id Uutil.Filesize.zero "ext";
- targetExistsOnRoot
- rootTo rootFrom (`MakeWriteableAndCheckNonempty, fspathTo, pathTo) >>= (fun b ->
- let prog =
- if b
- then Prefs.read copyprogrest
- else Prefs.read copyprog in
- let extraquotes = Prefs.read copyquoterem = "true"
- || ( Prefs.read copyquoterem = "default"
- && Util.findsubstring "rsync" prog <> None) in
- let addquotes root s =
- match root with
- | Common.Local, _ -> s
- | Common.Remote _, _ -> if extraquotes then Uutil.quotes s else s in
- let fromSpec =
- (formatConnectionInfo rootFrom)
- ^ (addquotes rootFrom
- (Fspath.toString (Fspath.concat (snd rootFrom) pathFrom))) in
- let toSpec =
- (formatConnectionInfo rootTo)
- ^ (addquotes rootTo
- (Fspath.toString (Fspath.concat fspathTo pathTo))) in
- let cmd = prog ^ " "
- ^ (Uutil.quotes fromSpec) ^ " "
- ^ (Uutil.quotes toSpec) in
- Trace.log (Printf.sprintf "%s\n" cmd);
- let _,log = External.runExternalProgram cmd in
- debug (fun() ->
- let l = Util.trimWhitespace log in
- Util.msg "transferFileUsingExternalCopyprog %s: returned...\n%s%s"
- (Path.toString pathFrom)
- l (if l="" then "" else "\n"));
- targetExistsOnRoot
- rootTo rootFrom (`CheckDataSize desc, fspathTo, pathTo)
- >>= (fun b ->
- if not b then
- raise (Util.Transient (Printf.sprintf
- "External copy program did not create target file (or bad length): %s"
- (Path.toString pathTo)));
- Uutil.showProgress id (Props.length desc) "ext";
- Lwt.return ()))
- end)
+ update desc fp ress id useExistingTarget =
+ Uutil.showProgress id Uutil.Filesize.zero "ext";
+ let prog =
+ if useExistingTarget then
+ Prefs.read copyprogrest
+ else
+ Prefs.read copyprog
+ in
+ let extraquotes = Prefs.read copyquoterem = "true"
+ || ( Prefs.read copyquoterem = "default"
+ && Util.findsubstring "rsync" prog <> None) in
+ let addquotes root s =
+ match root with
+ | Common.Local, _ -> s
+ | Common.Remote _, _ -> if extraquotes then Uutil.quotes s else s in
+ let fromSpec =
+ (formatConnectionInfo rootFrom)
+ ^ (addquotes rootFrom
+ (Fspath.toString (Fspath.concat (snd rootFrom) pathFrom))) in
+ let toSpec =
+ (formatConnectionInfo rootTo)
+ ^ (addquotes rootTo
+ (Fspath.toString (Fspath.concat fspathTo pathTo))) in
+ let cmd = prog ^ " "
+ ^ (Uutil.quotes fromSpec) ^ " "
+ ^ (Uutil.quotes toSpec) in
+ Trace.log (Printf.sprintf "%s\n" cmd);
+ let _,log = External.runExternalProgram cmd in
+ debug (fun() ->
+ let l = Util.trimWhitespace log in
+ Util.msg "transferFileUsingExternalCopyprog %s: returned...\n%s%s"
+ (Path.toString pathFrom)
+ l (if l="" then "" else "\n"));
+ Uutil.showProgress id (Props.length desc) "ext";
+ finishExternalTransferOnRoot rootTo rootFrom
+ (snd rootFrom, pathFrom, fspathTo, pathTo, realPathTo,
+ update, desc, fp, ress, id)
-(****)
+let transferFileLocal connFrom
+ (fspathFrom, pathFrom, fspathTo, pathTo, realPathTo,
+ update, desc, fp, ress, id) =
+ let (info, isTransferred) = fileIsTransferred fspathTo pathTo desc fp ress in
+ if isTransferred then begin
+ (* File is already fully transferred (from some interrupted
+ previous transfer). *)
+ (* Make sure permissions are right. *)
+ Trace.log (Printf.sprintf
+ "%s/%s has already been transferred\n"
+ (Fspath.toDebugString fspathTo) (Path.toString pathTo));
+ setFileinfo fspathTo pathTo realPathTo update desc;
+ Lwt.return (`DONE (Success info))
+ end else
+ match
+ tryCopyMovedFile fspathTo pathTo realPathTo update desc fp ress id
+ with
+ Some info ->
+ (* Transfer was performed by copying *)
+ Lwt.return (`DONE (Success info))
+ | None ->
+ if shouldUseExternalCopyprog update desc then
+ Lwt.return (`EXTERNAL (prepareExternalTransfer fspathTo pathTo))
+ else begin
+ reallyTransferFile
+ connFrom fspathFrom pathFrom fspathTo pathTo realPathTo
+ update desc fp ress id >>= fun status ->
+ Lwt.return (`DONE status)
+ end
+let transferFileOnRoot =
+ Remote.registerRootCmdWithConnection "transferFile" transferFileLocal
+
(* We limit the size of the output buffers to about 512 KB
(we cannot go above the limit below plus 64) *)
let transferFileReg = Lwt_util.make_region 440
@@ -642,35 +694,35 @@
+
8 (* Read buffer *)
-let transferFileLocal connFrom
- (fspathFrom, pathFrom, fspathTo, pathTo, realPathTo,
- update, desc, fp, ress, ressOnly, id) =
- if (not ressOnly)
- && tryCopyMovedFile fspathTo pathTo realPathTo update desc fp ress id
- then Lwt.return ()
- else reallyTransferFile
- connFrom fspathFrom pathFrom fspathTo pathTo realPathTo
- update desc fp ress ressOnly id
-
-let transferFileOnRoot =
- Remote.registerRootCmdWithConnection "transferFile" transferFileLocal
-
let transferFile
rootFrom pathFrom rootTo fspathTo pathTo realPathTo
- update desc fp ress ressOnly id =
- let bufSz = bufferSize (max (Props.length desc) (Osx.ressLength ress)) in
- (* This must be on the client: any lock on the server side may result
- in a deadlock under windows *)
- Lwt_util.run_in_region transferFileReg bufSz (fun () ->
+ update desc fp ress id =
+ let f () =
Abort.check id;
transferFileOnRoot rootTo rootFrom
(snd rootFrom, pathFrom, fspathTo, pathTo, realPathTo,
- update, desc, fp, ress, ressOnly, id))
+ update, desc, fp, ress, id) >>= fun status ->
+ match status with
+ `DONE status ->
+ Lwt.return status
+ | `EXTERNAL useExistingTarget ->
+ transferFileUsingExternalCopyprog
+ rootFrom pathFrom rootTo fspathTo pathTo realPathTo
+ update desc fp ress id useExistingTarget
+ in
+ (* When streaming, we only transfer one file at a time *)
+ if Prefs.read Remote.streamingActivated then
+ f ()
+ else
+ let bufSz = bufferSize (max (Props.length desc) (Osx.ressLength ress)) in
+ (* This must be on the client: any lock on the server side may result
+ in a deadlock under windows *)
+ Lwt_util.run_in_region transferFileReg bufSz f
(****)
let file rootFrom pathFrom rootTo fspathTo pathTo realPathTo
- update desc fp ress id =
+ update desc fp stamp ress id =
debug (fun() -> Util.msg "copyRegFile(%s,%s) -> (%s,%s,%s,%s,%s)\n"
(Common.root2string rootFrom) (Path.toString pathFrom)
(Common.root2string rootTo) (Path.toString realPathTo)
@@ -682,41 +734,21 @@
localFile
fspathFrom pathFrom fspathTo pathTo realPathTo
update desc (Osx.ressLength ress) (Some id);
- Lwt.return ()
+ paranoidCheck fspathTo pathTo desc fp ress
| _ ->
- (* Check whether we actually need to copy the file (or whether it
- already exists from some interrupted previous transfer) *)
- targetExistsOnRoot
- rootTo rootFrom (`CheckSize (desc,ress), fspathTo, pathTo) >>= (fun b ->
- if b then begin
- Trace.log (Printf.sprintf
- "%s/%s has already been transferred\n"
- (Fspath.toDebugString fspathTo) (Path.toString pathTo));
- (* Make sure the file information is right *)
- setFileinfoOnRoot rootTo rootFrom (fspathTo, pathTo, desc)
- (* Check whether we should use an external program to copy the
- file *)
- end else if shouldUseExternalCopyprog update desc then begin
- (* First use the external program to copy the data fork *)
- transferFileUsingExternalCopyprog
- rootFrom pathFrom rootTo fspathTo pathTo realPathTo
- update desc fp ress id >>= (fun () ->
- (* Now use the regular transport mechanism to copy the resource
- fork *)
- begin if (Osx.ressLength ress) > Uutil.Filesize.zero then begin
- transferFile
- rootFrom pathFrom rootTo fspathTo pathTo realPathTo
- update desc fp ress true id
- end else Lwt.return ()
- end >>= (fun() ->
- (* Finally, set the file info *)
- setFileinfoOnRoot rootTo rootFrom (fspathTo, pathTo, desc)))
- end else
- (* Just transfer the file in the usual way with Unison's
- built-in facilities *)
- transferFile
- rootFrom pathFrom rootTo fspathTo pathTo realPathTo
- update desc fp ress false id
- ) end >>= (fun () ->
+ transferFile
+ rootFrom pathFrom rootTo fspathTo pathTo realPathTo
+ update desc fp ress id
+ end >>= fun status ->
Trace.showTimer timer;
- Lwt.return ())
+ match status with
+ Success info ->
+ checkContentsChange rootFrom pathFrom desc fp stamp ress false
+ >>= fun () ->
+ Lwt.return info
+ | Failure reason ->
+ (* Maybe we failed because the source file was modified.
+ We check this before reporting a failure *)
+ checkContentsChange rootFrom pathFrom desc fp stamp ress true
+ >>= fun () ->
+ Lwt.fail (Util.Transient reason)
Modified: trunk/src/copy.mli
===================================================================
--- trunk/src/copy.mli 2009-06-09 08:53:55 UTC (rev 349)
+++ trunk/src/copy.mli 2009-06-09 15:41:29 UTC (rev 350)
@@ -10,9 +10,11 @@
-> [`Update of (Uutil.Filesize.t * Uutil.Filesize.t) | `Copy]
-> Props.t (* permissions for new file *)
-> Os.fullfingerprint (* fingerprint of file *)
+ -> Fileinfo.stamp option
+ (* source file stamp, if available *)
-> Osx.ressStamp (* ressource info of file *)
-> Uutil.File.t (* file's index in UI (for progress bars) *)
- -> unit Lwt.t
+ -> Fileinfo.t Lwt.t (* information regarding the transferred file *)
val localFile :
Fspath.t (* fspath of source *)
Modified: trunk/src/files.ml
===================================================================
--- trunk/src/files.ml 2009-06-09 08:53:55 UTC (rev 349)
+++ trunk/src/files.ml 2009-06-09 15:41:29 UTC (rev 350)
@@ -90,13 +90,12 @@
(* FIX: maybe we should rename the destination before making any check ? *)
let delete rootFrom pathFrom rootTo pathTo ui =
Update.transaction (fun id ->
- Update.replaceArchive rootFrom pathFrom None Update.NoArchive id true false
+ Update.replaceArchive rootFrom pathFrom Update.NoArchive id
>>= (fun _ ->
(* Unison do the next line cause we want to keep a backup of the file.
FIX: We only need this when we are making backups *)
Update.updateArchive rootTo pathTo ui id >>= (fun _ ->
- Update.replaceArchive
- rootTo pathTo None Update.NoArchive id true false
+ Update.replaceArchive rootTo pathTo Update.NoArchive id
>>= (fun localPathTo ->
(* Make sure the target is unchanged *)
(* (There is an unavoidable race condition here.) *)
@@ -151,12 +150,12 @@
Fs.chmod (Fspath.concat workingDir path)
(Props.perms info.Fileinfo.desc lor 0o700)
with Unix.Unix_error _ -> () end;
- Lwt.return info.Fileinfo.desc
+ Lwt.return (true, info.Fileinfo.desc)
end else begin
if info.Fileinfo.typ <> `ABSENT then
Os.delete workingDir path;
Os.createDir workingDir path Props.dirDefault;
- Lwt.return (Fileinfo.get false workingDir path).Fileinfo.desc
+ Lwt.return (false, (Fileinfo.get false workingDir path).Fileinfo.desc)
end)
let mkdir onRoot workingDir path = mkdirRemote onRoot (workingDir,path)
@@ -268,50 +267,6 @@
(* ------------------------------------------------------------ *)
-let checkContentsChangeLocal
- currfspath path archDesc archDig archStamp archRess =
- let info = Fileinfo.get true currfspath path in
- if Props.length archDesc <> Props.length info.Fileinfo.desc then
- raise (Util.Transient (Printf.sprintf
- "The file %s\nhas been modified during synchronization. \
- Transfer aborted."
- (Fspath.toPrintString (Fspath.concat currfspath path))));
- match archStamp with
- Fileinfo.InodeStamp inode
- when info.Fileinfo.inode = inode
- && Props.same_time info.Fileinfo.desc archDesc ->
- ()
- | _ ->
- (* Note that we fall back to the paranoid check (using a fingerprint)
- even if a CtimeStamp was provided, since we do not trust them
- completely. *)
- let (info, newDig) = Os.safeFingerprint currfspath path info None in
- if archDig <> newDig then
- raise (Util.Transient (Printf.sprintf
- "The file %s\nhas been modified during synchronization. \
- Transfer aborted.%s"
- (Fspath.toPrintString (Fspath.concat currfspath path))
- (if Update.useFastChecking ()
- && Props.same_time info.Fileinfo.desc archDesc
- then
- " If this happens repeatedly, try running once with the \
- fastcheck option set to 'no'"
- else
- "")))
-
-let checkContentsChangeOnHost =
- Remote.registerRootCmd
- "checkContentsChange"
- (fun (currfspath, (path, archDesc, archDig, archStamp, archRess)) ->
- checkContentsChangeLocal
- currfspath path archDesc archDig archStamp archRess;
- Lwt.return ())
-
-let checkContentsChange root path archDesc archDig archStamp archRess =
- checkContentsChangeOnHost root (path, archDesc, archDig, archStamp, archRess)
-
-(* ------------------------------------------------------------ *)
-
(* Calculate the target working directory and paths for the copy.
workingDir is an fspath naming the directory on the target
host where the copied file will actually live.
@@ -348,6 +303,44 @@
Os.symlink workingDir path l;
Lwt.return ())
+(* ------------------------------------------------------------ *)
+
+let deleteSpuriousChild fspathTo pathTo nm =
+ let path = (Path.child pathTo nm) in
+ debug (fun() -> Util.msg "Deleting spurious file %s/%s\n"
+ (Fspath.toDebugString fspathTo) (Path.toString path));
+ Os.delete fspathTo path
+
+let rec deleteSpuriousChildrenRec fspathTo pathTo archChildren children =
+ match archChildren, children with
+ archNm :: archRem, nm :: rem ->
+ let c = Name.compare archNm nm in
+ if c < 0 then
+ deleteSpuriousChildrenRec fspathTo pathTo archRem children
+ else if c = 0 then
+ deleteSpuriousChildrenRec fspathTo pathTo archChildren rem
+ else begin
+ deleteSpuriousChild fspathTo pathTo nm;
+ deleteSpuriousChildrenRec fspathTo pathTo archChildren rem
+ end
+ | [], nm :: rem ->
+ deleteSpuriousChild fspathTo pathTo nm;
+ deleteSpuriousChildrenRec fspathTo pathTo [] rem
+ | _, [] ->
+ ()
+
+let deleteSpuriousChildrenLocal (_, (fspathTo, pathTo, archChildren)) =
+List.iter (fun nm -> Format.eprintf "%s at ." (Name.toString nm)) archChildren;
+ deleteSpuriousChildrenRec
+ fspathTo pathTo archChildren
+ (List.sort Name.compare (Os.childrenOf fspathTo pathTo));
+ Lwt.return ()
+
+let deleteSpuriousChildren =
+ Remote.registerRootCmd "deleteSpuriousChildren" deleteSpuriousChildrenLocal
+
+(* ------------------------------------------------------------ *)
+
let copyReg = Lwt_util.make_region 50
let copy
@@ -383,87 +376,87 @@
Update.ArchiveFile (desc, dig, stamp, ress) ->
Lwt_util.run_in_region copyReg 1 (fun () ->
Abort.check id;
+ let stmp = if Update.useFastChecking () then Some stamp else None in
Copy.file
rootFrom pFrom rootTo workingDir pTo realPTo
- update desc dig ress id
- >>= (fun () ->
- checkContentsChange rootFrom pFrom desc dig stamp ress))
+ update desc dig stmp ress id
+ >>= fun info ->
+ let ress' = Osx.stamp info.Fileinfo.osX in
+ Lwt.return
+ (Update.ArchiveFile (Props.override info.Fileinfo.desc desc,
+ dig, Fileinfo.stamp info, ress')))
| Update.ArchiveSymlink l ->
Lwt_util.run_in_region copyReg 1 (fun () ->
debug (fun() -> Util.msg "Making symlink %s/%s -> %s\n"
(root2string rootTo) (Path.toString pTo) l);
Abort.check id;
- makeSymlink rootTo (workingDir, pTo, l))
+ makeSymlink rootTo (workingDir, pTo, l) >>= fun () ->
+ Lwt.return f)
| Update.ArchiveDir (desc, children) ->
Lwt_util.run_in_region copyReg 1 (fun () ->
debug (fun() -> Util.msg "Creating directory %s/%s\n"
(root2string rootTo) (Path.toString pTo));
- mkdir rootTo workingDir pTo) >>= (fun initialDesc ->
+ mkdir rootTo workingDir pTo) >>= fun (alreadyThere, initialDesc) ->
Abort.check id;
+ begin if alreadyThere then
+ let childNames =
+ Update.NameMap.fold (fun nm _ l -> nm :: l) children [] in
+ deleteSpuriousChildren rootTo (workingDir, pTo, childNames)
+ else
+ Lwt.return ()
+ end >>= fun () ->
+ Abort.check id;
let runningThreads = ref [] in
Lwt.catch
(fun () ->
- Update.NameMap.iter
- (fun name child ->
- let thread =
- copyRec (Path.child pFrom name)
- (Path.child pTo name)
- (Path.child realPTo name)
- child
- in
- runningThreads := thread :: !runningThreads)
- children;
- Lwt_util.join !runningThreads)
+ let ch =
+ Update.NameMap.mapi
+ (fun name child ->
+ let thread : Update.archive Lwt.t =
+ copyRec (Path.child pFrom name)
+ (Path.child pTo name)
+ (Path.child realPTo name)
+ child
+ in
+ runningThreads := thread :: !runningThreads;
+ thread)
+ children
+ in
+ Update.NameMap.fold
+ (fun nm arThr chThr ->
+ arThr >>= fun ar ->
+ chThr >>= fun ch ->
+ Lwt.return (Update.NameMap.add nm ar ch))
+ ch
+ (Lwt.return Update.NameMap.empty))
(fun e ->
(* If one thread fails (in a non-fatal way), we wait for
all other threads to terminate before continuing *)
- if not (Abort.testException e) then Abort.file id;
- match e with
- Util.Transient _ ->
- let e = ref e in
- Lwt_util.iter
- (fun act ->
- Lwt.catch
- (fun () -> act)
- (fun e' ->
- match e' with
- Util.Transient _ ->
- if Abort.testException !e then e := e';
- Lwt.return ()
- | _ ->
- Lwt.fail e'))
- !runningThreads >>= (fun () ->
- Lwt.fail !e)
- | _ ->
- Lwt.fail e) >>= (fun () ->
+ Abort.mergeErrors id e !runningThreads)
+ >>= fun newChildren ->
Lwt_util.run_in_region copyReg 1 (fun () ->
(* We use the actual file permissions so as to preserve
inherited bits *)
Abort.check id;
setPropRemote rootTo
- (workingDir, pTo, `Set initialDesc, desc))))
+ (workingDir, pTo, `Set initialDesc, desc)) >>= fun () ->
+ Lwt.return (Update.ArchiveDir (desc, newChildren))
| Update.NoArchive ->
assert false
in
- (* BCP (6/08): We used to have an unwindProtect here that would *always* do the
- final performDelete. This was removed so that failed partial transfers can
- be restarted. We instead remove individual failing files (not
- directories) inside replaceArchive. *)
Update.transaction (fun id ->
(* Update the archive on the source replica (but don't commit
the changes yet) and return the part of the new archive
corresponding to this path *)
Update.updateArchive rootFrom pathFrom uiFrom id
>>= fun (localPathFrom, archFrom) ->
- let make_backup =
+ let make_backup = (* FIX: this call should probably be removed... *)
(* Perform (asynchronously) a backup of the destination files *)
Update.updateArchive rootTo pathTo uiTo id
in
- copyRec localPathFrom tempPathTo realPathTo archFrom >>= fun () ->
+ copyRec localPathFrom tempPathTo realPathTo archFrom >>= fun archTo ->
make_backup >>= fun _ ->
- Update.replaceArchive
- rootTo pathTo (Some (workingDir, tempPathTo))
- archFrom id true true >>= fun _ ->
+ Update.replaceArchive rootTo pathTo archTo id >>= fun _ ->
rename rootTo pathTo localPathTo workingDir tempPathTo realPathTo uiTo)
(* ------------------------------------------------------------ *)
@@ -531,7 +524,8 @@
(Update.translatePath root2 path2 >>= (fun path2 ->
Copy.file root2 path2 root1 workingDir tmppath realPath
`Copy (Props.setLength Props.fileSafe (Props.length desc2))
- fp2 ress2 id));
+ fp2 None ress2 id) >>= fun info ->
+ Lwt.return ());
displayDiff
(Fspath.concat workingDir realPath)
(Fspath.concat workingDir tmppath);
@@ -549,7 +543,8 @@
(* Note that we don't need the resource fork *)
Copy.file root1 path1 root2 workingDir tmppath realPath
`Copy (Props.setLength Props.fileSafe (Props.length desc1))
- fp1 ress1 id));
+ fp1 None ress1 id >>= fun info ->
+ Lwt.return ()));
displayDiff
(Fspath.concat workingDir tmppath)
(Fspath.concat workingDir realPath);
@@ -635,9 +630,9 @@
let newprops = Props.setLength propsTo (Props.length info.Fileinfo.desc) in
Copy.file
(Local, fspathFrom) pathFrom rootTo workingDirForCopy tempPathTo realPathTo
- `Copy newprops fp stamp id >>= (fun () ->
- rename rootTo pathTo localPathTo workingDirForCopy tempPathTo realPathTo
- uiTo ))
+ `Copy newprops fp None stamp id >>= fun info ->
+ rename rootTo pathTo localPathTo workingDirForCopy tempPathTo realPathTo
+ uiTo )
let keeptempfilesaftermerge =
Prefs.createBool
@@ -700,12 +695,14 @@
Lwt_unix.run
(Copy.file
root1 localPath1 root1 workingDirForMerge working1 basep
- `Copy desc1 fp1 ress1 id);
+ `Copy desc1 fp1 None ress1 id >>= fun info ->
+ Lwt.return ());
Lwt_unix.run
(Update.translatePath root2 path >>= (fun path ->
Copy.file
root2 path root1 workingDirForMerge working2 basep
- `Copy desc2 fp2 ress2 id));
+ `Copy desc2 fp2 None ress2 id) >>= fun info ->
+ Lwt.return ());
(* retrieve the archive for this file, if any *)
let arch =
@@ -912,13 +909,11 @@
Osx.stamp infoarch.osX) in
Update.transaction
(fun transid ->
- Update.replaceArchive root1 path
- (Some(workingDirForMerge, workingarch))
- new_archive_entry transid false false >>= (fun _ ->
- Update.replaceArchive root2 path
- (Some(workingDirForMerge, workingarch))
- new_archive_entry transid false false >>= (fun _ ->
- Lwt.return ())))
+ Update.replaceArchive root1 path new_archive_entry transid
+ >>= fun _ ->
+ Update.replaceArchive root2 path new_archive_entry transid
+ >>= fun _ ->
+ Lwt.return ())
end else
(Lwt.return ()) )))) )
(fun _ ->
Modified: trunk/src/mkProjectInfo.ml
===================================================================
--- trunk/src/mkProjectInfo.ml 2009-06-09 08:53:55 UTC (rev 349)
+++ trunk/src/mkProjectInfo.ml 2009-06-09 15:41:29 UTC (rev 350)
@@ -5,8 +5,8 @@
let projectName = "unison"
let majorVersion = 2
-let minorVersion = 34
-let pointVersionOrigin = 332 (* Revision that corresponds to point version 0 *)
+let minorVersion = 35
+let pointVersionOrigin = 349 (* Revision that corresponds to point version 0 *)
(* Documentation:
This is a program to construct a version of the form Major.Minor.Point,
@@ -159,3 +159,4 @@
+
Modified: trunk/src/remote.ml
===================================================================
--- trunk/src/remote.ml 2009-06-09 08:53:55 UTC (rev 349)
+++ trunk/src/remote.ml 2009-06-09 15:41:29 UTC (rev 350)
@@ -35,14 +35,17 @@
(****)
-let intSize = 4
+let intSize = 5
+let intHash x = ((x * 791538121) lsr 23 + 17) land 255
+
let encodeInt m =
let int_buf = Bytearray.create intSize in
int_buf.{0} <- Char.chr ( m land 0xff);
int_buf.{1} <- Char.chr ((m lsr 8) land 0xff);
int_buf.{2} <- Char.chr ((m lsr 16) land 0xff);
int_buf.{3} <- Char.chr ((m lsr 24) land 0xff);
+ int_buf.{4} <- Char.chr (intHash m);
(int_buf, 0, intSize)
let decodeInt int_buf i =
@@ -50,7 +53,13 @@
let b1 = Char.code (int_buf.{i + 1}) in
let b2 = Char.code (int_buf.{i + 2}) in
let b3 = Char.code (int_buf.{i + 3}) in
- ((b3 lsl 24) lor (b2 lsl 16) lor (b1 lsl 8) lor b0)
+ let m = (b3 lsl 24) lor (b2 lsl 16) lor (b1 lsl 8) lor b0 in
+ if Char.code (int_buf.{i + 4}) <> intHash m then
+ raise (Util.Fatal
+ "Protocol error: corrupted message received;\n\
+ if it happens to you in a repeatable way, \n\
+ please post a report on the unison-users mailing list.");
+ m
(*************************************************************************)
(* LOW-LEVEL IO *)
@@ -171,6 +180,8 @@
Lwt.return ()
end
+let bufReg = Lwt_util.make_region 1
+
let rec fill_buffer conn l =
match l with
(s, pos, len) :: rem ->
@@ -182,6 +193,29 @@
| [] ->
Lwt.return ()
+let fill_buffer conn l =
+ Lwt_util.run_in_region bufReg 1 (fun () -> fill_buffer conn l)
+let send_output conn =
+ Lwt_util.run_in_region bufReg 1 (fun () -> send_output conn)
+
+
+let blockedStream = ref None
+
+let rec streamWaitForWrite conn =
+ if conn.canWrite then Lwt.return () else begin
+ debugE (fun() -> Util.msg "Stream: waiting for write token\n");
+ let w = Lwt.wait () in
+ blockedStream := Some w;
+ w >>= fun () ->
+ debugE (fun() -> Util.msg "Stream: restarting\n");
+ streamWaitForWrite conn
+ end
+
+let restartStream () =
+ match !blockedStream with
+ Some w -> blockedStream := None; Lwt.wakeup w ()
+ | None -> ()
+
(*
Flow-control mechanism (only active under windows).
Only one side is allowed to send message at any given time.
@@ -196,6 +230,13 @@
*)
let needFlowControl = windowsHack
+let rec flush_buffer_simpl conn =
+ if conn.outputLength > 0 then
+ send_output conn >>= fun () ->
+ flush_buffer_simpl conn
+ else
+ Lwt.return ()
+
(* Loop until the output buffer is empty *)
let rec flush_buffer conn =
if conn.tokens <= 0 && conn.canWrite then begin
@@ -235,23 +276,27 @@
(* We wait a bit before flushing everything, so that other packets
send just afterwards can be coalesced *)
Lwt_unix.yield () >>= (fun () ->
- try
- ignore (Queue.peek conn.outputQueue);
+ if not (Queue.is_empty conn.outputQueue) then
dump_rec conn
- with Queue.Empty ->
- flush_buffer conn)
+ else begin
+ flush_buffer conn >>= fun () ->
+ if not (Queue.is_empty conn.outputQueue) then
+ signalSomethingToWrite conn;
+ Lwt.return ()
+ end)
(* Start the thread that write all pending messages, if this thread is
not running at this time *)
-let signalSomethingToWrite conn =
+and signalSomethingToWrite conn =
if not conn.canWrite && conn.pendingOutput then
debugE
(fun () -> Util.msg "Something to write, but no write token (%d)\n"
conn.tokens);
- if conn.pendingOutput = false && conn.canWrite then begin
+ if not conn.pendingOutput && conn.canWrite then begin
conn.pendingOutput <- true;
Lwt.ignore_result (dump_rec conn)
- end
+ end;
+ if conn.canWrite then restartStream ()
(* Add a message to the output queue and schedule its emission *)
(* A message is a list of fragments of messages, represented by triplets
@@ -532,11 +577,17 @@
((Bytearray.t * int * int) list -> (Bytearray.t * int * int) list) Lwt.t
let serverCmds = ref (Util.StringMap.empty : servercmd Util.StringMap.t)
+type serverstream =
+ connection -> Bytearray.t -> unit
+let serverStreams = ref (Util.StringMap.empty : serverstream Util.StringMap.t)
+
type header =
NormalResult
| TransientExn of string
| FatalExn of string
| Request of string
+ | Stream of string
+ | StreamAbort
let ((marshalHeader, unmarshalHeader) : header marshalingFunctions) =
makeMarshalingFunctions defaultMarshalingFunctions "rsp"
@@ -562,6 +613,38 @@
| e ->
Lwt.fail e)
+let streamAbortedSrc = ref 0
+let streamAbortedDst = ref false
+
+let streamError = Hashtbl.create 7
+
+let abortStream conn id =
+ if not !streamAbortedDst then begin
+ streamAbortedDst := true;
+ let request = encodeInt id :: marshalHeader StreamAbort [] in
+ fill_buffer conn request >>= fun () ->
+ flush_buffer_simpl conn
+ end else
+ Lwt.return ()
+
+let processStream conn id cmdName buf =
+ let id = decodeInt id 0 in
+ if Hashtbl.mem streamError id then
+ abortStream conn id
+ else begin
+ begin try
+ let cmd =
+ try Util.StringMap.find cmdName !serverStreams
+ with Not_found -> raise (Util.Fatal (cmdName ^ " not registered!"))
+ in
+ cmd conn buf;
+ Lwt.return ()
+ with e ->
+ Hashtbl.add streamError id e;
+ abortStream conn id
+ end
+ end
+
(* Message ids *)
type msgId = int
module MsgIdMap = Map.Make (struct type t = msgId let compare = compare end)
@@ -628,6 +711,15 @@
debugV (fun() -> Util.msg "receive: Fatal remote error '%s']" s);
Lwt.wakeup_exn (find_receiver num_id) (Util.Fatal ("Server: " ^ s));
receive conn
+ | Stream cmdName ->
+ receivePacket conn >>= fun buf ->
+ if conn.flowControl then conn.tokens <- conn.tokens - 1;
+ processStream conn id cmdName buf >>= fun () ->
+ receive conn
+ | StreamAbort ->
+ if conn.flowControl then conn.tokens <- conn.tokens - 1;
+ streamAbortedSrc := num_id;
+ receive conn
end)
end))
@@ -731,7 +823,76 @@
| _ -> let conn = hostConnection (hostOfRoot localRoot) in
client0 conn args
+let streamReg = Lwt_util.make_region 1
+let streamingActivated =
+ Prefs.createBool "stream" true
+ ("!use a streaming protocol for transferring file contents")
+ "When this preference is set, Unison will use an experimental \
+ streaming protocol for transferring file contents more efficiently. \
+ The default value is \\texttt{true}."
+
+let registerStreamCmd
+ (cmdName : string)
+ marshalingFunctionsArgs
+ (serverSide : connection -> 'a -> unit)
+ =
+ let cmd =
+ registerSpecialServerCmd
+ cmdName marshalingFunctionsArgs defaultMarshalingFunctions
+ (fun conn v -> serverSide conn v; Lwt.return ())
+ in
+ let ping =
+ registerServerCmd (cmdName ^ "Ping")
+ (fun conn (id : int) ->
+ try
+ let e = Hashtbl.find streamError id in
+ Hashtbl.remove streamError id;
+ streamAbortedDst := false;
+ Lwt.fail e
+ with Not_found ->
+ Lwt.return ())
+ in
+ (* Check that this command name has not already been bound *)
+ if (Util.StringMap.mem cmdName !serverStreams) then
+ raise (Util.Fatal (cmdName ^ " already registered!"));
+ (* Create marshaling and unmarshaling functions *)
+ let ((marshalArgs,unmarshalArgs) : 'a marshalingFunctions) =
+ makeMarshalingFunctions marshalingFunctionsArgs (cmdName ^ "-str") in
+ (* Create a server function and remember it *)
+ let server conn buf =
+ let args = unmarshalArgs buf in
+ serverSide conn args
+ in
+ serverStreams := Util.StringMap.add cmdName server !serverStreams;
+ (* Create a client function and return it *)
+ let client conn id serverArgs =
+ if !streamAbortedSrc = id then raise (Util.Transient "Streaming aborted");
+ streamWaitForWrite conn >>= fun () ->
+ debugE (fun () -> Util.msg "Sending stream chunk (id: %d)\n" id);
+ if !streamAbortedSrc = id then raise (Util.Transient "Streaming aborted");
+ let request =
+ encodeInt id ::
+ marshalHeader (Stream cmdName) (marshalArgs serverArgs [])
+ in
+ fill_buffer conn request
+ in
+ fun conn sender ->
+ if not (Prefs.read streamingActivated) then
+ sender (fun v -> cmd conn v)
+ else begin
+ (* At most one active stream at a time *)
+ let id = newMsgId () in (* Message ID *)
+ Lwt.try_bind
+ (fun () ->
+ Lwt_util.run_in_region streamReg 1
+ (fun () ->
+ Lwt_unix.yield () >>= fun () ->
+ sender (fun v -> client conn id v)))
+ (fun v -> ping conn id >>= fun () -> Lwt.return v)
+ (fun e -> ping conn id >>= fun () -> Lwt.fail e)
+ end
+
(****************************************************************************
BUILDING CONNECTIONS TO THE SERVER
****************************************************************************)
Modified: trunk/src/remote.mli
===================================================================
--- trunk/src/remote.mli 2009-06-09 08:53:55 UTC (rev 349)
+++ trunk/src/remote.mli 2009-06-09 15:41:29 UTC (rev 350)
@@ -104,3 +104,13 @@
-> Common.root (* other root *)
-> 'a (* additional arguments *)
-> 'b Lwt.t (* result *)
+
+val streamingActivated : bool Prefs.t
+
+val registerStreamCmd :
+ string ->
+ ('a ->
+ (Bytearray.t * int * int) list -> (Bytearray.t * int * int) list * int) *
+ (Bytearray.t -> int -> 'a) ->
+ (connection -> 'a -> unit) ->
+ connection -> (('a -> unit Lwt.t) -> 'b Lwt.t) -> 'b Lwt.t
Modified: trunk/src/transfer.ml
===================================================================
--- trunk/src/transfer.ml 2009-06-09 08:53:55 UTC (rev 349)
+++ trunk/src/transfer.ml 2009-06-09 15:41:29 UTC (rev 350)
@@ -342,16 +342,6 @@
in
iter 0 arg 0 0
- let rec rev_split_rec accu1 accu2 n l =
- if n = 100000 then
- rev_split_rec (accu2 :: accu1) [] 0 l
- else
- match l with
- [] -> accu2 :: accu1
- | x :: r -> rev_split_rec accu1 (x :: accu2) (n + 1) r
-
- let rev_split l = rev_split_rec [] [] 0 l
-
(* Given a block size, get blocks from the old file and compute a
checksum and a fingerprint for each one. *)
let rsyncPreprocess infd =
@@ -369,7 +359,7 @@
(* Limit the number of block so that there is no overflow in
encodeInt3 *)
let rev_bi = blockIter infd addBlock [] (256*256*256) in
- let bi = rev_split rev_bi in
+ let bi = Safelist.rev rev_bi in
debugLog (fun() -> Util.msg "%d blocks\n" (Safelist.length bi));
Trace.showTimer timer;
bi
@@ -441,11 +431,6 @@
let hash checksum = checksum
- let rec sigLength sigs =
- match sigs with
- [] -> 0
- | x :: r -> Safelist.length x + sigLength r
-
(* Compute the hash table length as a function of the number of blocks *)
let hashTableLength signatures =
let rec upperPowerOfTwo n n2 =
@@ -454,23 +439,21 @@
else
upperPowerOfTwo n (2 * n2)
in
- 2 * (upperPowerOfTwo (sigLength signatures) 32)
+ 2 * (upperPowerOfTwo (Safelist.length signatures) 32)
(* Hash the block signatures into the hash table *)
let hashSig hashTableLength signatures =
let hashTable = Array.make hashTableLength [] in
- let rec addList k l l' =
- match l, l' with
- [], [] ->
+ let rec addList k l =
+ match l with
+ [] ->
()
- | [], r :: r' ->
- addList k r r'
- | ((cs, fp) :: r), _ ->
+ | (cs, fp) :: r ->
let h = (hash cs) land (hashTableLength - 1) in
hashTable.(h) <- (k, cs, fp)::(hashTable.(h));
- addList (k + 1) r l'
+ addList (k + 1) r
in
- addList 0 [] signatures;
+ addList 0 signatures;
hashTable
(* Given a key, retrieve the corresponding entry in the table *)
Modified: trunk/src/transfer.mli
===================================================================
--- trunk/src/transfer.mli 2009-06-09 08:53:55 UTC (rev 349)
+++ trunk/src/transfer.mli 2009-06-09 15:41:29 UTC (rev 350)
@@ -80,7 +80,7 @@
(* Compute block informations from the old file *)
val rsyncPreprocess :
in_channel (* old file descriptor *)
- -> rsync_block_info list
+ -> rsync_block_info
(* Interpret a transfer instruction *)
val rsyncDecompress :
@@ -95,7 +95,7 @@
(* Using block informations, parse the new file and send transfer
instructions accordingly *)
val rsyncCompress :
- rsync_block_info list
+ rsync_block_info
(* block info received from the destination *)
-> in_channel (* new file descriptor *)
-> Uutil.Filesize.t (* source file length *)
Modified: trunk/src/transport.ml
===================================================================
--- trunk/src/transport.ml 2009-06-09 08:53:55 UTC (rev 349)
+++ trunk/src/transport.ml 2009-06-09 15:41:29 UTC (rev 350)
@@ -87,7 +87,12 @@
let doAction (fromRoot,toRoot) path fromContents toContents id =
Lwt_util.resize_region actionReg (Prefs.read maxthreads);
- Lwt_util.resize_region Files.copyReg (Prefs.read maxthreads);
+ (* When streaming, we can transfer many file simultaneously:
+ as the contents of only one file is transferred in one direction
+ at any time, little ressource is consumed this way. *)
+ Lwt_util.resize_region Files.copyReg
+ (if Prefs.read Remote.streamingActivated then 4000 else
+ Prefs.read maxthreads);
Lwt_util.run_in_region actionReg 1 (fun () ->
if not !Trace.sendLogMsgsToStderr then
Trace.statusDetail (Path.toString path);
Modified: trunk/src/update.ml
===================================================================
--- trunk/src/update.ml 2009-06-09 08:53:55 UTC (rev 349)
+++ trunk/src/update.ml 2009-06-09 15:41:29 UTC (rev 350)
@@ -1830,46 +1830,7 @@
Tree.map (fun n -> n) (fun (uc1,uc2) -> uc2) equals])
end
-let rec replaceArchiveRec fspath path arch paranoid deleteBadTempFiles =
- match arch with
- ArchiveDir (desc, children) ->
- ArchiveDir (desc,
- NameMap.mapi
- (fun nm a ->
- replaceArchiveRec
- fspath (Path.child path nm) a paranoid deleteBadTempFiles)
- children)
- | ArchiveFile (desc, dig, stamp, ress) ->
- if paranoid then begin
- (* Paranoid check: recompute the file's digest to match it with
- the archive's *)
- let info = Fileinfo.get false fspath path in
- let dig' = Os.fingerprint fspath path info in
- let ress' = Osx.stamp info.Fileinfo.osX in
- if dig' <> dig then begin
- let savepath = Path.addSuffixToFinalName path "-bad" in
- (* if deleteBadTempFiles then Os.delete fspath path; *)
- if deleteBadTempFiles then
- Os.rename "save temp" fspath path fspath savepath;
- raise (Util.Transient (Printf.sprintf
- "The file %s was incorrectly transferred (fingerprint mismatch in %s)%s"
- (Path.toString path)
- (Os.reasonForFingerprintMismatch dig dig')
- (if deleteBadTempFiles
- then " -- temp file saved as" ^ Path.toString savepath
- else "")));
- end;
- ArchiveFile (Props.override info.Fileinfo.desc desc,
- dig, Fileinfo.stamp info, ress')
- end else begin
- ArchiveFile (desc, dig, stamp, ress)
- end
- | ArchiveSymlink l ->
- ArchiveSymlink l
- | NoArchive ->
- arch
-
-let replaceArchiveLocal fspath pathTo location arch id paranoid deleteBadTempFiles =
+let replaceArchiveLocal fspath pathTo arch id =
debug (fun() -> Util.msg
"replaceArchiveLocal %s %s\n"
(Fspath.toDebugString fspath)
@@ -1877,20 +1838,12 @@
);
let root = thisRootsGlobalName fspath in
let localPath = translatePathLocal fspath pathTo in
- let (workingDir, tempPathTo) =
- match location with
- None -> (fspath, localPath)
- | Some loc -> loc
- in
- let newArch =
- replaceArchiveRec workingDir tempPathTo arch paranoid deleteBadTempFiles in
let commit () =
debug (fun() -> Util.msg "replaceArchiveLocal: committing\n");
- let _ = Stasher.stashCurrentVersion fspath localPath (Some tempPathTo) in
let archive = getArchive root in
let archive, () =
updatePathInArchive archive fspath Path.empty pathTo
- (fun _ _ _ -> newArch, ())
+ (fun _ _ _ -> arch, ())
in
setArchiveLocal root archive
in
@@ -1900,13 +1853,11 @@
let replaceArchiveOnRoot =
Remote.registerRootCmd
"replaceArchive"
- (fun (fspath, (pathTo, location, arch, id, paranoid, deleteBadTempFiles)) ->
- Lwt.return (replaceArchiveLocal fspath pathTo location arch
- id paranoid deleteBadTempFiles))
+ (fun (fspath, (pathTo, arch, id)) ->
+ Lwt.return (replaceArchiveLocal fspath pathTo arch id))
-let replaceArchive root pathTo location archive id paranoid deleteBadTempFiles =
- replaceArchiveOnRoot root
- (pathTo, location, archive, id, paranoid, deleteBadTempFiles)
+let replaceArchive root pathTo archive id =
+ replaceArchiveOnRoot root (pathTo, archive, id)
(* Update the archive to reflect
- the last observed state of the file on disk (ui)
Modified: trunk/src/update.mli
===================================================================
--- trunk/src/update.mli 2009-06-09 08:53:55 UTC (rev 349)
+++ trunk/src/update.mli 2009-06-09 15:41:29 UTC (rev 350)
@@ -41,8 +41,7 @@
(Path.local * archive) Lwt.t
(* Replace a part of an archive by another archive *)
val replaceArchive :
- Common.root -> Path.t -> (Fspath.t * Path.local) option ->
- archive -> transaction -> bool -> bool -> Path.local Lwt.t
+ Common.root -> Path.t -> archive -> transaction -> Path.local Lwt.t
(* Update only some permissions *)
val updateProps :
Common.root -> Path.t -> Props.t option -> Common.updateItem ->
More information about the Unison-hackers
mailing list