From 2ff148115f6e444feba9e84d9629a3fff7d7ab9c Mon Sep 17 00:00:00 2001 From: Rose Hogenson Date: Tue, 15 Apr 2025 17:01:53 -0700 Subject: Perf improvements --- ccp.go | 49 ++++++++++++------------------------------------- internal/cp/cp.go | 45 ++++++++++++++++++++++++++------------------- 2 files changed, 38 insertions(+), 56 deletions(-) diff --git a/ccp.go b/ccp.go index 18e2248..69691d6 100644 --- a/ccp.go +++ b/ccp.go @@ -41,10 +41,10 @@ import ( "github.com/charmbracelet/bubbles/progress" tea "github.com/charmbracelet/bubbletea" "github.com/charmbracelet/lipgloss" - "github.com/rhogenson/container/deque" "github.com/rhogenson/ccp/internal/cp" "github.com/rhogenson/ccp/internal/wfs/osfs" "github.com/rhogenson/ccp/internal/wfs/sftpfs" + "github.com/rhogenson/container/deque" ) var f = flag.Bool("f", false, "if an existing destination file cannot be opened, remove it and try again") @@ -64,12 +64,8 @@ type model struct { // Every 500 milliseconds, the current progress is appended to // measurements for calculating ETA. measurements deque.Deque[measurement] - // copyingFiles holds the files currently being copied. Keys are source - // paths and values are the corresponding destination paths. - copyingFiles map[string]string - // copyingFile is an arbitrary entry from copyingFiles that we're - // currently showing to the user. Tracked in the state so that it - // doesn't change every time we update the view. + // copyingFile is a file that is or was being copied that we're + // currently showing to the user. copyingFile string // eta is the estimated time to completion, or -1 if we don't have // enough samples. @@ -95,10 +91,7 @@ type ( } // fileDoneMsg is sent whenever we finish copying a file. err indicates // any error that was encountered during the copy. - fileDoneMsg struct { - name string - err error - } + errorMsg struct{ error } // doneMsg is sent when all files are finished copying and it's time // to exit. doneMsg struct{} @@ -117,22 +110,9 @@ func (m *model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { case maxMsg: m.max = int64(msg) case fileStartMsg: - m.copyingFiles[msg.from] = msg.to - if m.copyingFile == "" { - m.copyingFile = msg.from - } - case fileDoneMsg: - delete(m.copyingFiles, msg.name) - if m.copyingFile == msg.name { - m.copyingFile = "" - for name := range m.copyingFiles { - m.copyingFile = name - break - } - } - if msg.err != nil { - m.errs = append(m.errs, msg.err.Error()) - } + m.copyingFile = msg.from + " -> " + msg.to + case errorMsg: + m.errs = append(m.errs, msg.Error()) case doneMsg: m.done = true var cmd tea.Cmd @@ -186,16 +166,12 @@ func (m *model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { var warningStyle = lipgloss.NewStyle().Foreground(lipgloss.Color("3")).Render func (m *model) View() string { - copying := "" - if m.copyingFile != "" { - copying = m.copyingFile + " -> " + m.copyingFiles[m.copyingFile] - } etaStr := "calculating..." if m.eta >= 0 { etaStr = m.eta.Round(time.Second).String() } return "\n" + - " " + copying + "\n" + + " " + m.copyingFile + "\n" + " " + m.progress.View() + "\n" + " " + "ETA: " + etaStr + "\n\n" + warningStyle(strings.Join(m.errs, "\n")) + "\n" @@ -219,8 +195,8 @@ func (pu *progressUpdater) FileStart(from, to string) { pu.p.Send(fileStartMsg{from, to}) } -func (pu *progressUpdater) FileDone(name string, err error) { - pu.p.Send(fileDoneMsg{name, err}) +func (pu *progressUpdater) Error(err error) { + pu.p.Send(errorMsg{err}) } // splitHostPath splits an scp target into host and path, e.g. user@host:/path/ @@ -270,9 +246,8 @@ func run() error { } dst := toFSPath(dstTarget, sftpHosts) m := &model{ - progress: progress.New(progress.WithDefaultGradient(), progress.WithoutPercentage()), - copyingFiles: make(map[string]string), - eta: -1, + progress: progress.New(progress.WithDefaultGradient(), progress.WithoutPercentage()), + eta: -1, } p := tea.NewProgram(m, tea.WithInput(nil), tea.WithOutput(os.Stderr)) go func() { diff --git a/internal/cp/cp.go b/internal/cp/cp.go index da164f4..7828787 100644 --- a/internal/cp/cp.go +++ b/internal/cp/cp.go @@ -10,6 +10,7 @@ import ( "path" "slices" "strings" + "time" "github.com/rhogenson/ccp/internal/wfs" "github.com/rhogenson/ccp/internal/wfs/sftpfs" @@ -24,11 +25,11 @@ type Progress interface { // Progress reports that n additional bytes have been copied. Progress(n int64) // FileStart reports that src is currently being copied to dst. Only - // called for regular files, not directories or symlinks. + // called for regular files, not directories or symlinks. cp also + // rate-limits calls to FileStart, so not all files will be reported. FileStart(src, dst string) - // FileDone is called when a regular file has finished copying - // successfully, or when there was an error copying a file. - FileDone(src string, err error) + // Error reports an error encountered. + Error(error) } // An FSPath is an abstraction over a file path that can point to multiple @@ -126,8 +127,9 @@ func (p FSPath) exists() bool { } type copier struct { - p Progress - force bool + p Progress + force bool + fileStartRateLimit *time.Ticker } func (c *copier) openWithRetry(path FSPath, fn func() error) error { @@ -141,7 +143,11 @@ func (c *copier) openWithRetry(path FSPath, fn func() error) error { } func (c *copier) copyRegularFile(src, dst FSPath) error { - c.p.FileStart(src.String(), dst.String()) + select { + case <-c.fileStartRateLimit.C: + c.p.FileStart(src.String(), dst.String()) + default: + } in, err := src.open() if err != nil { @@ -179,7 +185,6 @@ func (c *copier) copyRegularFile(src, dst FSPath) error { return err } c.p.Progress(1) - c.p.FileDone(src.String(), nil) return nil } @@ -211,13 +216,15 @@ func Copy(progress Progress, srcs []FSPath, dstRoot FSPath, force bool) { dstIsDir = err == nil && stat.IsDir() } - const maxConcurrency = 10 + const maxConcurrency = 500 // sem acts as a semaphore to limit the number of concurrent file copies sem := make(chan struct{}, maxConcurrency) c := &copier{ - p: progress, - force: force, + p: progress, + force: force, + fileStartRateLimit: time.NewTicker(500 * time.Millisecond), } + defer c.fileStartRateLimit.Stop() type roDir struct { path FSPath mode fs.FileMode @@ -233,14 +240,14 @@ func Copy(progress Progress, srcs []FSPath, dstRoot FSPath, force bool) { } srcRoot.Path = path.Clean(srcRoot.Path) if srcRoot == dstRoot { - progress.FileDone(srcRoot.String(), fmt.Errorf("%q and %q are the same file", srcRoot, dstRoot)) + progress.Error(fmt.Errorf("%q and %q are the same file", srcRoot, dstRoot)) continue } srcRoot.walkDir(func(srcPath string, d fs.DirEntry, err error) error { src := FSPath{srcRoot.FS, srcPath} dst := FSPath{dstRoot.FS, path.Join(dstRoot.Path, strings.TrimPrefix(srcPath, srcRoot.Path))} if err != nil { - progress.FileDone(src.String(), err) + progress.Error(err) return nil } switch d.Type() { @@ -249,14 +256,14 @@ func Copy(progress Progress, srcs []FSPath, dstRoot FSPath, force bool) { go func() { defer func() { <-sem }() if err := c.copyRegularFile(src, dst); err != nil { - progress.FileDone(src.String(), err) + progress.Error(err) } }() case fs.ModeDir: stat, err := d.Info() if err != nil { - progress.FileDone(src.String(), err) + progress.Error(err) return fs.SkipDir } hasWritePerm := stat.Mode()&0300 == 0300 @@ -276,7 +283,7 @@ func Copy(progress Progress, srcs []FSPath, dstRoot FSPath, force bool) { return dst.mkdir() } }); err != nil { - progress.FileDone(src.String(), err) + progress.Error(err) return fs.SkipDir } if hasWritePerm { @@ -286,10 +293,10 @@ func Copy(progress Progress, srcs []FSPath, dstRoot FSPath, force bool) { } case fs.ModeSymlink: if err := c.copySymlink(src, dst); err != nil { - progress.FileDone(src.String(), err) + progress.Error(err) } default: - progress.FileDone(src.String(), fmt.Errorf("%s: unknown file type %s", src, d.Type())) + progress.Error(fmt.Errorf("%s: unknown file type %s", src, d.Type())) } return nil }) @@ -302,7 +309,7 @@ func Copy(progress Progress, srcs []FSPath, dstRoot FSPath, force bool) { // parent directory itself. for _, d := range slices.Backward(roDirs) { if err := d.path.chmod(d.mode); err != nil { - progress.FileDone(d.path.String(), err) + progress.Error(err) continue } progress.Progress(1) -- cgit v1.3.1