闽江人才网,齐月宾,舞蹈春天在哪里串词
项目地址:
开发服务端程序的一个基本任务是处理并发连接,现在服务端网络编程处理并发连接主要有两种方式:
在线程很宝贵的情况下,常见的服务器编程模型有如下几种:
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,总体来说可以接受。
这是个典型的生产者-消费者问题。产生日志的线程将日志先存到缓冲区,日志消费线程将缓冲区中的日志写到磁盘。要保证两个线程的临界区尽可能小。
每条log_debug等语句对应创建一个匿名logwrapper对象,同时搜集日志信息保存到logevent对象中,匿名对象创建完毕就会调用析构函数,在logwrapper析构函数中将logevent送到logger中,logger再送往不同的目的地,比如控制台,文件等。
asyncfileappend对外提供append方法,前端logger只需要调用这个方法往里面塞日志,不用担心会被阻塞。
前端和后端都维护一个缓冲区。
第一种情况:前端写日志较慢,三秒内还没写满一个缓冲区。后端线程会被唤醒,进入临界区,在临界区内交换两个buffer的指针,出临界区后前端cur指向的缓冲区又是空的了,后端buffer指向的缓冲区为刚才搜集了日志的缓冲区,后端线程随后将buffer指向的缓冲区中的日志写到磁盘中。临界区内只交换两个指针,所以临界区很小。
第二种情况:前端写日志较快,三秒内已经写满了一个缓冲区。比如两秒的时候已经写满了第一个缓冲区,那么将cur指针保存到一个向量buffers_中,然后开辟一块新的缓冲区,另cur指向这块新缓冲区。然后唤醒后端消费线程,后端线程进入临界区,将cur和后端buffer_指针进行交换,将前端buffers_向量和后端persist_buffers_向量进行swap(对于std::vector也是指针交换)。出了临界区后,前端的cur始终指向一块干净的缓冲区,前端的向量buffers_也始终为空,后端的persist_buffers_向量中始终保存着有日志的缓冲区的指针。临界区同样很小仅仅是几个指针交换。
成员变量:
成员函数:
ucontext系列函数:
int getcontext(ucontext_t *ucp)
: 将此刻的上下文保存到ucp指向的结构中。int setcontext(const ucontext_t *ucp)
: 调用成功后不会返回,执行流转移到ucp指向的上下文。void makecontext(ucontext_t *ucp, void (*func)(), int argc, ...)
:重新设置ucp指向的上下文为func函数起始处。ucp结构由getcontext()获取。后续以ucp为参数调用setcontext()或者swapcontext()执行流将转到func函数。int swapcontext(ucontext_t *oucp, const ucontext_t *ucp)
:保存当前上下文到oucp,并激活ucp指向的上下文。不能太大:协程多了,内存浪费。
不能太小:使用者可能无意在栈上分配一个缓冲区,导致栈溢出。
暂时先固定为128k。
目前是非抢占式调度。只能由协程主动或者协程执行完毕,才会让出cpu。
两个协程间可能需要同步操作,比如协程1需要等待某个条件才能继续运行,线程2修改条件然后通知协程1。
目前实现了简陋的wait/notify机制,见coroutinecondition。
线程栈上的对象,线程退出后自动销毁,生命周期大可不必操心。
成员变量:
成员函数:
每个线程都有一个本地变量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(); } } }
可能出现这种情况:正在执行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将变为可读。
要想实现在协程中遇到耗时操作不阻塞当前io线程,需要对一些系统函数进行hook。
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说简单点就是将参数传给服务端,服务端根据参数找到对应的函数执行,得出一个响应,再将响应传回给客户端。客户端的参数对象如何通过网络传到服务端呢?这就涉及到序列化和反序列化。
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 客户端接收响并解析 <---------------- 服务端将响应发回给客户端
如对本文有疑问,请在下面进行留言讨论,广大热心网友会与你互动!! 点击进行留言回复
如何在没有core文件的情况下用dmesg+addr2line定位段错误
用QT制作3D点云显示器——QtDataVisualization
网友评论