一文教你搞懂 Go 中棧操做

轉載請聲明出處哦~,本篇文章發佈於luozhiyun的博客:https://www.luozhiyun.com/archives/513linux

本文使用的go的源碼15.7git

知識點

LInux 進程在內存佈局

Linux_stack

多任務操做系統中的每一個進程都在本身的內存沙盒中運行。在32位模式下,它老是4GB內存地址空間,內存分配是分配虛擬內存給進程,當進程真正訪問某一虛擬內存地址時,操做系統經過觸發缺頁中斷,在物理內存上分配一段相應的空間再與之創建映射關係,這樣進程訪問的虛擬內存地址,會被自動轉換變成有效物理內存地址,即可以進行數據的存儲與訪問了。github

Kernel space:操做系統內核地址空間;golang

Stack:棧空間,是用戶存放程序臨時建立的局部變量,棧的增加方向是從高位地址到地位地址向下進行增加。在現代主流機器架構上(例如x86)中,棧都是向下生長的。然而,也有一些處理器(例如B5000)棧是向上生長的,還有一些架構(例如System Z)容許自定義棧的生長方向,甚至還有一些處理器(例如SPARC)是循環棧的處理方式;編程

Heap:堆空間,堆是用於存放進程運行中被動態分配的內存段,它的大小並不固定,可動態擴張或縮減;數組

BBS segment:BSS段,存放的是全局或者靜態數據,可是存放的是全局/靜態未初始化數據;緩存

Data segment:數據段,一般是指用來存放程序中已初始化的全局變量的一塊內存區域;安全

Text segment:代碼段,指用來存放程序執行代碼的一塊內存區域。這部分區域的大小在程序運行前就已經肯定,而且內存區域屬於只讀。數據結構

棧的相關概念

In computer science, a call stack is a stack data structure that stores information about the active subroutines of a computer program.架構

In computer programming, a subroutine is a sequence of program instructions that performs a specific task, packaged as a unit.

調用棧call stack,簡稱棧,是一種棧數據結構,用於存儲有關計算機程序的活動 subroutines 信息。在計算機編程中,subroutines 是執行特定任務的一系列程序指令,打包爲一個單元。

A stack frame is a frame of data that gets pushed onto the stack. In the case of a call stack, a stack frame would represent a function call and its argument data.

棧幀stack frame又常被稱爲幀frame是在調用棧中儲存的函數之間的調用關係,每一幀對應了函數調用以及它的參數數據。

有了函數調用天然就要有調用者 caller 和被調用者 callee ,如在 函數 A 裏 調用 函數 B,A 是 caller,B 是 callee。

調用者與被調用者的棧幀結構以下圖所示:

Stack layout

Go 語言的彙編代碼中棧寄存器解釋的很是模糊,咱們大概只要知道兩個寄存器 BP 和 SP 的做用就能夠了:

BP:基準指針寄存器,維護當前棧幀的基準地址,以便用來索引變量和參數,就像一個錨點同樣,在其它架構中它等價於幀指針FP,只是在x86架構下,變量和參數均可以經過SP來索引;

SP:棧指針寄存器,老是指向棧頂;

Goroutine 棧操做

G Stack

在 Goroutine 中有一個 stack 數據結構,裏面有兩個屬性 lo 與 hi,描述了實際的棧內存地址:

  • stack.lo:棧空間的低地址;
  • stack.hi:棧空間的高地址;

在 Goroutine 中會經過 stackguard0 來判斷是否要進行棧增加:

  • stackguard0:stack.lo + StackGuard, 用於stack overlow的檢測;
  • StackGuard:保護區大小,常量Linux上爲 928 字節;
  • StackSmall:常量大小爲 128 字節,用於小函數調用的優化;
  • StackBig:常量大小爲 4096 字節;

根據被調用函數棧幀的大小來判斷是否須要擴容:

  1. 當棧幀大小(FramSzie)小於等於 StackSmall(128)時,若是 SP 小於 stackguard0 那麼就執行棧擴容;
  2. 當棧幀大小(FramSzie)大於 StackSmall(128)時,就會根據公式 SP - FramSzie + StackSmall 和 stackguard0 比較,若是小於 stackguard0 則執行擴容;
  3. 當棧幀大小(FramSzie)大於StackBig(4096)時,首先會檢查 stackguard0 是否已轉變成 StackPreempt 狀態了;而後根據公式 SP-stackguard0+StackGuard <= framesize + (StackGuard-StackSmall)判斷,若是是 true 則執行擴容;

須要注意的是,因爲棧是由高地址向低地址增加的,因此對比的時候,都是小於才執行擴容,這裏須要你們品品。

當執行棧擴容時,會在內存空間中分配更大的棧內存空間,而後將舊棧中的全部內容複製到新棧中,並修改指向舊棧對應變量的指針從新指向新棧,最後銷燬並回收舊棧的內存空間,從而實現棧的動態擴容。

彙編

這裏簡單講解一下後面分析中會用到的一些 Go 語言使用的 Plan 9 彙編,以避免看不太明白。

彙編函數

咱們先來看看 plan9 的彙編函數的定義:

function

stack frame size:包含局部變量以及額外調用函數的參數空間;

arguments size:包含參數以及返回值大小,例如入參是 3 個 int64 類型,返回值是 1 個 int64 類型,那麼返回值就是 sizeof(int64) * 4;

棧調整

棧的調整是經過對硬件 SP 寄存器進行運算來實現的,例如:

SUBQ    $24, SP  // 對 sp 作減法,爲函數分配函數棧幀 
...
ADDQ    $24, SP  // 對 sp 作加法 ,清除函數棧幀

因爲棧是往下增加的,因此 SUBQ 對 SP 作減法的時候其實是爲函數分配棧幀,ADDQ 則是清除棧幀。

常見指令

加減法操做

ADDQ  AX, BX   // BX += AX
SUBQ  AX, BX   // BX -= AX

數據搬運

常數在 plan9 彙編用 $num 表示,能夠爲負數,默認狀況下爲十進制。搬運的長度是由 MOV 的後綴決定。

MOVB $1, DI      // 1 byte
MOVW $0x10, BX   // 2 bytes
MOVD $1, DX      // 4 bytes
MOVQ $-10, AX     // 8 bytes

還有一點區別是在使用 MOVQ 的時候會有看到帶括號和不帶括號的區別。

// 加括號表明是指針的引用
MOVQ (AX), BX   // => BX = *AX 將AX指向的內存區域8byte賦值給BX
MOVQ 16(AX), BX // => BX = *(AX + 16)

//不加括號是值的引用
MOVQ AX, BX     // => BX = AX 將AX中存儲的內容賦值給BX,注意區別

跳轉

// 無條件跳轉
JMP addr   // 跳轉到地址,地址可爲代碼中的地址
JMP label  // 跳轉到標籤,能夠跳轉到同一函數內的標籤位置
JMP 2(PC)  // 以當前指令爲基礎,向前/後跳轉 x 行

// 有條件跳轉
JLS addr

地址運算:

LEAQ (AX)(AX*2), CX // => CX = AX + (AX * 2) = AX * 3

上面代碼中的 2 表明 scale,scale 只能是 0、二、四、8。

解析

G 的建立

由於棧都是在 Goroutine 上的,因此先從 G 的建立開始看如何建立以及初始化棧空間的。因爲我在《詳解Go語言調度循環源碼實現 https://www.luozhiyun.com/archives/448 》中已經講過 G 的建立,因此這裏只對棧的初始化部分的代碼進行講解。

G 的建立會調用 runtime·newproc進行建立:

runtime.newproc

func newproc(siz int32, fn *funcval) {
	argp := add(unsafe.Pointer(&fn), sys.PtrSize)
	gp := getg()
	// 獲取 caller 的 PC 寄存器
	pc := getcallerpc()
    // 切換到 G0 進行建立
	systemstack(func() {
		newg := newproc1(fn, argp, siz, gp, pc)
		...
	})
}

newproc 方法會切換到 G0 上調用 newproc1 函數進行 G 的建立。

runtime.newproc1

const _StackMin = 2048
func newproc1(fn *funcval, argp unsafe.Pointer, narg int32, callergp *g, callerpc uintptr) *g {
	_g_ := getg()
	...
	_p_ := _g_.m.p.ptr()
	// 從 P 的空閒鏈表中獲取一個新的 G
	newg := gfget(_p_)
	// 獲取不到則調用 malg 進行建立
	if newg == nil {
		newg = malg(_StackMin)
		casgstatus(newg, _Gidle, _Gdead)
		allgadd(newg) // publishes with a g->status of Gdead so GC scanner doesn't look at uninitialized stack.
	}
	...
	return newg
}

newproc1 方法很長,裏面主要是獲取 G ,而後對獲取到的 G 作一些初始化的工做。咱們這裏只看 malg 函數的調用。

在調用 malg 函數的時候會傳入一個最小棧大小的值:_StackMin(2048)。

runtime.malg

func malg(stacksize int32) *g {
	// 建立 G 結構體
	newg := new(g)
	if stacksize >= 0 {
		// 這裏會在 stacksize 的基礎上爲每一個棧預留系統調用所需的內存大小 _StackSystem
        // 在 Linux/Darwin 上( _StackSystem == 0 )本行不改變 stacksize 的大小
		stacksize = round2(_StackSystem + stacksize)
		// 切換到 G0 爲 newg 初始化棧內存
		systemstack(func() {
			newg.stack = stackalloc(uint32(stacksize))
		})
		// 設置 stackguard0 ,用來判斷是否要進行棧擴容
		newg.stackguard0 = newg.stack.lo + _StackGuard
		newg.stackguard1 = ^uintptr(0) 
		*(*uintptr)(unsafe.Pointer(newg.stack.lo)) = 0
	}
	return newg
}

在調用 malg 的時候會將傳入的內存大小加上一個 _StackSystem 值預留給系統調用使用,round2 函數會將傳入的值舍入爲 2 的指數。而後會切換到 G0 執行 stackalloc 函數進行棧內存分配。

分配完畢以後會設置 stackguard0 爲 stack.lo + _StackGuard,做爲判斷是否須要進行棧擴容使用,下面會談到。

棧的初始化

文件位置:src/runtime/stack.go

// 全局的棧緩存,分配 32KB如下內存
var stackpool [_NumStackOrders]struct {
	item stackpoolItem
	_    [cpu.CacheLinePadSize - unsafe.Sizeof(stackpoolItem{})%cpu.CacheLinePadSize]byte
}

//go:notinheap
type stackpoolItem struct {
	mu   mutex
	span mSpanList
} 

// 全局的棧緩存,分配 32KB 以上內存
var stackLarge struct {
	lock mutex
	free [heapAddrBits - pageShift]mSpanList // free lists by log_2(s.npages)
}

// 初始化stackpool/stackLarge全局變量
func stackinit() {
	if _StackCacheSize&_PageMask != 0 {
		throw("cache size must be a multiple of page size")
	}
	for i := range stackpool {
		stackpool[i].item.span.init()
		lockInit(&stackpool[i].item.mu, lockRankStackpool)
	}
	for i := range stackLarge.free {
		stackLarge.free[i].init()
		lockInit(&stackLarge.lock, lockRankStackLarge)
	}
}

在進行棧分配以前咱們先來看看棧初始化的時候會作什麼,須要注意的是,stackinit 是在調用 runtime·schedinit初始化的,是在調用 runtime·newproc以前進行的。

在執行棧初始化的時候會初始化兩個全局變量 stackpool 和 stackLarge。stackpool 能夠分配小於 32KB 的內存,stackLarge 用來分配大於 32KB 的棧空間。

棧的分配

從初始化的兩個兩個全局變量咱們也能夠知道,棧會根據大小的不一樣從不一樣的位置進行分配。

小棧內存分配

文件位置:src/runtime/stack.go

func stackalloc(n uint32) stack { 
    // 這裏的 G 是 G0
	thisg := getg()
	...
	var v unsafe.Pointer
	// 在 Linux 上,_FixedStack = 204八、_NumStackOrders = 四、_StackCacheSize = 32768
	// 若是申請的棧空間小於 32KB
	if n < _FixedStack<<_NumStackOrders && n < _StackCacheSize {
		order := uint8(0)
		n2 := n
		// 大於 2048 ,那麼 for 循環 將 n2 除 2,直到 n 小於等於 2048
		for n2 > _FixedStack {
			// order 表示除了多少次
			order++
			n2 >>= 1
		}
		var x gclinkptr
		//preemptoff != "", 在 GC 的時候會進行設置,表示若是在 GC 那麼從 stackpool 分配
		// thisg.m.p = 0 會在系統調用和 改變 P 的個數的時候調用,若是發生,那麼也從 stackpool 分配
		if stackNoCache != 0 || thisg.m.p == 0 || thisg.m.preemptoff != "" { 
			lock(&stackpool[order].item.mu)
			// 從 stackpool 分配
			x = stackpoolalloc(order)
			unlock(&stackpool[order].item.mu)
		} else {
			// 從 P 的 mcache 分配內存
			c := thisg.m.p.ptr().mcache
			x = c.stackcache[order].list
			if x.ptr() == nil {
				// 從堆上申請一片內存空間填充到stackcache中
				stackcacherefill(c, order)
				x = c.stackcache[order].list
			}
			// 移除鏈表的頭節點
			c.stackcache[order].list = x.ptr().next
			c.stackcache[order].size -= uintptr(n)
		}
		// 獲取到分配的span內存塊
		v = unsafe.Pointer(x)
	} else {
		...
	}
    ...
	return stack{uintptr(v), uintptr(v) + uintptr(n)}
}

stackalloc 會根據傳入的參數 n 的大小進行分配,在 Linux 上若是 n 小於 32768 bytes,也就是 32KB ,那麼會進入到小棧的分配邏輯中。

小棧指大小爲 2K/4K/8K/16K 的棧,在分配的時候,會根據大小計算不一樣的 order 值,若是棧大小是 2K,那麼 order 就是 0,4K 對應 order 就是 1,以此類推。這樣一方面能夠減小不一樣 Goroutine 獲取不一樣棧大小的鎖衝突,另外一方面能夠預先緩存對應大小的 span ,以便快速獲取。

thisg.m.p == 0可能發生在系統調用 exitsyscall 或改變 P 的個數 procresize 時,thisg.m.preemptoff != ""會發生在 GC 的時候。也就是說在發生在系統調用 exitsyscall 或改變 P 的個數在變更,亦或是在 GC 的時候,會從 stackpool 分配棧空間,不然從 mcache 中獲取。

若是 mcache 對應的 stackcache 獲取不到,那麼調用 stackcacherefill 從堆上申請一片內存空間填充到stackcache中。

主要注意的是,stackalloc 因爲切換到 G0 進行調用,因此 thisg 是 G0,咱們也能夠經過《如何編譯調試 Go runtime 源碼 https://www.luozhiyun.com/archives/506 》這一篇文章的方法來進行調試:

func stackalloc(n uint32) stack { 
	thisg := getg()
	// 添加一行打印
	if debug.schedtrace > 0 {
		print("stackalloc runtime: gp: gp=", thisg, ", goid=", thisg.goid, ", gp->atomicstatus=", readgstatus(thisg), "\n")
	}
	...
}

下面咱們分別看一下 stackpoolalloc 與 stackcacherefill 函數。

runtime.stackpoolalloc

func stackpoolalloc(order uint8) gclinkptr {
	list := &stackpool[order].item.span
	s := list.first
	lockWithRankMayAcquire(&mheap_.lock, lockRankMheap)
	if s == nil {
		// no free stacks. Allocate another span worth.
		// 從堆上分配 mspan
        // _StackCacheSize = 32 * 1024
		s = mheap_.allocManual(_StackCacheSize>>_PageShift, &memstats.stacks_inuse)
		if s == nil {
			throw("out of memory")
		}
		// 剛分配的 span 裏面分配對象個數確定爲 0
		if s.allocCount != 0 {
			throw("bad allocCount")
		}
		if s.manualFreeList.ptr() != nil {
			throw("bad manualFreeList")
		}
		//OpenBSD 6.4+ 系統須要作額外處理
		osStackAlloc(s)
		// Linux 中 _FixedStack = 2048
		s.elemsize = _FixedStack << order
		//_StackCacheSize =  32 * 1024
		// 這裏是將 32KB 大小的內存塊分紅了elemsize大小塊,用單向鏈表進行鏈接
		// 最後 s.manualFreeList 指向的是這塊內存的尾部
		for i := uintptr(0); i < _StackCacheSize; i += s.elemsize {
			x := gclinkptr(s.base() + i)
			x.ptr().next = s.manualFreeList
			s.manualFreeList = x
		}
		// 插入到 list 鏈表頭部
		list.insert(s)
	}
	x := s.manualFreeList
	// 表明被分配完畢
	if x.ptr() == nil {
		throw("span has no free stacks")
	}
	// 將 manualFreeList 日後移動一個單位
	s.manualFreeList = x.ptr().next
	// 統計被分配的內存塊
	s.allocCount++
	// 由於分配的時候第一個內存塊是 nil
	// 因此當指針爲nil 的時候表明被分配完畢
	// 那麼須要將該對象從 list 的頭節點移除
	if s.manualFreeList.ptr() == nil {
		// all stacks in s are allocated.
		list.remove(s)
	}
	return x
}

在 stackpoolalloc 函數中會去找 stackpool 對應 order 下標的 span 鏈表的頭節點,若是不爲空,那麼直接將頭節點的屬性 manualFreeList 指向的節點從鏈表中移除,並返回;

若是 list.first爲空,那麼調用 mheap_的 allocManual 函數從堆中分配 mspan,具體的內存分配相關的文章能夠看我這篇:《詳解Go中內存分配源碼實現 https://www.luozhiyun.com/archives/434 》。

從 allocManual 函數會分配 32KB 大小的內存塊,分配好新的 span 以後會根據 elemsize 大小將 32KB 內存進行切割,而後經過單向鏈表串起來並將最後一塊內存地址賦值給 manualFreeList 。

好比當前的 elemsize 所表明的內存大小是 8KB大小:

stackpool

runtime.stackcacherefill

func stackcacherefill(c *mcache, order uint8) { 
	var list gclinkptr
	var size uintptr
	lock(&stackpool[order].item.mu)
	//_StackCacheSize = 32 * 1024
	// 將 stackpool 分配的內存組成一個單向鏈表 list
	for size < _StackCacheSize/2 {
		x := stackpoolalloc(order)
		x.ptr().next = list
		list = x
		// _FixedStack = 2048
		size += _FixedStack << order
	}
	unlock(&stackpool[order].item.mu)
	c.stackcache[order].list = list
	c.stackcache[order].size = size
}

stackcacherefill 函數會調用 stackpoolalloc 從 stackpool 中獲取一半的空間組裝成 list 鏈表,而後放入到 stackcache 數組中。

大棧內存分配

func stackalloc(n uint32) stack { 
	thisg := getg() 
	var v unsafe.Pointer
	 
	if n < _FixedStack<<_NumStackOrders && n < _StackCacheSize {
		...
	} else {
		// 申請的內存空間過大,從 runtime.stackLarge 中檢查是否有剩餘的空間
		var s *mspan
		// 計算須要分配多少個 span 頁, 8KB 爲一頁
		npage := uintptr(n) >> _PageShift
		// 計算 npage 可以被2整除幾回,用來做爲不一樣大小內存的塊的索引
		log2npage := stacklog2(npage)
 
		lock(&stackLarge.lock)
		// 若是 stackLarge 對應的鏈表不爲空
		if !stackLarge.free[log2npage].isEmpty() {
			//獲取鏈表的頭節點,並將其從鏈表中移除
			s = stackLarge.free[log2npage].first
			stackLarge.free[log2npage].remove(s)
		}
		unlock(&stackLarge.lock)

		lockWithRankMayAcquire(&mheap_.lock, lockRankMheap)
		//這裏是stackLarge爲空的狀況
		if s == nil {
			// 從堆上申請新的內存 span
			s = mheap_.allocManual(npage, &memstats.stacks_inuse)
			if s == nil {
				throw("out of memory")
			}
			// OpenBSD 6.4+ 系統須要作額外處理
			osStackAlloc(s)
			s.elemsize = uintptr(n)
		}
		v = unsafe.Pointer(s.base())
	}
	...
	return stack{uintptr(v), uintptr(v) + uintptr(n)}
}

對於大棧內存分配,運行時會查看 stackLarge 中是否有剩餘的空間,若是不存在剩餘空間,它也會調用 mheap_.allocManual 從堆上申請新的內存。

棧的擴容

棧溢出檢測

編譯器會在目標代碼生成的時候執行:src/cmd/internal/obj/x86/obj6.go:stacksplit 根據函數棧幀大小插入相應的指令,檢查當前 goroutine 的棧空間是否足夠。

  1. 當棧幀大小(FramSzie)小於等於 StackSmall(128)時,若是 SP 小於 stackguard0 那麼就執行棧擴容;
  2. 當棧幀大小(FramSzie)大於 StackSmall(128)時,就會根據公式 SP - FramSzie + StackSmall 和 stackguard0 比較,若是小於 stackguard0 則執行擴容;
  3. 當棧幀大小(FramSzie)大於StackBig(4096)時,首先會檢查 stackguard0 是否已轉變成 StackPreempt 狀態了;而後根據公式 SP-stackguard0+StackGuard <= framesize + (StackGuard-StackSmall)判斷,若是是 true 則執行擴容;

咱們先來看看僞代碼會更清楚一些:

當棧幀大小(FramSzie)小於等於 StackSmall(128)時

CMPQ SP, stackguard
JEQ	label-of-call-to-morestack

當棧幀大小(FramSzie)大於 StackSmall(128)時:

LEAQ -xxx(SP), AX 
CMPQ AX, stackguard
JEQ	label-of-call-to-morestack

這裏 AX = SP - framesize + StackSmall,而後執行 CMPQ 指令讓 AX 與 stackguard 比較;

當棧幀大小(FramSzie)大於StackBig(4096)時

MOVQ	stackguard, SI // SI = stackguard
CMPQ	SI, $StackPreempt // compare SI ,StackPreempt
JEQ	label-of-call-to-morestack
LEAQ	StackGuard(SP), AX // AX = SP + StackGuard
SUBQ	SI, AX // AX = AX - SI =  SP + StackGuard -stackguard
CMPQ	AX, $(framesize+(StackGuard-StackSmall))

這裏的僞代碼會相對複雜一些,因爲 G 裏面的 stackguard0 在搶佔的時候可能會賦值成 StackPreempt,因此明確有沒有被搶佔,那麼須要將 stackguard0 和 StackPreempt進行比較。而後將執行比較: SP-stackguard+StackGuard <= framesize + (StackGuard-StackSmall),兩邊都加上 StackGuard 是爲了保證左邊的值是正數。

但願在理解完上面的代碼以前不要繼續往下看。

主要注意的是,在一些函數的執行代碼中,編譯器很智能的加上了NOSPLIT標記,打了這個標記以後就會禁用棧溢出檢測,能夠在以下代碼中發現這個標記的蹤跡:

代碼位置:cmd/internal/obj/x86/obj6.go

...
	if ctxt.Arch.Family == sys.AMD64 && autoffset < objabi.StackSmall && !p.From.Sym.NoSplit() {
		leaf := true
	LeafSearch:
		for q := p; q != nil; q = q.Link {
			...
		}

		if leaf {
			p.From.Sym.Set(obj.AttrNoSplit, true)
		}
	}
...

大體代碼邏輯應該是:當函數處於調用鏈的葉子節點,且棧幀小於StackSmall字節時,則自動標記爲NOSPLIT。一樣的,咱們在寫代碼的時候也能夠本身在函數上面加上//go:nosplit強制指定NOSPLIT屬性。

棧溢出實例

下面咱們寫一個簡單的例子:

func main() {
	a, b := 1, 2
	_ = add1(a, b)
	_ = add2(a, b)
	_ = add3(a, b)
}

func add1(x, y int) int {
	_ = make([]byte, 20)
	return x + y
}

func add2(x, y int) int {
	_ = make([]byte, 200)
	return x + y
}

func add3(x, y int) int {
	_ = make([]byte, 5000)
	return x + y
}

而後打印出它的彙編:

$ GOOS=linux GOARCH=amd64 go tool compile -S -N -l main.go

上面這個例子用三個方法調用解釋了上面所說的三種狀況:

main 函數

0x0000 00000 (main.go:3)        TEXT    "".main(SB), ABIInternal, $48-0
        0x0000 00000 (main.go:3)        MOVQ    (TLS), CX 
        0x0009 00009 (main.go:3)        CMPQ    SP, 16(CX) // SP < stackguard 則跳到 129執行
        0x0009 00009 (main.go:3)        CMPQ    SP, 16(CX)
        0x000d 00013 (main.go:3)        PCDATA  $0, $-2
        0x000d 00013 (main.go:3)        JLS     129
    	... 
        0x0081 00129 (main.go:3)        CALL    runtime.morestack_noctxt(SB)

首先,咱們從 TLS ( thread local storage) 變量中加載一個值至 CX 寄存器,而後將 SP 和 16(CX) 進行比較,那什麼是 TLS?16(CX) 又表明什麼?

其實TLS是一個僞寄存器,表示的是thread-local storage,它存放了 G 結構體。咱們看一看 runtime 源代碼中對於 G 的定義:

type g struct { 
	stack       stack   // offset known to runtime/cgo
	stackguard0 uintptr // offset known to liblink
	...
}
type stack struct {
	lo uintptr
	hi uintptr
}

能夠看到 stack 佔了 16bytes,因此 16(CX) 對應的是 g.stackguard0。因此 CMPQ SP, 16(CX)這一行代碼其實是比較 SP 和 stackguard 大小。若是 SP 小於 stackguard ,那麼說明到了增加的閾值,會執行 JLS 跳到 129 行,調用 runtime.morestack_noctxt 執行下一步棧擴容操做。

add1

0x0000 00000 (main.go:10)       TEXT    "".add1(SB), NOSPLIT|ABIInternal, $32-24

咱們看到 add1 的彙編函數,能夠看到它的棧大小隻有 32 ,沒有到達 StackSmall 128 bytes 的大小,而且它又是一個 callee 被調用者,因此能夠發它加上了NOSPLIT標記,也就印證了我上面結論。

add2

"".add2 STEXT size=148 args=0x18 locals=0xd0
        0x0000 00000 (main.go:15)       TEXT    "".add2(SB), ABIInternal, $208-24
        0x0000 00000 (main.go:15)       MOVQ    (TLS), CX
		// AX = SP  - 208 + 128 = SP -80
        0x0009 00009 (main.go:15)       LEAQ    -80(SP), AX // 棧大小大於StackSmall =128, 計算 SP - FramSzie + StackSmall 並放入AX寄存器							
        0x000e 00014 (main.go:15)       CMPQ    AX, 16(CX) // AX < stackguard 則跳到 138 執行
        0x0012 00018 (main.go:15)       PCDATA  $0, $-2
        0x0012 00018 (main.go:15)       JLS     138
		...
        0x008a 00138 (main.go:15)       CALL    runtime.morestack_noctxt(SB)

add2 函數的棧幀大小是 208,大於 StackSmall 128 bytes ,因此能夠看到首先從 TLS 變量中加載一個值至 CX 寄存器。

而後執行指令 LEAQ -80(SP), AX,可是這裏爲何是 -80 其實當時讓我蠻疑惑的,可是須要注意的是這裏的計算公式是: SP - FramSzie + StackSmall,直接代入以後會發現它就是 -80,而後將這個數值加載到 AX 寄存器中。

最後調用 CMPQ AX, 16(CX),16(CX) 咱們在上面已經講過了是等於 stackguard0 ,因此這裏是比較 AX 與 stackguard0 的小大,若是小於則直接跳轉到 138 行執行 runtime.morestack_noctxt

add3

"".add3 STEXT size=157 args=0x18 locals=0x1390
        0x0000 00000 (main.go:20)       TEXT    "".add3(SB), ABIInternal, $5008-24
        0x0000 00000 (main.go:20)       MOVQ    (TLS), CX 
        0x0009 00009 (main.go:20)       MOVQ    16(CX), SI // 將 stackguard 賦值給  SI
        0x000d 00013 (main.go:20)       PCDATA  $0, $-2 
        0x000d 00013 (main.go:20)       CMPQ    SI, $-1314 // 將 stackguard < stackPreempt 則跳轉到 147 執行
        0x0014 00020 (main.go:20)       JEQ     147 
        0x0016 00022 (main.go:20)       LEAQ    928(SP), AX // AX = SP +928
        0x001e 00030 (main.go:20)       SUBQ    SI, AX // AX -= stackguard
        0x0021 00033 (main.go:20)       CMPQ    AX, $5808 // framesize + 928 -128  = 5808,比較 AX < 5808,則執行147
        0x0027 00039 (main.go:20)       JLS     147
        ...
        0x0093 00147 (main.go:20)       CALL    runtime.morestack_noctxt(SB)

add3 函數是直接分配了一個 5000 bytes 的數組在棧上,因此開頭仍是同樣的,將從 TLS 變量中加載一個值至 CX 寄存器,而後將 stackguard0 賦值給 SI 寄存器;

接下來會執行指令 CMPQ SI, $-1314,這裏實際上比較 stackguard0 和 StackPreempt 的大小,至於爲啥是 -1314 實際上是直接在插入彙編代碼的時候會調用 StackPreempt 變量,這個變量是在代碼裏面寫死的:

代碼位置:cmd/internal/objabi/stack.go

const (
	StackPreempt = -1314 // 0xfff...fade
)

若是沒有被搶佔,那麼直接往下執行LEAQ 928(SP), AX,這句指令等於 AX = SP +_StackGuard,在 Linux 中 _StackGuard 等於 928;

接下來執行 SUBQ SI, AX,這一句指令等於 AX -= stackguard0

最後執行 CMPQ AX, $5808,這個 5808 其實是 framesize + _StackGuard - _StackSmall,若是 AX 小於 5808 那麼跳轉到 147 行執行 runtime.morestack_noctxt 函數。

到這裏棧溢出檢測就講解完畢了,我看了其餘的文章,應該都沒有我講解的全面,特別是棧幀大小大於 _StackBig 時的溢出檢測。

棧的擴張

runtime.morestack_noctxt 是用匯編實現的,它會調用到 runtime·morestack,下面咱們看看它的實現:

代碼位置:src/runtime/asm_amd64.s

TEXT runtime·morestack(SB),NOSPLIT,$0-0
	// Cannot grow scheduler stack (m->g0).
	// 沒法增加調度器的棧(m->g0)
	get_tls(CX)
	MOVQ	g(CX), BX
	MOVQ	g_m(BX), BX
	MOVQ	m_g0(BX), SI
	CMPQ	g(CX), SI
	JNE	3(PC)
	CALL	runtime·badmorestackg0(SB)
	CALL	runtime·abort(SB)
	// 省略signal stack、morebuf和sched的處理
	...
	// Call newstack on m->g0's stack.
	// 在 m->g0 棧上調用 newstack.
	MOVQ	m_g0(BX), BX
	MOVQ	BX, g(CX)
	MOVQ	(g_sched+gobuf_sp)(BX), SP
	CALL	runtime·newstack(SB)
	CALL	runtime·abort(SB)	// 若是 newstack 返回則崩潰 crash if newstack returns
	RET

runtime·morestack 作完校驗和賦值操做後會切換到 G0 調用 runtime·newstack來完成擴容的操做。

runtime·newstack

func newstack() {
	thisg := getg() 

	gp := thisg.m.curg
	 
	// 初始化寄存器相關變量
	morebuf := thisg.m.morebuf
	thisg.m.morebuf.pc = 0
	thisg.m.morebuf.lr = 0
	thisg.m.morebuf.sp = 0
	thisg.m.morebuf.g = 0
	...
	// 校驗是否被搶佔
	preempt := atomic.Loaduintptr(&gp.stackguard0) == stackPreempt
 
	// 若是被搶佔
	if preempt {
		// 校驗是否能夠安全的被搶佔
		// 若是 M 上有鎖
		// 若是正在進行內存分配
		// 若是明確禁止搶佔
		// 若是 P 的狀態不是 running
		// 那麼就不執行搶佔了
		if !canPreemptM(thisg.m) {
			// 到這裏表示不能被搶佔?
			// Let the goroutine keep running for now.
			// gp->preempt is set, so it will be preempted next time.
			gp.stackguard0 = gp.stack.lo + _StackGuard
			// 觸發調度器的調度
			gogo(&gp.sched) // never return
		}
	}

	if gp.stack.lo == 0 {
		throw("missing stack in newstack")
	}
	// 寄存器 sp
	sp := gp.sched.sp
	if sys.ArchFamily == sys.AMD64 || sys.ArchFamily == sys.I386 || sys.ArchFamily == sys.WASM {
		// The call to morestack cost a word.
		sp -= sys.PtrSize
	} 
	...
	if preempt {
		//須要收縮棧
		if gp.preemptShrink { 
			gp.preemptShrink = false
			shrinkstack(gp)
		}
		// 被 runtime.suspendG 函數掛起
		if gp.preemptStop {
			// 被動讓出當前處理器的控制權
			preemptPark(gp) // never returns
		}
 
		//主動讓出當前處理器的控制權
		gopreempt_m(gp) // never return
	}
 
	// 計算新的棧空間是原來的兩倍
	oldsize := gp.stack.hi - gp.stack.lo
	newsize := oldsize * 2 
	... 
	//將 Goroutine 切換至 _Gcopystack 狀態
	casgstatus(gp, _Grunning, _Gcopystack)
 
	//開始棧拷貝
	copystack(gp, newsize) 
	casgstatus(gp, _Gcopystack, _Grunning)
	gogo(&gp.sched)
}

newstack 函數的前半部分承擔了對 Goroutine 進行搶佔的任務,對於任務搶佔還不清楚的能夠看我這篇:《從源碼剖析Go語言基於信號搶佔式調度 https://www.luozhiyun.com/archives/485 》。

在開始執行棧拷貝以前會先計算新棧的大小是原來的兩倍,而後將 Goroutine 狀態切換至 _Gcopystack 狀態。

棧拷貝

func copystack(gp *g, newsize uintptr) { 
	old := gp.stack 
	// 當前已使用的棧空間大小
	used := old.hi - gp.sched.sp
 
	//分配新的棧空間
	new := stackalloc(uint32(newsize))
	...
 
	// 計算調整的幅度
	var adjinfo adjustinfo
	adjinfo.old = old
	// 新棧和舊棧的幅度來控制指針的移動
	adjinfo.delta = new.hi - old.hi
 
	// 調整 sudogs, 必要時與 channel 操做同步
	ncopy := used
	if !gp.activeStackChans {
		...
		adjustsudogs(gp, &adjinfo)
	} else {
		// 到這裏表明有被阻塞的 G 在當前 G 的channel 中,因此要防止併發操做,須要獲取 channel 的鎖
		 
		// 在全部 sudog 中找到地址最大的指針
		adjinfo.sghi = findsghi(gp, old) 
		// 對全部 sudog 關聯的 channel 上鎖,而後調整指針,而且複製 sudog 指向的部分舊棧的數據到新的棧上
		ncopy -= syncadjustsudogs(gp, used, &adjinfo)
	} 
	// 將源棧中的整片內存拷貝到新的棧中
	memmove(unsafe.Pointer(new.hi-ncopy), unsafe.Pointer(old.hi-ncopy), ncopy)
	// 繼續調整棧中 txt、defer、panic 位置的指針
	adjustctxt(gp, &adjinfo)
	adjustdefers(gp, &adjinfo)
	adjustpanics(gp, &adjinfo)
	if adjinfo.sghi != 0 {
		adjinfo.sghi += adjinfo.delta
	} 
	// 將 G 上的棧引用切換成新棧
	gp.stack = new
	gp.stackguard0 = new.lo + _StackGuard // NOTE: might clobber a preempt request
	gp.sched.sp = new.hi - used
	gp.stktopsp += adjinfo.delta
 
	// 在新棧重調整指針
	gentraceback(^uintptr(0), ^uintptr(0), 0, gp, 0, nil, 0x7fffffff, adjustframe, noescape(unsafe.Pointer(&adjinfo)), 0)
 
	if stackPoisonCopy != 0 {
		fillstack(old, 0xfc)
	}
	//釋放原始棧的內存空間
	stackfree(old)
}
  1. copystack 首先會計算一下使用棧空間大小,那麼在進行棧複製的時候只須要複製已使用的空間就行了;
  2. 而後調用 stackalloc 函數從堆上分配一片內存塊;
  3. 而後對比新舊棧的 hi 的值計算出兩塊內存之間的差值 delta,這個 delta 會在調用 adjustsudogs、adjustctxt 等函數的時候判斷舊棧的內存指針位置,而後加上 delta 而後就獲取到了新棧的指針位置,這樣就能夠將指針也調整到新棧了;

new_stack

  1. 調用 memmove 將源棧中的整片內存拷貝到新的棧中;
  2. 而後繼續調用調整指針的函數繼續調整棧中 txt、defer、panic 位置的指針;
  3. 接下來將 G 上的棧引用切換成新棧;
  4. 最後調用 stackfree 釋放原始棧的內存空間;

棧的收縮

棧的收縮發生在 GC 時對棧進行掃描的階段:

func scanstack(gp *g, gcw *gcWork) {
	... 
	// 進行棧收縮
	shrinkstack(gp)
	...
}

若是還不清楚 GC 的話不妨看一下我這篇文章:《Go語言GC實現原理及源碼分析 https://www.luozhiyun.com/archives/475 》。

runtime.shrinkstack

shrinkstack 這個函數我屏蔽了一些校驗函數,只留下面的核心邏輯:

func shrinkstack(gp *g) {
	...
	oldsize := gp.stack.hi - gp.stack.lo
	newsize := oldsize / 2 
	// 當收縮後的大小小於最小的棧的大小時,再也不進行收縮
	if newsize < _FixedStack {
		return
	}
	avail := gp.stack.hi - gp.stack.lo
	// 計算當前正在使用的棧數量,若是 gp 使用的當前棧少於四分之一,則對棧進行收縮
	// 當前使用的棧包括到 SP 的全部內容以及棧保護空間,以確保有 nosplit 功能的空間
	if used := gp.stack.hi - gp.sched.sp + _StackLimit; used >= avail/4 {
		return
	}
	// 將舊棧拷貝到新收縮後的棧上
	copystack(gp, newsize)
}

新棧的大小會縮小至原來的一半,若是小於 _FixedStack (2KB)那麼再也不進行收縮。除此以外還會計算一下當前棧的使用狀況是否不足 1/4 ,若是使用超過 1/4 那麼也不會進行收縮。

最後判斷肯定要進行收縮則調用 copystack 函數進行棧拷貝的邏輯。

總結

若是對於沒有了解過內存佈局的同窗,理解起來可能會比較吃力,由於咱們在看堆的時候內存增加都是從小往大增加,而棧的增加方向是相反的,致使在作棧指令操做的時候將 SP 減少反而是將棧幀增大。

除此以外就是 Go 使用的是 plan9 這種彙編,資料比較少,看起來很麻煩,想要更深刻了解這種彙編的能夠看我下面的 Reference 的資料。

Reference

聊一聊goroutine stack https://kirk91.github.io/posts/2d571d09/

Anatomy of a Program in Memory https://manybutfinite.com/post/anatomy-of-a-program-in-memory/

stack-frame-layout-on-x86-64 https://eli.thegreenplace.net/2011/09/06/stack-frame-layout-on-x86-64

深刻研究goroutine棧 http://www.huamo.online/2019/06/25/深刻研究goroutine棧/

x86-64 下函數調用及棧幀原理 https://zhuanlan.zhihu.com/p/27339191

Call stack https://en.wikipedia.org/wiki/Call_stack

plan9 assembly 徹底解析 https://github.com/cch123/golang-notes/blob/master/assembly.md

Go語言內幕(5):運行時啓動過程 https://studygolang.com/articles/7211

Go 彙編入門 https://github.com/go-internals-cn/go-internals/blob/master/chapter1_assembly_primer/README.md

Go Assembly by Example https://davidwong.fr/goasm/

https://golang.org/doc/asm

luozhiyun很酷

相關文章
相關標籤/搜索