From 15f293d2276e5a030fcf88ef2398706435310ec0 Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 13 Aug 2026 12:34:49 +0800 Subject: [PATCH 1/2] pool: bring a claim's volumes up concurrently Cocoon serializes the attach itself per VM, so the win is overlapping the CLI spawns, device settle waits, and guest mounts across volumes. Single-volume claims run inline; per-volume ordering (marker before attach before mount) is unchanged and pinned. --- sandboxd/pool/pool_test.go | 19 ++++- sandboxd/pool/volume.go | 45 +++++++---- sandboxd/pool/volume_test.go | 142 ++++++++++++++++++++++++++++++++--- 3 files changed, 177 insertions(+), 29 deletions(-) diff --git a/sandboxd/pool/pool_test.go b/sandboxd/pool/pool_test.go index f7d8e34..e1ce3cd 100644 --- a/sandboxd/pool/pool_test.go +++ b/sandboxd/pool/pool_test.go @@ -809,6 +809,7 @@ type fakeEngine struct { staleReconciles []string // VM names ReconcileStaleCreate was called on installCAErr error diskAttachErr error + diskAttachErrFor string // volume name whose DiskAttach fails; "" = never diskAttachCancel context.CancelFunc mountVolumeErr error unmountVolumeErr error @@ -826,6 +827,8 @@ type fakeEngine struct { tap string // non-empty: lifecycle records carry this NIC tap listCount int + attachRendezvous *sync.WaitGroup // non-nil: DiskAttach waits there until every attach has arrived + cloneStall chan struct{} // non-nil: Clone blocks until closed probeStall chan struct{} // non-nil: Probe blocks until closed hibernateStall chan struct{} // non-nil: Hibernate blocks until closed @@ -1043,14 +1046,22 @@ func (f *fakeEngine) InstallCACert(_ context.Context, vsockSocket string, _ []by func (f *fakeEngine) DiskAttach(_ context.Context, _ string, spec engine.VolumeSpec) error { f.mu.Lock() - defer f.mu.Unlock() f.volumeSpecs = append(f.volumeSpecs, spec) f.volumeOps = append(f.volumeOps, "attach:"+spec.Name) f.attachDirty[spec.Name] = volumeDirty(spec.Path) - if f.diskAttachCancel != nil { - f.diskAttachCancel() + cancel, rendezvous, err := f.diskAttachCancel, f.attachRendezvous, f.diskAttachErr + if spec.Name == f.diskAttachErrFor { + err = errors.New("attach failed") } - return f.diskAttachErr + f.mu.Unlock() + if cancel != nil { + cancel() + } + if rendezvous != nil { + rendezvous.Done() + rendezvous.Wait() + } + return err } func (f *fakeEngine) MountVolume(_ context.Context, _, name, mount string, rw bool) error { diff --git a/sandboxd/pool/volume.go b/sandboxd/pool/volume.go index 6d313e2..f74d0bf 100644 --- a/sandboxd/pool/volume.go +++ b/sandboxd/pool/volume.go @@ -11,6 +11,7 @@ import ( "time" "github.com/projecteru2/core/log" + "golang.org/x/sync/errgroup" "github.com/cocoonstack/sandbox/sandboxd/engine" "github.com/cocoonstack/sandbox/sandboxd/types" @@ -177,30 +178,48 @@ func (m *Manager) confirmVolumesClean(volumes []resolvedVolume) error { return nil } -// applyVolumes attaches and mounts the resolved set; applied is that same set +// applyVolumes brings the resolved set up, a goroutine per volume past the +// first: cocoon serializes the attach itself, so what overlaps is the CLI +// spawn, the device settle wait and the guest mount. applied is that same set // in request shape, recorded on the sandbox only once every mount is up. func (m *Manager) applyVolumes(ctx context.Context, sb *types.Sandbox, volumes []resolvedVolume, applied []types.Volume) error { - if len(volumes) == 0 { + switch len(volumes) { + case 0: return nil - } - for _, volume := range volumes { - // Write-ahead: the marker must be durable before any guest write can be. - if volume.disk.RW { - if err := markVolumeDirty(volume.disk.Path); err != nil { - return fmt.Errorf("mark volume %q dirty: %w", volume.applied.Name, err) - } + case 1: + if err := m.applyVolume(ctx, sb, volumes[0]); err != nil { + return err } - if err := m.eng.DiskAttach(ctx, sb.VMName, volume.disk); err != nil { - return fmt.Errorf("attach volume %q: %w", volume.applied.Name, err) + default: + group, groupCtx := errgroup.WithContext(ctx) + for _, volume := range volumes { + group.Go(func() error { return m.applyVolume(groupCtx, sb, volume) }) } - if err := m.eng.MountVolume(ctx, sb.VsockSocket, volume.applied.Name, volume.applied.Mount, volume.disk.RW); err != nil { - return fmt.Errorf("setup volume %q: %w", volume.applied.Name, err) + if err := group.Wait(); err != nil { + return err } } sb.Volumes = applied return nil } +// applyVolume keeps one volume's steps strictly ordered; siblings overlap freely. +func (m *Manager) applyVolume(ctx context.Context, sb *types.Sandbox, volume resolvedVolume) error { + // Write-ahead: the marker must be durable before any guest write can be. + if volume.disk.RW { + if err := markVolumeDirty(volume.disk.Path); err != nil { + return fmt.Errorf("mark volume %q dirty: %w", volume.applied.Name, err) + } + } + if err := m.eng.DiskAttach(ctx, sb.VMName, volume.disk); err != nil { + return fmt.Errorf("attach volume %q: %w", volume.applied.Name, err) + } + if err := m.eng.MountVolume(ctx, sb.VsockSocket, volume.applied.Name, volume.applied.Mount, volume.disk.RW); err != nil { + return fmt.Errorf("setup volume %q: %w", volume.applied.Name, err) + } + return nil +} + // quiesceVolumes unmounts a claim's writable mounts in reverse order and // returns the teardown its VM removal must finish. A failed unmount never // blocks teardown: the image keeps its marker and waits for a recovering diff --git a/sandboxd/pool/volume_test.go b/sandboxd/pool/volume_test.go index abb82dc..cbc25c7 100644 --- a/sandboxd/pool/volume_test.go +++ b/sandboxd/pool/volume_test.go @@ -8,13 +8,15 @@ import ( "path/filepath" "slices" "strings" + "sync" "testing" "github.com/cocoonstack/sandbox/sandboxd/config" + "github.com/cocoonstack/sandbox/sandboxd/engine" "github.com/cocoonstack/sandbox/sandboxd/types" ) -func TestClaimProvisionAppliesVolumesInOrder(t *testing.T) { +func TestClaimProvisionAppliesVolumesInRequestOrder(t *testing.T) { first := writeVolumeImage(t, "imagenet.img", "first") second := writeVolumeImage(t, "weights.img", "second") eng := newFakeEngine() @@ -35,19 +37,20 @@ func TestClaimProvisionAppliesVolumesInOrder(t *testing.T) { if !slices.Equal(sb.Volumes, wantApplied) { t.Errorf("volumes=%v, want %v", sb.Volumes, wantApplied) } - wantOps := []string{ - "provision", "probe", - "attach:weights", "mount:weights:/models", - "attach:imagenet", "mount:imagenet:/volumes/imagenet", + wantLead := []string{"provision", "probe"} + if ops := eng.volumeOpsLog(); len(ops) < len(wantLead) || !slices.Equal(ops[:len(wantLead)], wantLead) { + t.Errorf("operations=%v, want %v ahead of every volume op", ops, wantLead) } - if !slices.Equal(eng.volumeOps, wantOps) { - t.Errorf("operations=%v, want %v", eng.volumeOps, wantOps) - } - if !slices.Equal(eng.volumeMounts, wantApplied) { - t.Errorf("mounts=%v, want %v", eng.volumeMounts, wantApplied) + assertVolumeBringUp(t, eng, wantApplied) + wantSpecs := []engine.VolumeSpec{ + {Name: "imagenet", Path: first, DirectIO: "off"}, + {Name: "weights", Path: second, DirectIO: "on"}, } - if got := eng.volumeSpecs; len(got) != 2 || got[0].Path != second || got[0].DirectIO != "on" || got[1].Path != first { - t.Errorf("attached specs=%+v", got) + specs := slices.SortedFunc(slices.Values(eng.volumeSpecs), func(a, b engine.VolumeSpec) int { + return strings.Compare(a.Name, b.Name) + }) + if !slices.Equal(specs, wantSpecs) { + t.Errorf("attached specs=%+v, want %+v", specs, wantSpecs) } persisted, err := newClaimStore(m.dataDir).load() if err != nil { @@ -70,6 +73,90 @@ func TestClaimProvisionAppliesVolumesInOrder(t *testing.T) { } } +func TestClaimProvisionBringsMixedVolumesUpConcurrently(t *testing.T) { + dataset := writeVolumeImage(t, "dataset.img", "dataset") + weights := writeVolumeImage(t, "weights.img", "weights") + scratch := writeVolumeImage(t, "scratch.img", "scratch") + cache := writeVolumeImage(t, "cache.img", "cache") + eng := newFakeEngine() + m := newVolumeManager(t, eng, []config.VolumeSpec{ + {Name: "dataset", Path: dataset}, + {Name: "weights", Path: weights}, + {Name: "scratch", Path: scratch, Writable: true}, + {Name: "cache", Path: cache, Writable: true}, + }) + requested := []types.Volume{ + {Name: "weights", Mount: "/models"}, + {Name: "scratch", Mode: types.VolumeModeRW}, + {Name: "dataset"}, + {Name: "cache", Mount: "/cache", Mode: types.VolumeModeRW}, + } + wantApplied := []types.Volume{ + {Name: "weights", Mount: "/models"}, + {Name: "scratch", Mount: "/volumes/scratch", Mode: types.VolumeModeRW}, + {Name: "dataset", Mount: "/volumes/dataset"}, + {Name: "cache", Mount: "/cache", Mode: types.VolumeModeRW}, + } + // Every attach must be in flight at once: a sequential apply blocks here. + var attaches sync.WaitGroup + attaches.Add(len(requested)) + eng.attachRendezvous = &attaches + + sb, err := m.ClaimProvision(t.Context(), testKey, 0, "", "", requested) + if err != nil { + t.Fatalf("ClaimProvision: %v", err) + } + if !slices.Equal(sb.Volumes, wantApplied) { + t.Errorf("volumes=%v, want the request order %v", sb.Volumes, wantApplied) + } + assertVolumeBringUp(t, eng, wantApplied) + if !volumeDirty(scratch) || !volumeDirty(cache) { + t.Error("live writable claim left an image unmarked") + } + if volumeDirty(dataset) || volumeDirty(weights) { + t.Error("read-only volume of a mixed claim was marked dirty") + } +} + +func TestClaimProvisionOneVolumeAttachFailureFailsWholeClaim(t *testing.T) { + first := writeVolumeImage(t, "first.img", "first") + broken := writeVolumeImage(t, "broken.img", "broken") + third := writeVolumeImage(t, "third.img", "third") + eng := newFakeEngine() + eng.diskAttachErrFor = "broken" + m := newVolumeManager(t, eng, []config.VolumeSpec{ + {Name: "first", Path: first, Writable: true}, + {Name: "broken", Path: broken, Writable: true}, + {Name: "third", Path: third}, + }) + + _, err := m.ClaimProvision(t.Context(), testKey, 0, "", "", []types.Volume{ + {Name: "first", Mode: types.VolumeModeRW}, + {Name: "broken", Mode: types.VolumeModeRW}, + {Name: "third"}, + }) + if err == nil || !strings.Contains(err.Error(), `attach volume "broken"`) { + t.Fatalf("ClaimProvision: %v, want the failing volume's attach error", err) + } + if len(eng.removes) != 1 { + t.Errorf("removes=%v, want the failed VM destroyed", eng.removes) + } + if !volumeDirty(first) || !volumeDirty(broken) { + t.Error("failed claim cleared a writable marker, opening an image no writer flushed") + } + if volumeDirty(third) { + t.Error("read-only volume was marked dirty") + } + for _, name := range []string{"first", "broken", "third"} { + if holders := volumeHoldersOf(m, name); holders != (volumeHolders{}) { + t.Errorf("registry for %s after the failure=%+v, want empty", name, holders) + } + } + if _, gauges := m.Info(); gauges.Claimed != 0 { + t.Errorf("claimed=%d, want 0", gauges.Claimed) + } +} + func TestClaimWarmAppliesVolumesAndRefillsAfterFailure(t *testing.T) { path := writeVolumeImage(t, "data.img", "data") for _, tt := range []struct { @@ -400,6 +487,37 @@ func TestClaimProvisionRejectsMissingVolumePathBeforeProvision(t *testing.T) { } } +// assertVolumeBringUp pins each volume's own marker→attach→mount order and the +// completeness of the set; cross-volume order is free — they come up together. +func assertVolumeBringUp(t *testing.T, eng *fakeEngine, applied []types.Volume) { + t.Helper() + ops := eng.volumeOpsLog() + brought := 0 + for _, op := range ops { + if strings.HasPrefix(op, "attach:") || strings.HasPrefix(op, "mount:") { + brought++ + } + } + if brought != 2*len(applied) { + t.Errorf("bring-up ops=%v, want one attach and one mount per volume of %v", ops, applied) + } + for _, volume := range applied { + attach := slices.Index(ops, "attach:"+volume.Name) + mount := slices.Index(ops, "mount:"+volume.Name+":"+volume.Mount) + if attach < 0 || mount < attach { + t.Errorf("volume %s: attach=%d mount=%d in %v, want the attach first", volume.Name, attach, mount, ops) + } + if volume.RW() && !eng.dirtyAtAttach(volume.Name) { + t.Errorf("volume %s attached before its dirty marker was durable", volume.Name) + } + } + byName := func(a, b types.Volume) int { return strings.Compare(a.Name, b.Name) } + mounts := slices.SortedFunc(slices.Values(eng.volumeMounts), byName) + if want := slices.SortedFunc(slices.Values(applied), byName); !slices.Equal(mounts, want) { + t.Errorf("mounts=%v, want %v", mounts, want) + } +} + func newVolumeManager(t *testing.T, eng *fakeEngine, volumes []config.VolumeSpec) *Manager { t.Helper() return newVolumeManagerAt(t, eng, t.TempDir(), volumes) From c444e53b8e2a87965ec761ff65c587228ce17269 Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 13 Aug 2026 18:15:17 +0800 Subject: [PATCH 2/2] clean --- sandboxd/pool/volume.go | 7 +++---- sandboxd/pool/volume_test.go | 2 +- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/sandboxd/pool/volume.go b/sandboxd/pool/volume.go index f74d0bf..1984457 100644 --- a/sandboxd/pool/volume.go +++ b/sandboxd/pool/volume.go @@ -178,10 +178,9 @@ func (m *Manager) confirmVolumesClean(volumes []resolvedVolume) error { return nil } -// applyVolumes brings the resolved set up, a goroutine per volume past the -// first: cocoon serializes the attach itself, so what overlaps is the CLI -// spawn, the device settle wait and the guest mount. applied is that same set -// in request shape, recorded on the sandbox only once every mount is up. +// applyVolumes brings the resolved set up concurrently: cocoon serializes the +// attach per VM, so what overlaps is the CLI spawns, settle waits and guest +// mounts. applied is the request-shaped set, recorded once every mount is up. func (m *Manager) applyVolumes(ctx context.Context, sb *types.Sandbox, volumes []resolvedVolume, applied []types.Volume) error { switch len(volumes) { case 0: diff --git a/sandboxd/pool/volume_test.go b/sandboxd/pool/volume_test.go index cbc25c7..29ad7f2 100644 --- a/sandboxd/pool/volume_test.go +++ b/sandboxd/pool/volume_test.go @@ -488,7 +488,7 @@ func TestClaimProvisionRejectsMissingVolumePathBeforeProvision(t *testing.T) { } // assertVolumeBringUp pins each volume's own marker→attach→mount order and the -// completeness of the set; cross-volume order is free — they come up together. +// completeness of the set; cross-volume order is free. func assertVolumeBringUp(t *testing.T, eng *fakeEngine, applied []types.Volume) { t.Helper() ops := eng.volumeOpsLog()