Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| f2f07a0f9a | |||
| e889bd4e88 | |||
| 908adf51a1 | |||
| 2c3e341343 | |||
| 21c21e17dd | |||
| 7e73567f2a | |||
| 6405d41c20 | |||
| 10d84909c0 |
117
README.md
117
README.md
@ -1,9 +1,27 @@
|
||||
# WARNING
|
||||
THIS TOOL IS COMPLETELY VIBE-CODED!!!
|
||||
|
||||
THESE LINES ARE THE ONLY ONE I HAVE WRITTEN MYSELF!!!!
|
||||
|
||||
I SWEAR I DID NOT READ MORE THAN 20 LINES OF CODE!
|
||||
|
||||
USE AT YOUR OWN RISK (AS ALWAYS)
|
||||
|
||||
|
||||
I needed a Tool to sync 2 Files from a to b with differential transfer, all managed from a workstation c.
|
||||
|
||||
Maybe i use this to learn go...
|
||||
|
||||
Looks nice, builds fast, runs everywhere with one Binary.
|
||||
|
||||
|
||||
|
||||
# clonetool
|
||||
|
||||
Block-level sync for a large file or block device between two machines (or
|
||||
two paths on the same machine), driven from a third, passive "manager"
|
||||
machine. Single static Go binary, no runtime dependencies beyond the
|
||||
system `ssh` client for remote endpoints.
|
||||
system `ssh` client for remote endpoints. Linux only.
|
||||
|
||||
## How it works
|
||||
|
||||
@ -44,13 +62,25 @@ system `ssh` client for remote endpoints.
|
||||
is used at all — the source agent spawns the write-side helper as a
|
||||
plain local subprocess.
|
||||
- **Self-deploy:** if a remote endpoint has no runnable `clonetool` on
|
||||
`PATH` (or wherever `--remote-bin` points), the manager streams *this*
|
||||
binary to `~/.clonetool/bin/clonetool` on that host over the existing
|
||||
SSH connection and uses it — no install, no root. Disable with
|
||||
`--deploy=false`. If the copied binary won't execute there (wrong CPU
|
||||
architecture) the error says so; build one for the remote's arch
|
||||
`PATH` (or wherever `--remote-bin` points), **or the one that's there is
|
||||
a different build than this binary** (see below), the manager streams
|
||||
*this* binary to `~/.clonetool/bin/clonetool` on that host over the
|
||||
existing SSH connection and uses it — no install, no root. Disable with
|
||||
`--deploy=false` (a stale or missing binary is then an error, except a
|
||||
merely-stale `--remote-bin` is used anyway with a warning, since there's
|
||||
no way to correct it). If the copied binary won't execute there (wrong
|
||||
CPU architecture) the error says so; build one for the remote's arch
|
||||
(`CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build`) and put it on `PATH`
|
||||
or pass `--remote-bin`.
|
||||
- **Version/staleness check:** every binary has a build timestamp baked in
|
||||
(see Build below) and `clonetool version` prints it. Before using a
|
||||
remote `clonetool`, the manager runs its `version` there and compares
|
||||
that timestamp against its own — a mismatch (e.g. the remote was
|
||||
self-deployed from an older build of this tool) is treated the same as
|
||||
"not runnable" and triggers the self-deploy above, so the remote binary
|
||||
is kept in sync with whatever you're running locally. Only the build
|
||||
timestamp is compared, not `GOARCH` — that is expected to differ across
|
||||
a cross-compiled deploy.
|
||||
- **sudo:** if reading or writing an endpoint that is a **block device**
|
||||
fails with a permission error, `--sudo=auto` (the default) transparently
|
||||
restarts that side's agent — and the helper it spawns on the peer —
|
||||
@ -61,8 +91,8 @@ system `ssh` client for remote endpoints.
|
||||
start; `--sudo=never` never does.
|
||||
- Sizing rules:
|
||||
- Destination is a **block device**: it can't be resized, so if the
|
||||
source is larger the job fails; otherwise exactly `min(source, dest)`
|
||||
bytes are synced and the remainder of the device is left untouched.
|
||||
source is larger the job fails; otherwise exactly the source's size is
|
||||
synced and the remainder of the device is left untouched.
|
||||
- Destination is a **regular file**: it's truncated (created if
|
||||
missing) to exactly the source's size, growing or shrinking it.
|
||||
Shrinking an existing non-empty file prompts for confirmation unless
|
||||
@ -74,54 +104,27 @@ system `ssh` client for remote endpoints.
|
||||
CGO_ENABLED=0 go build -o clonetool .
|
||||
```
|
||||
|
||||
Cross-compile for another OS/arch by setting `GOOS`/`GOARCH` (no cgo, so
|
||||
these all work from any host):
|
||||
Cross-compile for another architecture by setting `GOARCH` (no cgo, so
|
||||
this works from any host):
|
||||
|
||||
```
|
||||
CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -o clonetool .
|
||||
CGO_ENABLED=0 GOOS=windows GOARCH=amd64 go build -o clonetool.exe .
|
||||
CGO_ENABLED=0 GOOS=darwin GOARCH=arm64 go build -o clonetool-darwin .
|
||||
CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -o clonetool .
|
||||
```
|
||||
|
||||
`./build.sh` writes all three of the above (plus linux/amd64) into `dist/`.
|
||||
`./build.sh` writes linux/amd64 and linux/arm64 binaries into `dist/`. It
|
||||
also stamps every binary it builds with the same build timestamp (via
|
||||
`-ldflags -X main.buildTime=...`), which is how the manager tells a stale
|
||||
self-deployed remote binary apart from a current one (see "Version/staleness
|
||||
check" above) — build all your binaries for a given release with one
|
||||
`./build.sh` run so they compare equal to each other. A plain `go build`
|
||||
with no `-ldflags` leaves the timestamp at `"dev"`, which still works for
|
||||
this check (it only ever compares equal to another literal copy of the same
|
||||
`dev` binary, never to a real timestamp).
|
||||
|
||||
Copy the resulting binary to the manager host. Source and destination
|
||||
hosts get it automatically (see self-deploy above), or place it yourself
|
||||
and point `--remote-bin` at it. The binary is architecture-specific —
|
||||
cross-compile if your hosts differ. Self-deploy streams a POSIX shell
|
||||
script over SSH, so a **Windows** host can be the manager or a local
|
||||
endpoint but cannot be an automatic deploy target — put `clonetool.exe`
|
||||
on it yourself and point `--remote-bin` at it.
|
||||
|
||||
## Windows
|
||||
|
||||
clonetool runs on Windows and can clone **physical drives and volumes**,
|
||||
not just files:
|
||||
|
||||
```
|
||||
# Whole disk to an image file
|
||||
clonetool sync --source \\.\PhysicalDrive2 --dest D:\backup\disk2.img
|
||||
|
||||
# Image file back onto a disk (must not be larger than the disk)
|
||||
clonetool sync --source D:\backup\disk2.img --dest \\.\PhysicalDrive2
|
||||
|
||||
# A single volume
|
||||
clonetool sync --source \\.\E: --dest \\.\F:
|
||||
```
|
||||
|
||||
- Raw-disk paths are `\\.\PhysicalDrive<n>` (whole disk) or `\\.\<X>:` (a
|
||||
volume). `\\?\` also works. Forward slashes are accepted.
|
||||
- **Run from an elevated (Administrator) console** to open a raw disk.
|
||||
There is no `sudo` fallback on Windows; `--sudo` is ignored. A
|
||||
permission error tells you to elevate.
|
||||
- A raw-disk **destination** should have no mounted filesystem in use
|
||||
(take the disk offline in Disk Management, or target a volume that
|
||||
nothing else has open) — Windows blocks writes to a disk region owned by
|
||||
a mounted volume. Reading a live disk as the **source** is fine.
|
||||
- Raw-disk I/O must be sector-aligned. `--block-size` must be a multiple
|
||||
of the drive's sector size (512 or 4096); the default 4M is. clonetool
|
||||
handles the final partial block itself.
|
||||
- Device size is read with `IOCTL_DISK_GET_LENGTH_INFO`.
|
||||
cross-compile if your hosts differ.
|
||||
|
||||
## Usage
|
||||
|
||||
@ -136,6 +139,9 @@ clonetool sync --source LOC --dest LOC [options]
|
||||
# Same machine
|
||||
clonetool sync --source /dev/sda --dest /dev/sdb
|
||||
|
||||
# Whole disk to an image file
|
||||
clonetool sync --source /dev/sda --dest /srv/sda.img
|
||||
|
||||
# Two remote machines, orchestrated from a third
|
||||
clonetool sync --source db1:/dev/vdb --dest backup-host:/srv/db1.img
|
||||
|
||||
@ -158,14 +164,19 @@ Options:
|
||||
| `--remote-bin` | `clonetool` | Path to clonetool on remote hosts. |
|
||||
| `--manager-host` | local hostname | Address a peer should use to reach this machine, needed only when source or dest is local to the manager *and* the other side is remote and ends up needing to dial back in (pull fallback). |
|
||||
|
||||
`clonetool version` prints the binary's `GOOS/GOARCH` (used internally for
|
||||
the self-deploy check).
|
||||
`clonetool version` prints the binary's `GOOS/GOARCH` and build timestamp,
|
||||
e.g. `clonetool linux/amd64 build=2024-06-01T12:00:00Z` (the timestamp is
|
||||
used internally for the self-deploy staleness check above).
|
||||
|
||||
The status line shows read throughput for both sides and write throughput
|
||||
for the destination separately: `rd(src)` is the source reading and
|
||||
comparing its blocks, `rd(dst)` is the destination fingerprinting its
|
||||
current content (only while the trailing `scan H/M` segment is present),
|
||||
and `wr(dst)` is the destination actually writing changed blocks.
|
||||
|
||||
## Caveats
|
||||
|
||||
- Block-device size detection is implemented on Linux (`BLKGETSIZE64`) and
|
||||
Windows (`IOCTL_DISK_GET_LENGTH_INFO`). On other systems only regular
|
||||
files can be synced.
|
||||
- Block-device size detection uses `BLKGETSIZE64`; the tool is Linux only.
|
||||
- If a destination path doesn't exist yet, it's created as a regular
|
||||
file — clonetool won't create device nodes, so double-check device
|
||||
paths for typos before running.
|
||||
|
||||
113
agent.go
113
agent.go
@ -19,6 +19,7 @@ func cmdAgent(args []string) error {
|
||||
role := fs.String("role", "", "control|sink|source-stream (internal)")
|
||||
path := fs.String("path", "", "path to read/write")
|
||||
size := fs.Int64("size", 0, "total sync size in bytes")
|
||||
base := fs.Int64("base", 0, "byte offset the window starts at")
|
||||
blockSize := fs.Int64("block-size", defaultBlockSize, "block size in bytes")
|
||||
if err := fs.Parse(args); err != nil {
|
||||
return err
|
||||
@ -28,11 +29,11 @@ func cmdAgent(args []string) error {
|
||||
case "control":
|
||||
return runControlAgent()
|
||||
case "sink":
|
||||
return runSinkRole(*path, *size, *blockSize)
|
||||
return runSinkRole(*path, *base, *size, *blockSize)
|
||||
case "source-stream":
|
||||
return runSourceStreamRole(*path, *size, *blockSize)
|
||||
return runSourceStreamRole(*path, *base, *size, *blockSize)
|
||||
default:
|
||||
return fmt.Errorf("agent: unknown or missing --role %q (want control|sink|source-stream)", *role)
|
||||
return fmt.Errorf("agent: unknown or missing --role %q", *role)
|
||||
}
|
||||
}
|
||||
|
||||
@ -127,6 +128,7 @@ func runPushDriver(req CtrlMsg, out *FrameWriter) {
|
||||
tailArgs := []string{
|
||||
"agent", "--role", "sink",
|
||||
"--path", req.PeerPath,
|
||||
"--base", strconv.FormatInt(req.PeerBase, 10),
|
||||
"--size", strconv.FormatInt(req.Size, 10),
|
||||
"--block-size", strconv.FormatInt(req.BlockSize, 10),
|
||||
}
|
||||
@ -231,7 +233,7 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
var pendingMu sync.Mutex
|
||||
pending := make(map[uint64]bool)
|
||||
|
||||
var copied, skipped int64
|
||||
var copied, skipped, srcRead int64
|
||||
var lastProgress time.Time
|
||||
maybeProgress := func() {
|
||||
if time.Since(lastProgress) < 500*time.Millisecond {
|
||||
@ -239,11 +241,24 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
}
|
||||
lastProgress = time.Now()
|
||||
_ = out.WriteJSON(CtrlMsg{
|
||||
Type: msgProgress, Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped),
|
||||
Type: msgProgress, Phase: "xfer",
|
||||
Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped), SrcRead: atomic.LoadInt64(&srcRead),
|
||||
TotalBlocks: blockCount,
|
||||
})
|
||||
}
|
||||
|
||||
// lastActivityNano records the last moment either side made progress: a
|
||||
// frame arriving from the sink (hashes, ACKs, relayed progress) or the
|
||||
// source loop reading and comparing a block. The idle watchdog below
|
||||
// treats a stall as fatal only when *neither* side has moved for a whole
|
||||
// window — otherwise a check-only run, where every block matches and so no
|
||||
// DATA frames and no ACKs are ever exchanged, trips the watchdog purely
|
||||
// because the wire went quiet while the two scans churned through
|
||||
// identical data.
|
||||
var lastActivityNano int64
|
||||
atomic.StoreInt64(&lastActivityNano, time.Now().UnixNano())
|
||||
markActivity := func() { atomic.StoreInt64(&lastActivityNano, time.Now().UnixNano()) }
|
||||
|
||||
// hashCh is sized to hold every block hash so the reader below never
|
||||
// blocks handing hashes off (which, since ACKs share the same stream,
|
||||
// would otherwise be able to deadlock against the in-flight-send limit).
|
||||
@ -274,6 +289,7 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
}
|
||||
return
|
||||
}
|
||||
markActivity()
|
||||
switch typ {
|
||||
case frameBlockHash:
|
||||
idx, h, derr := decodeBlockHashFrame(payload)
|
||||
@ -315,8 +331,10 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
go func() {
|
||||
sendErrCh <- runSourceLoop(sourceLoopParams{
|
||||
File: srcFile, Size: req.Size, BlockSize: req.BlockSize, Align: align, Hashes: hashCh, Out: fw,
|
||||
OnSkip: func(uint64) { atomic.AddInt64(&skipped, 1); maybeProgress() },
|
||||
OnRead: func(uint64) { markActivity(); atomic.AddInt64(&srcRead, 1); maybeProgress() },
|
||||
OnSkip: func(uint64) { markActivity(); atomic.AddInt64(&skipped, 1); maybeProgress() },
|
||||
OnSend: func(idx uint64, _ [32]byte) {
|
||||
markActivity()
|
||||
sem <- struct{}{}
|
||||
pendingMu.Lock()
|
||||
pending[idx] = true
|
||||
@ -332,7 +350,11 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
sendCh := sendErrCh
|
||||
ackCh := ackEvents
|
||||
var fatalErr error
|
||||
idle := time.NewTimer(120 * time.Second)
|
||||
// idleTimeout is how long *both* sides can be silent before the transfer
|
||||
// is declared stalled. The timer may fire sooner; a fire is only fatal
|
||||
// when lastActivityNano confirms nothing has moved for the whole window.
|
||||
const idleTimeout = 120 * time.Second
|
||||
idle := time.NewTimer(idleTimeout)
|
||||
defer idle.Stop()
|
||||
for sendCh != nil || ackCh != nil {
|
||||
if !idle.Stop() {
|
||||
@ -341,7 +363,7 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
default:
|
||||
}
|
||||
}
|
||||
idle.Reset(120 * time.Second)
|
||||
idle.Reset(idleTimeout)
|
||||
select {
|
||||
case sendErr := <-sendCh:
|
||||
sendCh = nil
|
||||
@ -367,7 +389,11 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
maybeProgress()
|
||||
}
|
||||
case <-idle.C:
|
||||
fatalErr = fmt.Errorf("timed out waiting for the sink")
|
||||
idleFor := time.Since(time.Unix(0, atomic.LoadInt64(&lastActivityNano)))
|
||||
if idleFor < idleTimeout {
|
||||
continue // progress somewhere within the window; re-arm and wait
|
||||
}
|
||||
fatalErr = fmt.Errorf("timed out waiting for the sink (no activity for %s)", idleFor.Round(time.Second))
|
||||
}
|
||||
if fatalErr != nil {
|
||||
break
|
||||
@ -384,7 +410,11 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
|
||||
return fmt.Errorf("sink closed the connection with %d block write confirmation(s) still outstanding", n)
|
||||
}
|
||||
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped), TotalBlocks: blockCount})
|
||||
_ = out.WriteJSON(CtrlMsg{
|
||||
Type: msgProgress, Phase: "xfer",
|
||||
Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped), SrcRead: atomic.LoadInt64(&srcRead),
|
||||
TotalBlocks: blockCount,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
@ -400,6 +430,7 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
|
||||
tailArgs := []string{
|
||||
"agent", "--role", "source-stream",
|
||||
"--path", req.PeerPath,
|
||||
"--base", strconv.FormatInt(req.PeerBase, 10),
|
||||
"--size", strconv.FormatInt(req.Size, 10),
|
||||
"--block-size", strconv.FormatInt(req.BlockSize, 10),
|
||||
}
|
||||
@ -448,7 +479,7 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
|
||||
|
||||
hashErrCh := make(chan error, 1)
|
||||
go func() {
|
||||
err := streamHashBlocks(dstFile, req.Size, req.BlockSize, align, func(bh blockHash) error {
|
||||
err := streamHashBlocks(dstFile, req.Base, req.Size, req.BlockSize, align, func(bh blockHash) error {
|
||||
return fw.WriteFrame(frameBlockHash, encodeBlockHashFrame(bh.index, bh.hash))
|
||||
}, scanProgressEmitter(out))
|
||||
if err == nil {
|
||||
@ -462,14 +493,24 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
|
||||
hashErrCh <- err
|
||||
}()
|
||||
|
||||
var written int64
|
||||
var lastWritten time.Time
|
||||
loopErr := runDestLoop(destLoopParams{
|
||||
File: dstFile, BlockSize: req.BlockSize, Align: align, In: fr,
|
||||
File: dstFile, Base: req.Base, BlockSize: req.BlockSize, Align: align, In: fr,
|
||||
OnCtrlMsg: func(m CtrlMsg) {
|
||||
if m.Type == msgProgress {
|
||||
_ = out.WriteJSON(m)
|
||||
}
|
||||
},
|
||||
OnWrite: func(uint64) {
|
||||
written++
|
||||
if time.Since(lastWritten) >= 500*time.Millisecond {
|
||||
lastWritten = time.Now()
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "write", Written: written})
|
||||
}
|
||||
},
|
||||
})
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "write", Written: written})
|
||||
hashErr := <-hashErrCh
|
||||
|
||||
if hashErr != nil {
|
||||
@ -496,7 +537,7 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
|
||||
// destination host. Dumb write endpoint: verify+pwrite+ack per block.
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
func runSinkRole(path string, size, blockSize int64) error {
|
||||
func runSinkRole(path string, base, size, blockSize int64) error {
|
||||
out := NewFrameWriter(os.Stdout)
|
||||
in := NewFrameReader(os.Stdin)
|
||||
|
||||
@ -519,11 +560,33 @@ func runSinkRole(path string, size, blockSize int64) error {
|
||||
// block is always hashed before a write for it can arrive (the source
|
||||
// only sends after seeing that block's hash), so the two accesses to f
|
||||
// never race on the same region.
|
||||
// While the destination scan runs, emit a heartbeat on a fixed cadence
|
||||
// even if streamHashBlocks is parked inside one slow ReadAt (a big /
|
||||
// non-sparse / remote target). The push driver's idle watchdog counts any
|
||||
// frame from the sink as activity, so this stops a legitimately slow scan
|
||||
// from being declared a stalled sink during a check-only run, where there
|
||||
// is no DATA/ACK traffic to feed the watchdog. A sink that has actually
|
||||
// died stops heartbeating and its pipe closes, so the watchdog still fires.
|
||||
scanDone := make(chan struct{})
|
||||
go func() {
|
||||
t := time.NewTicker(10 * time.Second)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-scanDone:
|
||||
return
|
||||
case <-t.C:
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "scan"})
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
hashErrCh := make(chan error, 1)
|
||||
go func() {
|
||||
err := streamHashBlocks(f, size, blockSize, align, func(bh blockHash) error {
|
||||
err := streamHashBlocks(f, base, size, blockSize, align, func(bh blockHash) error {
|
||||
return out.WriteFrame(frameBlockHash, encodeBlockHashFrame(bh.index, bh.hash))
|
||||
}, scanProgressEmitter(out))
|
||||
close(scanDone)
|
||||
if err == nil {
|
||||
err = out.WriteFrame(frameHashDone, nil)
|
||||
} else {
|
||||
@ -533,7 +596,19 @@ func runSinkRole(path string, size, blockSize int64) error {
|
||||
hashErrCh <- err
|
||||
}()
|
||||
|
||||
loopErr := runDestLoop(destLoopParams{File: f, BlockSize: blockSize, Align: align, In: in, AckOut: out})
|
||||
var written int64
|
||||
var lastWritten time.Time
|
||||
loopErr := runDestLoop(destLoopParams{
|
||||
File: f, BlockSize: blockSize, Align: align, In: in, AckOut: out,
|
||||
OnWrite: func(uint64) {
|
||||
written++
|
||||
if time.Since(lastWritten) >= 500*time.Millisecond {
|
||||
lastWritten = time.Now()
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "write", Written: written})
|
||||
}
|
||||
},
|
||||
})
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "write", Written: written})
|
||||
if hashErr := <-hashErrCh; hashErr != nil {
|
||||
return hashErr
|
||||
}
|
||||
@ -547,7 +622,7 @@ func runSinkRole(path string, size, blockSize int64) error {
|
||||
// writing straight to its own stdout.
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
func runSourceStreamRole(path string, size, blockSize int64) error {
|
||||
func runSourceStreamRole(path string, base, size, blockSize int64) error {
|
||||
out := NewFrameWriter(os.Stdout)
|
||||
in := NewFrameReader(os.Stdin)
|
||||
|
||||
@ -599,21 +674,23 @@ func runSourceStreamRole(path string, size, blockSize int64) error {
|
||||
}
|
||||
}()
|
||||
|
||||
var copied, skipped int64
|
||||
var copied, skipped, srcRead int64
|
||||
var lastProgress time.Time
|
||||
maybeProgress := func() {
|
||||
if time.Since(lastProgress) < 500*time.Millisecond {
|
||||
return
|
||||
}
|
||||
lastProgress = time.Now()
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Copied: copied, Skipped: skipped, TotalBlocks: blockCount})
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "xfer", Copied: copied, Skipped: skipped, SrcRead: srcRead, TotalBlocks: blockCount})
|
||||
}
|
||||
|
||||
loopErr := runSourceLoop(sourceLoopParams{
|
||||
File: f, Size: size, BlockSize: blockSize, Align: align, Hashes: hashCh, Out: out,
|
||||
OnRead: func(uint64) { srcRead++; maybeProgress() },
|
||||
OnSkip: func(uint64) { skipped++; maybeProgress() },
|
||||
OnSend: func(uint64, [32]byte) { copied++; maybeProgress() },
|
||||
})
|
||||
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "xfer", Copied: copied, Skipped: skipped, SrcRead: srcRead, TotalBlocks: blockCount})
|
||||
if readErr := <-readErrCh; readErr != nil {
|
||||
return readErr
|
||||
}
|
||||
|
||||
9
build.sh
9
build.sh
@ -4,18 +4,17 @@ set -e
|
||||
cd "$(dirname "$0")"
|
||||
mkdir -p dist
|
||||
|
||||
build_time=$(date -u +%Y-%m-%dT%H:%M:%SZ)
|
||||
|
||||
build() {
|
||||
os=$1 arch=$2 out=$3
|
||||
echo "==> $os/$arch -> dist/$out"
|
||||
CGO_ENABLED=0 GOOS="$os" GOARCH="$arch" go build -trimpath -ldflags "-s -w" -o "dist/$out" .
|
||||
CGO_ENABLED=0 GOOS="$os" GOARCH="$arch" go build -trimpath \
|
||||
-ldflags "-s -w -X main.buildTime=$build_time" -o "dist/$out" .
|
||||
}
|
||||
|
||||
build linux amd64 clonetool-linux-amd64
|
||||
build linux arm64 clonetool-linux-arm64
|
||||
build windows amd64 clonetool-windows-amd64.exe
|
||||
build windows arm64 clonetool-windows-arm64.exe
|
||||
build darwin amd64 clonetool-darwin-amd64
|
||||
build darwin arm64 clonetool-darwin-arm64
|
||||
|
||||
echo "done:"
|
||||
ls -la dist/
|
||||
|
||||
29
ctrlmsg.go
29
ctrlmsg.go
@ -19,8 +19,12 @@ type CtrlMsg struct {
|
||||
// connect_push (-> source agent) / connect_pull (-> dest agent)
|
||||
PeerHost string `json:"peerHost,omitempty"`
|
||||
PeerUser string `json:"peerUser,omitempty"`
|
||||
PeerPort int `json:"peerPort,omitempty"`
|
||||
PeerPath string `json:"peerPath,omitempty"`
|
||||
// Base/PeerBase shift every block offset on the local/peer side. Zero on
|
||||
// both sides (the whole-file case) is what `sync` uses; the sink and
|
||||
// source-stream roles still accept a non-zero base for a windowed copy.
|
||||
Base int64 `json:"base,omitempty"`
|
||||
PeerBase int64 `json:"peerBase,omitempty"`
|
||||
// PeerLocal is set when both source and dest are local to the manager,
|
||||
// so the driver (itself already a local child of the manager) can spawn
|
||||
// the sink/source-stream helper as a plain local subprocess instead of
|
||||
@ -39,18 +43,24 @@ type CtrlMsg struct {
|
||||
Reason string `json:"reason,omitempty"`
|
||||
Message string `json:"message,omitempty"`
|
||||
|
||||
// progress. Scan and transfer now overlap, so a run emits both "scan"
|
||||
// messages (Hashed/TotalBlocks) and transfer messages (Copied/Skipped)
|
||||
// interleaved; the printer keeps the latest of each.
|
||||
Phase string `json:"phase,omitempty"` // "scan" while hashing the destination; empty during transfer
|
||||
// progress. Scan, source read and destination write all overlap and are
|
||||
// tracked independently, so a run emits several kinds of "progress"
|
||||
// message distinguished by Phase, interleaved; the printer keeps the
|
||||
// latest of each:
|
||||
// "scan" - Hashed/TotalBlocks: destination read throughput while it
|
||||
// fingerprints its current content.
|
||||
// "xfer" - Copied/Skipped/SrcRead/TotalBlocks: source read
|
||||
// throughput as it reads and compares each block (Skipped:
|
||||
// matched the destination; Copied: differed and was sent).
|
||||
// "write" - Written: destination write throughput as changed blocks
|
||||
// are actually applied.
|
||||
Phase string `json:"phase,omitempty"`
|
||||
Hashed int64 `json:"hashed,omitempty"`
|
||||
Copied int64 `json:"copied,omitempty"`
|
||||
Skipped int64 `json:"skipped,omitempty"`
|
||||
SrcRead int64 `json:"srcRead,omitempty"`
|
||||
Written int64 `json:"written,omitempty"`
|
||||
TotalBlocks int64 `json:"totalBlocks,omitempty"`
|
||||
BytesCopied int64 `json:"bytesCopied,omitempty"`
|
||||
|
||||
// log
|
||||
Level string `json:"level,omitempty"`
|
||||
}
|
||||
|
||||
const (
|
||||
@ -65,7 +75,6 @@ const (
|
||||
msgPullOK = "pull_ok"
|
||||
msgPullFailed = "pull_failed"
|
||||
msgProgress = "progress"
|
||||
msgLog = "log"
|
||||
msgError = "error"
|
||||
msgClose = "close"
|
||||
msgBye = "bye"
|
||||
|
||||
31
deploy.go
31
deploy.go
@ -15,14 +15,31 @@ import (
|
||||
// no install, no root.
|
||||
const deployedRemoteBin = ".clonetool/bin/clonetool"
|
||||
|
||||
// resolveRemoteBin returns the path to a runnable clonetool on spec's host.
|
||||
// It checks the configured --remote-bin first, then a copy deployed by an
|
||||
// earlier run, and finally streams this binary over and verifies it runs.
|
||||
// resolveRemoteBin returns the path to a clonetool on spec's host that is
|
||||
// both runnable and the same build as this binary (compared via the
|
||||
// "build=" tag in `version` output, see versionString/remoteBuildTag — an
|
||||
// architecture mismatch alone, expected when cross-compiled, doesn't count
|
||||
// as a mismatch). It checks the configured --remote-bin first, then a copy
|
||||
// deployed by an earlier run, and finally streams this binary over and
|
||||
// verifies it runs. With --deploy=false it can't correct a stale or missing
|
||||
// binary, so it uses --remote-bin anyway if merely out of date (only a
|
||||
// non-runnable one is still an error).
|
||||
func resolveRemoteBin(cfg *SyncConfig, spec Spec, tag string) (string, error) {
|
||||
if _, code, err := runRemote(cfg, spec, cfg.RemoteBin, "version"); err != nil {
|
||||
localTag := remoteBuildTag(versionString())
|
||||
|
||||
if out, code, err := runRemote(cfg, spec, cfg.RemoteBin, "version"); err != nil {
|
||||
return "", fmt.Errorf("%s: cannot reach %s over ssh: %w", tag, spec.Host, err)
|
||||
} else if code == 0 {
|
||||
return cfg.RemoteBin, nil
|
||||
if remoteBuildTag(out) == localTag {
|
||||
return cfg.RemoteBin, nil
|
||||
}
|
||||
if !cfg.Deploy {
|
||||
fmt.Fprintf(os.Stderr, "%s: %s on %s reports %q, a different build than this one (%s); continuing anyway (--deploy=false)\n",
|
||||
tag, cfg.RemoteBin, spec.Host, firstLine(out), versionString())
|
||||
return cfg.RemoteBin, nil
|
||||
}
|
||||
fmt.Fprintf(os.Stderr, "%s: %s on %s reports %q, a different build than this one (%s); looking for an up-to-date copy ...\n",
|
||||
tag, cfg.RemoteBin, spec.Host, firstLine(out), versionString())
|
||||
}
|
||||
|
||||
if !cfg.Deploy {
|
||||
@ -30,11 +47,11 @@ func resolveRemoteBin(cfg *SyncConfig, spec Spec, tag string) (string, error) {
|
||||
"install clonetool there or pass --remote-bin", tag, cfg.RemoteBin, spec.Host)
|
||||
}
|
||||
|
||||
if _, code, err := runRemote(cfg, spec, deployedRemoteBin, "version"); err == nil && code == 0 {
|
||||
if out, code, err := runRemote(cfg, spec, deployedRemoteBin, "version"); err == nil && code == 0 && remoteBuildTag(out) == localTag {
|
||||
return deployedRemoteBin, nil
|
||||
}
|
||||
|
||||
fmt.Fprintf(os.Stderr, "%s: no runnable clonetool on %s; copying this binary there ...\n", tag, spec.Host)
|
||||
fmt.Fprintf(os.Stderr, "%s: no up-to-date clonetool on %s; copying this binary there ...\n", tag, spec.Host)
|
||||
if err := deployBinary(cfg, spec); err != nil {
|
||||
return "", fmt.Errorf("%s: copy clonetool to %s: %w", tag, spec.Host, err)
|
||||
}
|
||||
|
||||
25
device.go
25
device.go
@ -13,20 +13,16 @@ type PathInfo struct {
|
||||
Size int64
|
||||
}
|
||||
|
||||
func statPath(path string) (PathInfo, error) {
|
||||
// A raw disk handle (Windows \\.\PhysicalDrive0, \\.\C:) is not something
|
||||
// os.Stat can describe, so ask the platform for its size directly.
|
||||
if isDevicePath(path) {
|
||||
size, err := blockDeviceSize(path)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return PathInfo{Exists: false}, nil
|
||||
}
|
||||
return PathInfo{}, fmt.Errorf("stat device %s: %w", path, err)
|
||||
}
|
||||
return PathInfo{Exists: true, IsDevice: true, Size: size}, nil
|
||||
}
|
||||
// alignmentFor returns the offset/length alignment a path's handle requires
|
||||
// for positioned reads and writes. Linux block devices accept ordinary
|
||||
// buffered pread/pwrite at any alignment, so there is nothing to round to.
|
||||
func alignmentFor(string) int64 { return 1 }
|
||||
|
||||
// canElevate reports whether a permission failure opening a device is worth
|
||||
// retrying under `sudo` (see --sudo).
|
||||
func canElevate() bool { return true }
|
||||
|
||||
func statPath(path string) (PathInfo, error) {
|
||||
fi, err := os.Stat(path)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
@ -63,9 +59,6 @@ func prepareDest(path string, targetSize int64) error {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if isDevicePath(path) && !info.Exists {
|
||||
return fmt.Errorf("destination device %s not found", path)
|
||||
}
|
||||
f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY, 0644)
|
||||
if err != nil {
|
||||
return fmt.Errorf("open/create %s: %w", path, err)
|
||||
|
||||
@ -1,9 +0,0 @@
|
||||
//go:build !linux && !windows
|
||||
|
||||
package main
|
||||
|
||||
import "fmt"
|
||||
|
||||
func blockDeviceSize(path string) (int64, error) {
|
||||
return 0, fmt.Errorf("block device size detection is only implemented on linux and windows (got path %s)", path)
|
||||
}
|
||||
@ -1,18 +0,0 @@
|
||||
//go:build !windows
|
||||
|
||||
package main
|
||||
|
||||
// isDevicePath reports whether a path must be treated as a raw device from
|
||||
// its string form alone. On unix, os.Stat's mode bits already identify
|
||||
// device nodes accurately (symlinks, non-standard locations and all), so
|
||||
// this is always false and statPath relies on those instead.
|
||||
func isDevicePath(string) bool { return false }
|
||||
|
||||
// alignmentFor returns the offset/length alignment a path's handle requires
|
||||
// for positioned reads and writes. Unix block devices accept ordinary
|
||||
// buffered pread/pwrite at any alignment, so there is nothing to round to.
|
||||
func alignmentFor(string) int64 { return 1 }
|
||||
|
||||
// canElevate reports whether a permission failure opening a device is worth
|
||||
// retrying under `sudo` (see --sudo). Always true on unix.
|
||||
func canElevate() bool { return true }
|
||||
@ -1,99 +0,0 @@
|
||||
//go:build windows
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"syscall"
|
||||
"unsafe"
|
||||
)
|
||||
|
||||
// isDevicePath recognises the Win32 device namespaces clonetool can sync
|
||||
// against by string form: \\.\PhysicalDrive0 (a whole disk) and \\.\C: (a
|
||||
// volume). Both the \\.\ and \\?\ prefixes, and their forward-slash
|
||||
// spellings, are accepted. os.Stat can't describe these paths, so statPath
|
||||
// keys off this instead of mode bits on Windows.
|
||||
func isDevicePath(path string) bool {
|
||||
p := strings.ReplaceAll(path, "/", `\`)
|
||||
return strings.HasPrefix(p, `\\.\`) || strings.HasPrefix(p, `\\?\`)
|
||||
}
|
||||
|
||||
// canElevate is false on Windows: there is no `sudo` equivalent to re-exec
|
||||
// under, so a permission failure opening a raw disk is reported with a hint
|
||||
// to run from an elevated console instead of being retried.
|
||||
func canElevate() bool { return false }
|
||||
|
||||
const (
|
||||
ioctlDiskGetLengthInfo = 0x0007405C // IOCTL_DISK_GET_LENGTH_INFO
|
||||
ioctlDiskGetDriveGeometry = 0x00070000 // IOCTL_DISK_GET_DRIVE_GEOMETRY
|
||||
)
|
||||
|
||||
// diskGeometry mirrors DISK_GEOMETRY (24 bytes; Cylinders is a LARGE_INTEGER).
|
||||
type diskGeometry struct {
|
||||
Cylinders int64
|
||||
MediaType uint32
|
||||
TracksPerCylinder uint32
|
||||
SectorsPerTrack uint32
|
||||
BytesPerSector uint32
|
||||
}
|
||||
|
||||
// openDeviceHandle opens path for a metadata ioctl only: zero access rights
|
||||
// (which need no privilege and don't require the volume to be unlocked) and
|
||||
// shared read/write so it doesn't disturb a mounted filesystem.
|
||||
func openDeviceHandle(path string) (syscall.Handle, error) {
|
||||
p, err := syscall.UTF16PtrFromString(path)
|
||||
if err != nil {
|
||||
return syscall.InvalidHandle, err
|
||||
}
|
||||
return syscall.CreateFile(p, 0,
|
||||
syscall.FILE_SHARE_READ|syscall.FILE_SHARE_WRITE, nil,
|
||||
syscall.OPEN_EXISTING, 0, 0)
|
||||
}
|
||||
|
||||
// blockDeviceSize returns the byte length of a physical drive or volume via
|
||||
// DeviceIoControl(IOCTL_DISK_GET_LENGTH_INFO).
|
||||
func blockDeviceSize(path string) (int64, error) {
|
||||
h, err := openDeviceHandle(path)
|
||||
if err != nil {
|
||||
return 0, &os.PathError{Op: "open", Path: path, Err: err}
|
||||
}
|
||||
defer syscall.CloseHandle(h)
|
||||
|
||||
var length int64 // GET_LENGTH_INFORMATION is a single LARGE_INTEGER
|
||||
var ret uint32
|
||||
err = syscall.DeviceIoControl(h, ioctlDiskGetLengthInfo,
|
||||
nil, 0,
|
||||
(*byte)(unsafe.Pointer(&length)), uint32(unsafe.Sizeof(length)),
|
||||
&ret, nil)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("IOCTL_DISK_GET_LENGTH_INFO %s: %w", path, err)
|
||||
}
|
||||
return length, nil
|
||||
}
|
||||
|
||||
// alignmentFor returns the sector size a raw disk handle's positioned reads
|
||||
// and writes must be aligned to; 1 for an ordinary file path. It falls back
|
||||
// to 512 if the geometry query fails.
|
||||
func alignmentFor(path string) int64 {
|
||||
if !isDevicePath(path) {
|
||||
return 1
|
||||
}
|
||||
h, err := openDeviceHandle(path)
|
||||
if err != nil {
|
||||
return 512
|
||||
}
|
||||
defer syscall.CloseHandle(h)
|
||||
|
||||
var g diskGeometry
|
||||
var ret uint32
|
||||
err = syscall.DeviceIoControl(h, ioctlDiskGetDriveGeometry,
|
||||
nil, 0,
|
||||
(*byte)(unsafe.Pointer(&g)), uint32(unsafe.Sizeof(g)),
|
||||
&ret, nil)
|
||||
if err != nil || g.BytesPerSector == 0 {
|
||||
return 512
|
||||
}
|
||||
return int64(g.BytesPerSector)
|
||||
}
|
||||
3
main.go
3
main.go
@ -4,7 +4,6 @@ import (
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
"runtime"
|
||||
)
|
||||
|
||||
func main() {
|
||||
@ -20,7 +19,7 @@ func main() {
|
||||
case "agent":
|
||||
err = cmdAgent(os.Args[2:])
|
||||
case "version", "--version":
|
||||
fmt.Printf("clonetool %s/%s\n", runtime.GOOS, runtime.GOARCH)
|
||||
fmt.Println(versionString())
|
||||
return
|
||||
case "-h", "--help", "help":
|
||||
usage()
|
||||
|
||||
93
manager.go
93
manager.go
@ -153,11 +153,6 @@ func bringUpController(spec Spec, tag string, cfg *SyncConfig, remoteBin, probeP
|
||||
if err == nil {
|
||||
return c, info, sudo, nil
|
||||
}
|
||||
if errors.Is(err, errNeedPriv) && !canElevate() {
|
||||
c.Close()
|
||||
return nil, PathInfo{}, sudo, fmt.Errorf(
|
||||
"%w; on Windows, run clonetool from an elevated (Administrator) console to open a raw disk", err)
|
||||
}
|
||||
if errors.Is(err, errNeedPriv) && cfg.Sudo == "auto" && !sudo {
|
||||
fmt.Fprintf(os.Stderr, "%s: permission denied on %s; retrying via sudo ...\n", tag, probePath)
|
||||
c.Close()
|
||||
@ -234,28 +229,62 @@ func confirmShrink(dstSpec Spec, oldSize, newSize int64) bool {
|
||||
return line == "y" || line == "yes"
|
||||
}
|
||||
|
||||
// progressPrinter renders a single rewriting status line on stderr. Scan
|
||||
// and transfer now run concurrently, so it keeps the latest of each kind of
|
||||
// update (blocks hashed so far, blocks copied/skipped) and shows them on one
|
||||
// line, with a rolling copy rate; the "scan H/M" segment disappears once the
|
||||
// rateTracker smooths a monotonically increasing block counter into a
|
||||
// blocks/sec rate, sampling no more than a few times a second so a burst of
|
||||
// same-tick progress messages doesn't produce a noisy instantaneous rate.
|
||||
type rateTracker struct {
|
||||
lastT time.Time
|
||||
lastVal int64
|
||||
rate float64
|
||||
}
|
||||
|
||||
func (r *rateTracker) update(val int64) {
|
||||
now := time.Now()
|
||||
if r.lastT.IsZero() {
|
||||
r.lastT, r.lastVal = now, val
|
||||
return
|
||||
}
|
||||
dt := now.Sub(r.lastT).Seconds()
|
||||
if dt < 0.4 {
|
||||
return
|
||||
}
|
||||
inst := float64(val-r.lastVal) / dt
|
||||
if r.rate == 0 {
|
||||
r.rate = inst
|
||||
} else {
|
||||
r.rate = 0.6*r.rate + 0.4*inst
|
||||
}
|
||||
r.lastT, r.lastVal = now, val
|
||||
}
|
||||
|
||||
func (r *rateTracker) mib(blockSize int64) float64 {
|
||||
return r.rate * float64(blockSize) / (1024 * 1024)
|
||||
}
|
||||
|
||||
// progressPrinter renders a single rewriting status line on stderr. Scan,
|
||||
// source read and destination write all run concurrently, each tracked by
|
||||
// its own phase-tagged progress message (see CtrlMsg.Phase), so it keeps the
|
||||
// latest of each and shows them on one line with their own rolling
|
||||
// throughput; the "scan H/M" segment and its read rate disappear once the
|
||||
// destination fingerprint is complete.
|
||||
type progressPrinter struct {
|
||||
blockSize int64
|
||||
start time.Time
|
||||
lastT time.Time
|
||||
lastCopied int64
|
||||
rate float64 // copied blocks/sec, smoothed
|
||||
active bool
|
||||
blockSize int64
|
||||
active bool
|
||||
|
||||
hashed int64
|
||||
copied int64
|
||||
skipped int64
|
||||
srcRead int64
|
||||
written int64
|
||||
total int64
|
||||
|
||||
dstReadRate rateTracker // destination scan/hash throughput
|
||||
srcReadRate rateTracker // source read/compare throughput
|
||||
dstWriteRate rateTracker // destination write throughput
|
||||
}
|
||||
|
||||
func newProgressPrinter(blockSize int64) *progressPrinter {
|
||||
now := time.Now()
|
||||
return &progressPrinter{blockSize: blockSize, start: now, lastT: now}
|
||||
return &progressPrinter{blockSize: blockSize}
|
||||
}
|
||||
|
||||
func (p *progressPrinter) print(m CtrlMsg) {
|
||||
@ -263,22 +292,22 @@ func (p *progressPrinter) print(m CtrlMsg) {
|
||||
if m.TotalBlocks > 0 {
|
||||
p.total = m.TotalBlocks
|
||||
}
|
||||
if m.Phase == "scan" {
|
||||
switch m.Phase {
|
||||
case "scan":
|
||||
if m.Hashed > p.hashed {
|
||||
p.hashed = m.Hashed
|
||||
}
|
||||
} else {
|
||||
p.copied, p.skipped = m.Copied, m.Skipped
|
||||
now := time.Now()
|
||||
if dt := now.Sub(p.lastT).Seconds(); dt >= 0.4 {
|
||||
inst := float64(m.Copied-p.lastCopied) / dt
|
||||
if p.rate == 0 {
|
||||
p.rate = inst
|
||||
} else {
|
||||
p.rate = 0.6*p.rate + 0.4*inst
|
||||
}
|
||||
p.lastT, p.lastCopied = now, m.Copied
|
||||
if p.total > 0 && p.hashed >= p.total {
|
||||
p.dstReadRate.rate = 0 // scan finished; stop showing a frozen rate
|
||||
} else {
|
||||
p.dstReadRate.update(p.hashed)
|
||||
}
|
||||
case "xfer":
|
||||
p.copied, p.skipped, p.srcRead = m.Copied, m.Skipped, m.SrcRead
|
||||
p.srcReadRate.update(p.srcRead)
|
||||
case "write":
|
||||
p.written = m.Written
|
||||
p.dstWriteRate.update(p.written)
|
||||
}
|
||||
|
||||
done := p.copied + p.skipped
|
||||
@ -286,15 +315,15 @@ func (p *progressPrinter) print(m CtrlMsg) {
|
||||
if total <= 0 {
|
||||
total = done
|
||||
}
|
||||
mib := p.rate * float64(p.blockSize) / (1024 * 1024)
|
||||
|
||||
scan := ""
|
||||
if p.total > 0 && p.hashed < p.total {
|
||||
scan = fmt.Sprintf(" scan %s/%s", formatCount(p.hashed), formatCount(p.total))
|
||||
}
|
||||
fmt.Fprintf(os.Stderr, "\r syncing %s %s/%s copied=%s skipped=%s%s %6.1f MiB/s ",
|
||||
fmt.Fprintf(os.Stderr, "\r syncing %s %s/%s copied=%s skipped=%s%s rd(src) %5.1f rd(dst) %5.1f wr(dst) %5.1f MiB/s ",
|
||||
progressBar(done, total), formatCount(done), formatCount(total),
|
||||
formatCount(p.copied), formatCount(p.skipped), scan, mib)
|
||||
formatCount(p.copied), formatCount(p.skipped), scan,
|
||||
p.srcReadRate.mib(p.blockSize), p.dstReadRate.mib(p.blockSize), p.dstWriteRate.mib(p.blockSize))
|
||||
}
|
||||
|
||||
// finish ends the current status line with a newline so following output
|
||||
|
||||
43
syncside.go
43
syncside.go
@ -81,12 +81,14 @@ type blockHash struct {
|
||||
hash [32]byte
|
||||
}
|
||||
|
||||
// streamHashBlocks reads f in blockSize-byte blocks up to size, hashing each
|
||||
// one and handing it to onHash the moment it is computed (this is what lets
|
||||
// the peer start comparing before the whole side has been scanned). It keeps
|
||||
// no hash state of its own. onProgress, if non-nil, is called with
|
||||
// (blocksHashed, totalBlocks) before the first block and after each one.
|
||||
func streamHashBlocks(f *os.File, size, blockSize, align int64, onHash func(bh blockHash) error, onProgress func(done, total int64)) error {
|
||||
// streamHashBlocks reads f in blockSize-byte blocks over the window
|
||||
// [base, base+size), hashing each one and handing it to onHash the moment it
|
||||
// is computed (this is what lets the peer start comparing before the whole
|
||||
// side has been scanned). It keeps no hash state of its own. onProgress, if
|
||||
// non-nil, is called with (blocksHashed, totalBlocks) before the first block
|
||||
// and after each one. base is 0 for a whole-file sync; a non-zero base
|
||||
// fingerprints just a byte window within a larger handle.
|
||||
func streamHashBlocks(f *os.File, base, size, blockSize, align int64, onHash func(bh blockHash) error, onProgress func(done, total int64)) error {
|
||||
if err := checkBlockAlign(blockSize, align); err != nil {
|
||||
return err
|
||||
}
|
||||
@ -96,7 +98,7 @@ func streamHashBlocks(f *os.File, size, blockSize, align int64, onHash func(bh b
|
||||
onProgress(0, blockCount)
|
||||
}
|
||||
for i := int64(0); i < blockCount; i++ {
|
||||
n, err := readBlockAt(f, buf, i*blockSize, size, align)
|
||||
n, err := readBlockAt(f, buf, base+i*blockSize, base+size, align)
|
||||
if err != nil {
|
||||
return fmt.Errorf("hash %s at block %d: %w", f.Name(), i, err)
|
||||
}
|
||||
@ -130,12 +132,18 @@ func scanProgressEmitter(out *FrameWriter) func(done, total int64) {
|
||||
// never read twice.
|
||||
type sourceLoopParams struct {
|
||||
File *os.File // already opened for reading by the caller
|
||||
Size int64
|
||||
Base int64 // byte offset the window starts at (0 for a whole-file sync)
|
||||
Size int64 // window length in bytes
|
||||
BlockSize int64
|
||||
Align int64
|
||||
Hashes <-chan blockHash
|
||||
Out *FrameWriter
|
||||
OnSkip func(index uint64)
|
||||
// OnRead is called immediately after a block has been read and hashed
|
||||
// for comparison — whether it turns out to match or differ — so
|
||||
// callers can track true source read throughput independent of the
|
||||
// send/ack timing tracked by OnSkip/OnSend below.
|
||||
OnRead func(index uint64)
|
||||
OnSkip func(index uint64)
|
||||
// OnSend is called for a differing block immediately before its DATA
|
||||
// frame goes on the wire — never after. The push driver relies on this
|
||||
// ordering to record the block as awaiting confirmation before the peer
|
||||
@ -151,12 +159,15 @@ func runSourceLoop(p sourceLoopParams) error {
|
||||
f := p.File
|
||||
buf := make([]byte, p.BlockSize)
|
||||
for bh := range p.Hashes {
|
||||
offset := int64(bh.index) * p.BlockSize
|
||||
n, err := readBlockAt(f, buf, offset, p.Size, p.Align)
|
||||
offset := p.Base + int64(bh.index)*p.BlockSize
|
||||
n, err := readBlockAt(f, buf, offset, p.Base+p.Size, p.Align)
|
||||
if err != nil {
|
||||
return fmt.Errorf("read %s at block %d: %w", f.Name(), bh.index, err)
|
||||
}
|
||||
hash := sha256.Sum256(buf[:n])
|
||||
if p.OnRead != nil {
|
||||
p.OnRead(bh.index)
|
||||
}
|
||||
if hash == bh.hash {
|
||||
if p.OnSkip != nil {
|
||||
p.OnSkip(bh.index)
|
||||
@ -181,6 +192,7 @@ func runSourceLoop(p sourceLoopParams) error {
|
||||
// same process, before OnWritten is called).
|
||||
type destLoopParams struct {
|
||||
File *os.File // already opened for read/write by the caller
|
||||
Base int64 // byte offset the window starts at (0 for a whole-file sync)
|
||||
BlockSize int64
|
||||
Align int64 // sector alignment for a raw disk handle, else 0/1
|
||||
In *FrameReader
|
||||
@ -189,6 +201,10 @@ type destLoopParams struct {
|
||||
// pull mode, where the remote source-stream has no other channel back
|
||||
// to the manager for progress updates). Sink-role callers leave it nil.
|
||||
OnCtrlMsg func(CtrlMsg)
|
||||
// OnWrite is called right after a block is durably written — the true
|
||||
// destination write-throughput measurement point, as opposed to
|
||||
// whatever "copied" count a driver derives from the send/ack protocol.
|
||||
OnWrite func(index uint64)
|
||||
}
|
||||
|
||||
func runDestLoop(p destLoopParams) error {
|
||||
@ -212,12 +228,15 @@ func runDestLoop(p destLoopParams) error {
|
||||
}
|
||||
return fmt.Errorf("block %d: hash mismatch after transfer", index)
|
||||
}
|
||||
if err := writeBlockAt(f, block, int64(index)*p.BlockSize, p.Align); err != nil {
|
||||
if err := writeBlockAt(f, block, p.Base+int64(index)*p.BlockSize, p.Align); err != nil {
|
||||
if p.AckOut != nil {
|
||||
_ = p.AckOut.WriteFrame(frameErr, encodeErrFrame(index, err.Error()))
|
||||
}
|
||||
return fmt.Errorf("write block %d: %w", index, err)
|
||||
}
|
||||
if p.OnWrite != nil {
|
||||
p.OnWrite(index)
|
||||
}
|
||||
if p.AckOut != nil {
|
||||
if err := p.AckOut.WriteFrame(frameAck, encodeIndexFrame(index)); err != nil {
|
||||
return err
|
||||
|
||||
16
util.go
16
util.go
@ -2,9 +2,25 @@ package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os/exec"
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// runCmd runs name with args, returning trimmed combined output.
|
||||
func runCmd(name string, args ...string) (string, error) {
|
||||
out, err := exec.Command(name, args...).CombinedOutput()
|
||||
return strings.TrimSpace(string(out)), err
|
||||
}
|
||||
|
||||
// runCmdStdin is runCmd with a fixed stdin string (for tools that prompt).
|
||||
func runCmdStdin(stdin, name string, args ...string) (string, error) {
|
||||
c := exec.Command(name, args...)
|
||||
c.Stdin = strings.NewReader(stdin)
|
||||
out, err := c.CombinedOutput()
|
||||
return strings.TrimSpace(string(out)), err
|
||||
}
|
||||
|
||||
// limitedBuffer keeps only the last maxLen bytes written to it — used to
|
||||
// capture a bounded tail of a subprocess's stderr for error messages
|
||||
// without risking unbounded memory growth from a noisy remote command.
|
||||
|
||||
35
version.go
Normal file
35
version.go
Normal file
@ -0,0 +1,35 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"runtime"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// buildTime is stamped at build time via -ldflags "-X main.buildTime=..."
|
||||
// (build.sh does this). A plain "go build" with no ldflags leaves it at
|
||||
// "dev" — still a meaningful, comparable value: two "dev" builds are only
|
||||
// ever the same binary bytes copied around (self-deploy), never two
|
||||
// independently built ones landing on the same value by chance.
|
||||
var buildTime = "dev"
|
||||
|
||||
// versionString is what `clonetool version` prints, and what a remote
|
||||
// host's own `version` output is compared against (see remoteBuildTag) to
|
||||
// tell whether it's running the same build as this binary.
|
||||
func versionString() string {
|
||||
return fmt.Sprintf("clonetool %s/%s build=%s", runtime.GOOS, runtime.GOARCH, buildTime)
|
||||
}
|
||||
|
||||
// remoteBuildTag extracts the "build=..." tag from a clonetool `version`
|
||||
// line, ignoring its GOOS/GOARCH — those are expected to legitimately
|
||||
// differ across a cross-compiled deploy — and any surrounding ssh/shell
|
||||
// noise on other lines. Returns "" if the line isn't a clonetool version
|
||||
// line at all (e.g. "command not found").
|
||||
func remoteBuildTag(out string) string {
|
||||
line := firstLine(out)
|
||||
i := strings.Index(line, "build=")
|
||||
if i < 0 {
|
||||
return ""
|
||||
}
|
||||
return strings.TrimSpace(line[i+len("build="):])
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user