Golang中的runtime.LockOSThread 和 runtime.UnlockOSThread

在runtime中有 runtime.LockOSThreadruntime.UnlockOSThread 两个函数,这两个函数有什么作用呢?我们看一下标准库中对它们的解释。

runtime.LockOSThread

// LockOSThread wires the calling goroutine to its current operating system thread.
// The calling goroutine will always execute in that thread,
// and no other goroutine will execute in it,
// until the calling goroutine has made as many calls to
// UnlockOSThread as to LockOSThread.
// If the calling goroutine exits without unlocking the thread,
// the thread will be terminated.
//
// All init functions are run on the startup thread. Calling LockOSThread
// from an init function will cause the main function to be invoked on
// that thread.
//
// A goroutine should call LockOSThread before calling OS services or
// non-Go library functions that depend on per-thread state.

调用 LockOSThread绑定 当前 goroutine 到当前 操作系统线程,此 goroutine 将始终在此线程执行,其它 goroutine 则无法在此线程中得到执行,直到当前调用线程执行了 UnlockOSThread 为止(也就是说 LockOSThread 可以指定一个goroutine 独占 一个系统线程);

Continue reading

认识无锁队列

无锁队列lock-free 中最基本的数据结构,一般应用在需要一款高性能队列的场景下。

对于多线程用户来说,无锁队列的入队和出队操作是线程安全的,不用再加锁控制

什么是无锁队列

队列每个开发者都知道,那么什么又是无锁队列呢?字面理解起来就是一个无锁状态的队列,多个线程(消费者)同时操作数据的时候不需要加锁,因为加/解锁都是一个很消耗资源的动作。

实现原理

我们先看一下无锁队列的底层实现数据结构。

数据结构

无锁队列底层的数据结构实现方式主要有两种:数组链接

Continue reading

Runtime: goroutine的暂停和恢复源码剖析

上一节《GC 对根对象扫描实现的源码分析》中,我们提到过在GC的时候,在对一些goroutine 栈进行扫描时,会在其扫描前触发 G 的暂停(suspendG)和恢复(resumeG)。

// markroot scans the i'th root.
//
// Preemption must be disabled (because this uses a gcWork).
//
// nowritebarrier is only advisory here.
//
//go:nowritebarrier
func markroot(gcw *gcWork, i uint32) {
	baseFlushCache := uint32(fixedRootCount)
	baseData := baseFlushCache + uint32(work.nFlushCacheRoots)
	baseBSS := baseData + uint32(work.nDataRoots)
	baseSpans := baseBSS + uint32(work.nBSSRoots)
	baseStacks := baseSpans + uint32(work.nSpanRoots)
	end := baseStacks + uint32(work.nStackRoots)

	// Note: if you add a case here, please also update heapdump.go:dumproots.
	switch {
		......

	default:
		var gp *g
		if baseStacks <= i && i < end {
			gp = allgs[i-baseStacks]
		} else {
			throw("markroot: bad index")
		}

		status := readgstatus(gp) // We are not in a scan state
		if (status == _Gwaiting || status == _Gsyscall) && gp.waitsince == 0 {
			gp.waitsince = work.tstart
		}

		// scanstack must be done on the system stack in case
		// we're trying to scan our own stack.
		systemstack(func() {
			userG := getg().m.curg
			selfScan := gp == userG && readgstatus(userG) == _Grunning
			if selfScan {
				casgstatus(userG, _Grunning, _Gwaiting)
				userG.waitreason = waitReasonGarbageCollectionScan
			}

			// TODO: suspendG blocks (and spins) until gp
			// stops, which may take a while for
			// running goroutines. Consider doing this in
			// two phases where the first is non-blocking:
			// we scan the stacks we can and ask running
			// goroutines to scan themselves; and the
			// second blocks.
			stopped := suspendG(gp)
			if stopped.dead {
				gp.gcscandone = true
				return
			}
			if gp.gcscandone {
				throw("g already scanned")
			}
			scanstack(gp, gcw)
			gp.gcscandone = true
			resumeG(stopped)

			if selfScan {
				casgstatus(userG, _Gwaiting, _Grunning)
			}
		})
	}
}

那么它在暂停和恢复一个goroutine时都做了些什么工作呢,今天我们通过源码来详细看一下。 go version 1.16.2

G的抢占

一个G可以在任何 安全点(safe-point) 被抢占,目前安全点可以分为以下几类:

  1. 阻塞安全点出现在 goroutine 被取消调度、同步阻塞或系统调用期间;
  2. 同步安全点出现在运行goroutine检查抢占请求时;
  3. 异步安全点出现在用户代码中的任何指令上,其中G可以安全的暂停且可以保证堆栈和寄存器扫描找到 stack root(这个很重要,GC扫描开始的地方)。runtime 可以通过一个信号在一个异步安全点暂停一个G。

这里将安全点分为 阻塞安全点同步安全点异步安全点,每种安全点都出现在不同的场景。

阻塞安全点和同步安全点,一个G的CPU状态是最小的(无法理解这里最小的意思)。垃圾回收器拥有整个stack的完整信息。这样就有可能使用最小的空间重新调度G,并精确的扫描G的 栈。

Continue reading

goroutine栈的申请与释放

对于提高对 stack 的使用效率,避免重复从heap中分配与释放,对其使用了 pool 的概念,runtime 里为共提供了两个pool, 分别为 stackpool ,另一个为 stackLarge

stack pool

stackpool: 16b~32k 对应通用的大小的stack。获取时通过调用 stackpoolalloc(), 释放时调用 stackpoolfree()

stackLarge:对应 > 32K 的 stack

在程序全局调度器初始化时会通过调用 stackinit() 实现对 stack 初始化。

当我们执行一个 go func() 语句的时候,runtime 会通过调用 newproc() 函数来创建G。而内部真正创建G的函数为 newproc1(),在没有G可以复用的情况下,会通过 newg = malg(_StackMin) 语句创建一个包含stack的G。

// Allocate a new g, with a stack big enough for stacksize bytes.
func malg(stacksize int32) *g {
	newg := new(g)
	if stacksize >= 0 {
		stacksize = round2(_StackSystem + stacksize)
		systemstack(func() {
			newg.stack = stackalloc(uint32(stacksize))
		})
		newg.stackguard0 = newg.stack.lo + _StackGuard
		newg.stackguard1 = ^uintptr(0)
		// Clear the bottom word of the stack. We record g
		// there on gsignal stack during VDSO on ARM and ARM64.
		*(*uintptr)(unsafe.Pointer(newg.stack.lo)) = 0
	}
	return newg
}

对于新创建的g,需要通过调用 stackalloc() 函数为其分配 stacksize 大小stack,那么分配操作它又是如何工作的呢?

stack的申请

根据申请stack的大小分两种情况,一种是 small stack,另一种是 large stack,两者采用不同的申请策略。主要涉及了内存申请策略,如果对golang 的内存管理比较了解的话,这块理解起来就显的太过于简单了。建议先阅读一下这篇文章《Golang 内存组件之mspan、mcache、mcentral 和 mheap 数据结构》。

func stackalloc(n uint32) stack {
	...

	var v unsafe.Pointer
	if n < _FixedStack<<_NumStackOrders && n < _StackCacheSize {
		order := uint8(0)
		n2 := n
		for n2 > _FixedStack {
			order++
			n2 >>= 1
		}
		var x gclinkptr
		if stackNoCache != 0 || thisg.m.p == 0 || thisg.m.preemptoff != "" {
			// thisg.m.p == 0 can happen in the guts of exitsyscall
			// or procresize. Just get a stack from the global pool.
			// Also don't touch stackcache during gc
			// as it's flushed concurrently.
			lock(&stackpool[order].item.mu)
			x = stackpoolalloc(order)
			unlock(&stackpool[order].item.mu)
		} else {
			c := thisg.m.p.ptr().mcache
			x = c.stackcache[order].list
			if x.ptr() == nil {
				stackcacherefill(c, order)
				x = c.stackcache[order].list
			}
			c.stackcache[order].list = x.ptr().next
			c.stackcache[order].size -= uintptr(n)
		}
		v = unsafe.Pointer(x)
	} else {
		...
	}
	return stack{uintptr(v), uintptr(v) + uintptr(n)}
}

对于 small stack 以可以分两种情况:

  • P:会直接通过从当前 G 绑定的P中的 mcache 字段申请,这个字段可以理解为内存资源中心,里面包含有多种不同规格大小的内存块,根据申请大小找到一个可以满足其大小的最小规格的内存区域。
  • P:调用 stackpoolalloc() 会直接从全局 stack pool 中获取
Continue reading

Golang的GPM 模型在网络编程中存在的问题

现状

目前在网络编程中,golang采用的是一种 goroutine-per-connection 的模式,即为每一个连接都分配一个goroutine,一个连接就是一个goroutine,多个连接之间没有关系。

package main

import (
	"fmt"
	"io/ioutil"
	"net"
	"time"
)

//模拟server端
func main() {
	tcpServer, _ := net.ResolveTCPAddr("tcp4", ":8080")
	listener, _ := net.ListenTCP("tcp", tcpServer)

	for {
		//当有新客户端请求时拿到与客户端的连接
		conn, err := listener.Accept()
		if err != nil {
			fmt.Println(err)
			continue
		}

		// 处理逻辑 goroutine-per-connection
		go handle(conn)
	}
}

func handle(conn net.Conn) {
	defer conn.Close()

	//读取客户端传送的消息
	go func() {
		response, _ := ioutil.ReadAll(conn)
		fmt.Println(string(response))
	}()

	//向客户端发送消息
	time.Sleep(1 * time.Second)
	now := time.Now().String()
	conn.Write([]byte(now))
}

这种模式看起来非常的舒服,简单易懂,大大减少了开发者的心智负担,深受开发者的喜爱。

存在的问题

对于 goroutine-per-connection 这种模式,真的就不有一点问题的吗?

假如现在只有两个 P,正常情况下每秒可以处理 10000 个 goroutine 连接。突然有1秒并发量极其的大,此时有 100万 个请求发送过来(极少出现这种极端情况,但并不表示不存在),根据 goroutine-per-connection 模式服务端会为这些连接创建 100万 个 goroutine,每个goroutine表示一个请求连接。

Continue reading