diff --git a/cmd/opensave-app/frontend/src/lib/restoreprogress.js b/cmd/opensave-app/frontend/src/lib/restoreprogress.js new file mode 100644 index 00000000..c690e1f6 --- /dev/null +++ b/cmd/opensave-app/frontend/src/lib/restoreprogress.js @@ -0,0 +1,29 @@ +// How far a restore has come, in words: the phase, percent and, once it can +// be told, roughly how long is left (daemon/restoreprogress.go). +// +// The words are the catalog's (restoreProgress.*): the caller passes its +// translate function, `$t` in a component, so the line follows the chosen +// language as it changes. + +const phases = { + rebuilding: 'restoreProgress.rebuilding', + checking: 'restoreProgress.checking', + writing: 'restoreProgress.writing' +}; + +export function restoreProgressText(p, t) { + if (!p || !p.phase) return ''; + const params = { + phase: t(phases[p.phase] ?? 'restoreProgress.restoring'), + percent: p.total > 0 ? Math.floor((p.done / p.total) * 100) : 0, + done: p.done.toLocaleString(), + total: p.total.toLocaleString() + }; + if (p.secondsLeft > 0) return t('restoreProgress.lineWithTimeLeft', { ...params, left: timeLeft(p.secondsLeft, t) }); + return t('restoreProgress.line', params); +} + +function timeLeft(s, t) { + if (s < 60) return t('restoreProgress.underAMinute'); + return t('restoreProgress.aboutMinutes', { n: Math.round(s / 60) }); +} diff --git a/cmd/opensave-app/frontend/src/lib/restoreprogress.test.js b/cmd/opensave-app/frontend/src/lib/restoreprogress.test.js new file mode 100644 index 00000000..570bb7a2 --- /dev/null +++ b/cmd/opensave-app/frontend/src/lib/restoreprogress.test.js @@ -0,0 +1,39 @@ +import { describe, it, expect } from 'vitest'; +import { get } from 'svelte/store'; +import { locale, t } from './i18n.js'; +import { restoreProgressText } from './restoreprogress.js'; + +// The line in a given language, as a component would say it with `$t`. +const say = (l, p) => { + locale.set(l); + return restoreProgressText(p, get(t)); +}; + +describe('restoreProgressText', () => { + it('says nothing when no restore is under way', () => { + expect(say('en', null)).toBe(''); + }); + it('says the phase, percent and time left', () => { + const line = say('en', { phase: 'writing', done: 60000, total: 240000, secondsLeft: 150 }); + expect(line).toContain('Writing the files'); + expect(line).toContain('25%'); + expect(line).toContain('about 3 minutes left'); + }); + it('says one minute, and less than one, in words', () => { + expect(say('en', { phase: 'writing', done: 1, total: 2, secondsLeft: 70 })).toContain('about a minute left'); + expect(say('en', { phase: 'writing', done: 1, total: 2, secondsLeft: 20 })).toContain('under a minute left'); + }); + it('leaves out the time while it cannot be told', () => { + expect(say('en', { phase: 'checking', done: 10, total: 100 })).not.toContain('left'); + }); + it('calls a phase it does not know restoring', () => { + expect(say('en', { phase: 'later', done: 0, total: 0 })).toContain('Restoring'); + }); + it('says the line in the chosen language', () => { + const line = say('zh-CN', { phase: 'writing', done: 60000, total: 240000, secondsLeft: 150 }); + expect(line).toContain('正在写入文件'); + expect(line).toContain('25%'); + expect(line).toContain('约 3 分钟'); + expect(line).not.toContain('Writing'); + }); +}); diff --git a/cmd/opensave-app/frontend/src/locales/en.json b/cmd/opensave-app/frontend/src/locales/en.json index 2157edb9..e699ecf9 100644 --- a/cmd/opensave-app/frontend/src/locales/en.json +++ b/cmd/opensave-app/frontend/src/locales/en.json @@ -211,5 +211,16 @@ "home.findGame.pickAnother": "← Pick another game", "home.findGame.already": "Already tracked here as \"{tracked}\". Tracking it again would sync the same files twice; link that entry to {name} instead, so it gets the game database’s name, App ID and cover art.", "home.findGame.link": "Link \"{tracked}\" to {name}", - "home.findGame.linked": "\"{tracked}\" is now {name}" + "home.findGame.linked": "\"{tracked}\" is now {name}", + "restoreProgress.rebuilding": "Rebuilding the snapshot", + "restoreProgress.checking": "Checking the snapshot", + "restoreProgress.writing": "Writing the files", + "restoreProgress.restoring": "Restoring", + "restoreProgress.line": "{phase} — {percent}% ({done} of {total})", + "restoreProgress.lineWithTimeLeft": "{phase} — {percent}% ({done} of {total}), {left} left", + "restoreProgress.underAMinute": "under a minute", + "restoreProgress.aboutMinutes": { + "one": "about a minute", + "other": "about {n} minutes" + } } diff --git a/cmd/opensave-app/frontend/src/locales/zh-CN.json b/cmd/opensave-app/frontend/src/locales/zh-CN.json index b7399448..f35d728e 100644 --- a/cmd/opensave-app/frontend/src/locales/zh-CN.json +++ b/cmd/opensave-app/frontend/src/locales/zh-CN.json @@ -190,5 +190,13 @@ "home.findGame.pickAnother": "← 选择其他游戏", "home.findGame.already": "此处已作为“{tracked}”跟踪。再次跟踪会将相同的文件同步两次;请改为将该条目关联到 {name},使其获得游戏数据库中的名称、App ID 和封面。", "home.findGame.link": "将“{tracked}”关联到 {name}", - "home.findGame.linked": "“{tracked}”现在是 {name}" + "home.findGame.linked": "“{tracked}”现在是 {name}", + "restoreProgress.rebuilding": "正在重建快照", + "restoreProgress.checking": "正在检查快照", + "restoreProgress.writing": "正在写入文件", + "restoreProgress.restoring": "正在恢复", + "restoreProgress.line": "{phase} — {percent}%({done}/{total})", + "restoreProgress.lineWithTimeLeft": "{phase} — {percent}%({done}/{total}),还剩{left}", + "restoreProgress.underAMinute": "不到一分钟", + "restoreProgress.aboutMinutes": "约 {n} 分钟" } diff --git a/cmd/opensave-app/frontend/src/views/game/GameHeader.svelte b/cmd/opensave-app/frontend/src/views/game/GameHeader.svelte index 3db66f67..558381a8 100644 --- a/cmd/opensave-app/frontend/src/views/game/GameHeader.svelte +++ b/cmd/opensave-app/frontend/src/views/game/GameHeader.svelte @@ -10,6 +10,9 @@ import { timeAgo } from '../../lib/timeago.js'; import { playLength } from '../../lib/format.js'; import { canLaunch } from '../../lib/gameactions.js'; + import Spinner from '../../components/ui/Spinner.svelte'; + import { restoreProgressText } from '../../lib/restoreprogress.js'; + import { t } from '../../lib/i18n.js'; import ArrowLeft from 'lucide-svelte/icons/arrow-left'; import Play from 'lucide-svelte/icons/play'; import RefreshCw from 'lucide-svelte/icons/refresh-cw'; @@ -133,6 +136,9 @@ {#if activity?.state === 'running'} · syncing {activity.percentage ?? 0}% {/if} + {#if game.restoring} + · {restoreProgressText(game.restoring, $t)} + {/if} {#if game.playingSince} · in session since {new Date(game.playingSince).toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' })} {:else if game.lastPlayedAt} diff --git a/internal/api/server.go b/internal/api/server.go index 0663ae75..645cc12c 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -578,6 +578,8 @@ func (s *Server) gamePayload(g store.Game) map[string]any { "playingSince": playingSince, "playtimeMs": play.PlaytimeMs, "playSessions": play.Sessions, + // A restore of it under way, and how far it has come. + "restoring": s.Daemon.RestoringNow(g.ID), // Whether the game is installed on this device: "found", "not-found", // or "" when there is no telling (daemon.InstallState). A game that is // not here is one nobody plays here, which is why it shows no play. diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index dde8a586..39e7ad63 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -72,6 +72,9 @@ type Daemon struct { relocateFor map[string]string relocateDismissed map[string]string + // Restores under way; see restoreprogress.go. + restoring restoreState + // Play sessions; see sessions.go. sessions sessionState @@ -171,6 +174,8 @@ func New(opts Options) (*Daemon, error) { // A restore or a branch switch rewrites a save folder; nothing syncing it // may read it half-way (syncengine/settle.go). snaps.WriteGate = d.P2P.Sync.Writing + // How far a restore has come, for the window (restoreprogress.go). + snaps.OnRestoreProgress = d.noteRestoreProgress // A paired peer untracking/re-tracking a game mirrors here. d.P2P.OnUntrackRequest = d.untrackFromPeer diff --git a/internal/daemon/restoreprogress.go b/internal/daemon/restoreprogress.go new file mode 100644 index 00000000..3801df5d --- /dev/null +++ b/internal/daemon/restoreprogress.go @@ -0,0 +1,79 @@ +package daemon + +import ( + "sync" + "time" +) + +// How far a restore has come, for the window: a large save is put back in +// minutes, and a spinner alone left it unclear whether anything was +// happening at all. + +// RestoreProgress is a restore under way. +type RestoreProgress struct { + Phase string `json:"phase"` // snapshot.PhaseRebuilding, PhaseChecking, PhaseWriting + Done int `json:"done"` + Total int `json:"total"` + // SecondsLeft estimates how long the phase still takes; 0 until there + // is enough to go on. + SecondsLeft int `json:"secondsLeft,omitempty"` +} + +type restoreState struct { + mu sync.Mutex + games map[string]*restoreEntry +} + +type restoreEntry struct { + RestoreProgress + phaseAt time.Time // when the phase began + sentAt time.Time // when the window was last told +} + +// noteRestoreProgress records a restore's progress (snapshot.Manager's +// OnRestoreProgress), telling the window about once a second. +func (d *Daemon) noteRestoreProgress(gameID, phase string, done, total int) { + now := time.Now() + d.restoring.mu.Lock() + if d.restoring.games == nil { + d.restoring.games = map[string]*restoreEntry{} + } + tell := false + if phase == "" { + delete(d.restoring.games, gameID) + tell = true + } else { + e := d.restoring.games[gameID] + if e == nil || e.Phase != phase { + e = &restoreEntry{phaseAt: now} + d.restoring.games[gameID] = e + tell = true + } + e.Phase, e.Done, e.Total = phase, done, total + e.SecondsLeft = 0 + if elapsed := now.Sub(e.phaseAt); done > 0 && total > done && elapsed > 2*time.Second { + e.SecondsLeft = int(elapsed.Seconds() / float64(done) * float64(total-done)) + } + if now.Sub(e.sentAt) >= time.Second { + tell = true + } + if tell { + e.sentAt = now + } + } + d.restoring.mu.Unlock() + if tell && d.OnGameChanged != nil { + d.OnGameChanged(gameID) + } +} + +// RestoringNow is the progress of a restore of gameID under way, or nil. +func (d *Daemon) RestoringNow(gameID string) *RestoreProgress { + d.restoring.mu.Lock() + defer d.restoring.mu.Unlock() + if e := d.restoring.games[gameID]; e != nil { + p := e.RestoreProgress + return &p + } + return nil +} diff --git a/internal/iopar/drive.go b/internal/iopar/drive.go new file mode 100644 index 00000000..14d9a0f5 --- /dev/null +++ b/internal/iopar/drive.go @@ -0,0 +1,52 @@ +// Package iopar runs many small file operations — a save of a quarter of a +// million files restored, checked, written — as many at a time as the drive +// they are on does best: one at a time on a spinning disk, whose head would +// otherwise be sent back and forth across it; many at once on an SSD, which +// works through several requests in parallel and is otherwise left waiting +// on each file's own small delays (its creation, a virus scanner's look at +// it). How many is a starting point by kind of drive, adjusted as the work +// goes by what it actually achieves (Run). +package iopar + +// Kind is the kind of drive a path is on. +type Kind int + +const ( + // Unknown: it could not be told. Treated as a SATA SSD — parallel, but + // modestly so. + Unknown Kind = iota + HDD + SATA + NVMe +) + +func (k Kind) String() string { + switch k { + case HDD: + return "hard disk" + case SATA: + return "SATA SSD" + case NVMe: + return "NVMe SSD" + } + return "unknown drive" +} + +// start is how many operations run at once to begin with. +func (k Kind) start() int { + switch k { + case HDD: + return 1 + case NVMe: + return 32 + } + return 8 +} + +// adapts reports whether the number is adjusted as the work goes. A spinning +// disk is not: it does best one at a time, and finding that out would cost +// the seeking it exists to avoid. +func (k Kind) adapts() bool { return k != HDD } + +// DriveKind says what kind of drive path is on. +func DriveKind(path string) Kind { return driveKind(path) } diff --git a/internal/iopar/drive_other.go b/internal/iopar/drive_other.go new file mode 100644 index 00000000..77cf455b --- /dev/null +++ b/internal/iopar/drive_other.go @@ -0,0 +1,56 @@ +//go:build !windows + +package iopar + +import ( + "os" + "path/filepath" + "strings" + "syscall" +) + +// On Linux the kernel says whether a block device rotates; an NVMe drive is +// named so. Elsewhere it is not told. +func driveKind(path string) Kind { + var st syscall.Stat_t + if err := syscall.Stat(path, &st); err != nil { + return Unknown + } + major, minor := (st.Dev>>8)&0xfff, (st.Dev&0xff)|((st.Dev>>12)&0xfff00) + dev, err := os.Readlink("/sys/dev/block/" + itoa(uint64(major)) + ":" + itoa(uint64(minor))) + if err != nil { + return Unknown + } + // A partition: its disk is the folder above it. + disk := filepath.Base(dev) + sys := filepath.Join("/sys/class/block", disk) + if _, err := os.Stat(filepath.Join(sys, "partition")); err == nil { + disk = filepath.Base(filepath.Dir(dev)) + sys = filepath.Join("/sys/class/block", disk) + } + if strings.HasPrefix(disk, "nvme") { + return NVMe + } + b, err := os.ReadFile(filepath.Join(sys, "queue", "rotational")) + if err != nil { + return Unknown + } + if strings.TrimSpace(string(b)) == "1" { + return HDD + } + return SATA +} + +func itoa(n uint64) string { + if n == 0 { + return "0" + } + var b [20]byte + i := len(b) + for n > 0 { + i-- + b[i] = byte('0' + n%10) + n /= 10 + } + return string(b[i:]) +} diff --git a/internal/iopar/drive_windows.go b/internal/iopar/drive_windows.go new file mode 100644 index 00000000..8bb019cf --- /dev/null +++ b/internal/iopar/drive_windows.go @@ -0,0 +1,94 @@ +//go:build windows + +package iopar + +import ( + "path/filepath" + "strings" + "unsafe" + + "golang.org/x/sys/windows" +) + +// Asked of the volume a path is on, as Windows' own optimiser asks it: does +// it incur a seek penalty (a spinning disk), and over which bus is it +// attached (NVMe or otherwise). + +const ( + ioctlStorageQueryProperty = 0x002D1400 + + storageDeviceProperty = 0 + storageDeviceSeekPenaltyProperty = 7 + propertyStandardQuery = 0 + + busTypeNvme = 17 +) + +type storagePropertyQuery struct { + PropertyID uint32 + QueryType uint32 + Additional [1]byte +} + +type deviceSeekPenaltyDescriptor struct { + Version uint32 + Size uint32 + IncursSeekPenalty byte +} + +type storageDeviceDescriptor struct { + Version uint32 + Size uint32 + DeviceType byte + DeviceTypeModifier byte + RemovableMedia byte + CommandQueueing byte + VendorIDOffset uint32 + ProductIDOffset uint32 + ProductRevisionOffset uint32 + SerialNumberOffset uint32 + BusType uint32 + RawPropertiesLength uint32 +} + +func driveKind(path string) Kind { + abs, err := filepath.Abs(path) + if err != nil { + return Unknown + } + vol := filepath.VolumeName(abs) + if len(vol) != 2 || vol[1] != ':' { + return Unknown // a network share or the like: no drive to ask + } + name, err := windows.UTF16PtrFromString(`\\.\` + strings.ToUpper(vol)) + if err != nil { + return Unknown + } + // No access rights are needed to ask these properties. + h, err := windows.CreateFile(name, 0, windows.FILE_SHARE_READ|windows.FILE_SHARE_WRITE, nil, windows.OPEN_EXISTING, 0, 0) + if err != nil { + return Unknown + } + defer windows.CloseHandle(h) + + var seek deviceSeekPenaltyDescriptor + if query(h, storageDeviceSeekPenaltyProperty, unsafe.Pointer(&seek), uint32(unsafe.Sizeof(seek))) && seek.IncursSeekPenalty != 0 { + return HDD + } + var dev storageDeviceDescriptor + if query(h, storageDeviceProperty, unsafe.Pointer(&dev), uint32(unsafe.Sizeof(dev))) { + if dev.BusType == busTypeNvme { + return NVMe + } + return SATA + } + return Unknown +} + +func query(h windows.Handle, property uint32, out unsafe.Pointer, size uint32) bool { + q := storagePropertyQuery{PropertyID: property, QueryType: propertyStandardQuery} + var returned uint32 + err := windows.DeviceIoControl(h, ioctlStorageQueryProperty, (*byte)(unsafe.Pointer(&q)), uint32(unsafe.Sizeof(q)), + (*byte)(out), size, &returned, nil) + return err == nil && returned >= 8 +} diff --git a/internal/iopar/run.go b/internal/iopar/run.go new file mode 100644 index 00000000..3cd15028 --- /dev/null +++ b/internal/iopar/run.go @@ -0,0 +1,234 @@ +package iopar + +import ( + "context" + "sync" + "sync/atomic" + "time" +) + +// The adjusting: every window the number of operations finished is +// measured, and the number running at once moves the way that made it +// faster — on while it keeps helping, back when it hurt. A drive that slows +// as it goes, its cache full or a scanner busy, gets fewer; one with more to +// give gets more. Once it settles it still looks the other way now and then, +// so a change later in a long run is followed too. + +const ( + // maxAtOnce bounds how many run at once. + maxAtOnce = 64 + // Changes smaller than this are noise, not a reason to move. + better = 1.05 + worse = 0.95 + // probeAfter windows without change before looking the other way. + probeAfter = 4 +) + +// window is how long each measurement runs; a variable, for tests. +var window = 750 * time.Millisecond + +// Stats is how a run went, for the log. +type Stats struct { + Kind Kind + Start int // operations at once to begin with + Settled int // at the end + Peak float64 + Duration time.Duration +} + +// Run calls work for each of n items, as many at a time as suits the drive +// path is on. progress, if not nil, is told how many are done now and then +// and at the end. The first error stops the run and is returned; work +// already started finishes first. +func Run(ctx context.Context, path string, n int, work func(i int) error, progress func(done, total int)) (Stats, error) { + return RunKind(ctx, DriveKind(path), n, work, progress) +} + +// RunFixed runs work with exactly at operations at once, not adjusting: for +// measuring what a drive does at each number. +func RunFixed(ctx context.Context, at, n int, work func(i int) error, progress func(done, total int)) (Stats, error) { + return run(ctx, Unknown, at, false, n, work, progress) +} + +// RunKind is Run for a drive whose kind is known. +func RunKind(ctx context.Context, kind Kind, n int, work func(i int) error, progress func(done, total int)) (Stats, error) { + return run(ctx, kind, kind.start(), kind.adapts(), n, work, progress) +} + +func run(ctx context.Context, kind Kind, start int, adapts bool, n int, work func(i int) error, progress func(done, total int)) (Stats, error) { + started := time.Now() + st := Stats{Kind: kind, Start: start} + if n <= 0 { + return st, nil + } + p := &pool{limit: min(start, n)} + p.cond = sync.NewCond(&p.mu) + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + var next, done atomic.Int64 + var firstErr error + var errOnce sync.Once + fail := func(err error) { + errOnce.Do(func() { + firstErr = err + cancel() + p.cond.Broadcast() // wake the ones waiting, to stop + }) + } + + workers := min(maxAtOnce, n) + if !adapts { + workers = p.limit + } + var wg sync.WaitGroup + for w := 0; w < workers; w++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + if !p.acquire(ctx) { + return + } + i := int(next.Add(1) - 1) + if i >= n { + p.release() + return + } + if err := work(i); err != nil { + p.release() + fail(err) + return + } + done.Add(1) + p.release() + } + }() + } + + // Adjusting, and telling progress, until the workers are done. + finished := make(chan struct{}) + go func() { + wg.Wait() + close(finished) + }() + tick := time.NewTicker(window) + defer tick.Stop() + c := climber{dir: 1} + last, lastAt := int64(0), time.Now() +loop: + for { + select { + case <-finished: + break loop + case now := <-tick.C: + d := done.Load() + rate := float64(d-last) / now.Sub(lastAt).Seconds() + last, lastAt = d, now + if rate > st.Peak { + st.Peak = rate + } + if progress != nil { + progress(int(d), n) + } + if adapts { + p.setLimit(c.next(p.getLimit(), rate)) + } + } + } + if progress != nil { + progress(int(done.Load()), n) + } + st.Settled = p.getLimit() + st.Duration = time.Since(started) + if firstErr != nil { + return st, firstErr + } + return st, ctx.Err() +} + +// climber moves the number running at once towards where the most finishes. +type climber struct { + dir int // +1 more, -1 fewer + prevRate float64 + still int // windows without a move +} + +func (c *climber) next(limit int, rate float64) int { + step := max(1, limit/4) + switch { + case c.prevRate == 0: + // The first window: try more. + case rate >= c.prevRate*better: + // That helped: go on the same way. + c.still = 0 + case rate <= c.prevRate*worse: + // That hurt: back the other way. + c.dir = -c.dir + c.still = 0 + default: + // No difference worth the name: stay, and look the other way after + // a while. + c.still++ + if c.still < probeAfter { + c.prevRate = rate + return limit + } + c.still = 0 + c.dir = -c.dir + } + c.prevRate = rate + limit += c.dir * step + if limit < 1 { + limit, c.dir = 1, 1 + } + if limit > maxAtOnce { + limit, c.dir = maxAtOnce, -1 + } + return limit +} + +// pool lets at most limit workers work at once; the limit can move while +// they do. +type pool struct { + mu sync.Mutex + cond *sync.Cond + active int + limit int +} + +func (p *pool) acquire(ctx context.Context) bool { + p.mu.Lock() + defer p.mu.Unlock() + for p.active >= p.limit { + if ctx.Err() != nil { + return false + } + p.cond.Wait() + } + if ctx.Err() != nil { + return false + } + p.active++ + return true +} + +func (p *pool) release() { + p.mu.Lock() + p.active-- + p.mu.Unlock() + p.cond.Broadcast() +} + +func (p *pool) setLimit(n int) { + p.mu.Lock() + p.limit = n + p.mu.Unlock() + p.cond.Broadcast() +} + +func (p *pool) getLimit() int { + p.mu.Lock() + defer p.mu.Unlock() + return p.limit +} diff --git a/internal/iopar/run_test.go b/internal/iopar/run_test.go new file mode 100644 index 00000000..48f3b74d --- /dev/null +++ b/internal/iopar/run_test.go @@ -0,0 +1,117 @@ +package iopar + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" +) + +// A drive that finishes the most with best operations at once: each +// operation takes longer the more run beside it, slowly up to best and +// steeply beyond. +func simulated(best int) func(i int) error { + var active atomic.Int64 + return func(int) error { + a := float64(active.Add(1)) + defer active.Add(-1) + over := a / float64(best) + lat := 200 * time.Microsecond + if over > 1 { + lat = time.Duration(float64(lat) * over * over * over) + } + time.Sleep(lat) + return nil + } +} + +// The simulated drive is timed with sub-millisecond sleeps, so a machine busy +// with other work (a full test run on a slow CPU) can make one run's windows +// look alike and leave it far off. Each kind gets three runs and one has to +// settle near the best; an adaptation that is actually broken misses in all. +func TestRun_SettlesNearWhatTheDriveDoesBest(t *testing.T) { + defer func(w time.Duration) { window = w }(window) + window = 60 * time.Millisecond + for _, kind := range []Kind{NVMe, SATA} { + near := false + for attempt := 1; attempt <= 3 && !near; attempt++ { + st, err := RunKind(context.Background(), kind, 40000, simulated(12), nil) + if err != nil { + t.Fatal(err) + } + t.Logf("%s, run %d: started at %d, settled at %d, peak %.0f/s in %v", kind, attempt, st.Start, st.Settled, st.Peak, st.Duration) + near = st.Settled >= 5 && st.Settled <= 28 + } + if !near { + t.Errorf("%s: no run settled near the 12 the drive does best", kind) + } + } +} + +func TestRun_HardDiskOneAtATime(t *testing.T) { + var active, most atomic.Int64 + st, err := RunKind(context.Background(), HDD, 300, func(int) error { + a := active.Add(1) + defer active.Add(-1) + for { + m := most.Load() + if a <= m || most.CompareAndSwap(m, a) { + break + } + } + time.Sleep(100 * time.Microsecond) + return nil + }, nil) + if err != nil { + t.Fatal(err) + } + if most.Load() != 1 || st.Settled != 1 { + t.Errorf("a hard disk ran %d at once (settled %d), want 1", most.Load(), st.Settled) + } +} + +func TestRun_EveryItemOnceAndProgress(t *testing.T) { + const n = 5000 + var seen [n]atomic.Int32 + var lastDone int + _, err := RunKind(context.Background(), NVMe, n, func(i int) error { + seen[i].Add(1) + return nil + }, func(done, total int) { + if total != n || done < lastDone { + t.Errorf("progress %d/%d after %d", done, total, lastDone) + } + lastDone = done + }) + if err != nil { + t.Fatal(err) + } + for i := range seen { + if seen[i].Load() != 1 { + t.Fatalf("item %d done %d times", i, seen[i].Load()) + } + } + if lastDone != n { + t.Errorf("progress ended at %d of %d", lastDone, n) + } +} + +func TestRun_StopsAtTheFirstError(t *testing.T) { + boom := errors.New("disk full") + var calls atomic.Int64 + _, err := RunKind(context.Background(), SATA, 100000, func(i int) error { + calls.Add(1) + if i == 50 { + return boom + } + time.Sleep(50 * time.Microsecond) + return nil + }, nil) + if !errors.Is(err, boom) { + t.Fatalf("err = %v, want the work's error", err) + } + if calls.Load() > 5000 { + t.Errorf("went on for %d items after the error", calls.Load()) + } +} diff --git a/internal/snapshot/extract.go b/internal/snapshot/extract.go new file mode 100644 index 00000000..f7ab5120 --- /dev/null +++ b/internal/snapshot/extract.go @@ -0,0 +1,137 @@ +package snapshot + +import ( + "archive/zip" + "context" + "fmt" + "io" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "sync/atomic" + + "github.com/opensave/opensave/internal/iopar" +) + +// Progress is told how far a long operation on a save has come: which phase +// ("rebuilding", "checking", "writing"), how many of how many done. +type Progress func(phase string, done, total int) + +// Phases of a restore, as Progress names them. +const ( + PhaseRebuilding = "rebuilding" + PhaseChecking = "checking" + PhaseWriting = "writing" +) + +// extractJob is one archive entry to write: under destDir, at rel. +type extractJob struct { + f *zip.File + destDir string + rel string +} + +// extractAll writes archive entries into place. Every folder is made first, +// once — not looked up again for each of a quarter of a million files — and +// the files are then written as many at a time as suits the drive they go to +// (iopar): one at a time on a spinning disk, many at once on an SSD, where +// written one after another a large save took most of a quarter of an hour +// at a fraction of what the drive can do. +func extractAll(jobs []extractJob, onDrive string, progress Progress) error { + type target struct { + f *zip.File + dest string + } + var files []target + dirs := map[string]struct{}{} + for _, j := range jobs { + dest, err := entryDest(j.f.Name, j.destDir, j.rel) + if err != nil { + return err + } + if j.f.FileInfo().IsDir() { + dirs[dest] = struct{}{} + continue + } + dirs[filepath.Dir(dest)] = struct{}{} + files = append(files, target{j.f, dest}) + } + ordered := make([]string, 0, len(dirs)) + for d := range dirs { + ordered = append(ordered, d) + } + sort.Strings(ordered) // a parent before what is inside it + for _, d := range ordered { + if err := os.MkdirAll(d, 0o777); err != nil { + return err + } + } + + var report func(done, total int) + if progress != nil { + report = func(done, total int) { progress(PhaseWriting, done, total) } + } + st, err := iopar.RunKind(context.Background(), driveKindOf(onDrive), len(files), func(i int) error { + return writeEntry(files[i].f, files[i].dest) + }, report) + lastExtract.Store(&st) + return err +} + +// driveKindOf is iopar.DriveKind; a variable, so a test can compare one at a +// time with many on the same drive. +var driveKindOf = iopar.DriveKind + +// lastExtract is how the last extraction went, for a test to report. +var lastExtract atomic.Pointer[iopar.Stats] + +// entryDest is where an entry is written: rel under destDir, refused if it +// would land outside it (zip-slip). Checked on the name actually written, +// since that is the one that decides where bytes land. +func entryDest(name, destDir, rel string) (string, error) { + clean := filepath.Clean(filepath.FromSlash(rel)) + if strings.HasPrefix(clean, "..") || filepath.IsAbs(clean) { + return "", fmt.Errorf("zip entry %q escapes destination", name) + } + return filepath.Join(destDir, clean), nil +} + +var copyBuffers = sync.Pool{New: func() any { b := make([]byte, 64<<10); return &b }} + +// writeEntry writes one file entry to dest, whose folder exists. +func writeEntry(f *zip.File, dest string) error { + src, err := f.Open() + if err != nil { + return err + } + defer src.Close() + dst, err := os.OpenFile(dest, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o666) + if err != nil { + return err + } + buf := copyBuffers.Get().(*[]byte) + defer copyBuffers.Put(buf) + if _, err := io.CopyBuffer(dst, src, *buf); err != nil { + dst.Close() + return err + } + return dst.Close() +} + +// progressFor is how a restore of gameID tells OnRestoreProgress how far it +// has come; nil when nobody listens. +func (m *Manager) progressFor(gameID string) Progress { + if m.OnRestoreProgress == nil { + return nil + } + return func(phase string, done, total int) { m.OnRestoreProgress(gameID, phase, done, total) } +} + +// progressDone tells OnRestoreProgress the restore of gameID is over. +func (m *Manager) progressDone(gameID string) { + if m.OnRestoreProgress != nil { + m.OnRestoreProgress(gameID, "", 0, 0) + } +} diff --git a/internal/snapshot/extractable.go b/internal/snapshot/extractable.go index e491a19f..4e682200 100644 --- a/internal/snapshot/extractable.go +++ b/internal/snapshot/extractable.go @@ -20,6 +20,11 @@ import ( // (GitHub #60). The same goes for an entry that would land outside its folder, // which extraction refuses. So the layout is checked here, as extraction would // lay it out, and a restore or branch switch stops before touching the save. +// +// A file listed twice is refused too, also when the two names differ only in +// case on a filesystem that ignores it. Extraction writes files in parallel, +// so the two entries would be two writers on one file at once, and what is +// left on disk could be neither of them. func CheckExtractable(zipPath string) error { r, err := zip.OpenReader(zipPath) if err != nil { @@ -52,6 +57,9 @@ func CheckExtractable(zipPath string) error { if other, ok := dirs[place]; ok { return fmt.Errorf("%w: %s is a file, but %s needs a folder of that name", ErrDamaged, f.Name, other) } + if other, ok := files[place]; ok { + return fmt.Errorf("%w: %s and %s are the same file", ErrDamaged, other, f.Name) + } files[place] = f.Name } // Every folder above the entry has to be a folder. diff --git a/internal/snapshot/restore_safety_test.go b/internal/snapshot/restore_safety_test.go index 96cf54bc..6ec4c75e 100644 --- a/internal/snapshot/restore_safety_test.go +++ b/internal/snapshot/restore_safety_test.go @@ -4,6 +4,7 @@ import ( "archive/zip" "os" "path/filepath" + "runtime" "strings" "testing" ) @@ -168,13 +169,16 @@ func TestCheckExtractable(t *testing.T) { entries []string ok bool }{ - "ordinary files and folders": {[]string{"a.sav", "slots/", "slots/1.sav", "slots/2.sav"}, true}, - "a file, then a file inside it": {[]string{"block", "block/child.sav"}, false}, - "a file inside, then the file": {[]string{"block/child.sav", "block"}, false}, - "a folder entry and a file": {[]string{"block/", "block"}, false}, - "an entry escaping its folder": {[]string{"../outside.sav"}, false}, - "the same clash in an extra one": {[]string{RootPrefix + "config/x", RootPrefix + "config/x/y"}, false}, - "the same name in two locations": {[]string{"x", RootPrefix + "config/x/y"}, true}, + "ordinary files and folders": {[]string{"a.sav", "slots/", "slots/1.sav", "slots/2.sav"}, true}, + "a file, then a file inside it": {[]string{"block", "block/child.sav"}, false}, + "a file inside, then the file": {[]string{"block/child.sav", "block"}, false}, + "a folder entry and a file": {[]string{"block/", "block"}, false}, + "an entry escaping its folder": {[]string{"../outside.sav"}, false}, + "the same clash in an extra one": {[]string{RootPrefix + "config/x", RootPrefix + "config/x/y"}, false}, + "the same name in two locations": {[]string{"x", RootPrefix + "config/x/y"}, true}, + "the same file twice": {[]string{"slots/1.sav", "slots/1.sav"}, false}, + "the same file twice in an extra": {[]string{RootPrefix + "config/x", RootPrefix + "config/x"}, false}, + "one file in two locations": {[]string{"x", RootPrefix + "config/x"}, true}, } { path := filepath.Join(dir, strings.ReplaceAll(name, " ", "_")+".zip") writeZip(t, path, c.entries...) @@ -183,3 +187,16 @@ func TestCheckExtractable(t *testing.T) { } } } + +// On a filesystem that ignores case, two names differing only in case are one +// file, and parallel extraction would write it twice at once. +func TestCheckExtractable_NamesDifferingOnlyInCase(t *testing.T) { + path := filepath.Join(t.TempDir(), "case.zip") + writeZip(t, path, "Slots/1.sav", "slots/1.sav") + err := CheckExtractable(path) + if caseInsensitive := runtime.GOOS == "windows" || runtime.GOOS == "darwin"; caseInsensitive && err == nil { + t.Errorf("CheckExtractable = nil, want the two names refused as one file") + } else if !caseInsensitive && err != nil { + t.Errorf("CheckExtractable = %v, want two different files accepted", err) + } +} diff --git a/internal/snapshot/shared.go b/internal/snapshot/shared.go index 81de0691..b3d8504d 100644 --- a/internal/snapshot/shared.go +++ b/internal/snapshot/shared.go @@ -140,6 +140,12 @@ var reading = struct { // with it. A missing snapshot is an error that matches fs.ErrNotExist; a // compacted one whose shared files are damaged, one that matches ErrDamaged. func OpenArchive(zipPath string) (path string, done func(), err error) { + return OpenArchiveProgress(zipPath, nil) +} + +// OpenArchiveProgress is OpenArchive telling progress how far a rebuild of a +// compacted archive has come. +func OpenArchiveProgress(zipPath string, progress Progress) (path string, done func(), err error) { reading.Lock() if _, statErr := os.Stat(zipPath); statErr == nil { reading.n[zipPath]++ @@ -165,7 +171,7 @@ func OpenArchive(zipPath string) (path string, done func(), err error) { return "", func() {}, err } name := tmp.Name() - err = rebuild(zipPath, tmp) + err = rebuild(zipPath, tmp, progress) if closeErr := tmp.Close(); err == nil { err = closeErr } @@ -250,7 +256,7 @@ func validSum(s string) bool { } // rebuild writes a compacted snapshot's archive back out whole. -func rebuild(zipPath string, out io.Writer) error { +func rebuild(zipPath string, out io.Writer, progress Progress) error { man, err := readManifest(manifestPath(zipPath)) if err != nil { return err @@ -265,7 +271,10 @@ func rebuild(zipPath string, out io.Writer) error { } w := zip.NewWriter(out) next := 0 - for _, e := range man.Entries { + for i, e := range man.Entries { + if progress != nil && i%500 == 0 { + progress(PhaseRebuilding, i, len(man.Entries)) + } if e.Blob != "" { if err := copyShared(w, store, e); err != nil { return err @@ -280,6 +289,9 @@ func rebuild(zipPath string, out io.Writer) error { } next++ } + if progress != nil { + progress(PhaseRebuilding, len(man.Entries), len(man.Entries)) + } return w.Close() } @@ -603,7 +615,7 @@ func sameArchive(zipPath string, orig *zip.ReadCloser) error { return err } defer os.Remove(tmp.Name()) - err = rebuild(zipPath, tmp) + err = rebuild(zipPath, tmp, nil) if closeErr := tmp.Close(); err == nil { err = closeErr } diff --git a/internal/snapshot/snapshot.go b/internal/snapshot/snapshot.go index 694a6ab7..b48bf4ad 100644 --- a/internal/snapshot/snapshot.go +++ b/internal/snapshot/snapshot.go @@ -44,8 +44,12 @@ type UploadHook func(zipPath, remoteFileName string) // Manager performs snapshot/branch operations against the store and // filesystem. type Manager struct { - Store *store.Store - OnUpload UploadHook + Store *store.Store + // OnRestoreProgress is told how far a restore of a game has come: its + // phase (PhaseRebuilding, PhaseChecking, PhaseWriting) and how many of + // how many done; phase "" when it is over. Optional. + OnRestoreProgress func(gameID, phase string, done, total int) + OnUpload UploadHook // OnCreated fires for every snapshot written, before its upload starts. // current is false for a copy kept of a save about to be replaced (see // CreateBeforeReplacing). Optional; runs on the snapshotting goroutine @@ -747,14 +751,19 @@ func (m *Manager) Restore(gameID, snapshotID string) (store.Snapshot, error) { // restore empties the save folder and then extracts; an archive found // damaged part-way through would leave neither the save that was there // nor this one. - archive, done, err := OpenArchive(snap.ZipPath) + progress := m.progressFor(gameID) + defer m.progressDone(gameID) + archive, done, err := OpenArchiveProgress(snap.ZipPath, progress) if err == nil { defer done() // Read back whole, and laid out as extraction would lay it out: an // archive whose checksums are fine can still hold a file where a // folder has to go, found only after the save had been cleared // (GitHub #60). - err = checkRestorable(archive) + err = VerifyArchiveProgress(archive, progress) + if err == nil { + err = CheckExtractable(archive) + } } else { err = damagedArchive(snap.ZipPath, err) } @@ -789,7 +798,7 @@ func (m *Manager) Restore(gameID, snapshotID string) (store.Snapshot, error) { if rootsErr != nil { restoreRoots = nil } - unplaced, err := UnzipRoots(restoreZip, game.SavePath, restoreRoots) + unplaced, err := UnzipRootsProgress(restoreZip, game.SavePath, restoreRoots, progress) for _, name := range unplaced { // Named but not placed: the snapshot holds files for a location this // device has no folder for. Saying so is the only honest option — diff --git a/internal/snapshot/verify.go b/internal/snapshot/verify.go index 69f1ff47..ecce0bf9 100644 --- a/internal/snapshot/verify.go +++ b/internal/snapshot/verify.go @@ -2,6 +2,7 @@ package snapshot import ( "archive/zip" + "context" "errors" "fmt" "io" @@ -10,6 +11,7 @@ import ( "sync" "time" + "github.com/opensave/opensave/internal/iopar" "github.com/opensave/opensave/internal/store" ) @@ -34,6 +36,14 @@ var ErrDamaged = errors.New("snapshot is damaged") // by its parts: its own small files read in full, and every shared file it // names read back against the hash of the bytes it was stored with. func VerifyArchive(path string) error { + return VerifyArchiveProgress(path, nil) +} + +// VerifyArchiveProgress is VerifyArchive telling progress how far it is. The +// entries are read as many at a time as suits the drive the archive is on +// (iopar): each is a small read of its own, and one after another a large +// save's quarter of a million of them took minutes. +func VerifyArchiveProgress(path string, progress Progress) error { r, err := zip.OpenReader(path) if err != nil { if errors.Is(err, fs.ErrNotExist) { @@ -45,15 +55,25 @@ func VerifyArchive(path string) error { return fmt.Errorf("%w: its archive cannot be opened: %v", ErrDamaged, err) } defer r.Close() + var files []*zip.File for _, f := range r.File { - if f.FileInfo().IsDir() { - continue + if !f.FileInfo().IsDir() { + files = append(files, f) } + } + var report func(done, total int) + if progress != nil { + report = func(done, total int) { progress(PhaseChecking, done, total) } + } + _, err = iopar.RunKind(context.Background(), driveKindOf(path), len(files), func(i int) error { + f := files[i] rc, err := f.Open() if err != nil { return fmt.Errorf("%w: %s cannot be read: %v", ErrDamaged, f.Name, err) } - _, err = io.Copy(io.Discard, rc) + buf := copyBuffers.Get().(*[]byte) + _, err = io.CopyBuffer(io.Discard, rc, *buf) + copyBuffers.Put(buf) rc.Close() if err != nil { if errors.Is(err, zip.ErrChecksum) { @@ -61,8 +81,9 @@ func VerifyArchive(path string) error { } return fmt.Errorf("%w: %s cannot be read: %v", ErrDamaged, f.Name, err) } - } - return nil + return nil + }, report) + return err } // Verify checks one snapshot and records what it found. diff --git a/internal/snapshot/zip.go b/internal/snapshot/zip.go index 803c60a8..f2d81dde 100644 --- a/internal/snapshot/zip.go +++ b/internal/snapshot/zip.go @@ -261,6 +261,11 @@ func newSnapshotWriter(out io.Writer) *zip.Writer { // directory and extracts into it — both matching unzipDirectory() in the // JS app. func UnzipTo(zipPath, targetPath string) error { + return UnzipToProgress(zipPath, targetPath, nil) +} + +// UnzipToProgress is UnzipTo telling progress how far it is. +func UnzipToProgress(zipPath, targetPath string, progress Progress) error { // Defence in depth, not the load-bearing guard. Extraction does not // restore the archived modification time — the written file gets the // current one — so the hash cache would notice on its own today. It is @@ -321,12 +326,11 @@ func UnzipTo(zipPath, targetPath string) error { } } + jobs := make([]extractJob, 0, len(r.File)) for _, entry := range r.File { - if err := extractEntry(entry, destDir); err != nil { - return err - } + jobs = append(jobs, extractJob{f: entry, destDir: destDir, rel: entry.Name}) } - return nil + return extractAll(jobs, destDir, progress) } func extractEntry(entry *zip.File, destDir string) error { diff --git a/internal/snapshot/zip_roots.go b/internal/snapshot/zip_roots.go index 8137c5f7..a83a8e0a 100644 --- a/internal/snapshot/zip_roots.go +++ b/internal/snapshot/zip_roots.go @@ -208,6 +208,11 @@ func rootOfEntry(entry string) (string, bool) { // restore — the operation someone reaches for when things have already gone // wrong. func UnzipRoots(zipPath, primary string, extra map[string]string) (unplaced []string, err error) { + return UnzipRootsProgress(zipPath, primary, extra, nil) +} + +// UnzipRootsProgress is UnzipRoots telling progress how far it is. +func UnzipRootsProgress(zipPath, primary string, extra map[string]string, progress Progress) (unplaced []string, err error) { // Every location this can write into, dropped from the hash cache. See // UnzipTo for why this is defence in depth rather than the guard that is // currently doing the work. @@ -231,7 +236,7 @@ func UnzipRoots(zipPath, primary string, extra map[string]string) (unplaced []st return nil, err } if len(roots) == 0 { - return nil, UnzipTo(zipPath, primary) + return nil, UnzipToProgress(zipPath, primary, progress) } r, err := zip.OpenReader(zipPath) @@ -268,6 +273,7 @@ func UnzipRoots(zipPath, primary string, extra map[string]string) (unplaced []st cleared := map[string]bool{} missing := map[string]bool{} + var jobs []extractJob for _, f := range r.File { name, isRoot := rootOfEntry(f.Name) target := primary @@ -303,9 +309,10 @@ func UnzipRoots(zipPath, primary string, extra map[string]string) (unplaced []st continue } } - if err := extractEntryAs(f, target, entry); err != nil { - return nil, err - } + jobs = append(jobs, extractJob{f: f, destDir: target, rel: entry}) + } + if err := extractAll(jobs, primary, progress); err != nil { + return nil, err } for name := range missing {