Goroutines To OS Threads: The 73% Latency Drop We Measured By Promoting Work

// This looks so clean and lightweight, right?  
go func() { // Spawn a new goroutine - appears cheap  
    processMarketData(data)  // We need this done in under 10ms - critical path  
}() // But what's actually happening under the hood?  

// Here's the reality:  
// 1. Goroutine gets created (this part IS cheap)  
// 2. Gets placed on the global run queue (lock contention starts)  
// 3. Now it waits for the scheduler to notice it  
// 4. Eventually gets scheduled to an M (context switch overhead)  
// 5. Runs for its 10ms quantum if lucky  
// 6. Gets preempted if it exceeds the quantum  
// 7. Back to the run queue it goes  
// 8. The whole dance repeats

import ( // Import required packages  
    "runtime"  // Need this for accessing Go runtime stats  
    "runtime/trace"  // This one's for detailed scheduler tracing  
)  

type SchedulerStats struct { // Structure to hold scheduler metrics  
    StartTime     time.Time  // When we started measuring  
    Schedules     uint64     // How many scheduler decisions made  
    PreemptCount  uint64     // Times goroutines got forcibly preempted  
    RunqueueLen   int        // Length of run queue - high means contention  
    ThreadCount   int        // How many goroutines exist  
}  
func measureSchedulerOverhead() SchedulerStats { // Function to capture current scheduler state  
    var stats runtime.MemStats  // Allocate memory stats struct on stack  
    runtime.ReadMemStats(&stats)  // Populate it with current statistics  

    return SchedulerStats { // Return snapshot of current state  
        StartTime:    time.Now(),  // Capture current timestamp for baseline  
        Schedules:    stats.NumGC,  // Using NumGC as a proxy metric  
        RunqueueLen:  runtime.GOMAXPROCS(0),  // Query number of Ps without changing  
        ThreadCount:  runtime.NumGoroutine(),  // Total goroutines in system  
    }  
}  
func traceSchedulerActivity() { // Detailed tracing function  
    f, _ := os.Create("trace.out")  // Create output file for trace data  
    defer f.Close()  // Make sure we close when done  

    trace.Start(f)  // Begin capturing trace events  
    defer trace.Stop()  // Stop tracing when function exits  

    processMarketData()  // Run the workload we're measuring  
}

type PriorityWorker struct { // Structure for dedicated worker  
    name     string  // Human-readable name for debugging  
    work     chan WorkItem  // Buffered channel for incoming work  
    affinity int  // Which CPU core to stick to  
}  

func (pw *PriorityWorker) Run() { // Main worker loop  
    runtime.LockOSThread()  // Lock this goroutine to its own OS thread  
    defer runtime.UnlockOSThread()  // Clean up to avoid thread leaks  

    setCPUAffinity(pw.affinity)  // Pin to specific CPU core  
    setThreadPriority(ThreadPriorityHigh)  // Bump up OS-level priority  

    for work := range pw.work {  // Block waiting for work  
        start := time.Now()  // Capture start time for latency measurement  
        result := work.Process()  // Execute without scheduler interference  

        latency := time.Since(start)  // Calculate how long it took  
        if latency > 10*time.Millisecond {  // Check against SLA  
            log.Printf("High latency: %v", latency)  // Log for investigation  
        }  

        work.Respond(result)  // Send result back to caller  
    }  
}  
func setCPUAffinity(cpu int) { // Linux-specific CPU pinning  
    var mask unix.CPUSet  // Create a CPU set bitmask  
    mask.Set(cpu)  // Set the bit for our target CPU  
    unix.SchedSetaffinity(0, &mask)  // Apply to current thread  
}

type OSThreadPool struct { // Pool structure  
    workers   []*OSWorker  // Slice of worker pointers  
    workQueue chan WorkItem  // Main queue for all work  
    size      int  // Fixed number of workers  
}  

func NewOSThreadPool(size int) *OSThreadPool { // Constructor  
    pool := &OSThreadPool{  // Create pool structure  
        workers:   make([]*OSWorker, size),  // Pre-allocate workers  
        workQueue: make(chan WorkItem, 10000),  // Large buffer  
        size:      size,  // Store size  
    }  

    for i := 0; i < size; i++ {  // Create all workers upfront  
        worker := &OSWorker{  // Allocate new worker  
            id:       i,  // Assign sequential ID  
            pool:     pool,  // Link back to parent  
            workChan: make(chan WorkItem, 100),  // Each worker gets own queue  
            affinity: i % runtime.NumCPU(),  // Spread across CPUs  
        }  

        pool.workers[i] = worker  // Store in pool  
        go worker.run()  // Start worker on its OS thread  
    }  

    go pool.distribute()  // Start work distributor  
    return pool  // Return initialized pool  
}  
func (w *OSWorker) run() { // Worker execution loop  
    runtime.LockOSThread()  // Lock immediately when starting  
    defer runtime.UnlockOSThread()  // Cleanup on exit  

    setCPUAffinity(w.affinity)  // Pin to CPU core  

    for work := range w.workChan {  // Process work forever  
        start := time.Now()  // Start latency tracking  
        result := work.Execute()  // Execute on dedicated thread  

        latency := time.Since(start)  // Calculate execution time  
        metrics.RecordLatency("os_thread_worker", latency)  // Send to metrics  
        work.Complete(result)  // Return result  
    }  
}

type HybridScheduler struct { // Intelligent routing system  
    normalPool  *sync.Pool  // For batching low-priority work  
    priorityPool *OSThreadPool  // Dedicated OS threads  
    classifier  WorkClassifier  // Decides routing for each item  
}  

func (hs *HybridScheduler) Schedule(work WorkItem) { // Main routing function  
    priority := hs.classifier.Classify(work)  // Evaluate work priority  

    switch priority {  // Route based on classification  
    case PriorityCritical:  // Ultra-low latency needed  
        hs.priorityPool.Submit(work)  // Send to OS thread pool  

    case PriorityHigh:  // Time-sensitive but not critical  
        go func() {  // Create goroutine but lock it  
            runtime.LockOSThread()  // Lock for this work  
            work.Execute()  // Do the work  
            runtime.UnlockOSThread()  // Release lock  
        }()  

    case PriorityNormal:  // Regular latency is fine  
        go work.Execute()  // Normal goroutine works  

    case PriorityLow:  // Can be batched  
        hs.normalPool.Put(work)  // Add to batch pool  
    }  
}

// Bad approach - creates massive contention  
runtime.GOMAXPROCS(runtime.NumCPU() * 4)  // 4x OS threads causes thrashing  
// Best for our use case - fewer Ps plus dedicated threads  
runtime.GOMAXPROCS(runtime.NumCPU() / 2)  // Leave room for OS threads  
dedicatedThreads := runtime.NumCPU() / 2  // Use other half for dedicated work