From 0cbe6fc835840b3b99b458229eb785e9859faf0d Mon Sep 17 00:00:00 2001 From: Andrei Vagin Date: Tue, 21 Mar 2023 22:08:06 -0700 Subject: [PATCH] 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 | 8 + pkg/sentry/platform/systrap/subprocess.go | 17 ++ pkg/sentry/platform/systrap/sysmsg/sysmsg.go | 3 + pkg/sentry/platform/systrap/sysmsg/sysmsg.h | 4 +- .../platform/systrap/sysmsg/sysmsg_lib.c | 148 ++++++++++++++---- pkg/sentry/platform/systrap/systrap.go | 3 + 6 files changed, 155 insertions(+), 28 deletions(-) diff --git a/pkg/sentry/platform/systrap/stub_unsafe.go b/pkg/sentry/platform/systrap/stub_unsafe.go index 2f8aac0d7..7db687dc6 100644 --- a/pkg/sentry/platform/systrap/stub_unsafe.go +++ b/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) diff --git a/pkg/sentry/platform/systrap/subprocess.go b/pkg/sentry/platform/systrap/subprocess.go index b520b1eaa..70ebfe760 100644 --- a/pkg/sentry/platform/systrap/subprocess.go +++ b/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). diff --git a/pkg/sentry/platform/systrap/sysmsg/sysmsg.go b/pkg/sentry/platform/systrap/sysmsg/sysmsg.go index 7448a1232..7c1e8fa8f 100644 --- a/pkg/sentry/platform/systrap/sysmsg/sysmsg.go +++ b/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. diff --git a/pkg/sentry/platform/systrap/sysmsg/sysmsg.h b/pkg/sentry/platform/systrap/sysmsg/sysmsg.h index 445d2cbbc..16d973282 100644 --- a/pkg/sentry/platform/systrap/sysmsg/sysmsg.h +++ b/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() { diff --git a/pkg/sentry/platform/systrap/sysmsg/sysmsg_lib.c b/pkg/sentry/platform/systrap/sysmsg/sysmsg_lib.c index 265e2a2d8..8be2c9b95 100644 --- a/pkg/sentry/platform/systrap/sysmsg/sysmsg_lib.c +++ b/pkg/sentry/platform/systrap/sysmsg/sysmsg_lib.c @@ -17,6 +17,7 @@ #include #include #include +#include #include #include #include @@ -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); diff --git a/pkg/sentry/platform/systrap/systrap.go b/pkg/sentry/platform/systrap/systrap.go index eea34b3cd..28d8070a8 100644 --- a/pkg/sentry/platform/systrap/systrap.go +++ b/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