systrap: disable the fast path in stub threads when it isn't effective

Here are two conditions when the fast path is effective:
* the side that has to change a state is running on CPU when another side
  is polling the state. This is why the Sentry threads have a higher
  priority than stub threads.
* The sentry handles events faster than the overhead of scheduling another
  stub thread.

This patch addresses the second condition. The fast path in stub threads is
disabled when we reach the limit of stub threads. The idea is that more stub
threads can generate more events to the sentry.

PiperOrigin-RevId: 529120681

pkg/sentry/platform/systrap/BUILD

@@ -18,6 +18,18 @@ go_template_instance(
     },
 )
 
+go_template_instance(
+    name = "subprocess_list",
+    out = "subprocess_list.go",
+    package = "systrap",
+    prefix = "subprocess",
+    template = "//pkg/ilist:generic_list",
+    types = {
+        "Element": "*subprocess",
+        "Linker": "*subprocess",
+    },
+)
+
 go_template_instance(
     name = "subprocess_refs",
     out = "subprocess_refs.go",
@@ -51,6 +63,7 @@ go_library(
         "subprocess_arm64.go",
         "subprocess_linux.go",
         "subprocess_linux_unsafe.go",
+        "subprocess_list.go",
         "subprocess_pool.go",
         "subprocess_refs.go",
         "subprocess_unsafe.go",

pkg/sentry/platform/systrap/context_queue.go

@@ -50,6 +50,7 @@ type contextQueue struct {
 
 	fastPathDisabledTS  uint64
 	fastPathFailedInRow uint32
+	fastPathDisabled    uint32
 	ringbuffer          [maxContextQueueEntries]uint32
 }
 
@@ -66,6 +67,7 @@ func (q *contextQueue) init() {
 	atomic.StoreUint32(&q.numActiveThreads, 0)
 	atomic.StoreUint32(&q.numActiveContexts, 0)
 	atomic.StoreUint32(&q.numAwakeContexts, 0)
+	atomic.StoreUint32(&q.fastPathDisabled, 0)
 }
 
 func (q *contextQueue) isEmpty() bool {
@@ -76,12 +78,13 @@ func (q *contextQueue) queuedContexts() uint32 {
 	return (atomic.LoadUint32(&q.end) + maxContextQueueEntries - atomic.LoadUint32(&q.start)) % maxContextQueueEntries
 }
 
-func (q *contextQueue) add(ctx *sharedContext) uint32 {
-	contextID := ctx.contextID
-	if ctx.sleeping {
-		ctx.sleeping = false
-		atomic.AddUint32(&q.numAwakeContexts, 1)
+func (q *contextQueue) add(ctx *sharedContext, stubFastPathEnabled bool) uint32 {
+	if stubFastPathEnabled {
+		q.enableFastPath()
+	} else {
+		q.disableFastPath()
 	}
+	contextID := ctx.contextID
 	atomic.AddUint32(&q.numActiveContexts, 1)
 	next := atomic.AddUint32(&q.end, 1)
 	if (next % maxContextQueueEntries) ==
@@ -93,3 +96,11 @@ func (q *contextQueue) add(ctx *sharedContext) uint32 {
 	atomic.StoreUint32(&q.ringbuffer[next], contextID)
 	return next // remove me
 }
+
+func (q *contextQueue) disableFastPath() {
+	atomic.StoreUint32(&q.fastPathDisabled, 1)
+}
+
+func (q *contextQueue) enableFastPath() {
+	atomic.StoreUint32(&q.fastPathDisabled, 0)
+}

pkg/sentry/platform/systrap/shared_context.go

@@ -16,6 +16,7 @@ package systrap
 
 import (
 	"fmt"
+	"runtime"
 	"sync"
 	"sync/atomic"
 
@@ -99,7 +100,7 @@ func (sc *sharedContext) release() {
 		return
 	}
 	if !sc.sleeping {
-		atomic.AddUint32(&sc.subprocess.contextQueue.numAwakeContexts, ^uint32(0))
+		sc.subprocess.decAwakeContexts()
 
 	}
 	sc.subprocess.threadContextPool.Put(uint64(sc.contextID))
@@ -199,8 +200,7 @@ func (sc *sharedContext) sleepOnState(state sysmsg.ContextState) {
 	}
 }
 
-type fastPathContextQueue struct {
-
+type fastPathDispatcher struct {
 	// list is used only from the loop method and so it isn't protected by
 	// any lock.
 	list contextList
@@ -214,15 +214,84 @@ type fastPathContextQueue struct {
 	// entrants contains new contexts that haven't been added to `list` yet.
 	// +checklocks:mu
 	entrants contextList
+
+	// fastPathDisabledTS is the time stamp when the stub fast path was
+	// disabled. It is zero if the fast path is enabled.
+	fastPathDisabledTS atomic.Uint64
+
+	subprocessListMu sync.Mutex
+	// subprocessList contains subprocesses with at least one awake context.
+	// +checklocks:subprocessListMu
+	subprocessList subprocessList
 }
 
-var dispatcher fastPathContextQueue
+var dispatcher fastPathDispatcher
+
+// fastPathContextLimit is the maximum number of contexts after which the fast
+// path in stub threads is disabled. Its value can be higher than the number of
+// CPU-s, because the Sentry is running with higher priority than stub threads,
+// deepSleepTimeout is much shorter than the Linux scheduler timeslice, so the
+// only thing that matters here is whether the Sentry handles syscall faster
+// than the overhead of scheduling another stub thread.
+var fastPathContextLimit = uint32(runtime.GOMAXPROCS(0) * 2)
+
+// fastPathDisabledTimeout is the timeout after which the fast path in stub
+// processes will be re-enabled.
+const fastPathDisabledTimeout = uint64(200 * 1000 * 1000) // 100ms for 2GHz.
+
+// nrMaxAwakeStubThreads is the maximum number of awake stub threads over all
+// subprocesses at the this moment.
+var nrMaxAwakeStubThreads atomic.Uint32
+
+// stubFastPathEnabled returns true if the fast path in stub processes is
+// enabled. If the fast path is disabled, it revises whether it has to be
+// re-enabled or not.
+func (q *fastPathDispatcher) stubFastPathEnabled() bool {
+	ts := q.fastPathDisabledTS.Load()
+	if ts != 0 {
+		if uint64(cputicks())-ts < fastPathDisabledTimeout {
+			return false
+		}
+		if nrMaxAwakeStubThreads.Load() > fastPathContextLimit {
+			q.fastPathDisabledTS.Store(uint64(cputicks()))
+			return false
+		}
+		q.fastPathDisabledTS.Store(0)
+	}
+	return true
+}
+
+// disableStubFastPath disables the fast path over all subprocesses with active
+// contexts.
+func (q *fastPathDispatcher) disableStubFastPath() {
+	q.subprocessListMu.Lock()
+	defer q.subprocessListMu.Unlock()
+
+	for s := q.subprocessList.Front(); s != nil; s = s.Next() {
+		s.contextQueue.disableFastPath()
+	}
+	q.fastPathDisabledTS.Store(uint64(cputicks()))
+}
+
+func (q *fastPathDispatcher) activateSubprocess(s *subprocess) {
+	q.subprocessListMu.Lock()
+	defer q.subprocessListMu.Unlock()
+
+	q.subprocessList.PushBack(s)
+}
+
+func (q *fastPathDispatcher) deactivateSubprocess(s *subprocess) {
+	q.subprocessListMu.Lock()
+	defer q.subprocessListMu.Unlock()
+
+	q.subprocessList.Remove(s)
+}
 
 // loop is processing contexts in the queue. Only one instance of it can be
 // running, because it has exclusive access to the list.
 //
 // target is the context associated with the current go-routine.
-func (q *fastPathContextQueue) loop(target *sharedContext) {
+func (q *fastPathDispatcher) loop(target *sharedContext) {
 	done := false
 	processed := 0
 	slowPath := false
@@ -287,7 +356,7 @@ func (q *fastPathContextQueue) loop(target *sharedContext) {
 	}
 }
 
-func (q *fastPathContextQueue) waitFor(ctx *sharedContext) syncevent.Set {
+func (q *fastPathDispatcher) waitFor(ctx *sharedContext) syncevent.Set {
 	events := syncevent.Set(0)
 
 	q.mu.Lock()

pkg/sentry/platform/systrap/subprocess.go

@@ -121,6 +121,7 @@ const (
 type subprocess struct {
 	platform.NoAddressSpaceIO
 	subprocessRefs
+	subprocessEntry
 
 	// requests is used to signal creation of new threads.
 	requests chan any
@@ -690,6 +691,33 @@ func (t *thread) NotifyInterrupt() {
 	unix.Tgkill(int(t.tgid), int(t.tid), unix.Signal(platform.SignalInterrupt))
 }
 
+func (s *subprocess) incAwakeContexts() {
+	nr := atomic.AddUint32(&s.contextQueue.numAwakeContexts, 1)
+	if nr > uint32(maxSysmsgThreads) {
+		return
+	}
+	if nr == 1 {
+		dispatcher.activateSubprocess(s)
+	}
+	nr = nrMaxAwakeStubThreads.Add(1)
+	if nr > fastPathContextLimit {
+		if dispatcher.stubFastPathEnabled() {
+			dispatcher.disableStubFastPath()
+		}
+	}
+}
+
+func (s *subprocess) decAwakeContexts() {
+	nr := atomic.AddUint32(&s.contextQueue.numAwakeContexts, ^uint32(0))
+	if nr >= uint32(maxSysmsgThreads) {
+		return
+	}
+	if nr == 0 {
+		dispatcher.deactivateSubprocess(s)
+	}
+	nrMaxAwakeStubThreads.Add(^uint32(0))
+}
+
 // switchToApp is called from the main SwitchToApp entrypoint.
 //
 // This function returns true on a system call, false on a signal.
@@ -725,9 +753,14 @@ func (s *subprocess) switchToApp(c *context, ac *arch.Context64) (isSyscall bool
 		restoreFPState(nil, ctx, 0, c, ac)
 
 		// Place the context onto the context queue.
+		if ctx.sleeping {
+			ctx.sleeping = false
+			s.incAwakeContexts()
+		}
+		stubFastPathEnabled := dispatcher.stubFastPathEnabled()
 		ctx.setState(sysmsg.ContextStateNone)
-		s.contextQueue.add(ctx)
-		s.waitOnState(ctx)
+		s.contextQueue.add(ctx, stubFastPathEnabled)
+		s.waitOnState(ctx, stubFastPathEnabled)
 
 		// Check if there's been an error.
 		threadID := ctx.threadID()
@@ -788,17 +821,12 @@ func (s *subprocess) switchToApp(c *context, ac *arch.Context64) (isSyscall bool
 	return false, false, nil
 }
 
-const (
-	fastPathFailedInRowLimit = 5
-	fastPathDisabledTimeout  = 20 * 1000 * 1000 // 10ms
-)
-
-func (s *subprocess) waitOnState(ctx *sharedContext) {
+func (s *subprocess) waitOnState(ctx *sharedContext, stubFastPathEnabled bool) {
 	ctx.kicked = false
 	slowPath := false
 	start := cputicks()
 	ctx.startWaitingTS = start
-	if atomic.LoadUint32(&s.contextQueue.numActiveThreads) == 0 {
+	if !stubFastPathEnabled || atomic.LoadUint32(&s.contextQueue.numActiveThreads) == 0 {
 		ctx.kicked = s.kickSysmsgThread()
 	}
 	for curState := ctx.state(); curState == sysmsg.ContextStateNone; curState = ctx.state() {

pkg/sentry/platform/systrap/sysmsg/sysmsg_lib.c

@@ -44,6 +44,7 @@ struct context_queue {
   uint32_t num_awake_contexts;
   uint64_t fast_path_disalbed_ts;
   uint32_t fast_path_failed_in_row;
+  uint32_t fast_path_disabled;
   uint32_t ringbuffer[MAX_CONTEXT_QUEUE_ENTRIES];
 };
 
@@ -220,6 +221,11 @@ static struct thread_context *get_context_fast(struct sysmsg *sysmsg,
       return ctx;
     }
 
+    if (__atomic_load_n(&queue->fast_path_disabled, __ATOMIC_ACQUIRE) != 0 &&
+        spinning_queue_remove_first(0)) {
+      break;
+    }
+
     nr_active_threads =
         __atomic_load_n(&queue->num_active_threads, __ATOMIC_ACQUIRE);
     nr_awake_contexts =
@@ -287,6 +293,9 @@ struct thread_context *get_context(struct sysmsg *sysmsg) {
         fast_path_enabled = false;
       }
     }
+    if (__atomic_load_n(&queue->fast_path_disabled, __ATOMIC_ACQUIRE) != 0) {
+      fast_path_enabled = false;
+    }
 
     nr_active_threads = NR_IF_THREAD_IS_ACTIVE;
     if (fast_path_enabled) {

pkg/sentry/platform/systrap/systrap.go

@@ -52,7 +52,6 @@ import (
 	"fmt"
 	"os"
 	"sync"
-	"sync/atomic"
 
 	"gvisor.dev/gvisor/pkg/abi/linux"
 	pkgcontext "gvisor.dev/gvisor/pkg/context"
@@ -295,10 +294,9 @@ func (c *context) PrepareSleep() {
 		if ctx == nil {
 			return
 		}
-		s := ctx.subprocess
 		if !ctx.sleeping {
 			ctx.sleeping = true
-			atomic.AddUint32(&s.contextQueue.numAwakeContexts, ^uint32(0))
+			ctx.subprocess.decAwakeContexts()
 		}
 		return
 	} else if c.sysmsgThread != nil {