Align Context API with kernel internals.

This change adapts the existing context to use more suitable non-channel-based
methods. This is a requisite for migrating the kernel internals to a
sleeper-based notification mechanism.

The last uses of amutex outside those migrated as part of this change were
dropped in a previous change. Since amutex depends on the channel-based
implementation, this package is also deleted as part of this change.

PiperOrigin-RevId: 415189675

pkg/amutex/BUILD

@@ -1,21 +0,0 @@
-load("//tools:defs.bzl", "go_library", "go_test")
-
-package(licenses = ["notice"])
-
-go_library(
-    name = "amutex",
-    srcs = ["amutex.go"],
-    visibility = ["//:sandbox"],
-    deps = [
-        "//pkg/context",
-        "//pkg/errors/linuxerr",
-    ],
-)
-
-go_test(
-    name = "amutex_test",
-    size = "small",
-    srcs = ["amutex_test.go"],
-    library = ":amutex",
-    deps = ["//pkg/sync"],
-)

pkg/amutex/amutex.go

@@ -1,113 +0,0 @@
-// Copyright 2018 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 amutex provides the implementation of an abortable mutex. It allows
-// the Lock() function to be canceled while it waits to acquire the mutex.
-package amutex
-
-import (
-	"sync/atomic"
-
-	"gvisor.dev/gvisor/pkg/context"
-	"gvisor.dev/gvisor/pkg/errors/linuxerr"
-)
-
-// Sleeper must be implemented by users of the abortable mutex to allow for
-// cancellation of waits.
-type Sleeper = context.ChannelSleeper
-
-// NoopSleeper is a stateless no-op implementation of Sleeper for anonymous
-// embedding in other types that do not support cancelation.
-type NoopSleeper = context.Context
-
-// Block blocks until either receiving from ch succeeds (in which case it
-// returns nil) or sleeper is interrupted (in which case it returns
-// linuxerr.ErrInterrupted).
-func Block(sleeper Sleeper, ch <-chan struct{}) error {
-	cancel := sleeper.SleepStart()
-	select {
-	case <-ch:
-		sleeper.SleepFinish(true)
-		return nil
-	case <-cancel:
-		sleeper.SleepFinish(false)
-		return linuxerr.ErrInterrupted
-	}
-}
-
-// AbortableMutex is an abortable mutex. It allows Lock() to be aborted while it
-// waits to acquire the mutex.
-type AbortableMutex struct {
-	v  int32
-	ch chan struct{}
-}
-
-// Init initializes the abortable mutex.
-func (m *AbortableMutex) Init() {
-	m.v = 1
-	m.ch = make(chan struct{}, 1)
-}
-
-// Lock attempts to acquire the mutex, returning true on success. If something
-// is written to the "c" while Lock waits, the wait is aborted and false is
-// returned instead.
-func (m *AbortableMutex) Lock(s Sleeper) bool {
-	// Uncontended case.
-	if atomic.AddInt32(&m.v, -1) == 0 {
-		return true
-	}
-
-	var c <-chan struct{}
-	if s != nil {
-		c = s.SleepStart()
-	}
-
-	for {
-		// Try to acquire the mutex again, at the same time making sure
-		// that m.v is negative, which indicates to the owner of the
-		// lock that it is contended, which ill force it to try to wake
-		// someone up when it releases the mutex.
-		if v := atomic.LoadInt32(&m.v); v >= 0 && atomic.SwapInt32(&m.v, -1) == 1 {
-			if s != nil {
-				s.SleepFinish(true)
-			}
-			return true
-		}
-
-		// Wait for the owner to wake us up before trying again, or for
-		// the wait to be aborted by the provided channel.
-		select {
-		case <-m.ch:
-		case <-c:
-			// s must be non-nil, otherwise c would be nil and we'd
-			// never reach this path.
-			s.SleepFinish(false)
-			return false
-		}
-	}
-}
-
-// Unlock releases the mutex.
-func (m *AbortableMutex) Unlock() {
-	if atomic.SwapInt32(&m.v, 1) == 0 {
-		// There were no pending waiters.
-		return
-	}
-
-	// Wake some waiter up.
-	select {
-	case m.ch <- struct{}{}:
-	default:
-	}
-}

pkg/amutex/amutex_test.go

@@ -1,98 +0,0 @@
-// Copyright 2018 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 amutex
-
-import (
-	"testing"
-	"time"
-
-	"gvisor.dev/gvisor/pkg/sync"
-)
-
-type sleeper struct {
-	ch chan struct{}
-}
-
-func (s *sleeper) SleepStart() <-chan struct{} {
-	return s.ch
-}
-
-func (*sleeper) SleepFinish(bool) {
-}
-
-func (s *sleeper) Interrupted() bool {
-	return len(s.ch) != 0
-}
-
-func TestMutualExclusion(t *testing.T) {
-	var m AbortableMutex
-	m.Init()
-
-	// Test mutual exclusion by running "gr" goroutines concurrently, and
-	// have each one increment a counter "iters" times within the critical
-	// section established by the mutex.
-	//
-	// If at the end of the counter is not gr * iters, then we know that
-	// goroutines ran concurrently within the critical section.
-	//
-	// If one of the goroutines doesn't complete, it's likely a bug that
-	// causes it to wait forever.
-	const gr = 1000
-	const iters = 100000
-	v := 0
-	var wg sync.WaitGroup
-	for i := 0; i < gr; i++ {
-		wg.Add(1)
-		go func() {
-			for j := 0; j < iters; j++ {
-				m.Lock(nil)
-				v++
-				m.Unlock()
-			}
-			wg.Done()
-		}()
-	}
-
-	wg.Wait()
-
-	if v != gr*iters {
-		t.Fatalf("Bad count: got %v, want %v", v, gr*iters)
-	}
-}
-
-func TestAbortWait(t *testing.T) {
-	var s sleeper
-	var m AbortableMutex
-	m.Init()
-
-	// Lock the mutex.
-	m.Lock(&s)
-
-	// Lock again, but this time cancel after 500ms.
-	s.ch = make(chan struct{}, 1)
-	go func() {
-		time.Sleep(500 * time.Millisecond)
-		s.ch <- struct{}{}
-	}()
-	if v := m.Lock(&s); v {
-		t.Fatalf("Lock succeeded when it should have failed")
-	}
-
-	// Lock again, but cancel right away.
-	s.ch <- struct{}{}
-	if v := m.Lock(&s); v {
-		t.Fatalf("Lock succeeded when it should have failed")
-	}
-}

pkg/context/BUILD

@@ -4,9 +4,12 @@ package(licenses = ["notice"])
 
 go_library(
     name = "context",
-    srcs = ["context.go"],
+    srcs = [
+        "context.go",
+    ],
     visibility = ["//:sandbox"],
     deps = [
         "//pkg/log",
+        "//pkg/waiter",
     ],
 )

pkg/context/context.go

@@ -27,9 +27,112 @@ import (
 	"time"
 
 	"gvisor.dev/gvisor/pkg/log"
+	"gvisor.dev/gvisor/pkg/waiter"
 )
 
-// A Context represents a thread of execution (hereafter "goroutine" to reflect
+// Blocker represents an object with control flow hooks.
+//
+// These may be used to perform blocking operations, sleep or otherwise
+// wait, since there may be asynchronous events that require processing.
+type Blocker interface {
+	// Interrupt interrupts any Block operations.
+	Interrupt()
+
+	// Interrupted notes whether this context is Interrupted.
+	Interrupted() bool
+
+	// BlockOn blocks until one of the previously registered events occurs,
+	// or some external interrupt (cancellation).
+	//
+	// The return value should indicate whether the wake-up occurred as a
+	// result of the requested event (versus an external interrupt).
+	BlockOn(waiter.Waitable, waiter.EventMask) bool
+
+	// BlockWithTimeoutOn blocks until either the conditions of Block are
+	// satisfied, or the timeout is hit. Note that deadlines are not supported
+	// since the notion of "with respect to what clock" is not resolved.
+	//
+	// The return value is per BlockOn.
+	BlockWithTimeoutOn(waiter.Waitable, waiter.EventMask, time.Duration) (time.Duration, bool)
+
+	// UninterruptibleSleepStart indicates the beginning of an uninterruptible
+	// sleep state (equivalent to Linux's TASK_UNINTERRUPTIBLE). If deactivate
+	// is true and the Context represents a Task, the Task's AddressSpace is
+	// deactivated.
+	UninterruptibleSleepStart(deactivate bool)
+
+	// UninterruptibleSleepFinish indicates the end of an uninterruptible sleep
+	// state that was begun by a previous call to UninterruptibleSleepStart. If
+	// activate is true and the Context represents a Task, the Task's
+	// AddressSpace is activated. Normally activate is the same value as the
+	// deactivate parameter passed to UninterruptibleSleepStart.
+	UninterruptibleSleepFinish(activate bool)
+}
+
+// NoTask is an implementation of Blocker that does not block.
+type NoTask struct {
+	cancel chan struct{}
+}
+
+// Interrupt implements Blocker.Interrupt.
+func (nt *NoTask) Interrupt() {
+	select {
+	case nt.cancel <- struct{}{}:
+	default:
+	}
+}
+
+// Interrupted implements Blocker.Interrupted.
+func (nt *NoTask) Interrupted() bool {
+	return nt.cancel != nil && len(nt.cancel) > 0
+}
+
+// BlockOn implements Blocker.BlockOn.
+func (nt *NoTask) BlockOn(w waiter.Waitable, mask waiter.EventMask) bool {
+	if nt.cancel == nil {
+		nt.cancel = make(chan struct{}, 1)
+	}
+	e, ch := waiter.NewChannelEntry(mask)
+	w.EventRegister(&e)
+	defer w.EventUnregister(&e)
+	select {
+	case <-nt.cancel:
+		return false // Interrupted.
+	case _, ok := <-ch:
+		return ok
+	}
+}
+
+// BlockWithTimeoutOn implements Blocker.BlockWithTimeoutOn.
+func (nt *NoTask) BlockWithTimeoutOn(w waiter.Waitable, mask waiter.EventMask, duration time.Duration) (time.Duration, bool) {
+	if nt.cancel == nil {
+		nt.cancel = make(chan struct{}, 1)
+	}
+	e, ch := waiter.NewChannelEntry(mask)
+	w.EventRegister(&e)
+	defer w.EventUnregister(&e)
+	start := time.Now() // In system time.
+	t := time.AfterFunc(duration, func() { ch <- struct{}{} })
+	select {
+	case <-nt.cancel:
+		return time.Since(start), false // Interrupted.
+	case _, ok := <-ch:
+		if ok && t.Stop() {
+			// Timer never fired.
+			return time.Since(start), ok
+		}
+		// Timer fired, remain is zero.
+		return time.Duration(0), ok
+	}
+}
+
+// UninterruptibleSleepStart implmenents Blocker.UninterruptedSleepStart.
+func (*NoTask) UninterruptibleSleepStart(bool) {}
+
+// UninterruptibleSleepFinish implmenents Blocker.UninterruptibleSleepFinish.
+func (*NoTask) UninterruptibleSleepFinish(bool) {}
+
+// Context represents a thread of execution (hereafter "goroutine" to reflect
 // Go idiosyncrasy). It carries state associated with the goroutine across API
 // boundaries.
 //
@@ -46,96 +149,23 @@ import (
 //
 // In both cases, values extracted from the Context should be used instead.
 type Context interface {
-	log.Logger
 	context.Context
-
-	ChannelSleeper
-
-	// UninterruptibleSleepStart indicates the beginning of an uninterruptible
-	// sleep state (equivalent to Linux's TASK_UNINTERRUPTIBLE). If deactivate
-	// is true and the Context represents a Task, the Task's AddressSpace is
-	// deactivated.
-	UninterruptibleSleepStart(deactivate bool)
-
-	// UninterruptibleSleepFinish indicates the end of an uninterruptible sleep
-	// state that was begun by a previous call to UninterruptibleSleepStart. If
-	// activate is true and the Context represents a Task, the Task's
-	// AddressSpace is activated. Normally activate is the same value as the
-	// deactivate parameter passed to UninterruptibleSleepStart.
-	UninterruptibleSleepFinish(activate bool)
-}
-
-// A ChannelSleeper represents a goroutine that may sleep interruptibly, where
-// interruption is indicated by a channel becoming readable.
-type ChannelSleeper interface {
-	// SleepStart is called before going to sleep interruptibly. If SleepStart
-	// returns a non-nil channel and that channel becomes ready for receiving
-	// while the goroutine is sleeping, the goroutine should be woken, and
-	// SleepFinish(false) should be called. Otherwise, SleepFinish(true) should
-	// be called after the goroutine stops sleeping.
-	SleepStart() <-chan struct{}
-
-	// SleepFinish is called after an interruptibly-sleeping goroutine stops
-	// sleeping, as documented by SleepStart.
-	SleepFinish(success bool)
-
-	// Interrupted returns true if the channel returned by SleepStart is
-	// ready for receiving.
-	Interrupted() bool
-}
-
-// NoopSleeper is a noop implementation of ChannelSleeper and
-// Context.UninterruptibleSleep* methods for anonymous embedding in other types
-// that do not implement special behavior around sleeps.
-type NoopSleeper struct{}
-
-// SleepStart implements ChannelSleeper.SleepStart.
-func (NoopSleeper) SleepStart() <-chan struct{} {
-	return nil
-}
-
-// SleepFinish implements ChannelSleeper.SleepFinish.
-func (NoopSleeper) SleepFinish(success bool) {}
-
-// Interrupted implements ChannelSleeper.Interrupted.
-func (NoopSleeper) Interrupted() bool {
-	return false
-}
-
-// UninterruptibleSleepStart implements Context.UninterruptibleSleepStart.
-func (NoopSleeper) UninterruptibleSleepStart(deactivate bool) {}
-
-// UninterruptibleSleepFinish implements Context.UninterruptibleSleepFinish.
-func (NoopSleeper) UninterruptibleSleepFinish(activate bool) {}
-
-// Deadline implements context.Context.Deadline.
-func (NoopSleeper) Deadline() (time.Time, bool) {
-	return time.Time{}, false
-}
-
-// Done implements context.Context.Done.
-func (NoopSleeper) Done() <-chan struct{} {
-	return nil
-}
-
-// Err returns context.Context.Err.
-func (NoopSleeper) Err() error {
-	return nil
+	log.Logger
+	Blocker
 }
 
 // logContext implements basic logging.
 type logContext struct {
+	NoTask
 	log.Logger
-	NoopSleeper
-}
-
-// Value implements Context.Value.
-func (logContext) Value(key interface{}) interface{} {
-	return nil
+	context.Context
 }
 
 // bgContext is the context returned by context.Background.
-var bgContext = &logContext{Logger: log.Log()}
+var bgContext Context = &logContext{
+	Context: context.Background(),
+	Logger:  log.Log(),
+}
 
 // Background returns an empty context using the default logger.
 // Generally, one should use the Task as their context when available, or avoid

pkg/sentry/fs/fdpipe/pipe_opener.go

@@ -24,6 +24,7 @@ import (
 	"gvisor.dev/gvisor/pkg/errors/linuxerr"
 	"gvisor.dev/gvisor/pkg/fd"
 	"gvisor.dev/gvisor/pkg/sentry/fs"
+	"gvisor.dev/gvisor/pkg/waiter"
 )
 
 // NonBlockingOpener is a generic host file opener used to retry opening host
@@ -37,8 +38,11 @@ type NonBlockingOpener interface {
 // Open blocks until a host pipe can be opened or the action was cancelled.
 // On success, returns fs.FileOperations wrapping the opened host pipe.
 func Open(ctx context.Context, opener NonBlockingOpener, flags fs.FileFlags) (fs.FileOperations, error) {
-	p := &pipeOpenState{}
-	canceled := false
+	var (
+		p        pipeOpenState
+		q        waiter.NeverReady
+		canceled bool
+	)
 	for {
 		if file, err := p.TryOpen(ctx, opener, flags); err != linuxerr.ErrWouldBlock {
 			return file, err
@@ -54,24 +58,9 @@ func Open(ctx context.Context, opener NonBlockingOpener, flags fs.FileFlags) (fs
 			return nil, linuxerr.ErrInterrupted
 		}
 
-		cancel := ctx.SleepStart()
-		select {
-		case <-cancel:
-			// The cancellation request received here really says
-			// "cancel from now on (or ASAP)". Any environmental
-			// changes happened before receiving it, that might have
-			// caused open to not block anymore, should still be
-			// respected. So we cannot just return here. We have to
-			// give open another try below first.
+		// Block for up to the requested amount of time.
+		if left, ok := ctx.BlockWithTimeoutOn(&q, waiter.ReadableEvents, 100*time.Millisecond); !ok && left != 0 {
 			canceled = true
-			ctx.SleepFinish(false)
-		case <-time.After(100 * time.Millisecond):
-			// If we would block, then delay retrying for a bit, since there
-			// is no way to know when the pipe would be ready to be
-			// re-opened. This is identical to sending an event notification
-			// to stop blocking in Task.Block, given that this routine will
-			// stop retrying if a cancelation is received.
-			ctx.SleepFinish(true)
 		}
 	}
 }

pkg/sentry/fs/fdpipe/pipe_state.go

@@ -32,6 +32,7 @@ func (p *pipeOperations) beforeSave() {
 		}
 		p.readAheadBuffer = append(p.readAheadBuffer, data...)
 	} else if p.flags.Write {
+		// It's not really possible to evaluate what can be reopened on restore.
 		file, err := p.opener.NonBlockingOpen(context.Background(), fs.PermMask{Write: true})
 		if err != nil {
 			panic(&fs.ErrSaveRejection{

pkg/sentry/fs/lock/BUILD

@@ -43,6 +43,7 @@ go_library(
     deps = [
         "//pkg/abi/linux",
         "//pkg/context",
+        "//pkg/errors/linuxerr",
         "//pkg/log",
         "//pkg/sync",
         "//pkg/waiter",
@@ -58,5 +59,8 @@ go_test(
         "lock_test.go",
     ],
     library = ":lock",
-    deps = ["@org_golang_x_sys//unix:go_default_library"],
+    deps = [
+        "//pkg/errors/linuxerr",
+        "@org_golang_x_sys//unix:go_default_library",
+    ],
 )

pkg/sentry/fs/lock/lock.go

@@ -56,6 +56,7 @@ import (
 	"golang.org/x/sys/unix"
 	"gvisor.dev/gvisor/pkg/abi/linux"
 	"gvisor.dev/gvisor/pkg/context"
+	"gvisor.dev/gvisor/pkg/errors/linuxerr"
 	"gvisor.dev/gvisor/pkg/sync"
 	"gvisor.dev/gvisor/pkg/waiter"
 )
@@ -135,48 +136,31 @@ type Locks struct {
 	blockedQueue waiter.Queue
 }
 
-// Blocker is the interface used for blocking locks. Passing a nil Blocker
-// will be treated as non-blocking.
-type Blocker interface {
-	Block(C <-chan struct{}) error
-}
-
-const (
-	// EventMaskAll is the mask we will always use for locks, by using the
-	// same mask all the time we can wake up everyone anytime the lock
-	// changes state.
-	EventMaskAll waiter.EventMask = 0xFFFF
-)
-
-// LockRegion attempts to acquire a typed lock for the uid on a region
-// of a file. Returns true if successful in locking the region. If false
-// is returned, the caller should normally interpret this as "try again later" if
-// acquiring the lock in a non-blocking mode or "interrupted" if in a blocking mode.
-// Blocker is the interface used to provide blocking behavior, passing a nil Blocker
-// will result in non-blocking behavior.
-func (l *Locks) LockRegion(uid UniqueID, ownerPID int32, t LockType, r LockRange, block Blocker) bool {
+// LockRegion attempts to acquire a typed lock for the uid on a region of a
+// file. Returns nil if successful in locking the region, otherwise an
+// appropriate error is returned.
+func (l *Locks) LockRegion(ctx context.Context, uid UniqueID, ownerPID int32, t LockType, r LockRange, block bool) error {
+	l.mu.Lock()
+	defer l.mu.Unlock()
 	for {
-		l.mu.Lock()
 
 		// Blocking locks must run in a loop because we'll be woken up whenever an unlock event
 		// happens for this lock. We will then attempt to take the lock again and if it fails
 		// continue blocking.
-		res := l.locks.lock(uid, ownerPID, t, r)
-		if !res && block != nil {
-			e, ch := waiter.NewChannelEntry(EventMaskAll)
-			l.blockedQueue.EventRegister(&e)
-			l.mu.Unlock()
-			if err := block.Block(ch); err != nil {
-				// We were interrupted, the caller can translate this to EINTR if applicable.
-				l.blockedQueue.EventUnregister(&e)
-				return false
+		err := l.locks.lock(uid, ownerPID, t, r)
+		if err == linuxerr.ErrWouldBlock && block {
+			// Note: we release the lock in EventRegister below, in
+			// order to avoid a possible race.
+			ok := ctx.BlockOn(l, waiter.EventIn)
+			l.mu.Lock() // +checklocksforce: see above.
+			if ok {
+				continue // Try again now that someone has unlocked.
 			}
-			l.blockedQueue.EventUnregister(&e)
-			continue // Try again now that someone has unlocked.
+			// Must be interrupted.
+			return linuxerr.ErrInterrupted
 		}
 
-		l.mu.Unlock()
-		return res
+		return err
 	}
 }
 
@@ -184,8 +168,25 @@ func (l *Locks) LockRegion(uid UniqueID, ownerPID int32, t LockType, r LockRange
 // F_GETLK (and does not care about storing PIDs as a result).
 //
 // TODO(gvisor.dev/issue/1624): Delete.
-func (l *Locks) LockRegionVFS1(uid UniqueID, t LockType, r LockRange, block Blocker) bool {
-	return l.LockRegion(uid, 0 /* ownerPID */, t, r, block)
+func (l *Locks) LockRegionVFS1(ctx context.Context, uid UniqueID, t LockType, r LockRange, block bool) error {
+	return l.LockRegion(ctx, uid, 0 /* ownerPID */, t, r, block)
+}
+
+// Readiness always returns zero.
+func (l *Locks) Readiness(waiter.EventMask) waiter.EventMask {
+	return 0
+}
+
+// EventRegister implements waiter.Waitable.EventRegister.
+func (l *Locks) EventRegister(e *waiter.Entry) error {
+	defer l.mu.Unlock() // +checklocksforce: see above.
+	l.blockedQueue.EventRegister(e)
+	return nil
+}
+
+// EventUnregister implements waiter.Waitable.EventUnregister.
+func (l *Locks) EventUnregister(e *waiter.Entry) {
+	l.blockedQueue.EventUnregister(e)
 }
 
 // UnlockRegion attempts to release a lock for the uid on a region of a file.
@@ -197,7 +198,9 @@ func (l *Locks) UnlockRegion(uid UniqueID, r LockRange) {
 	l.locks.unlock(uid, r)
 
 	// Now that we've released the lock, we need to wake up any waiters.
-	l.blockedQueue.Notify(EventMaskAll)
+	// We track how many notifications have happened since the last attempt
+	// to acquire the lock, in order to ensure that we avoid races.
+	l.blockedQueue.Notify(waiter.EventIn)
 }
 
 // makeLock returns a new typed Lock that has either uid as its only reader
@@ -332,11 +335,11 @@ func (l *Lock) isOnlyReader(uid UniqueID) bool {
 	return ok
 }
 
-// lock returns true if uid took a lock of type t on the entire range of
-// LockRange.
+// lock returns nil if uid took a lock of type t on the entire range of
+// LockRange. Otherwise, linuxerr.ErrWouldBlock is returned.
 //
 // Preconditions: r.Start <= r.End (will panic otherwise).
-func (l *LockSet) lock(uid UniqueID, ownerPID int32, t LockType, r LockRange) bool {
+func (l *LockSet) lock(uid UniqueID, ownerPID int32, t LockType, r LockRange) error {
 	if r.Start > r.End {
 		panic(fmt.Sprintf("lock: r.Start %d > r.End %d", r.Start, r.End))
 	}
@@ -344,15 +347,16 @@ func (l *LockSet) lock(uid UniqueID, ownerPID int32, t LockType, r LockRange) bo
 	// Don't attempt to insert anything with a range of 0 and treat this
 	// as a successful no-op.
 	if r.Length() == 0 {
-		return true
+		return nil
 	}
 
-	// Do a first-pass check.  We *could* hold onto the segments we
-	// checked if canLock would return true, but traversing the segment
-	// set should be fast and this keeps things simple.
+	// Do a first-pass check. We *could* hold onto the segments we checked
+	// if canLock would return true, but traversing the segment set should
+	// be fast and this keeps things simple.
 	if !l.canLock(uid, t, r) {
-		return false
+		return linuxerr.ErrWouldBlock
 	}
+
 	// Get our starting point.
 	seg, gap := l.Find(r.Start)
 	if gap.Ok() {
@@ -381,7 +385,8 @@ func (l *LockSet) lock(uid UniqueID, ownerPID int32, t LockType, r LockRange) bo
 			seg = gap.NextSegment()
 		}
 	}
-	return true
+
+	return nil
 }
 
 // unlock is always successful.  If uid has no locks held for the range LockRange,

pkg/sentry/fs/lock/lock_test.go

@@ -17,6 +17,8 @@ package lock
 import (
 	"reflect"
 	"testing"
+
+	"gvisor.dev/gvisor/pkg/errors/linuxerr"
 )
 
 type entry struct {
@@ -182,13 +184,11 @@ func TestSetLock(t *testing.T) {
 		uid UniqueID
 		// lock type requested.
 		lockType LockType
-
-		// success is true if taking the above
-		// lock should succeed.
-		success bool
+		// err is the expected results.
+		err error
 
 		// Expected layout of the set after locking
-		// if success is true.
+		// if err is nil.
 		after []entry
 	}{
 		{
@@ -197,7 +197,6 @@ func TestSetLock(t *testing.T) {
 			end:      0,
 			uid:      0,
 			lockType: ReadLock,
-			success:  true,
 		},
 		{
 			name:     "set zero length WriteLock on empty set",
@@ -205,7 +204,6 @@ func TestSetLock(t *testing.T) {
 			end:      0,
 			uid:      0,
 			lockType: WriteLock,
-			success:  true,
 		},
 		{
 			name:     "set ReadLock on empty set",
@@ -213,7 +211,6 @@ func TestSetLock(t *testing.T) {
 			end:      LockEOF,
 			uid:      0,
 			lockType: ReadLock,
-			success:  true,
 			// + ----------------------------------------- +
 			// | Readers 0                                 |
 			// + ----------------------------------------- +
@@ -231,7 +228,6 @@ func TestSetLock(t *testing.T) {
 			end:      LockEOF,
 			uid:      0,
 			lockType: WriteLock,
-			success:  true,
 			// + ----------------------------------------- +
 			// | Writer  0                                 |
 			// + ----------------------------------------- +
@@ -259,7 +255,6 @@ func TestSetLock(t *testing.T) {
 			end:      4096,
 			uid:      0,
 			lockType: ReadLock,
-			success:  true,
 			// + ----------- + --------------------------- +
 			// | Readers 0   | Writer 0                    |
 			// + ----------- + --------------------------- +
@@ -291,7 +286,6 @@ func TestSetLock(t *testing.T) {
 			end:      4096,
 			uid:      0,
 			lockType: WriteLock,
-			success:  true,
 			// + ----------- + --------------------------- +
 			// | Writer 0    | Readers 0                   |
 			// + ----------- + --------------------------- +
@@ -323,7 +317,7 @@ func TestSetLock(t *testing.T) {
 			end:      4096,
 			uid:      1,
 			lockType: ReadLock,
-			success:  false,
+			err:      linuxerr.ErrWouldBlock,
 		},
 		{
 			name: "set WriteLock on ReadLock different uid",
@@ -341,7 +335,7 @@ func TestSetLock(t *testing.T) {
 			end:      4096,
 			uid:      1,
 			lockType: WriteLock,
-			success:  false,
+			err:      linuxerr.ErrWouldBlock,
 		},
 		{
 			name: "split ReadLock for overlapping lock at start 0",
@@ -359,7 +353,6 @@ func TestSetLock(t *testing.T) {
 			end:      4096,
 			uid:      1,
 			lockType: ReadLock,
-			success:  true,
 			// + -------------- + --------------------------- +
 			// | Readers 0 & 1  | Readers 0                   |
 			// + -------------- + --------------------------- +
@@ -391,7 +384,6 @@ func TestSetLock(t *testing.T) {
 			end:      8192,
 			uid:      1,
 			lockType: ReadLock,
-			success:  true,
 			// + ---------- + -------------- + ----------- +
 			// | Readers 0  | Readers 0 & 1  | Readers 0   |
 			// + ---------- + -------------- + ----------- +
@@ -427,7 +419,6 @@ func TestSetLock(t *testing.T) {
 			end:      8192,
 			uid:      0,
 			lockType: ReadLock,
-			success:  true,
 			// + ----------------------------------------- +
 			// | Readers 0                                 |
 			// + ----------------------------------------- +
@@ -455,7 +446,6 @@ func TestSetLock(t *testing.T) {
 			end:      LockEOF,
 			uid:      0,
 			lockType: ReadLock,
-			success:  true,
 			// Note that this is not merged after lock does a Split.  This is
 			// fine because the two locks will still *behave* as one.  In other
 			// words we can fragment any lock all we want and semantically it
@@ -492,7 +482,6 @@ func TestSetLock(t *testing.T) {
 			end:      4096,
 			uid:      1,
 			lockType: ReadLock,
-			success:  true,
 			// + --------- + ------------- + ------------- +
 			// | Reader 1  | Readers 0 & 1 | Reader 0      |
 			// + ----------+ ------------- + ------------- +
@@ -536,7 +525,6 @@ func TestSetLock(t *testing.T) {
 			end:      4096,
 			uid:      0,
 			lockType: WriteLock,
-			success:  true,
 			// + ------------- + -------- + ------------- +
 			// | Readers 0 & 1 | Writer 0 | Readers 0 & 2 |
 			// + ------------- + -------- + ------------- +
@@ -580,7 +568,6 @@ func TestSetLock(t *testing.T) {
 			end:      3072,
 			uid:      0,
 			lockType: WriteLock,
-			success:  true,
 			// + ------------- + -------- + --- + ------------- +
 			// | Readers 0 & 1 | Writer 0 | gap | Readers 0 & 2 |
 			// + ------------- + -------- + --- + ------------- +
@@ -620,7 +607,7 @@ func TestSetLock(t *testing.T) {
 			end:      2048,
 			uid:      0,
 			lockType: WriteLock,
-			success:  false,
+			err:      linuxerr.ErrWouldBlock,
 		},
 		{
 			name: "take WriteLock on whole file if all uids are the same",
@@ -650,7 +637,6 @@ func TestSetLock(t *testing.T) {
 			end:      LockEOF,
 			uid:      0,
 			lockType: WriteLock,
-			success:  true,
 			// We do not manually merge locks.  Semantically a fragmented lock
 			// held by the same uid will behave as one lock so it makes no difference.
 			//
@@ -676,7 +662,7 @@ func TestSetLock(t *testing.T) {
 			l := fill(test.before)
 
 			r := LockRange{Start: test.start, End: test.end}
-			success := l.lock(test.uid, 0 /* ownerPID */, test.lockType, r)
+			err := l.lock(test.uid, 0 /* ownerPID */, test.lockType, r)
 			var got []entry
 			for seg := l.FirstSegment(); seg.Ok(); seg = seg.NextSegment() {
 				got = append(got, entry{
@@ -685,12 +671,12 @@ func TestSetLock(t *testing.T) {
 				})
 			}
 
-			if success != test.success {
-				t.Errorf("setlock(%v, %+v, %d, %d) got success %v, want %v", test.before, r, test.uid, test.lockType, success, test.success)
+			if err != test.err {
+				t.Errorf("setlock(%v, %+v, %d, %d) got err %v, want %v", test.before, r, test.uid, test.lockType, err, test.err)
 				return
 			}
 
-			if success {
+			if err == nil {
 				if !equals(got, test.after) {
 					t.Errorf("got set %+v, want %+v", got, test.after)
 				}

pkg/sentry/fsimpl/gofer/gofer.go

@@ -2536,7 +2536,7 @@ func (fd *fileDescription) RemoveXattr(ctx context.Context, name string) error {
 }
 
 // LockBSD implements vfs.FileDescriptionImpl.LockBSD.
-func (fd *fileDescription) LockBSD(ctx context.Context, uid fslock.UniqueID, ownerPID int32, t fslock.LockType, block fslock.Blocker) error {
+func (fd *fileDescription) LockBSD(ctx context.Context, uid fslock.UniqueID, ownerPID int32, t fslock.LockType, block bool) error {
 	fd.lockLogging.Do(func() {
 		log.Infof("File lock using gofer file handled internally.")
 	})
@@ -2544,7 +2544,7 @@ func (fd *fileDescription) LockBSD(ctx context.Context, uid fslock.UniqueID, own
 }
 
 // LockPOSIX implements vfs.FileDescriptionImpl.LockPOSIX.
-func (fd *fileDescription) LockPOSIX(ctx context.Context, uid fslock.UniqueID, ownerPID int32, t fslock.LockType, r fslock.LockRange, block fslock.Blocker) error {
+func (fd *fileDescription) LockPOSIX(ctx context.Context, uid fslock.UniqueID, ownerPID int32, t fslock.LockType, r fslock.LockRange, block bool) error {
 	fd.lockLogging.Do(func() {
 		log.Infof("Range lock using gofer file handled internally.")
 	})

pkg/sentry/fsimpl/gofer/host_named_pipe.go

@@ -22,6 +22,7 @@ import (
 	"golang.org/x/sys/unix"
 	"gvisor.dev/gvisor/pkg/context"
 	"gvisor.dev/gvisor/pkg/errors/linuxerr"
+	"gvisor.dev/gvisor/pkg/waiter"
 )
 
 // Global pipe used by blockUntilNonblockingPipeHasWriter since we can't create
@@ -51,6 +52,7 @@ func blockUntilNonblockingPipeHasWriter(ctx context.Context, fd int32) error {
 		if ok {
 			return nil
 		}
+		// Delay before trying again.
 		if sleepErr := sleepBetweenNamedPipeOpenChecks(ctx); sleepErr != nil {
 			// Another application thread may have opened this pipe for
 			// writing, succeeded because we previously opened the pipe for
@@ -99,15 +101,10 @@ func nonblockingPipeHasWriter(fd int32) (bool, error) {
 }
 
 func sleepBetweenNamedPipeOpenChecks(ctx context.Context) error {
-	t := time.NewTimer(100 * time.Millisecond)
-	defer t.Stop()
-	cancel := ctx.SleepStart()
-	select {
-	case <-t.C:
-		ctx.SleepFinish(true)
-		return nil
-	case <-cancel:
-		ctx.SleepFinish(false)
+	var q waiter.NeverReady
+	left, ok := ctx.BlockWithTimeoutOn(&q, waiter.EventIn, 100*time.Millisecond)
+	if !ok && left != 0 {
 		return linuxerr.ErrInterrupted
 	}
+	return nil
 }

pkg/sentry/fsimpl/tmpfs/regular_file_test.go

@@ -140,32 +140,32 @@ func TestLocks(t *testing.T) {
 
 	uid1 := 123
 	uid2 := 456
-	if err := fd.Impl().LockBSD(ctx, uid1, 0 /* ownerPID */, lock.ReadLock, nil); err != nil {
+	if err := fd.Impl().LockBSD(ctx, uid1, 0 /* ownerPID */, lock.ReadLock, false /* block */); err != nil {
 		t.Fatalf("fd.Impl().LockBSD failed: err = %v", err)
 	}
-	if err := fd.Impl().LockBSD(ctx, uid2, 0 /* ownerPID */, lock.ReadLock, nil); err != nil {
+	if err := fd.Impl().LockBSD(ctx, uid2, 0 /* ownerPID */, lock.ReadLock, false /* block */); err != nil {
 		t.Fatalf("fd.Impl().LockBSD failed: err = %v", err)
 	}
-	if got, want := fd.Impl().LockBSD(ctx, uid2, 0 /* ownerPID */, lock.WriteLock, nil), linuxerr.ErrWouldBlock; got != want {
+	if got, want := fd.Impl().LockBSD(ctx, uid2, 0 /* ownerPID */, lock.WriteLock, false /* block */), linuxerr.ErrWouldBlock; got != want {
 		t.Fatalf("fd.Impl().LockBSD failed: got = %v, want = %v", got, want)
 	}
 	if err := fd.Impl().UnlockBSD(ctx, uid1); err != nil {
 		t.Fatalf("fd.Impl().UnlockBSD failed: err = %v", err)
 	}
-	if err := fd.Impl().LockBSD(ctx, uid2, 0 /* ownerPID */, lock.WriteLock, nil); err != nil {
+	if err := fd.Impl().LockBSD(ctx, uid2, 0 /* ownerPID */, lock.WriteLock, false /* block */); err != nil {
 		t.Fatalf("fd.Impl().LockBSD failed: err = %v", err)
 	}
 
-	if err := fd.Impl().LockPOSIX(ctx, uid1, 0 /* ownerPID */, lock.ReadLock, lock.LockRange{Start: 0, End: 1}, nil); err != nil {
+	if err := fd.Impl().LockPOSIX(ctx, uid1, 0 /* ownerPID */, lock.ReadLock, lock.LockRange{Start: 0, End: 1}, false /* block */); err != nil {
 		t.Fatalf("fd.Impl().LockPOSIX failed: err = %v", err)
 	}
-	if err := fd.Impl().LockPOSIX(ctx, uid2, 0 /* ownerPID */, lock.ReadLock, lock.LockRange{Start: 1, End: 2}, nil); err != nil {
+	if err := fd.Impl().LockPOSIX(ctx, uid2, 0 /* ownerPID */, lock.ReadLock, lock.LockRange{Start: 1, End: 2}, false /* block */); err != nil {
 		t.Fatalf("fd.Impl().LockPOSIX failed: err = %v", err)
 	}
-	if err := fd.Impl().LockPOSIX(ctx, uid1, 0 /* ownerPID */, lock.WriteLock, lock.LockRange{Start: 0, End: 1}, nil); err != nil {
+	if err := fd.Impl().LockPOSIX(ctx, uid1, 0 /* ownerPID */, lock.WriteLock, lock.LockRange{Start: 0, End: 1}, false /* block */); err != nil {
 		t.Fatalf("fd.Impl().LockPOSIX failed: err = %v", err)
 	}
-	if got, want := fd.Impl().LockPOSIX(ctx, uid2, 0 /* ownerPID */, lock.ReadLock, lock.LockRange{Start: 0, End: 1}, nil), linuxerr.ErrWouldBlock; got != want {
+	if got, want := fd.Impl().LockPOSIX(ctx, uid2, 0 /* ownerPID */, lock.ReadLock, lock.LockRange{Start: 0, End: 1}, false /* block */), linuxerr.ErrWouldBlock; got != want {
 		t.Fatalf("fd.Impl().LockPOSIX failed: got = %v, want = %v", got, want)
 	}
 	if err := fd.Impl().UnlockPOSIX(ctx, uid1, lock.LockRange{Start: 0, End: 1}); err != nil {

pkg/sentry/fsimpl/verity/verity.go

@@ -1520,7 +1520,7 @@ func (fd *fileDescription) SupportsLocks() bool {
 }
 
 // LockBSD implements vfs.FileDescriptionImpl.LockBSD.
-func (fd *fileDescription) LockBSD(ctx context.Context, uid fslock.UniqueID, ownerPID int32, t fslock.LockType, block fslock.Blocker) error {
+func (fd *fileDescription) LockBSD(ctx context.Context, uid fslock.UniqueID, ownerPID int32, t fslock.LockType, block bool) error {
 	return fd.lowerFD.LockBSD(ctx, ownerPID, t, block)
 }
 
@@ -1530,7 +1530,7 @@ func (fd *fileDescription) UnlockBSD(ctx context.Context, uid fslock.UniqueID) e
 }
 
 // LockPOSIX implements vfs.FileDescriptionImpl.LockPOSIX.
-func (fd *fileDescription) LockPOSIX(ctx context.Context, uid fslock.UniqueID, ownerPID int32, t fslock.LockType, r fslock.LockRange, block fslock.Blocker) error {
+func (fd *fileDescription) LockPOSIX(ctx context.Context, uid fslock.UniqueID, ownerPID int32, t fslock.LockType, r fslock.LockRange, block bool) error {
 	return fd.lowerFD.LockPOSIX(ctx, uid, ownerPID, t, r, block)
 }
 

pkg/sentry/kernel/kernel.go

@@ -806,28 +806,27 @@ type CreateProcessArgs struct {
 
 // NewContext returns a context.Context that represents the task that will be
 // created by args.NewContext(k).
-func (args *CreateProcessArgs) NewContext(k *Kernel) *createProcessContext {
+func (args *CreateProcessArgs) NewContext(k *Kernel) context.Context {
 	return &createProcessContext{
-		Logger: log.Log(),
-		k:      k,
-		args:   args,
+		Context: context.Background(),
+		kernel:  k,
+		args:    args,
 	}
 }
 
 // createProcessContext is a context.Context that represents the context
 // associated with a task that is being created.
 type createProcessContext struct {
-	context.NoopSleeper
-	log.Logger
-	k    *Kernel
-	args *CreateProcessArgs
+	context.Context
+	kernel *Kernel
+	args   *CreateProcessArgs
 }
 
 // Value implements context.Context.Value.
 func (ctx *createProcessContext) Value(key interface{}) interface{} {
 	switch key {
 	case CtxKernel:
-		return ctx.k
+		return ctx.kernel
 	case CtxPIDNamespace:
 		return ctx.args.PIDNamespace
 	case CtxUTSNamespace:
@@ -852,34 +851,34 @@ func (ctx *createProcessContext) Value(key interface{}) interface{} {
 		root.IncRef()
 		return root
 	case vfs.CtxMountNamespace:
-		if ctx.k.globalInit == nil {
+		if ctx.kernel.globalInit == nil {
 			return nil
 		}
-		mntns := ctx.k.GlobalInit().Leader().MountNamespaceVFS2()
+		mntns := ctx.kernel.GlobalInit().Leader().MountNamespaceVFS2()
 		mntns.IncRef()
 		return mntns
 	case fs.CtxDirentCacheLimiter:
-		return ctx.k.DirentCacheLimiter
+		return ctx.kernel.DirentCacheLimiter
 	case inet.CtxStack:
-		return ctx.k.RootNetworkNamespace().Stack()
+		return ctx.kernel.RootNetworkNamespace().Stack()
 	case ktime.CtxRealtimeClock:
-		return ctx.k.RealtimeClock()
+		return ctx.kernel.RealtimeClock()
 	case limits.CtxLimits:
 		return ctx.args.Limits
 	case pgalloc.CtxMemoryFile:
-		return ctx.k.mf
+		return ctx.kernel.mf
 	case pgalloc.CtxMemoryFileProvider:
-		return ctx.k
+		return ctx.kernel
 	case platform.CtxPlatform:
-		return ctx.k
+		return ctx.kernel
 	case uniqueid.CtxGlobalUniqueID:
-		return ctx.k.UniqueID()
+		return ctx.kernel.UniqueID()
 	case uniqueid.CtxGlobalUniqueIDProvider:
-		return ctx.k
+		return ctx.kernel
 	case uniqueid.CtxInotifyCookie:
-		return ctx.k.GenerateInotifyCookie()
+		return ctx.kernel.GenerateInotifyCookie()
 	case unimpl.CtxEvents:
-		return ctx.k
+		return ctx.kernel
 	default:
 		return nil
 	}
@@ -1539,9 +1538,9 @@ func (k *Kernel) MemoryFile() *pgalloc.MemoryFile {
 // Callers are responsible for ensuring that the returned Context is not used
 // concurrently with changes to the Kernel.
 func (k *Kernel) SupervisorContext() context.Context {
-	return supervisorContext{
+	return &supervisorContext{
+		Kernel: k,
 		Logger: log.Log(),
-		k:      k,
 	}
 }
 
@@ -1641,13 +1640,28 @@ func (k *Kernel) ListSockets() []*SocketRecord {
 
 // supervisorContext is a privileged context.
 type supervisorContext struct {
-	context.NoopSleeper
+	context.NoTask
 	log.Logger
-	k *Kernel
+	*Kernel
+}
+
+// Deadline implements context.Context.Deadline.
+func (*Kernel) Deadline() (time.Time, bool) {
+	return time.Time{}, false
+}
+
+// Done implements context.Context.Done.
+func (*Kernel) Done() <-chan struct{} {
+	return nil
+}
+
+// Err implements context.Context.Err.
+func (*Kernel) Err() error {
+	return nil
 }
 
 // Value implements context.Context.
-func (ctx supervisorContext) Value(key interface{}) interface{} {
+func (ctx *supervisorContext) Value(key interface{}) interface{} {
 	switch key {
 	case CtxCanTrace:
 		// The supervisor context can trace anything. (None of
@@ -1655,60 +1669,60 @@ func (ctx supervisorContext) Value(key interface{}) interface{} {
 		// permissions are required for certain file accesses.)
 		return func(*Task, bool) bool { return true }
 	case CtxKernel:
-		return ctx.k
+		return ctx.Kernel
 	case CtxPIDNamespace:
-		return ctx.k.tasks.Root
+		return ctx.Kernel.tasks.Root
 	case CtxUTSNamespace:
-		return ctx.k.rootUTSNamespace
+		return ctx.Kernel.rootUTSNamespace
 	case ipc.CtxIPCNamespace:
-		ipcns := ctx.k.rootIPCNamespace
+		ipcns := ctx.Kernel.rootIPCNamespace
 		ipcns.IncRef()
 		return ipcns
 	case auth.CtxCredentials:
 		// The supervisor context is global root.
-		return auth.NewRootCredentials(ctx.k.rootUserNamespace)
+		return auth.NewRootCredentials(ctx.Kernel.rootUserNamespace)
 	case fs.CtxRoot:
-		if ctx.k.globalInit != nil {
-			return ctx.k.globalInit.mounts.Root()
+		if ctx.Kernel.globalInit != nil {
+			return ctx.Kernel.globalInit.mounts.Root()
 		}
 		return nil
 	case vfs.CtxRoot:
-		if ctx.k.globalInit == nil {
+		if ctx.Kernel.globalInit == nil {
 			return vfs.VirtualDentry{}
 		}
-		root := ctx.k.GlobalInit().Leader().MountNamespaceVFS2().Root()
+		root := ctx.Kernel.GlobalInit().Leader().MountNamespaceVFS2().Root()
 		root.IncRef()
 		return root
 	case vfs.CtxMountNamespace:
-		if ctx.k.globalInit == nil {
+		if ctx.Kernel.globalInit == nil {
 			return nil
 		}
-		mntns := ctx.k.GlobalInit().Leader().MountNamespaceVFS2()
+		mntns := ctx.Kernel.GlobalInit().Leader().MountNamespaceVFS2()
 		mntns.IncRef()
 		return mntns
 	case fs.CtxDirentCacheLimiter:
-		return ctx.k.DirentCacheLimiter
+		return ctx.Kernel.DirentCacheLimiter
 	case inet.CtxStack:
-		return ctx.k.RootNetworkNamespace().Stack()
+		return ctx.Kernel.RootNetworkNamespace().Stack()
 	case ktime.CtxRealtimeClock:
-		return ctx.k.RealtimeClock()
+		return ctx.Kernel.RealtimeClock()
 	case limits.CtxLimits:
 		// No limits apply.
 		return limits.NewLimitSet()
 	case pgalloc.CtxMemoryFile:
-		return ctx.k.mf
+		return ctx.Kernel.mf
 	case pgalloc.CtxMemoryFileProvider:
-		return ctx.k
+		return ctx.Kernel
 	case platform.CtxPlatform:
-		return ctx.k
+		return ctx.Kernel
 	case uniqueid.CtxGlobalUniqueID:
-		return ctx.k.UniqueID()
+		return ctx.Kernel.UniqueID()
 	case uniqueid.CtxGlobalUniqueIDProvider:
-		return ctx.k
+		return ctx.Kernel
 	case uniqueid.CtxInotifyCookie:
-		return ctx.k.GenerateInotifyCookie()
+		return ctx.Kernel.GenerateInotifyCookie()
 	case unimpl.CtxEvents:
-		return ctx.k
+		return ctx.Kernel
 	default:
 		return nil
 	}

pkg/sentry/kernel/pipe/BUILD

@@ -19,7 +19,6 @@ go_library(
     visibility = ["//pkg/sentry:internal"],
     deps = [
         "//pkg/abi/linux",
-        "//pkg/amutex",
         "//pkg/context",
         "//pkg/errors/linuxerr",
         "//pkg/hostarch",

pkg/sentry/kernel/pipe/node.go

@@ -20,7 +20,7 @@ import (
 	"gvisor.dev/gvisor/pkg/errors/linuxerr"
 	"gvisor.dev/gvisor/pkg/sentry/fs"
 	"gvisor.dev/gvisor/pkg/sentry/fs/fsutil"
-	"gvisor.dev/gvisor/pkg/sync"
+	"gvisor.dev/gvisor/pkg/waiter"
 )
 
 // inodeOperations implements fs.InodeOperations for pipes.
@@ -41,34 +41,22 @@ type inodeOperations struct {
 	// even if they have been unlinked. We can get away with this because
 	// their state exists entirely within the sentry.
 	fsutil.InodeVirtual `state:"nosave"`
-
 	fsutil.InodeSimpleAttributes
 
-	// mu protects the fields below.
-	mu sync.Mutex `state:"nosave"`
-
-	// p is the underlying Pipe object representing this fifo.
+	// p is the underlying Pipe object representing this fifo. This field
+	// may have methods called on it, but the pointer is immutable.
 	p *Pipe
-
-	// Channels for synchronizing the creation of new readers and writers of
-	// this fifo. See waitFor and newHandleLocked.
-	//
-	// These are not saved/restored because all waiters are unblocked on save,
-	// and either automatically restart (via ERESTARTSYS) or return EINTR on
-	// resume. On restarts via ERESTARTSYS, the appropriate channel will be
-	// recreated.
-	rWakeup chan struct{} `state:"nosave"`
-	wWakeup chan struct{} `state:"nosave"`
 }
 
 var _ fs.InodeOperations = (*inodeOperations)(nil)
 
 // NewInodeOperations returns a new fs.InodeOperations for a given pipe.
 func NewInodeOperations(ctx context.Context, perms fs.FilePermissions, p *Pipe) *inodeOperations {
-	return &inodeOperations{
+	i := &inodeOperations{
 		InodeSimpleAttributes: fsutil.NewInodeSimpleAttributes(ctx, fs.FileOwnerFromContext(ctx), perms, linux.PIPEFS_MAGIC),
 		p:                     p,
 	}
+	return i
 }
 
 // GetFile implements fs.InodeOperations.GetFile. Named pipes have special blocking
@@ -83,16 +71,11 @@ func NewInodeOperations(ctx context.Context, perms fs.FilePermissions, p *Pipe)
 // leaves this behavior undefined. This can be used to open a FIFO for writing
 // while there are no readers available." - fifo(7)
 func (i *inodeOperations) GetFile(ctx context.Context, d *fs.Dirent, flags fs.FileFlags) (*fs.File, error) {
-	i.mu.Lock()
-	defer i.mu.Unlock()
-
 	switch {
 	case flags.Read && !flags.Write: // O_RDONLY.
 		r := i.p.Open(ctx, d, flags)
-		newHandleLocked(&i.rWakeup)
-
-		if i.p.isNamed && !flags.NonBlocking && !i.p.HasWriters() {
-			if !waitFor(&i.mu, &i.wWakeup, ctx) {
+		for i.p.isNamed && !flags.NonBlocking && !i.p.HasWriters() {
+			if !ctx.BlockOn((*waitQueue)(i.p), waiter.EventInternal) {
 				r.DecRef(ctx)
 				return nil, linuxerr.ErrInterrupted
 			}
@@ -105,17 +88,14 @@ func (i *inodeOperations) GetFile(ctx context.Context, d *fs.Dirent, flags fs.Fi
 
 	case flags.Write && !flags.Read: // O_WRONLY.
 		w := i.p.Open(ctx, d, flags)
-		newHandleLocked(&i.wWakeup)
-
-		if i.p.isNamed && !i.p.HasReaders() {
+		for i.p.isNamed && !i.p.HasReaders() {
 			// On a nonblocking, write-only open, the open fails with ENXIO if the
 			// read side isn't open yet.
 			if flags.NonBlocking {
 				w.DecRef(ctx)
 				return nil, linuxerr.ENXIO
 			}
-
-			if !waitFor(&i.mu, &i.rWakeup, ctx) {
+			if !ctx.BlockOn((*waitQueue)(i.p), waiter.EventInternal) {
 				w.DecRef(ctx)
 				return nil, linuxerr.ErrInterrupted
 			}
@@ -125,8 +105,6 @@ func (i *inodeOperations) GetFile(ctx context.Context, d *fs.Dirent, flags fs.Fi
 	case flags.Read && flags.Write: // O_RDWR.
 		// Pipes opened for read-write always succeeds without blocking.
 		rw := i.p.Open(ctx, d, flags)
-		newHandleLocked(&i.rWakeup)
-		newHandleLocked(&i.wWakeup)
 		return rw, nil
 
 	default:

pkg/sentry/kernel/pipe/node_test.go

@@ -24,33 +24,6 @@ import (
 	"gvisor.dev/gvisor/pkg/sentry/fs"
 )
 
-type sleeper struct {
-	context.Context
-	ch chan struct{}
-}
-
-func newSleeperContext(t *testing.T) context.Context {
-	return &sleeper{
-		Context: contexttest.Context(t),
-		ch:      make(chan struct{}),
-	}
-}
-
-func (s *sleeper) SleepStart() <-chan struct{} {
-	return s.ch
-}
-
-func (s *sleeper) SleepFinish(bool) {
-}
-
-func (s *sleeper) Cancel() {
-	s.ch <- struct{}{}
-}
-
-func (s *sleeper) Interrupted() bool {
-	return len(s.ch) != 0
-}
-
 type openResult struct {
 	*fs.File
 	error
@@ -96,6 +69,7 @@ func newAnonPipe(t *testing.T) *Pipe {
 // blockDuration. This is useful for checking that a goroutine that is supposed
 // to be executing a blocking operation is actually blocking.
 func assertRecvBlocks(t *testing.T, c <-chan struct{}, blockDuration time.Duration, failMsg string) {
+	t.Helper()
 	select {
 	case <-c:
 		t.Fatalf(failMsg)
@@ -105,7 +79,7 @@ func assertRecvBlocks(t *testing.T, c <-chan struct{}, blockDuration time.Durati
 }
 
 func TestReadOpenBlocksForWriteOpen(t *testing.T) {
-	ctx := newSleeperContext(t)
+	ctx := contexttest.Context(t)
 	f := NewInodeOperations(ctx, perms, newNamedPipe(t))
 
 	rDone := make(chan struct{})
@@ -123,7 +97,7 @@ func TestReadOpenBlocksForWriteOpen(t *testing.T) {
 }
 
 func TestWriteOpenBlocksForReadOpen(t *testing.T) {
-	ctx := newSleeperContext(t)
+	ctx := contexttest.Context(t)
 	f := NewInodeOperations(ctx, perms, newNamedPipe(t))
 
 	wDone := make(chan struct{})
@@ -141,7 +115,7 @@ func TestWriteOpenBlocksForReadOpen(t *testing.T) {
 }
 
 func TestMultipleWriteOpenDoesntCountAsReadOpen(t *testing.T) {
-	ctx := newSleeperContext(t)
+	ctx := contexttest.Context(t)
 	f := NewInodeOperations(ctx, perms, newNamedPipe(t))
 
 	rDone1 := make(chan struct{})
@@ -163,7 +137,7 @@ func TestMultipleWriteOpenDoesntCountAsReadOpen(t *testing.T) {
 }
 
 func TestClosedReaderBlocksWriteOpen(t *testing.T) {
-	ctx := newSleeperContext(t)
+	ctx := contexttest.Context(t)
 	f := NewInodeOperations(ctx, perms, newNamedPipe(t))
 
 	rFile, _ := testOpenOrDie(ctx, t, f, fs.FileFlags{Read: true, NonBlocking: true}, nil)
@@ -184,7 +158,7 @@ func TestClosedReaderBlocksWriteOpen(t *testing.T) {
 }
 
 func TestReadWriteOpenNeverBlocks(t *testing.T) {
-	ctx := newSleeperContext(t)
+	ctx := contexttest.Context(t)
 	f := NewInodeOperations(ctx, perms, newNamedPipe(t))
 
 	rwDone := make(chan struct{})
@@ -195,7 +169,7 @@ func TestReadWriteOpenNeverBlocks(t *testing.T) {
 }
 
 func TestReadWriteOpenUnblocksReadOpen(t *testing.T) {
-	ctx := newSleeperContext(t)
+	ctx := contexttest.Context(t)
 	f := NewInodeOperations(ctx, perms, newNamedPipe(t))
 
 	rDone := make(chan struct{})
@@ -209,7 +183,7 @@ func TestReadWriteOpenUnblocksReadOpen(t *testing.T) {
 }
 
 func TestReadWriteOpenUnblocksWriteOpen(t *testing.T) {
-	ctx := newSleeperContext(t)
+	ctx := contexttest.Context(t)
 	f := NewInodeOperations(ctx, perms, newNamedPipe(t))
 
 	wDone := make(chan struct{})
@@ -223,7 +197,7 @@ func TestReadWriteOpenUnblocksWriteOpen(t *testing.T) {
 }
 
 func TestBlockedOpenIsCancellable(t *testing.T) {
-	ctx := newSleeperContext(t)
+	ctx := contexttest.Context(t)
 	f := NewInodeOperations(ctx, perms, newNamedPipe(t))
 
 	done := make(chan openResult)
@@ -235,7 +209,8 @@ func TestBlockedOpenIsCancellable(t *testing.T) {
 		// Ok.
 	}
 
-	ctx.(*sleeper).Cancel()
+	ctx.Interrupt()
+
 	// If the cancel on the sleeper didn't work, the open for read would never
 	// return.
 	res := <-done
@@ -246,7 +221,7 @@ func TestBlockedOpenIsCancellable(t *testing.T) {
 }
 
 func TestNonblockingReadOpenFileNoWriters(t *testing.T) {
-	ctx := newSleeperContext(t)
+	ctx := contexttest.Context(t)
 	f := NewInodeOperations(ctx, perms, newNamedPipe(t))
 
 	if _, err := testOpen(ctx, t, f, fs.FileFlags{Read: true, NonBlocking: true}, nil); err != nil {
@@ -255,7 +230,7 @@ func TestNonblockingReadOpenFileNoWriters(t *testing.T) {
 }
 
 func TestNonblockingWriteOpenFileNoReaders(t *testing.T) {
-	ctx := newSleeperContext(t)
+	ctx := contexttest.Context(t)
 	f := NewInodeOperations(ctx, perms, newNamedPipe(t))
 
 	if _, err := testOpen(ctx, t, f, fs.FileFlags{Write: true, NonBlocking: true}, nil); !linuxerr.Equals(linuxerr.ENXIO, err) {
@@ -264,7 +239,7 @@ func TestNonblockingWriteOpenFileNoReaders(t *testing.T) {
 }
 
 func TestNonBlockingReadOpenWithWriter(t *testing.T) {
-	ctx := newSleeperContext(t)
+	ctx := contexttest.Context(t)
 	f := NewInodeOperations(ctx, perms, newNamedPipe(t))
 
 	wDone := make(chan struct{})
@@ -283,7 +258,7 @@ func TestNonBlockingReadOpenWithWriter(t *testing.T) {
 }
 
 func TestNonBlockingWriteOpenWithReader(t *testing.T) {
-	ctx := newSleeperContext(t)
+	ctx := contexttest.Context(t)
 	f := NewInodeOperations(ctx, perms, newNamedPipe(t))
 
 	rDone := make(chan struct{})
@@ -302,7 +277,7 @@ func TestNonBlockingWriteOpenWithReader(t *testing.T) {
 }
 
 func TestAnonReadOpen(t *testing.T) {
-	ctx := newSleeperContext(t)
+	ctx := contexttest.Context(t)
 	f := NewInodeOperations(ctx, perms, newAnonPipe(t))
 
 	if _, err := testOpen(ctx, t, f, fs.FileFlags{Read: true}, nil); err != nil {
@@ -311,7 +286,7 @@ func TestAnonReadOpen(t *testing.T) {
 }
 
 func TestAnonWriteOpen(t *testing.T) {
-	ctx := newSleeperContext(t)
+	ctx := contexttest.Context(t)
 	f := NewInodeOperations(ctx, perms, newAnonPipe(t))
 
 	if _, err := testOpen(ctx, t, f, fs.FileFlags{Write: true}, nil); err != nil {

pkg/sentry/kernel/pipe/pipe.go

@@ -49,13 +49,42 @@ const (
 	atomicIOBytes = 4096
 )
 
+// waitQueue is a wrapper around Pipe.
+//
+// This is used for ctx.Block operations that require the synchronization of
+// readers and writers, along with the careful grabbing and releasing of locks.
+type waitQueue Pipe
+
+// Readiness implements waiter.Waitable.Readiness.
+func (wq *waitQueue) Readiness(mask waiter.EventMask) waiter.EventMask {
+	return ((*Pipe)(wq)).rwReadiness() & mask
+}
+
+// EventRegister implements waiter.Waitable.EventRegister.
+func (wq *waitQueue) EventRegister(e *waiter.Entry) error {
+	((*Pipe)(wq)).queue.EventRegister(e)
+
+	// Notify synchronously.
+	if ((*Pipe)(wq)).HasReaders() || ((*Pipe)(wq)).HasWriters() {
+		e.NotifyEvent(waiter.EventInternal)
+	}
+
+	return nil
+}
+
+// EventUnregister implements waiter.Waitable.EventUnregister.
+func (wq *waitQueue) EventUnregister(e *waiter.Entry) {
+	((*Pipe)(wq)).queue.EventUnregister(e)
+}
+
 // Pipe is an encapsulation of a platform-independent pipe.
 // It manages a buffered byte queue shared between a reader/writer
 // pair.
 //
 // +stateify savable
 type Pipe struct {
-	waiter.Queue
+	// queue is the waiter queue.
+	queue waiter.Queue
 
 	// isNamed indicates whether this is a named pipe.
 	//
@@ -183,7 +212,8 @@ func (p *Pipe) Open(ctx context.Context, d *fs.Dirent, flags fs.FileFlags) *fs.F
 //
 // peekLocked does not mutate the pipe; if the read consumes bytes from the
 // pipe, then the caller is responsible for calling p.consumeLocked() and
-// p.Notify(waiter.WritableEvents). (The latter must be called with p.mu unlocked.)
+// p.queue.Notify(waiter.WritableEvents). (The latter must be called with p.mu
+// unlocked.)
 //
 // Preconditions:
 // * p.mu must be locked.
@@ -237,7 +267,7 @@ func (p *Pipe) consumeLocked(n int64) {
 // Unlike peekLocked, writeLocked assumes that f returns the number of bytes
 // written to the pipe, and increases the number of bytes stored in the pipe
 // accordingly. Callers are still responsible for calling
-// p.Notify(waiter.ReadableEvents) with p.mu unlocked.
+// p.queue.Notify(waiter.ReadableEvents) with p.mu unlocked.
 //
 // Preconditions:
 // * p.mu must be locked.
@@ -315,28 +345,32 @@ func (p *Pipe) writeLocked(count int64, f func(safemem.BlockSeq) (uint64, error)
 // rOpen signals a new reader of the pipe.
 func (p *Pipe) rOpen() {
 	atomic.AddInt32(&p.readers, 1)
+
+	// Notify for blocking openers.
+	p.queue.Notify(waiter.EventInternal)
 }
 
 // wOpen signals a new writer of the pipe.
 func (p *Pipe) wOpen() {
 	p.mu.Lock()
-	defer p.mu.Unlock()
 	p.hadWriter = true
 	atomic.AddInt32(&p.writers, 1)
+	p.mu.Unlock()
+
+	// Notify for blocking openers.
+	p.queue.Notify(waiter.EventInternal)
 }
 
 // rClose signals that a reader has closed their end of the pipe.
 func (p *Pipe) rClose() {
-	newReaders := atomic.AddInt32(&p.readers, -1)
-	if newReaders < 0 {
+	if newReaders := atomic.AddInt32(&p.readers, -1); newReaders < 0 {
 		panic(fmt.Sprintf("Refcounting bug, pipe has negative readers: %v", newReaders))
 	}
 }
 
 // wClose signals that a writer has closed their end of the pipe.
 func (p *Pipe) wClose() {
-	newWriters := atomic.AddInt32(&p.writers, -1)
-	if newWriters < 0 {
+	if newWriters := atomic.AddInt32(&p.writers, -1); newWriters < 0 {
 		panic(fmt.Sprintf("Refcounting bug, pipe has negative writers: %v.", newWriters))
 	}
 }
@@ -407,6 +441,17 @@ func (p *Pipe) rwReadiness() waiter.EventMask {
 	return p.rReadinessLocked() | p.wReadinessLocked()
 }
 
+// EventRegister implements waiter.Waitable.EventRegister.
+func (p *Pipe) EventRegister(e *waiter.Entry) error {
+	p.queue.EventRegister(e)
+	return nil
+}
+
+// EventUnregister implements waiter.Waitable.EventUnregister.
+func (p *Pipe) EventUnregister(e *waiter.Entry) {
+	p.queue.EventUnregister(e)
+}
+
 // queued returns the amount of queued data.
 func (p *Pipe) queued() int64 {
 	p.mu.Lock()
@@ -444,9 +489,3 @@ func (p *Pipe) SetFifoSize(size int64) (int64, error) {
 	p.max = size
 	return size, nil
 }
-
-// EventRegister implements waiter.Waitable.
-func (p *Pipe) EventRegister(e *waiter.Entry) error {
-	p.Queue.EventRegister(e)
-	return nil
-}

pkg/sentry/kernel/pipe/pipe_util.go

@@ -20,13 +20,11 @@ import (
 
 	"golang.org/x/sys/unix"
 	"gvisor.dev/gvisor/pkg/abi/linux"
-	"gvisor.dev/gvisor/pkg/amutex"
 	"gvisor.dev/gvisor/pkg/context"
 	"gvisor.dev/gvisor/pkg/errors/linuxerr"
 	"gvisor.dev/gvisor/pkg/marshal/primitive"
 	"gvisor.dev/gvisor/pkg/safemem"
 	"gvisor.dev/gvisor/pkg/sentry/arch"
-	"gvisor.dev/gvisor/pkg/sync"
 	"gvisor.dev/gvisor/pkg/usermem"
 	"gvisor.dev/gvisor/pkg/waiter"
 )
@@ -40,14 +38,14 @@ func (p *Pipe) Release(context.Context) {
 	p.wClose()
 
 	// Wake up readers and writers.
-	p.Notify(waiter.ReadableEvents | waiter.WritableEvents)
+	p.queue.Notify(waiter.ReadableEvents | waiter.WritableEvents)
 }
 
 // Read reads from the Pipe into dst.
 func (p *Pipe) Read(ctx context.Context, dst usermem.IOSequence) (int64, error) {
 	n, err := dst.CopyOutFrom(ctx, p)
 	if n > 0 {
-		p.Notify(waiter.WritableEvents)
+		p.queue.Notify(waiter.WritableEvents)
 	}
 	return n, err
 }
@@ -76,7 +74,7 @@ func (p *Pipe) WriteTo(ctx context.Context, w io.Writer, count int64, dup bool)
 		return safemem.FromIOWriter{w}.WriteFromBlocks(srcs)
 	}, !dup /* removeFromSrc */)
 	if n > 0 && !dup {
-		p.Notify(waiter.WritableEvents)
+		p.queue.Notify(waiter.WritableEvents)
 	}
 	return n, err
 }
@@ -85,7 +83,7 @@ func (p *Pipe) WriteTo(ctx context.Context, w io.Writer, count int64, dup bool)
 func (p *Pipe) Write(ctx context.Context, src usermem.IOSequence) (int64, error) {
 	n, err := src.CopyInTo(ctx, p)
 	if n > 0 {
-		p.Notify(waiter.ReadableEvents)
+		p.queue.Notify(waiter.ReadableEvents)
 	}
 	if linuxerr.Equals(linuxerr.EPIPE, err) {
 		// If we are returning EPIPE send SIGPIPE to the task.
@@ -116,7 +114,7 @@ func (p *Pipe) ReadFrom(ctx context.Context, r io.Reader, count int64) (int64, e
 		return safemem.FromIOReader{r}.ReadToBlocks(dsts)
 	})
 	if n > 0 {
-		p.Notify(waiter.ReadableEvents)
+		p.queue.Notify(waiter.ReadableEvents)
 	}
 	return n, err
 }
@@ -149,63 +147,3 @@ func (p *Pipe) Ioctl(ctx context.Context, io usermem.IO, args arch.SyscallArgume
 		return 0, unix.ENOTTY
 	}
 }
-
-// waitFor blocks until the underlying pipe has at least one reader/writer is
-// announced via 'wakeupChan', or until 'sleeper' is cancelled. Any call to this
-// function will block for either readers or writers, depending on where
-// 'wakeupChan' points.
-//
-// mu must be held by the caller. waitFor returns with mu held, but it will
-// drop mu before blocking for any reader/writers.
-// +checklocks:mu
-func waitFor(mu *sync.Mutex, wakeupChan *chan struct{}, sleeper amutex.Sleeper) bool {
-	// Ideally this function would simply use a condition variable. However, the
-	// wait needs to be interruptible via 'sleeper', so we must sychronize via a
-	// channel. The synchronization below relies on the fact that closing a
-	// channel unblocks all receives on the channel.
-
-	// Does an appropriate wakeup channel already exist? If not, create a new
-	// one. This is all done under f.mu to avoid races.
-	if *wakeupChan == nil {
-		*wakeupChan = make(chan struct{})
-	}
-
-	// Grab a local reference to the wakeup channel since it may disappear as
-	// soon as we drop f.mu.
-	wakeup := *wakeupChan
-
-	// Drop the lock and prepare to sleep.
-	mu.Unlock()
-	cancel := sleeper.SleepStart()
-
-	// Wait for either a new reader/write to be signalled via 'wakeup', or
-	// for the sleep to be cancelled.
-	select {
-	case <-wakeup:
-		sleeper.SleepFinish(true)
-	case <-cancel:
-		sleeper.SleepFinish(false)
-	}
-
-	// Take the lock and check if we were woken. If we were woken and
-	// interrupted, the former takes priority.
-	mu.Lock()
-	select {
-	case <-wakeup:
-		return true
-	default:
-		return false
-	}
-}
-
-// newHandleLocked signals a new pipe reader or writer depending on where
-// 'wakeupChan' points. This unblocks any corresponding reader or writer
-// waiting for the other end of the channel to be opened, see Fifo.waitFor.
-//
-// Precondition: the mutex protecting wakeupChan must be held.
-func newHandleLocked(wakeupChan *chan struct{}) {
-	if *wakeupChan != nil {
-		close(*wakeupChan)
-		*wakeupChan = nil
-	}
-}