systrap: introduce a spinning queue

The spinning queue is a queue of spinning threads. It solves the
fragmentation problem. The idea is to minimize the number of threads
processing requests. We can't control how system threads are scheduled, so
can't distribute requests efficiently. The spinning queue emulates virtual
threads sorted by their spinning time.

PiperOrigin-RevId: 518470754

pkg/sentry/platform/systrap/stub_unsafe.go

@@ -134,6 +134,7 @@ func stubInit() {
 	// Add a guard page.
 	mapLen += hostarch.PageSize
 	stubSysmsgStack = mapLen
+
 	// Allocate maxGuestThreads plus ONE because each per-thread stack
 	// has to be aligned to sysmsg.PerThreadMemSize.
 	// Look at sysmsg/sighandler.c:sysmsg_addr() for more details.
@@ -144,6 +145,9 @@ func stubInit() {
 		stubContextQueueRegion = mapLen
 		stubContextQueueRegionLen, _ = hostarch.PageRoundUp(unsafe.Sizeof(contextQueue{}))
 		mapLen += stubContextQueueRegionLen
+
+		stubSpinningThreadQueueAddr = mapLen
+		mapLen += sysmsg.SpinningQueueMemSize
 	}
 
 	// Allocate thread context region
@@ -200,6 +204,7 @@ func stubInit() {
 	stubSysmsgStack += stubStart
 	stubROMapEnd += stubStart
 	stubContextQueueRegion += stubStart
+	stubSpinningThreadQueueAddr += stubStart
 	stubContextRegion += stubStart
 
 	// Align stubSysmsgStack to the per-thread stack size.
@@ -224,6 +229,9 @@ func stubInit() {
 		*exp = 1
 		contextQueue := (*uint64)(unsafe.Pointer(stubSysmsgStart + uintptr(sysmsg.Sighandler_blob_offset____export_context_queue_addr)))
 		*contextQueue = uint64(stubContextQueueRegion)
+
+		p = (*uint64)(unsafe.Pointer(stubSysmsgStart + uintptr(sysmsg.Sighandler_blob_offset____export_spinning_queue_addr)))
+		*p = uint64(stubSpinningThreadQueueAddr)
 	}
 
 	prepareSeccompRules(stubSysmsgStart, stubSysmsgRules, stubSysmsgRulesLen)

pkg/sentry/platform/systrap/subprocess.go

@@ -311,6 +311,7 @@ func newSubprocess(create func() (*thread, error), memoryFile *pgalloc.MemoryFil
 	sp.unmap()
 	sp.usertrap = usertrap.New()
 	sp.mapSharedRegions()
+	sp.mapPrivateRegions()
 
 	// Create the initial sysmsg thread.
 	if contextDecouplingExp {
@@ -391,6 +392,22 @@ func (s *subprocess) mapSharedRegions() {
 	s.threadContextRegion = sentryThreadContextRegionAddr
 }
 
+func (s *subprocess) mapPrivateRegions() {
+	if contextDecouplingExp {
+		_, err := s.syscallThread.syscall(
+			unix.SYS_MMAP,
+			arch.SyscallArgument{Value: uintptr(stubSpinningThreadQueueAddr)},
+			arch.SyscallArgument{Value: uintptr(sysmsg.SpinningQueueMemSize)},
+			arch.SyscallArgument{Value: uintptr(unix.PROT_READ | unix.PROT_WRITE)},
+			arch.SyscallArgument{Value: uintptr(unix.MAP_PRIVATE | unix.MAP_ANONYMOUS | unix.MAP_FIXED)},
+			arch.SyscallArgument{Value: 0},
+			arch.SyscallArgument{Value: 0})
+		if err != nil {
+			panic(fmt.Sprintf("failed to mmap spinning queue region into syscall thread: %v", err))
+		}
+	}
+}
+
 // unmap unmaps non-stub regions of the process.
 //
 // This will panic on failure (which should never happen).

pkg/sentry/platform/systrap/sysmsg/sysmsg.go

@@ -65,6 +65,9 @@ const (
 	// MsgOffsetFromStack is the offset of the Msg structure on
 	// the thread stack.
 	MsgOffsetFromSharedStack = PerThreadMemSize - hostarch.PageSize - PerThreadSharedStackOffset
+
+	// SpinningQueueMemSize is the size of a spinning queue memory region.
+	SpinningQueueMemSize = hostarch.PageSize
 )
 
 // StackAddrToMsg returns an address of a sysmsg structure.

pkg/sentry/platform/systrap/sysmsg/sysmsg.h

@@ -111,6 +111,7 @@ struct thread_context {
 #define GUARD_SIZE (PAGE_SIZE)
 #define MSG_OFFSET_FROM_START (PER_THREAD_MEM_SIZE - PAGE_SIZE)
 
+#define SPINNING_QUEUE_MEM_SIZE PAGE_SIZE
 // LINT.ThenChange(sysmsg.go)
 
 #define FAULT_OPCODE 0x06  // "push %es" on x32 and invalid opcode on x64.
@@ -123,7 +124,8 @@ extern uint64_t __export_pr_sched_core;
 extern uint64_t __export_deep_sleep_timeout;
 extern struct arch_state __export_arch_state;
 extern uint64_t __export_context_decoupling_exp;
-extern uint64_t __export_context_queue_addr;
+struct context_queue;
+extern struct context_queue *__export_context_queue_addr;
 
 // NOLINTBEGIN(runtime/int)
 static void *sysmsg_sp() {

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

@@ -17,6 +17,7 @@
 #include <linux/futex.h>
 #include <linux/unistd.h>
 #include <signal.h>
+#include <stdbool.h>
 #include <stddef.h>
 #include <stdint.h>
 #include <stdlib.h>
@@ -27,7 +28,6 @@
 // polling and fall asleep.
 uint64_t __export_deep_sleep_timeout;
 uint64_t __export_handshake_timeout;
-uint64_t __export_context_queue_addr;
 
 // LINT.IfChange
 #define MAX_STUB_THREADS (4096)
@@ -44,6 +44,9 @@ struct context_queue {
   uint32_t num_sleeping_threads;
   uint32_t ringbuffer[MAX_CONTEXT_QUEUE_ENTRIES];
 };
+
+struct context_queue *__export_context_queue_addr;
+
 // LINT.ThenChange(../context_queue.go)
 
 uint32_t is_empty(struct context_queue *queue) {
@@ -82,41 +85,132 @@ void memcpy(uint8_t *dest, uint8_t *src, size_t n) {
   }
 }
 
+// The spinning queue is a queue of spinning threads. It solves the
+// fragmentation problem. The idea is to minimize the number of threads
+// processing requests. We can't control how system threads are scheduled, so
+// can't distribute requests efficiently. The spinning queue emulates virtual
+// threads sorted by their spinning time.
+//
+// This queue is lock-less to be sure that any thread scheduled out
+// from CPU doesn't block others.
+#define SPINNING_QUEUE_SIZE 128
+
+// MAX_SPINNING_THREADS is half of SPINNING_QUEUE_SIZE to be sure that the tail
+// doesn't catch the head. More details are in spinning_queue_remove_first.
+#define MAX_SPINNING_THREADS (SPINNING_QUEUE_SIZE / 2)
+struct spinning_queue {
+  uint32_t start;
+  uint32_t end;
+  uint64_t start_times[SPINNING_QUEUE_SIZE];
+};
+
+struct spinning_queue *__export_spinning_queue_addr;
+
+// spinning_queue_push adds a new thread to the queue. It returns false if the
+// queue if full.
+static bool spinning_queue_push() __attribute__((warn_unused_result));
+static bool spinning_queue_push(void) {
+  struct spinning_queue *queue = __export_spinning_queue_addr;
+  uint32_t idx, start, end;
+
+  BUILD_BUG_ON(sizeof(struct spinning_queue) > SPINNING_QUEUE_MEM_SIZE);
+
+  end = __atomic_add_fetch(&queue->end, 1, __ATOMIC_SEQ_CST);
+  start = __atomic_load_n(&queue->start, __ATOMIC_SEQ_CST);
+  if (end - start > MAX_SPINNING_THREADS) {
+    __atomic_sub_fetch(&queue->end, 1, __ATOMIC_SEQ_CST);
+    return false;
+  }
+
+  idx = end - 1;
+  __atomic_store_n(&queue->start_times[idx % SPINNING_QUEUE_SIZE], rdtsc(),
+                   __ATOMIC_SEQ_CST);
+  return true;
+}
+
+// spinning_queue_pop() removes one thread from a queue that has been spinning
+// the shortest time.
+static void spinning_queue_pop() {
+  struct spinning_queue *queue = __export_spinning_queue_addr;
+
+  __atomic_add_fetch(&queue->end, -1, __ATOMIC_SEQ_CST);
+}
+
+// spinning_queue_remove_first removes one thread from a queue that has been
+// spinning longer than others and longer than a specified timeout.
+//
+// Returns true if one thread has been removed from the queue.
+static bool spinning_queue_remove_first(uint64_t timeout)
+    __attribute__((warn_unused_result));
+static bool spinning_queue_remove_first(uint64_t timeout) {
+  struct spinning_queue *queue = __export_spinning_queue_addr;
+  uint64_t ts;
+  uint32_t idx;
+
+  idx = __atomic_load_n(&queue->start, __ATOMIC_SEQ_CST);
+  ts = __atomic_load_n(&queue->start_times[idx % SPINNING_QUEUE_SIZE],
+                       __ATOMIC_SEQ_CST);
+  if (ts == 0 || rdtsc() - ts < timeout) return false;
+
+  // The current thread is still in a queue and the length of the queue is twice
+  // of the maximum number of threads, so we can zero the element and be sure
+  // that nobody is trying to set it in a  non-zero value.
+  __atomic_store_n(&queue->start_times[idx % SPINNING_QUEUE_SIZE], 0,
+                   __ATOMIC_SEQ_CST);
+  if (!__atomic_compare_exchange_n(&queue->start, &idx, idx + 1, false,
+                                   __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
+    return false;
+  }
+
+  return true;
+}
+
+struct thread_context *queue_get_context(struct sysmsg *sysmsg) {
+  struct context_queue *queue = __export_context_queue_addr;
+
+  while (!is_empty(queue)) {
+    uint32_t next = __atomic_load_n(&queue->start, __ATOMIC_ACQUIRE) %
+                    MAX_CONTEXT_QUEUE_ENTRIES;
+    uint32_t context_id = __atomic_exchange_n(
+        &queue->ringbuffer[next], INVALID_CONTEXT_ID, __ATOMIC_ACQ_REL);
+
+    if (context_id == INVALID_CONTEXT_ID) continue;
+
+    __atomic_add_fetch(&queue->start, 1, __ATOMIC_ACQ_REL);
+    if (context_id > MAX_STUB_THREADS) {
+      panic(context_id);
+    }
+    sysmsg->context_id = context_id;
+    struct thread_context *ctx = thread_context_addr(sysmsg);
+    __atomic_store_n(&ctx->acked, 1, __ATOMIC_RELEASE);
+    __atomic_store_n(&ctx->thread_id, sysmsg->thread_id, __ATOMIC_RELEASE);
+    return ctx;
+  }
+  return NULL;
+}
+
 // get_context retrieves a context that is ready to be restored to the user.
 // This populates sysmsg->thread_context_id.
 struct thread_context *get_context(struct sysmsg *sysmsg) {
-  struct context_queue *queue =
-      (struct context_queue *)(__export_context_queue_addr);
+  struct context_queue *queue = __export_context_queue_addr;
+
   for (;;) {
+    struct thread_context *ctx;
+
     // Change sysmsg thread state just to indicate thread is not asleep.
     __atomic_store_n(&sysmsg->state, THREAD_STATE_PREP, __ATOMIC_RELEASE);
-    unsigned long start = rdtsc();
-    for (;;) {
-      if (!is_empty(queue)) {
-        uint32_t next = __atomic_load_n(&queue->start, __ATOMIC_ACQUIRE) %
-                        MAX_CONTEXT_QUEUE_ENTRIES;
-        uint32_t context_id = __atomic_exchange_n(
-            &queue->ringbuffer[next], INVALID_CONTEXT_ID, __ATOMIC_ACQ_REL);
-        if (context_id != INVALID_CONTEXT_ID) {
-          __atomic_add_fetch(&queue->start, 1, __ATOMIC_ACQ_REL);
-          if (context_id > MAX_STUB_THREADS) {
-            panic(context_id);
-          }
-          sysmsg->context_id = context_id;
-          struct thread_context *ctx = thread_context_addr(sysmsg);
-          __atomic_store_n(&ctx->acked, 1, __ATOMIC_RELEASE);
-          __atomic_store_n(&ctx->thread_id, sysmsg->thread_id,
-                           __ATOMIC_RELEASE);
+    ctx = queue_get_context(sysmsg);
+    if (ctx) return ctx;
+    if (spinning_queue_push()) {
+      while (!spinning_queue_remove_first(__export_deep_sleep_timeout)) {
+        ctx = queue_get_context(sysmsg);
+        if (ctx) {
+          spinning_queue_pop();
           return ctx;
-        } else {
-          continue;
         }
-      }
-      if ((rdtsc() - start) > __export_deep_sleep_timeout) {
-        break;
-      }
 
-      spinloop();
+        spinloop();
+      }
     }
     __atomic_store_n(&sysmsg->state, THREAD_STATE_ASLEEP, __ATOMIC_RELEASE);
 

pkg/sentry/platform/systrap/systrap.go

@@ -88,6 +88,9 @@ var (
 	stubSysmsgRules    uintptr
 	stubSysmsgRulesLen uintptr
 
+	stubSpinningThreadQueueAddr uintptr
+	stubSpinningThreadQueueSize uintptr
+
 	// stubROMapEnd is the end address of the read-only stub region that
 	// contains the code and precompiled seccomp rules.
 	stubROMapEnd uintptr