diff --git a/pkg/aio/BUILD b/pkg/aio/BUILD new file mode 100644 index 000000000..850235267 --- /dev/null +++ b/pkg/aio/BUILD @@ -0,0 +1,30 @@ +load("//tools:defs.bzl", "go_library", "go_test") + +package( + default_applicable_licenses = ["//:license"], + licenses = ["notice"], +) + +go_library( + name = "aio", + srcs = [ + "aio.go", + "aio_unsafe.go", + ], + visibility = ["//pkg/sentry:internal"], + deps = [ + "//pkg/sync", + "@org_golang_x_sys//unix:go_default_library", + ], +) + +go_test( + name = "aio_test", + size = "small", + srcs = ["aio_test.go"], + library = ":aio", + deps = [ + "//pkg/bitmap", + "@org_golang_x_sys//unix:go_default_library", + ], +) diff --git a/pkg/aio/aio.go b/pkg/aio/aio.go new file mode 100644 index 000000000..fcd321ce0 --- /dev/null +++ b/pkg/aio/aio.go @@ -0,0 +1,185 @@ +// Copyright 2024 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package aio provides asynchronous I/O on host file descriptors. +package aio + +import ( + "fmt" + + "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/sync" +) + +// A Queue provides the ability to concurrently execute multiple read/write +// operations on host file descriptors. +// +// Queues are not safe to use concurrently in multiple goroutines. +type Queue interface { + // Destroy cancels all inflight operations and releases resources owned by + // the Queue. Destroy waits for cancelation, so the Queue will not access + // memory corresponding to inflight operations after Destroy returns. + Destroy() + + // Cap returns the Queue's capacity, which is the maximum number of + // concurrent operations supported by the Queue. + Cap() int + + // Add enqueues an inflight operation. + // + // Note that some Queue implementations may not begin execution of new + // Requests until the following call to Wait. + // + // Preconditions: + // - The current number of inflight operations < Cap(). + Add(req Request) + + // Wait blocks until at least minCompletions inflight operations have + // completed, then appends all completed inflight operations to cs and + // returns the updated slice. + // + // If Wait returns a non-nil error, no Queue methods may be subsequently + // called except Destroy. + // + // Preconditions: + // - 0 <= minCompletions <= Cap(). + Wait(cs []Completion, minCompletions int) ([]Completion, error) +} + +// Request is defined in aio_unsafe.go. + +// Op selects an asynchronous I/O operation. +type Op uint8 + +// Possible values for Request.Op. +const ( + // OpRead represents a read into addresses [Buf, Buf+Len). + OpRead Op = iota + + // OpWrite represents a write from addresses [Buf, Buf+Len). + OpWrite + + // OpReadv represents a read, where the destination addresses are given by + // the struct iovec array at address Buf of length Len. + OpReadv + + // OpWritev represents a write, where the source addresses are given by the + // struct iovec array at address Buf of length Len. + OpWritev +) + +// Completion provides outputs from an asynchronous I/O operation. +type Completion struct { + ID uint64 // copied from Request.ID in the corresponding Request + Result int64 // number of bytes or negative errno +} + +// Err returns an error representing the Completion. If Err returns nil, +// c.Result is the number of bytes completed by the operation. +func (c Completion) Err() error { + if c.Result >= 0 { + return nil + } + return unix.Errno(-c.Result) +} + +// GoQueue implements Queue using a pool of worker goroutines. +type GoQueue struct { + requests chan Request + completions chan Completion + shutdown chan struct{} + workers sync.WaitGroup +} + +// NewGoQueue returns a new GoQueue with the given capacity. +func NewGoQueue(cap int) *GoQueue { + q := &GoQueue{ + requests: make(chan Request, cap), + completions: make(chan Completion, cap), + shutdown: make(chan struct{}), + } + q.workers.Add(cap) + for range cap { + go q.workerMain() + } + return q +} + +func (q *GoQueue) workerMain() { + defer q.workers.Done() + for { + select { + case <-q.shutdown: + return + case r := <-q.requests: + var sysno uintptr + switch r.Op { + case OpRead: + sysno = unix.SYS_PREAD64 + case OpWrite: + sysno = unix.SYS_PWRITE64 + case OpReadv: + sysno = unix.SYS_PREADV2 + case OpWritev: + sysno = unix.SYS_PWRITEV2 + default: + panic(fmt.Sprintf("unknown op %v", r.Op)) + } + n, _, e := unix.Syscall6(sysno, uintptr(r.FD), uintptr(r.Buf), uintptr(r.Len), uintptr(r.Off), 0 /* pos_h */, 0 /* flags/unused */) + c := Completion{ + ID: r.ID, + Result: int64(n), + } + if e != 0 { + c.Result = -int64(e) + } + q.completions <- c + } + } +} + +// Destroy implements Queue.Destroy. +func (q *GoQueue) Destroy() { + close(q.shutdown) + q.workers.Wait() +} + +// Cap implements Queue.Cap. +func (q *GoQueue) Cap() int { + return cap(q.requests) +} + +// Add implements Queue.Add. +func (q *GoQueue) Add(r Request) { + q.requests <- r +} + +// Wait implements Queue.Wait. +func (q *GoQueue) Wait(cs []Completion, minCompletions int) ([]Completion, error) { + i := 0 + for { + if i < minCompletions { + cs = append(cs, <-q.completions) + i++ + } else { + select { + case c := <-q.completions: + cs = append(cs, c) + i++ + default: + return cs, nil + } + } + } +} diff --git a/pkg/aio/aio_test.go b/pkg/aio/aio_test.go new file mode 100644 index 000000000..fc7b5d1b6 --- /dev/null +++ b/pkg/aio/aio_test.go @@ -0,0 +1,280 @@ +// Copyright 2024 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package aio + +import ( + "bytes" + "io" + "math/rand" + "os" + "testing" + + "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/bitmap" +) + +func TestRead(t *testing.T) { + // Create a temp file. + testFile, err := os.CreateTemp(t.TempDir(), "aio_test_read") + if err != nil { + t.Fatalf("failed to create temp file: %v", err) + } + defer testFile.Close() + defer os.Remove(testFile.Name()) + + // Create random data. + const chunkSize = 4096 + const dataLen = 1024 * chunkSize + data := make([]byte, dataLen) + _, _ = rand.Read(data) + + // Write data to the file using sync writes. + if _, err := testFile.Write(data); err != nil { + t.Fatalf("failed to write temp file: %v", err) + } + + // Read data from the file using async reads. + q := NewGoQueue(8) + defer q.Destroy() + qavail := q.Cap() + off := int64(0) + fd := int32(testFile.Fd()) + buf := make([]byte, dataLen) + added := 0 + done := 0 + var cs []Completion + for done < dataLen { + for qavail > 0 && added < dataLen { + Read(q, 0 /* id */, fd, off, buf[added:added+chunkSize]) + qavail-- + off += chunkSize + added += chunkSize + } + cs, err := q.Wait(cs[:0], 1 /* minCompletions */) + if err != nil { + t.Fatalf("Queue.Wait failed: %v", err) + } + for _, c := range cs { + if err := c.Err(); err != nil { + t.Fatalf("Queue returned completion with error: %v", err) + } + if c.Result != chunkSize { + t.Fatalf("Queue returned completion of %d bytes, want %d", c.Result, chunkSize) + } + qavail++ + done += chunkSize + } + } + if bytes.Compare(data, buf) != 0 { + t.Errorf("bytes differ") + } +} + +func TestReadv(t *testing.T) { + // Create a temp file. + testFile, err := os.CreateTemp(t.TempDir(), "aio_test_readv") + if err != nil { + t.Fatalf("failed to create temp file: %v", err) + } + defer testFile.Close() + defer os.Remove(testFile.Name()) + + // Create random data. + const chunkSize = 4096 + const dataLen = 1024 * chunkSize + data := make([]byte, dataLen) + _, _ = rand.Read(data) + + // Write data to the file using sync writes. + if _, err := testFile.Write(data); err != nil { + t.Fatalf("failed to write temp file: %v", err) + } + + // Read data from the file using async vectored reads. + q := NewGoQueue(8) + defer q.Destroy() + qavail := q.Cap() + iovecsData := make([][2]unix.Iovec, qavail) + iovecsBusy := bitmap.New(uint32(qavail)) + off := int64(0) + fd := int32(testFile.Fd()) + buf := make([]byte, dataLen) + added := 0 + done := 0 + var cs []Completion + for done < dataLen { + for qavail > 0 && added < dataLen { + id, err := iovecsBusy.FirstZero(0) + if err != nil { + t.Fatalf("all iovecs busy with qavail=%d", qavail) + } + iovecsBusy.Add(id) + iovecs := &iovecsData[id] + iovecs[0].Base = &buf[added] + iovecs[0].Len = chunkSize + iovecs[1].Base = &buf[added+chunkSize] + iovecs[1].Len = chunkSize + Readv(q, uint64(id), fd, off, iovecs[:]) + qavail-- + off += 2 * chunkSize + added += 2 * chunkSize + } + cs, err := q.Wait(cs[:0], 1 /* minCompletions */) + if err != nil { + t.Fatalf("Queue.Wait failed: %v", err) + } + for _, c := range cs { + if err := c.Err(); err != nil { + t.Fatalf("Queue returned completion with error: %v", err) + } + if c.Result != 2*chunkSize { + t.Fatalf("Queue returned completion of %d bytes, want %d", c.Result, 2*chunkSize) + } + qavail++ + iovecsBusy.Remove(uint32(c.ID)) + done += 2 * chunkSize + } + } + if bytes.Compare(data, buf) != 0 { + t.Errorf("bytes differ") + } +} + +func TestWrite(t *testing.T) { + // Create a temp file. + testFile, err := os.CreateTemp(t.TempDir(), "aio_test_write") + if err != nil { + t.Fatalf("failed to create temp file: %v", err) + } + defer testFile.Close() + defer os.Remove(testFile.Name()) + + // Create random data. + const chunkSize = 4096 + const dataLen = 1024 * chunkSize + data := make([]byte, dataLen) + _, _ = rand.Read(data) + + // Write data to the file using async writes. + q := NewGoQueue(8) + defer q.Destroy() + qavail := q.Cap() + off := int64(0) + fd := int32(testFile.Fd()) + added := 0 + done := 0 + var cs []Completion + for done < dataLen { + for qavail > 0 && added < dataLen { + Write(q, 0 /* id */, fd, off, data[added:added+chunkSize]) + qavail-- + off += chunkSize + added += chunkSize + } + cs, err := q.Wait(cs[:0], 1 /* minCompletions */) + if err != nil { + t.Fatalf("Queue.Wait failed: %v", err) + } + for _, c := range cs { + if err := c.Err(); err != nil { + t.Fatalf("Queue returned completion with error: %v", err) + } + if c.Result != chunkSize { + t.Fatalf("Queue returned completion of %d bytes, want %d", c.Result, chunkSize) + } + qavail++ + done += chunkSize + } + } + + // Read data from the file using sync reads. + buf := make([]byte, dataLen) + if n, err := io.ReadFull(testFile, buf); err != nil { + t.Fatalf("failed to read temp file after %d bytes: %v", n, err) + } + if bytes.Compare(data, buf) != 0 { + t.Errorf("bytes differ") + } +} + +func TestWritev(t *testing.T) { + // Create a temp file. + testFile, err := os.CreateTemp(t.TempDir(), "aio_test_writev") + if err != nil { + t.Fatalf("failed to create temp file: %v", err) + } + defer testFile.Close() + defer os.Remove(testFile.Name()) + + // Create random data. + const chunkSize = 4096 + const dataLen = 1024 * chunkSize + data := make([]byte, dataLen) + _, _ = rand.Read(data) + + // Write data to the file using async vectored writes. + q := NewGoQueue(8) + defer q.Destroy() + qavail := q.Cap() + iovecsData := make([][2]unix.Iovec, qavail) + iovecsBusy := bitmap.New(uint32(qavail)) + off := int64(0) + fd := int32(testFile.Fd()) + added := 0 + done := 0 + var cs []Completion + for done < dataLen { + for qavail > 0 && added < dataLen { + id, err := iovecsBusy.FirstZero(0) + if err != nil { + t.Fatalf("all iovecs busy with qavail=%d", qavail) + } + iovecsBusy.Add(id) + iovecs := &iovecsData[id] + iovecs[0].Base = &data[added] + iovecs[0].Len = chunkSize + iovecs[1].Base = &data[added+chunkSize] + iovecs[1].Len = chunkSize + Writev(q, uint64(id), fd, off, iovecs[:]) + qavail-- + off += 2 * chunkSize + added += 2 * chunkSize + } + cs, err := q.Wait(cs[:0], 1 /* minCompletions */) + if err != nil { + t.Fatalf("Queue.Wait failed: %v", err) + } + for _, c := range cs { + if err := c.Err(); err != nil { + t.Fatalf("Queue returned completion with error: %v", err) + } + if c.Result != 2*chunkSize { + t.Fatalf("Queue returned completion of %d bytes, want %d", c.Result, 2*chunkSize) + } + qavail++ + iovecsBusy.Remove(uint32(c.ID)) + done += 2 * chunkSize + } + } + + // Read data from the file using sync reads. + buf := make([]byte, dataLen) + if n, err := io.ReadFull(testFile, buf); err != nil { + t.Fatalf("failed to read temp file after %d bytes: %v", n, err) + } + if bytes.Compare(data, buf) != 0 { + t.Errorf("bytes differ") + } +} diff --git a/pkg/aio/aio_unsafe.go b/pkg/aio/aio_unsafe.go new file mode 100644 index 000000000..cdea24eed --- /dev/null +++ b/pkg/aio/aio_unsafe.go @@ -0,0 +1,93 @@ +// Copyright 2024 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package aio + +import ( + "unsafe" + + "golang.org/x/sys/unix" +) + +// Request provides inputs to an asynchronous I/O operation. +type Request struct { + ID uint64 // copied to Completion.ID in the corresponding Completion + Op Op + FD int32 // host file descriptor + Off int64 // offset into FD + Buf unsafe.Pointer // depends on Op + Len int // depends on Op +} + +// Read enqueues a read. The caller must ensure that the memory referred to by +// dst remains valid until the read is complete. +// +// Preconditions: As for q.Add(). +func Read(q Queue, id uint64, fd int32, off int64, dst []byte) { + q.Add(Request{ + ID: id, + Op: OpRead, + FD: fd, + Off: off, + Buf: unsafe.Pointer(unsafe.SliceData(dst)), + Len: len(dst), + }) +} + +// Write enqueues a write. The caller must ensure that the memory referred to +// by src remains valid until the write is complete. +// +// Preconditions: As for q.Add(). +func Write(q Queue, id uint64, fd int32, off int64, src []byte) { + q.Add(Request{ + ID: id, + Op: OpWrite, + FD: fd, + Off: off, + Buf: unsafe.Pointer(unsafe.SliceData(src)), + Len: len(src), + }) +} + +// Readv enqueues a vectored read. The caller must ensure that the struct iovec +// array referred to by dst, and the memory that those struct iovecs refer to, +// remain valid until the read is complete. +// +// Preconditions: As for q.Add(). +func Readv(q Queue, id uint64, fd int32, off int64, dst []unix.Iovec) { + q.Add(Request{ + ID: id, + Op: OpReadv, + FD: fd, + Off: off, + Buf: unsafe.Pointer(unsafe.SliceData(dst)), + Len: len(dst), + }) +} + +// Writev enqueues a vectored write. The caller must ensure that the struct +// iovec array referred to by src, and the memory that those struct iovecs +// refer to, remain valid until the write is complete. +// +// Preconditions: As for q.Add(). +func Writev(q Queue, id uint64, fd int32, off int64, src []unix.Iovec) { + q.Add(Request{ + ID: id, + Op: OpWritev, + FD: fd, + Off: off, + Buf: unsafe.Pointer(unsafe.SliceData(src)), + Len: len(src), + }) +} diff --git a/pkg/sentry/kernel/BUILD b/pkg/sentry/kernel/BUILD index f9c561d95..3b85c43a2 100644 --- a/pkg/sentry/kernel/BUILD +++ b/pkg/sentry/kernel/BUILD @@ -386,7 +386,6 @@ go_library( "//pkg/sentry/usage", "//pkg/sentry/vfs", "//pkg/state", - "//pkg/state/statefile", "//pkg/state/wire", "//pkg/sync", "//pkg/sync/locking", diff --git a/pkg/sentry/kernel/kernel.go b/pkg/sentry/kernel/kernel.go index 05e71fb77..26cd4786c 100644 --- a/pkg/sentry/kernel/kernel.go +++ b/pkg/sentry/kernel/kernel.go @@ -75,7 +75,6 @@ import ( "gvisor.dev/gvisor/pkg/sentry/uniqueid" "gvisor.dev/gvisor/pkg/sentry/vfs" "gvisor.dev/gvisor/pkg/state" - "gvisor.dev/gvisor/pkg/state/statefile" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" ) @@ -597,7 +596,7 @@ func savePrivateMFs(ctx context.Context, w io.Writer, pw io.Writer, mfsToSave ma return nil } -func loadPrivateMFs(ctx context.Context, r io.Reader, pr *statefile.AsyncReader) error { +func loadPrivateMFs(ctx context.Context, r io.Reader, opts *pgalloc.LoadOpts) error { // Load the metadata. var meta privateMemoryFileMetadata if _, err := state.Load(ctx, r, &meta); err != nil { @@ -614,7 +613,7 @@ func loadPrivateMFs(ctx context.Context, r io.Reader, pr *statefile.AsyncReader) if !ok { return fmt.Errorf("saved memory file for %q was not configured on restore", fsID) } - if err := mf.LoadFrom(ctx, r, pr); err != nil { + if err := mf.LoadFrom(ctx, r, opts); err != nil { return err } } @@ -767,9 +766,16 @@ func (k *Kernel) invalidateUnsavableMappings(ctx context.Context) error { } // LoadFrom returns a new Kernel loaded from args. -func (k *Kernel) LoadFrom(ctx context.Context, r, pagesMetadata io.Reader, pagesFile *fd.FD, timeReady chan struct{}, net inet.Stack, clocks sentrytime.Clocks, vfsOpts *vfs.CompleteRestoreOptions, saveRestoreNet bool) error { +// +// LoadFrom takes ownership of pagesFile. +func (k *Kernel) LoadFrom(ctx context.Context, r, pagesMetadata io.Reader, pagesFile *fd.FD, background bool, timeReady chan struct{}, net inet.Stack, clocks sentrytime.Clocks, vfsOpts *vfs.CompleteRestoreOptions, saveRestoreNet bool) error { loadStart := time.Now() + defer func() { + if pagesFile != nil { + pagesFile.Close() + } + }() var ( mfLoadWg sync.WaitGroup mfLoadErr error @@ -780,7 +786,8 @@ func (k *Kernel) LoadFrom(ctx context.Context, r, pagesMetadata io.Reader, pages mfLoadWg.Add(1) go func() { defer mfLoadWg.Done() - mfLoadErr = k.loadMemoryFiles(ctx, r, pagesMetadata, pagesFile) + mfLoadErr = k.loadMemoryFiles(ctx, r, pagesMetadata, pagesFile, background) + pagesFile = nil // transferred to k.loadMemoryFiles() }() // Defer a Wait() so we wait for k.loadMemoryFiles() to complete even if we // error out without reaching the other Wait() below. @@ -824,7 +831,8 @@ func (k *Kernel) LoadFrom(ctx context.Context, r, pagesMetadata io.Reader, pages if parallelMfLoad { mfLoadWg.Wait() } else { - mfLoadErr = k.loadMemoryFiles(ctx, r, pagesMetadata, pagesFile) + mfLoadErr = k.loadMemoryFiles(ctx, r, pagesMetadata, pagesFile, background) + pagesFile = nil // transferred to k.loadMemoryFiles() } if mfLoadErr != nil { return mfLoadErr @@ -869,28 +877,40 @@ func (k *Kernel) LoadFrom(ctx context.Context, r, pagesMetadata io.Reader, pages return nil } -func (k *Kernel) loadMemoryFiles(ctx context.Context, r, pagesMetadata io.Reader, pagesFile *fd.FD) error { - // Load the memory files' state. +// loadMemoryFiles takes ownership of pagesFile. +func (k *Kernel) loadMemoryFiles(ctx context.Context, r, pagesMetadata io.Reader, pagesFile *fd.FD, background bool) error { memoryStart := time.Now() pmr := r if pagesMetadata != nil { pmr = pagesMetadata } - var pr *statefile.AsyncReader - if pagesFile != nil { - pr = statefile.NewAsyncReader(pagesFile, 0 /* off */) - defer pr.Close() + var ( + pagesFileUsers atomicbitops.Int64 + asyncPageLoadWG sync.WaitGroup + ) + opts := pgalloc.LoadOpts{ + PagesFile: pagesFile, + OnAsyncPageLoadStart: func() { + pagesFileUsers.Add(1) + asyncPageLoadWG.Add(1) + }, + OnAsyncPageLoadDone: func(error) { + if n := pagesFileUsers.Add(-1); n == 0 { + pagesFile.Close() + } else if n < 0 { + panic("pagesFileUsers < 0") + } + asyncPageLoadWG.Done() + }, } - if err := k.mf.LoadFrom(ctx, pmr, pr); err != nil { + if err := k.mf.LoadFrom(ctx, pmr, &opts); err != nil { return err } - if err := loadPrivateMFs(ctx, pmr, pr); err != nil { + if err := loadPrivateMFs(ctx, pmr, &opts); err != nil { return err } - if pr != nil { - if err := pr.Wait(); err != nil { - return err - } + if !background { + asyncPageLoadWG.Wait() } log.Infof("Memory files load took [%s].", time.Since(memoryStart)) return nil diff --git a/pkg/sentry/pgalloc/BUILD b/pkg/sentry/pgalloc/BUILD index a0abd7683..48ac49640 100644 --- a/pkg/sentry/pgalloc/BUILD +++ b/pkg/sentry/pgalloc/BUILD @@ -8,10 +8,10 @@ package( ) declare_mutex( - name = "mappings_mutex", - out = "mappings_mutex.go", + name = "apl_shared_mutex", + out = "apl_shared_mutex.go", package = "pgalloc", - prefix = "mappings", + prefix = "aplShared", ) declare_mutex( @@ -21,6 +21,23 @@ declare_mutex( prefix = "memoryFile", ) +go_template_instance( + name = "apl_unloaded_set", + out = "apl_unloaded_set.go", + imports = { + "memmap": "gvisor.dev/gvisor/pkg/sentry/memmap", + }, + package = "pgalloc", + prefix = "aplUnloaded", + template = "//pkg/segment:generic_set", + types = { + "Key": "uint64", + "Range": "memmap.FileRange", + "Value": "aplUnloadedInfo", + "Functions": "aplUnloadedSetFunctions", + }, +) + go_template_instance( name = "evictable_range", out = "evictable_range.go", @@ -51,7 +68,6 @@ go_template_instance( out = "memacct_set.go", consts = { "minDegree": "10", - "trackGaps": "1", }, imports = { "memmap": "gvisor.dev/gvisor/pkg/sentry/memmap", @@ -112,10 +128,12 @@ go_template_instance( go_library( name = "pgalloc", srcs = [ + "apl_shared_mutex.go", + "apl_unloaded_set.go", "context.go", + "debug.go", "evictable_range.go", "evictable_range_set.go", - "mappings_mutex.go", "memacct_set.go", "memory_file_mutex.go", "pgalloc.go", @@ -127,21 +145,26 @@ go_library( visibility = ["//pkg/sentry:internal"], deps = [ "//pkg/abi/linux", + "//pkg/aio", "//pkg/atomicbitops", + "//pkg/bitmap", "//pkg/context", "//pkg/errors/linuxerr", + "//pkg/fd", + "//pkg/goid", "//pkg/hostarch", "//pkg/log", + "//pkg/ringdeque", "//pkg/safemem", "//pkg/sentry/arch", "//pkg/sentry/hostmm", "//pkg/sentry/memmap", "//pkg/sentry/usage", "//pkg/state", - "//pkg/state/statefile", "//pkg/state/wire", "//pkg/sync", "//pkg/sync/locking", + "//pkg/syncevent", "//pkg/usermem", "@org_golang_x_sys//unix:go_default_library", ], diff --git a/pkg/sentry/pgalloc/debug.go b/pkg/sentry/pgalloc/debug.go new file mode 100644 index 000000000..99faad321 --- /dev/null +++ b/pkg/sentry/pgalloc/debug.go @@ -0,0 +1,19 @@ +// Copyright 2024 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package pgalloc + +// If logAwaitedLoads is true, events relevant to awaited async page loads will +// be logged at level Info. +const logAwaitedLoads = false diff --git a/pkg/sentry/pgalloc/pgalloc.go b/pkg/sentry/pgalloc/pgalloc.go index c6bb96e84..e7d8f8dc6 100644 --- a/pkg/sentry/pgalloc/pgalloc.go +++ b/pkg/sentry/pgalloc/pgalloc.go @@ -183,6 +183,10 @@ type MemoryFile struct { // immutable. stopNotifyPressure func() + // If asyncPageLoad is non-nil, it tracks the state of in-progress or + // failed async page loading. + asyncPageLoad atomic.Pointer[aplShared] + // file is the backing file. The file pointer is immutable. file *os.File @@ -1162,7 +1166,11 @@ func (f *MemoryFile) IncRef(fr memmap.FileRange, memCgID uint32) { f.mu.Lock() defer f.mu.Unlock() + f.incRefLocked(fr) +} +// Preconditions: f.mu must be locked. +func (f *MemoryFile) incRefLocked(fr memmap.FileRange) { f.forEachChunk(fr, func(chunk *chunkInfo, chunkFR memmap.FileRange) bool { unfree := &f.unfreeSmall if chunk.huge { @@ -1219,6 +1227,10 @@ func (f *MemoryFile) DecRef(fr memmap.FileRange) { ma.wasteOrReleasing = true return true }) + // Cancel any pending async load on waste pages. + if apl := f.asyncPageLoad.Load(); apl != nil { + apl.cancelWasteLoad(wasteFR) + } } return true }) @@ -1418,6 +1430,12 @@ func (f *MemoryFile) MapInternal(fr memmap.FileRange, at hostarch.AccessType) (s return safemem.BlockSeq{}, linuxerr.EACCES } + if apl := f.asyncPageLoad.Load(); apl != nil { + if err := apl.awaitLoad(f, fr); err != nil { + return safemem.BlockSeq{}, err + } + } + chunks := ((fr.End + chunkMask) / chunkSize) - (fr.Start / chunkSize) if chunks == 1 { // Avoid an unnecessary slice allocation. @@ -1753,6 +1771,11 @@ func (f *MemoryFile) File() *os.File { // DataFD implements memmap.File.DataFD. func (f *MemoryFile) DataFD(fr memmap.FileRange) (int, error) { + if apl := f.asyncPageLoad.Load(); apl != nil { + if err := apl.awaitLoad(f, fr); err != nil { + return -1, err + } + } return f.FD(), nil } diff --git a/pkg/sentry/pgalloc/pgalloc_unsafe.go b/pkg/sentry/pgalloc/pgalloc_unsafe.go index 73a1cf140..48fb3964f 100644 --- a/pkg/sentry/pgalloc/pgalloc_unsafe.go +++ b/pkg/sentry/pgalloc/pgalloc_unsafe.go @@ -36,3 +36,11 @@ func mincore(s []byte, buf []byte, off uint64, wasCommitted bool) error { } return nil } + +func sliceFromIovec(iov unix.Iovec) []byte { + return unsafe.Slice(iov.Base, iov.Len) +} + +func canMergeIovecAndSlice(iov unix.Iovec, bs []byte) bool { + return uintptr(unsafe.Pointer(iov.Base))+uintptr(iov.Len) == uintptr(unsafe.Pointer(unsafe.SliceData(bs))) +} diff --git a/pkg/sentry/pgalloc/save_restore.go b/pkg/sentry/pgalloc/save_restore.go index 6e56e58fa..0495aa602 100644 --- a/pkg/sentry/pgalloc/save_restore.go +++ b/pkg/sentry/pgalloc/save_restore.go @@ -19,19 +19,26 @@ import ( "context" "fmt" "io" + "math" "runtime" "time" "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/aio" "gvisor.dev/gvisor/pkg/atomicbitops" + "gvisor.dev/gvisor/pkg/bitmap" + "gvisor.dev/gvisor/pkg/errors/linuxerr" + "gvisor.dev/gvisor/pkg/fd" + "gvisor.dev/gvisor/pkg/goid" "gvisor.dev/gvisor/pkg/hostarch" "gvisor.dev/gvisor/pkg/log" + "gvisor.dev/gvisor/pkg/ringdeque" "gvisor.dev/gvisor/pkg/sentry/memmap" "gvisor.dev/gvisor/pkg/sentry/usage" "gvisor.dev/gvisor/pkg/state" - "gvisor.dev/gvisor/pkg/state/statefile" "gvisor.dev/gvisor/pkg/state/wire" "gvisor.dev/gvisor/pkg/sync" + "gvisor.dev/gvisor/pkg/syncevent" ) // SaveOpts provides options to MemoryFile.SaveTo(). @@ -51,6 +58,10 @@ type SaveOpts struct { // SaveTo writes f's state to the given stream. func (f *MemoryFile) SaveTo(ctx context.Context, w io.Writer, pw io.Writer, opts SaveOpts) error { + if err := f.AwaitLoadAll(); err != nil { + return fmt.Errorf("previous async page loading failed: %w", err) + } + // Wait for memory release. f.mu.Lock() defer f.mu.Unlock() @@ -138,7 +149,7 @@ func (f *MemoryFile) SaveTo(ctx context.Context, w io.Writer, pw io.Writer, opts if err != nil { return err } - log.Debugf("MemoryFile.SaveTo: scanned %d bytes, decommitted %d bytes in %d syscalls, %s", scanTotal, decommitTotal, decommitCount, time.Since(timeScanStart)) + log.Infof("MemoryFile(%p): saving scanned %d bytes, decommitted %d bytes in %d syscalls, %s", f, scanTotal, decommitTotal, decommitCount, time.Since(timeScanStart)) // Save metadata. timeMetadataStart := time.Now() @@ -169,7 +180,7 @@ func (f *MemoryFile) SaveTo(ctx context.Context, w io.Writer, pw io.Writer, opts if _, err := state.Save(ctx, w, f.chunks.Load()); err != nil { return err } - log.Debugf("MemoryFile.SaveTo: saved metadata in %s", time.Since(timeMetadataStart)) + log.Infof("MemoryFile(%p): saved metadata in %s", f, time.Since(timeMetadataStart)) // Dump out committed pages. ww := wire.Writer{Writer: w} @@ -197,7 +208,7 @@ func (f *MemoryFile) SaveTo(ctx context.Context, w io.Writer, pw io.Writer, opts savedBytes += maseg.Range().Length() } durPages := time.Since(timePagesStart) - log.Debugf("MemoryFile.SaveTo: saved pages in %s (%d bytes, %f bytes/second)", durPages, savedBytes, float64(savedBytes)/durPages.Seconds()) + log.Infof("MemoryFile(%p): saved pages in %s (%d bytes, %f bytes/second)", f, durPages, savedBytes, float64(savedBytes)/durPages.Seconds()) return nil } @@ -221,8 +232,24 @@ func (f *MemoryFile) RestoreID() string { return f.opts.RestoreID } +// LoadOpts provides options to MemoryFile.LoadFrom(). +type LoadOpts struct { + // If PagesFile is not nil, then page contents will be read from PagesFile, + // starting at PagesFileOffset, rather than from r. If LoadFrom returns a + // nil error, it increments PagesFileOffset by the number of bytes that + // will be read out of PagesFile. PagesFile may be read even after LoadFrom + // returns; OnAsyncPageLoadStart will be called before reading from + // PagesFile begins, and OnAsyncPageLoadDone will be called after all reads + // are complete. Callers must ensure that PagesFile remains valid until + // OnAsyncPageLoadDone is called. + PagesFile *fd.FD + PagesFileOffset uint64 + OnAsyncPageLoadStart func() + OnAsyncPageLoadDone func(error) +} + // LoadFrom loads MemoryFile state from the given stream. -func (f *MemoryFile) LoadFrom(ctx context.Context, r io.Reader, pr *statefile.AsyncReader) error { +func (f *MemoryFile) LoadFrom(ctx context.Context, r io.Reader, opts *LoadOpts) error { timeMetadataStart := time.Now() // Clear sets since non-empty sets will panic if loaded into. @@ -262,7 +289,7 @@ func (f *MemoryFile) LoadFrom(ctx context.Context, r io.Reader, pr *statefile.As return err } f.chunks.Store(&chunks) - log.Debugf("MemoryFile.LoadFrom: loaded metadata in %s", time.Since(timeMetadataStart)) + log.Infof("MemoryFile(%p): loaded metadata in %s", f, time.Since(timeMetadataStart)) if err := f.file.Truncate(int64(len(chunks)) * chunkSize); err != nil { return err } @@ -305,10 +332,52 @@ func (f *MemoryFile) LoadFrom(ctx context.Context, r io.Reader, pr *statefile.As } defer madviseWG.Wait() + // Start async page loading if a pages file has been provided. + // + // Future work: In practice, all restored MemoryFiles in a given Kernel + // will share the same pages file (see Kernel.loadMemoryFiles()), so each + // MemoryFile will maintain its own AIO queue and async page loader. In + // addition to resource usage downsides, this means that awaited loads may + // not be consistently prioritized: they'll be prioritized by their + // originating MemoryFile, but may compete with unawaited loads from other + // MemoryFiles. I think the best way to fix this would be to have a single + // AIO queue and async page loader per pages file (still in this package), + // and have it schedule reads between multiple MemoryFiles. As of this + // writing, this doesn't seem to be a problem since our workloads of + // interest have relatively small root overlay MemoryFiles (the most common + // private MemoryFile). + var ( + aplg *aplGoroutine + apl *aplShared + ) + if opts.PagesFile != nil { + aplg = &aplGoroutine{ + f: f, + q: aio.NewGoQueue(aplQueueCapacity), + doneCallback: opts.OnAsyncPageLoadDone, + qavail: aplQueueCapacity, + fd: int32(opts.PagesFile.FD()), + opsBusy: bitmap.New(aplQueueCapacity), + } + apl = &aplg.apl + // Mark ops in opsBusy that don't actually exist as permanently busy. + for i, n := aplQueueCapacity, aplg.opsBusy.Size(); i < n; i++ { + aplg.opsBusy.Add(uint32(i)) + } + aplg.lfStatus.Init() + defer aplg.lfStatus.Notify(aplLFDone) + f.asyncPageLoad.Store(apl) + if opts.OnAsyncPageLoadStart != nil { + opts.OnAsyncPageLoadStart() + } + go aplg.main() + } + // Load committed pages. wr := wire.Reader{Reader: r} timePagesStart := time.Now() loadedBytes := uint64(0) + defer func() { opts.PagesFileOffset += loadedBytes }() for maseg := f.memAcct.FirstSegment(); maseg.Ok(); maseg = maseg.NextSegment() { if !maseg.ValuePtr().knownCommitted { continue @@ -322,41 +391,705 @@ func (f *MemoryFile) LoadFrom(ctx context.Context, r io.Reader, pr *statefile.As // Not expected. return fmt.Errorf("unexpected object") } - if expected := uint64(maseg.Range().Length()); length != expected { + maFR := maseg.Range() + amount := maFR.Length() + if length != amount { // Size mismatch. - return fmt.Errorf("mismatched segment: expected %d, got %d", expected, length) + return fmt.Errorf("mismatched segment: expected %d, got %d", amount, length) } // Wait for all chunks spanned by this segment to be madvised. - for madviseEnd.Load() < maseg.End() { + for madviseEnd.Load() < maFR.End { <-madviseChan } - // Read data. - var ioErr error - f.forEachMappingSlice(maseg.Range(), func(s []byte) { - if ioErr != nil { - return - } - if pr != nil { - pr.ReadAsync(s) - } else { + if apl != nil { + // Record where to read data. + apl.mu.Lock() + apl.unloaded.InsertRange(maFR, aplUnloadedInfo{ + off: opts.PagesFileOffset + loadedBytes, + }) + apl.mu.Unlock() + aplg.lfStatus.Notify(aplLFPending) + } else { + // Read data. + var ioErr error + f.forEachMappingSlice(maFR, func(s []byte) { + if ioErr != nil { + return + } _, ioErr = io.ReadFull(r, s) + }) + if ioErr != nil { + return ioErr } - }) - if ioErr != nil { - return ioErr } // Update accounting for restored pages. We need to do this here since // these segments are marked as "known committed", and will be skipped // over on accounting scans. - amount := maseg.Range().Length() loadedBytes += amount if !f.opts.DisableMemoryAccounting { usage.MemoryAccounting.Inc(amount, maseg.ValuePtr().kind, maseg.ValuePtr().memCgID) } } durPages := time.Since(timePagesStart) - log.Debugf("MemoryFile.LoadFrom: loaded pages in %s (%d bytes, %f bytes/second)", durPages, loadedBytes, float64(loadedBytes)/durPages.Seconds()) + if apl != nil { + log.Infof("MemoryFile(%p): loaded page file offsets in %s; async loading %d bytes", f, durPages, loadedBytes) + } else { + log.Infof("MemoryFile(%p): loaded pages in %s (%d bytes, %f bytes/second)", f, durPages, loadedBytes, float64(loadedBytes)/durPages.Seconds()) + } return nil } + +// aplShared holds asynchronous page loading state that is shared with +// users of the MemoryFile. +type aplShared struct { + // minUnloaded is the MemoryFile offset of the first unloaded byte. + minUnloaded atomicbitops.Uint64 + + // mu protects the following fields. + mu aplSharedMutex + + // If err is not nil, it is an error that has terminated asynchronous page + // loading. err can only be set by the async page loader goroutine, and can + // only transition from nil to non-nil once, after which it is immutable. + err error + + // unloaded tracks pages that have not been loaded. + unloaded aplUnloadedSet + + // priority contains possibly-unstarted ranges in unloaded with at least + // one waiter. + priority ringdeque.Deque[memmap.FileRange] +} + +// aplUnloadedInfo is the value type of aplShared.unloaded. +type aplUnloadedInfo struct { + // off is the offset into the pages file at which the represented pages + // begin. + off uint64 + + // started is true if a read has been enqueued for these pages. + started bool + + // waiters queues goroutines waiting for these pages to be loaded. + waiters []*aplWaiter +} + +type aplWaiter struct { + // wakeup is used by a caller of MemoryFile.awaitLoad() to block until all + // pages in fr are loaded. + wakeup syncevent.Waiter + fr memmap.FileRange + + // pending is the number of unloaded bytes that this waiter is waiting for. + pending uint64 +} + +var aplWaiterPool = sync.Pool{ + New: func() any { + var w aplWaiter + w.wakeup.Init() + return &w + }, +} + +// AwaitLoadAll blocks until async page loading has completed. If async page +// loading is not in progress, AwaitLoadAll returns immediately. +func (f *MemoryFile) AwaitLoadAll() error { + if apl := f.asyncPageLoad.Load(); apl != nil { + return apl.awaitLoad(f, memmap.FileRange{0, hostarch.PageRoundDown(uint64(math.MaxUint64))}) + } + return nil +} + +// awaitLoad blocks until data has been loaded for all pages in fr. +// +// Preconditions: At least one reference must be held on all unloaded pages in +// fr. +func (apl *aplShared) awaitLoad(f *MemoryFile, fr memmap.FileRange) error { + // Lockless fast path: + if fr.End <= apl.minUnloaded.Load() { + return nil + } + + // fr might not be page-aligned; everything else involved in async page + // loading requires page-aligned FileRanges. + fr.Start = hostarch.PageRoundDown(fr.Start) + fr.End = hostarch.MustPageRoundUp(fr.End) + + apl.mu.Lock() + if err := apl.err; err != nil { + if apl.unloaded.IsEmptyRange(fr) { + // fr is already loaded. + apl.mu.Unlock() + return nil + } + // A previous error means that fr will never be loaded. + apl.mu.Unlock() + return err + } + w := aplWaiterPool.Get().(*aplWaiter) + defer aplWaiterPool.Put(w) + w.fr = fr + w.pending = 0 + apl.unloaded.MutateRange(fr, func(ulseg aplUnloadedIterator) bool { + ul := ulseg.ValuePtr() + ulFR := ulseg.Range() + if len(ul.waiters) == 0 && !ul.started { + apl.priority.PushBack(ulFR) + if logAwaitedLoads { + log.Infof("MemoryFile(%p): prioritize %v", f, ulFR) + } + } + ul.waiters = append(ul.waiters, w) + w.pending += ulFR.Length() + return true + }) + pending := w.pending != 0 + apl.mu.Unlock() + if pending { + var startWaitTime time.Time + if logAwaitedLoads { + startWaitTime = time.Now() + log.Infof("MemoryFile(%p): awaitLoad goid %d start: %v (%d bytes)", f, goid.Get(), fr, fr.Length()) + } + w.wakeup.WaitAndAckAll() + if logAwaitedLoads { + log.Infof("MemoryFile(%p): awaitLoad goid %d waited %v: %v (%d bytes)", f, goid.Get(), time.Since(startWaitTime), fr, fr.Length()) + } + } + return apl.err +} + +const ( + // When a pages file is provided, reads from it will be issued + // asynchronously via an aio.Queue of capacity aplQueueCapacity, and each + // read will be of size aplReadMaxBytes when possible; reads may be smaller + // in some circumstances but will never be larger. + // TODO: Pass these via LoadOpts and make them flag-controlled. + aplReadMaxBytes = 256 * 1024 + aplQueueCapacity = 128 + + aplOpMaxIovecs = aplReadMaxBytes / hostarch.PageSize +) + +// aplGoroutine holds state for the async page loader goroutine. +type aplGoroutine struct { + apl aplShared + _ [hostarch.CacheLineSize]byte // padding + + f *MemoryFile // immutable + q *aio.GoQueue // immutable + doneCallback func(error) // immutable + + // lfStatus communicates state from Memory.LoadFrom() to the goroutine. + lfStatus syncevent.Waiter + + // qavail is unused capacity in q. + qavail int + + // The async page loader combines multiple loads with contiguous pages file + // offsets (the common case) into a single read, even if their + // corresponding memmap.FileRanges and mappings are discontiguous. If curOp + // is not nil, it is the current aplOp under construction, and curOpID is + // its index into ops. + curOp *aplOp + curOpID uint32 + + // fd is the host file descriptor for the pages file. + fd int32 // immutable + + // opsBusy tracks which aplOps in ops are in use (correspond to + // inflight operations or curOp). + opsBusy bitmap.Bitmap + + // ops stores all aplOps. + ops [aplQueueCapacity]aplOp +} + +// Possible events in aplGoroutine.lfStatus: +const ( + aplLFPending syncevent.Set = 1 << iota + aplLFDone +) + +// aplOp tracks async page load state corresponding to a single AIO read +// operation. +type aplOp struct { + // total is the number of bytes to be read by the operation. + total uint64 + + // end is the pages file offset at which the read ends. + end uint64 + + // frs() = frsData[:frsLen] are the MemoryFile ranges being loaded. + frsData [aplOpMaxIovecs]memmap.FileRange + frsLen uint8 + + // iovecsLen is described below, but stored here to minimize alignment + // padding. + iovecsLen uint8 + + // If tempRef is true, a temporary reference is held on pages in frs() that + // should be dropped after completion. + tempRef bool + + // iovecs() = iovecsData[:iovecsLen] contains mappings of frs(). + iovecsData [aplOpMaxIovecs]unix.Iovec +} + +func (op *aplOp) off() int64 { + return int64(op.end - op.total) +} + +func (op *aplOp) frs() []memmap.FileRange { + return op.frsData[:op.frsLen] +} + +func (op *aplOp) iovecs() []unix.Iovec { + return op.iovecsData[:op.iovecsLen] +} + +func (g *aplGoroutine) canEnqueue() bool { + return g.qavail > 0 +} + +// Preconditions: g.canEnqueue() == true. +func (g *aplGoroutine) enqueueCurOp() { + if g.qavail <= 0 { + panic("queue full") + } + op := g.curOp + if op.total == 0 { + panic("invalid read of 0 bytes") + } + if op.total > aplReadMaxBytes { + panic(fmt.Sprintf("read of %d bytes exceeds per-read limit of %d bytes", op.total, aplReadMaxBytes)) + } + + g.qavail-- + g.curOp = nil + if op.iovecsLen == 1 { + // Perform a non-vectorized read to save an indirection (and + // userspace-to-kernelspace copy) in the aio.Queue implementation. + aio.Read(g.q, uint64(g.curOpID), g.fd, op.off(), sliceFromIovec(op.iovecsData[0])) + } else { + aio.Readv(g.q, uint64(g.curOpID), g.fd, op.off(), op.iovecs()) + } + if logAwaitedLoads && !op.tempRef { + log.Infof("MemoryFile(%p): awaited opid %d start, read %d bytes: %v", g.f, g.curOpID, op.total, op.frs()) + } +} + +// Preconditions: +// - g.canEnqueue() == true. +// - fr.Length() > 0. +// - fr must be page-aligned. +func (g *aplGoroutine) enqueueRange(fr memmap.FileRange, off uint64, tempRef bool) uint64 { + for { + if g.curOp == nil { + id, err := g.opsBusy.FirstZero(0) + if err != nil { + panic(fmt.Sprintf("all ops busy with qavail=%d: %v", g.qavail, err)) + } + g.opsBusy.Add(id) + op := &g.ops[id] + op.total = 0 + op.frsLen = 0 + op.iovecsLen = 0 + g.curOp = op + g.curOpID = id + } + n := g.combine(fr, off, tempRef) + if n > 0 { + return n + } + // Flush the existing (conflicting) op and try again with a new one. + g.enqueueCurOp() + if !g.canEnqueue() { + return 0 + } + } +} + +// combine adds as much of the given load as possible to g.curOp and returns +// the number of bytes added. +// +// Preconditions: +// - fr.Length() > 0. +// - fr must be page-aligned. +// +// Postconditions: +// - combine() never returns (0, false). +func (g *aplGoroutine) combine(fr memmap.FileRange, off uint64, tempRef bool) uint64 { + op := g.curOp + if op.total != 0 { + if op.end != off { + // Non-contiguous in the pages file. + return 0 + } + if int(op.frsLen) == len(op.frsData) && op.frsData[op.frsLen-1].End != fr.Start { + // Non-contiguous in the MemoryFile, and we're out of space for + // FileRanges. + return 0 + } + if op.tempRef != tempRef { + // Incompatible reference-counting semantics. We could handle this + // by making tempRef per-FileRange, but it's very unlikely that an + // awaited load (tempRef=false) will happen to be followed by an + // unawaited load (tempRef=true) at the correct offset. + return 0 + } + } + + // Apply direct length limits. + n := fr.Length() + if op.total+n >= aplReadMaxBytes { + n = aplReadMaxBytes - op.total + } + if n == 0 { + return 0 + } + fr.End = fr.Start + n + + // Collect iovecs, which may further limit length. + n = 0 + g.f.forEachMappingSlice(fr, func(bs []byte) { + if op.iovecsLen > 0 { + if canMergeIovecAndSlice(op.iovecsData[op.iovecsLen-1], bs) { + op.iovecsData[op.iovecsLen-1].Len += uint64(len(bs)) + n += uint64(len(bs)) + return + } + if int(op.iovecsLen) == len(op.iovecsData) { + return + } + } + op.iovecsData[op.iovecsLen].Base = &bs[0] + op.iovecsData[op.iovecsLen].SetLen(len(bs)) + op.iovecsLen++ + n += uint64(len(bs)) + }) + if n == 0 { + return 0 + } + fr.End = fr.Start + n + + // With the length decided, finish updating op. + if op.total == 0 { + op.end = off + } + op.end += n + op.total += n + op.tempRef = tempRef + if op.frsLen > 0 && op.frsData[op.frsLen-1].End == fr.Start { + op.frsData[op.frsLen-1].End = fr.End + } else { + op.frsData[op.frsLen] = fr + op.frsLen++ + } + return n +} + +func (g *aplGoroutine) main() { + apl := &g.apl + f := g.f + q := g.q + defer func() { + // Destroy q first since this synchronously stops inflight I/O. + q.Destroy() + // Wake up any remaining waiters so that they can observe apl.err. + // Leave all segments in unloaded so that new callers of + // f.awaitLoad(apl) will still observe the correct (permanently + // unloaded) segments. + apl.mu.Lock() + for ulseg := apl.unloaded.FirstSegment(); ulseg.Ok(); ulseg = ulseg.NextSegment() { + ul := ulseg.ValuePtr() + ullen := ulseg.Range().Length() + for _, w := range ul.waiters { + w.pending -= ullen + if w.pending == 0 { + w.wakeup.Notify(1) + } + } + ul.waiters = nil + } + apl.mu.Unlock() + if g.doneCallback != nil { + g.doneCallback(apl.err) + } + }() + + minUnstarted := uint64(0) + + // Storage reused between main loop iterations: + var completions []aio.Completion + var wakeups []*aplWaiter + var decRefs []memmap.FileRange + + dropDelayedDecRefs := func() { + if len(decRefs) != 0 { + for _, fr := range decRefs { + f.DecRef(fr) + } + decRefs = decRefs[:0] + } + } + defer dropDelayedDecRefs() + + timeStart := time.Now() + loadedBytes := uint64(0) + log.Debugf("MemoryFile(%p): async page loading started", f) + for { + // Enqueue as many reads as possible. + if !g.canEnqueue() { + panic("main loop invariant failed") + } + // Prioritize reading pages with waiters. + apl.mu.Lock() + for g.canEnqueue() && !apl.priority.Empty() { + fr := apl.priority.PopFront() + // All pages in apl.priority have non-zero waiters and were split + // around fr by f.awaitLoad(), and apl.unloaded never merges + // segments with waiters. Thus, we don't need to split around fr + // again, and fr.Intersect(ulseg.Range()) == ulseg.Range(). + ulseg := apl.unloaded.LowerBoundSegment(fr.Start) + for ulseg.Ok() && ulseg.Start() < fr.End { + ul := ulseg.ValuePtr() + ulFR := ulseg.Range() + if ul.started { + fr.Start = ulFR.End + ulseg = ulseg.NextSegment() + continue + } + // Awaited pages are guaranteed to have a reference held (by + // f.awaitLoad() precondition), so they can't become waste (which would + // allow them to be racily released or recycled). + n := g.enqueueRange(ulFR, ul.off, false /* tempRef */) + if n == 0 { + // Try again in the next iteration of the main loop, when + // we have space in the queue again. + apl.priority.PushFront(fr) + break + } + ulFR.End = ulFR.Start + n + ulseg = apl.unloaded.SplitAfter(ulseg, ulFR.End) + ulseg.ValuePtr().started = true + fr.Start = ulFR.End + if fr.Length() > 0 { + // Cycle the rest of fr to the end of apl.priority. This + // prevents large awaited reads from starving other + // waiters. + apl.priority.PushBack(fr) + break + } + ulseg = ulseg.NextSegment() + } + } + apl.mu.Unlock() + // Fill remaining queue with reads for pages with no waiters. + if g.canEnqueue() { + f.mu.Lock() + apl.mu.Lock() + ulseg := apl.unloaded.LowerBoundSegment(minUnstarted) + for ulseg.Ok() { + ul := ulseg.ValuePtr() + ulFR := ulseg.Range() + if ul.started { + minUnstarted = ulFR.End + ulseg = ulseg.NextSegment() + continue + } + // We need to take page references during reading to prevent + // pages from becoming waste due to concurrent dropping of the + // last reference. + n := g.enqueueRange(ulFR, ul.off, true /* tempRef */) + if n == 0 { + break + } + ulFR.End = ulFR.Start + n + ulseg = apl.unloaded.SplitAfter(ulseg, ulFR.End) + ulseg.ValuePtr().started = true + minUnstarted = ulFR.End + f.incRefLocked(ulFR) + if !g.canEnqueue() { + break + } + ulseg = ulseg.NextSegment() + } + apl.mu.Unlock() + f.mu.Unlock() + } + // Flush pending op. + if g.curOp != nil { + g.enqueueCurOp() + } + + if g.qavail == q.Cap() { + // We are out of work to do. + ev := g.lfStatus.Wait() + if ev&aplLFPending != 0 { + // We may have raced with MemoryFile.LoadFrom() inserting into + // apl.unloaded. + g.lfStatus.Ack(aplLFPending) + continue + } + if ev&aplLFDone != 0 { + // MemoryFile.LoadFrom() finished inserting into apl.unloaded, + // so async page loading has completed successfully. + apl.minUnloaded.Store(math.MaxUint64) + f.asyncPageLoad.Store(nil) + dur := time.Since(timeStart) + log.Infof("MemoryFile(%p): async page loading completed in %s (%d bytes, %f bytes/second)", f, dur, loadedBytes, float64(loadedBytes)/dur.Seconds()) + return + } + panic(fmt.Sprintf("unknown events in lfStatus: %#x", ev)) + } + + // Wait for any number of reads to complete. + var err error + completions, err = q.Wait(completions[:0], 1 /* minCompletions */) + if err != nil { + log.Warningf("MemoryFile(%p): async page loading: aio.Queue.Wait failed: %v", f, err) + apl.mu.Lock() + apl.err = linuxerr.EIO + apl.mu.Unlock() + return + } + + // Process completions. + apl.mu.Lock() + for _, c := range completions { + op := g.ops[c.ID] + g.opsBusy.Remove(uint32(c.ID)) + g.qavail++ + if op.tempRef { + // Delay f.DecRef(fr) until after dropping locks. This is + // required to avoid lock recursion via dropping the last + // reference => apl.cancelWasteLoad() => apl.mu.Lock(). + decRefs = append(decRefs, op.frs()...) + } + if err := c.Err(); err != nil { + log.Warningf("MemoryFile(%p): async page loading: read for pages %v failed: %v", f, op.frs(), err) + apl.err = err + apl.mu.Unlock() + return + } + if uint64(c.Result) != op.total { + // TODO: Is this something we actually have to worry about? If + // so, we need to reissue the remainder of the read... + log.Warningf("MemoryFile(%p): async page loading: read for pages %v (total %d bytes) returned %d bytes", f, op.frs(), op.total, c.Result) + apl.err = linuxerr.EIO + apl.mu.Unlock() + return + } + haveWaiters := false + for _, fr := range op.frs() { + // All pages in fr have been started and were split around fr + // when they were started (above), and apl.unloaded never + // merges started segments. Thus, we don't need to split around + // fr again, and fr.Intersect(ulseg.Range()) == ulseg.Range(). + for ulseg := apl.unloaded.FindSegment(fr.Start); ulseg.Ok() && ulseg.Start() < fr.End; ulseg = apl.unloaded.Remove(ulseg).NextSegment() { + ul := ulseg.ValuePtr() + ullen := ulseg.Range().Length() + loadedBytes += ullen + if !ul.started { + panic(fmt.Sprintf("completion of %v includes pages %v that were never started", fr, ulseg.Range())) + } + for _, w := range ul.waiters { + haveWaiters = true + w.pending -= ullen + if w.pending == 0 { + wakeups = append(wakeups, w) + } + } + } + } + if logAwaitedLoads && haveWaiters { + log.Infof("MemoryFile(%p): awaited opid %d complete, read %d bytes: %v", g.f, c.ID, op.total, op.frs()) + } + } + // Keep apl.minUnloaded up to date. We can only determine this + // accurately if insertions into apl.unloaded are complete. + if g.lfStatus.Pending()&aplLFDone != 0 { + if apl.unloaded.IsEmpty() { + apl.minUnloaded.Store(math.MaxUint64) + } else { + apl.minUnloaded.Store(apl.unloaded.FirstSegment().Start()) + } + } + apl.mu.Unlock() + for _, w := range wakeups { + w.wakeup.Notify(1) + } + wakeups = wakeups[:0] + dropDelayedDecRefs() + } +} + +// Preconditions: +// - All pages in fr must be becoming waste pages. +// - fr must be page-aligned. +func (apl *aplShared) cancelWasteLoad(fr memmap.FileRange) { + // Lockless fast path: + if fr.End <= apl.minUnloaded.Load() { + return + } + + apl.mu.Lock() + defer apl.mu.Unlock() + apl.unloaded.RemoveRangeWith(fr, func(ulseg aplUnloadedIterator) { + ul := ulseg.ValuePtr() + if ul.started { + // This shouldn't be possible since page references are held while + // reading (see MemoryFile.asyncPageLoadMain()). + panic(fmt.Sprintf("pages %v becoming waste during inflight read from async loading", ulseg.Range())) + } + if n := len(ul.waiters); n != 0 { + // This shouldn't be possible since the waiters should hold page + // references. + panic(fmt.Sprintf("pages %v becoming waste with %d async load waiters", ulseg.Range(), n)) + } + }) +} + +type aplUnloadedSetFunctions struct{} + +func (aplUnloadedSetFunctions) MinKey() uint64 { + return 0 +} + +func (aplUnloadedSetFunctions) MaxKey() uint64 { + return math.MaxUint64 +} + +func (aplUnloadedSetFunctions) ClearValue(ul *aplUnloadedInfo) { + ul.waiters = nil +} + +func (aplUnloadedSetFunctions) Merge(fr1 memmap.FileRange, ul1 aplUnloadedInfo, fr2 memmap.FileRange, ul2 aplUnloadedInfo) (aplUnloadedInfo, bool) { + if ul1.off+fr1.Length() != ul2.off { + return aplUnloadedInfo{}, false + } + if ul1.started || ul2.started || len(ul1.waiters) != 0 || len(ul2.waiters) != 0 { + // Merging would be counterproductive, since we expect that these + // segments will shortly be removed (separately) based on AIO + // completions, which would just necessitate splitting again. + return aplUnloadedInfo{}, false + } + return ul1, true +} + +func (aplUnloadedSetFunctions) Split(fr memmap.FileRange, ul aplUnloadedInfo, splitAt uint64) (aplUnloadedInfo, aplUnloadedInfo) { + ul2 := aplUnloadedInfo{ + off: ul.off + (splitAt - fr.Start), + started: ul.started, + // Setting cap(ul2.waiters) == len(ul2.waiters) makes ul2 + // "copy-on-append", saving an allocation if ul2 is never appended-to. + // This is safe since existing elements in ul.waiters will never be + // mutated. + waiters: ul.waiters[:len(ul.waiters):len(ul.waiters)], + } + return ul, ul2 +} diff --git a/pkg/sentry/state/state.go b/pkg/sentry/state/state.go index ffd465a32..4f82ca21d 100644 --- a/pkg/sentry/state/state.go +++ b/pkg/sentry/state/state.go @@ -144,12 +144,25 @@ type LoadOpts struct { // PagesFile is non-nil. Otherwise this content is stored in Source. PagesFile *fd.FD + // If Background is true, the sentry may read from PagesFile after Load has + // returned. + Background bool + // Key is used for state integrity check. Key []byte } // Load loads the given kernel, setting the provided platform and stack. +// +// Load takes ownership of (and unsets) opts.PagesFile. func (opts LoadOpts) Load(ctx context.Context, k *kernel.Kernel, timeReady chan struct{}, n inet.Stack, clocks time.Clocks, vfsOpts *vfs.CompleteRestoreOptions, saveRestoreNet bool) error { + defer func() { + if opts.PagesFile != nil { + opts.PagesFile.Close() + opts.PagesFile = nil + } + }() + // Open the file. r, m, err := statefile.NewReader(opts.Source, opts.Key) if err != nil { @@ -167,5 +180,7 @@ func (opts LoadOpts) Load(ctx context.Context, k *kernel.Kernel, timeReady chan previousMetadata = m // Restore the Kernel object graph. - return k.LoadFrom(ctx, r, pagesMetadata, opts.PagesFile, timeReady, n, clocks, vfsOpts, saveRestoreNet) + err = k.LoadFrom(ctx, r, pagesMetadata, opts.PagesFile, opts.Background, timeReady, n, clocks, vfsOpts, saveRestoreNet) + opts.PagesFile = nil // transferred to k.LoadFrom() + return err } diff --git a/pkg/shim/extension/extension.go b/pkg/shim/extension/extension.go index c0ec2ad4c..753f684cf 100644 --- a/pkg/shim/extension/extension.go +++ b/pkg/shim/extension/extension.go @@ -42,8 +42,9 @@ type Process interface { // RestoreConfig is the configuration for a restore request. type RestoreConfig struct { - ImagePath string - Direct bool + ImagePath string + Direct bool + Background bool } // TaskServiceExt extends TaskRequest with extra functionality required by the shim. diff --git a/pkg/shim/proc/init.go b/pkg/shim/proc/init.go index ac55f7e12..a10b97d25 100644 --- a/pkg/shim/proc/init.go +++ b/pkg/shim/proc/init.go @@ -240,9 +240,10 @@ func (p *Init) start(ctx context.Context, restoreConf *extension.RestoreConfig) } } else { if err := p.runtime.Restore(ctx, p.id, cio, &runsccmd.RestoreOpts{ - ImagePath: restoreConf.ImagePath, - Detach: true, - Direct: restoreConf.Direct, + ImagePath: restoreConf.ImagePath, + Detach: true, + Direct: restoreConf.Direct, + Background: restoreConf.Background, }); err != nil { return p.runtimeError(err, "OCI runtime restore failed") } diff --git a/pkg/shim/runsccmd/runsc.go b/pkg/shim/runsccmd/runsc.go index e401096ef..8fe2530c5 100644 --- a/pkg/shim/runsccmd/runsc.go +++ b/pkg/shim/runsccmd/runsc.go @@ -233,9 +233,10 @@ func (r *Runsc) start(context context.Context, cio runc.IO, cmd *exec.Cmd) error // RestoreOpts is a set of options to runsc.Restore(). type RestoreOpts struct { - ImagePath string - Detach bool - Direct bool + ImagePath string + Detach bool + Direct bool + Background bool } func (o *RestoreOpts) args() []string { @@ -249,6 +250,9 @@ func (o *RestoreOpts) args() []string { if o.Direct { out = append(out, "--direct") } + if o.Background { + out = append(out, "--background") + } return out } diff --git a/pkg/state/statefile/BUILD b/pkg/state/statefile/BUILD index bb56eee20..07c4c4437 100644 --- a/pkg/state/statefile/BUILD +++ b/pkg/state/statefile/BUILD @@ -8,28 +8,18 @@ package( go_library( name = "statefile", srcs = [ - "async_io.go", "statefile.go", ], visibility = ["//:sandbox"], - deps = [ - "//pkg/compressio", - "//pkg/fd", - "//pkg/sync", - ], + deps = ["//pkg/compressio"], ) go_test( name = "statefile_test", size = "small", srcs = [ - "async_io_test.go", "statefile_test.go", ], library = ":statefile", - deps = [ - "//pkg/compressio", - "//pkg/fd", - "@org_golang_x_sys//unix:go_default_library", - ], + deps = ["//pkg/compressio"], ) diff --git a/pkg/state/statefile/async_io.go b/pkg/state/statefile/async_io.go deleted file mode 100644 index 78b655047..000000000 --- a/pkg/state/statefile/async_io.go +++ /dev/null @@ -1,94 +0,0 @@ -// Copyright 2024 The gVisor Authors. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package statefile - -import ( - "runtime" - "sync/atomic" - - "gvisor.dev/gvisor/pkg/fd" - "gvisor.dev/gvisor/pkg/sync" -) - -type chunk struct { - dst []byte - off int64 -} - -// AsyncReader can be used to do reads asynchronously. It does not change the -// underlying file's offset. -type AsyncReader struct { - // in is the backing file which contains all pages. - in *fd.FD - // off is the offset being read. - off int64 - // q is the work queue. - q chan chunk - // err stores the latest IO error that occured during async read. - err atomic.Pointer[error] - // wg tracks all in flight work. - wg sync.WaitGroup -} - -// NewAsyncReader initializes a new AsyncReader. -func NewAsyncReader(in *fd.FD, off int64) *AsyncReader { - workers := runtime.GOMAXPROCS(0) - r := &AsyncReader{ - in: in, - off: off, - q: make(chan chunk, workers), - } - for i := 0; i < workers; i++ { - go r.work() - } - return r -} - -// ReadAsync schedules a read of len(p) bytes from current offset into p. -func (r *AsyncReader) ReadAsync(p []byte) { - r.wg.Add(1) - r.q <- chunk{off: r.off, dst: p} - r.off += int64(len(p)) -} - -// Wait blocks until all in flight work is complete and then returns any IO -// errors that occurred since the last call to Wait(). -func (r *AsyncReader) Wait() error { - r.wg.Wait() - if err := r.err.Swap(nil); err != nil { - return *err - } - return nil -} - -// Close calls Wait() and additionally cleans up all worker goroutines. -func (r *AsyncReader) Close() error { - err := r.Wait() - close(r.q) - return err -} - -func (r *AsyncReader) work() { - for { - c := <-r.q - if c.dst == nil { - return - } - if _, err := r.in.ReadAt(c.dst, c.off); err != nil { - r.err.Store(&err) - } - r.wg.Done() - } -} diff --git a/pkg/state/statefile/async_io_test.go b/pkg/state/statefile/async_io_test.go deleted file mode 100644 index 84adc1693..000000000 --- a/pkg/state/statefile/async_io_test.go +++ /dev/null @@ -1,64 +0,0 @@ -// Copyright 2024 The gVisor Authors. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package statefile - -import ( - "bytes" - "math/rand" - "os" - "testing" - - "golang.org/x/sys/unix" - "gvisor.dev/gvisor/pkg/fd" -) - -func TestAsyncReader(t *testing.T) { - // Create random data. - const chunkSize = 4096 - const dataLen = 1024 * chunkSize - data := make([]byte, dataLen) - _, _ = rand.Read(data) - - // Create a temp file with the data. - testFile, err := os.CreateTemp(t.TempDir(), "source") - if err != nil { - t.Fatalf("failed to create temp source file: %v", err) - } - if _, err := testFile.Write(data); err != nil { - t.Fatalf("failed to write temp source file: %v", err) - } - testFilePath := testFile.Name() - if err := testFile.Close(); err != nil { - t.Fatalf("failed to close temp source file: %v", err) - } - - // Read the data from the file using async reads. - sourceFD, err := fd.Open(testFilePath, unix.O_RDONLY, 0) - if err != nil { - t.Fatalf("failed to open source file %q: %v", testFilePath, err) - } - ar := NewAsyncReader(sourceFD, 0 /* off */) - defer ar.Close() - p := make([]byte, dataLen) - for i := 0; i < dataLen; i += chunkSize { - ar.ReadAsync(p[i : i+chunkSize]) - } - if err := ar.Wait(); err != nil { - t.Fatalf("AsyncReader.Wait failed: %v", err) - } - if ret := bytes.Compare(p, data); ret != 0 { - t.Errorf("bytes differ") - } -} diff --git a/runsc/boot/controller.go b/runsc/boot/controller.go index d4a6fff2c..d43e6cc23 100644 --- a/runsc/boot/controller.go +++ b/runsc/boot/controller.go @@ -488,6 +488,7 @@ type RestoreOpts struct { urpc.FilePayload HavePagesFile bool HaveDeviceFile bool + Background bool } // Restore loads a container from a statefile. @@ -524,7 +525,11 @@ func (cm *containerManager) Restore(o *RestoreOpts, _ *struct{}) error { return fmt.Errorf("statefile cannot be empty") } - cm.restorer = &restorer{restoreDone: cm.onRestoreDone, stateFile: stateFile} + cm.restorer = &restorer{ + restoreDone: cm.onRestoreDone, + stateFile: stateFile, + background: o.Background, + } cm.l.restoreWaiters = sync.NewCond(&cm.l.mu) cm.l.state = restoring // Release `cm.l.mu`. diff --git a/runsc/boot/restore.go b/runsc/boot/restore.go index 9389a1d59..eec2cff99 100644 --- a/runsc/boot/restore.go +++ b/runsc/boot/restore.go @@ -76,6 +76,10 @@ type restorer struct { pagesMetadata *fd.FD pagesFile *fd.FD + // If background is true, pagesFile may continue to be read after + // restorer.restore() returns. + background bool + // deviceFile is the required to start the platform. deviceFile *fd.FD @@ -231,8 +235,15 @@ func (r *restorer) restore(l *Loader) error { ctx = context.WithValue(ctx, devutil.CtxDevGoferClientProvider, l.k) // Load the state. - loadOpts := state.LoadOpts{Source: r.stateFile, PagesMetadata: r.pagesMetadata, PagesFile: r.pagesFile} - if err := loadOpts.Load(ctx, l.k, nil, oldInetStack, time.NewCalibratedClocks(), &vfs.CompleteRestoreOptions{}, l.saveRestoreNet); err != nil { + loadOpts := state.LoadOpts{ + Source: r.stateFile, + PagesMetadata: r.pagesMetadata, + PagesFile: r.pagesFile, + Background: r.background, + } + err = loadOpts.Load(ctx, l.k, nil, oldInetStack, time.NewCalibratedClocks(), &vfs.CompleteRestoreOptions{}, l.saveRestoreNet) + r.pagesFile = nil // transferred to loadOpts.Load() + if err != nil { return err } diff --git a/runsc/cmd/restore.go b/runsc/cmd/restore.go index bc04e2e56..0ecdb697c 100644 --- a/runsc/cmd/restore.go +++ b/runsc/cmd/restore.go @@ -46,6 +46,13 @@ type Restore struct { // network block device). Usually the restore is done only once, so the cost // of adding the checkpoint files to the page cache can be redundant. direct bool + + // If background is true, the container image may continue to be read after + // the restore command exits. For large images, this significantly shortens + // the amount of time taken by the restore command. The checkpoint must be + // uncompressed for background to work; if the checkpoint is compressed, + // background has no effect. + background bool } // Name implements subcommands.Command.Name. @@ -70,6 +77,7 @@ func (r *Restore) SetFlags(f *flag.FlagSet) { f.StringVar(&r.imagePath, "image-path", "", "directory path to saved container image") f.BoolVar(&r.detach, "detach", false, "detach from the container's process") f.BoolVar(&r.direct, "direct", false, "use O_DIRECT for reading checkpoint pages file") + f.BoolVar(&r.background, "background", false, "allow image loading to continue after restore exits (requires uncompressed checkpoint)") // Unimplemented flags necessary for compatibility with docker. @@ -146,7 +154,7 @@ func (r *Restore) Execute(_ context.Context, f *flag.FlagSet, args ...any) subco } log.Debugf("Restore: %v", r.imagePath) - if err := c.Restore(conf, r.imagePath, r.direct); err != nil { + if err := c.Restore(conf, r.imagePath, r.direct, r.background); err != nil { return util.Errorf("starting container: %v", err) } diff --git a/runsc/container/container.go b/runsc/container/container.go index 8429c9b6d..318d9d659 100644 --- a/runsc/container/container.go +++ b/runsc/container/container.go @@ -433,11 +433,11 @@ func (c *Container) Start(conf *config.Config) error { // Restore takes a container and replaces its kernel and file system // to restore a container from its state file. -func (c *Container) Restore(conf *config.Config, imagePath string, direct bool) error { +func (c *Container) Restore(conf *config.Config, imagePath string, direct, background bool) error { log.Debugf("Restore container, cid: %s", c.ID) restore := func(conf *config.Config) error { - return c.Sandbox.Restore(conf, c.ID, imagePath, direct) + return c.Sandbox.Restore(conf, c.ID, imagePath, direct, background) } return c.startImpl(conf, "restore", restore, c.Sandbox.RestoreSubcontainer) } diff --git a/runsc/container/container_test.go b/runsc/container/container_test.go index 139b416dd..6f88b50d9 100644 --- a/runsc/container/container_test.go +++ b/runsc/container/container_test.go @@ -1103,7 +1103,7 @@ func testCheckpointRestore(t *testing.T, conf *config.Config, compression statef } defer cont2.Destroy() - if err := cont2.Restore(conf, dir, false /* direct */); err != nil { + if err := cont2.Restore(conf, dir, false /* direct */, false /* background */); err != nil { t.Fatalf("error restoring container: %v", err) } @@ -1146,7 +1146,7 @@ func testCheckpointRestore(t *testing.T, conf *config.Config, compression statef } defer cont3.Destroy() - if err := cont3.Restore(conf, dir, false /* direct */); err != nil { + if err := cont3.Restore(conf, dir, false /* direct */, false /* background */); err != nil { t.Fatalf("error restoring container: %v", err) } @@ -1264,7 +1264,7 @@ func TestCheckpointRestoreExecKilled(t *testing.T) { } defer cont2.Destroy() - if err := cont2.Restore(conf, dir, false /* direct */); err != nil { + if err := cont2.Restore(conf, dir, false /* direct */, false /* background */); err != nil { t.Fatalf("error restoring container: %v", err) } @@ -1349,7 +1349,7 @@ func TestCheckpointRestoreCreateMountPoint(t *testing.T) { } defer cont2.Destroy() - if err := cont2.Restore(conf, dir, false /* direct */); err != nil { + if err := cont2.Restore(conf, dir, false /* direct */, false /* background */); err != nil { t.Fatalf("error restoring container: %v", err) } @@ -1461,7 +1461,7 @@ func TestUnixDomainSockets(t *testing.T) { } defer contRestore.Destroy() - if err := contRestore.Restore(conf, dir, false /* direct */); err != nil { + if err := contRestore.Restore(conf, dir, false /* direct */, false /* background */); err != nil { t.Fatalf("error restoring container: %v", err) } @@ -2811,7 +2811,7 @@ func TestUsageFD(t *testing.T) { } defer cont2.Destroy() - if err := cont2.Restore(conf, dir, false /* direct */); err != nil { + if err := cont2.Restore(conf, dir, false /* direct */, false /* background */); err != nil { t.Fatalf("error restoring container: %v", err) } diff --git a/runsc/container/multi_container_test.go b/runsc/container/multi_container_test.go index b09a4b379..ada560b48 100644 --- a/runsc/container/multi_container_test.go +++ b/runsc/container/multi_container_test.go @@ -131,7 +131,7 @@ func restoreContainers(conf *config.Config, specs []*specs.Spec, ids []string, i cu.Add(func() { cont.Destroy() }) containers = append(containers, cont) - if err := cont.Restore(conf, imagePath, false /* direct */); err != nil { + if err := cont.Restore(conf, imagePath, false /* direct */, false /* background */); err != nil { return nil, nil, fmt.Errorf("error restoring container: %v", err) } diff --git a/runsc/sandbox/sandbox.go b/runsc/sandbox/sandbox.go index 4ce9c67ce..7a2350599 100644 --- a/runsc/sandbox/sandbox.go +++ b/runsc/sandbox/sandbox.go @@ -461,7 +461,7 @@ func (s *Sandbox) StartSubcontainer(spec *specs.Spec, conf *config.Config, cid s } // Restore sends the restore call for a container in the sandbox. -func (s *Sandbox) Restore(conf *config.Config, cid string, imagePath string, direct bool) error { +func (s *Sandbox) Restore(conf *config.Config, cid string, imagePath string, direct, background bool) error { log.Debugf("Restore sandbox %q from path %q", s.ID, imagePath) stateFileName := path.Join(imagePath, boot.CheckpointStateFileName) @@ -475,6 +475,7 @@ func (s *Sandbox) Restore(conf *config.Config, cid string, imagePath string, dir FilePayload: urpc.FilePayload{ Files: []*os.File{sf}, }, + Background: background, } // If the pages file exists, we must pass it in.