go 網路 network poller

来源:https://www.cnblogs.com/studyios/archive/2023/12/04/17875360.html
-Advertisement-
Play Games

常用操作文件目錄的函數 1. CreateDirectory 創建文件夾 原型: BOOL CreateDirectory( LPCTSTR lpPathName, LPSECURITY_ATTRIBUTES lpSecurityAttributes ); 參數說明: lpPathName 要創建的 ...


網路基礎

協議架構

tcp鏈接

假如需要開發者去實現一套新的網路協議(例如 redis 的resp), 是基於TCP的, 那tcp這層的協議,是否需要開發者自己去實現?

這層如果自己實現, 其實很複雜, 會涉及很多演算法相關.

因此, 出現了 socket 對傳輸層進行了抽象, 開發者不需要關註傳輸層具體的實現, 使用socket提供的介面, socket內部會實現,比如三次握手, 四次揮手.

Socket

  很多系統都提供 Socket 作為 TCP(也有UDP) 網路連接的抽象,

  Linux-> Internet domain socket -> SOCK STREAM

  Linux 中 Socket 以 “文件描述符〞FD 作為標識

每建立一次連接接, sever都會創建一個新的 socket 專門和client通信, 原來監聽的socket還是一直處於監聽狀態.

socket 之前如何進行通信, 假設client 給 server發送消息, sever 沒有回覆, 那client是阻塞, 還是先執行別的工作, 這就是IO模型的範疇了.

IO模型

阻塞

非阻塞

多路復用

阻塞

當read中沒有數據過來, 線程(sever或者client都一樣)會阻塞等待.

同步讀寫Socket時，線程陷入內核態

當讀寫成功後，切換回用戶態，繼續執行

優點：開發難度小，代碼簡單

缺點：內核態切換開銷大

非阻塞

如果暫時無法收發數據，會返回錯誤

應用會不斷輪詢，直到Socket可以讀寫

優點：不會陷入內核態，自由度高

缺點：需要自旋輪詢

多路復用

大廠面試經常喜歡問的 epoll就是 多路復用的一種實現, 還有 select, poll 都是, 這三個主要的區別: 在返回數據時候的讀取方式.

業務把關註的事件註冊到 event poll池中, 當epoll接受到對應事件後,通知業務.

註冊多個Socket事件

調用epool，當有事件發生，返回

優點：提供了事件列表，不需要查詢輪詢各個Scoket

缺點：開發難度大，邏輯複雜

epoll 在 Mac: kqueue ; Windows: IOCP

小結

操作系統提供了Socket作為TCP和UDP通信的抽象

IO模型指的是操作Socket的方案

阻塞模型最利於業務編寫，但是性能差

多路復用性能好，但業務編寫麻煩

go socket的實現

實現方法

  1. 對系統的 epoll進行封裝, 因為go是可以在多個系統上運行的, 底層需要對平臺差異化封裝.
  1. 如果用戶去調用 封裝好的 epoll方法, 需要開發者執行去關註 epoll的狀態, 瞭解epoll的大致實現原理,肯定知道還需要去 解析epoll的返回, 然後 分析那些 socket 有事件來了. 這些工作都是通用性的,所以,go把這層一起封裝在了它的 網路層中.
  1. 然後,go為了更加簡化使用, 採用了 阻塞的思想, 一個協程對應一個 socket連接 , 當epoll 沒有對應事件返回時候, 就休眠等待, 有事件來了,就喚醒處理. 這樣就簡化了上層開發者的使用.

在底層使用操作系統的多路復用10

在協程層次使用阻塞模型

阻塞協程時，休眠協程

具體抽象 go代碼相關

多路復用器

各個系統的多路復用都有以下功能： 系統原生

1. 新建多路復用器 epoll create()
2. 往多路復用器里插入需要監聽的事件 epoll ctl()
3. 查詢發生了什麼事件 epoll wait()

Go Network Poller 多路復用器的抽象

epoll create() -> netpollinit()  // 新建

epoll cti0 -> netpollopen()   // 插入事件

epoll wait -> netpoll()        // 查詢

netpollinit 新建多路復用器

func netpollinit() {
	var errno uintptr
    // 新建了 epoll 系統底層實現 
	epfd, errno = syscall.EpollCreate1(syscall.EPOLL_CLOEXEC)
	 // 這裡的 epfd是個全局變數,初始化之後, 在操作時候就用這個 epfd

    // 新建管道 Linux的管道,用於多線程多進程間的通信, 用於結束這個epoll
	r, w, errpipe := nonblockingPipe()
	
	ev := syscall.EpollEvent{
		Events: syscall.EPOLLIN,
	}

	*(**uintptr)(unsafe.Pointer(&ev.Data)) = &netpollBreakRd
     // 管道中加入了事件 
	errno = syscall.EpollCtl(epfd, syscall.EPOLL_CTL_ADD, r, &ev)
	
	netpollBreakRd = uintptr(r)
	netpollBreakWr = uintptr(w)
}

歸納:

新建了一個底層 epoll對象  epfd

新建一個pipe管道用於中斷Epoll

將“管道有數據到達〞 事件註冊在Epoll中

netpollopen() 插入事件

func netpollopen(fd uintptr, pd *pollDesc) uintptr {
    // fd 是socket的對象  pd是 socket 和協程的對應關係

	var ev syscall.EpollEvent
	ev.Events = syscall.EPOLLIN | syscall.EPOLLOUT | syscall.EPOLLRDHUP | syscall.EPOLLET

	tp := taggedPointerPack(unsafe.Pointer(pd), pd.fdseq.Load())
	*(*taggedPointer)(unsafe.Pointer(&ev.Data)) = tp
    //  上面這兩行代碼 是在把 ev.Data 和  pd 聯繫起來,  後面epoll 返回時候, 就能直接和pd的協程聯繫起來

    // 將fd要關註的事件, 註冊到 epoll中, 調用的是 EpollCtl 和上面也對上了
	return syscall.EpollCtl(epfd, syscall.EPOLL_CTL_ADD, int32(fd), &ev)
}


type pollDesc struct {
	_     sys.NotInHeap
	link  *pollDesc      // in pollcache, protected by pollcache.lock
	fd    uintptr        //  socket ID
	fdseq atomic.Uintptr // protects against stale pollDesc

	atomicInfo atomic.Uint32 // atomic pollInfo
    // 想要讀的協程
	rg atomic.Uintptr // pdReady, pdWait, G waiting for read or pdNil
    // 想要寫的協程
	wg atomic.Uintptr // pdReady, pdWait, G waiting for write or pdNil

	lock    mutex // protects the following fields
	closing bool
	user    uint32    // user settable cookie
	rseq    uintptr   // protects from stale read timers
	rt      timer     // read deadline timer (set if rt.f != nil)
	rd      int64     // read deadline (a nanotime in the future, -1 when expired)
	wseq    uintptr   // protects from stale write timers
	wt      timer     // write deadline timer
	wd      int64     // write deadline (a nanotime in the future, -1 when expired)
	self    *pollDesc // storage for indirect interface. See (*pollDesc).makeArg.
}

歸納:

傳入一個Socket的FD，和pollDesc指針

pollDesc指針是Socket相關詳細信息

pollIDesc中記錄了哪個協程休眠在等待此Socket

將Socket可讀、可寫、斷開事件註冊到Epoll中

netpoll 查詢

這裡有必要講下 epoll的使用的一些特點, 能更好理解go中的實現.

epoll_wait 是阻塞函數, 但是可以傳一個 超時時間, 超過時間後就先返回, 不阻塞了。

int epoll_wait（int epfd，struct epoll_event * events， int maxevents，int timeout)

epfd epoll的id ,就是創建時候的 epfd.
epoll_event 是傳回有事件的fd和even的記錄
timeout 是 超時時間
返回值 是指示 epoll_event 中,前多個元素,是有值的, 避免遍歷.

// 超時時間 delay
 func netpoll(delay int64) gList {

	var events [128]syscall.EpollEvent
retry:
  // 去調了系統的 EpollWait 然後 看參數, 第二個參數裡面是包含了 有事件的even
	n, errno := syscall.EpollWait(epfd, events[:], int32(len(events)), waitms)
  // n 就表示 events 前多少個是有值的
	var toRun gList
	for i := int32(0); i < n; i++ {
	        var mode int32
		if ev.Events&(syscall.EPOLLIN|syscall.EPOLLRDHUP|syscall.EPOLLHUP|syscall.EPOLLERR) != 0 {
			mode += 'r'
		}
		if ev.Events&(syscall.EPOLLOUT|syscall.EPOLLHUP|syscall.EPOLLERR) != 0 {
			mode += 'w'
		}
		if mode != 0 {
			tp := *(*taggedPointer)(unsafe.Pointer(&ev.Data))
			pd := (*pollDesc)(tp.pointer())
			tag := tp.tag()
          // 通過ev.Data  和 pd關聯了起來  參入事件的時候, 有講過這個
			if pd.fdseq.Load() == tag {
				pd.setEventErr(ev.Events == syscall.EPOLLERR, tag)
				netpollready(&toRun, pd, mode) // 通過pd ,獲取到了對應的 協程隊列
			}
          }
	}
  return toRun // 返回的是 關註這個事件的協程隊列
}

歸納:

調用epoll wait()，查詢有哪些事件發生

根據Socket相關的pollDesc信息，返回哪些協程可以喚醒

Go Network Poller 如何工作

收發數據

場景 1：Socket 已經可讀寫

  1. netpoll 方法需要迴圈執行,這樣才能及時獲得那些事件有返回了, 才能通知對應的協程.
    誰來調這個方法, 入口方法放到了 gcStart , 因為 go會保證一段時間,一定會調下 gc, 在講 go 記憶體,垃圾回收時候講過.

再來看 netpoll() 中 netpollready() 的實現:

func netpollready(toRun *gList, pd *pollDesc, mode int32) {
	var rg, wg *g
	if mode == 'r' || mode == 'r'+'w' { // 可讀
		rg = netpollunblock(pd, 'r', true)
	}
	if mode == 'w' || mode == 'r'+'w' { // 可寫
		wg = netpollunblock(pd, 'w', true)
	}
     // 通過上面的方法,標記好了 pdReady之後,
	if rg != nil {
		toRun.push(rg) //把對應的協程,加入到一個可執行的 隊列中,一個鏈表
	}
	if wg != nil {
		toRun.push(wg)
	}
}

func netpollunblock(pd *pollDesc, mode int32, ioready bool) *g {
	gpp := &pd.rg // 預設是 可讀協程
	if mode == 'w' { //如果是 可寫 w, 就取可寫的協程
		gpp = &pd.wg
	}
    // 下麵的代碼就是把 對應的 rg 或者 wg 這個欄位改成了  pdReady, 標記下有事件來了.
	for {
        // 這裡的old值 , 第一次取可能是一個等待協程的地址 ,通過下麵的情景能看到
		old := gpp.Load()
		if old == pdReady {
			return nil
		}
		if old == pdNil && !ioready {
			return nil
		}
		var new uintptr
		if ioready {
			new = pdReady
		}
		if gpp.CompareAndSwap(old, new) { // 如果值相等,就寫入新的值
			if old == pdWait {
				old = pdNil
			}
			return (*g)(unsafe.Pointer(old)) // 當old為一個協程的地址,這裡返回的就是一個協程
		}
	}
}

上面方法就是把 pollDesc 的 rg或者wg置為  pdReady

當協程去調用 poll_runtime_pollWait() 方法 詢問時候, 發現已經是 ready狀態,就開始進行 讀寫

//go:linkname poll_runtime_pollWait internal/poll.runtime_pollWait
// 這種方式 可以調用聲明包中的小寫方法
func poll_runtime_pollWait(pd *pollDesc, mode int) int {
errcode := netpollcheckerr(pd, int32(mode))

for !netpollblock(pd, int32(mode), false) { // 這裡就是判斷這個 pd的rg或者wg 是否是 ready狀態
	errcode = netpollcheckerr(pd, int32(mode))
	if errcode != pollNoError {
		return errcode
	}
}
return pollNoError

}

歸納:

runtime 迴圈調用 netpoll() 方法 （g0協程 的gcStart 垃圾回收開始方法）

發現Socket可讀寫時，給對應的rg或者wg置為pdReady(1)

協程調用poll_runtime_pollWait()

判斷rg或者wg已經置為pdReady(1)，返向0

場景 2：Socket 暫時無法讀寫

需要深入看上面那個判斷 pd的rg是否為ready狀態的方法
當協程去調用 poll_runtime_pollWait()

會跳轉到:

func netpollblock(pd *pollDesc, mode int32, waitio bool) bool {
	gpp := &pd.rg
	if mode == 'w' {
		gpp = &pd.wg
	}

	for {
		
		if gpp.CompareAndSwap(pdReady, pdNil) {
			return true
		}
		if gpp.CompareAndSwap(pdNil, pdWait) {
			break
		}
		if v := gpp.Load(); v != pdReady && v != pdNil {
			throw("runtime: double wait")
		}
	}

	// 如果不是ready ,那麼協程就會休眠等待,  並把協程的地址,賦值給 pd的rg或wg,在`netpollblockcommit`中
	if waitio || netpollcheckerr(pd, mode) == pollNoError {
		gopark(netpollblockcommit, unsafe.Pointer(gpp), waitReasonIOWait, traceBlockNet, 5)
	}
	// be careful to not lose concurrent pdReady notification
	old := gpp.Swap(pdNil)
	if old > pdWait {
		throw("runtime: corrupted polldesc")
	}
	return old == pdReady
}

歸納:

runtime 迴圈調用netpoll()方法 （g0協程）
協程調用poll_runtime_pollWait()
發現對應的rg或者wg為0
給對應的rg或者wg置為協程地址
休眠等待

當有事件來的時候, netpoll 中返回了 對應事件的 glist, 然後,runtime就會通知這些協程開始工作.

runtime 迴圈調用 netpoll() 方法 (g0協程）
發現Socket可讀寫時，給對應的查看對應的rg或者wg
若為協程地址，返回協程地址
調度器開始調度對應協程

大體的結構如下:


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 前端組件wolfTable的style格式文檔
    最近有個需求需要實現自定義首頁佈局，需要將屏幕按照 6 列 4 行進行等分成多個格子，然後將組件可拖拽對應格子進行渲染展示。 示例 對比一些已有的插件，發現想要實現產品的交互效果，沒有現成可用的。本身功能並不是太過複雜，於是決定自己基於 vue 手擼一個簡易的 Grid 拖拽佈局。 完整源碼在此，在 ...
  • Arcpy使用入門
    2.7Python(目前ArcGIS使用)代碼轉化為3.5Python(目前ArcGIS Pro使用)代碼 Analyze Tools For Pro (2to3命令) 基本操作 調用ArcToolbox的兩種形式 #arcpy.ToolboxAlias.ToolName() #arcpy.Tool ...
  • DDD學習與感悟——總是覺得自己在CRUD怎麼辦？
    我們有時候也會看到一些博客看到或者聽到一些同事在說：這個業務有什麼難的，不就是CRUD麽？在軟體生命周期初期，我們通過CRUD這種方式我們可以快速的實現業務規則，交付項目，但隨著業務逐漸複雜，通過CRUD這種粗暴方式不可避免地會淹沒業務核心規則，產生很多祖傳（屎山）代碼，系統交接的時候我們經常會聽到... ...
  • 2分鐘，快速認識什麼是SQL
    結構化查詢語言，簡稱SQL，它是與關係資料庫管理系統通信的黃金標準語言。今天就來一起快速認識一下什麼是SQL，您可以通過以下的文字內容學習，也可以通過文末的視頻學習，希望本文對您有所幫助。 您可能聽說過 MySQL、Postgres、Microsoft SQL Server 和 Oracle 等數據 ...
  • 從0開始用Maven
    一.Maven的介紹即相關概念 Maven是一款構建和管理Java項目的工具,它將項目開發和管理過程抽象成一個項目對象模型（POM),提供了一種統一的項目結構。 Maven官網 1.為什麼使用Maven/Maven的作用 (1)多模塊支持：當項目非常龐大的時候，就不適合使用package來劃分模塊， ...
  • LeetCode-Java：55.跳躍游戲
    題目 給你一個非負整數數組 nums ，你最初位於數組的 第一個下標 。數組中的每個元素代表你在該位置可以跳躍的最大長度。 判斷你是否能夠到達最後一個下標，如果可以，返回 true ；否則，返回 false 。 示例 1： 輸入：nums = [2,3,1,1,4] 輸出：true 解釋：可以先跳 ...
  • Django學習(二) 之 模板的使用
    寫在前面 昨晚應該是睡的最好一天吧，最近一個月睡眠好差，睡不著不說，而且半夜總醒，搞的我第二天就會超沒精神。 昨天下午去姐姐家，我剛進屋，小外甥直接就問我說： 老舅，你都很長時間沒來啦，**(前女友)哪去了， 我們都好久沒出溜溜了！ 我頓了下說，她不喜歡我們了，等以後天暖和，我們再去溜溜。 才發現， ...
  • Python 解析JSON實現主機管理
    JSON（JavaScript Object Notation）是一種輕量級的數據交換格式，它以易於閱讀和編寫的文本形式表示數據。JSON 是一種獨立於編程語言的數據格式，因此在不同的編程語言中都有對應的解析器和生成器。JSON 格式的設計目標是易於理解、支持複雜數據結構和具有良好的可擴展性。 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類