diff --git a/deploy/csi/README.md b/deploy/csi/README.md index e97ebd0..d20ccb9 100644 --- a/deploy/csi/README.md +++ b/deploy/csi/README.md @@ -77,6 +77,11 @@ On the lab node (`onggi pull --base`, the same pull the plugin runs): Through the node cache, a pull that needs only a few chunks finds them by scanning the cache's pack indexes instead of loading its whole index (3.3 M chunks), and index lookups search a contiguous array of digest prefixes (#42). +In the lab cluster (#41; `npu-node-1`, `--rebase-idle 1m`): a pod with decode B8 was Running 12 s after creation, +taking over the idle B2 artifact (15.0 GB left in place, 28.3 MB read, 7.5 s pull; byte-identical, read-only), against +about a minute before. A B2 pod started while the B8 pod ran did not touch the B8 artifact. A prefill pod took over +nothing: the plugin logs why (`took over nothing: ... keeps x %` or `no artifact to take over`, #49). + ## Limits of v0 - The files are a second copy next to the node cache, which holds packed and compressed chunks. Reflinks or diff --git a/pkg/csi/materializer.go b/pkg/csi/materializer.go index 6baa18f..f739805 100644 --- a/pkg/csi/materializer.go +++ b/pkg/csi/materializer.go @@ -190,8 +190,12 @@ func (m *Materializer) materialize(d digest.Digest) error { for _, c := range *claimed { os.RemoveAll(c) // what the pull did not move of a consumed base } - if st.Base != "" { + switch { + case st.Base != "": m.logf("pull of %s took over %s: %d B left in place", d, st.Base, st.Reused) + case err == nil && offer != nil: + m.logf("pull of %s took over nothing: of %d idle artifacts the best keeps %d of %d B at the same offsets (%.1f %%, %.0f %% needed)", + d, len(offer.Bases), st.BaseBest, st.Bytes, 100*float64(st.BaseBest)/float64(max(st.Bytes, 1)), 100*store.DefaultMinSame) } if err != nil { m.logf("materializing %s failed after %s: %v", d, time.Since(start).Round(time.Millisecond), err) @@ -326,13 +330,23 @@ func (m *Materializer) offer(d digest.Digest) (*Offer, *[]string) { m.mu.Lock() ds, _ := m.materialized() var bases []store.Base + inUse, recent := 0, 0 for _, b := range ds { - if m.idle(b, time.Now()) && b != d { + switch { + case b == d: + case m.idle(b, time.Now()): bases = append(bases, store.Base{Manifest: b, Dir: m.Dir(b)}) + case m.busy(b): + inUse++ + default: + recent++ } } m.mu.Unlock() if len(bases) == 0 { + if len(ds) > 0 { + m.logf("pull of %s: no artifact to take over (%d in use, %d used within %s)", d, inUse, recent, m.RebaseIdle) + } return nil, claimed } sort.Slice(bases, func(i, j int) bool { return m.lastUsed(bases[i].Manifest).After(m.lastUsed(bases[j].Manifest)) }) @@ -373,11 +387,17 @@ func (m *Materializer) offer(d digest.Digest) (*Offer, *[]string) { // idle says whether artifact d may be consumed: materialized, no volume, no pull, unused for // RebaseIdle. The caller holds m.mu. func (m *Materializer) idle(d digest.Digest, now time.Time) bool { - if _, busy := m.inflight[d]; busy || !isDir(m.Dir(d)) { - return false - } - if n, err := m.users(d); err != nil || n > 0 { + if m.busy(d) || !isDir(m.Dir(d)) { return false } return now.Sub(m.lastUsed(d)) >= m.RebaseIdle } + +// busy says whether a volume or a pull uses artifact d. The caller holds m.mu. +func (m *Materializer) busy(d digest.Digest) bool { + if _, ok := m.inflight[d]; ok { + return true + } + n, err := m.users(d) + return err != nil || n > 0 +} diff --git a/pkg/csi/rebase_test.go b/pkg/csi/rebase_test.go index 15765f3..98ae3eb 100644 --- a/pkg/csi/rebase_test.go +++ b/pkg/csi/rebase_test.go @@ -2,12 +2,40 @@ package csi import ( "context" + "fmt" "os" "path/filepath" + "strings" + "sync" "testing" "time" + + "github.com/ai-computing/onggi/pkg/store" ) +// logs collects a materializer's log lines. +type logs struct { + mu sync.Mutex + lines []string +} + +func (l *logs) logf(format string, args ...any) { + l.mu.Lock() + defer l.mu.Unlock() + l.lines = append(l.lines, fmt.Sprintf(format, args...)) +} + +func (l *logs) has(s string) bool { + l.mu.Lock() + defer l.mu.Unlock() + for _, x := range l.lines { + if strings.Contains(x, s) { + return true + } + } + return false +} + // An idle artifact is turned into the next one; then back again (#41). func TestEnsureRebasesFromIdleArtifact(t *testing.T) { ctx := context.Background() @@ -54,6 +82,8 @@ func TestEnsureKeepsBusyArtifacts(t *testing.T) { f := newFixture(t) m, c := newMaterializer(t, f) m.RebaseIdle = time.Hour + var l logs + m.Logf = l.logf if _, err := m.Ensure(ctx, f.a); err != nil { t.Fatal(err) } @@ -63,6 +93,9 @@ func TestEnsureKeepsBusyArtifacts(t *testing.T) { if c.pulls[1].Base != "" || !isDir(m.Dir(f.a)) { t.Fatalf("took a recently used artifact: %+v", c.pulls[1]) } + if !l.has("no artifact to take over (0 in use, 1 used within 1h0m0s)") { + t.Fatalf("no reason logged: %q", l.lines) + } m2, c2 := newMaterializer(t, f) m2.RebaseIdle = time.Nanosecond @@ -78,4 +111,23 @@ func TestEnsureKeepsBusyArtifacts(t *testing.T) { if c2.pulls[1].Base != "" || !isDir(m2.Dir(f.a)) { t.Fatalf("took an artifact in use: %+v", c2.pulls[1]) } + // in use, then released: idle at once with RebaseIdle 1 ns, but too different from a third + // artifact to be worth taking over + if _, _, _, err := m2.Release("csi-vol1"); err != nil { + t.Fatal(err) + } + var l2 logs + m2.Logf = l2.logf + src := t.TempDir() + os.WriteFile(filepath.Join(src, "model.bin"), random(9, 1<<20), 0o644) + other, _, err := f.s.Push(ctx, src, store.PushOptions{}) + if err != nil { + t.Fatal(err) + } + if _, err := m2.Ensure(ctx, other); err != nil { + t.Fatal(err) + } + if !l2.has("took over nothing: of 2 idle artifacts the best keeps 0 of") { + t.Fatalf("no reason logged: %q", l2.lines) + } } diff --git a/pkg/store/pull.go b/pkg/store/pull.go index 8b1847b..7df96f7 100644 --- a/pkg/store/pull.go +++ b/pkg/store/pull.go @@ -51,6 +51,7 @@ type PullStats struct { Cached int64 `json:"cached"` // stored bytes added to the node cache Base string `json:"base,omitempty"` // the artifact this pull turned into the new one Reused int64 `json:"reused"` // bytes left in place in the base's files + BaseBest int64 `json:"base_best"` // bytes the best offered base keeps at the same offsets, taken or not (#49) Ranges int `json:"ranges"` Duration time.Duration `json:"duration_ns"` } @@ -146,7 +147,9 @@ func (s *Store) Pull(ctx context.Context, manifest digest.Digest, dest string, o if minSame <= 0 { minSame = DefaultMinSame } - if p, ok := s.chooseBase(ctx, m, extentsOf(nodes, len(m.Files), false), o.Bases, minSame); ok { + p, ok := s.chooseBase(ctx, m, extentsOf(nodes, len(m.Files), false), o.Bases, minSame) + st.BaseBest = p.same + if ok { if dir, ok := o.Claim(p.base); ok { st.Base = p.base.Manifest.String() if st.Reused, err = rebase(p, dir, m, nodes, handles, done); err != nil { diff --git a/pkg/store/rebase.go b/pkg/store/rebase.go index 908c8df..3678160 100644 --- a/pkg/store/rebase.go +++ b/pkg/store/rebase.go @@ -32,8 +32,8 @@ type rebasePlan struct { same int64 } -// chooseBase returns the base that keeps the most bytes in place, if that is at least minSame -// of m's bytes. +// chooseBase returns the base that keeps the most bytes in place, and whether that is at least +// minSame of m's bytes. func (s *Store) chooseBase(ctx context.Context, m *Manifest, newExt [][]Extent, bases []Base, minSame float64) (rebasePlan, bool) { var best rebasePlan for _, b := range bases { @@ -64,10 +64,7 @@ func (s *Store) chooseBase(ctx context.Context, m *Manifest, newExt [][]Extent, best = p } } - if best.same == 0 || float64(best.same) < minSame*float64(m.Bytes) { - return rebasePlan{}, false - } - return best, true + return best, best.same > 0 && float64(best.same) >= minSame*float64(m.Bytes) } // sameAt says whether base holds e's chunk at e's offset. diff --git a/pkg/store/rebase_test.go b/pkg/store/rebase_test.go index 497002c..9d771b1 100644 --- a/pkg/store/rebase_test.go +++ b/pkg/store/rebase_test.go @@ -51,7 +51,8 @@ func TestRebaseWritesOnlyWhatChanged(t *testing.T) { } sameTree(t, src, out) t.Logf("%+v", st) - if claimed != 1 || st.Base != b.Manifest.String() || st.Reused < int64(len(v))*9/10 || st.ReadBytes > int64(len(v))/10 { + if claimed != 1 || st.Base != b.Manifest.String() || st.Reused < int64(len(v))*9/10 || st.ReadBytes > int64(len(v))/10 || + st.BaseBest < st.Reused { t.Fatalf("rebase: claimed %d, %+v", claimed, st) } if _, err := os.Stat(filepath.Join(b.Dir, "decode_B2.rbln")); !os.IsNotExist(err) { @@ -74,7 +75,7 @@ func TestRebaseNeedsEnoughInPlace(t *testing.T) { t.Fatal(err) } sameTree(t, src, out) - if st.Base != "" || st.Reused != 0 { + if st.Base != "" || st.Reused != 0 || st.BaseBest >= 2<<20 { // the best keeps less than half t.Fatalf("%+v", st) } }