[Golang] 劍走偏鋒 -- IoComplete ports

来源:https://www.cnblogs.com/cnblogs-wangzhipeng/archive/2020/01/17/12206337.html
-Advertisement-
Play Games

前言 Golang 目前的主要應用領域還是後臺微服務，雖然在業務領域也有所應用但仍然是比較小衆的選擇。大多數的服務運行環境都是linux，而在windows中golang應用更少，而作者因爲特殊情況，不得已要在widows環境中用golang去寫本地代理服務。在我的使用場景中實時性要求非常高（視頻通 ...


前言

Golang 目前的主要應用領域還是後臺微服務，雖然在業務領域也有所應用但仍然是比較小衆的選擇。大多數的服務運行環境都是linux，而在windows中golang應用更少，而作者因爲特殊情況，不得已要在widows環境中用golang去寫本地代理服務。在我的使用場景中實時性要求非常高（視頻通信），對tcp數據處理要足夠快，否則會造成TCP 服務端的 Receive Buffer 溢出造成 Packet loss，影響實時性和數據的完整性。

作者閲讀了golang 在windows 環境下 tcp 部分syscall 的實現，最終確認它的底層模型是用了完成埠（異步IO模型）的。
但是由於作者本人比較喜歡折騰，所以用golang 底層的syscall 實現了一下tcp 完成埠服務。

IoCompletion Port

以下為windows環境下用golang實現的 IoCompletion Port Server

IoCompletionRootContext

管理指定 Port 上所有 accepted socket:


type IoCompletionRootContext struct {
    socket     windows.Handle
    socketAddr windows.SockaddrInet4
    ioSet      []*IoCompletionContext

    sync.Mutex
}


func (root *IoCompletionRootContext) NewIoContext() *IoCompletionContext {
    root.Lock()
    defer root.Unlock()
    res := &IoCompletionContext{
        data: make([]byte, 65535),
        overlapped: windows.Overlapped{
            Internal:     0,
            InternalHigh: 0,
            Offset:       0,
            OffsetHigh:   0,
            HEvent:       0,
        },
    }

    res.wsaBuf.Buf = &res.data[0]
    res.wsaBuf.Len = uint32(65535)

    root.ioSet = append(root.ioSet, res)
    return res
}

func NewRootContext() *IoCompletionRootContext {
    return &IoCompletionRootContext{
        ioSet: make([]*IoCompletionContext, 0),
    }
}

IoCompletionContext

accepted socket 的上下文：

    type IoCompletionContext struct {
        socket     windows.Handle
        socketAddr windows.SockaddrInet4
        wsaBuf     windows.WSABuf
        data       []byte
        opType     OP_TYPE
        overlapped windows.Overlapped
    }

IoCompletionServer

完成埠服務:

type IoCompletionServer struct {
    Addr     string
    Port     int
    recvFunc func(data []byte) error
    rootCtx  *IoCompletionRootContext
    // 爲了防止記憶體移動，採用此種方式
    accepts           sync.Map
    hIOCompletionPort windows.Handle
}


func (ss *IoCompletionServer) saveIoRootCtx(id uint32, ctx *IoCompletionRootContext) {
    ss.accepts.Store(id, ctx)
}

func (ss *IoCompletionServer) loadIoRootCtx(id uint32) *IoCompletionRootContext {
    if id == uint32(ss.rootCtx.socket) {
        return ss.rootCtx
    }

    if v, isOk := ss.accepts.Load(id); isOk {
        if res, isOk := v.(*IoCompletionRootContext); isOk {
            return res
        }
    }

    return nil
}

func (ss *IoCompletionServer) remove(id uint32) {
    ss.accepts.Delete(id)
}

func (ss *IoCompletionServer) RegisterReceiveFunc(rfunc func([]byte) error) {
    ss.recvFunc = rfunc
}

func (ss *IoCompletionServer) Listen() {
    dwBytesTransfered := uint32(0)
    var ctxId uint32
    var overlapped *windows.Overlapped
    for {
        err := windows.GetQueuedCompletionStatus(ss.hIOCompletionPort, &dwBytesTransfered,
            &ctxId, &overlapped, windows.INFINITE)
        if err != nil {
            fmt.Printf("syscall.GetQueuedCompletionStatus: %v\n", err)
        }

        if overlapped == nil {
            continue
        }

        // 通過位移取得ioCtx
        ioCtx := (*IoCompletionContext)(unsafe.Pointer(uintptr(unsafe.Pointer(overlapped)) - unsafe.Offsetof(IoCompletionContext{}.overlapped)))
        switch ioCtx.opType {
        case ACCEPT_POSTED:
            {
                ss.DoAcceptEx(ss.loadIoRootCtx(ctxId), ioCtx)
            }
        case RECV_POSTED:
            {
                ss.DoReceive(ss.loadIoRootCtx(ctxId), ioCtx)
            }
        case SEND_POSTED:
        case NULL_POSTED:
        default:
        }
    }
}

func (ss *IoCompletionServer) DoAcceptEx(rootCtx *IoCompletionRootContext, ioCtx *IoCompletionContext) (err error) {
    nFdCtx := NewRootContext()
    nFdCtx.socket = ioCtx.socket
    addrSize := uint32(unsafe.Sizeof(windows.RawSockaddrAny{}))

    var localAddr, remoteAddr *windows.RawSockaddrAny
    lrsalen := int32(addrSize)
    rrsalen := int32(addrSize)

    // 與windows C++ 不同，此處函數無需去函數指針即可使用
    windows.GetAcceptExSockaddrs(ioCtx.wsaBuf.Buf, ioCtx.wsaBuf.Len-(addrSize+16)*2,
        addrSize+16, addrSize+16, &localAddr, &lrsalen, &remoteAddr, &rrsalen)

    if ss.recvFunc != nil {
        ss.recvFunc(ioCtx.data[:ioCtx.overlapped.InternalHigh])
    }

    // 繼承listen socket的屬性
    err = windows.Setsockopt(nFdCtx.socket, windows.SOL_SOCKET, windows.SO_UPDATE_ACCEPT_CONTEXT,
        (*byte)(unsafe.Pointer(&ss.rootCtx.socket)), int32(unsafe.Sizeof(ss.rootCtx.socket)))
    if err != nil {
        return errors.Wrap(err, "syscall.AcceptEx")
    }

    err = windows.SetsockoptInt(nFdCtx.socket, windows.SOL_SOCKET, windows.SO_RCVBUF, 65535)
    if err != nil {
        return errors.Wrap(err, "windows.SetsockoptInt")
    }

    // 綁定到完成埠, 此步驟很關鍵
    handle, err := windows.CreateIoCompletionPort(nFdCtx.socket,
        ss.hIOCompletionPort, uint32(nFdCtx.socket), 0)
    if err != nil {
        return errors.Wrap(err, "syscall.CreateIoCompletionPort")
    } else {
        fmt.Println(handle, rootCtx.socket)
    }

    // 投遞接收請求, 此處可以自行修改
    for i := 0; i < 16; i++ {
        nFdIoCtx := nFdCtx.NewIoContext()
        nFdIoCtx.socket = nFdCtx.socket
        if err = ss.Receive(nFdIoCtx); err != nil {
            return err
        }
    }

    //投遞接收連接請求
    if err = ss.AcceptEx(ioCtx); err != nil {
        return err
    }

    // 保存到context中
    ss.saveIoRootCtx(uint32(nFdCtx.socket), nFdCtx)
    return nil
}

func (ss *IoCompletionServer) AcceptEx(ctx *IoCompletionContext) (err error) {
    ctx.socket = windows.Handle(C.mWSASocket())

    dwBytes := uint32(0)
    addrSize := uint32(unsafe.Sizeof(windows.RawSockaddrAny{}))
    ctx.opType = ACCEPT_POSTED
    //err = syscall.AcceptEx(ss.rootCtx.socket, ctx.socket, ctx.wsaBuf.Buf,
    //  ctx.wsaBuf.Len-2*(addrSize+16), addrSize+16,
    //  addrSize+16, &dwBytes, &ctx.overlapped)

    //windows.WSAIoctl(ss.rootCtx.socket, windows.SIO_GET_EXTENSION_FUNCTION_POINTER)
    err = windows.AcceptEx(ss.rootCtx.socket, ctx.socket, ctx.wsaBuf.Buf,
        ctx.wsaBuf.Len-2*(addrSize+16), addrSize+16,
        addrSize+16, &dwBytes, &ctx.overlapped)
    if err != nil {
        if err == windows.Errno(997) { // ERROR_IO_PENDING 表示尚未接收到鏈接
            err = nil
        } else {
            err = errors.Wrap(err, "syscall.AcceptEx")
        }
    }

    return err
}

func (ss *IoCompletionServer) DoReceive(rootCtx *IoCompletionRootContext, ctx *IoCompletionContext) {
    if ctx.overlapped.InternalHigh == 0 {
        if rootCtx != nil {
            ss.remove(uint32(rootCtx.socket))
            C.mClose(C.int(rootCtx.socket))
        }
        return
    }

    if ss.recvFunc != nil {
        ss.recvFunc(ctx.data[:ctx.overlapped.InternalHigh])
    }

    ss.Receive(ctx)
}

func (ss *IoCompletionServer) Receive(ioCtx *IoCompletionContext) error {
    recv := uint32(0)
    flags := uint32(0)
    ioCtx.opType = RECV_POSTED

    err := windows.WSARecv(ioCtx.socket, &ioCtx.wsaBuf,
        1, &recv, &flags, &ioCtx.overlapped, nil)
    if err != nil {
        if err == windows.Errno(997) { // ERROR_IO_PENDING 表示尚未接收到數據
            err = nil
        } else {
            err = errors.Wrap(err, "syscall.AcceptEx")
        }
    }

    return err
}

func setDefaultSockOpt(handle windows.Handle) error {
    err := windows.SetsockoptInt(handle, windows.SOL_SOCKET, windows.SO_REUSEADDR, 1)
    if err != nil {
        return errors.Wrap(err, "syscall.SetsockoptInt")
    }

    //err = windows.SetsockoptInt(handle, windows.SOL_SOCKET, windows.SO, 1)
    //if err != nil {
    //  return errors.Wrap(err, "syscall.SetsockoptInt")
    //}

    return nil
}

func (ss *IoCompletionServer) Start() error {
    fmt.Println(windows.WSAStartup(2, &windows.WSAData{}))

    // 初始創建一個用於綁定的 listen socket 的 IoCompletion 句柄
    hIOCompletionPort, err := windows.CreateIoCompletionPort(windows.InvalidHandle, 0, 0, 0)
    if err != nil {
        return errors.Wrap(err, "syscall.CreateIoCompletionPort")
    }

    ss.hIOCompletionPort = hIOCompletionPort

    rootCtx := NewRootContext()
    rootCtx.socket = windows.Handle(C.mWSASocket())
    setDefaultSockOpt(rootCtx.socket)
    ss.rootCtx = rootCtx

    handle, err := windows.CreateIoCompletionPort(rootCtx.socket,
        hIOCompletionPort, uint32(ss.rootCtx.socket), 0)
    if err != nil {
        return errors.Wrap(err, "syscall.CreateIoCompletionPort")
    } else {
        fmt.Println(handle, rootCtx.socket)
    }

    sockAddr := windows.SockaddrInet4{}
    sockAddr.Port = ss.Port

    if err := windows.Bind(rootCtx.socket, &sockAddr); err != nil {
        return errors.Wrap(err, "syscall.Bind")
    }

    if err := windows.Listen(rootCtx.socket, MAX_POST_ACCEPT); err != nil {
        return errors.Wrap(err, "windows.Listen")
    }

    ss.rootCtx = rootCtx

    if err := ss.AcceptEx(rootCtx.NewIoContext()); err != nil {
        return err
    }
    return nil
}

example

完成埠服務使用示例:

ss = &StreamServer{
    Addr: "127.0.0.1:10050",
    Port: 10050,
    accepts: sync.Map{},
}

ss.RegisterReceiveFunc(func(data []byte) error {
    fmt.Println("receive data len:", string(data))
    return nil
})

// 可以啓動多個攜程來接收請求，但是需要特別註意的是
// 多攜程可能會導致接受數據包時序發生亂序
ss.Listen()

結尾

以上代碼經過實際測試檢驗，可以正常使用，尚未與標準庫進行 效率\性能 對比，沒有實現 send 功能,此處需要提醒的是,使用 IoCompletion Port 發送數據要註意時序的把握。

IoCompletion Port 是windows 系統中十分優秀的IO模型, 深入瞭解其工作機制及原理, 也有助於我們對操作系統 IO 數據處理的機制有更清晰的認知。

參考


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

-Advertisement-
Play Games
更多相關文章
  • 2020為什麼要學Python？這個理由至關重要！
    學Python的很多，不只是程式員學Python，運營、產品、測試都在學Python 除了互聯網，辦公一族都開始學Python了，學Python的理由千萬條。 因為人生苦短 這句話最初出自《Java 編程思想》作者Bruce Eckel。 有因為Python牛到飛起的 再看看Python鳥怎麼喝水的 ...
  • Python 教你 4 行代碼開發新聞網站通用爬蟲
    \ ​ GNE（GeneralNewsExtractor）是一個通用新聞網站正文抽取模塊，輸入一篇新聞網頁的 HTML， 輸出正文內容、標題、作者、發佈時間、正文中的圖片地址和正文所在的標簽源代碼。GNE在提取今日頭條、網易新聞、游民星空、 觀察者網、鳳凰網、騰訊新聞、ReadHub、新浪新聞等數百 ...
  • Android中常用ADB命令集錦
    收集的Android測試或者開發中常用的aadb命令，可以使用Ctrl+F快速搜索### ADB命令集錦： adb --help //adb幫助 adb start-server //啟動adb server adb kill-server //關閉adb server adb devices // ...
  • Python 3.8 已發佈，現在是切換至新版本的好時機嗎？
    Python 3.8 已於前兩周正式發佈，不過目前大多數開發者主要使用的仍是 Python 3.7.x 或更舊的版本。 ！ 事實上，立刻切換到使用 Python 3.8 可能會引發一些問題。想知道何時切換至 Python 3.8？下文將簡要概述切換主要的 Python 版本以及進行切換時可能會遇到的 ...
  • 深入delphi編程理解之消息（一）WINDOWS原生視窗編寫及消息處理過程
    通過以sdk方式編製windows視窗程式，對理解windows消息驅動機制和delphi消息編程有很大的幫助。 sdk編製windows視窗程式的步驟： 1、對TWndClass對象進行賦值; 2、向系統註冊wndclass對象(RegisterClass)； 3、CreateWindow創建視窗 ...
  • java提取字元串數字，Java獲取字元串中的數字
    ©Copyright 蕃薯耀 2020-01-17 https://www.cnblogs.com/fanshuyao/ 具體的方法如下： /** * 把字元串數字類型的數字取出來（只取遇到非數字字元前，包括空格） * @param str * <li>"1-0我5013我24a5c6" 》 1</ ...
  • 樂優商城1 —— 工程搭建
    你可以訪問 "碼雲 樂優商城" 來獲取關於樂優商城的工程代碼。 你可以訪問 "百度雲 樂優優商城" 密碼：ppzy 來獲取關於樂優商城的資料。 一、創建父工程 Maven Project 用來管理依賴 GroupId：項目中唯一標識符，對應的是java中的包結構，在這裡表示項目中的結構 Artifa ...
  • SpringBoot啟動後自動打開瀏覽器訪問項目
    之前我們用SSM或者SSH進行JAVA WEB開發的時候,IDEA 需要配置Tomcat然後把項目放到tomcat運行，tomcat啟動的時候會自動打開瀏覽器去訪問項目，但是SpringBoot是內嵌tomcat的，項目啟動成功後無法自主訪問，需要我們手動打開瀏覽器輸入url訪問，我覺得這樣很不習慣... ...
一周排行
    -Advertisement-
    Play Games
  • Dapr Outbox 執行流程
    Dapr Outbox 是1.12中的功能。 本文只介紹Dapr Outbox 執行流程，Dapr Outbox基本用法請閱讀官方文檔 。本文中appID=order-processor，topic=orders 本文前提知識：熟悉Dapr狀態管理、Dapr發佈訂閱和Outbox 模式。 Outbo ...
  • Advanced .Net Debugging 7：托管堆與垃圾收集
    引言 在前幾章我們深度講解了單元測試和集成測試的基礎知識，這一章我們來講解一下代碼覆蓋率,代碼覆蓋率是單元測試運行的度量值,覆蓋率通常以百分比表示，用於衡量代碼被測試覆蓋的程度，幫助開發人員評估測試用例的質量和代碼的健壯性。常見的覆蓋率包括語句覆蓋率（Line Coverage）、分支覆蓋率（Bra ...
  • C#S7.NET實現西門子PLCDB塊數據採集的完整步驟
    前言 本文介紹瞭如何使用S7.NET庫實現對西門子PLC DB塊數據的讀寫，記錄了使用電腦模擬，模擬PLC，自至完成測試的詳細流程，並重點介紹了在這個過程中的易錯點，供參考。 用到的軟體： 1.Windows環境下鏈路層網路訪問的行業標準工具（WinPcap_4_1_3.exe）下載鏈接：http ...
  • WPF控制項：密碼框綁定MVVM
    從依賴倒置原則（Dependency Inversion Principle, DIP）到控制反轉（Inversion of Control, IoC）再到依賴註入（Dependency Injection, DI）的演進過程，我們可以理解為一種逐步抽象和解耦的設計思想。這種思想在C#等面向對象的編 ...
  • Python中的私有屬性與私有方法
    關於Python中的私有屬性和私有方法 Python對於類的成員沒有嚴格的訪問控制限制，這與其他面相對對象語言有區別。關於私有屬性和私有方法，有如下要點： 1、通常我們約定，兩個下劃線開頭的屬性是私有的（private）。其他為公共的（public）； 2、類內部可以訪問私有屬性（方法）； 3、類外 ...
  • C++ 訪問說明符詳解：封裝數據，控制訪問，提升安全性
    C++ 訪問說明符 訪問說明符是 C++ 中控制類成員（屬性和方法）可訪問性的關鍵字。它們用於封裝類數據並保護其免受意外修改或濫用。 三種訪問說明符： public：允許從類外部的任何地方訪問成員。 private：僅允許在類內部訪問成員。 protected：允許在類內部及其派生類中訪問成員。 示 ...
  • static_cast與dynamic_cast到底是什麼？
    寫這個隨筆說一下C++的static_cast和dynamic_cast用在子類與父類的指針轉換時的一些事宜。首先，【static_cast，dynamic_cast】【父類指針，子類指針】，兩兩一組，共有4種組合：用 static_cast 父類轉子類、用 static_cast 子類轉父類、使用 ...
  • C語言數據結構：雙向鏈表的增刪操作
    /******************************************************************************************************** * * * 設計雙向鏈表的介面 * * * * Copyright (c) 2023-2 ...
  • @ComponentScan註解的實現，Spring掃描包的過程
    相信接觸過spring做開發的小伙伴們一定使用過@ComponentScan註解 @ComponentScan("com.wangm.lifecycle") public class AppConfig { } @ComponentScan指定basePackage，將包下的類按照一定規則註冊成Be ...
  • opensips開啟python支持
    操作系統 ：CentOS 7.6_x64 opensips版本： 2.4.9 python版本：2.7.5 python作為腳本語言，使用起來很方便，查了下opensips的文檔，支持使用python腳本寫邏輯代碼。今天整理下CentOS7環境下opensips2.4.9的python模塊筆記及使用 ...
所有分類