已开启
feat: support incremental memory checkpoints for StratoVirt #155
Yekelu创建于 8月11日
feat: support incremental memory checkpoints for StratoVirt #155
已开启
Yekelu创建于 8月11日
57 个文件变更+6486-194
@@ -1,4 +1,4 @@
1-.PHONY: build clean test fmt vet lint help gen-proto gen-proto-go gen-proto-py mod-tidy mod-vendor build-conch-init-initramfs1+.PHONY: build clean test fmt vet lint help gen-proto gen-proto-go gen-proto-py mod-tidy mod-vendor build-conch-cow build-conch-init-initramfs
2 2 
3# project name3# project name
4PROJECT_NAME := Conch4PROJECT_NAME := Conch
@@ -111,6 +111,11 @@ build-%: ## Build specific binary (e.g., make build-conchd)
111 @mkdir -p $(BIN_DIR)111 @mkdir -p $(BIN_DIR)
112 $(GOBUILD) -ldflags "$(VERSION_LDFLAGS)" -o $(BIN_DIR)/$* ./cmd/$*112 $(GOBUILD) -ldflags "$(VERSION_LDFLAGS)" -o $(BIN_DIR)/$* ./cmd/$*
113 113 
114+build-conch-cow: ## Build the conch-cow daemon
115+ @echo "building cmd/conch-cow..."
116+ @mkdir -p $(BIN_DIR)
117+ $(GOBUILD) -ldflags "$(VERSION_LDFLAGS)" -o $(BIN_DIR)/conch-cow ./cmd/conch-cow
118+ 
114build-conch-init-initramfs: ## Build minimal initramfs that runs conch-init as PID 1119build-conch-init-initramfs: ## Build minimal initramfs that runs conch-init as PID 1
115 @echo "building static conch-init for initramfs..."120 @echo "building static conch-init for initramfs..."
116 @mkdir -p $(BIN_DIR)121 @mkdir -p $(BIN_DIR)
@@ -0,0 +1,75 @@
1+package main
2+ 
3+import (
4+ "context"
5+ "errors"
6+ "flag"
7+ "fmt"
8+ "log"
9+ "os"
10+ "os/signal"
11+ "syscall"
12+ 
13+ "github.com/openeuler/Conch/internal/cow"
14+)
15+ 
16+type commandOptions struct {
17+ socketPath string
18+}
19+ 
20+func main() {
21+ ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
22+ defer stop()
23+ if err := run(ctx, os.Args[1:]); err != nil {
24+ log.Printf("conch-cow: %v", err)
25+ os.Exit(1)
26+ }
27+}
28+ 
29+func parseOptions(args []string) (commandOptions, bool, error) {
30+ flags := flag.NewFlagSet("conch-cow", flag.ContinueOnError)
31+ flags.SetOutput(os.Stdout)
32+ flags.Usage = func() {
33+ _, _ = fmt.Fprintln(flags.Output(), "Usage: conch-cow [--socket PATH]")
34+ _, _ = fmt.Fprintln(flags.Output(), "Serve on-demand Guest memory pages for Conch.")
35+ flags.PrintDefaults()
36+ }
37+ socketPath := flags.String("socket", cow.DefaultSocketPath, "Unix control socket path")
38+ if err := flags.Parse(args); err != nil {
39+ if errors.Is(err, flag.ErrHelp) {
40+ return commandOptions{}, true, nil
41+ }
42+ return commandOptions{}, false, err
43+ }
44+ if flags.NArg() != 0 {
45+ return commandOptions{}, false, fmt.Errorf("positional arguments are not accepted: %v", flags.Args())
46+ }
47+ return commandOptions{socketPath: *socketPath}, false, nil
48+}
49+ 
50+func run(ctx context.Context, args []string) error {
51+ options, help, err := parseOptions(args)
52+ if err != nil || help {
53+ return err
54+ }
55+ server := cow.NewServer(options.socketPath)
56+ serveDone := make(chan error, 1)
57+ go func() { serveDone <- server.Serve(ctx) }()
58+ 
59+ select {
60+ case <-server.Ready():
61+ case serveErr := <-serveDone:
62+ return errors.Join(serveErr, server.Close())
63+ case <-ctx.Done():
64+ closeErr := server.Close()
65+ return errors.Join(closeErr, <-serveDone)
66+ }
67+ 
68+ select {
69+ case serveErr := <-serveDone:
70+ return errors.Join(serveErr, server.Close())
71+ case <-ctx.Done():
72+ closeErr := server.Close()
73+ return errors.Join(closeErr, <-serveDone)
74+ }
75+}
@@ -0,0 +1,57 @@
1+package main
2+ 
3+import (
4+ "context"
5+ "errors"
6+ "os"
7+ "path/filepath"
8+ "testing"
9+ "time"
10+ 
11+ "github.com/openeuler/Conch/internal/cow"
12+)
13+ 
14+func TestParseOptionsUsesCowSocketDefault(t *testing.T) {
15+ options, help, err := parseOptions(nil)
16+ if err != nil || help {
17+ t.Fatalf("parseOptions() help=%v err=%v", help, err)
18+ }
19+ if options.socketPath != cow.DefaultSocketPath {
20+ t.Fatalf("socket = %q, want %q", options.socketPath, cow.DefaultSocketPath)
21+ }
22+ if _, _, err := parseOptions([]string{"extra"}); err == nil {
23+ t.Fatal("parseOptions accepted a positional argument")
24+ }
25+ if _, help, err := parseOptions([]string{"--help"}); err != nil || !help {
26+ t.Fatalf("parseOptions(--help) help=%v err=%v", help, err)
27+ }
28+}
29+ 
30+func TestRunServesUntilContextCancellation(t *testing.T) {
31+ socketPath := filepath.Join(t.TempDir(), "cow.sock")
32+ ctx, cancel := context.WithCancel(context.Background())
33+ done := make(chan error, 1)
34+ go func() { done <- run(ctx, []string{"--socket", socketPath}) }()
35+ deadline := time.Now().Add(time.Second)
36+ for {
37+ if _, err := os.Stat(socketPath); err == nil {
38+ break
39+ }
40+ if time.Now().After(deadline) {
41+ t.Fatal("conch-cow did not create its socket")
42+ }
43+ time.Sleep(time.Millisecond)
44+ }
45+ cancel()
46+ select {
47+ case err := <-done:
48+ if err != nil {
49+ t.Fatal(err)
50+ }
51+ case <-time.After(2 * time.Second):
52+ t.Fatal("conch-cow did not stop after cancellation")
53+ }
54+ if _, err := os.Lstat(socketPath); !errors.Is(err, os.ErrNotExist) {
55+ t.Fatalf("socket remains after run returned: %v", err)
56+ }
57+}
@@ -42,6 +42,12 @@ vmm:
42 binary: /usr/bin/stratovirt42 binary: /usr/bin/stratovirt
43 43 
44sandbox:44sandbox:
45+ # Global StratoVirt memory policy: full, auto, or incremental. Defaults to full.
Z
Zzhaoyifan2825 天前

暂不实现auto,让用户显式指定,不支持就报错

likedislike
46+ memory_mode: full
47+ # conch-cow binary started and stopped by conchd.
48+ cow_binary: /usr/bin/conch-cow
49+ # Local conch-cow control socket.
50+ cow_socket: /run/conch/cow.sock
45 vsock_signal_retry: 10ms51 vsock_signal_retry: 10ms
46 vsock_signal_timeout: 60s52 vsock_signal_timeout: 60s
47 request_timeout: 60s53 request_timeout: 60s
Mgo.mod+1-1
@@ -15,6 +15,7 @@ require (
15 github.com/coreos/go-systemd/v22 v22.7.015 github.com/coreos/go-systemd/v22 v22.7.0
16 github.com/creack/pty v1.1.2416 github.com/creack/pty v1.1.24
17 github.com/erofs/erofs-container-toolkit v0.0.0-20260123120957-823f29fa15cb17 github.com/erofs/erofs-container-toolkit v0.0.0-20260123120957-823f29fa15cb
18+ github.com/google/uuid v1.6.0
18 github.com/moby/sys/mountinfo v0.7.219 github.com/moby/sys/mountinfo v0.7.2
19 github.com/opencontainers/go-digest v1.0.020 github.com/opencontainers/go-digest v1.0.0
20 github.com/opencontainers/image-spec v1.1.121 github.com/opencontainers/image-spec v1.1.1
@@ -58,7 +59,6 @@ require (
58 github.com/godbus/dbus/v5 v5.1.0 // indirect59 github.com/godbus/dbus/v5 v5.1.0 // indirect
59 github.com/gogo/protobuf v1.3.2 // indirect60 github.com/gogo/protobuf v1.3.2 // indirect
60 github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 // indirect61 github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 // indirect
61- github.com/google/uuid v1.6.0 // indirect
62 github.com/intel/goresctrl v0.12.0 // indirect62 github.com/intel/goresctrl v0.12.0 // indirect
63 github.com/klauspost/compress v1.18.5 // indirect63 github.com/klauspost/compress v1.18.5 // indirect
64 github.com/mdlayher/socket v0.5.1 // indirect64 github.com/mdlayher/socket v0.5.1 // indirect
@@ -17,6 +17,7 @@ import (
17 "github.com/openeuler/Conch/internal/daemon/state"17 "github.com/openeuler/Conch/internal/daemon/state"
18 conchimage "github.com/openeuler/Conch/internal/image"18 conchimage "github.com/openeuler/Conch/internal/image"
19 "github.com/openeuler/Conch/internal/image/erofsconvert"19 "github.com/openeuler/Conch/internal/image/erofsconvert"
20+ "github.com/openeuler/Conch/internal/memorymode"
20 "github.com/openeuler/Conch/internal/netstack"21 "github.com/openeuler/Conch/internal/netstack"
21 "github.com/openeuler/Conch/internal/runtimeapi"22 "github.com/openeuler/Conch/internal/runtimeapi"
22 "github.com/openeuler/Conch/internal/sandbox"23 "github.com/openeuler/Conch/internal/sandbox"
@@ -33,6 +34,10 @@ type SandboxOps interface {
33 Checkpoint(sandbox.CheckpointRequest) (sandbox.CheckpointResult, error)34 Checkpoint(sandbox.CheckpointRequest) (sandbox.CheckpointResult, error)
34}35}
35 36 
37+type checkpointCompleter interface {
38+ CompleteCheckpoint(sandbox.LifecycleRequest) error
39+}
40+ 
36// ErrTemplateIDRequired reports that neither the request nor conchd supplied41// ErrTemplateIDRequired reports that neither the request nor conchd supplied
37// a usable template ID for sandbox creation.42// a usable template ID for sandbox creation.
38var ErrTemplateIDRequired = errors.New("template_id is required and no default_template_id is configured")43var ErrTemplateIDRequired = errors.New("template_id is required and no default_template_id is configured")
@@ -44,13 +49,17 @@ type SnapshotOps interface {
44}49}
45 50 
46type Service struct {51type Service struct {
47- Sandbox SandboxOps52+ Sandbox SandboxOps
48- Containerd *containerdclient.Client53+ Containerd *containerdclient.Client
49- Snapshot SnapshotOps54+ Snapshot SnapshotOps
50- Store state.Store55+ Store state.Store
51- Templates conchtemplate.Store56+ Templates conchtemplate.Store
52- SandboxDefaults SandboxDefaults57+ SandboxDefaults SandboxDefaults
53- lifecycleLocks sandboxLifecycleLocks58+ lifecycleLocks sandboxLifecycleLocks
59+ memoryPolicyConfigured bool
60+ memoryRequested memorymode.RequestedMode
61+ memoryCapabilities memorymode.CapabilityProvider
62+ memoryBootInspector func(context.Context, string) (conchimage.BootIndexInfo, error)
54}63}
55 64 
56var ErrSandboxAlreadyExists = errors.New("sandbox already exists")65var ErrSandboxAlreadyExists = errors.New("sandbox already exists")
@@ -106,6 +115,15 @@ func (s *Service) SetSandboxDefaults(defaults SandboxDefaults) {
106 s.SandboxDefaults = defaults115 s.SandboxDefaults = defaults
107}116}
108 117 
118+func (s *Service) SetMemoryPolicy(requested memorymode.RequestedMode, capabilities memorymode.CapabilityProvider) {
119+ if s == nil {
120+ return
121+ }
122+ s.memoryPolicyConfigured = true
123+ s.memoryRequested = requested
124+ s.memoryCapabilities = capabilities
125+}
126+ 
109func (s *Service) CreateSandbox(ctx context.Context, opts SandboxCreateOptions) (SandboxCreateResult, error) {127func (s *Service) CreateSandbox(ctx context.Context, opts SandboxCreateOptions) (SandboxCreateResult, error) {
110 if s == nil || s.Sandbox == nil {128 if s == nil || s.Sandbox == nil {
111 return SandboxCreateResult{}, fmt.Errorf("sandbox service is not configured")129 return SandboxCreateResult{}, fmt.Errorf("sandbox service is not configured")
@@ -134,6 +152,10 @@ func (s *Service) CreateSandbox(ctx context.Context, opts SandboxCreateOptions)
134 if opts.TemplateID == "" {152 if opts.TemplateID == "" {
135 return SandboxCreateResult{}, ErrTemplateIDRequired153 return SandboxCreateResult{}, ErrTemplateIDRequired
136 }154 }
155+ memoryMode, err := s.resolveCreateMemoryMode(ctx, opts)
156+ if err != nil {
157+ return SandboxCreateResult{}, err
158+ }
137 if err := netstack.ValidateSandboxNetworkInputConfig(ctx, opts.Network); err != nil {159 if err := netstack.ValidateSandboxNetworkInputConfig(ctx, opts.Network); err != nil {
138 return SandboxCreateResult{}, err160 return SandboxCreateResult{}, err
139 }161 }
@@ -154,6 +176,7 @@ func (s *Service) CreateSandbox(ctx context.Context, opts SandboxCreateOptions)
154 Env: copyMap(opts.Env),176 Env: copyMap(opts.Env),
155 VolumeMounts: opts.VolumeMounts,177 VolumeMounts: opts.VolumeMounts,
156 Network: opts.Network,178 Network: opts.Network,
179+ MemoryMode: string(memoryMode),
157 }180 }
158 181 
159 createdAt := time.Now().UnixNano()182 createdAt := time.Now().UnixNano()
@@ -191,6 +214,46 @@ func (s *Service) CreateSandbox(ctx context.Context, opts SandboxCreateOptions)
191 }, nil214 }, nil
192}215}
193 216 
217+func (s *Service) resolveCreateMemoryMode(ctx context.Context, opts SandboxCreateOptions) (memorymode.EffectiveMode, error) {
218+ if !s.memoryPolicyConfigured {
219+ return memorymode.EffectiveFull, nil
220+ }
221+ if s.Templates == nil {
222+ return "", fmt.Errorf("template store is not configured")
223+ }
224+ entry, err := s.Templates.Get(ctx, opts.TemplateID)
225+ if err != nil {
226+ return "", fmt.Errorf("get template for memory mode: %w", err)
227+ }
228+ inspector := s.memoryBootInspector
229+ if inspector == nil {
230+ if s.Containerd == nil {
231+ return "", fmt.Errorf("containerd client is not configured for memory mode preflight")
232+ }
233+ inspector = func(ctx context.Context, digest string) (conchimage.BootIndexInfo, error) {
234+ return conchimage.InspectBootIndex(ctx, s.Containerd, digest)
235+ }
236+ }
237+ info, err := inspector(ctx, entry.BootIndexDigest)
238+ if err != nil {
239+ return "", fmt.Errorf("inspect template boot index for memory mode: %w", err)
240+ }
241+ vmmName := opts.VMMName
242+ if info.Resume {
243+ vmmName = info.VMMName
244+ }
245+ mode, err := memorymode.Resolve(ctx, s.memoryCapabilities, memorymode.Input{
246+ Requested: s.memoryRequested,
247+ VMMName: vmmName,
248+ Resume: info.Resume,
249+ ArtifactFormat: info.MemoryFormat,
250+ })
251+ if err != nil {
252+ return "", err
253+ }
254+ return mode, nil
255+}
256+ 
194func (s *Service) UpdateSandboxNetworkConfig(ctx context.Context, opts SandboxNetworkUpdateOptions) error {257func (s *Service) UpdateSandboxNetworkConfig(ctx context.Context, opts SandboxNetworkUpdateOptions) error {
195 if s == nil || s.Sandbox == nil {258 if s == nil || s.Sandbox == nil {
196 return fmt.Errorf("sandbox service is not configured")259 return fmt.Errorf("sandbox service is not configured")
@@ -353,7 +416,11 @@ func (s *Service) CheckpointSandbox(ctx context.Context, opts SandboxCheckpointO
353 if err != nil {416 if err != nil {
354 return SandboxCheckpointResult{}, err417 return SandboxCheckpointResult{}, err
355 }418 }
356- defer os.RemoveAll(captured.MemRootPath)419+ cleanupPath := captured.CleanupPath
420+ if strings.TrimSpace(cleanupPath) == "" {
421+ cleanupPath = captured.MemRootPath
422+ }
423+ defer os.RemoveAll(cleanupPath)
357 424 
358 bootIndexTag := "localhost/conch/template:" + templateID425 bootIndexTag := "localhost/conch/template:" + templateID
359 published, err := conchimage.PublishCheckpointBootIndex(ctx, s.Containerd, conchimage.PublishCheckpointBootIndexOptions{426 published, err := conchimage.PublishCheckpointBootIndex(ctx, s.Containerd, conchimage.PublishCheckpointBootIndexOptions{
@@ -362,6 +429,7 @@ func (s *Service) CheckpointSandbox(ctx context.Context, opts SandboxCheckpointO
362 MemRoot: captured.MemRootPath,429 MemRoot: captured.MemRootPath,
363 VMMName: captured.VMMName,430 VMMName: captured.VMMName,
364 MemorySizeMB: captured.MemorySizeMB,431 MemorySizeMB: captured.MemorySizeMB,
432+ MemoryFormat: captured.MemoryFormat,
365 })433 })
366 if err != nil {434 if err != nil {
367 return SandboxCheckpointResult{}, err435 return SandboxCheckpointResult{}, err
@@ -394,6 +462,13 @@ func (s *Service) CheckpointSandbox(ctx context.Context, opts SandboxCheckpointO
394 captured.MemorySizeMB,462 captured.MemorySizeMB,
395 )463 )
396 }464 }
465+ if info.MemoryFormat != captured.MemoryFormat {
466+ return SandboxCheckpointResult{}, fmt.Errorf(
467+ "validated checkpoint memory format %s does not match captured format %s",
468+ info.MemoryFormat,
469+ captured.MemoryFormat,
470+ )
471+ }
397 if err := s.Store.PublishCheckpoint(ctx, conchtemplate.Entry{472 if err := s.Store.PublishCheckpoint(ctx, conchtemplate.Entry{
398 ID: templateID,473 ID: templateID,
399 Origin: conchtemplate.OriginCheckpoint,474 Origin: conchtemplate.OriginCheckpoint,
@@ -407,6 +482,15 @@ func (s *Service) CheckpointSandbox(ctx context.Context, opts SandboxCheckpointO
407 }); err != nil {482 }); err != nil {
408 return SandboxCheckpointResult{}, err483 return SandboxCheckpointResult{}, err
409 }484 }
485+ if captured.MemoryFormat == conchimage.MemoryFormatIncrementalV1 {
486+ completer, ok := s.Sandbox.(checkpointCompleter)
487+ if !ok {
488+ return SandboxCheckpointResult{}, fmt.Errorf("sandbox manager cannot complete incremental checkpoint")
489+ }
490+ if err := completer.CompleteCheckpoint(sandbox.LifecycleRequest{SandboxID: sandboxID}); err != nil {
491+ return SandboxCheckpointResult{}, fmt.Errorf("complete incremental checkpoint: %w", err)
492+ }
493+ }
410 return SandboxCheckpointResult{494 return SandboxCheckpointResult{
411 TemplateID: templateID,495 TemplateID: templateID,
412 BootIndexDigest: info.BootIndexDigest,496 BootIndexDigest: info.BootIndexDigest,
@@ -15,8 +15,10 @@ import (
15 "github.com/opencontainers/go-digest"15 "github.com/opencontainers/go-digest"
16 containerdclient "github.com/openeuler/Conch/internal/adapters/containerd/client"16 containerdclient "github.com/openeuler/Conch/internal/adapters/containerd/client"
17 containerdhost "github.com/openeuler/Conch/internal/adapters/containerd/host"17 containerdhost "github.com/openeuler/Conch/internal/adapters/containerd/host"
18+ "github.com/openeuler/Conch/internal/cow"
18 "github.com/openeuler/Conch/internal/daemon/state"19 "github.com/openeuler/Conch/internal/daemon/state"
19 conchimage "github.com/openeuler/Conch/internal/image"20 conchimage "github.com/openeuler/Conch/internal/image"
21+ "github.com/openeuler/Conch/internal/memorymode"
20 "github.com/openeuler/Conch/internal/netstack"22 "github.com/openeuler/Conch/internal/netstack"
21 "github.com/openeuler/Conch/internal/sandbox"23 "github.com/openeuler/Conch/internal/sandbox"
22 conchtemplate "github.com/openeuler/Conch/internal/template"24 conchtemplate "github.com/openeuler/Conch/internal/template"
@@ -33,6 +35,8 @@ type fakeSandboxOps struct {
33 deleteErr error35 deleteErr error
34 updateReq sandbox.NetworkUpdateRequest36 updateReq sandbox.NetworkUpdateRequest
35 updateErr error37 updateErr error
38+ completeRequests []sandbox.LifecycleRequest
39+ completeErr error
36}40}
37 41 
38type serializedDeleteOps struct {42type serializedDeleteOps struct {
@@ -96,6 +100,11 @@ func (f *fakeSandboxOps) Checkpoint(req sandbox.CheckpointRequest) (sandbox.Chec
96 return sandbox.CheckpointResult{}, nil100 return sandbox.CheckpointResult{}, nil
97}101}
98 102 
103+func (f *fakeSandboxOps) CompleteCheckpoint(req sandbox.LifecycleRequest) error {
104+ f.completeRequests = append(f.completeRequests, req)
105+ return f.completeErr
106+}
107+ 
99func TestCheckpointSandboxPublishesCaptureAndAtomicallyAdvancesHead(t *testing.T) {108func TestCheckpointSandboxPublishesCaptureAndAtomicallyAdvancesHead(t *testing.T) {
100 ctx := context.Background()109 ctx := context.Background()
101 host := newRuntimeImageHost(t)110 host := newRuntimeImageHost(t)
@@ -168,6 +177,50 @@ func TestCheckpointSandboxPublishesCaptureAndAtomicallyAdvancesHead(t *testing.T
168 }177 }
169}178}
170 179 
180+func TestCheckpointSandboxClearsIncrementalPoisonOnlyAfterHeadAdvances(t *testing.T) {
181+ ctx := context.Background()
182+ host := newRuntimeImageHost(t)
183+ t0Digest := buildColdBootIndex(t, host, "incremental-checkpoint-t0")
184+ memRoot := t.TempDir()
185+ if err := os.WriteFile(filepath.Join(memRoot, "state"), []byte("state"), 0o600); err != nil {
186+ t.Fatal(err)
187+ }
188+ if err := os.WriteFile(filepath.Join(memRoot, "memory"), []byte("metadata"), 0o600); err != nil {
189+ t.Fatal(err)
190+ }
191+ completeErr := errors.New("completion failed")
192+ ops := &fakeSandboxOps{
193+ checkpointResults: []sandbox.CheckpointResult{{
194+ MemRootPath: memRoot, CleanupPath: memRoot, VMMName: "stratovirt", MemorySizeMB: 256,
195+ MemoryFormat: conchimage.MemoryFormatIncrementalV1,
196+ }},
197+ completeErr: completeErr,
198+ }
199+ store := newTestStore(t)
200+ svc := New(ops, host.Client(), store)
201+ seedTemplate(t, ctx, svc.Templates, "t0", t0Digest, conchtemplate.BootModeCold)
202+ if err := store.UpsertSandbox(ctx, state.SandboxRecord{
203+ SandboxID: "sandbox-incremental", CheckpointHeadTemplateID: "t0", CheckpointHeadBootIndexDigest: t0Digest,
204+ }); err != nil {
205+ t.Fatal(err)
206+ }
207+ 
208+ _, err := svc.CheckpointSandbox(ctx, SandboxCheckpointOptions{SandboxID: "sandbox-incremental"})
209+ if !errors.Is(err, completeErr) {
210+ t.Fatalf("CheckpointSandbox() error = %v", err)
211+ }
212+ if len(ops.completeRequests) != 1 || ops.completeRequests[0].SandboxID != "sandbox-incremental" {
213+ t.Fatalf("completion requests = %#v", ops.completeRequests)
214+ }
215+ record, err := store.GetSandbox(ctx, "sandbox-incremental")
216+ if err != nil {
217+ t.Fatal(err)
218+ }
219+ if record.CheckpointHeadTemplateID == "t0" || record.CheckpointHeadBootIndexDigest == t0Digest {
220+ t.Fatalf("checkpoint head was not advanced before completion: %#v", record)
221+ }
222+}
223+ 
171func TestCheckpointSandboxDoesNotPersistBeforeValidationSucceeds(t *testing.T) {224func TestCheckpointSandboxDoesNotPersistBeforeValidationSucceeds(t *testing.T) {
172 ctx := context.Background()225 ctx := context.Background()
173 host := newRuntimeImageHost(t)226 host := newRuntimeImageHost(t)
@@ -218,8 +271,8 @@ func TestCheckpointSandboxBuildsConsecutiveTemplateLineage(t *testing.T) {
218 memRoot1 := t.TempDir()271 memRoot1 := t.TempDir()
219 memRoot2 := t.TempDir()272 memRoot2 := t.TempDir()
220 sandboxOps := &fakeSandboxOps{checkpointResults: []sandbox.CheckpointResult{273 sandboxOps := &fakeSandboxOps{checkpointResults: []sandbox.CheckpointResult{
221- {MemRootPath: memRoot1, VMMName: "stratovirt", MemorySizeMB: 256},274+ {MemRootPath: memRoot1, VMMName: "stratovirt", MemorySizeMB: 256, MemoryFormat: conchimage.MemoryFormatFullV1},
222- {MemRootPath: memRoot2, VMMName: "stratovirt", MemorySizeMB: 256},275+ {MemRootPath: memRoot2, VMMName: "stratovirt", MemorySizeMB: 256, MemoryFormat: conchimage.MemoryFormatFullV1},
223 }}276 }}
224 store := newTestStore(t)277 store := newTestStore(t)
225 svc := New(sandboxOps, host.Client(), store)278 svc := New(sandboxOps, host.Client(), store)
@@ -394,6 +447,79 @@ func TestCreateSandboxStoresAPIAndCheckpointMetadata(t *testing.T) {
394 }447 }
395}448}
396 449 
450+func TestCreateSandboxResolvesGlobalMemoryModeBeforeRuntimeAllocation(t *testing.T) {
451+ tests := []struct {
452+ name string
453+ requested memorymode.RequestedMode
454+ info conchimage.BootIndexInfo
455+ capability cow.Capabilities
456+ capabilityErr error
457+ wantMode string
458+ precondition bool
459+ wantError bool
460+ }{
461+ {
462+ name: "auto falls back only for explicit unsupported capability", requested: memorymode.RequestedAuto,
463+ info: conchimage.BootIndexInfo{VMMName: "", Resume: false},
464+ capability: cow.Capabilities{IncrementalMemory: cow.CapabilityUnsupported}, wantMode: "full",
465+ },
466+ {
467+ name: "auto operational error fails before create", requested: memorymode.RequestedAuto,
468+ info: conchimage.BootIndexInfo{Resume: false}, capabilityErr: errors.New("cow unavailable"), wantError: true,
469+ },
470+ {
471+ name: "incremental cannot restore full artifact", requested: memorymode.RequestedIncremental,
472+ info: conchimage.BootIndexInfo{Resume: true, VMMName: "stratovirt", MemoryFormat: conchimage.MemoryFormatFullV1},
473+ precondition: true, wantError: true,
474+ },
475+ }
476+ for _, test := range tests {
477+ t.Run(test.name, func(t *testing.T) {
478+ store := newTestStore(t)
479+ ops := &fakeSandboxOps{createResult: sandbox.CreateResult{BootIndexDigest: digest.FromString(t.Name()).String()}}
480+ svc := New(ops, nil, store)
481+ entry, err := svc.Templates.Create(context.Background(), conchtemplate.Entry{
482+ ID: "tmpl-memory", Origin: conchtemplate.OriginImage, BootMode: conchtemplate.BootModeCold,
483+ BootIndexDigest: digest.FromString("memory-policy").String(), ImageName: "local/memory-policy",
484+ })
485+ if err != nil {
486+ t.Fatal(err)
487+ }
488+ provider := &runtimeMemoryCapabilities{capabilities: test.capability, err: test.capabilityErr}
489+ svc.SetMemoryPolicy(test.requested, provider)
490+ svc.memoryBootInspector = func(context.Context, string) (conchimage.BootIndexInfo, error) { return test.info, nil }
491+ 
492+ _, err = svc.CreateSandbox(context.Background(), SandboxCreateOptions{
493+ SandboxID: "sandbox-memory", TemplateID: entry.ID, VMMName: "stratovirt", RamMB: 256,
494+ })
495+ if test.wantError {
496+ if err == nil || errors.Is(err, memorymode.ErrPrecondition) != test.precondition {
497+ t.Fatalf("CreateSandbox() error = %v, precondition=%v", err, errors.Is(err, memorymode.ErrPrecondition))
498+ }
499+ if ops.createCalls != 0 {
500+ t.Fatalf("runtime Create() calls = %d, want 0", ops.createCalls)
501+ }
502+ return
503+ }
504+ if err != nil {
505+ t.Fatalf("CreateSandbox() error = %v", err)
506+ }
507+ if ops.req.MemoryMode != test.wantMode {
508+ t.Fatalf("runtime memory mode = %q, want %q", ops.req.MemoryMode, test.wantMode)
509+ }
510+ })
511+ }
512+}
513+ 
514+type runtimeMemoryCapabilities struct {
515+ capabilities cow.Capabilities
516+ err error
517+}
518+ 
519+func (provider *runtimeMemoryCapabilities) Capabilities(context.Context) (cow.Capabilities, error) {
520+ return provider.capabilities, provider.err
521+}
522+ 
397func TestCreateSandboxFailureDoesNotPersistRecord(t *testing.T) {523func TestCreateSandboxFailureDoesNotPersistRecord(t *testing.T) {
398 ctx := context.Background()524 ctx := context.Background()
399 store := newTestStore(t)525 store := newTestStore(t)
@@ -95,6 +95,9 @@ const (
95)95)
96 96 
97type SandboxConfig struct {97type SandboxConfig struct {
98+ MemoryMode string `yaml:"memory_mode"`
99+ CowBinary string `yaml:"cow_binary"`
100+ CowSocket string `yaml:"cow_socket"`
98 VsockSignalRetry time.Duration `yaml:"vsock_signal_retry"`101 VsockSignalRetry time.Duration `yaml:"vsock_signal_retry"`
99 VsockSignalTimeout time.Duration `yaml:"vsock_signal_timeout"`102 VsockSignalTimeout time.Duration `yaml:"vsock_signal_timeout"`
100 RequestTimeout time.Duration `yaml:"request_timeout"`103 RequestTimeout time.Duration `yaml:"request_timeout"`
@@ -153,6 +156,9 @@ func DefaultConfig() *Config {
153 StateDir: "/run/conch/containerd",156 StateDir: "/run/conch/containerd",
154 },157 },
155 Sandbox: SandboxConfig{158 Sandbox: SandboxConfig{
159+ MemoryMode: "full",
160+ CowBinary: "/usr/bin/conch-cow",
161+ CowSocket: "/run/conch/cow.sock",
156 VsockSignalRetry: 10 * time.Millisecond,162 VsockSignalRetry: 10 * time.Millisecond,
157 VsockSignalTimeout: 60 * time.Second,163 VsockSignalTimeout: 60 * time.Second,
158 RequestTimeout: 60 * time.Second,164 RequestTimeout: 60 * time.Second,
@@ -268,6 +274,15 @@ func LoadConfig(configPath string) (*Config, error) {
268 if cfg.Containerd.StateDir == "" {274 if cfg.Containerd.StateDir == "" {
269 cfg.Containerd.StateDir = defaultCfg.Containerd.StateDir275 cfg.Containerd.StateDir = defaultCfg.Containerd.StateDir
270 }276 }
277+ if strings.TrimSpace(cfg.Sandbox.MemoryMode) == "" {
278+ cfg.Sandbox.MemoryMode = defaultCfg.Sandbox.MemoryMode
279+ }
280+ if strings.TrimSpace(cfg.Sandbox.CowBinary) == "" {
281+ cfg.Sandbox.CowBinary = defaultCfg.Sandbox.CowBinary
282+ }
283+ if strings.TrimSpace(cfg.Sandbox.CowSocket) == "" {
284+ cfg.Sandbox.CowSocket = defaultCfg.Sandbox.CowSocket
285+ }
271 if cfg.Sandbox.VsockSignalRetry == 0 {286 if cfg.Sandbox.VsockSignalRetry == 0 {
272 cfg.Sandbox.VsockSignalRetry = defaultCfg.Sandbox.VsockSignalRetry287 cfg.Sandbox.VsockSignalRetry = defaultCfg.Sandbox.VsockSignalRetry
273 }288 }
@@ -318,6 +333,19 @@ func LoadConfig(configPath string) (*Config, error) {
318}333}
319 334 
320func validateConfig(cfg *Config) error {335func validateConfig(cfg *Config) error {
336+ switch cfg.Sandbox.MemoryMode {
337+ case "full", "auto", "incremental":
338+ default:
339+ return fmt.Errorf("invalid sandbox.memory_mode=%q: must be full, auto, or incremental", cfg.Sandbox.MemoryMode)
340+ }
341+ if !filepath.IsAbs(cfg.Sandbox.CowBinary) {
342+ return fmt.Errorf("invalid sandbox.cow_binary=%q: must be an absolute path", cfg.Sandbox.CowBinary)
343+ }
344+ cfg.Sandbox.CowBinary = filepath.Clean(cfg.Sandbox.CowBinary)
345+ if !filepath.IsAbs(cfg.Sandbox.CowSocket) {
346+ return fmt.Errorf("invalid sandbox.cow_socket=%q: must be an absolute path", cfg.Sandbox.CowSocket)
347+ }
348+ cfg.Sandbox.CowSocket = filepath.Clean(cfg.Sandbox.CowSocket)
321 if cfg.Network.WarmPoolSize < 0 {349 if cfg.Network.WarmPoolSize < 0 {
322 return fmt.Errorf("invalid network.warm_pool_size=%d: must be greater than or equal to 0", cfg.Network.WarmPoolSize)350 return fmt.Errorf("invalid network.warm_pool_size=%d: must be greater than or equal to 0", cfg.Network.WarmPoolSize)
323 }351 }
@@ -478,3 +478,56 @@ func TestDefaultVMMNameStaysStratovirt(t *testing.T) {
478 t.Fatalf("DefaultConfig().Sandbox.DefaultVMMName = %q, want stratovirt", got)478 t.Fatalf("DefaultConfig().Sandbox.DefaultVMMName = %q, want stratovirt", got)
479 }479 }
480}480}
481+ 
482+func TestSandboxMemoryConfigDefaultsAndValidation(t *testing.T) {
483+ defaults := DefaultConfig()
484+ if defaults.Sandbox.MemoryMode != "full" {
485+ t.Fatalf("default memory mode = %q, want full", defaults.Sandbox.MemoryMode)
486+ }
487+ if defaults.Sandbox.CowSocket != "/run/conch/cow.sock" {
488+ t.Fatalf("default cow socket = %q", defaults.Sandbox.CowSocket)
489+ }
490+ if defaults.Sandbox.CowBinary != "/usr/bin/conch-cow" {
491+ t.Fatalf("default cow binary = %q", defaults.Sandbox.CowBinary)
492+ }
493+ 
494+ configPath := filepath.Join(t.TempDir(), "config.yaml")
495+ if err := os.WriteFile(configPath, []byte("sandbox:\n memory_mode: auto\n cow_binary: /opt/conch/conch-cow\n cow_socket: /tmp/custom-cow.sock\n"), 0o640); err != nil {
496+ t.Fatal(err)
497+ }
498+ loaded, err := LoadConfig(configPath)
499+ if err != nil {
500+ t.Fatal(err)
501+ }
502+ if loaded.Sandbox.MemoryMode != "auto" || loaded.Sandbox.CowBinary != "/opt/conch/conch-cow" || loaded.Sandbox.CowSocket != "/tmp/custom-cow.sock" {
503+ t.Fatalf("loaded sandbox memory config = %#v", loaded.Sandbox)
504+ }
505+ 
506+ relativeBinaryPath := filepath.Join(t.TempDir(), "config.yaml")
507+ if err := os.WriteFile(relativeBinaryPath, []byte("sandbox:\n cow_binary: bin/conch-cow\n"), 0o640); err != nil {
508+ t.Fatal(err)
509+ }
510+ if _, err := LoadConfig(relativeBinaryPath); err == nil || !strings.Contains(err.Error(), "sandbox.cow_binary") {
511+ t.Fatalf("LoadConfig(relative cow_binary) error = %v, want sandbox.cow_binary", err)
512+ }
513+ 
514+ for _, value := range []string{"unknown", "FULL"} {
515+ t.Run(value, func(t *testing.T) {
516+ path := filepath.Join(t.TempDir(), "config.yaml")
517+ if err := os.WriteFile(path, []byte("sandbox:\n memory_mode: "+value+"\n"), 0o640); err != nil {
518+ t.Fatal(err)
519+ }
520+ if _, err := LoadConfig(path); err == nil || !strings.Contains(err.Error(), "sandbox.memory_mode") {
521+ t.Fatalf("LoadConfig() error = %v, want sandbox.memory_mode", err)
522+ }
523+ })
524+ }
525+ 
526+ legacyPath := filepath.Join(t.TempDir(), "config.yaml")
527+ if err := os.WriteFile(legacyPath, []byte("sandbox:\n memd_socket: /run/conch/memd.sock\n"), 0o640); err != nil {
528+ t.Fatal(err)
529+ }
530+ if _, err := LoadConfig(legacyPath); err == nil || !strings.Contains(err.Error(), "field memd_socket not found") {
531+ t.Fatalf("LoadConfig(legacy memd_socket) error = %v", err)
532+ }
533+}
@@ -0,0 +1,142 @@
1+package cow
2+ 
3+import (
4+ "context"
5+ "fmt"
6+ "net"
7+ "os"
8+ "time"
9+ 
10+ "github.com/google/uuid"
11+)
12+ 
13+const (
14+ requestTimeout = 5 * time.Second
15+ defaultUFFDAcceptTimeout = 30 * time.Second
16+ waitReadyTimeoutMargin = 2 * time.Second
17+)
18+ 
19+type Client struct {
20+ socketPath string
21+ requestTimeout time.Duration
22+ waitTimeout time.Duration
23+}
24+ 
25+func NewClient(socketPath string) *Client {
26+ if socketPath == "" {
27+ socketPath = DefaultSocketPath
28+ }
29+ return &Client{
30+ socketPath: socketPath,
31+ requestTimeout: requestTimeout,
32+ waitTimeout: defaultUFFDAcceptTimeout + waitReadyTimeoutMargin,
33+ }
34+}
35+ 
36+func (client *Client) Capabilities(ctx context.Context) (Capabilities, error) {
37+ response, err := client.simple(ctx, Request{Type: RequestCapabilities}, nil)
38+ if err != nil {
39+ return Capabilities{}, err
40+ }
41+ if response.Capabilities == nil {
42+ return Capabilities{}, fmt.Errorf("cow capabilities response is missing report")
43+ }
44+ switch response.Capabilities.IncrementalMemory {
45+ case CapabilitySupported, CapabilityUnsupported, CapabilityUnknown:
46+ default:
47+ return Capabilities{}, fmt.Errorf(
48+ "cow capabilities response has invalid incremental memory state %q",
49+ response.Capabilities.IncrementalMemory,
50+ )
51+ }
52+ return *response.Capabilities, nil
53+}
54+ 
55+func (client *Client) Attach(ctx context.Context, request Request) (*os.File, Response, error) {
56+ request.Type = RequestAttach
57+ conn, request, err := client.start(ctx, request, nil)
58+ if err != nil {
59+ return nil, Response{}, err
60+ }
61+ defer conn.Close()
62+ var response Response
63+ fds, err := readFrame(conn, &response)
64+ if err != nil {
65+ return nil, Response{}, err
66+ }
67+ expectedFDs := 0
68+ if response.OK {
69+ expectedFDs = 1
70+ }
71+ if err := validateResponse(request.RequestID, response, fds, expectedFDs); err != nil {
72+ return nil, Response{}, err
73+ }
74+ if !response.OK {
75+ return nil, response, fmt.Errorf("cow Attach failed: %s", response.Error)
76+ }
77+ file := os.NewFile(uintptr(fds[0]), "conch-cow-memory")
78+ if file == nil {
79+ closeFDs(fds)
80+ return nil, response, fmt.Errorf("wrap cow memory descriptor")
81+ }
82+ return file, response, nil
83+}
84+ 
85+func (client *Client) WaitAttachmentReady(ctx context.Context, token, sandboxID string) (Response, error) {
86+ return client.simple(ctx, Request{Type: RequestWaitAttachmentReady, Token: token, SandboxID: sandboxID}, nil)
87+}
88+ 
89+func (client *Client) Detach(ctx context.Context, token string) (Response, error) {
90+ return client.simple(ctx, Request{Type: RequestDetach, Token: token}, nil)
91+}
92+ 
93+func (client *Client) simple(ctx context.Context, request Request, fds []int) (Response, error) {
94+ conn, request, err := client.start(ctx, request, fds)
Z
Zzhaoyifan2825 天前

client.start不涉及fds参数,应移除

likedislike
95+ if err != nil {
96+ return Response{}, err
97+ }
98+ defer conn.Close()
99+ var response Response
100+ responseFDs, err := readFrame(conn, &response)
101+ if err != nil {
102+ return Response{}, err
103+ }
104+ if err := validateResponse(request.RequestID, response, responseFDs, 0); err != nil {
105+ return Response{}, err
106+ }
107+ if !response.OK {
108+ return response, fmt.Errorf("cow %s failed: %s", request.Type, response.Error)
109+ }
110+ return response, nil
111+}
112+ 
113+func (client *Client) start(ctx context.Context, request Request, fds []int) (*net.UnixConn, Request, error) {
114+ timeout := client.requestTimeout
115+ if request.Type == RequestWaitAttachmentReady {
116+ timeout = client.waitTimeout
117+ }
118+ requestContext, cancel := context.WithTimeout(ctx, timeout)
119+ defer cancel()
120+ request.ProtocolVersion = ProtocolVersion
121+ if request.RequestID == "" {
122+ request.RequestID = uuid.NewString()
123+ }
124+ dialer := net.Dialer{}
125+ connection, err := dialer.DialContext(requestContext, "unix", client.socketPath)
126+ if err != nil {
127+ return nil, request, fmt.Errorf("dial cow: %w", err)
128+ }
129+ conn, ok := connection.(*net.UnixConn)
130+ if !ok {
131+ _ = connection.Close()
132+ return nil, request, fmt.Errorf("cow connection is not a Unix socket")
133+ }
134+ if deadline, ok := requestContext.Deadline(); ok {
135+ _ = conn.SetDeadline(deadline)
136+ }
137+ if err := writeFrame(conn, request, fds); err != nil {
Z
Zzhaoyifan2825 天前

client.start不涉及fds参数,可以传nil

likedislike
138+ _ = conn.Close()
139+ return nil, request, err
140+ }
141+ return conn, request, nil
142+}
@@ -0,0 +1,153 @@
1+package cow
2+ 
3+import (
4+ "bytes"
5+ "encoding/binary"
6+ "encoding/json"
7+ "fmt"
8+ "io"
9+ "net"
10+ 
11+ "golang.org/x/sys/unix"
12+)
13+ 
14+const (
15+ maxFrameSize = 1 << 20
16+ maxFrameFDs = 8
17+)
18+ 
19+var defaultFrameOOBLen = unix.CmsgSpace(maxFrameFDs * 4)
20+ 
21+func writeFrame(conn *net.UnixConn, message any, fds []int) error {
22+ payload, err := json.Marshal(message)
23+ if err != nil {
24+ return fmt.Errorf("marshal frame: %w", err)
25+ }
26+ if len(payload) > maxFrameSize {
27+ return fmt.Errorf("frame too large: %d", len(payload))
28+ }
29+ if len(fds) > maxFrameFDs {
30+ return fmt.Errorf("too many frame descriptors: %d", len(fds))
31+ }
32+ frame := make([]byte, 4+len(payload))
33+ binary.BigEndian.PutUint32(frame[:4], uint32(len(payload)))
34+ copy(frame[4:], payload)
35+ var rights []byte
36+ if len(fds) != 0 {
37+ rights = unix.UnixRights(fds...)
38+ }
39+ written, _, err := conn.WriteMsgUnix(frame, rights, nil)
40+ if err != nil {
41+ return fmt.Errorf("write frame: %w", err)
42+ }
43+ if written < len(frame) {
44+ if _, err := conn.Write(frame[written:]); err != nil {
45+ return fmt.Errorf("finish frame write: %w", err)
46+ }
47+ }
48+ return nil
49+}
50+ 
51+func readFrame(conn *net.UnixConn, message any) ([]int, error) {
52+ return readFrameWithOOBSize(conn, message, defaultFrameOOBLen)
53+}
54+ 
55+func readFrameWithOOBSize(conn *net.UnixConn, message any, oobSize int) ([]int, error) {
Z
Zzhaoyifan2825 天前

该函数可内联

likedislike
56+ header := make([]byte, 4)
57+ oob := make([]byte, oobSize)
58+ read, oobRead, flags, _, err := conn.ReadMsgUnix(header, oob)
59+ if err != nil {
60+ return nil, fmt.Errorf("read frame header: %w", err)
61+ }
62+ fds, err := parseFDs(oob[:oobRead])
63+ if err != nil {
64+ return nil, err
65+ }
66+ if flags&unix.MSG_CTRUNC != 0 {
67+ closeFDs(fds)
68+ return nil, fmt.Errorf("frame ancillary data truncated")
69+ }
70+ if flags&unix.MSG_TRUNC != 0 {
71+ closeFDs(fds)
72+ return nil, fmt.Errorf("frame header truncated")
73+ }
74+ if read < len(header) {
75+ if _, err := io.ReadFull(conn, header[read:]); err != nil {
76+ closeFDs(fds)
77+ return nil, fmt.Errorf("read frame header: %w", err)
78+ }
79+ }
80+ length := binary.BigEndian.Uint32(header)
81+ if length > maxFrameSize {
82+ closeFDs(fds)
83+ return nil, fmt.Errorf("frame too large: %d", length)
84+ }
85+ payload := make([]byte, int(length))
86+ if _, err := io.ReadFull(conn, payload); err != nil {
87+ closeFDs(fds)
88+ return nil, fmt.Errorf("read frame payload: %w", err)
89+ }
90+ decoder := json.NewDecoder(bytes.NewReader(payload))
91+ decoder.DisallowUnknownFields()
92+ if err := decoder.Decode(message); err != nil {
93+ closeFDs(fds)
94+ return nil, fmt.Errorf("decode frame: %w", err)
95+ }
96+ if err := decoder.Decode(&struct{}{}); err != io.EOF {
97+ closeFDs(fds)
98+ if err == nil {
99+ return nil, fmt.Errorf("decode frame: multiple JSON values")
100+ }
101+ return nil, fmt.Errorf("decode frame trailing data: %w", err)
102+ }
103+ return fds, nil
104+}
105+ 
106+func parseFDs(oob []byte) ([]int, error) {
107+ if len(oob) == 0 {
108+ return nil, nil
109+ }
110+ messages, err := unix.ParseSocketControlMessage(oob)
111+ if err != nil {
112+ return nil, fmt.Errorf("parse socket control message: %w", err)
113+ }
114+ var fds []int
115+ for _, message := range messages {
116+ if message.Header.Level != unix.SOL_SOCKET || message.Header.Type != unix.SCM_RIGHTS {
117+ closeFDs(fds)
118+ return nil, fmt.Errorf("unexpected socket control message")
119+ }
120+ rights, err := unix.ParseUnixRights(&message)
121+ if err != nil {
122+ closeFDs(fds)
123+ return nil, fmt.Errorf("parse SCM_RIGHTS: %w", err)
124+ }
125+ for _, fd := range rights {
126+ unix.CloseOnExec(fd)
127+ }
128+ fds = append(fds, rights...)
129+ }
130+ return fds, nil
131+}
132+ 
133+func validateResponse(requestID string, response Response, fds []int, expectedFDs int) error {
134+ if response.ProtocolVersion != ProtocolVersion {
135+ closeFDs(fds)
136+ return fmt.Errorf("response protocol version %d is not v%d", response.ProtocolVersion, ProtocolVersion)
137+ }
138+ if response.RequestID != requestID {
139+ closeFDs(fds)
140+ return fmt.Errorf("response request ID %q does not match %q", response.RequestID, requestID)
141+ }
142+ if len(fds) != expectedFDs {
143+ closeFDs(fds)
144+ return fmt.Errorf("response returned %d descriptors, expected %d", len(fds), expectedFDs)
145+ }
146+ return nil
147+}
148+ 
149+func closeFDs(fds []int) {
150+ for _, fd := range fds {
151+ _ = unix.Close(fd)
152+ }
153+}
@@ -0,0 +1,172 @@
1+package cow
2+ 
3+import (
4+ "errors"
5+ "fmt"
6+ "io"
7+ "os"
8+ "unsafe"
9+ 
10+ "github.com/openeuler/Conch/internal/memsnap"
11+ "golang.org/x/sys/unix"
12+)
13+ 
14+const (
15+ uffdAPI = uint64(0xaa)
16+ uffdFeaturePagefaultFlagWP = uint64(1 << 0)
17+ uffdFeatureMissingShmem = uint64(1 << 5)
18+ uffdFeatureWPHugetlbfsShmem = uint64(1 << 12)
19+ uffdFeatureWPAsync = uint64(1 << 15)
20+ uffdioAPIRequest = uintptr(0xc018aa3f)
21+)
22+ 
23+const requiredUFFDFeatures = uffdFeatureMissingShmem |
24+ uffdFeaturePagefaultFlagWP |
25+ uffdFeatureWPHugetlbfsShmem |
26+ uffdFeatureWPAsync
27+ 
28+var requiredUFFDFeatureNames = []string{
29+ "uffd.missing_shmem",
30+ "uffd.pagefault_flag_wp",
31+ "uffd.wp_hugetlbfs_shmem",
32+ "uffd.wp_async",
33+}
34+ 
35+const (
36+ pagemapFeatureName = "pagemap.read"
37+ hostPageSizeFeatureName = "host-page-size.snapshot-block-compatible"
38+)
39+ 
40+type uffdioAPI struct {
41+ API uint64
42+ Features uint64
43+ Ioctls uint64
44+}
45+ 
46+type probeOps struct {
47+ pageSize func() int
48+ userfaultfd func(flags int) (int, error)
49+ ioctlAPI func(fd int, api *uffdioAPI) error
50+ openPagemap func() (io.ReadCloser, error)
51+ closeFD func(fd int) error
52+}
53+ 
54+func productionProbeOps() probeOps {
55+ return probeOps{
56+ pageSize: os.Getpagesize,
57+ userfaultfd: func(flags int) (int, error) {
58+ fd, _, errno := unix.Syscall(unix.SYS_USERFAULTFD, uintptr(flags), 0, 0)
59+ if errno != 0 {
60+ return -1, errno
61+ }
62+ return int(fd), nil
63+ },
64+ ioctlAPI: func(fd int, api *uffdioAPI) error {
65+ _, _, errno := unix.Syscall(unix.SYS_IOCTL, uintptr(fd), uffdioAPIRequest, uintptr(unsafe.Pointer(api)))
66+ if errno != 0 {
67+ return errno
68+ }
69+ return nil
70+ },
71+ openPagemap: func() (io.ReadCloser, error) { return os.Open("/proc/self/pagemap") },
72+ closeFD: unix.Close,
73+ }
74+}
75+ 
76+func ProbeCapabilities() Capabilities {
77+ return probeCapabilities(productionProbeOps())
78+}
79+ 
80+func probeCapabilities(ops probeOps) Capabilities {
81+ fd, err := ops.userfaultfd(unix.O_CLOEXEC | unix.O_NONBLOCK)
82+ if err != nil {
83+ if errors.Is(err, unix.ENOSYS) {
84+ return unsupportedCapabilities(requiredUFFDFeatureNames)
85+ }
86+ return unknownCapabilities("create userfaultfd", err)
87+ }
88+ closed := false
89+ closeUFFD := func() error {
90+ if closed {
91+ return nil
92+ }
93+ closed = true
94+ return ops.closeFD(fd)
95+ }
96+ 
97+ api := uffdioAPI{API: uffdAPI}
98+ if err := ops.ioctlAPI(fd, &api); err != nil {
99+ closeErr := closeUFFD()
100+ if errors.Is(err, unix.ENOSYS) && closeErr == nil {
101+ return unsupportedCapabilities(requiredUFFDFeatureNames)
102+ }
103+ return unknownCapabilities("negotiate UFFDIO_API", errors.Join(err, closeErr))
104+ }
105+ if api.API != uffdAPI {
106+ return unknownCapabilities("negotiate UFFDIO_API", errors.Join(fmt.Errorf("kernel returned API %#x", api.API), closeUFFD()))
107+ }
108+ if missing := missingUFFDFeatures(api.Features); len(missing) != 0 {
109+ if err := closeUFFD(); err != nil {
110+ return unknownCapabilities("close temporary userfaultfd", err)
111+ }
112+ return unsupportedCapabilities(missing)
113+ }
114+ pageSize := ops.pageSize()
115+ if pageSize <= 0 || !compatibleHostPageSize(uint64(pageSize), memsnap.DefaultBlockSize) {
116+ if err := closeUFFD(); err != nil {
117+ return unknownCapabilities("close temporary userfaultfd", err)
118+ }
119+ return unsupportedCapabilities([]string{hostPageSizeFeatureName})
120+ }
121+ pagemap, err := ops.openPagemap()
122+ if err != nil {
123+ return unknownPagemapCapabilities("open pagemap", errors.Join(err, closeUFFD()))
124+ }
125+ var entry [8]byte
126+ _, readErr := io.ReadFull(pagemap, entry[:])
127+ closeErr := errors.Join(pagemap.Close(), closeUFFD())
128+ if readErr != nil {
129+ return unknownPagemapCapabilities("read pagemap", errors.Join(readErr, closeErr))
130+ }
131+ if closeErr != nil {
132+ return unknownCapabilities("close probe descriptors", closeErr)
133+ }
134+ return Capabilities{IncrementalMemory: CapabilitySupported}
135+}
136+ 
137+func missingUFFDFeatures(features uint64) []string {
138+ tests := []struct {
139+ bit uint64
140+ name string
141+ }{
142+ {uffdFeatureMissingShmem, "uffd.missing_shmem"},
143+ {uffdFeaturePagefaultFlagWP, "uffd.pagefault_flag_wp"},
144+ {uffdFeatureWPHugetlbfsShmem, "uffd.wp_hugetlbfs_shmem"},
145+ {uffdFeatureWPAsync, "uffd.wp_async"},
146+ }
147+ var missing []string
148+ for _, feature := range tests {
149+ if features&feature.bit == 0 {
150+ missing = append(missing, feature.name)
151+ }
152+ }
153+ return missing
154+}
155+ 
156+func unsupportedCapabilities(missing []string) Capabilities {
157+ return Capabilities{IncrementalMemory: CapabilityUnsupported, MissingFeatures: append([]string(nil), missing...)}
158+}
159+ 
160+func unknownCapabilities(operation string, err error) Capabilities {
161+ return Capabilities{IncrementalMemory: CapabilityUnknown, ProbeError: fmt.Sprintf("%s: %v", operation, err)}
162+}
163+ 
164+func unknownPagemapCapabilities(operation string, err error) Capabilities {
165+ result := unknownCapabilities(operation, err)
166+ result.MissingFeatures = []string{pagemapFeatureName}
167+ return result
168+}
169+ 
170+func compatibleHostPageSize(pageSize, blockSize uint64) bool {
171+ return pageSize != 0 && pageSize&(pageSize-1) == 0 && pageSize >= blockSize && pageSize%blockSize == 0
172+}
@@ -0,0 +1,89 @@
1+package cow
2+ 
3+import (
4+ "errors"
5+ "io"
6+ "strings"
7+ "testing"
8+ 
9+ "golang.org/x/sys/unix"
10+)
11+ 
12+func TestProbeCapabilitiesClassifiesUnsupportedAndOperationalErrors(t *testing.T) {
13+ tests := []struct {
14+ name string
15+ userfaultfd error
16+ ioctlErr error
17+ api uint64
18+ features uint64
19+ pageSize int
20+ pagemapErr error
21+ wantState string
22+ wantMissing []string
23+ wantError bool
24+ }{
25+ {name: "supported", api: uffdAPI, features: requiredUFFDFeatures, pageSize: 4096, wantState: CapabilitySupported},
26+ {name: "64K host page", api: uffdAPI, features: requiredUFFDFeatures, pageSize: 64 * 1024, wantState: CapabilitySupported},
27+ {name: "syscall absent", userfaultfd: unix.ENOSYS, wantState: CapabilityUnsupported, wantMissing: requiredUFFDFeatureNames},
28+ {name: "missing async write protect", api: uffdAPI, features: requiredUFFDFeatures &^ uffdFeatureWPAsync, pageSize: 4096, wantState: CapabilityUnsupported, wantMissing: []string{"uffd.wp_async"}},
29+ {name: "incompatible host page", api: uffdAPI, features: requiredUFFDFeatures, pageSize: 6000, wantState: CapabilityUnsupported, wantMissing: []string{hostPageSizeFeatureName}},
30+ {name: "permission", userfaultfd: unix.EPERM, wantState: CapabilityUnknown, wantError: true},
31+ {name: "malformed API", api: 0, features: requiredUFFDFeatures, pageSize: 4096, wantState: CapabilityUnknown, wantError: true},
32+ {name: "pagemap permission", api: uffdAPI, features: requiredUFFDFeatures, pageSize: 4096, pagemapErr: unix.EACCES, wantState: CapabilityUnknown, wantMissing: []string{pagemapFeatureName}, wantError: true},
33+ }
34+ for _, test := range tests {
35+ t.Run(test.name, func(t *testing.T) {
36+ closed := false
37+ ops := probeOps{
38+ pageSize: func() int { return test.pageSize },
39+ userfaultfd: func(flags int) (int, error) {
40+ if flags != unix.O_CLOEXEC|unix.O_NONBLOCK {
41+ t.Fatalf("userfaultfd flags = %#x", flags)
42+ }
43+ return 91, test.userfaultfd
44+ },
45+ ioctlAPI: func(fd int, api *uffdioAPI) error {
46+ api.API = test.api
47+ api.Features = test.features
48+ return test.ioctlErr
49+ },
50+ openPagemap: func() (io.ReadCloser, error) {
51+ if test.pagemapErr != nil {
52+ return nil, test.pagemapErr
53+ }
54+ return &probeTestReader{closed: &closed}, nil
55+ },
56+ closeFD: func(int) error { return nil },
57+ }
58+ got := probeCapabilities(ops)
59+ if got.IncrementalMemory != test.wantState {
60+ t.Fatalf("state = %q, want %q (%+v)", got.IncrementalMemory, test.wantState, got)
61+ }
62+ if strings.Join(got.MissingFeatures, ",") != strings.Join(test.wantMissing, ",") {
63+ t.Fatalf("missing = %v, want %v", got.MissingFeatures, test.wantMissing)
64+ }
65+ if (got.ProbeError != "") != test.wantError {
66+ t.Fatalf("ProbeError = %q, want present=%v", got.ProbeError, test.wantError)
67+ }
68+ })
69+ }
70+}
71+ 
72+type probeTestReader struct {
73+ closed *bool
74+}
75+ 
76+func (reader *probeTestReader) Read(buffer []byte) (int, error) {
77+ if len(buffer) != 8 {
78+ return 0, errors.New("pagemap read must be exactly one entry")
79+ }
80+ for index := range buffer {
81+ buffer[index] = 1
82+ }
83+ return len(buffer), nil
84+}
85+ 
86+func (reader *probeTestReader) Close() error {
87+ *reader.closed = true
88+ return nil
89+}
@@ -0,0 +1,109 @@
1+package cow
2+ 
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+ "os"
8+ "os/exec"
9+ "sync"
10+ "syscall"
11+ "time"
12+)
13+ 
14+const (
15+ processStartupTimeout = 5 * time.Second
16+ processReadyRetry = 10 * time.Millisecond
17+)
18+ 
19+// Process owns one conch-cow child started by conchd.
20+type Process struct {
21+ cmd *exec.Cmd
22+ waitDone chan error
23+ 
24+ closeOnce sync.Once
25+ closeErr error
26+}
27+ 
28+// StartProcess starts conch-cow and waits until its control protocol responds.
29+func StartProcess(ctx context.Context, binaryPath, socketPath string) (*Process, error) {
30+ if binaryPath == "" {
31+ return nil, fmt.Errorf("cow binary path is required")
32+ }
33+ if socketPath == "" {
34+ return nil, fmt.Errorf("cow socket path is required")
35+ }
36+ 
37+ cmd := exec.Command(binaryPath, "--socket", socketPath)
38+ cmd.Stdout = os.Stdout
39+ cmd.Stderr = os.Stderr
40+ if err := cmd.Start(); err != nil {
41+ return nil, fmt.Errorf("start conch-cow: %w", err)
42+ }
43+ process := &Process{cmd: cmd, waitDone: make(chan error, 1)}
44+ go func() {
45+ process.waitDone <- cmd.Wait()
46+ }()
47+ 
48+ startupCtx, cancel := context.WithTimeout(ctx, processStartupTimeout)
49+ defer cancel()
50+ client := NewClient(socketPath)
51+ var lastErr error
52+ for {
53+ if _, err := client.Capabilities(startupCtx); err == nil {
54+ return process, nil
55+ } else {
56+ lastErr = err
57+ }
58+ 
59+ timer := time.NewTimer(processReadyRetry)
60+ select {
61+ case waitErr := <-process.waitDone:
62+ timer.Stop()
63+ if waitErr == nil {
64+ return nil, fmt.Errorf("conch-cow exited before becoming ready")
65+ }
66+ return nil, fmt.Errorf("conch-cow exited before becoming ready: %w", waitErr)
67+ case <-startupCtx.Done():
68+ timer.Stop()
69+ cleanupErr := process.stopAndWait()
70+ return nil, errors.Join(fmt.Errorf("wait for conch-cow readiness: %w: %v", startupCtx.Err(), lastErr), cleanupErr)
71+ case <-timer.C:
72+ }
73+ }
74+}
75+ 
76+// Close asks conch-cow to stop normally and reaps the child process.
77+func (process *Process) Close() error {
78+ if process == nil {
79+ return nil
80+ }
81+ process.closeOnce.Do(func() {
82+ process.closeErr = process.stopAndWait()
83+ })
84+ return process.closeErr
85+}
86+ 
87+func (process *Process) stopAndWait() error {
88+ if process == nil || process.cmd == nil || process.cmd.Process == nil {
89+ return nil
90+ }
91+ select {
92+ case waitErr := <-process.waitDone:
93+ if waitErr != nil {
94+ return fmt.Errorf("wait for conch-cow: %w", waitErr)
95+ }
96+ return nil
97+ default:
98+ }
99+ 
100+ signalErr := process.cmd.Process.Signal(syscall.SIGTERM)
101+ if errors.Is(signalErr, os.ErrProcessDone) {
102+ signalErr = nil
103+ }
104+ waitErr := <-process.waitDone
105+ if waitErr != nil {
106+ waitErr = fmt.Errorf("wait for conch-cow: %w", waitErr)
107+ }
108+ return errors.Join(signalErr, waitErr)
109+}
@@ -0,0 +1,81 @@
1+package cow
2+ 
3+import (
4+ "context"
5+ "errors"
6+ "os"
7+ "os/signal"
8+ "path/filepath"
9+ "strings"
10+ "syscall"
11+ "testing"
12+ "time"
13+ 
14+ "golang.org/x/sys/unix"
15+)
16+ 
17+func TestMain(m *testing.M) {
18+ for index, argument := range os.Args {
19+ if argument == "--socket" && index+1 < len(os.Args) {
20+ os.Exit(runManagedCowTestProcess(os.Args[index+1]))
21+ }
22+ }
23+ os.Exit(m.Run())
24+}
25+ 
26+func runManagedCowTestProcess(socketPath string) int {
27+ ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
28+ defer stop()
29+ server := newServer(socketPath, Capabilities{IncrementalMemory: CapabilityUnsupported})
30+ if err := server.Serve(ctx); err != nil {
31+ return 1
32+ }
33+ if err := server.Close(); err != nil {
34+ return 1
35+ }
36+ return 0
37+}
38+ 
39+func TestStartProcessWaitsForCapabilitiesAndClosesChild(t *testing.T) {
40+ socketPath := filepath.Join(t.TempDir(), "cow.sock")
41+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
42+ defer cancel()
43+ process, err := StartProcess(ctx, os.Args[0], socketPath)
44+ if err != nil {
45+ t.Fatal(err)
46+ }
47+ pid := process.cmd.Process.Pid
48+ t.Cleanup(func() {
49+ if process.cmd.ProcessState == nil {
50+ _ = process.cmd.Process.Kill()
51+ _, _ = process.cmd.Process.Wait()
52+ }
53+ })
54+ 
55+ capabilities, err := NewClient(socketPath).Capabilities(context.Background())
56+ if err != nil {
57+ t.Fatal(err)
58+ }
59+ if capabilities.IncrementalMemory != CapabilityUnsupported {
60+ t.Fatalf("Capabilities() = %+v", capabilities)
61+ }
62+ if err := process.Close(); err != nil {
63+ t.Fatal(err)
64+ }
65+ if process.cmd.ProcessState == nil || !process.cmd.ProcessState.Exited() {
66+ t.Fatalf("cow process state = %#v, want exited", process.cmd.ProcessState)
67+ }
68+ if err := unix.Kill(pid, 0); !errors.Is(err, unix.ESRCH) {
69+ t.Fatalf("cow pid %d remains after Close: %v", pid, err)
70+ }
71+}
72+ 
73+func TestStartProcessFailsWhenChildExitsBeforeReady(t *testing.T) {
74+ socketPath := filepath.Join(t.TempDir(), "cow.sock")
75+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
76+ defer cancel()
77+ process, err := StartProcess(ctx, "/bin/false", socketPath)
78+ if process != nil || err == nil || !strings.Contains(err.Error(), "before becoming ready") {
79+ t.Fatalf("StartProcess() process=%v error=%v", process, err)
80+ }
81+}
@@ -0,0 +1,79 @@
1+package cow
2+ 
3+import "fmt"
4+ 
5+const (
6+ ProtocolVersion = 1
7+ DefaultSocketPath = "/run/conch/cow.sock"
8+)
9+ 
10+const (
11+ RequestCapabilities = "Capabilities"
12+ RequestAttach = "Attach"
13+ RequestWaitAttachmentReady = "WaitAttachmentReady"
14+ RequestDetach = "Detach"
15+)
16+ 
17+const (
Z
Zzhaoyifan2825 天前

Capability探测并没有必须在conch-cow做的理由,建议放到conchd来做降低复杂度

likedislike
18+ CapabilitySupported = "supported"
19+ CapabilityUnsupported = "unsupported"
20+ CapabilityUnknown = "unknown"
Z
Zzhaoyifan2825 天前

同上,unknown不用单独区分,可以直接和unsupported一起导致报错退出

likedislike
21+)
22+ 
23+type Request struct {
Z
Zzhaoyifan2825 天前

现在Request和Response看起来就是所有接口的输入/输出参数并集,应该按照不同接口做区分

likedislike
24+ Type string `json:"type"`
25+ ProtocolVersion int `json:"protocol_version"`
26+ RequestID string `json:"request_id"`
27+ MemorySnapshotRoot string `json:"memory_snapshot_root,omitempty"`
28+ Token string `json:"token,omitempty"`
29+ SandboxID string `json:"sandbox_id,omitempty"`
30+}
31+ 
32+type Capabilities struct {
33+ IncrementalMemory string `json:"incremental_memory"`
34+ MissingFeatures []string `json:"missing_features,omitempty"`
35+ ProbeError string `json:"probe_error,omitempty"`
36+}
37+ 
38+type Response struct {
39+ OK bool `json:"ok"`
40+ ProtocolVersion int `json:"protocol_version"`
41+ RequestID string `json:"request_id"`
42+ Error string `json:"error,omitempty"`
43+ Capabilities *Capabilities `json:"capabilities,omitempty"`
44+ Token string `json:"token,omitempty"`
45+ UFFDSocketPath string `json:"uffd_socket_path,omitempty"`
46+ MemorySize uint64 `json:"memory_size,omitempty"`
47+ BlockSize uint64 `json:"block_size,omitempty"`
48+}
49+ 
50+func validateRequest(request Request, fds []int) error {
51+ if request.ProtocolVersion != ProtocolVersion {
52+ return fmt.Errorf("unsupported protocol version %d", request.ProtocolVersion)
53+ }
54+ if request.RequestID == "" {
55+ return fmt.Errorf("request ID is required")
56+ }
57+ expectedFDs := 0
58+ switch request.Type {
59+ case RequestCapabilities:
60+ case RequestAttach:
61+ if request.MemorySnapshotRoot == "" {
62+ return fmt.Errorf("memory snapshot root is required")
63+ }
64+ case RequestWaitAttachmentReady:
65+ if request.Token == "" || request.SandboxID == "" {
66+ return fmt.Errorf("token and sandbox ID are required")
67+ }
68+ case RequestDetach:
69+ if request.Token == "" {
70+ return fmt.Errorf("token is required")
71+ }
72+ default:
73+ return fmt.Errorf("unknown request type %q", request.Type)
74+ }
75+ if len(fds) != expectedFDs {
76+ return fmt.Errorf("%s request has %d descriptors, expected %d", request.Type, len(fds), expectedFDs)
77+ }
78+ return nil
79+}
@@ -0,0 +1,217 @@
1+package cow
2+ 
3+import (
4+ "context"
5+ "encoding/binary"
6+ "net"
7+ "os"
8+ "path/filepath"
9+ "strings"
10+ "testing"
11+ "time"
12+ 
13+ "golang.org/x/sys/unix"
14+)
15+ 
16+func TestFrameReadsPartialHeaderAndPayload(t *testing.T) {
17+ reader, writer := testUnixSocketPair(t)
18+ want := Request{Type: RequestCapabilities, ProtocolVersion: ProtocolVersion, RequestID: "partial"}
19+ payload := []byte(`{"type":"Capabilities","protocol_version":1,"request_id":"partial"}`)
20+ header := make([]byte, 4)
21+ binary.BigEndian.PutUint32(header, uint32(len(payload)))
22+ done := make(chan error, 1)
23+ go func() {
24+ for _, piece := range [][]byte{header[:2], header[2:], payload[:7], payload[7:]} {
25+ if _, err := writer.Write(piece); err != nil {
26+ done <- err
27+ return
28+ }
29+ }
30+ done <- nil
31+ }()
32+ var got Request
33+ fds, err := readFrame(reader, &got)
34+ if err != nil {
35+ t.Fatal(err)
36+ }
37+ defer closeFDs(fds)
38+ if err := <-done; err != nil {
39+ t.Fatal(err)
40+ }
41+ if got != want {
42+ t.Fatalf("request = %#v, want %#v", got, want)
43+ }
44+}
45+ 
46+func TestFrameRejectsOversizeAndTrailingJSON(t *testing.T) {
47+ t.Run("oversize", func(t *testing.T) {
48+ reader, writer := testUnixSocketPair(t)
49+ header := make([]byte, 4)
50+ binary.BigEndian.PutUint32(header, maxFrameSize+1)
51+ if _, err := writer.Write(header); err != nil {
52+ t.Fatal(err)
53+ }
54+ var request Request
55+ if _, err := readFrame(reader, &request); err == nil || !strings.Contains(err.Error(), "too large") {
56+ t.Fatalf("readFrame() error = %v, want too large", err)
57+ }
58+ })
59+ t.Run("trailing JSON", func(t *testing.T) {
60+ reader, writer := testUnixSocketPair(t)
61+ payload := []byte(`{"type":"Capabilities","protocol_version":1,"request_id":"first"}{}`)
62+ header := make([]byte, 4)
63+ binary.BigEndian.PutUint32(header, uint32(len(payload)))
64+ if _, err := writer.Write(append(header, payload...)); err != nil {
65+ t.Fatal(err)
66+ }
67+ var request Request
68+ if fds, err := readFrame(reader, &request); err == nil {
69+ closeFDs(fds)
70+ t.Fatal("readFrame accepted multiple JSON values")
71+ }
72+ })
73+}
74+ 
75+func TestValidateRequestRequiresOperationFieldsAndFDs(t *testing.T) {
76+ valid := []struct {
77+ request Request
78+ fds int
79+ }{
80+ {request: Request{Type: RequestCapabilities, ProtocolVersion: ProtocolVersion, RequestID: "cap"}},
81+ {request: Request{Type: RequestAttach, ProtocolVersion: ProtocolVersion, RequestID: "attach", MemorySnapshotRoot: "/memory"}},
82+ {request: Request{Type: RequestWaitAttachmentReady, ProtocolVersion: ProtocolVersion, RequestID: "wait", Token: "token", SandboxID: "sandbox"}},
83+ {request: Request{Type: RequestDetach, ProtocolVersion: ProtocolVersion, RequestID: "detach", Token: "token"}},
84+ }
85+ for _, test := range valid {
86+ if err := validateRequest(test.request, make([]int, test.fds)); err != nil {
87+ t.Fatalf("validateRequest(%s): %v", test.request.Type, err)
88+ }
89+ }
90+ invalid := []struct {
91+ name string
92+ request Request
93+ fds int
94+ }{
95+ {name: "old version", request: Request{Type: RequestCapabilities, RequestID: "id"}},
96+ {name: "future version", request: Request{Type: RequestCapabilities, ProtocolVersion: 2, RequestID: "id"}},
97+ {name: "unknown operation", request: Request{Type: "Future", ProtocolVersion: ProtocolVersion, RequestID: "id"}},
98+ {name: "attach root", request: Request{Type: RequestAttach, ProtocolVersion: ProtocolVersion, RequestID: "id"}},
99+ {name: "wait sandbox", request: Request{Type: RequestWaitAttachmentReady, ProtocolVersion: ProtocolVersion, RequestID: "id", Token: "token"}},
100+ {name: "wait token", request: Request{Type: RequestWaitAttachmentReady, ProtocolVersion: ProtocolVersion, RequestID: "id", SandboxID: "sandbox"}},
101+ {name: "wait descriptor", request: Request{Type: RequestWaitAttachmentReady, ProtocolVersion: ProtocolVersion, RequestID: "id", Token: "token", SandboxID: "sandbox"}, fds: 1},
102+ {name: "detach token", request: Request{Type: RequestDetach, ProtocolVersion: ProtocolVersion, RequestID: "id"}},
103+ }
104+ for _, test := range invalid {
105+ t.Run(test.name, func(t *testing.T) {
106+ if err := validateRequest(test.request, make([]int, test.fds)); err == nil {
107+ t.Fatal("validateRequest accepted invalid request")
108+ }
109+ })
110+ }
111+}
112+ 
113+func TestValidateResponseClosesUnexpectedFDs(t *testing.T) {
114+ fds := testPipeFDs(t, 2)
115+ response := Response{OK: true, ProtocolVersion: ProtocolVersion, RequestID: "request"}
116+ if err := validateResponse("request", response, fds, 0); err == nil {
117+ t.Fatal("validateResponse accepted unexpected descriptors")
118+ }
119+ for _, fd := range fds {
120+ if _, err := unix.FcntlInt(uintptr(fd), unix.F_GETFD, 0); err == nil {
121+ t.Fatalf("descriptor %d remained open", fd)
122+ }
123+ }
124+}
125+ 
126+func TestClientWaitAttachmentReadySendsNoDescriptors(t *testing.T) {
127+ socketPath := filepath.Join(t.TempDir(), "cow.sock")
128+ listener, err := net.ListenUnix("unix", &net.UnixAddr{Name: socketPath, Net: "unix"})
129+ if err != nil {
130+ t.Fatal(err)
131+ }
132+ t.Cleanup(func() { _ = listener.Close() })
133+ serverDone := make(chan error, 1)
134+ go func() {
135+ conn, err := listener.AcceptUnix()
136+ if err != nil {
137+ serverDone <- err
138+ return
139+ }
140+ defer conn.Close()
141+ var request Request
142+ fds, err := readFrame(conn, &request)
143+ if err != nil {
144+ serverDone <- err
145+ return
146+ }
147+ defer closeFDs(fds)
148+ if err := validateRequest(request, fds); err != nil {
149+ serverDone <- err
150+ return
151+ }
152+ if request.Type != RequestWaitAttachmentReady || request.Token != "token" || request.SandboxID != "sandbox" || len(fds) != 0 {
153+ serverDone <- &testError{"unexpected WaitAttachmentReady request"}
154+ return
155+ }
156+ serverDone <- writeFrame(conn, Response{OK: true, ProtocolVersion: ProtocolVersion, RequestID: request.RequestID}, nil)
157+ }()
158+ 
159+ client := NewClient(socketPath)
160+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
161+ defer cancel()
162+ if _, err := client.WaitAttachmentReady(ctx, "token", "sandbox"); err != nil {
163+ t.Fatal(err)
164+ }
165+ if err := <-serverDone; err != nil {
166+ t.Fatal(err)
167+ }
168+}
169+ 
170+type testError struct{ message string }
171+ 
172+func (err *testError) Error() string { return err.message }
173+ 
174+func testUnixSocketPair(t *testing.T) (*net.UnixConn, *net.UnixConn) {
175+ t.Helper()
176+ fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_CLOEXEC, 0)
177+ if err != nil {
178+ t.Fatal(err)
179+ }
180+ leftFile := os.NewFile(uintptr(fds[0]), "left")
181+ rightFile := os.NewFile(uintptr(fds[1]), "right")
182+ leftConnection, err := net.FileConn(leftFile)
183+ _ = leftFile.Close()
184+ if err != nil {
185+ _ = rightFile.Close()
186+ t.Fatal(err)
187+ }
188+ rightConnection, err := net.FileConn(rightFile)
189+ _ = rightFile.Close()
190+ if err != nil {
191+ _ = leftConnection.Close()
192+ t.Fatal(err)
193+ }
194+ left := leftConnection.(*net.UnixConn)
195+ right := rightConnection.(*net.UnixConn)
196+ t.Cleanup(func() { _ = left.Close(); _ = right.Close() })
197+ return left, right
198+}
199+ 
200+func testPipeFDs(t *testing.T, count int) []int {
201+ t.Helper()
202+ fds := make([]int, 0, count)
203+ for range count {
204+ readEnd, writeEnd, err := os.Pipe()
205+ if err != nil {
206+ t.Fatal(err)
207+ }
208+ _ = writeEnd.Close()
209+ fd, err := unix.Dup(int(readEnd.Fd()))
210+ _ = readEnd.Close()
211+ if err != nil {
212+ t.Fatal(err)
213+ }
214+ fds = append(fds, fd)
215+ }
216+ return fds
217+}
@@ -0,0 +1,486 @@
1+package cow
2+ 
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+ "net"
8+ "os"
9+ "path/filepath"
10+ "sync"
11+ "time"
12+ 
13+ "github.com/google/uuid"
14+ "github.com/openeuler/Conch/internal/memsnap"
15+ "golang.org/x/sys/unix"
16+)
17+ 
18+type attachment struct {
19+ token string
20+ sandboxID string
21+ pinned *memsnap.PinnedManifest
22+ memfd *os.File
23+ 
24+ uffdSocketPath string
25+ uffdListener *net.UnixListener
26+ uffd *os.File
27+ uffdRanges []uffdRange
28+ handoffPrepared bool
29+ handoffDone chan struct{}
30+ handoffErr error
31+ workerStop chan struct{}
32+ workerDone chan struct{}
33+ closeOnce sync.Once
34+ closeErr error
35+}
36+ 
37+type Server struct {
38+ socketPath string
39+ capabilities Capabilities
40+ 
41+ mu sync.Mutex
42+ attachments map[string]*attachment
43+ listener *net.UnixListener
44+ closed bool
45+ connections sync.WaitGroup
46+ activeConnections map[*net.UnixConn]struct{}
47+ uffdOps uffdOperations
48+ uffdSequence uint64
49+ controlSocketIdentity unixSocketIdentity
50+ controlSocketOwned bool
51+ ready chan struct{}
52+ readyOnce sync.Once
53+}
54+ 
55+type unixSocketIdentity struct {
56+ device uint64
57+ inode uint64
58+}
59+ 
60+func NewServer(socketPath string) *Server {
61+ if socketPath == "" {
62+ socketPath = DefaultSocketPath
63+ }
64+ return newServer(socketPath, ProbeCapabilities())
65+}
66+ 
67+func newServer(socketPath string, capabilities Capabilities) *Server {
68+ return &Server{
69+ socketPath: socketPath,
70+ capabilities: capabilities,
71+ attachments: make(map[string]*attachment),
72+ activeConnections: make(map[*net.UnixConn]struct{}),
73+ uffdOps: productionUFFDOperations(),
74+ ready: make(chan struct{}),
75+ }
76+}
77+ 
78+func (server *Server) Ready() <-chan struct{} {
79+ return server.ready
80+}
81+ 
82+func baseResponse(request Request) Response {
83+ return Response{ProtocolVersion: ProtocolVersion, RequestID: request.RequestID}
84+}
85+ 
86+func (server *Server) Serve(ctx context.Context) error {
87+ if server.socketPath == "" {
88+ return fmt.Errorf("socket path is required")
89+ }
90+ if err := os.MkdirAll(filepath.Dir(server.socketPath), 0o755); err != nil {
91+ return fmt.Errorf("create cow socket directory: %w", err)
92+ }
93+ if err := prepareControlSocketPath(server.socketPath); err != nil {
94+ return err
95+ }
96+ listener, err := net.ListenUnix("unix", &net.UnixAddr{Name: server.socketPath, Net: "unix"})
97+ if err != nil {
98+ return fmt.Errorf("listen cow socket: %w", err)
99+ }
100+ listener.SetUnlinkOnClose(false)
101+ identity, err := controlSocketIdentity(server.socketPath)
102+ if err != nil {
103+ _ = listener.Close()
104+ return fmt.Errorf("identify cow socket: %w", err)
105+ }
106+ if err := os.Chmod(server.socketPath, 0o600); err != nil {
107+ _ = listener.Close()
108+ _ = removeControlSocketIfOwned(server.socketPath, identity)
109+ return fmt.Errorf("chmod cow socket: %w", err)
110+ }
111+ server.mu.Lock()
112+ if server.closed {
113+ server.mu.Unlock()
114+ _ = listener.Close()
115+ _ = removeControlSocketIfOwned(server.socketPath, identity)
116+ return fmt.Errorf("server is closed")
117+ }
118+ server.listener = listener
119+ server.controlSocketIdentity = identity
120+ server.controlSocketOwned = true
121+ server.mu.Unlock()
122+ server.readyOnce.Do(func() { close(server.ready) })
123+ 
124+ stop := make(chan struct{})
125+ go func() {
126+ select {
127+ case <-ctx.Done():
128+ _ = listener.Close()
129+ case <-stop:
130+ }
131+ }()
132+ defer close(stop)
133+ defer server.removeOwnedControlSocket()
134+ for {
135+ conn, err := listener.AcceptUnix()
136+ if err != nil {
137+ if ctx.Err() != nil || server.isClosed() || errors.Is(err, net.ErrClosed) {
138+ return nil
139+ }
140+ return fmt.Errorf("accept cow connection: %w", err)
141+ }
142+ server.mu.Lock()
143+ if server.closed {
144+ server.mu.Unlock()
145+ _ = conn.Close()
146+ continue
147+ }
148+ server.connections.Add(1)
149+ server.activeConnections[conn] = struct{}{}
150+ server.mu.Unlock()
151+ go func(conn *net.UnixConn) {
152+ defer server.connections.Done()
153+ defer func() {
154+ server.mu.Lock()
155+ delete(server.activeConnections, conn)
156+ server.mu.Unlock()
157+ }()
158+ server.serveConnection(conn)
159+ }(conn)
160+ }
161+}
162+ 
163+func (server *Server) serveConnection(conn *net.UnixConn) {
164+ defer conn.Close()
165+ _ = conn.SetDeadline(time.Now().Add(requestTimeout))
166+ var request Request
167+ received, err := readFrame(conn, &request)
168+ if err != nil {
169+ return
170+ }
171+ response := baseResponse(request)
172+ if err := validateRequest(request, received); err != nil {
173+ closeFDs(received)
174+ response.Error = err.Error()
175+ _ = writeFrame(conn, response, nil)
176+ return
177+ }
178+ if request.Type == RequestWaitAttachmentReady {
179+ _ = conn.SetDeadline(time.Now().Add(server.uffdOps.acceptTimeout + waitReadyTimeoutMargin))
180+ }
181+ var sendFDs []int
182+ switch request.Type {
183+ case RequestCapabilities:
184+ response.OK = true
185+ capabilities := server.capabilities
186+ capabilities.MissingFeatures = append([]string(nil), capabilities.MissingFeatures...)
187+ response.Capabilities = &capabilities
188+ case RequestAttach:
189+ response, sendFDs = server.handleAttach(request, response)
190+ case RequestWaitAttachmentReady:
191+ response = server.handleWaitAttachmentReady(request, response)
192+ case RequestDetach:
193+ response = server.handleDetach(request, response)
194+ }
195+ closeFDs(received)
196+ defer closeFDs(sendFDs)
197+ if err := writeFrame(conn, response, sendFDs); err != nil && request.Type == RequestAttach && response.OK {
198+ server.releaseAttachment(response.Token)
199+ }
200+}
201+ 
202+func (server *Server) handleAttach(request Request, response Response) (Response, []int) {
203+ if request.SandboxID == "" || request.MemorySnapshotRoot == "" {
204+ response.Error = "sandbox ID and memory snapshot root are required"
205+ return response, nil
206+ }
207+ if server.capabilities.IncrementalMemory != CapabilitySupported {
208+ response.Error = fmt.Sprintf("incremental memory is %s", server.capabilities.IncrementalMemory)
209+ return response, nil
210+ }
211+ pinned, err := memsnap.LoadAndPin(request.MemorySnapshotRoot)
212+ if err != nil {
213+ response.Error = fmt.Sprintf("load and pin manifest: %v", err)
214+ return response, nil
215+ }
216+ token := uuid.NewString()
217+ fd, err := unix.MemfdCreate("conch-memory-"+token, unix.MFD_CLOEXEC|unix.MFD_ALLOW_SEALING)
218+ if err != nil {
219+ _ = pinned.Close()
220+ response.Error = fmt.Sprintf("create memory memfd: %v", err)
221+ return response, nil
222+ }
223+ owner := os.NewFile(uintptr(fd), "conch-memory-owner")
224+ if owner == nil {
225+ _ = unix.Close(fd)
226+ _ = pinned.Close()
227+ response.Error = "wrap memory memfd"
228+ return response, nil
229+ }
230+ cleanup := func() {
231+ _ = owner.Close()
232+ _ = pinned.Close()
233+ }
234+ if pinned.Manifest.MemorySize > uint64(^uint64(0)>>1) {
235+ cleanup()
236+ response.Error = "memory size exceeds supported file size"
237+ return response, nil
238+ }
239+ if err := unix.Ftruncate(fd, int64(pinned.Manifest.MemorySize)); err != nil {
240+ cleanup()
241+ response.Error = fmt.Sprintf("size memory memfd: %v", err)
242+ return response, nil
243+ }
244+ if _, err := unix.FcntlInt(uintptr(fd), unix.F_ADD_SEALS, unix.F_SEAL_GROW|unix.F_SEAL_SHRINK); err != nil {
245+ cleanup()
246+ response.Error = fmt.Sprintf("seal memory memfd: %v", err)
247+ return response, nil
248+ }
249+ duplicate, err := unix.FcntlInt(uintptr(fd), unix.F_DUPFD_CLOEXEC, 0)
250+ if err != nil {
251+ cleanup()
252+ response.Error = fmt.Sprintf("duplicate memory memfd: %v", err)
253+ return response, nil
254+ }
255+ item := &attachment{token: token, sandboxID: request.SandboxID, pinned: pinned, memfd: owner}
256+ server.mu.Lock()
257+ if server.closed {
258+ server.mu.Unlock()
259+ _ = unix.Close(duplicate)
260+ cleanup()
261+ response.Error = "server is closed"
262+ return response, nil
263+ }
264+ server.attachments[token] = item
265+ server.mu.Unlock()
266+ prepared := server.prepareUFFDHandoff(item, response)
267+ if !prepared.OK {
268+ _ = unix.Close(duplicate)
269+ server.releaseAttachment(token)
270+ return prepared, nil
271+ }
272+ prepared.Token = token
273+ prepared.MemorySize = pinned.Manifest.MemorySize
274+ prepared.BlockSize = pinned.Manifest.BlockSize
275+ return prepared, []int{duplicate}
276+}
277+ 
278+func (server *Server) handleDetach(request Request, response Response) Response {
279+ server.releaseAttachment(request.Token)
280+ response.OK = true
281+ return response
282+}
283+ 
284+func (server *Server) releaseAttachment(token string) bool {
285+ server.mu.Lock()
286+ item := server.attachments[token]
287+ if item != nil {
288+ delete(server.attachments, token)
289+ }
290+ server.mu.Unlock()
291+ if item == nil {
292+ return false
293+ }
294+ _ = server.closeAttachment(item)
295+ return true
296+}
297+ 
298+func (server *Server) handleWaitAttachmentReady(request Request, response Response) Response {
299+ server.mu.Lock()
300+ if server.closed {
301+ server.mu.Unlock()
302+ response.Error = "server is closed"
303+ return response
304+ }
305+ item := server.attachments[request.Token]
306+ if item == nil || !item.handoffPrepared || item.sandboxID != request.SandboxID {
307+ server.mu.Unlock()
308+ response.Error = "UFFD handoff is not prepared for sandbox"
309+ return response
310+ }
311+ handoffDone := item.handoffDone
312+ server.mu.Unlock()
313+ if handoffDone != nil {
314+ timer := time.NewTimer(server.uffdOps.acceptTimeout + time.Second)
315+ defer timer.Stop()
316+ select {
317+ case <-handoffDone:
318+ case <-timer.C:
319+ response.Error = "timed out waiting for UFFD handoff"
320+ return response
321+ }
322+ }
323+ server.mu.Lock()
324+ defer server.mu.Unlock()
325+ if server.closed || server.attachments[request.Token] != item {
326+ response.Error = "attachment changed while waiting for UFFD"
327+ return response
328+ }
329+ if item.handoffErr != nil {
330+ response.Error = fmt.Sprintf("UFFD handoff failed: %v", item.handoffErr)
331+ return response
332+ }
333+ if item.uffd == nil {
334+ response.Error = "UFFD handoff did not provide a fault descriptor"
335+ return response
336+ }
337+ response.OK = true
338+ return response
339+}
340+ 
341+func (server *Server) closeAttachment(item *attachment) error {
342+ if item == nil {
343+ return nil
344+ }
345+ item.closeOnce.Do(func() {
346+ var result error
347+ server.mu.Lock()
348+ listener := item.uffdListener
349+ handoffDone := item.handoffDone
350+ socketPath := item.uffdSocketPath
351+ server.mu.Unlock()
352+ if listener != nil {
353+ _ = listener.Close()
354+ }
355+ if socketPath != "" {
356+ if err := os.Remove(socketPath); err != nil && !errors.Is(err, os.ErrNotExist) {
357+ result = errors.Join(result, err)
358+ }
359+ }
360+ if handoffDone != nil {
361+ <-handoffDone
362+ }
363+ server.mu.Lock()
364+ workerStop := item.workerStop
365+ workerDone := item.workerDone
366+ uffd := item.uffd
367+ server.mu.Unlock()
368+ if workerStop != nil {
369+ close(workerStop)
370+ }
371+ if workerDone != nil {
372+ <-workerDone
373+ } else if uffd != nil {
374+ result = errors.Join(result, uffd.Close())
375+ }
376+ if item.memfd != nil {
377+ result = errors.Join(result, item.memfd.Close())
378+ }
379+ if item.pinned != nil {
380+ result = errors.Join(result, item.pinned.Close())
381+ }
382+ item.closeErr = result
383+ })
384+ return item.closeErr
385+}
386+ 
387+func (server *Server) Close() error {
388+ server.mu.Lock()
389+ firstClose := !server.closed
390+ server.closed = true
391+ listener := server.listener
392+ var handoffListeners []*net.UnixListener
393+ var activeConnections []*net.UnixConn
394+ for _, item := range server.attachments {
395+ if item.uffdListener != nil {
396+ handoffListeners = append(handoffListeners, item.uffdListener)
397+ }
398+ }
399+ for conn := range server.activeConnections {
400+ activeConnections = append(activeConnections, conn)
401+ }
402+ server.mu.Unlock()
403+ if firstClose && listener != nil {
404+ _ = listener.Close()
405+ }
406+ for _, handoffListener := range handoffListeners {
407+ _ = handoffListener.Close()
408+ }
409+ for _, conn := range activeConnections {
410+ _ = conn.Close()
411+ }
412+ server.connections.Wait()
413+ server.mu.Lock()
414+ attachments := make(map[string]*attachment, len(server.attachments))
415+ for token, item := range server.attachments {
416+ attachments[token] = item
417+ }
418+ server.mu.Unlock()
419+ var result error
420+ for token, item := range attachments {
421+ result = errors.Join(result, server.closeAttachment(item))
422+ server.mu.Lock()
423+ if server.attachments[token] == item {
424+ delete(server.attachments, token)
425+ }
426+ server.mu.Unlock()
427+ }
428+ return errors.Join(result, server.removeOwnedControlSocket())
429+}
430+ 
431+func controlSocketIdentity(path string) (unixSocketIdentity, error) {
432+ var stat unix.Stat_t
433+ if err := unix.Lstat(path, &stat); err != nil {
434+ return unixSocketIdentity{}, err
435+ }
436+ if stat.Mode&unix.S_IFMT != unix.S_IFSOCK {
437+ return unixSocketIdentity{}, fmt.Errorf("path is not a Unix socket")
438+ }
439+ return unixSocketIdentity{device: uint64(stat.Dev), inode: stat.Ino}, nil
440+}
441+ 
442+func prepareControlSocketPath(path string) error {
443+ identity, err := controlSocketIdentity(path)
444+ if errors.Is(err, unix.ENOENT) {
445+ return nil
446+ }
447+ if err != nil {
448+ return fmt.Errorf("control socket path %q already exists", path)
449+ }
450+ connection, dialErr := net.DialTimeout("unix", path, 100*time.Millisecond)
451+ if dialErr == nil {
452+ _ = connection.Close()
453+ return fmt.Errorf("control socket path %q already exists", path)
454+ }
455+ if !errors.Is(dialErr, unix.ECONNREFUSED) && !errors.Is(dialErr, os.ErrNotExist) {
456+ return fmt.Errorf("control socket path %q already exists: %w", path, dialErr)
457+ }
458+ return removeControlSocketIfOwned(path, identity)
459+}
460+ 
461+func removeControlSocketIfOwned(path string, expected unixSocketIdentity) error {
Z
Zzhaoyifan2824 天前

这里为了处理socket被其他进程占用的边界情况引入过多复杂度,建议删除

likedislike
462+ actual, err := controlSocketIdentity(path)
463+ if errors.Is(err, unix.ENOENT) || err != nil || actual != expected {
464+ return nil
465+ }
466+ return unix.Unlink(path)
467+}
468+ 
469+func (server *Server) removeOwnedControlSocket() error {
470+ server.mu.Lock()
471+ if !server.controlSocketOwned {
472+ server.mu.Unlock()
473+ return nil
474+ }
475+ server.controlSocketOwned = false
476+ identity := server.controlSocketIdentity
477+ path := server.socketPath
478+ server.mu.Unlock()
479+ return removeControlSocketIfOwned(path, identity)
480+}
481+ 
482+func (server *Server) isClosed() bool {
483+ server.mu.Lock()
484+ defer server.mu.Unlock()
485+ return server.closed
486+}
@@ -0,0 +1,235 @@
1+package cow
2+ 
3+import (
4+ "context"
5+ "encoding/json"
6+ "errors"
7+ "net"
8+ "os"
9+ "path/filepath"
10+ "strings"
11+ "testing"
12+ "time"
13+ 
14+ "github.com/openeuler/Conch/internal/memsnap"
15+ "golang.org/x/sys/unix"
16+)
17+ 
18+func TestServerCloseReleasesPartialControlFrame(t *testing.T) {
19+ socketPath := filepath.Join(t.TempDir(), "cow.sock")
20+ server := newServer(socketPath, Capabilities{IncrementalMemory: CapabilitySupported})
21+ ctx, cancel := context.WithCancel(context.Background())
22+ defer cancel()
23+ serveDone := make(chan error, 1)
24+ go func() { serveDone <- server.Serve(ctx) }()
25+ select {
26+ case <-server.Ready():
27+ case <-time.After(time.Second):
28+ t.Fatal("server did not become ready")
29+ }
30+ conn, err := net.DialUnix("unix", nil, &net.UnixAddr{Name: socketPath, Net: "unix"})
31+ if err != nil {
32+ t.Fatal(err)
33+ }
34+ defer conn.Close()
35+ if _, err := conn.Write([]byte{0, 0}); err != nil {
36+ t.Fatal(err)
37+ }
38+ closeDone := make(chan error, 1)
39+ go func() { closeDone <- server.Close() }()
40+ select {
41+ case err := <-closeDone:
42+ if err != nil {
43+ t.Fatal(err)
44+ }
45+ case <-time.After(500 * time.Millisecond):
46+ _ = conn.Close()
47+ <-closeDone
48+ t.Fatal("Server.Close blocked on a partial control frame")
49+ }
50+ if err := <-serveDone; err != nil {
51+ t.Fatal(err)
52+ }
53+}
54+ 
55+func TestServerSignalsReadyAndReportsCapabilities(t *testing.T) {
56+ socketPath := filepath.Join(t.TempDir(), "cow.sock")
57+ server := newServer(socketPath, Capabilities{IncrementalMemory: CapabilityUnsupported, MissingFeatures: []string{"uffd.wp_async"}})
58+ ctx, cancel := context.WithCancel(context.Background())
59+ done := make(chan error, 1)
60+ go func() { done <- server.Serve(ctx) }()
61+ select {
62+ case <-server.Ready():
63+ case <-time.After(time.Second):
64+ t.Fatal("server did not become ready")
65+ }
66+ capabilities, err := NewClient(socketPath).Capabilities(context.Background())
67+ if err != nil {
68+ t.Fatal(err)
69+ }
70+ if capabilities.IncrementalMemory != CapabilityUnsupported || len(capabilities.MissingFeatures) != 1 {
71+ t.Fatalf("Capabilities() = %+v", capabilities)
72+ }
73+ cancel()
74+ if err := <-done; err != nil {
75+ t.Fatal(err)
76+ }
77+ if err := server.Close(); err != nil {
78+ t.Fatal(err)
79+ }
80+ if _, err := os.Lstat(socketPath); !errors.Is(err, os.ErrNotExist) {
81+ t.Fatalf("control socket remains after close: %v", err)
82+ }
83+}
84+ 
85+func TestClientCapabilitiesRejectsInvalidIncrementalMemoryState(t *testing.T) {
86+ socketPath := filepath.Join(t.TempDir(), "cow.sock")
87+ server := newServer(socketPath, Capabilities{IncrementalMemory: "invalid"})
88+ ctx, cancel := context.WithCancel(context.Background())
89+ done := make(chan error, 1)
90+ go func() { done <- server.Serve(ctx) }()
91+ select {
92+ case <-server.Ready():
93+ case <-time.After(time.Second):
94+ t.Fatal("server did not become ready")
95+ }
96+ 
97+ _, err := NewClient(socketPath).Capabilities(context.Background())
98+ if err == nil || !strings.Contains(err.Error(), "invalid incremental memory state") {
99+ t.Fatalf("Capabilities() error = %v", err)
100+ }
101+ 
102+ cancel()
103+ if err := <-done; err != nil {
104+ t.Fatal(err)
105+ }
106+ if err := server.Close(); err != nil {
107+ t.Fatal(err)
108+ }
109+}
110+ 
111+func TestAttachCreatesIndependentSealedMemfdsAndDetachIsIdempotent(t *testing.T) {
112+ root := writeServerManifest(t)
113+ server := newServer(filepath.Join(t.TempDir(), "cow.sock"), Capabilities{IncrementalMemory: CapabilitySupported})
114+ server.uffdOps.acceptTimeout = time.Second
115+ t.Cleanup(func() { _ = server.Close() })
116+ 
117+ attach := func(sandboxID string) (Response, []int) {
118+ request := Request{RequestID: sandboxID, SandboxID: sandboxID, MemorySnapshotRoot: root}
119+ response, fds := server.handleAttach(request, baseResponse(request))
120+ if !response.OK || len(fds) != 1 || response.UFFDSocketPath == "" {
121+ closeFDs(fds)
122+ t.Fatalf("Attach(%s) = %+v, fds=%v", sandboxID, response, fds)
123+ }
124+ return response, fds
125+ }
126+ first, firstFDs := attach("vm-a")
127+ second, secondFDs := attach("vm-b")
128+ t.Cleanup(func() { closeFDs(firstFDs); closeFDs(secondFDs) })
129+ if first.Token == second.Token {
130+ t.Fatal("independent attaches share a token")
131+ }
132+ for _, fd := range []int{firstFDs[0], secondFDs[0]} {
133+ seals, err := unix.FcntlInt(uintptr(fd), unix.F_GET_SEALS, 0)
134+ if err != nil {
135+ t.Fatal(err)
136+ }
137+ want := unix.F_SEAL_GROW | unix.F_SEAL_SHRINK
138+ if seals&want != want {
139+ t.Fatalf("memfd seals = %#x, want %#x", seals, want)
140+ }
141+ }
142+ for range 2 {
143+ response := server.handleDetach(Request{Token: first.Token}, baseResponse(Request{}))
144+ if !response.OK {
145+ t.Fatalf("Detach() = %+v", response)
146+ }
147+ }
148+ server.mu.Lock()
149+ _, firstExists := server.attachments[first.Token]
150+ _, secondExists := server.attachments[second.Token]
151+ server.mu.Unlock()
152+ if firstExists || !secondExists {
153+ t.Fatalf("Detach affected wrong attachment: first=%v second=%v", firstExists, secondExists)
154+ }
155+}
156+ 
157+func TestWaitAttachmentReadyMatchesSandboxAndCompletedHandoff(t *testing.T) {
158+ server := newServer(filepath.Join(t.TempDir(), "cow.sock"), Capabilities{IncrementalMemory: CapabilitySupported})
159+ handoffDone := make(chan struct{})
160+ close(handoffDone)
161+ uffdRead, uffdWrite, err := os.Pipe()
162+ if err != nil {
163+ t.Fatal(err)
164+ }
165+ defer uffdWrite.Close()
166+ attachment := &attachment{
167+ token: "token",
168+ sandboxID: "sandbox",
169+ handoffPrepared: true,
170+ handoffDone: handoffDone,
171+ uffd: uffdRead,
172+ }
173+ server.attachments[attachment.token] = attachment
174+ response := server.handleWaitAttachmentReady(Request{Token: "token", SandboxID: "sandbox"}, baseResponse(Request{}))
175+ if !response.OK {
176+ t.Fatalf("WaitAttachmentReady() = %+v", response)
177+ }
178+ wrong := server.handleWaitAttachmentReady(Request{Token: "token", SandboxID: "other"}, baseResponse(Request{}))
179+ if wrong.OK {
180+ t.Fatalf("WaitAttachmentReady() accepted wrong sandbox: %+v", wrong)
181+ }
182+}
183+ 
184+func TestWaitAttachmentReadyReportsHandoffFailure(t *testing.T) {
185+ server := newServer(filepath.Join(t.TempDir(), "cow.sock"), Capabilities{IncrementalMemory: CapabilitySupported})
186+ handoffDone := make(chan struct{})
187+ close(handoffDone)
188+ item := &attachment{
189+ token: "token",
190+ sandboxID: "sandbox",
191+ handoffPrepared: true,
192+ handoffDone: handoffDone,
193+ handoffErr: errors.New("handoff failed"),
194+ }
195+ server.attachments[item.token] = item
196+ response := server.handleWaitAttachmentReady(Request{Token: "token", SandboxID: "sandbox"}, baseResponse(Request{}))
197+ if response.OK {
198+ t.Fatalf("WaitAttachmentReady() = %+v, want failure", response)
199+ }
200+}
201+ 
202+func writeServerManifest(t *testing.T) string {
203+ t.Helper()
204+ root := t.TempDir()
205+ if err := os.Mkdir(filepath.Join(root, memsnap.LayerDirName), 0o700); err != nil {
206+ t.Fatal(err)
207+ }
208+ for index, value := range []byte{0x31, 0x72} {
209+ layer := make([]byte, 2*memsnap.DefaultBlockSize)
210+ for offset := uint64(0); offset < memsnap.DefaultBlockSize; offset++ {
211+ layer[uint64(index)*memsnap.DefaultBlockSize+offset] = value
212+ }
213+ if err := os.WriteFile(filepath.Join(root, memsnap.LayerDirName, string(rune('0'+index))+".mem"), layer, 0o600); err != nil {
214+ t.Fatal(err)
215+ }
216+ }
217+ manifest := memsnap.Manifest{
218+ SchemaVersion: memsnap.SchemaVersion,
219+ MemorySize: 2 * memsnap.DefaultBlockSize,
220+ BlockSize: memsnap.DefaultBlockSize,
221+ Layers: []string{"layers/0.mem", "layers/1.mem"},
222+ BuildMap: []memsnap.BuildRange{
223+ {Offset: 0, Length: memsnap.DefaultBlockSize, LayerIndex: 0},
224+ {Offset: memsnap.DefaultBlockSize, Length: memsnap.DefaultBlockSize, LayerIndex: 1},
225+ },
226+ }
227+ data, err := json.Marshal(manifest)
228+ if err != nil {
229+ t.Fatal(err)
230+ }
231+ if err := os.WriteFile(filepath.Join(root, memsnap.ManifestFileName), data, 0o600); err != nil {
232+ t.Fatal(err)
233+ }
234+ return root
235+}
@@ -0,0 +1,466 @@
1+package cow
2+ 
3+import (
4+ "bytes"
5+ "crypto/sha256"
6+ "encoding/base64"
7+ "encoding/binary"
8+ "encoding/json"
9+ "errors"
10+ "fmt"
11+ "io"
12+ "net"
13+ "os"
14+ "path/filepath"
15+ "strconv"
16+ "time"
17+ "unsafe"
18+ 
19+ "github.com/openeuler/Conch/pkg/ulog"
20+ "golang.org/x/sys/unix"
21+)
22+ 
23+const (
24+ uffdEventPagefault = byte(0x12)
25+ uffdMessageSize = 32
26+ uffdReadyACK = byte(1)
27+ uffdPagefaultFlagWrite = uint64(1 << 0)
28+ uffdioWakeRequest = uintptr(0x8010aa02)
29+ linuxUnixPathMax = 107
30+ uffdBindAttempts = 8
31+)
32+ 
33+type uffdRange struct {
34+ HVA uint64
35+ Length uint64
36+ GuestOffset uint64
37+ HostPageSize uint64
38+}
39+ 
40+type uffdOperations struct {
41+ wake func(fd int, start, length uint64) error
42+ poll func(fds []unix.PollFd, timeout int) (int, error)
43+ read func(fd int, buffer []byte) (int, error)
44+ acceptTimeout time.Duration
45+ pollTimeout int
46+}
47+ 
48+func productionUFFDOperations() uffdOperations {
49+ return uffdOperations{
50+ wake: wakeUFFDRange,
51+ poll: unix.Poll,
52+ read: unix.Read,
53+ acceptTimeout: defaultUFFDAcceptTimeout,
54+ pollTimeout: 100,
55+ }
56+}
57+ 
58+type directUFFDMapping struct {
59+ BaseHVA uint64 `json:"base_host_virt_addr"`
60+ Size uint64 `json:"size"`
61+ Offset uint64 `json:"offset"`
62+ PageSizeBytes uint64 `json:"page_size_kib"`
63+}
64+ 
65+type wrappedUFFDMappings struct {
66+ Mappings []wrappedUFFDMapping `json:"mappings"`
67+}
68+ 
69+type wrappedUFFDMapping struct {
70+ BaseHVA uint64 `json:"base-host-virt-addr"`
71+ Size uint64 `json:"size"`
72+ Offset uint64 `json:"offset"`
73+ PageSize uint64 `json:"page-size"`
74+}
75+ 
76+func receiveUFFDHandoff(conn *net.UnixConn, memorySize, blockSize uint64) (*os.File, []uffdRange, error) {
77+ return receiveUFFDHandoffForHostPage(conn, memorySize, blockSize, uint64(os.Getpagesize()))
78+}
79+ 
80+func receiveUFFDHandoffForHostPage(conn *net.UnixConn, memorySize, blockSize, hostPageSize uint64) (*os.File, []uffdRange, error) {
81+ if err := conn.SetReadDeadline(time.Now().Add(requestTimeout)); err != nil {
82+ return nil, nil, fmt.Errorf("set UFFD handoff deadline: %w", err)
83+ }
84+ firstPayload := make([]byte, 4096)
85+ oob := make([]byte, unix.CmsgSpace(2*4))
86+ read, oobRead, flags, _, err := conn.ReadMsgUnix(firstPayload, oob)
87+ if err != nil {
88+ return nil, nil, fmt.Errorf("read UFFD handoff: %w", err)
89+ }
90+ fds, err := parseFDs(oob[:oobRead])
91+ if err != nil {
92+ return nil, nil, err
93+ }
94+ if flags&unix.MSG_CTRUNC != 0 {
95+ closeFDs(fds)
96+ return nil, nil, fmt.Errorf("UFFD handoff ancillary data truncated")
97+ }
98+ if flags&unix.MSG_TRUNC != 0 {
99+ closeFDs(fds)
100+ return nil, nil, fmt.Errorf("UFFD handoff first payload truncated")
101+ }
102+ if len(fds) != 1 {
103+ closeFDs(fds)
104+ return nil, nil, fmt.Errorf("expected exactly one UFFD descriptor, got %d", len(fds))
105+ }
106+ payload, err := readOneUFFDJSONValue(conn, firstPayload[:read])
107+ if err != nil {
108+ closeFDs(fds)
109+ return nil, nil, err
110+ }
111+ ranges, err := decodeUFFDRanges(payload, memorySize, blockSize, hostPageSize)
112+ if err != nil {
113+ closeFDs(fds)
114+ return nil, nil, err
115+ }
116+ uffd := os.NewFile(uintptr(fds[0]), "stratovirt-uffd")
117+ if uffd == nil {
118+ closeFDs(fds)
119+ return nil, nil, fmt.Errorf("wrap UFFD descriptor")
120+ }
121+ return uffd, ranges, nil
122+}
123+ 
124+type countingReader struct {
125+ reader io.Reader
126+ read int64
127+}
128+ 
129+func (reader *countingReader) Read(buffer []byte) (int, error) {
130+ read, err := reader.reader.Read(buffer)
131+ reader.read += int64(read)
132+ return read, err
133+}
134+ 
135+func readOneUFFDJSONValue(conn *net.UnixConn, first []byte) ([]byte, error) {
136+ if len(first) > maxFrameSize {
137+ return nil, fmt.Errorf("UFFD handoff payload too large")
138+ }
139+ firstReader := bytes.NewReader(first)
140+ connectionReader := &countingReader{reader: conn}
141+ limited := io.LimitReader(io.MultiReader(firstReader, connectionReader), maxFrameSize+1)
142+ decoder := json.NewDecoder(limited)
143+ var payload json.RawMessage
144+ if err := decoder.Decode(&payload); err != nil {
145+ return nil, fmt.Errorf("stream UFFD mappings: %w", err)
146+ }
147+ if int64(len(first))+connectionReader.read > int64(maxFrameSize) || len(payload) > maxFrameSize {
148+ return nil, fmt.Errorf("UFFD handoff payload too large")
149+ }
150+ buffered, err := io.ReadAll(decoder.Buffered())
151+ if err != nil {
152+ return nil, fmt.Errorf("inspect UFFD mapping suffix: %w", err)
153+ }
154+ unreadFirst, err := io.ReadAll(firstReader)
155+ if err != nil {
156+ return nil, fmt.Errorf("inspect initial UFFD mapping suffix: %w", err)
157+ }
158+ if len(bytes.TrimSpace(buffered)) != 0 || len(bytes.TrimSpace(unreadFirst)) != 0 {
159+ return nil, fmt.Errorf("multiple JSON values in UFFD handoff")
160+ }
161+ return payload, nil
162+}
163+ 
164+func decodeUFFDRanges(payload []byte, memorySize, blockSize, hostPageSize uint64) ([]uffdRange, error) {
165+ if memorySize == 0 || blockSize == 0 || hostPageSize == 0 {
166+ return nil, fmt.Errorf("invalid attachment memory geometry")
167+ }
168+ trimmed := bytes.TrimSpace(payload)
169+ if len(trimmed) == 0 {
170+ return nil, fmt.Errorf("empty UFFD mapping payload")
171+ }
172+ var ranges []uffdRange
173+ if trimmed[0] == '[' {
174+ var mappings []directUFFDMapping
175+ if err := decodeStrictPayload(trimmed, &mappings); err != nil {
176+ return nil, fmt.Errorf("decode direct UFFD mappings: %w", err)
177+ }
178+ for _, mapping := range mappings {
179+ ranges = append(ranges, uffdRange{HVA: mapping.BaseHVA, Length: mapping.Size, GuestOffset: mapping.Offset, HostPageSize: mapping.PageSizeBytes})
180+ }
181+ } else {
182+ var wrapped wrappedUFFDMappings
183+ if err := decodeStrictPayload(trimmed, &wrapped); err != nil {
184+ return nil, fmt.Errorf("decode wrapped UFFD mappings: %w", err)
185+ }
186+ for _, mapping := range wrapped.Mappings {
187+ ranges = append(ranges, uffdRange{HVA: mapping.BaseHVA, Length: mapping.Size, GuestOffset: mapping.Offset, HostPageSize: mapping.PageSize})
188+ }
189+ }
190+ if len(ranges) == 0 {
191+ return nil, fmt.Errorf("UFFD mappings are empty")
192+ }
193+ nextGuest := uint64(0)
194+ previousHVAEnd := uint64(0)
195+ for index, item := range ranges {
196+ if item.HostPageSize != hostPageSize || !compatibleHostPageSize(item.HostPageSize, blockSize) || item.Length == 0 ||
197+ item.HVA%item.HostPageSize != 0 || item.Length%item.HostPageSize != 0 || item.GuestOffset%item.HostPageSize != 0 {
198+ return nil, fmt.Errorf("UFFD mapping %d has invalid geometry", index)
199+ }
200+ if item.HVA > ^uint64(0)-item.Length || item.GuestOffset > memorySize || item.Length > memorySize-item.GuestOffset {
201+ return nil, fmt.Errorf("UFFD mapping %d overflows its address space", index)
202+ }
203+ if index != 0 && item.HVA < previousHVAEnd {
204+ return nil, fmt.Errorf("UFFD HVA mappings are unsorted or overlapping")
205+ }
206+ if item.GuestOffset != nextGuest {
207+ return nil, fmt.Errorf("UFFD Guest mappings are not a sorted contiguous cover")
208+ }
209+ previousHVAEnd = item.HVA + item.Length
210+ nextGuest = item.GuestOffset + item.Length
211+ }
212+ if nextGuest != memorySize {
213+ return nil, fmt.Errorf("UFFD Guest mappings do not cover memory")
214+ }
215+ return ranges, nil
216+}
217+ 
218+func decodeStrictPayload(payload []byte, destination any) error {
219+ decoder := json.NewDecoder(bytes.NewReader(payload))
220+ decoder.DisallowUnknownFields()
221+ if err := decoder.Decode(destination); err != nil {
222+ return err
223+ }
224+ if err := decoder.Decode(&struct{}{}); err != io.EOF {
225+ if err == nil {
226+ return fmt.Errorf("multiple JSON values")
227+ }
228+ return err
229+ }
230+ return nil
231+}
232+ 
233+func (server *Server) prepareUFFDHandoff(item *attachment, response Response) Response {
234+ directory := filepath.Dir(server.socketPath)
235+ var listener *net.UnixListener
236+ var socketPath string
237+ for range uffdBindAttempts {
238+ server.mu.Lock()
239+ server.uffdSequence++
240+ sequence := server.uffdSequence
241+ server.mu.Unlock()
242+ candidate, err := uffdSocketCandidate(directory, item.token, sequence)
243+ if err != nil {
244+ response.Error = err.Error()
245+ return response
246+ }
247+ listener, err = net.ListenUnix("unix", &net.UnixAddr{Name: candidate, Net: "unix"})
248+ if err == nil {
249+ socketPath = candidate
250+ break
251+ }
252+ if !errors.Is(err, unix.EADDRINUSE) {
253+ response.Error = fmt.Sprintf("listen UFFD socket: %v", err)
254+ return response
255+ }
256+ }
257+ if listener == nil {
258+ response.Error = "unable to bind generated UFFD socket"
259+ return response
260+ }
261+ listener.SetUnlinkOnClose(false)
262+ if err := os.Chmod(socketPath, 0o600); err != nil {
263+ _ = listener.Close()
264+ _ = os.Remove(socketPath)
265+ response.Error = fmt.Sprintf("chmod UFFD socket: %v", err)
266+ return response
267+ }
268+ server.mu.Lock()
269+ if server.closed || server.attachments[item.token] != item || item.handoffPrepared {
270+ server.mu.Unlock()
271+ _ = listener.Close()
272+ _ = os.Remove(socketPath)
273+ response.Error = "attachment changed during UFFD preparation"
274+ return response
275+ }
276+ item.handoffPrepared = true
277+ item.uffdListener = listener
278+ item.uffdSocketPath = socketPath
279+ item.handoffDone = make(chan struct{})
280+ server.mu.Unlock()
281+ go server.acceptUFFDHandoff(item, listener, socketPath)
282+ response.OK = true
283+ response.UFFDSocketPath = socketPath
284+ return response
285+}
286+ 
287+func uffdSocketCandidate(directory, token string, sequence uint64) (string, error) {
288+ digest := sha256.Sum256([]byte(token))
289+ encoded := base64.RawURLEncoding.EncodeToString(digest[:12])
290+ path := filepath.Join(directory, "u-"+encoded+"-"+strconv.FormatUint(sequence, 36))
291+ if len(path) > linuxUnixPathMax {
292+ return "", fmt.Errorf("UFFD Unix socket path is %d bytes; maximum is %d", len(path), linuxUnixPathMax)
293+ }
294+ return path, nil
295+}
296+ 
297+func (server *Server) acceptUFFDHandoff(item *attachment, listener *net.UnixListener, socketPath string) {
298+ server.mu.Lock()
299+ done := item.handoffDone
300+ server.mu.Unlock()
301+ var handoffErr error
302+ defer func() {
303+ server.mu.Lock()
304+ if server.attachments[item.token] == item {
305+ item.handoffErr = handoffErr
306+ }
307+ server.mu.Unlock()
308+ close(done)
309+ }()
310+ defer listener.Close()
311+ defer os.Remove(socketPath)
312+ _ = listener.SetDeadline(time.Now().Add(server.uffdOps.acceptTimeout))
313+ conn, err := listener.AcceptUnix()
314+ if err != nil {
315+ handoffErr = fmt.Errorf("accept UFFD handoff: %w", err)
316+ return
317+ }
318+ defer conn.Close()
319+ uffd, ranges, err := receiveUFFDHandoff(conn, item.pinned.Manifest.MemorySize, item.pinned.Manifest.BlockSize)
320+ if err != nil {
321+ handoffErr = err
322+ return
323+ }
324+ if _, err := conn.Write([]byte{uffdReadyACK}); err != nil {
325+ _ = uffd.Close()
326+ handoffErr = fmt.Errorf("acknowledge UFFD handoff: %w", err)
327+ return
328+ }
329+ server.mu.Lock()
330+ if server.attachments[item.token] != item || item.uffd != nil {
331+ server.mu.Unlock()
332+ _ = uffd.Close()
333+ handoffErr = fmt.Errorf("attachment changed during UFFD handoff")
334+ return
335+ }
336+ item.uffd = uffd
337+ item.uffdRanges = ranges
338+ item.workerStop = make(chan struct{})
339+ item.workerDone = make(chan struct{})
340+ go server.serveUFFD(item, uffd, ranges, item.workerStop, item.workerDone)
341+ server.mu.Unlock()
342+}
343+ 
344+func (server *Server) serveUFFD(item *attachment, uffd *os.File, ranges []uffdRange, stop, done chan struct{}) {
345+ defer close(done)
346+ defer func() {
347+ _ = uffd.Close()
348+ server.mu.Lock()
349+ if item.uffd == uffd {
350+ item.uffd = nil
351+ }
352+ server.mu.Unlock()
353+ }()
354+ if err := server.runUFFD(uffd, item, ranges, stop); err != nil {
355+ logger := ulog.GetLogger()
356+ logger.Error("UFFD worker failed", ulog.F("sandbox_id", item.sandboxID), ulog.F("error", err))
357+ }
358+}
359+ 
360+func (server *Server) runUFFD(uffd *os.File, item *attachment, ranges []uffdRange, stop <-chan struct{}) error {
361+ fd := int(uffd.Fd())
362+ pollFDs := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
363+ message := make([]byte, uffdMessageSize)
364+ for {
365+ select {
366+ case <-stop:
367+ return nil
368+ default:
369+ }
370+ count, err := server.uffdOps.poll(pollFDs, server.uffdOps.pollTimeout)
371+ if err != nil {
372+ if errors.Is(err, unix.EINTR) {
373+ continue
374+ }
375+ return fmt.Errorf("poll UFFD: %w", err)
376+ }
377+ if count == 0 {
378+ continue
379+ }
380+ if pollFDs[0].Revents&(unix.POLLERR|unix.POLLHUP|unix.POLLNVAL) != 0 {
381+ return fmt.Errorf("UFFD poll failed with events %#x", pollFDs[0].Revents)
382+ }
383+ if pollFDs[0].Revents&unix.POLLIN == 0 {
384+ continue
385+ }
386+ read, err := server.uffdOps.read(fd, message)
387+ if err != nil {
388+ if errors.Is(err, unix.EINTR) || errors.Is(err, unix.EAGAIN) {
389+ continue
390+ }
391+ return fmt.Errorf("read UFFD message: %w", err)
392+ }
393+ if read != len(message) {
394+ return fmt.Errorf("read UFFD message: got %d bytes, want %d", read, len(message))
395+ }
396+ if err := server.serviceUFFDMessage(item, fd, ranges, message); err != nil {
397+ return fmt.Errorf("service UFFD message: %w", err)
398+ }
399+ }
400+}
401+ 
402+func (server *Server) serviceUFFDMessage(item *attachment, uffd int, ranges []uffdRange, message []byte) error {
403+ if len(message) != uffdMessageSize {
404+ return fmt.Errorf("invalid UFFD message size %d", len(message))
405+ }
406+ if message[0] != uffdEventPagefault {
407+ return nil
408+ }
409+ flags := binary.LittleEndian.Uint64(message[8:16])
410+ if flags&^uffdPagefaultFlagWrite != 0 {
411+ return fmt.Errorf("unsupported non-missing UFFD pagefault flags %#x", flags)
412+ }
413+ return server.handlePageFault(item, uffd, binary.LittleEndian.Uint64(message[16:24]), ranges)
414+}
415+ 
416+func (server *Server) handlePageFault(item *attachment, uffd int, faultAddress uint64, ranges []uffdRange) error {
417+ blockSize := item.pinned.Manifest.BlockSize
418+ if blockSize == 0 || blockSize > uint64(^uint(0)>>1) {
419+ return fmt.Errorf("unsupported fault block size %d", blockSize)
420+ }
421+ var faultPage, guestOffset, hostPageSize uint64
422+ for _, candidate := range ranges {
423+ if faultAddress >= candidate.HVA && faultAddress-candidate.HVA < candidate.Length {
424+ hostPageSize = candidate.HostPageSize
425+ pageOffset := faultAddress - candidate.HVA
426+ pageOffset -= pageOffset % hostPageSize
427+ faultPage = candidate.HVA + pageOffset
428+ guestOffset = candidate.GuestOffset + pageOffset
429+ break
430+ }
431+ }
432+ if hostPageSize == 0 || !compatibleHostPageSize(hostPageSize, blockSize) || hostPageSize > item.pinned.Manifest.MemorySize || guestOffset > item.pinned.Manifest.MemorySize-hostPageSize {
433+ return fmt.Errorf("fault address %#x is outside registered Guest memory", faultAddress)
434+ }
435+ page := make([]byte, int(hostPageSize))
436+ for offset := uint64(0); offset < hostPageSize; offset += blockSize {
437+ if err := item.pinned.ReadPage(guestOffset+offset, page[int(offset):int(offset+blockSize)]); err != nil {
438+ return fmt.Errorf("read sparse page at %#x: %w", guestOffset+offset, err)
439+ }
440+ }
441+ written, err := item.memfd.WriteAt(page, int64(guestOffset))
442+ if err != nil {
443+ return fmt.Errorf("write memfd page at %#x: %w", guestOffset, err)
444+ }
445+ if written != len(page) {
446+ return fmt.Errorf("short memfd page write at %#x: wrote %d of %d", guestOffset, written, len(page))
447+ }
448+ if err := server.uffdOps.wake(uffd, faultPage, hostPageSize); err != nil {
449+ return fmt.Errorf("UFFDIO_WAKE page at %#x: %w", faultPage, err)
450+ }
451+ return nil
452+}
453+ 
454+type uffdioRange struct {
455+ Start uint64
456+ Length uint64
457+}
458+ 
459+func wakeUFFDRange(fd int, start, length uint64) error {
460+ request := uffdioRange{Start: start, Length: length}
461+ _, _, errno := unix.Syscall(unix.SYS_IOCTL, uintptr(fd), uffdioWakeRequest, uintptr(unsafe.Pointer(&request)))
462+ if errno != 0 {
463+ return errno
464+ }
465+ return nil
466+}
@@ -0,0 +1,130 @@
1+package cow
2+ 
3+import (
4+ "bytes"
5+ "errors"
6+ "os"
7+ "path/filepath"
8+ "strings"
9+ "testing"
10+ "time"
11+ 
12+ "github.com/openeuler/Conch/internal/memsnap"
13+ "golang.org/x/sys/unix"
14+)
15+ 
16+func TestReceiveUFFDHandoffAcceptsStratoVirtMappingShapes(t *testing.T) {
17+ shapes := map[string]string{
18+ "direct": `[{"base_host_virt_addr":65536,"size":4096,"offset":0,"page_size_kib":4096},{"base_host_virt_addr":131072,"size":4096,"offset":4096,"page_size_kib":4096}]`,
19+ "wrapped": `{"mappings":[{"base-host-virt-addr":65536,"size":4096,"offset":0,"page-size":4096},{"base-host-virt-addr":131072,"size":4096,"offset":4096,"page-size":4096}]}`,
20+ }
21+ for name, payload := range shapes {
22+ t.Run(name, func(t *testing.T) {
23+ receiver, sender := testUnixSocketPair(t)
24+ readEnd, writeEnd, err := os.Pipe()
25+ if err != nil {
26+ t.Fatal(err)
27+ }
28+ defer readEnd.Close()
29+ defer writeEnd.Close()
30+ if _, _, err := sender.WriteMsgUnix([]byte(payload), unix.UnixRights(int(readEnd.Fd())), nil); err != nil {
31+ t.Fatal(err)
32+ }
33+ uffd, ranges, err := receiveUFFDHandoffForHostPage(receiver, 2*memsnap.DefaultBlockSize, memsnap.DefaultBlockSize, memsnap.DefaultBlockSize)
34+ if err != nil {
35+ t.Fatal(err)
36+ }
37+ defer uffd.Close()
38+ if len(ranges) != 2 || ranges[0].GuestOffset != 0 || ranges[1].GuestOffset != memsnap.DefaultBlockSize {
39+ t.Fatalf("ranges = %#v", ranges)
40+ }
41+ })
42+ }
43+}
44+ 
45+func TestDecodeUFFDRangesRejectsInvalidCoverage(t *testing.T) {
46+ tests := map[string]string{
47+ "empty": `[]`,
48+ "HVA overlap": `[{"base_host_virt_addr":65536,"size":8192,"offset":0,"page_size_kib":4096},{"base_host_virt_addr":69632,"size":4096,"offset":4096,"page_size_kib":4096}]`,
49+ "guest gap": `[{"base_host_virt_addr":65536,"size":4096,"offset":0,"page_size_kib":4096},{"base_host_virt_addr":131072,"size":4096,"offset":8192,"page_size_kib":4096}]`,
50+ "incomplete guest": `[{"base_host_virt_addr":65536,"size":4096,"offset":0,"page_size_kib":4096}]`,
51+ }
52+ for name, payload := range tests {
53+ t.Run(name, func(t *testing.T) {
54+ if _, err := decodeUFFDRanges([]byte(payload), 2*memsnap.DefaultBlockSize, memsnap.DefaultBlockSize, memsnap.DefaultBlockSize); err == nil {
55+ t.Fatal("decodeUFFDRanges accepted invalid ranges")
56+ }
57+ })
58+ }
59+}
60+ 
61+func TestPageFaultWritesBuildMapOwnerToMemfdBeforeWake(t *testing.T) {
62+ const hostPageSize = uint64(64 * 1024)
63+ root := t.TempDir()
64+ manifest, err := memsnap.CreateBaseLayer(root, hostPageSize, memsnap.DefaultBlockSize, func(sink memsnap.PageSink) error {
65+ for offset := uint64(0); offset < hostPageSize; offset += memsnap.DefaultBlockSize {
66+ page := bytes.Repeat([]byte{byte(offset/memsnap.DefaultBlockSize + 1)}, int(memsnap.DefaultBlockSize))
67+ if err := sink.WritePage(offset, page); err != nil {
68+ return err
69+ }
70+ }
71+ return nil
72+ })
73+ if err != nil {
74+ t.Fatal(err)
75+ }
76+ if err := memsnap.WriteManifestAtomic(filepath.Join(root, memsnap.ManifestFileName), manifest); err != nil {
77+ t.Fatal(err)
78+ }
79+ server := newServer(filepath.Join(t.TempDir(), "cow.sock"), Capabilities{IncrementalMemory: CapabilitySupported})
80+ server.uffdOps.acceptTimeout = time.Second
81+ response, fds := server.handleAttach(Request{SandboxID: "sandbox", MemorySnapshotRoot: root}, Response{})
82+ defer closeFDs(fds)
83+ defer server.Close()
84+ var wakeFD int
85+ var wakeStart, wakeLength uint64
86+ server.uffdOps.wake = func(fd int, start, length uint64) error {
87+ wakeFD, wakeStart, wakeLength = fd, start, length
88+ return nil
89+ }
90+ ranges := []uffdRange{{HVA: 0x20000, Length: hostPageSize, GuestOffset: 0, HostPageSize: hostPageSize}}
91+ item := server.attachments[response.Token]
92+ if err := server.handlePageFault(item, 9, 0x23456, ranges); err != nil {
93+ t.Fatal(err)
94+ }
95+ page := make([]byte, hostPageSize)
96+ read, err := unix.Pread(int(item.memfd.Fd()), page, 0)
97+ if err != nil {
98+ t.Fatal(err)
99+ }
100+ if read != int(hostPageSize) || page[0] != 1 || page[memsnap.DefaultBlockSize] != 2 {
101+ t.Fatalf("memfd size/bytes = %d/%d/%d", read, page[0], page[memsnap.DefaultBlockSize])
102+ }
103+ if wakeFD != 9 || wakeStart != 0x20000 || wakeLength != hostPageSize {
104+ t.Fatalf("wake = fd %d start %#x length %d", wakeFD, wakeStart, wakeLength)
105+ }
106+}
107+ 
108+func TestPageFaultReturnsWakeFailure(t *testing.T) {
109+ const hostPageSize = uint64(memsnap.DefaultBlockSize)
110+ root := t.TempDir()
111+ manifest, err := memsnap.CreateBaseLayer(root, hostPageSize, memsnap.DefaultBlockSize, func(sink memsnap.PageSink) error {
112+ return sink.WriteZeroPage(0)
113+ })
114+ if err != nil {
115+ t.Fatal(err)
116+ }
117+ if err := memsnap.WriteManifestAtomic(filepath.Join(root, memsnap.ManifestFileName), manifest); err != nil {
118+ t.Fatal(err)
119+ }
120+ server := newServer(filepath.Join(t.TempDir(), "cow.sock"), Capabilities{IncrementalMemory: CapabilitySupported})
121+ server.uffdOps.acceptTimeout = time.Second
122+ response, fds := server.handleAttach(Request{SandboxID: "sandbox", MemorySnapshotRoot: root}, Response{})
123+ defer closeFDs(fds)
124+ defer server.Close()
125+ server.uffdOps.wake = func(int, uint64, uint64) error { return errors.New("wake failed") }
126+ ranges := []uffdRange{{HVA: 0x20000, Length: hostPageSize, GuestOffset: 0, HostPageSize: hostPageSize}}
127+ if err := server.handlePageFault(server.attachments[response.Token], 9, 0x20000, ranges); err == nil || !strings.Contains(err.Error(), "wake failed") {
128+ t.Fatalf("wake error = %v", err)
129+ }
130+}
@@ -25,8 +25,10 @@ import (
25 "github.com/openeuler/Conch/internal/cleanupdiag"25 "github.com/openeuler/Conch/internal/cleanupdiag"
26 "github.com/openeuler/Conch/internal/conchruntime"26 "github.com/openeuler/Conch/internal/conchruntime"
27 "github.com/openeuler/Conch/internal/config"27 "github.com/openeuler/Conch/internal/config"
28+ "github.com/openeuler/Conch/internal/cow"
28 "github.com/openeuler/Conch/internal/daemon/state"29 "github.com/openeuler/Conch/internal/daemon/state"
29 conchimage "github.com/openeuler/Conch/internal/image"30 conchimage "github.com/openeuler/Conch/internal/image"
31+ "github.com/openeuler/Conch/internal/memorymode"
30 "github.com/openeuler/Conch/internal/netstack"32 "github.com/openeuler/Conch/internal/netstack"
31 "github.com/openeuler/Conch/internal/runtimeapi"33 "github.com/openeuler/Conch/internal/runtimeapi"
32 conchsandbox "github.com/openeuler/Conch/internal/sandbox"34 conchsandbox "github.com/openeuler/Conch/internal/sandbox"
@@ -51,6 +53,7 @@ type Daemon struct {
51 listener net.Listener53 listener net.Listener
52 unixSocketPath string54 unixSocketPath string
53 cleanupOnce sync.Once55 cleanupOnce sync.Once
56+ cowProcess *cow.Process
54 57 
55 // TODO: need ListCachedBuilds()58 // TODO: need ListCachedBuilds()
56}59}
@@ -99,6 +102,19 @@ func New(cfg *config.Config) (*Daemon, error) {
99 s.routes()102 s.routes()
100 103 
101 logger := ulog.GetLogger()104 logger := ulog.GetLogger()
105+ cowProcess, err := cow.StartProcess(ctx, cfg.Sandbox.CowBinary, cfg.Sandbox.CowSocket)
106+ if err != nil {
107+ cancel()
108+ return nil, fmt.Errorf("start conch-cow: %w", err)
109+ }
110+ s.cowProcess = cowProcess
111+ startupComplete := false
112+ defer func() {
113+ if !startupComplete {
114+ _ = cowProcess.Close()
115+ }
116+ }()
117+ logger.Info("conch-cow initialized", ulog.F("binary", cfg.Sandbox.CowBinary), ulog.F("socket", cfg.Sandbox.CowSocket))
102 118 
103 store, err := state.OpenBolt(cfg.State.Path)119 store, err := state.OpenBolt(cfg.State.Path)
104 if err != nil {120 if err != nil {
@@ -137,6 +153,7 @@ func New(cfg *config.Config) (*Daemon, error) {
137 VsockSignalTimeout: cfg.Sandbox.VsockSignalTimeout,153 VsockSignalTimeout: cfg.Sandbox.VsockSignalTimeout,
138 RequestTimeout: cfg.Sandbox.RequestTimeout,154 RequestTimeout: cfg.Sandbox.RequestTimeout,
139 VolumeManager: s.volumeManager,155 VolumeManager: s.volumeManager,
156+ CowSocket: cfg.Sandbox.CowSocket,
140 },157 },
141 })158 })
142 if err != nil {159 if err != nil {
@@ -159,6 +176,7 @@ func New(cfg *config.Config) (*Daemon, error) {
159 VCPUMax: cfg.Sandbox.DefaultVCPUMax,176 VCPUMax: cfg.Sandbox.DefaultVCPUMax,
160 RamMB: cfg.Sandbox.DefaultRAMMB,177 RamMB: cfg.Sandbox.DefaultRAMMB,
161 })178 })
179+ s.runtimeService.SetMemoryPolicy(memorymode.RequestedMode(cfg.Sandbox.MemoryMode), cow.NewClient(cfg.Sandbox.CowSocket))
162 180 
163 manager := host.SandboxManager()181 manager := host.SandboxManager()
164 if manager != nil {182 if manager != nil {
@@ -196,6 +214,7 @@ func New(cfg *config.Config) (*Daemon, error) {
196 214 
197 handleSignals(ctx, cancel, s)215 handleSignals(ctx, cancel, s)
198 216 
217+ startupComplete = true
199 logger.Info("Server initialized successfully")218 logger.Info("Server initialized successfully")
200 return s, nil219 return s, nil
201}220}
@@ -331,6 +350,15 @@ func (s *Daemon) Shutdown() {
331 }350 }
332 }351 }
333 352 
353+ if s.cowProcess != nil {
354+ finish := cleanupdiag.Start("daemon.cow.close")
355+ err := s.cowProcess.Close()
356+ finish(err)
357+ if err != nil {
358+ logger.Error("conch-cow cleanup error", ulog.F("error", err))
359+ }
360+ }
361+ 
334 if s.containerdHost != nil {362 if s.containerdHost != nil {
335 finish := cleanupdiag.Start("daemon.containerd_host.close")363 finish := cleanupdiag.Start("daemon.containerd_host.close")
336 err := s.containerdHost.Close()364 err := s.containerdHost.Close()
@@ -443,6 +471,8 @@ func (s *Daemon) handleCreateSandbox(w http.ResponseWriter, r *http.Request) {
443 status := http.StatusInternalServerError471 status := http.StatusInternalServerError
444 if errors.Is(err, conchruntime.ErrSandboxAlreadyExists) {472 if errors.Is(err, conchruntime.ErrSandboxAlreadyExists) {
445 status = http.StatusConflict473 status = http.StatusConflict
474+ } else if errors.Is(err, memorymode.ErrPrecondition) {
475+ status = http.StatusPreconditionFailed
446 }476 }
447 http.Error(w, "Failed to create sandbox: "+err.Error(), status)477 http.Error(w, "Failed to create sandbox: "+err.Error(), status)
448 return478 return
@@ -4,6 +4,7 @@ import (
4 "bytes"4 "bytes"
5 "context"5 "context"
6 "encoding/json"6 "encoding/json"
7+ "fmt"
7 "net/http"8 "net/http"
8 "net/http/httptest"9 "net/http/httptest"
9 "testing"10 "testing"
@@ -11,6 +12,7 @@ import (
11 "github.com/openeuler/Conch/internal/conchruntime"12 "github.com/openeuler/Conch/internal/conchruntime"
12 "github.com/openeuler/Conch/internal/config"13 "github.com/openeuler/Conch/internal/config"
13 "github.com/openeuler/Conch/internal/daemon/state"14 "github.com/openeuler/Conch/internal/daemon/state"
15+ "github.com/openeuler/Conch/internal/memorymode"
14)16)
15 17 
16func TestHandleCreateSandboxReturnsGeneratedSandboxID(t *testing.T) {18func TestHandleCreateSandboxReturnsGeneratedSandboxID(t *testing.T) {
@@ -39,6 +41,21 @@ func TestHandleCreateSandboxReturnsGeneratedSandboxID(t *testing.T) {
39 }41 }
40}42}
41 43 
44+func TestHandleCreateSandboxMapsMemoryPreconditionToHTTP412(t *testing.T) {
45+ sandboxOps := &fakeSandboxOps{createErr: fmt.Errorf("resolve memory mode: %w", memorymode.ErrPrecondition)}
46+ runtimeService := conchruntime.New(sandboxOps, nil, nil)
47+ runtimeService.SetSandboxDefaults(conchruntime.SandboxDefaults{TemplateID: "tmpl-default"})
48+ server := &Daemon{router: http.NewServeMux(), runtimeService: runtimeService}
49+ server.routes()
50+ 
51+ recorder := httptest.NewRecorder()
52+ request := httptest.NewRequest(http.MethodPost, "/api/v1/sandboxes", bytes.NewBufferString(`{}`))
53+ server.router.ServeHTTP(recorder, request)
54+ if recorder.Code != http.StatusPreconditionFailed {
55+ t.Fatalf("status = %d, body = %s", recorder.Code, recorder.Body.String())
56+ }
57+}
58+ 
42func TestRemoveAllSandboxesDeletesRuntimeAndStateRecords(t *testing.T) {59func TestRemoveAllSandboxesDeletesRuntimeAndStateRecords(t *testing.T) {
43 store, err := state.OpenBolt(t.TempDir() + "/state.db")60 store, err := state.OpenBolt(t.TempDir() + "/state.db")
44 if err != nil {61 if err != nil {