發佈一個基於協程和事件迴圈的c++網路庫

来源:https://www.cnblogs.com/gatsby123/archive/2019/12/20/12074574.html
-Advertisement-
Play Games

[TOC] 項目地址：https://github.com/gatsbyd/melon 介紹 開發服務端程式的一個基本任務是處理併發連接，現在服務端網路編程處理併發連接主要有兩種方式： 1. 當“線程”很廉價時，一臺機器上可以創建遠高於CPU數目的“線程”。這時一個線程只處理一個TCP連接，通常使用 ...


目錄

項目地址：https://github.com/gatsbyd/melon

介紹

開發服務端程式的一個基本任務是處理併發連接，現在服務端網路編程處理併發連接主要有兩種方式：

  1. 當“線程”很廉價時，一臺機器上可以創建遠高於CPU數目的“線程”。這時一個線程只處理一個TCP連接，通常使用阻塞IO。例如Go goroutine。這裡的“線程”由語言的runtime自行調度。
  2. 當線程很寶貴時，一臺機器上只能創建與CPU數目相當的線程。這時一個線程要處理多個TCP連接上的IO，通常使用非阻塞IO和IO multiplexing。C++編程主要採用這種方式。

線上程很寶貴的情況下，常見的伺服器編程模型有如下幾種：

  1. 每個請求創建一個線程，使用阻塞式IO操作（或者叫thread per connection）。這種模型的優點是可以使用阻塞操作，缺點是伸縮性不強，每台機器能創建的線程是有限的，32位的機器應該不超過400個。
  2. 非阻塞IO+IO多路復用（或者叫one loop per thread或者Reactor）+ 線程池。

melon是基於Reactor模式的Linux C++網路服務框架，集合了上述兩種方式，實現了協程的概念，對一些函數進行了hook，所以可以像操作阻塞IO一樣進行編程。

使用

在工程主目錄下新建build目錄，進入build目錄，

cmake ..
make  all

編譯完成後，example和test中的可執行程式分別位於build目錄下的example和test中。

以echo服務端為例，

void handleClient(TcpConnection::Ptr conn){
    conn->setTcpNoDelay(true);
    Buffer::Ptr buffer = std::make_shared<Buffer>();
    while (conn->read(buffer) > 0) {
        conn->write(buffer);
    }

    conn->close();
}


int main(int args, char* argv[]) {
    if (args != 2) {
        printf("Usage: %s threads\n", argv[0]);
        return 0;
    }
    Logger::setLogLevel(LogLevel::INFO);
    Singleton<Logger>::getInstance()->addAppender("console", LogAppender::ptr(new ConsoleAppender()));

    IpAddress listen_addr(5000);
    int threads_num = std::atoi(argv[1]);

    Scheduler scheduler(threads_num);
    scheduler.startAsync();
    TcpServer server(listen_addr, &scheduler);
    server.setConnectionHandler(handleClient);
    server.start();

    scheduler.wait();
    return 0;
}

只需要為TcpServer設置連接處理函數，在連接處理函數中，參數TcpConnection::Ptr conn代表此次連接，可以像阻塞IO一樣進行讀寫，如果發生阻塞，當前協程會被切出去，直到可讀或者可寫事件到來時，該協程會被重新執行。

性能

硬體環境：Intel Core i7-8550U CPU 1.80GHz，8核，8G RAM
軟體環境：操作系統為Ubuntu 16.04.2 LTS，g++版本5.4.0
測試對象：asio 1.14.0, melon 0.1.0

測試方法：
根據asio的測試方法，用echo協議來測試。客戶端和服務端建立連接，客戶端向服務端發送一些數據，服務端收到後將數據原封不動地發回給客戶端，客戶端收到後再將數據發給服務端，直到一方斷開連接位置。
melon的測試代碼在test/TcpClient_test.cpp和test/TcpServer_test.cpp。
asio的測試代碼在/src/tests/performance目錄下的client.cpp和server.cpp。

測試1：客戶端和伺服器運行在同一臺機器上，均為單線程，測試併發數為1/10/100/1000/10000的吞吐量。

吞吐量（MiB/s） 1 10 100 1000
melon 202 388 376 327
asio 251 541 489 436

測試2：客戶端和伺服器運行在同一臺機器上，均為開啟兩個線程，測試併發連接數100的吞吐量。

吞吐量（MiB/s） 2個線程
melon 499
asio 587

從數據看目前melon的性能還不及asio，但是考慮到melon存在協程切換的成本和0.1.0版本沒有上epoll，協程切換也是用的ucontext，總體來說可以接受。

實現

日誌庫

需求

  1. 有多種日誌級別，DEBUG, INFO, WARN, ERROR, FATAL。
  2. 可以有多個目的地，比如文件，控制台，可以拓展。
  3. 日誌文件達到指定大小時自動roll。
  4. 時間戳精確到微秒。使用gettimeofday(2)，在x86-64Linux上不會陷入內核。
  5. 線程安全。
  6. 寫日誌過程不能是同步的，否則會阻塞IO線程。

這是個典型的生產者-消費者問題。產生日誌的線程將日誌先存到緩衝區，日誌消費線程將緩衝區中的日誌寫到磁碟。要保證兩個線程的臨界區儘可能小。

總體結構如下

日誌結構

每條LOG_DEBUG等語句對應創建一個匿名LogWrapper對象，同時搜集日誌信息保存到LogEvent對象中，匿名對象創建完畢就會調用析構函數，在LogWrapper析構函數中將LogEvent送到Logger中，Logger再送往不同的目的地，比如控制台，文件等。

非同步文件Appender實現

AsyncFileAppend對外提供append方法，前端Logger只需要調用這個方法往裡面塞日誌，不用擔心會被阻塞。

前端和後端都維護一個緩衝區。
第一種情況：前端寫日誌較慢，三秒內還沒寫滿一個緩衝區。後端線程會被喚醒，進入臨界區，在臨界區內交換兩個buffer的指針，出臨界區後前端cur指向的緩衝區又是空的了，後端buffer指向的緩衝區為剛纔搜集了日誌的緩衝區，後端線程隨後將buffer指向的緩衝區中的日誌寫到磁碟中。臨界區內只交換兩個指針，所以臨界區很小。
情況1

第二種情況：前端寫日誌較快，三秒內已經寫滿了一個緩衝區。比如兩秒的時候已經寫滿了第一個緩衝區，那麼將cur指針保存到一個向量buffers_中，然後開闢一塊新的緩衝區，另cur指向這塊新緩衝區。然後喚醒後端消費線程，後端線程進入臨界區，將cur和後端buffer_指針進行交換，將前端buffers_向量和後端persist_buffers_向量進行swap(對於std::vector也是指針交換)。出了臨界區後，前端的cur始終指向一塊乾凈的緩衝區，前端的向量buffers_也始終為空，後端的persist_buffers_向量中始終保存著有日誌的緩衝區的指針。臨界區同樣很小僅僅是幾個指針交換。
情況2

協程

類圖

協程類圖

成員變數：

  1. c_id_：當前協程id。
  2. context_：協程上下文。
  3. cb_：協程執行的函數。
  4. stack_size_：協程棧大小。
  5. statck_：協程棧。
  6. state_：協程狀態。

成員函數：

  1. swapIn()：執行當前協程，只能由主協程調用。
  2. SwapOut()：靜態函數，讓出當前協程的CPU，執行主協程，主協程會進行協程調度，將CPU控制權轉到另一個協程。
  3. GetCurrentCoroutine()：獲取當前線程正在執行的協程。
  4. GetMainCoroutine()：獲取當前線程的的主協程。

原理

ucontext系列函數：

  1. int getcontext(ucontext_t *ucp)： 將此刻的上下文保存到ucp指向的結構中。
  2. int setcontext(const ucontext_t *ucp)： 調用成功後不會返回，執行流轉移到ucp指向的上下文。
  3. void makecontext(ucontext_t *ucp, void (*func)(), int argc, ...)：重新設置ucp指向的上下文為func函數起始處。ucp結構由getcontext()獲取。後續以ucp為參數調用setcontext()或者swapcontext()執行流將轉到func函數。
  4. int swapcontext(ucontext_t *oucp, const ucontext_t *ucp)：保存當前上下文到oucp，並激活ucp指向的上下文。

需要考慮的問題

協程棧大小

不能太大：協程多了，記憶體浪費。
不能太小：使用者可能無意在棧上分配一個緩衝區，導致棧溢出。
暫時先固定為128K。

調度策略

目前是非搶占式調度。只能由協程主動或者協程執行完畢，才會讓出CPU。

協程同步

兩個協程間可能需要同步操作，比如協程1需要等待某個條件才能繼續運行，線程2修改條件然後通知協程1。
目前實現了簡陋的wait/notify機制，見CoroutineCondition。

協程調度

類圖

協程調度

Processer

線程棧上的對象，線程退出後自動銷毀，生命周期大可不必操心。

成員變數：

  1. poller_：Poller。
  2. coroutines_：當前線程待執行的協程隊列。

成員函數：

  1. addTask()：添加任務。
  2. run()：開始進行協程調度。

協程調度示意圖

協程調度示意圖

每個線程都有一個本地變數t_cur_cotourine指向當前正在執行的協程對象。

調度過程

Processer.run()函數作為Main協程進行調度，沒有協程在協程隊列時，執行Poll協程，該協程執行poll()函數。以read操作為例，某個協程在執行read的操作時，如果數據沒有準備好，就會將<fd， 當前協程對象>對註冊到Poller中，然後掛起。如果所有協程都阻塞了，那麼會執行Poll協程等待poll()函數返回，poll()函數返回後，如果有事件發生，會根據之前註冊的<fd， 協程對象>，將協程對象重新加入調度隊列，此時read已經有數據可讀了。

Main協程對應的代碼邏輯如下：

void Processer::run() {
    if (GetProcesserOfThisThread() != nullptr) {
        LOG_FATAL << "run two processer in one thread";
    } else {
        GetProcesserOfThisThread() = this;
    }
    melon::setHookEnabled(true);
    Coroutine::Ptr cur;

    //沒有可以執行協程時調用poll協程
    Coroutine::Ptr poll_coroutine = std::make_shared<Coroutine>(std::bind(&Poller::poll, &poller_, kPollTimeMs), "Poll");

    while (!stop_) {
        {
            MutexGuard guard(mutex_);
            //沒有協程時執行poll協程
            if (coroutines_.empty()) {
                cur = poll_coroutine;
                poller_.setPolling(true);
            } else {
                for (auto it = coroutines_.begin();
                        it != coroutines_.end();
                            ++it) {
                    cur = *it;
                    coroutines_.erase(it);
                    break;
                }
            }
        }
        cur->swapIn();
        if (cur->getState() == CoroutineState::TERMINATED) {
            load_--;
        }
    }
}

Poll協程對應的代碼邏輯如下：

void PollPoller::poll(int timeout) {
    while (!processer_->stoped()) {
        is_polling_ = true;
        int num = ::poll(&*pollfds_.begin(), pollfds_.size(), timeout);
        is_polling_ = false;
        if (num == 0) {
        } else if (num < 0) {
            if (errno != EINTR) {
                LOG_ERROR << "poll error, errno: " << errno << ", error str:" << strerror(errno);
            }
        } else {
            std::vector<int> active_fds;
            for (const auto& pollfd : pollfds_) {
                if (pollfd.revents > 0) {
                    --num;
                    active_fds.push_back(pollfd.fd);
                    if (num == 0) {
                        break;
                    }
                }
            }
            for (const auto& active_fd : active_fds) {
                auto coroutine = fd_to_coroutine_[active_fd];
                assert(coroutine != nullptr);

                removeEvent(active_fd);
                processer_->addTask(coroutine);
            }   
        }
        Coroutine::SwapOut();
    }
}
}
為什麼需要一個wake協程

可能出現這種情況：正在執行Poll協程，並且沒有事件到達，這時新加入一個協程，如果沒有機制將Poll協程從poll()函數中喚醒，那麼這個新的協程將無法得到執行。wake協程會read eventfd，此時會將<eventfd, wake協程>註冊到Poller中，如果有新的協程加入，會往eventfd寫1位元組的數據，那麼poll()函數就會被喚醒，從而Poll協程讓出CPU，新加入的協程被調度。

定時器

原理

#include <sys/timerfd.h>
int timerfd_create(int clockid, int flags); //創建一個timer對象，返回一個文件描述符timer fd代表這個timer對象。
int timerfd_settime(int fd, int flags,
                           const struct itimerspec *new_value,
                           struct itimerspec *old_value);  //為timer對象設置一個時間間隔，倒計時結束後timer fd將變為可讀。

定時器

  1. 定時器專門占用一個線程。這個線程中加入一個定時器協程，該協程會去讀取timer fd，可讀後說明有定時器超時，然後執行定時器對應的任務。
  2. TimerManager維護一個定時器隊列。每一項包含定時器觸發時間和對應的回調。
  3. TimerManager.addTimer()將新的<timer, 回調>加入到隊列中。如果這個定時器是最先到期的那麼調用timerfd_settime()重新設置timer fd的到期時間。timer fd到期後，將從Poll協程中返回，然後執行定時器協程，該協程中讀取timer fd，然後根據現在的時間，將定時器隊列中超時的項刪除，並將超時的項的回調作為新的協程執行。
  4. 這個隊列可以由multimap來實現，multimap由紅黑樹實現，內部是有序的。紅黑樹本質就是一顆二叉樹，只不過為了防止多次的操作變得不平衡，增加了一些維持平衡的操作。
  5. 如何刪除定時器，每個定時器分配一個id，TimerManager內部維護一個id到定時器時間戳的映射sequence_2_timestamp_。cancel()時，根據id去sequence_2_timestamp_中找有沒有對應的定時器，如果有，將這個時間戳從時間戳隊列中刪除，必要時重新調用timerfd_settime()。

Hook

要想實現在協程中遇到耗時操作不阻塞當前IO線程，需要對一些系統函數進行hook。

  1. 可以用dlsym(3)來獲取想要hook的函數的函數指針，先保存起來，如果想要用到原函數，可以通過保存的函數指針進行調用。
  2. 定義自己的同名函數，覆蓋想要hook的函數。以sleep(3)為例。
unsigned int sleep(unsigned int seconds) {
    melon::Processer* processer = melon::Processer::GetProcesserOfThisThread();
    if (!melon::isHookEnabled()) {
        return sleep_f(seconds);
    }

    melon::Scheduler* scheduler = processer->getScheduler();
    assert(scheduler != nullptr);
    scheduler->runAt(melon::Timestamp::now() + seconds * melon::Timestamp::kMicrosecondsPerSecond, melon::Coroutine::GetCurrentCoroutine());
    melon::Coroutine::SwapOut();
    return 0;
}

我們自己定義的sleep不會阻塞線程，而是將當前協程切出去，讓CPU執行其它協程，等時間到了再執行當前協程。這樣就模擬了sleep的操作，同時不會阻塞當前線程。

RPC實現

參數序列化及反序列化

rpc說簡單點就是將參數傳給服務端，服務端根據參數找到對應的函數執行，得出一個響應，再將響應傳回給客戶端。客戶端的參數對象如何通過網路傳到服務端呢？這就涉及到序列化和反序列化。
melon選擇Protobuf，Protobuf具有很強的反射能力，在僅知道typename的情況下就能創建typename對應的對象。

google::protobuf::Message* ProtobufCodec::createMessage(const std::string& typeName) {
    google::protobuf::Message* message = nullptr;
    const google::protobuf::Descriptor* descriptor =
            google::protobuf::DescriptorPool::generated_pool()->FindMessageTypeByName(typeName);
    if (descriptor) {
        const google::protobuf::Message* prototype =
            google::protobuf::MessageFactory::generated_factory()->GetPrototype(descriptor);
        if (prototype) {
            message = prototype->New();
        }
    }
    return message;
}

上述函數根據參數typename就能創建一個Protobuf對象，這個新建的對象結合序列化後的Protobuf數據就能在服務端生成一個和客戶端一樣的Protobuf對象。

數據格式

|-------------------|
|   total  byte     |        總的位元組數
|-------------------|
|     typename      |         類型名
|-------------------|
|    typename len   |         類型名長度
|-------------------|
|   protobuf data   |          Protobuf對象序列化後的數據
|-------------------|
|       checksum    |        整個消息的checksum
|-------------------|

某次rpc的過程如下：

客戶端包裝請求併發送    ---------------->     服務端接收請求
                                            服務端解析請求，找到並執行對應的service::method
客戶端接收響並解析      <----------------     服務端將響應發回給客戶端

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

-Advertisement-
Play Games
更多相關文章
  • SpringCloud+Eureka+Feign+Ribbon的簡化搭建流程和CRUD練習
    作者:個人微信公眾號:程式猿的月光寶盒 環境:win10 idea2019 jdk8 1.搭建Eureka服務模塊 1.1 新建eureka服務模塊(Sping Initializr) 取名為 ,並添加如下Dependencies: 1.2 配置 1.3 在啟動類前加上註解 1.4 開啟啟動器,並訪 ...
  • java嵌套介面
    java嵌套介面 package object; class A { //嵌套在類中的介面，可以被private,protected,default和public四種許可權修飾 interface B { void f(); } public class BImp implements B { pub ...
  • Python學習之while
    count = 0while count <= 5: print('人生苦短，我用Python!', 'count = ', count) count += 1 ...
  • 基於Vue的前後端分離項目實踐
    https://blog.csdn.net/u011244682/article/details/82853213 ...
  • 高併發與多線程的關係、區別、高併發的技術方案
    什麼是高併發？ 高併發（High Concurrency）是一種系統運行過程中遇到的一種“短時間內遇到大量操作請求”的情況，主要發生在web系統集中大量訪問收到大量請求（例如：12306的搶票情況；天貓雙十一活動）。該情況的發生會導致系統在這段時間內執行大量操作，例如對資源的請求，資料庫的操作等。 ...
  • Redis有哪幾種數據類型
    Redis支持五種數據類型：string（字元串），hash（哈希），list（列表），set（集合）及zset(sorted set：有序集合)。 String（字元串） string 是 redis 最基本的類型，你可以理解成與 Memcached 一模一樣的類型，一個 key 對應一個 val ...
  • 一些實用的 Laravel 小技巧
    Laravel 中一些常用的小技巧，說不定你就用上了。 1.側欄 網站一般都有側欄，用來顯示分類，標簽，熱門文章，熱門評論啥的，但是這些側欄都是相對獨立的模塊，如果在每一個引入側欄的視圖中都單獨導入與視圖有關的數據的話，未免太冗餘了。所以最佳的做法是：新建一個widgets視圖文件夾，再利用Lara ...
  • django 用戶與許可權管理
    django中使用content_type表進行存儲app與model的關係。在permission表中設立了name（許可權的名字，中英文均可）content_type_id（與content_type是外鍵關係，用於綁定model）codename（在寫代碼時使用的許可權名）那麼如何自定義許可權呢？ ...
一周排行
    -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.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類