Java & Go 线程模式对比

// 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; // GC 中的 ThreadLocal 数据
 protected:
	void*       _real_malloc_address; // 真实分配地址
 protected:
  OSThread* _osthread; // 和 Thread 关联的 os Thread
}

// JavaThread
class JavaThread: public Thread {
  friend class VMStructs;
  friend class JVMCIVMStructs;
  friend class WhiteBox;
 private:
  JavaThread*    _next;            // Threads列表中的下一个 thread
  bool           _on_thread_list;  // 当前 JavaThread 是否在Thread列表中
  oop            _threadObj;       // Java级别的线程对象
}

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;

			// 创建 Hotstop 中的 JavaThread，同时创建对应的 _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());
  }

	// 开始执行，执行 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;
    
		// 线程池自身的状态 state
		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

		// 去掉前3位状态，获取工作线程数量
		private static int workerCountOf(int c)  { return c & COUNT_MASK; }
		// rs-state (RUNNING/SHUTDOWN/STOP/TIDYING/TERMINATED)
		// 如果wc=0，则此方法返回线程池state；如果w=workerCountOf(c)，则此方法返回 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 最大数量
	ncpu       int32         // cpu 数
    forcegc    forcegcstate  // 强制 GC
    sched      schedt        // 全局调度器
	allp []*p                // p
	allpLock mutex           // 读写 allp 时用的锁
	idlepMask pMask          // 空闲 p，每个 p 对应 pMask 中的
	timerpMask pMask         // 
)

type g struct {
    stack       stack   // goroutine 栈内存
	_panic    *_panic // goroutine 内部 panic
	_defer    *_defer // goroutine 内部 defer
	m         *m      // 当前 g 使用的 m
	sched     gobuf   // goroutine 调度数据
	param        unsafe.Pointer  // param 是用于传递的通用指针参数字段
	atomicstatus atomic.Uint32 // goroutine 状态
	goid         uint64        // goroutine ID
	waitsince    int64      // goroutine 阻塞时间
	preempt       bool // 抢占信号
	......
}

type m struct {
	g0      *g     // goroutine 
	morebuf gobuf  // gobuf arg to morestack
	tls           [tlsSlots]uintptr // thread-local 线程本地存储
	curg          *g // 正在执行的 g
	p             puintptr // 用于执行 go 代码的 p
	nextp         puintptr 
	oldp          puintptr // 在执行 syscall 之前的 p
	mstartfn      func()   // m 启动函数
	id            int64

	// 信号抢占
	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 所在 m 的 缓存
	pcache      pageCache
	goidcache    uint64 // goroutine ids 缓存
	// goroutines 队列
	runqhead uint32
	runqtail uint32
	runq     [256]guintptr
	gcw gcWork  // GC 执行时的缓冲区
	wbBuf wbBuf // GC 写入缓冲区，未来可能会放在 g 中
	sysmontick  sysmontick // 记录上一次 sysmon 抢占的信号
	......
}

type schedt struct {
	goidgen   atomic.Uint64 // 全局 goroutine id 生成
	lock mutex // 全局调度器的锁
	midle        muintptr // 等待工作的空闲 m
	maxmcount    int32    // 允许的 m 的最大数量

	pidle        puintptr // 空闲 p

	// 全局的 runnable 队列
	runq     gQueue
	runqsize int32

	// dead G 全局缓存，避免每次创建 goroutine 时重新分配内存
	gFree struct {
		lock    mutex
		stack   gList // Gs with stacks
		noStack gList // Gs without stacks
		n       int32
	}

	// sudog 缓存，sudog 代表等待列表中的goroutine
	sudoglock  mutex
	sudogcache *sudog

	freem *m // 等待释放的 m 列表

	idleTime atomic.Int64 // p 的空闲时间

	// sysmon 用于协作抢占
	sysmonwait atomic.Bool
	sysmonnote note
	sysmonlock mutex
	......
}

// src/runtime/proc.go

// 退出 syscall 后的调度
func exitsyscall() {
	...
	if sched.disable.user && !schedEnabled(gp) {
			// 发起调度
			Gosched()
	}
	...
}

// main goroutine 中监控触发 panic 的 defer 的 goroutine
func main() {
	...
	if runningPanicDefers.Load() != 0 {
		// Running deferred functions should not take long.
		for c := 0; c < 1000; c++ {
			if runningPanicDefers.Load() == 0 {
				break
			}
			// 发起调度
			Gosched()
		}
	}
	...
}

// 开始调度
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)
	// 删除 g 和 m 的关联关系
	dropg()
	lock(&sched.lock)
	// 将 g 放回全局队列 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 { // 如果超过 10ms，抢占 preempt
				// 抢占调度
				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 

// 触发信号事件，进行抢占信号的修改
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)
		}
	}

	// 修改 g 关联 m 的 抢占信号
	// 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

// 监控抢占信号状态，并执行抢占动作
func suspendG(gp *g) suspendGState {
    ......
	for i := 0; ; i++ {
		switch s := readgstatus(gp); s {
			case _Grunning:
					// Prepare for asynchronous preemption.
				asyncM2 := gp.m
				// 获取 m 中的 preemptGen 
				asyncGen2 := asyncM2.preemptGen.Load()
				needAsync := asyncM != asyncM2 || asyncGen != asyncGen2
				asyncM = asyncM2
				asyncGen = asyncGen2
				// cas 确认 g 状态
				casfrom_Gscanstatus(gp, _Gscanrunning, _Grunning)
				if preemptMSupported && debug.asyncpreemptoff == 0 && needAsync {
					now := nanotime()
					if now >= nextPreemptM {
						nextPreemptM = now + yieldDelay/2
						// 抢占 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()
		}
		// 尝试偷 待执行的 goroutine 或者 timer 从任一 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()
		}
	})
}