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