您好,登錄后才能下訂單哦!
小編給大家分享一下golang如何將多路復異步io轉成阻塞io,相信大部分人都還不怎么了解,因此分享這篇文章給大家參考一下,希望大家閱讀完這篇文章后大有收獲,下面讓我們一起去了解一下吧!
示例如下
package main import ( "net" ) func handleConnection(c net.Conn) { //讀寫數據 buffer := make([]byte, 1024) c.Read(buffer) c.Write([]byte("Hello from server")) } func main() { l, err := net.Listen("tcp", "host:port") if err != nil { return } defer l.Close() for { c, err := l.Accept() if err!= nil { return } go handleConnection(c) } }
對于我們都會寫上面的代碼,很簡單,的確golang的網絡部分對于我們隱藏了太多東西,我們不用像c++一樣去調用底層的socket函數,也不用去使用epoll等復雜的io多路復用相關的邏輯,但是上面的代碼真的就像我們看起來的那樣在調用accept和read時阻塞嗎?
// Multiple goroutines may invoke methods on a Conn simultaneously. //官方注釋:多個goroutines可能同時調用方法在一個連接上,我的理解就是所謂的驚群效應吧 //換句話說就是你多個goroutines監聽同一個連接同一個事件,所有的goroutines都會觸發, //這只是我的猜測,有待驗證。 type Conn interface { Read(b []byte) (n int, err error) Write(b []byte) (n int, err error) Close() error LocalAddr() Addr RemoteAddr() Addr SetDeadline(t time.Time) error SetReadDeadline(t time.Time) error SetWriteDeadline(t time.Time) error } type conn struct { fd *netFD }
這里面又一個Conn接口,下面conn實現了這個接口,里面只有一個成員netFD.
// Network file descriptor. type netFD struct { // locking/lifetime of sysfd + serialize access to Read and Write methods fdmu fdMutex // immutable until Close sysfd int family int sotype int isConnected bool net string laddr Addr raddr Addr // wait server pd pollDesc } func (fd *netFD) accept() (netfd *netFD, err error) { //................ for { s, rsa, err = accept(fd.sysfd) if err != nil { nerr, ok := err.(*os.SyscallError) if !ok { return nil, err } switch nerr.Err { /* 如果錯誤是EAGAIN說明Socket的緩沖區為空,未讀取到任何數據 則調用fd.pd.WaitRead,*/ case syscall.EAGAIN: if err = fd.pd.waitRead(); err == nil { continue } case syscall.ECONNABORTED: continue } return nil, err } break } //......... //代碼過長不再列出,感興趣看go的源碼,runtime 下的fd_unix.go return netfd, nil }
上面代碼段是accept部分,這里我們注意當accept有錯誤發生的時候,會檢查這個錯誤是否是syscall.EAGAIN
,如果是,則調用WaitRead將當前讀這個fd的goroutine在此等待,直到這個fd上的讀事件再次發生為止。當這個socket上有新數據到來的時候,WaitRead調用返回,繼續for循環的執行,這樣以來就讓調用netFD的Read的地方變成了同步“阻塞”。有興趣的可以看netFD的讀和寫方法,都有同樣的實現。
到這里所有的疑問都集中到了pollDesc上,它到底是什么呢?
const ( pdReady uintptr = 1 pdWait uintptr = 2 ) // Network poller descriptor. type pollDesc struct { link *pollDesc // in pollcache, protected by pollcache.lock lock mutex // protects the following fields fd uintptr closing bool seq uintptr // protects from stale timers and ready notifications rg uintptr // pdReady, pdWait, G waiting for read or nil rt timer // read deadline timer (set if rt.f != nil) rd int64 // read deadline wg uintptr // pdReady, pdWait, G waiting for write or nil wt timer // write deadline timer wd int64 // write deadline user uint32 // user settable cookie } type pollCache struct { lock mutex first *pollDesc }
pollDesc網絡輪詢器是Golang中針對每個socket文件描述符建立的輪詢機制。 此處的輪詢并不是一般意義上的輪詢,而是Golang的runtime在調度goroutine或者GC完成之后或者指定時間之內,調用epoll_wait獲取所有產生IO事件的socket文件描述符。當然在runtime輪詢之前,需要將socket文件描述符和當前goroutine的相關信息加入epoll維護的數據結構中,并掛起當前goroutine,當IO就緒后,通過epoll返回的文件描述符和其中附帶的goroutine的信息,重新恢復當前goroutine的執行。這里我們可以看到pollDesc中有兩個變量wg和rg,其實我們可以把它們看作信號量,這兩個變量有幾種不同的狀態:
pdReady:io就緒
pdWait:當前的goroutine正在準備掛起在信號量上,但是還沒有掛起。
G pointer:當我們把它改為指向當前goroutine的指針時,當前goroutine掛起
繼續接著上面的WaitRead調用說起,go在這里到底做了什么讓當前的goroutine掛起了呢。
func net_runtime_pollWait(pd *pollDesc, mode int) int { err := netpollcheckerr(pd, int32(mode)) if err != 0 { return err } // As for now only Solaris uses level-triggered IO. if GOOS == "solaris" { netpollarm(pd, mode) } for !netpollblock(pd, int32(mode), false) { err = netpollcheckerr(pd, int32(mode)) if err != 0 { return err } // Can happen if timeout has fired and unblocked us, // but before we had a chance to run, timeout has been reset. // Pretend it has not happened and retry. } return 0 } // returns true if IO is ready, or false if timedout or closed // waitio - wait only for completed IO, ignore errors func netpollblock(pd *pollDesc, mode int32, waitio bool) bool { //根據讀寫模式獲取相應的pollDesc中的讀寫信號量 gpp := &pd.rg if mode == 'w' { gpp = &pd.wg } for { old := *gpp //已經準備好直接返回true if old == pdReady { *gpp = 0 return true } if old != 0 { throw("netpollblock: double wait") } //設置gpp pdWait if atomic.Casuintptr(gpp, 0, pdWait) { break } } if waitio || netpollcheckerr(pd, mode) == 0 { gopark(netpollblockcommit, unsafe.Pointer(gpp), "IO wait", traceEvGoBlockNet, 5) } old := atomic.Xchguintptr(gpp, 0) if old > pdWait { throw("netpollblock: corrupted state") } return old == pdReady }
當調用WaitRead時經過一段匯編最重調用了上面的net_runtime_pollWait函數,該函數循環調用了netpollblock函數,返回true表示io已準備好,返回false表示錯誤或者超時,在netpollblock中調用了gopark函數,gopark函數調用了mcall的函數,該函數用匯編來實現,具體功能就是把當前的goroutine掛起,然后去執行其他可執行的goroutine。到這里整個goroutine掛起的過程已經結束,那當goroutine可讀的時候是如何通知該goroutine呢,這就是epoll的功勞了。
func netpoll(block bool) *g { if epfd == -1 { return nil } waitms := int32(-1) if !block { waitms = 0 } var events [128]epollevent retry: //每次最多監聽128個事件 n := epollwait(epfd, &events[0], int32(len(events)), waitms) if n < 0 { if n != -_EINTR { println("runtime: epollwait on fd", epfd, "failed with", -n) throw("epollwait failed") } goto retry } var gp guintptr for i := int32(0); i < n; i++ { ev := &events[i] if ev.events == 0 { continue } var mode int32 //讀事件 if ev.events&(_EPOLLIN|_EPOLLRDHUP|_EPOLLHUP|_EPOLLERR) != 0 { mode += 'r' } //寫事件 if ev.events&(_EPOLLOUT|_EPOLLHUP|_EPOLLERR) != 0 { mode += 'w' } if mode != 0 { //把epoll中的data轉換成pollDesc pd := *(**pollDesc)(unsafe.Pointer(&ev.data)) netpollready(&gp, pd, mode) } } if block && gp == 0 { goto retry } return gp.ptr() }
這里就是熟悉的代碼了,epoll的使用,看起來親民多了。pd:=*(**pollDesc)(unsafe.Pointer(&ev.data))
這是最關鍵的一句,我們在這里拿到當前可讀時間的pollDesc,上面我們已經說了,當pollDesc的讀寫信號量保存為G pointer時當前goroutine就會掛起。而在這里我們調用了netpollready函數,函數中把相應的讀寫信號量G指針擦出,置為pdReady,G-pointer狀態被抹去,當前goroutine的G指針就放到可運行隊列中,這樣goroutine就被喚醒了。
可以看到雖然我們在寫tcp server看似一個阻塞的網絡模型,在其底層實際上是基于異步多路復用的機制來實現的,只是把它封裝成了跟阻塞io相似的開發模式,這樣是使得我們不用去關注異步io,多路復用等這些復雜的概念以及混亂的回調函數。
以上是“golang如何將多路復異步io轉成阻塞io”這篇文章的所有內容,感謝各位的閱讀!相信大家都有了一定的了解,希望分享的內容對大家有所幫助,如果還想學習更多知識,歡迎關注億速云行業資訊頻道!
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。