अतुल्यकालिक: वापस भविष्य के लिए

परिचय

तुल्यकालिक एकल-थ्रेडेड सर्वर

 typedef std::string Buffer; // forward declaration struct Acceptor; struct Socket { friend struct Acceptor; Socket(); Socket(Socket&& s); //     void read(Buffer&); //    ,     void readSome(Buffer&); //      until int readUntil(Buffer&, const Buffer& until); //     void write(const Buffer&); //   void close(); private: boost::asio::ip::tcp::socket socket; }; struct Acceptor { //      explicit Acceptor(int port); //      void accept(Socket& socket); private: boost::asio::ip::tcp::acceptor acceptor; }; 

 boost::asio::io_service& service() { return single<boost::asio::io_service>(); } Socket::Socket() : socket(service()) { } Socket::Socket(Socket&& s) : socket(std::move(s.socket)) { } void Socket::read(Buffer& buffer) { boost::asio::read(socket, boost::asio::buffer(&buffer[0], buffer.size())); } void Socket::readSome(Buffer& buffer) { buffer.resize(socket.read_some(boost::asio::buffer(&buffer[0], buffer.size()))); } bool hasEnd(size_t posEnd, const Buffer& b, const Buffer& end) { return posEnd >= end.size() && b.rfind(end, posEnd - end.size()) != std::string::npos; } int Socket::readUntil(Buffer& buffer, const Buffer& until) { size_t offset = 0; while (true) { size_t bytes = socket.read_some(boost::asio::buffer(&buffer[offset], buffer.size() - offset)); offset += bytes; if (hasEnd(offset, buffer, until)) { buffer.resize(offset); return offset; } if (offset == buffer.size()) { LOG("not enough size: " << buffer.size()); buffer.resize(buffer.size() * 2); } } } void Socket::write(const Buffer& buffer) { boost::asio::write(socket, boost::asio::buffer(&buffer[0], buffer.size())); } void Socket::close() { socket.close(); } Acceptor::Acceptor(int port) : acceptor(service(), boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port)) { } void Acceptor::accept(Socket& socket) { acceptor.accept(socket.socket); } 

 #define HTTP_DELIM "\r\n" #define HTTP_DELIM_BODY HTTP_DELIM HTTP_DELIM //   Buffer httpContent(const Buffer& body) { std::ostringstream o; o << "HTTP/1.1 200 Ok" HTTP_DELIM "Content-Type: text/html" HTTP_DELIM "Content-Length: " << body.size() << HTTP_DELIM_BODY << body; return o.str(); } //  8800  ( 80 ?) Acceptor acceptor(8800); LOG("accepting"); while (true) { Socket socket; acceptor.accept(socket); try { LOG("accepted"); Buffer buffer(4000, 0); socket.readUntil(buffer, HTTP_DELIM_BODY); socket.write(httpContent("<h1>Hello sync singlethread!</h1>")); socket.close(); } catch (std::exception& e) { LOG("error: " << e.what()); } } 

सिंक्रोनस मल्टीथ्रेडेड सर्वर

1. एक बार में कई कनेक्शन संसाधित करना संभव नहीं है।
2. क्लाइंट अधिक प्रभावी इंटरैक्शन के लिए कनेक्शन का पुन: उपयोग कर सकता है, और हम हमेशा इसे बंद कर देते हैं।

 typedef std::function<void ()> Handler; void go(Handler handler) { LOG("sync::go"); std::thread([handler] { try { LOG("new thread had been created"); handler(); LOG("thread was ended successfully"); } catch (std::exception& e) { LOG("thread was ended with error: " << e.what()); } }).detach(); } 

 Acceptor acceptor(8800); LOG("accepting"); while (true) { Socket* toAccept = new Socket; acceptor.accept(*toAccept); LOG("accepted"); go([toAccept] { try { Socket socket = std::move(*toAccept); delete toAccept; Buffer buffer; while (true) { buffer.resize(4000); socket.readUntil(buffer, HTTP_DELIM_BODY); socket.write(httpContent("<h1>Hello sync multithread!</h1>")); } } catch (std::exception& e) { LOG("error: " << e.what()); } }); } 

अतुल्यकालिक सर्वर

 typedef boost::system::error_code Error; typedef std::function<void(const Error&)> IoHandler; struct Acceptor; struct Socket { friend struct Acceptor; Socket(); Socket(Socket&&); void read(Buffer&, IoHandler); void readSome(Buffer&, IoHandler); void readUntil(Buffer&, Buffer until, IoHandler); void write(const Buffer&, IoHandler); void close(); private: boost::asio::ip::tcp::socket socket; }; struct Acceptor { explicit Acceptor(int port); void accept(Socket&, IoHandler); private: boost::asio::ip::tcp::acceptor acceptor; }; 

1. त्रुटि से निपटने में अब काफी अंतर है। एक तुल्यकालिक दृष्टिकोण के मामले में, हमारे पास 2 विकल्प हैं: एक त्रुटि कोड वापस करना या एक अपवाद फेंकना (यह लेख की शुरुआत में उपयोग की जाने वाली विधि है)। एक अतुल्यकालिक कॉल के मामले में, एक ही तरीका है: हैंडलर के माध्यम से त्रुटि को पारित करना। यानी परिणाम के माध्यम से भी नहीं, लेकिन हैंडलर के लिए एक इनपुट पैरामीटर के रूप में। और अगर आप चाहते हैं, यदि आप नहीं चाहते हैं, तो अच्छे पुराने दिनों की तरह त्रुटियों को संभालने के लिए पर्याप्त रहें, जब कोई अपवाद नहीं थे: हर एक के लिए एक चेक था। लेकिन सबसे दिलचस्प, ज़ाहिर है, यह नहीं है; दिलचस्प यह है कि जब हैंडलर में कोई त्रुटि हुई और उसे संसाधित करने की आवश्यकता है। संदर्भ को याद करना अतुल्यकालिक प्रोग्रामिंग का एक पसंदीदा कार्य है!
2. एक सुसंगत दृष्टिकोण के लिए, मैंने IoHandler उपयोग किया, जो कोड को सरल और अधिक सार्वभौमिक बनाता है।

 Socket::Socket() : socket(service()) { } Socket::Socket(Socket&& s) : socket(std::move(s.socket)) { } void Socket::read(Buffer& buffer, IoHandler handler) { boost::asio::async_read(socket, boost::asio::buffer(&buffer[0], buffer.size()), [&buffer, handler](const Error& error, std::size_t) { handler(error); }); } void Socket::readSome(Buffer& buffer, IoHandler handler) { socket.async_read_some(boost::asio::buffer(&buffer[0], buffer.size()), [&buffer, handler](const Error& error, std::size_t bytes) { buffer.resize(bytes); handler(error); }); } bool hasEnd(size_t posEnd, const Buffer& b, const Buffer& end) { return posEnd >= end.size() && b.rfind(end, posEnd - end.size()) != std::string::npos; } void Socket::readUntil(Buffer& buffer, Buffer until, IoHandler handler) { VERIFY(buffer.size() >= until.size(), "Buffer size is smaller than expected"); struct UntilHandler { UntilHandler(Socket& socket_, Buffer& buffer_, Buffer until_, IoHandler handler_) : offset(0), socket(socket_), buffer(buffer_), until(std::move(until_)), handler(std::move(handler_)) { } void read() { LOG("read at offset: " << offset); socket.socket.async_read_some(boost::asio::buffer(&buffer[offset], buffer.size() - offset), *this); } void complete(const Error& error) { handler(error); } void operator()(const Error& error, std::size_t bytes) { if (!!error) { return complete(error); } offset += bytes; VERIFY(offset <= buffer.size(), "Offset outside buffer size"); LOG("buffer: '" << buffer.substr(0, offset) << "'"); if (hasEnd(offset, buffer, until)) { // found end buffer.resize(offset); return complete(error); } if (offset == buffer.size()) { LOG("not enough size: " << buffer.size()); buffer.resize(buffer.size() * 2); } read(); } private: size_t offset; Socket& socket; Buffer& buffer; Buffer until; IoHandler handler; }; UntilHandler(*this, buffer, std::move(until), std::move(handler)).read(); } void Socket::write(const Buffer& buffer, IoHandler handler) { boost::asio::async_write(socket, boost::asio::buffer(&buffer[0], buffer.size()), [&buffer, handler](const Error& error, std::size_t) { handler(error); }); } void Socket::close() { socket.close(); } Acceptor::Acceptor(int port) : acceptor(service(), boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port)) { } void Acceptor::accept(Socket& socket, IoHandler handler) { acceptor.async_accept(socket.socket, handler); } 

 void go(Handler); void dispatch(int threadCount = 0); 

 void go(Handler handler) { LOG("async::go"); service().post(std::move(handler)); } void run() { service().run(); } void dispatch(int threadCount) { int threads = threadCount > 0 ? threadCount : int(std::thread::hardware_concurrency()); RLOG("Threads: " << threads); for (int i = 1; i < threads; ++ i) sync::go(run); run(); } 

 Acceptor acceptor(8800); LOG("accepting"); Handler accepting = [&acceptor, &accepting] { struct Connection { Buffer buffer; Socket socket; void handling() { buffer.resize(4000); socket.readUntil(buffer, HTTP_DELIM_BODY, [this](const Error& error) { if (!!error) { LOG("error on reading: " << error.message()); delete this; return; } LOG("read"); buffer = httpContent("<h1>Hello async!</h1>"); socket.write(buffer, [this](const Error& error) { if (!!error) { LOG("error on writing: " << error.message()); delete this; return; } LOG("written"); handling(); }); }); } }; Connection* conn = new Connection; acceptor.accept(conn->socket, [conn, &accepting](const Error& error) { if (!!error) { LOG("error on accepting: " << error.message()); delete conn; return; } LOG("accepted"); conn->handling(); accepting(); }); }; accepting(); dispatch(); 

1. एक निरपेक्ष प्लस (और यह, वास्तव में, यह सब पीड़ा के लिए है) - उत्पादकता है। इसके अलावा, यह सिर्फ कई गुना अधिक नहीं है, यह परिमाण के आदेशों से अधिक है!
2. खैर, अब अंत। ठीक एक माइनस है - जटिल और भ्रामक कोड, जिसे डिबग करना भी मुश्किल है।

coroutine

Coroutine कार्यान्वयन

 //    void yield(); //  ,      bool isInsideCoro(); //  struct Coro { //   ,   friend void yield(); Coro(); //     Coro(Handler); //   ~Coro(); //   void start(Handler); //   (    yield) void resume(); //  ,     bool isStarted() const; private: ... }; 

 void coro() { std::cout << '2'; yield(); std::cout << '4'; } 

 std::cout << '1'; Coro c(coro); std::cout << '3'; c.resume(); std::cout << '5'; 

 12345

 void Coro::start(Handler handler) { VERIFY(!isStarted(), "Trying to start already started coro"); context = boost::context::make_fcontext(&stack.back(), stack.size(), &starterWrapper0); jump0(reinterpret_cast<intptr_t>(&handler)); } 

 TLS Coro* t_coro; void Coro::starterWrapper0(intptr_t p) { t_coro->starter0(p); } 

 void Coro::jump0(intptr_t p) { Coro* old = this; std::swap(old, t_coro); running = true; boost::context::jump_fcontext(&savedContext, context, p); running = false; std::swap(old, t_coro); if (exc != std::exception_ptr()) std::rethrow_exception(exc); } 

 void Coro::starter0(intptr_t p) { started = true; try { Handler handler = std::move(*reinterpret_cast<Handler*>(p)); handler(); } catch (...) { exc = std::current_exception(); } started = false; yield0(); } 

 //     void yield() { VERIFY(isInsideCoro(), "yield() outside coro"); t_coro->yield0(); } //  ,      bool isInsideCoro() { return t_coro != nullptr; } //    yield void Coro::resume() { VERIFY(started, "Cannot resume: not started"); VERIFY(!running, "Cannot resume: in running state"); jump0(); } //  ,      bool Coro::isStarted() const { return started || running; } //     void Coro::yield0() { boost::context::jump_fcontext(context, &savedContext, 0); } 

Synca: async इसके विपरीत

कार्यान्वयन

 void go(Handler handler) { LOG("synca::go"); async::go([handler] { coro::Coro* coro = new coro::Coro(std::move(handler)); onCoroComplete(coro); }); } 

 typedef std::function<void(coro::Coro*)> CoroHandler; TLS CoroHandler* t_deferHandler; void onCoroComplete(coro::Coro* coro) { VERIFY(!coro::isInsideCoro(), "Complete inside coro"); VERIFY(coro->isStarted() == (t_deferHandler != nullptr), "Unexpected condition in defer/started state"); if (t_deferHandler != nullptr) { LOG("invoking defer handler"); (*t_deferHandler)(coro); t_deferHandler = nullptr; LOG("completed defer handler"); } else { LOG("nothing to do, deleting coro"); delete coro; } } 

 TLS const Error* t_error; void handleError() { if (t_error) throw boost::system::system_error(*t_error, "synca"); } void defer(CoroHandler handler) { VERIFY(coro::isInsideCoro(), "defer() outside coro"); VERIFY(t_deferHandler == nullptr, "There is unexecuted defer handler"); t_deferHandler = &handler; coro::yield(); handleError(); } 

 void onComplete(coro::Coro* coro, const Error& error) { LOG("async completed, coro: " << coro << ", error: " << error.message()); VERIFY(coro != nullptr, "Coro is null"); VERIFY(!coro::isInsideCoro(), "Completion inside coro"); t_error = error ? &error : nullptr; coro->resume(); LOG("after resume"); onCoroComplete(coro); } async::IoHandler onCompleteHandler(coro::Coro* coro) { return [coro](const Error& error) { onComplete(coro, error); }; } void Acceptor::accept(Socket& socket) { VERIFY(coro::isInsideCoro(), "accept must be called inside coro"); defer([this, &socket](coro::Coro* coro) { VERIFY(!coro::isInsideCoro(), "accept completion must be called outside coro"); acceptor.accept(socket.socket, onCompleteHandler(coro)); LOG("accept scheduled"); }); } 

 void Socket::readSome(Buffer& buffer) { VERIFY(coro::isInsideCoro(), "readSome must be called inside coro"); defer([this, &buffer](coro::Coro* coro) { VERIFY(!coro::isInsideCoro(), "readSome completion must be called outside coro"); socket.readSome(buffer, onCompleteHandler(coro)); LOG("readSome scheduled"); }); } void Socket::readUntil(Buffer& buffer, Buffer until) { VERIFY(coro::isInsideCoro(), "readUntil must be called inside coro"); defer([this, &buffer, until](coro::Coro* coro) { VERIFY(!coro::isInsideCoro(), "readUntil completion must be called outside coro"); socket.readUntil(buffer, std::move(until), onCompleteHandler(coro)); LOG("readUntil scheduled"); }); } void Socket::write(const Buffer& buffer) { VERIFY(coro::isInsideCoro(), "write must be called inside coro"); defer([this, &buffer](coro::Coro* coro) { VERIFY(!coro::isInsideCoro(), "write completion must be called outside coro"); socket.write(buffer, onCompleteHandler(coro)); LOG("write scheduled"); }); } 

के उपयोग

 Acceptor acceptor(8800); LOG("accepting"); go([&acceptor] { while (true) { Socket* toAccept = new Socket; acceptor.accept(*toAccept); LOG("accepted"); go([toAccept] { try { Socket socket = std::move(*toAccept); delete toAccept; Buffer buffer; while (true) { buffer.resize(4000); socket.readUntil(buffer, HTTP_DELIM_BODY); socket.write(httpContent("<h1>Hello synca!</h1>")); } } catch (std::exception& e) { LOG("error: " << e.what()); } }); } }); dispatch(); 

 Acceptor acceptor(8800); LOG("accepting"); while (true) { Socket* toAccept = new Socket; acceptor.accept(*toAccept); LOG("accepted"); go([toAccept] { try { Socket socket = std::move(*toAccept); delete toAccept; Buffer buffer; while (true) { buffer.resize(4000); socket.readUntil(buffer, HTTP_DELIM_BODY); socket.write(httpContent( "<h1>Hello sync multithread!</h1>")); } } catch (std::exception& e) { LOG("error: " << e.what()); } }); } 

 Acceptor acceptor(8800); LOG("accepting"); go([&acceptor] { while (true) { Socket* toAccept = new Socket; acceptor.accept(*toAccept); LOG("accepted"); go([toAccept] { try { Socket socket = std::move(*toAccept); delete toAccept; Buffer buffer; while (true) { buffer.resize(4000); socket.readUntil(buffer, HTTP_DELIM_BODY); socket.write(httpContent( "<h1>Hello synca!</h1>")); } } catch (std::exception& e) { LOG("error: " << e.what()); } }); } }); dispatch(); 

सुधार

 async::IoHandler onCompleteGoHandler(coro::Coro* coro, Handler handler) { return [coro, handler](const Error& error) { if (!error) go(std::move(handler)); onComplete(coro, error); }; } struct Acceptor { typedef std::function<void(Socket&)> Handler; // ... }; void Acceptor::goAccept(Handler handler) { VERIFY(coro::isInsideCoro(), "goAccept must be called inside coro"); defer([this, handler](coro::Coro* coro) { VERIFY(!coro::isInsideCoro(), "goAccept completion must be called outside coro"); Socket* socket = new Socket; acceptor.accept(socket->socket, onCompleteGoHandler(coro, [socket, handler] { Socket s = std::move(*socket); delete socket; handler(s); })); LOG("accept scheduled"); }); } 

 Acceptor acceptor(8800); LOG("accepting"); go([&acceptor] { while (true) { acceptor.goAccept([](Socket& socket) { try { Buffer buffer; while (true) { buffer.resize(4000); socket.readUntil(buffer, HTTP_DELIM_BODY); socket.write(httpContent("<h1>Hello synca!</h1>")); } } catch (std::exception& e) { LOG("error: " << e.what()); } }); } }); dispatch(); 

प्रश्न 1. और प्रदर्शन के बारे में क्या?

1. सर्वर 30K RPS के निरंतर लोड के तहत चलता है (यानी प्रति सेकंड 30 हजार अनुरोध)।
2. हम asyncऔर के मामले में प्रोसेसर लोड को देखते हैं synca।

प्रश्न 2. अच्छा, चलिए बताते हैं। क्या इस तरह से अधिक जटिल अतुल्यकालिक कार्यों को हल करना संभव है?

प्रमेय। किसी भी अतुल्यकालिक कार्य को कोरटाइन का उपयोग करके हल किया जा सकता है।

 //    async(..., handler); //    

 //    async(..., handler); 

 //    synca(...); handler(); 

 //    go { async(..., handler); } //    

 //    go { synca(...); handler(); } //    

निष्कर्ष