From 95783225d5135a5971c3c6474abf1fcb1f5b9567 Mon Sep 17 00:00:00 2001 From: Alexander Gabriel Date: Sun, 6 Sep 2026 00:39:34 +0200 Subject: [PATCH] show progress --- README.md | 3 +- agent.go | 35 +++++++++++++-------- ctrlmsg.go | 9 +++--- manager.go | 87 ++++++++++++++++++++++++++++++++++++++++++++++++++--- syncside.go | 25 +++++++++++++-- 5 files changed, 135 insertions(+), 24 deletions(-) diff --git a/README.md b/README.md index db34766..fd7da02 100644 --- a/README.md +++ b/README.md @@ -119,7 +119,8 @@ the self-deploy check). and rejected if they later change, same as normal SSH behavior. - Because every run re-hashes the whole destination, a re-sync costs a full read of both sides even when little changed — the win is in the - bytes transferred, not the bytes read. + bytes transferred, not the bytes read. That read shows on the status + line as a `scanning destination` phase before `syncing` begins. - `agent` is an internal subcommand spawned automatically by `sync`; it's not meant to be run by hand, though it will work standalone for debugging. diff --git a/agent.go b/agent.go index c89f28d..668bb2a 100644 --- a/agent.go +++ b/agent.go @@ -75,16 +75,27 @@ func runControlAgent() error { } } -// readHashFrame reads one frameHashTable off fr and expands it. -func readHashFrame(fr *FrameReader) ([][32]byte, error) { - typ, payload, err := fr.ReadFrame() - if err != nil { - return nil, fmt.Errorf("read hash table: %w", err) +// readHashFrame reads frames off fr until the hash table arrives, relaying +// any "scan" progress frames the sink sends while it hashes the destination +// onward to the manager via out. +func readHashFrame(fr *FrameReader, out *FrameWriter) ([][32]byte, error) { + for { + typ, payload, err := fr.ReadFrame() + if err != nil { + return nil, fmt.Errorf("read hash table: %w", err) + } + switch typ { + case frameHashTable: + return unflattenHashes(payload) + case frameCtrlJSON: + var m CtrlMsg + if json.Unmarshal(payload, &m) == nil && m.Type == msgProgress { + _ = out.WriteJSON(m) + } + default: + return nil, fmt.Errorf("expected hash table frame, got type %d", typ) + } } - if typ != frameHashTable { - return nil, fmt.Errorf("expected hash table frame, got type %d", typ) - } - return unflattenHashes(payload) } func handleStat(out *FrameWriter, m CtrlMsg) { @@ -174,7 +185,7 @@ func runPushDriver(req CtrlMsg, out *FrameWriter) { // Handshake succeeded: we're committed to push for this run. The sink // now sends the current per-block hashes of the destination it just // read; the source loop compares against those to decide what to send. - hashes, err := readHashFrame(fr) + hashes, err := readHashFrame(fr, out) if err != nil { _ = cmd.Process.Kill() _ = cmd.Wait() @@ -422,7 +433,7 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) { } defer dstFile.Close() - hashes, err := hashFileBlocks(dstFile, req.Size, req.BlockSize) + hashes, err := hashFileBlocks(dstFile, req.Size, req.BlockSize, scanProgressEmitter(out)) if err != nil { _ = cmd.Process.Kill() _ = cmd.Wait() @@ -480,7 +491,7 @@ func runSinkRole(path string, size, blockSize int64) error { return err } - hashes, err := hashFileBlocks(f, size, blockSize) + hashes, err := hashFileBlocks(f, size, blockSize, scanProgressEmitter(out)) if err != nil { fmt.Fprintf(os.Stderr, "sink: hash %s: %v\n", path, err) return err diff --git a/ctrlmsg.go b/ctrlmsg.go index 50b39c2..0553082 100644 --- a/ctrlmsg.go +++ b/ctrlmsg.go @@ -40,10 +40,11 @@ type CtrlMsg struct { Message string `json:"message,omitempty"` // progress - Copied int64 `json:"copied,omitempty"` - Skipped int64 `json:"skipped,omitempty"` - TotalBlocks int64 `json:"totalBlocks,omitempty"` - BytesCopied int64 `json:"bytesCopied,omitempty"` + Phase string `json:"phase,omitempty"` // "scan" while hashing the destination; empty during transfer + Copied int64 `json:"copied,omitempty"` + Skipped int64 `json:"skipped,omitempty"` + TotalBlocks int64 `json:"totalBlocks,omitempty"` + BytesCopied int64 `json:"bytesCopied,omitempty"` // log Level string `json:"level,omitempty"` diff --git a/manager.go b/manager.go index ebd9bb0..28683e7 100644 --- a/manager.go +++ b/manager.go @@ -6,6 +6,7 @@ import ( "fmt" "os" "strings" + "time" ) const defaultBlockSize = 4 * 1024 * 1024 @@ -85,7 +86,8 @@ func runSync(cfg SyncConfig) error { blockCount := (targetSize + cfg.BlockSize - 1) / cfg.BlockSize - cb := transferCallbacks{onProgress: func(m CtrlMsg) { printProgress(m) }} + pp := newProgressPrinter(cfg.BlockSize) + cb := transferCallbacks{onProgress: pp.print} bothLocal := srcSpec.IsLocal() && dstSpec.IsLocal() srcHost, srcUser := resolveConnectHost(srcSpec, &cfg) @@ -107,6 +109,7 @@ func runSync(cfg SyncConfig) error { return err } if !ok { + pp.finish() fmt.Fprintf(os.Stderr, "push not possible (%s); trying pull %s <- %s ...\n", reason, dstSpec, srcSpec) pullReq := CtrlMsg{ Path: dstSpec.Path, Size: targetSize, BlockSize: cfg.BlockSize, @@ -126,7 +129,7 @@ func runSync(cfg SyncConfig) error { } } - fmt.Fprintln(os.Stderr) + pp.finish() label := cfg.Job if label == "" { label = fmt.Sprintf("%s -> %s", srcSpec, dstSpec) @@ -226,10 +229,84 @@ func confirmShrink(dstSpec Spec, oldSize, newSize int64) bool { return line == "y" || line == "yes" } -func printProgress(m CtrlMsg) { +// progressPrinter renders a single rewriting status line on stderr for both +// phases of a run: "scan" while a side hashes the destination, then the +// block transfer itself (with a rolling copy rate). +type progressPrinter struct { + blockSize int64 + start time.Time + lastT time.Time + lastCopied int64 + rate float64 // copied blocks/sec, smoothed + active bool +} + +func newProgressPrinter(blockSize int64) *progressPrinter { + now := time.Now() + return &progressPrinter{blockSize: blockSize, start: now, lastT: now} +} + +func (p *progressPrinter) print(m CtrlMsg) { + p.active = true + if m.Phase == "scan" { + fmt.Fprintf(os.Stderr, "\r scanning destination %s %s/%s blocks ", + progressBar(m.Copied, m.TotalBlocks), + formatCount(m.Copied), formatCount(m.TotalBlocks)) + return + } + + done := m.Copied + m.Skipped total := m.TotalBlocks if total <= 0 { - total = m.Copied + m.Skipped + total = done + } + + 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 + } + mib := p.rate * float64(p.blockSize) / (1024 * 1024) + + fmt.Fprintf(os.Stderr, "\r syncing %s %s/%s copied=%s skipped=%s %6.1f MiB/s ", + progressBar(done, total), formatCount(done), formatCount(total), + formatCount(m.Copied), formatCount(m.Skipped), mib) +} + +// finish ends the current status line with a newline so following output +// (and the shell prompt) starts clean. +func (p *progressPrinter) finish() { + if p.active { + fmt.Fprintln(os.Stderr) + p.active = false + } +} + +func progressBar(done, total int64) string { + const w = 22 + pct := 0 + if total > 0 { + pct = int(done * 100 / total) + } + if pct > 100 { + pct = 100 + } + n := pct * w / 100 + return fmt.Sprintf("[%s%s] %3d%%", strings.Repeat("=", n), strings.Repeat(" ", w-n), pct) +} + +func formatCount(n int64) string { + switch { + case n >= 1_000_000: + return fmt.Sprintf("%.1fM", float64(n)/1e6) + case n >= 10_000: + return fmt.Sprintf("%.0fk", float64(n)/1e3) + default: + return fmt.Sprintf("%d", n) } - fmt.Fprintf(os.Stderr, "\rblocks: %d/%d copied=%d skipped=%d ", m.Copied+m.Skipped, total, m.Copied, m.Skipped) } diff --git a/syncside.go b/syncside.go index d38efc5..a898a3b 100644 --- a/syncside.go +++ b/syncside.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "os" + "time" ) // readBlockAt reads exactly min(len(buf), totalSize-offset) bytes at offset, @@ -30,21 +31,41 @@ func readBlockAt(f *os.File, buf []byte, offset, totalSize int64) (int, error) { // hashFileBlocks reads f in blockSize-byte blocks up to size and returns the // SHA-256 of each. This is how a destination fingerprints its *current* // content at the start of every sync — clonetool keeps no hash state of its -// own between runs. -func hashFileBlocks(f *os.File, size, blockSize int64) ([][32]byte, error) { +// own between runs. onProgress, if non-nil, is called with (blocksHashed, +// totalBlocks) before the first block and after each one. +func hashFileBlocks(f *os.File, size, blockSize int64, onProgress func(done, total int64)) ([][32]byte, error) { blockCount := (size + blockSize - 1) / blockSize out := make([][32]byte, blockCount) buf := make([]byte, blockSize) + if onProgress != nil { + onProgress(0, blockCount) + } for i := int64(0); i < blockCount; i++ { n, err := readBlockAt(f, buf, i*blockSize, size) if err != nil { return nil, fmt.Errorf("hash %s at block %d: %w", f.Name(), i, err) } out[i] = sha256.Sum256(buf[:n]) + if onProgress != nil { + onProgress(i+1, blockCount) + } } return out, nil } +// scanProgressEmitter returns an onProgress callback for hashFileBlocks that +// forwards "scan" phase progress onto out, throttled to ~3/second. +func scanProgressEmitter(out *FrameWriter) func(done, total int64) { + var last time.Time + return func(done, total int64) { + if done != 0 && done != total && time.Since(last) < 350*time.Millisecond { + return + } + last = time.Now() + _ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "scan", Copied: done, TotalBlocks: total}) + } +} + // sourceLoopParams drives the single read -> hash -> compare -> maybe-send // pass over a source path. It is used both by the standalone // "source-stream" role (writing to its own stdout) and by a source control