發布一個基于協程和事件回圈的c++網路庫

目錄

• 介紹
• 使用
• 性能
• 實作
  • 日志庫
    • 需求
    • 總體結構如下
    • 異步檔案Appender實作
  • 協程
    • 類圖
    • 原理
    • 需要考慮的問題
      • 協程堆疊大小
      • 調度策略
      • 協程同步
  • 協程調度
    • 類圖
      • Processer
    • 協程調度示意圖
      • 調度程序
      • 為什么需要一個wake協程
  • 定時器
    • 原理
  • Hook
  • RPC實作
    • 引數序列化及反序列化
    • 資料格式

專案地址：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的吞吐量，

測驗2：客戶端和服務器運行在同一臺機器上，均為開啟兩個執行緒，測驗并發連接數100的吞吐量，

從資料看目前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指向的緩沖區中的日志寫到磁盤中，臨界區內只交換兩個指標，所以臨界區很小，

第二種情況：前端寫日志較快，三秒內已經寫滿了一個緩沖區，比如兩秒的時候已經寫滿了第一個緩沖區，那么將cur指標保存到一個向量buffers_中，然后開辟一塊新的緩沖區，另cur指向這塊新緩沖區，然后喚醒后端消費執行緒，后端執行緒進入臨界區，將cur和后端buffer_指標進行交換，將前端buffers_向量和后端persist_buffers_向量進行swap(對于std::vector也是指標交換)，出了臨界區后，前端的cur始終指向一塊干凈的緩沖區，前端的向量buffers_也始終為空，后端的persist_buffers_向量中始終保存著有日志的緩沖區的指標，臨界區同樣很小僅僅是幾個指標交換，

協程

類圖

成員變數：

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
客戶端接收響并決議      <----------------     服務端將回應發回給客戶端

標籤：C++

上一篇：c++-面向物件類的示例-求周長面積，判斷體積相等-檔案操作和一般操作

下一篇：購買電腦注冊時使用中文用戶名，怎樣將檔案修改成英文