From 41f01d8f9c5aee4f7a31ec6183fb50bbc6f9b851 Mon Sep 17 00:00:00 2001 From: Jamie Liu Date: Thu, 26 Sep 2024 15:46:56 -0700 Subject: [PATCH] pgalloc: integrate async page loading When a pages file is provided to `runsc restore`, reads from that file are asynchronous (via statefile.AsyncReader) in order to maximize throughput. However, all such reads must complete before Kernel.LoadFrom() returns, so applications cannot execute before MemoryFile loading is complete. The main objective of this CL is to allow reads to continue after Kernel.LoadFrom() returns, allowing applications to execute while MemoryFile loading is still in progress. This behavior is user-visible: it affects whether deleting the pages file frees disk space immediately on POSIX filesystems, may affect whether deletion is possible on non-POSIX filesystems, and prevents unmounting regardless. Thus it is flag-guarded as `runsc restore --background`. MemoryFile ranges that have yet to be loaded, but that are being waited-for by applications, should be prioritized over ranges for which no application is waiting. This requires that application requests for data (calls to MemoryFile.(memmap.File).DataFD/MapInternal()) are able to determine which ranges have not yet been loaded, request reads for such ranges with elevated priority, and wait for only those reads to be completed; none of these are supported by the existing statefile.AsyncReader. Thus: - Add //pkg/sentry/pgalloc/aio, which provides an async I/O API that is designed to be easily implementable using a goroutine pool, Linux native AIO, or io_uring, though only includes a goroutine pool implementation. (io_uring is widely disabled due to security vulnerabilities. In my testing, Linux native AIO is slower than the goroutine pool, but this may change with lower GOMAXPROCS which needs further testing.) - Move I/O scheduling into pgalloc: introduce an async page loader goroutine that is started by MemoryFile.LoadFrom() when async page loading is requested (implicitly, via the existence of a pages file), which is responsible for driving submission of read requests and handling their completions. PiperOrigin-RevId: 679321884 --- pkg/aio/BUILD | 30 + pkg/aio/aio.go | 185 ++++++ pkg/aio/aio_test.go | 280 +++++++++ pkg/aio/aio_unsafe.go | 93 +++ pkg/sentry/kernel/BUILD | 1 - pkg/sentry/kernel/kernel.go | 56 +- pkg/sentry/pgalloc/BUILD | 35 +- pkg/sentry/pgalloc/debug.go | 19 + pkg/sentry/pgalloc/pgalloc.go | 23 + pkg/sentry/pgalloc/pgalloc_unsafe.go | 8 + pkg/sentry/pgalloc/save_restore.go | 779 +++++++++++++++++++++++- pkg/sentry/state/state.go | 17 +- pkg/shim/extension/extension.go | 5 +- pkg/shim/proc/init.go | 7 +- pkg/shim/runsccmd/runsc.go | 10 +- pkg/state/statefile/BUILD | 14 +- pkg/state/statefile/async_io.go | 94 --- pkg/state/statefile/async_io_test.go | 64 -- runsc/boot/controller.go | 7 +- runsc/boot/restore.go | 15 +- runsc/cmd/restore.go | 10 +- runsc/container/container.go | 4 +- runsc/container/container_test.go | 12 +- runsc/container/multi_container_test.go | 2 +- runsc/sandbox/sandbox.go | 3 +- 25 files changed, 1532 insertions(+), 241 deletions(-) create mode 100644 pkg/aio/BUILD create mode 100644 pkg/aio/aio.go create mode 100644 pkg/aio/aio_test.go create mode 100644 pkg/aio/aio_unsafe.go create mode 100644 pkg/sentry/pgalloc/debug.go delete mode 100644 pkg/state/statefile/async_io.go delete mode 100644 pkg/state/statefile/async_io_test.go 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.