Java & Go thread model comparison

// Class hierarchy
// - Thread
//   - NamedThread
//     - VMThread
//     - ConcurrentGCThread
//     - WorkerThread
//       - GangWorker
//       - GCTaskThread
//   - JavaThread
//     - various subclasses eg CompilerThread, ServiceThread
//   - WatcherThread

// Thread
class Thread: public ThreadShadow {
  friend class VMStructs;
  friend class JVMCIVMStructs;
 private:
    GCThreadLocalData _gc_data; // ThreadLocal data in GC
 protected:
    void*       _real_malloc_address; // Real allocation address
 protected:
  OSThread* _osthread; // The os Thread associated with the Thread
}

// JavaThread
class JavaThread: public Thread {
  friend class VMStructs;
  friend class JVMCIVMStructs;
  friend class WhiteBox;
 private:
  JavaThread* _next; // The next thread in the thread list
  bool _on_thread_list; // Whether the current JavaThread is in the thread list
  oop _threadObj; // Java-level thread object
}

JavaThread::JavaThread(bool is_attaching_via_jni) :
                       Thread() {
  initialize();
  if (is_attaching_via_jni) {
    _jni_attach_state = _attaching_via_jni;
  } else {
    _jni_attach_state = _not_attaching_via_jni;
  }
  assert(deferred_card_mark().is_empty(), "Default MemRegion ctor");
}

public:
  // Constructor
  JavaThread();                            // delegating constructor
  JavaThread(bool is_attaching_via_jni);   // for main thread and JNI attached threads
  JavaThread(ThreadFunction entry_point, size_t stack_size = 0);
  ~JavaThread();

JavaThread::JavaThread(bool is_attaching_via_jni) :
                       Thread() {
  initialize();
  if (is_attaching_via_jni) {
    _jni_attach_state = _attaching_via_jni;
  } else {
    _jni_attach_state = _not_attaching_via_jni;
  }
  assert(deferred_card_mark().is_empty(), "Default MemRegion ctor");
}

JavaThread::JavaThread(ThreadFunction entry_point, size_t stack_sz) :
                       Thread() {
  initialize();
  _jni_attach_state = _not_attaching_via_jni;
  set_entry_point(entry_point);
  // Create the native thread itself.
  // %note runtime_23
  os::ThreadType thr_type = os::java_thread;
  thr_type = entry_point == &compiler_thread_entry ? os::compiler_thread :
                                                     os::java_thread;
  os::create_thread(this, thr_type, stack_sz);
}

public class Thread implements Runnable {

	public synchronized void start() {
		if (threadStatus != 0)
            throw new IllegalThreadStateException();

        group.add(this);

        boolean started = false;
        try {
		    // native 方法 创建 
            start0();
            started = true;
        } finally {
            try {
                if (!started) {
                    group.threadStartFailed(this);
                }
            } catch (Throwable ignore) {
            }
        }
	}
		
	private native void start0();
}

static JNINativeMethod methods[] = {
    {"start0",           "()V",        (void *)&JVM_StartThread},
    {"stop0",            "(" OBJ ")V", (void *)&JVM_StopThread},
    {"isAlive",          "()Z",        (void *)&JVM_IsThreadAlive},
    {"suspend0",         "()V",        (void *)&JVM_SuspendThread},
    {"resume0",          "()V",        (void *)&JVM_ResumeThread},
    {"setPriority0",     "(I)V",       (void *)&JVM_SetThreadPriority},
    {"yield",            "()V",        (void *)&JVM_Yield},
    {"sleep",            "(J)V",       (void *)&JVM_Sleep},
    {"currentThread",    "()" THD,     (void *)&JVM_CurrentThread},
    {"countStackFrames", "()I",        (void *)&JVM_CountStackFrames},
    {"interrupt0",       "()V",        (void *)&JVM_Interrupt},
    {"isInterrupted",    "(Z)Z",       (void *)&JVM_IsInterrupted},
    {"holdsLock",        "(" OBJ ")Z", (void *)&JVM_HoldsLock},
    {"getThreads",        "()[" THD,   (void *)&JVM_GetAllThreads},
    {"dumpThreads",      "([" THD ")[[" STE, (void *)&JVM_DumpThreads},
    {"setNativeName",    "(" STR ")V", (void *)&JVM_SetNativeThreadName},
};

JVM_ENTRY(void, JVM_StartThread(JNIEnv* env, jobject jthread))
  JVMWrapper("JVM_StartThread");
  JavaThread *native_thread = NULL;
  bool throw_illegal_thread_state = false;
  // We must release the Threads_lock before we can post a jvmti event
  // in Thread::start.
  {
    MutexLocker mu(Threads_lock);

    if (java_lang_Thread::thread(JNIHandles::resolve_non_null(jthread)) != NULL) {
      throw_illegal_thread_state = true;
    } else {
      jlong size =
             java_lang_Thread::stackSize(JNIHandles::resolve_non_null(jthread));
      NOT_LP64(if (size > SIZE_MAX) size = SIZE_MAX;)
      size_t sz = size > 0 ? (size_t) size : 0;

          // Create the JavaThread in Hotspot and also create the corresponding _osthread
      native_thread = new JavaThread(&thread_entry, sz);
      if (native_thread->osthread() != NULL) {
        // Note: the current thread is not being used within "prepare".
        native_thread->prepare(jthread);
      }
    }
  }

  if (throw_illegal_thread_state) {
    THROW(vmSymbols::java_lang_IllegalThreadStateException());
  }

  assert(native_thread != NULL, "Starting null thread?");

  if (native_thread->osthread() == NULL) {
    native_thread->smr_delete();
    if (JvmtiExport::should_post_resource_exhausted()) {
      JvmtiExport::post_resource_exhausted(
        JVMTI_RESOURCE_EXHAUSTED_OOM_ERROR | JVMTI_RESOURCE_EXHAUSTED_THREADS,
        os::native_thread_creation_failed_msg());
    }
    THROW_MSG(vmSymbols::java_lang_OutOfMemoryError(),
              os::native_thread_creation_failed_msg());
  }

    // Start execution. Execute os::start_thread
  Thread::start(native_thread);

JVM_END

public class Thread implements Runnable {
	/*
     * Java thread status for tools, default indicates thread 'not yet started'
     */
    private volatile int threadStatus;

		public enum State {
				// Thread state for a thread which has not yet started.
        NEW,

        // Thread state for a runnable thread.
        RUNNABLE,

        // Thread state for a thread blocked waiting for a monitor lock.
        BLOCKED,

        // Thread state for a waiting thread.
        WAITING,

        // Thread state for a waiting thread with a specified waiting time.
        TIMED_WAITING,

        // The thread has completed execution.
        TERMINATED;
	}
}

public class VM {
	/* The threadStatus field is set by the VM at state transition
     * in the hotspot implementation. Its value is set according to
     * the JVM TI specification GetThreadState function.
     */
    private static final int JVMTI_THREAD_STATE_ALIVE = 0x0001;
    private static final int JVMTI_THREAD_STATE_TERMINATED = 0x0002;
    private static final int JVMTI_THREAD_STATE_RUNNABLE = 0x0004;
    private static final int JVMTI_THREAD_STATE_BLOCKED_ON_MONITOR_ENTER = 0x0400;
    private static final int JVMTI_THREAD_STATE_WAITING_INDEFINITELY = 0x0010;
    private static final int JVMTI_THREAD_STATE_WAITING_WITH_TIMEOUT = 0x0020;

	public static Thread.State toThreadState(int threadStatus) {
        if ((threadStatus & JVMTI_THREAD_STATE_RUNNABLE) != 0) {
            return RUNNABLE;
        } else if ((threadStatus & JVMTI_THREAD_STATE_BLOCKED_ON_MONITOR_ENTER) != 0) {
            return BLOCKED;
        } else if ((threadStatus & JVMTI_THREAD_STATE_WAITING_INDEFINITELY) != 0) {
            return WAITING;
        } else if ((threadStatus & JVMTI_THREAD_STATE_WAITING_WITH_TIMEOUT) != 0) {
            return TIMED_WAITING;
        } else if ((threadStatus & JVMTI_THREAD_STATE_TERMINATED) != 0) {
            return TERMINATED;
        } else if ((threadStatus & JVMTI_THREAD_STATE_ALIVE) == 0) {
            return NEW;
        } else {
            return RUNNABLE;
        }
    }
}

public class ThreadPoolExecutor extends AbstractExecutorService {
       private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
       private final ReentrantLock mainLock = new ReentrantLock();
       private final HashSet<Worker> workers = new HashSet<>();

       private volatile int corePoolSize;
       private volatile int maximumPoolSize;
       private final BlockingQueue<Runnable> workQueue;
       private volatile long keepAliveTime;
       private volatile ThreadFactory threadFactory;
       private volatile RejectedExecutionHandler handler;

       private static final int COUNT_BITS = Integer.SIZE - 3;      // 29
       // 11111...111   2 ^ 29 - 1
       private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;
    
       // The state of the thread pool itself
       private static final int RUNNING    = -1 << COUNT_BITS;    // 101
       private static final int SHUTDOWN   =  0 << COUNT_BITS;    // 000
       private static final int STOP       =  1 << COUNT_BITS;    // 001
       private static final int TIDYING    =  2 << COUNT_BITS;    // 010
       private static final int TERMINATED =  3 << COUNT_BITS;    // 011

       // Remove the first 3 bits of the state to get the number of worker threads
       private static int workerCountOf(int c)  { return c & COUNT_MASK; }
       // rs-state (RUNNING/SHUTDOWN/STOP/TIDYING/TERMINATED)
       // If wc = 0, this method returns the thread pool state; if w = workerCountOf(c), this method returns the real value of ctl
       private static int ctlOf(int rs, int wc) { return rs | wc; }

       public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }
       
}

var(
    allm       *m            // m
    gomaxprocs int32         // p Maximum number
    ncpu       int32         // cpu Number
    forcegc    forcegcstate  // Force GC
    sched      schedt        // Global scheduler
    allp []*p                // p
    allpLock mutex           // Lock used when reading and writing allp
    idlepMask pMask          // idle p, each p corresponds to pMask
    timerpMask pMask         //
)

type g struct {
    stack       stack   // goroutine stack memory
    _panic    *_panic // goroutine internal panic
    _defer    *_defer // goroutine internal defer
    m         *m      // The m currently used by g
    sched     gobuf   // goroutine scheduling data
    param        unsafe.Pointer  // param is a generic pointer parameter field for passing
    atomicstatus atomic.Uint32 // goroutine status
    goid         uint64        // goroutine ID
    waitsince    int64      // goroutine blocking time
    preempt       bool // preemptive signal
    ...
}

type m struct {
    g0      *g     // goroutine
    morebuf gobuf  // gobuf arg to morestack
    tls           [tlsSlots]uintptr // thread-local storage
    curg          *g // The g being executed
    p             puintptr // p for executing go code
    nextp         puintptr
    oldp          puintptr // p before executing syscall
    mstartfn      func()   // m start function
    id            int64

    // signal preemption
    gsignal *g                // signal-handling g
    goSigStack gsignalStack      // Go-allocated signal handling stack
    sigmask sigset            // storage for saved signal mask
    signalPending atomic.Uint32 // Whether this is a pending preemption signal on this M.
    preemptGen atomic.Uint32   // counts the number of completed preemption signals
    ......
}

type p struct {
    id          int32
    status      uint32
    m           muintptr   // back-link to associated m (nil if idle)
    mcache      *mcache    // p's m cache
    pcache      pageCache
    goidcache    uint64 // goroutine ids cache
    // goroutines queue
    runqhead uint32
    runqtail uint32
    runq     [256]guintptr
    gcw gcWork  // GC execution buffer
    wbBuf wbBuf // GC write buffer, may be placed in g in the future
    sysmontick  sysmontick // Record the signal last sysmon preempted
    ...
}

type schedt struct {
    goidgen   atomic.Uint64 // Global goroutine id generation
    lock mutex // Global scheduler lock
    midle        muintptr // Idle m waiting for work
    maxmcount    int32    // Maximum number of m allowed

    pidle        puintptr // Idle p

    // global runnable queue
    runq     gQueue
    runqsize int32

    // dead G global cache to avoid reallocating memory every time a goroutine is created
    gFree struct {
       lock    mutex
       stack   gList // Gs with stacks
       noStack gList // Gs without stacks
       n       int32
    }

    // sudog cache, sudog stands for goroutine in the waiting list
    sudoglock  mutex
    sudogcache *sudog

    freem *m // list of m waiting to be released

    idleTime atomic.Int64 // idle time of p

    // sysmon is used for cooperative preemption
    sysmonwait atomic.Bool
    sysmonnote note
    sysmonlock mutex
    ...
}

// src/runtime/proc.go

// Scheduling after exiting the syscall
func exitsyscall() {
    ...
    if sched.disable.user && !schedEnabled(gp) {
          // Initiate scheduling
          Gosched()
    }
    ...
}

// The main goroutine monitors the goroutine that triggers the panic for defer
func main() {
    ...
    if runningPanicDefers.Load() != 0 {
       // Running deferred functions should not take long.
       for c := 0; c < 1000; c++ {
          if runningPanicDefers.Load() == 0 {
             break
          }
          // Initiate scheduling
          Gosched()
       }
    }
    ...
}

// Start scheduling
func Gosched() {
    checkTimeouts()
    mcall(gosched_m)
}

func gosched_m(gp *g) {
    if trace.enabled {
       traceGoSched()
    }
    goschedImpl(gp)
}

func goschedImpl(gp *g) {
    status := readgstatus(gp)
    if status&^_Gscan != _Grunning {
       dumpgstatus(gp)
       throw("bad g status")
    }
    casgstatus(gp, _Grunning, _Grunnable)
    // Remove the association between g and m
    dropg()
    lock(&sched.lock)
    // Put g back on the global queue runq
    globrunqput(gp)
    unlock(&sched.lock)

    schedule()
}

// src/runtime/proc.go

// The main goroutine.
func main() {
    ...
    if GOARCH != "wasm" { // no threads on wasm yet, so no sysmon
        systemstack(func() {
            newm(sysmon, nil, -1)
        })
    }
    ...
}

func sysmon() {
...
// retake P's blocked in syscalls
// and preempt long running G's
    if retake(now) != 0 {
        idle = 0
    } else {
        idle++
    }
    ...
}

const forcePreemptNS = 10 * 1000 * 1000 // 10ms

func retake(now int64) uint32 {
    ...
    for i := 0; i < len(allp); i++ {
        pp := allp[i]
        if pp == nil {
            continue
        }
        pd := &pp.sysmontick
        s := pp.status
        sysretake := false
        if s == _Prunning || s == _Psyscall {
            // Preempt G if it's running for too long.
            t := int64(pp.schedtick)
            if int64(pd.schedtick) != t {
                pd.schedtick = uint32(t)
                pd.schedwhen = now
			} else if pd.schedwhen+forcePreemptNS <= now { // If it's more than 10ms, preempt
                // preempt scheduling
                preemptone(pp)
                // In case of syscall, preemptone() doesn't
                // work, because there is no M wired to P.
                sysretake = true
            }
        }
    }
    ...
}

// src/runtime/signal_unix.go 

// Trigger signal events and modify preempt signals
func sighandler(sig uint32, info *siginfo, ctxt unsafe.Pointer, gp *g) {
    ...
    if sig == sigPreempt && debug.asyncpreemptoff == 0 && !delayedSignal {
       // Might be a preemption signal.
       doSigPreempt(gp, c)
    }
    ...
}

func doSigPreempt(gp *g, ctxt *sigctxt) {
    // Check if this G wants to be preempted and is safe to
    // preempt.
    if wantAsyncPreempt(gp) {
       if ok, newpc := isAsyncSafePoint(gp, ctxt.sigpc(), ctxt.sigsp(), ctxt.siglr()); ok {
          // Adjust the PC and inject a call to asyncPreempt.
          ctxt.pushCall(abi.FuncPCABI0(asyncPreempt), newpc)
       }
    }

    // Modify the preemption signal associated with m in g
    // Acknowledge the preemption.
    gp.m.preemptGen.Add(1)
    gp.m.signalPending.Store(0)

    if GOOS == "darwin" || GOOS == "ios" {
       pendingPreemptSignals.Add(-1)
    }
}

// src/runtime/preempt.go

// Monitor preempt signal status and perform preempt actions
func suspendG(gp *g) suspendGState {
    ......
    for i := 0; ; i++ {
       switch s := readgstatus(gp); s {
          case _Grunning:
                // Prepare for asynchronous preemption.
             asyncM2 := gp.m
             // Get preemptGen in m 
             asyncGen2 := asyncM2.preemptGen.Load()
             needAsync := asyncM != asyncM2 || asyncGen != asyncGen2
             asyncM = asyncM2
             asyncGen = asyncGen2
             // cas to check g status
             casfrom_Gscanstatus(gp, _Gscanrunning, _Grunning)
             if preemptMSupported && debug.asyncpreemptoff == 0 && needAsync {
                now := nanotime()
                if now >= nextPreemptM {
                   nextPreemptM = now + yieldDelay/2
                   // preempt M
                   preemptM(asyncM)
                }
             }  
       }
       ...
    }
    ...
}

func preemptM(mp *m) {
    if GOOS == "darwin" || GOOS == "ios" {
       execLock.rlock()
    }

    if mp.signalPending.CompareAndSwap(0, 1) {
       if GOOS == "darwin" || GOOS == "ios" {
          pendingPreemptSignals.Add(1)
       }
    
       signalM(mp, sigPreempt)
    }

    if GOOS == "darwin" || GOOS == "ios" {
       execLock.runlock()
    }
}

func schedule() {
    ...
    gp, inheritTime, tryWakeP := findRunnable() // blocks until work is available
    ...
}

func findRunnable() (gp *g, inheritTime, tryWakeP bool) {
    ...
    // Spinning Ms: steal work from other Ps.
    if mp.spinning || 2*sched.nmspinning.Load() < gomaxprocs-sched.npidle.Load() {
       if !mp.spinning {
          mp.becomeSpinning()
       }
       // Try to steal the goroutine or timer waiting to be executed from any of the p
       gp, inheritTime, tnow, w, newWork := stealWork(now)
       if gp != nil {
          // Successfully stole.
          return gp, inheritTime, false
       }
       ...
    }
}

// src/cmd/compile/internal/ssagen/ssa.go

func (s *state) stmt(n ir.Node) {
    ...
    s.stmtList(n.Init())
    switch n.Op() {
        case ir.OGO:
            n := n.(*ir.GoDeferStmt)
            s.callResult(n.Call.(*ir.CallExpr), callGo)
    }
}

func (s *state) call(n *ir.CallExpr, k callKind, returnResultAddr bool) *ssa.Value {
    ...
    switch {
        case k == callGo:
            aux := ssa.StaticAuxCall(ir.Syms.Newproc, s.f.ABIDefault.ABIAnalyzeTypes(nil, ACArgs, ACResults))
            call = s.newValue0A(ssa.OpStaticLECall, aux.LateExpansionResultType(), aux)
    }
}

// Create a new g running fn.
// Put it on the queue of g's waiting to run.
// The compiler turns a go statement into a call to this.
func newproc(fn *funcval) {
    gp := getg()
    pc := getcallerpc()
    systemstack(func() {
       newg := newproc1(fn, gp, pc)

       pp := getg().m.p.ptr()
       runqput(pp, newg, true)

       if mainStarted {
          wakep()
       }
    })
}