Delta apply: log why nothing was taken over; cluster results (#49)

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

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
+}

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)
+	}
 }

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 {

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.

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)
 	}
 }