diff --git a/cmd/bdrive/cmds.go b/cmd/bdrive/cmds.go index 35ae9c6..a3caffa 100644 --- a/cmd/bdrive/cmds.go +++ b/cmd/bdrive/cmds.go @@ -27,6 +27,7 @@ func syncCmd() *cobra.Command { return err } defer closeSession(sess) + sess.OnProgress = progressReporter() res, err := sess.Cycle(cmd.Context()) if err != nil { return err diff --git a/cmd/bdrive/progress.go b/cmd/bdrive/progress.go new file mode 100644 index 0000000..ee0e0a5 --- /dev/null +++ b/cmd/bdrive/progress.go @@ -0,0 +1,58 @@ +package main + +import ( + "fmt" + "os" + "strings" + "sync" + "time" + + "github.com/mattn/go-isatty" + + "github.com/runbear-io/beardrive/internal/syncer" +) + +// progressReporter returns a syncer.OnProgress callback that shows upload +// progress: an in-place bar on a TTY, periodic percentage lines otherwise. +// It's safe to call concurrently from upload workers, and renders are +// throttled so many small blobs don't thrash the terminal. It stays silent +// when there's nothing to upload. +func progressReporter() func(syncer.Progress) { + tty := isatty.IsTerminal(os.Stderr.Fd()) + var mu sync.Mutex + var lastDraw time.Time + lastPct := -1 + return func(p syncer.Progress) { + if p.Total == 0 { + return + } + mu.Lock() + defer mu.Unlock() + final := p.Done >= p.Total + pct := p.Done * 100 / p.Total + if tty { + if !final && time.Since(lastDraw) < 100*time.Millisecond { + return + } + lastDraw = time.Now() + const w = 24 + filled := pct * w / 100 + bar := strings.Repeat("█", filled) + strings.Repeat("░", w-filled) + fmt.Fprintf(os.Stderr, "\r uploading [%s] %3d%% %d/%d files %s / %s", + bar, pct, p.Done, p.Total, humanBytes(p.Bytes), humanBytes(p.ToBytes)) + if final { + fmt.Fprintln(os.Stderr) + } + return + } + // Non-TTY: announce the total once, then a line every ~10% and at the end. + if lastPct < 0 { + fmt.Fprintf(os.Stderr, " uploading %d files (%s)\n", p.Total, humanBytes(p.ToBytes)) + } + if final || pct/10 > lastPct/10 { + fmt.Fprintf(os.Stderr, " %d%% %d/%d files %s / %s\n", + pct, p.Done, p.Total, humanBytes(p.Bytes), humanBytes(p.ToBytes)) + } + lastPct = pct + } +} diff --git a/cmd/bdrive/sync_run.go b/cmd/bdrive/sync_run.go index 6e9d522..661307e 100644 --- a/cmd/bdrive/sync_run.go +++ b/cmd/bdrive/sync_run.go @@ -32,6 +32,7 @@ func startSync(ctx context.Context, folder string, proj config.Project, foregrou if err != nil { return err } + sess.OnProgress = progressReporter() // the initial import is the slow one res, err := sess.Cycle(ctx) closeSession(sess) if err != nil { diff --git a/go.mod b/go.mod index 2d0fb22..71c44e0 100644 --- a/go.mod +++ b/go.mod @@ -14,6 +14,7 @@ require ( github.com/spf13/cobra v1.10.2 github.com/yuin/goldmark v1.8.2 golang.org/x/crypto v0.51.0 + golang.org/x/sync v0.21.0 google.golang.org/api v0.284.0 modernc.org/sqlite v1.53.0 ) @@ -79,7 +80,6 @@ require ( go.opentelemetry.io/otel/trace v1.43.0 // indirect golang.org/x/net v0.55.0 // indirect golang.org/x/oauth2 v0.36.0 // indirect - golang.org/x/sync v0.21.0 // indirect golang.org/x/sys v0.45.0 // indirect golang.org/x/term v0.43.0 // indirect golang.org/x/text v0.37.0 // indirect diff --git a/internal/syncer/syncer.go b/internal/syncer/syncer.go index c61ea80..ce9cf9c 100644 --- a/internal/syncer/syncer.go +++ b/internal/syncer/syncer.go @@ -21,14 +21,30 @@ import ( "path" "path/filepath" "strings" + "sync/atomic" "time" + "golang.org/x/sync/errgroup" + "github.com/runbear-io/beardrive/internal/config" "github.com/runbear-io/beardrive/internal/journal" "github.com/runbear-io/beardrive/internal/remote" "github.com/runbear-io/beardrive/internal/store" ) +// pushConcurrency bounds how many blobs upload at once. The initial import of +// many files is latency-bound on serial round-trips, so uploading in parallel +// is the main speedup. +const pushConcurrency = 16 + +// Progress reports upload progress during a cycle's push phase, so the CLI can +// draw a bar. Total/TotalBytes are set once when the push starts; Done/Bytes +// climb as blobs finish. Nil OnProgress means no reporting (the daemon). +type Progress struct { + Done, Total int + Bytes, ToBytes int64 +} + // Session ties a working folder to its volume store and (optionally) remote. type Session struct { Folder string @@ -40,6 +56,10 @@ type Session struct { // Device.Author remains the fallback identity. Account config.Settings Backend remote.Backend // nil = work offline + // OnProgress, when set, is called during push with upload progress. It may + // be invoked concurrently from upload workers, so it must be safe to call + // from multiple goroutines. + OnProgress func(Progress) } func (s *Session) mountID() string { @@ -516,18 +536,44 @@ func (s *Session) push(ctx context.Context, myOps []journal.Op, st *store.SyncSt if st.PushedOps > int64(len(myOps)) { st.PushedOps = int64(len(myOps)) } - uploaded := map[string]bool{} + // Collect the unique, not-yet-pushed blobs to upload (deduped by content + // hash). The backend's Put is idempotent and already skips content that's + // present remotely (the hub reports it during signing), so we don't pay a + // separate existence round-trip per blob. + seen := map[string]bool{} + type blobJob struct { + blob string + size int64 + } + var jobs []blobJob + var totalBytes int64 for _, op := range myOps[st.PushedOps:] { - if op.Kind != journal.KindPut || op.Blob == "" || uploaded[op.Blob] { + if op.Kind != journal.KindPut || op.Blob == "" || seen[op.Blob] { continue } - key := "blobs/" + op.Blob - ok, err := s.Backend.Exists(ctx, key) - if err != nil { - return err + seen[op.Blob] = true + jobs = append(jobs, blobJob{op.Blob, op.Size}) + totalBytes += op.Size + } + + var done, bytesDone int64 + report := func() { + if s.OnProgress != nil { + s.OnProgress(Progress{ + Done: int(atomic.LoadInt64(&done)), Total: len(jobs), + Bytes: atomic.LoadInt64(&bytesDone), ToBytes: totalBytes, + }) } - if !ok { - f, err := s.Store.OpenBlob(op.Blob) + } + report() // announce the total up front (0 / N) + + // Upload blobs in parallel — the initial import is bound on serial + // round-trips, not bandwidth, so concurrency is the win. + g, gctx := errgroup.WithContext(ctx) + g.SetLimit(pushConcurrency) + for _, j := range jobs { + g.Go(func() error { + f, err := s.Store.OpenBlob(j.blob) if err != nil { return err } @@ -536,13 +582,19 @@ func (s *Session) push(ctx context.Context, myOps []journal.Op, st *store.SyncSt f.Close() return err } - err = s.Backend.Put(ctx, key, f, fi.Size()) + err = s.Backend.Put(gctx, "blobs/"+j.blob, f, fi.Size()) f.Close() if err != nil { return err } - } - uploaded[op.Blob] = true + atomic.AddInt64(&done, 1) + atomic.AddInt64(&bytesDone, fi.Size()) + report() + return nil + }) + } + if err := g.Wait(); err != nil { + return err } jp := s.Store.JournalPath(s.Device.ID) diff --git a/internal/syncer/syncer_test.go b/internal/syncer/syncer_test.go index 39e7459..6464717 100644 --- a/internal/syncer/syncer_test.go +++ b/internal/syncer/syncer_test.go @@ -2,9 +2,11 @@ package syncer import ( "context" + "fmt" "os" "path/filepath" "strings" + "sync" "testing" "time" @@ -13,6 +15,40 @@ import ( "github.com/runbear-io/beardrive/internal/store" ) +// TestPushProgress verifies the push phase reports upload progress: the total +// is the number of unique blobs, Done climbs to that total, and byte totals +// are populated. (Done isn't strictly ordered across the parallel workers, so +// we only assert it reaches the total.) +func TestPushProgress(t *testing.T) { + a := newDevice(t, "deva", sharedRemote(t)) + const n = 25 + for i := 0; i < n; i++ { + write(t, a.Folder, fmt.Sprintf("f%02d.txt", i), fmt.Sprintf("unique content for file %d — pad pad pad", i)) + } + var mu sync.Mutex + var total, maxDone int + var toBytes int64 + a.OnProgress = func(p Progress) { + mu.Lock() + defer mu.Unlock() + total = p.Total + toBytes = p.ToBytes + if p.Done > maxDone { + maxDone = p.Done + } + } + cycle(t, a) + if total != n { + t.Fatalf("progress Total = %d, want %d", total, n) + } + if maxDone != n { + t.Fatalf("progress reached Done = %d, want %d", maxDone, n) + } + if toBytes == 0 { + t.Fatal("progress ToBytes should be > 0") + } +} + // newDevice simulates one device: its own folder, volume store, and identity, // all syncing through a shared file:// remote. func newDevice(t *testing.T, name string, backend remote.Backend) *Session {