在开发高负载的多线程或分布式应用程序领域，经常会出现有关异步编程的讨论。 今天，我们将详细探讨异步，并研究异步的发生时间，它如何影响我们使用的代码和编程语言。 我们将弄清为什么需要期货和承诺，并介绍了协程和操作系统。 这将使在软件开发过程中出现的权衡更加明确。
该材料基于Yandex数据分析学校的老师Ivan Puzyrevsky的报告的笔录。
录像录像
1.内容
2.简介
大家好，我叫Ivan Puzyrevsky，我为Yandex工作。 在过去的六年中，我一直从事数据存储和处理的基础架构，现在我转向产品-搜索旅行，酒店和门票。 自从我在基础架构中工作了很长时间以来，我在如何编写不同的已加载应用程序方面积累了很多经验。 我们的基础架构每天不停地在数千台机器上连续运行24*7*365 。 自然，您需要编写代码，以使其可靠，高效地工作并解决公司提出的任务。
今天我们将讨论异步。 什么是异步？ 这是某物与某物在时间上的错配。 根据这种描述，通常不清楚我今天要谈论什么。 为了以某种方式澄清问题，我需要一个示例，例如“ Hello，world！”。 异步通常发生在编写网络应用程序的上下文中，因此我将得到一个类似“ Hello，world！”的网络。 这是一个乒乓球应用程序。 代码如下：
socket s; string x; x = read_from_socket(s, 4); if (x == "ping") { write_to_socket(s, "pong"); } return;
我创建了一个套接字，从那里读取一条线，然后检查它是否可以ping，然后在响应中写入pong。 非常简单明了。 当您在计算机屏幕上看到此类代码时会发生什么？ 我们将此代码视为以下步骤的序列：
从实际物理时间的角度来看，一切都有些偏颇。
那些实际编写并运行了这样的代码的人知道，在读取步骤之后和步骤之后
从我们的代码的角度来看，程序似乎什么也不做，但是在机械操作的内部（我们称为“输入-输出”），写是一个相当明显的时间间隔。
在I / O期间，数据包通过网络和所有随之而来的繁重的低级工作进行交换。 让我们进行一个思想实验：采用这样一个程序，在一个物理处理器上运行它，并假装我们没有任何操作系统，将会发生什么？ 处理器无法停止，它会继续执行措施而没有遵循任何说明，只是白白浪费了能量。
问题是我们在这段时间内是否能做些有用的事情。 这是一个很自然的问题，答案是我们可以节省处理器功率，并将其用于有用的东西，而我们的应用程序似乎无能为力。
3.基本概念
3.1。 执行线程
我们如何完成这项任务？ 让我们调和概念。 我将说“执行流程”，指的是一些有意义的基本操作或步骤序列。 有意义的程度将取决于我所说的执行流程的上下文。 也就是说，如果我们正在谈论单线程算法（Aho-Korasik，图形搜索），那么该算法本身已经是执行线程。 他采取了一些措施来解决该问题。
如果我在谈论数据库，那么执行线程可能是数据库为服务一个传入请求而执行的操作的一部分。 Web服务器也是如此。 如果我正在编写某种移动或Web应用程序，则可以为一个用户的操作提供服务，例如，单击按钮，网络交互，与本地存储的交互等。 从我的移动应用程序的角度来看，这些动作的顺序也将是一个单独的有意义的执行流程。 从操作系统的角度来看，进程或进程线程也是有意义的执行线程。
3.2。 多任务和并发
生产力的基石是执行这种技巧的能力：当我有一个执行线程的物理时间扫描中包含空白时，然后用一些有用的东西填充这些空白-遵循其他执行线程的步骤。
数据库通常同时服务于许多客户端。 如果我们可以在一个更高级别的执行线程的框架内将多个执行线程上的工作结合起来，那么这称为多任务处理。 也就是说，多任务是指我在一个较大的执行流框架内执行从属于较小任务的解决方案的动作时。
重要的是不要将多任务处理的概念与并行性混淆。 并发-
这些是运行时环境的属性，它使您可以在一个时间段或一个步骤中在不同的执行线程中取得进展。 如果我有两个物理处理器，则它们可以在一个时钟周期内执行两条指令。 如果程序在一个处理器上运行，则将需要两个时钟周期来执行相同的两条指令。
重要的是不要混淆这些概念，因为它们属于不同的类别。 多任务处理是程序的一项功能，它在内部结构为可处理不同任务的变量。 并发是运行时环境的属性，使您可以在一个时钟周期内处理多个任务。
在许多方面，异步代码和编写异步代码就是编写多任务代码。 主要的困难是我如何编码任务以及如何管理它们。 因此，今天我们将讨论这一点-编写多任务代码。
4.阻止和等待
让我们从一个简单的例子开始。 返回乒乓球：
socket s; string x; x = read_from_socket(s, 4); if (x == "ping") { write_to_socket(s, "pong"); } return;
正如我们已经讨论的那样，在执行读取线程和白线之后，执行线程进入睡眠状态，该线程被阻塞。 通常我们说“流量被阻止”。
socket s; string x; x = read_from_socket(s, 4); if (x == "ping") { write_to_socket(s, "pong"); } return;
这意味着执行流程已达到必须进行任何事件才能继续执行的地步。 特别是在我们的网络应用程序中，数据必须通过网络到达，或者相反，我们有一个空闲缓冲区，用于将数据写入网络。 事件可能有所不同。 如果我们在谈论时间方面，可以等待计时器触发或其他过程完成。 这里的事件是一种抽象的事物，对于它们而言，重要的是要理解它们是可以预期的。
当我们编写简单的代码时，我们隐式地将事件期望的控制权赋予更高的级别。 在我们的例子中，是操作系统。 作为较高级别的实体，她负责选择下一步要执行的任务，并且还负责跟踪事件的发生。
我们以开发人员的身份编写的代码是同时针对一项任务进行结构化的。 该示例中的代码片段处理一个连接：它从一个连接读取ping并将pong写入一个连接。
代码很清楚。 您可以阅读并了解它的作用，如何工作，解决了什么问题，具有哪些不变性，等等。 同时，我们在这种模型中管理任务计划的能力很差。 通常，操作系统具有优先级的概念，但是如果您编写了软实时系统，那么您就会知道Linux中可用的工具不足以创建足够合理的实时系统。
此外，操作系统是一件很复杂的事情，将上下文从我们的应用程序切换到内核需要花费几微秒的时间，通过一些简单的计算，我们可以估算出每秒大约20万至10万个上下文切换。 这意味着，如果我们编写一个Web服务器，那么假设处理请求的成本是系统的十倍，那么一秒钟之内，我们就可以处理大约2万个请求。
4.1。 非阻塞等待
如果遇到需要更有效地使用网络的情况，则可以开始在Internet上寻求帮助，并使用select / epoll。 在Internet上写道，如果您想同时服务数千个连接，则需要epoll，因为它是一种很好的机制，依此类推。 您打开文档，看到类似以下内容：
int select(int nfds, fd_set* readfds, fd_set* writefds, fd_set* exceptfds, struct timeval* timeout); void FD_CLR(int fd, fd_set* set); int FD_ISSET(int fd, fd_set* set); void FD_SET(int fd, fd_set* set); void FD_ZERO(fd_set* set); int epoll_ctl(int epfd, int op, int fd, struct epoll_event* event); int epoll_wait(int epfd, struct epoll_event* events, int maxevents, int timeout);
接口所包含的函数要么包含很多您要使用的描述符（对于select而言），要么包含大量传递的事件
跨越应用程序的边界，即您需要处理的操作系统内核（对于epoll）。
还值得补充的是，您可以不选择/ epoll，而是进入libuv之类的库，该库在API中没有任何事件，但将具有许多回调。 库接口将说：“亲爱的朋友，提供一个回调来读取套接字，当数据出现时我将调用它。”
int uv_timer_start(uv_timer_t* handle, uv_timer_cb cb, uint64_t timeout, uint64_t repeat); typedef void (*uv_timer_cb)(uv_timer_t* handle); int uv_read_start(uv_stream_t* stream, uv_alloc_cb alloc_cb, uv_read_cb read_cb); int uv_read_stop(uv_stream_t*); typedef void (*uv_read_cb)(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf); int uv_write(uv_write_t* req, uv_stream_t* handle, const uv_buf_t bufs[], unsigned int nbufs, uv_write_cb cb); typedef void (*uv_write_cb)(uv_write_t* req, int status);
与上一章的同步代码相比，发生了什么变化？ 该代码已变为异步。 这意味着我们将逻辑带入应用程序来确定监视事件的时间点。 明确的select / epoll调用是我们要求操作系统提供有关已发生事件的信息的地方。 我们还将应用程序代码纳入了下一步要处理的任务的选择。
从接口示例中，您可以看到基本上有两种引入多任务处理的机制。 当我们
我们绘制出许多我们正在等待的事件，然后以某种方式对它们做出反应。 通过这种方法，很容易将开销分摊一次
一个事件，因此在有关已发生事件的通信中实现了高吞吐量。 通常，所有网络元素（例如内核与网卡的交互或您与操作系统的交互）都基于轮询机制。
第二种方式是一种“推送”机制，当某个外部实体明显进入时，它会中断执行流程并说：“现在，请处理刚刚到达的事件。” 当外部实体明显侵入您的执行线程并说：“现在，请我们正在处理此事件时，这是一种使用回调，使用unix信号，在处理器级别中断的方法。” 为了减少事件发生与反应之间的延迟，出现了这种方法。
为什么我们的C ++开发人员编写和解决特定的应用程序问题可能想要将事件模型拖到我们的代码中？ 如果我们将许多任务的工作拖放到代码中并进行管理，那么由于缺乏向内核过渡的能力，反之亦然，我们可以更快地工作并每单位时间执行更多有用的操作。
就我们编写的代码而言，这会导致什么？ 以nginx为例，这是非常常见的高性能HTTP服务器。 如果阅读其代码，则它是基于异步模型构建的。 该代码很难阅读。 当您问自己处理单个HTTP请求时究竟发生了什么时，结果表明该代码在代码库的不同角度具有许多片段，这些片段被分成不同的文件。 作为服务整个HTTP请求的一部分，每个片段都会做少量的工作。 例如：
static void ngx_http_request_handler(ngx_event_t *ev) { … if (c->close) { ngx_http_terminate_request(r, 0); return; } if (ev->write) { r->write_event_handler(r); } else { r->read_event_handler(r); } ... } typedef void (*ngx_http_event_handler_pt)(ngx_http_request_t *r); struct ngx_http_request_s { ngx_http_event_handler_pt read_event_handler; }; r->read_event_handler = ngx_http_request_empty_handler; r->read_event_handler = ngx_http_block_reading; r->read_event_handler = ngx_http_test_reading; r->read_event_handler = ngx_http_discarded_request_body_handler; r->read_event_handler = ngx_http_read_client_request_body_handler; r->read_event_handler = ngx_http_upstream_rd_check_broken_connection; r->read_event_handler = ngx_http_upstream_read_request_handler;
有一个请求结构，当套接字发出读或写访问信号时，该结构将转发到事件处理程序。 此外，该处理程序根据请求处理的状态在程序的过程中不断切换。 我们要么读取标题，要么读取请求的正文，或者我们向上游请求数据-通常，存在许多不同的状态。
这样的代码很难阅读，因为它实质上是根据对事件的反应来描述的。 我们处于这样的状态，以某种方式对已经发生的事件做出反应。 没有完整的HTTP请求处理过程的完整描述。
JavaScript中经常使用的另一种选择是，当我们将回调转发到接口调用时，构建基于回调的代码，在接口调用中通常有针对该事件的其他嵌套回调，依此类推。
int LibuvStreamWrap::ReadStart() { return uv_read_start(stream(), [](uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { static_cast<LibuvStreamWrap*>(handle->data)->OnUvAlloc(suggested_size, buf); }, [](uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf) { static_cast<LibuvStreamWrap*>(stream->data)->OnUvRead(nread, buf); }); } for (p=data; p != data + len; p++) { ch = *p; reexecute: switch (CURRENT_STATE()) { case s_start_req_or_res: case s_res_or_resp_H: case s_res_HT: case s_res_HTT: case s_res_HTTP: case s_res_http_major: case s_res_http_dot:
该代码又是非常分散的，无法理解我们如何处理请求的当前状态。 许多信息都是通过闭包传输的，因此您需要花大力气来重构处理单个请求的逻辑。
因此，将多任务处理引入我们的代码（选择工作任务并对其进行多路复用的逻辑），我们获得了有效的代码并控制了任务优先级，但是在可读性方面却损失了很多。 此代码难以阅读且难以维护。
怎么了 假设我有一个简单的案例，例如，我读取了一个文件并通过网络传输了它。 在非阻塞版本中，这种情况将对应于这样的线性状态机：
- 初始状态
- 开始读取文件，
- 等待文件系统的响应，
- 将文件写入套接字，
- 最终状态。
现在，假设我想将数据库中的信息添加到此文件中。 一个简单的选择：
- 初始状态
- 读取文件
- 读取文件
- 从数据库中读取
- 从数据库中读取
- 我用插座工作
- 写到套接字。
看起来像是线性代码，但是状态数增加了。
然后您开始认为并行执行这两个步骤会很好-从文件和数据库读取。 组合运算的奇迹开始了：您处于初始状态，要求从数据库中读取文件和数据。 然后，您可以进入一种状态，即数据库中有数据但没有文件，反之亦然-有文件中的数据但没有数据库的数据。 接下来，您需要进入具有以下两种状态之一的状态。 同样，这是两个状态。 然后，您需要进入同时具有两种成分的状态。 然后将它们写入套接字等等。
应用程序越复杂，状态越多，需要在您的头脑中组合的代码段就越多。 不方便。 或者您正在编写回调面条，这不方便阅读。 如果编写了一个分支系统，那么总有一天您将无法忍受。
5.期货/承诺
要解决该问题，您需要更轻松地了解情况。
有一个程序，它有黑色和红色圆圈。 我们的执行流程是黑眼圈。 有时，当流无法继续工作时，它们会交替显示为红色。 问题在于，对于我们的黑执行线程，您需要进入下一个黑圈，何时不知道该黑圈。
问题在于，当我们用编程语言编写代码时，我们向计算机解释了现在该怎么做。 计算机是相对简单的事物，它期望我们以编程语言编写指令。 她正在等待下一个循环的指令，而在我们的编程语言中，没有足够的钱说：“将来，请在发生某些事情时做点什么。”
在编程语言中，我们以可理解的瞬时动作进行操作：调用函数，算术运算等。 他们描述了具体的下一步。 同时，要处理应用程序逻辑，有必要不描述下一个物理步骤，而是描述下一个逻辑步骤：例如，当来自数据库的数据出现时，我们应该怎么做。
因此，我们需要一些机制来组合这些片段。 在编写同步代码的情况下，我们将问题完全隐藏在后台，并说操作系统会这样做，允许它中断并重新安排线程。
在第1级中，我们打开了这个Pandora的盒子，它为代码带来了很多切换，案例，条件，分支，状态。 我希望做出一些让步，以便使代码相对易读，但保留第1级的所有优点。
对我们来说幸运的是，在1988年，涉及分布式系统的人们Barbara Liskov和Luba Shirir意识到了这个问题，并提出了需要进行语言更改的要求。 必须将结构添加到编程语言中，以表达事件之间的时间关系-在当前时间和未来的不确定时间。
这些被称为承诺。 这个概念很酷，但是已经在架子上收集灰尘二十年了。 — , Twitter, Ruby on Rails Scala, , , , future . Your Server as a Function. , .
Scala, , ++ ?
, Future. T c : , - .
template <class T> class Future <T>
, , , . , «», , . Future «», Promise — «». ; , JavaScript, Promise — , Java – Future.
, . , , boost::future ( std::future) — , .
5.1。 Future & Promise
template <class T> class Future { bool IsSet() const; const T& Get() const; T* TryGet() const; void Subscribe(std::function<void(const T&)> cb); template <class R> Future<R> Then( std::function<R(const T&)> f); template <class R> Future<R> Then( std::function<Future<R>(const T&)> f); }; template <class T> Future<T> MakeFuture(const T& value);
, , - , . , , , . , , — , , . Then, .
template <class T> class Promise { bool IsSet() const; void Set(const T& value); bool TrySet(const T& value); Future<T> ToFuture() const; }; template <class T> Promise<T> NewPromise();
. , . «, , , ».
5.2。
? , . Then — , .
, — future --, - t — . , , , f, - r.
t f. , , r.
: t, , r . :
template <class T> template <class R> Future<R> Future<T>::Then(std::function<R(const T&)> f) { auto promise = NewPromise<R>(); this->Subscribe([promise] (const T& t) { auto r = f(t); promise.Set(r); }); return promise.ToFuture(); }
:
Promise R ,Future<T> t ,- ,
r = f(t) , r Promise ,Promise .
f , R , Future<R> , R . :
template <class T> template <class R> Future<R> Future<T>::Then(std::function<Future<R>(const T&)> f) { auto promise = NewPromise<R>(); this->Subscribe([promise] (const T& t) { auto that = f(t); that.Subscribe([promise] (R r) { promise.Set(r); }); }); return promise.ToFuture(); }
, - t. f, r, . , , .
, Then :
Promise ,Subscribe -,Promise , Future .
, . , , , .
, , , -. , , -, Subscribe. , , , - . , .
5.3。 例子
AsyncComputeValue, GPU, . Then, , (2v+1) 2 .
Future<int> value = AsyncComputeValue();
. , : (2v+1) 2 . , .
, , . .
. : , ; ; .
Future<int> GetDbKey(); Future<string> LoadDbValue(int key); Future<void> SendToMars(string message); Future<void> ExploreOuterSpace() { return GetDbKey()
— ExploreOuterSpace. Then; — — , . ( ) . .
5.4。 Any-
: Future , , . , , :
template <class T> Future<T> Any(Future<T> f1, Future<T> f2) { auto promise = NewPromise<T>(); f1.Subscribe([promise] (const T& t) { promise.TrySet(t); }); f2.Subscribe([promise] (const T& t) { promise.TrySet(t); }); return promise.ToFuture(); }
, Any-, Future : , . , , .
, , , , , . « DB1, DB2, — - ».
5.5。 All-
. , , , ( T1 T2), T1 T2 , , .
template <class T1, class T2> Future<std::tuple<T1, T2>> All(Future<T1> f1, Future<T2> f2) { auto promise = NewPromise<std::tuple<T1, T2>>(); auto result = std::make_shared< std::tuple<T1, T2> >(); auto counter = std::make_shared< std::atomic<int> >(2); f1.Subscribe([promise, result, counter] (const T1& t1) { std::get<0>(*result) = t1; if (--(*counter) == 0) { promise.Set(*result)); } }); f2.Subscribe([promise, result, counter] (const T2& t2) { } return promise.ToFuture(); }
nginx. , , . nginx « », « », « » . All- , . .
5.6.
Future Promises — legacy-, . callback- , , : Future, , callback- Future.
: , Future .
6.
, , . .
Future<Request> GetRequest(); Future<Payload> QueryBackend(Request req); Future<Response> HandlePayload(Payload pld); Future<void> Reply(Request req, Response rsp);
. Request, - . , . , , , . , - .
, , . ? — , request payload, — , .
, Java Netty. , , . , , .
, GetRequest, QueryBackend, HandlePayload Reply , Future.
, , Future T — WaitFor.
Future<Request> GetRequest(); Future<Payload> QueryBackend(Request req); Future<Response> HandlePayload(Payload pld); Future<void> Reply(Request req, Response rsp); template <class T> T WaitFor(Future<T> future);
:
Future<Request> GetRequest(); Future<Payload> QueryBackend(Request req); Future<Response> HandlePayload(Payload pld); Future<void> Reply(Request req, Response rsp); template <class T> T WaitFor(Future<T> future); auto req = WaitFor(GetRequest()); auto pld = WaitFor(QueryBackend(req)); auto rsp = WaitFor(HandlePayload(pld)); WaitFor(Reply(req, rsp));
: Future, . . , . .
. . - 0, , , mutex+cvar future. . , .
6.1。
, . , , , , - , . , - .
— «» , , . . . : boost::asio boost::fiber.
, . 怎么做？
6.2。 WaitFor
, , boost::context, : , ; , . x86/64 , , .
, goto: , , , .
, - . Fiber — . +Future. , , Future, .
class Fiber { MachineContext context_; Future<void> future_; };
class Scheduler { void WaitFor(Future<void> future); void Loop(); MachineContext loop_context_; Fiber* current_fiber_; std::deque<Fiber*> run_queue_; };
Future , , , . : Loop, , , , , .
WaitFor?
thread_local Scheduler* ThisScheduler; template <class T> T WaitFor(Future<T> future) { ThisScheduler->WaitFor(future.As<void>()); return future.Get(); } void Scheduler::WaitFor(Future<void> future) { current_fiber_->future_ = future; SwitchContext(¤t_fiber_->context_, &loop_context_); }
: , - , , Future void, . .
Future<void> , , - .
WaitFor : : « Fiber Future», ( ) .
, :
ThisScheduler->WaitFor return future.Get() , .
? , Future, .
6.3。
- , , , - , . SwitchContext , 2 — .
void Scheduler::Loop() { while (true) {
? , , , Future, Future, , , .
void Scheduler::Loop() { while (true) {
, . :
WaitFor — .
Switch- .
Future ( ), , . - Fiber.
WaitFor Future , - , Future . :
Future<Request> GetRequest(); Future<Payload> QueryBackend(Request req); Future<Response> HandlePayload(Payload pld); Future<void> Reply(Request req, Response rsp); template <class T> T WaitFor(Future<T> future); auto req = WaitFor( GetRequest()); auto pld = WaitFor( QueryBackend(req)); auto rsp = WaitFor( HandlePayload(pld)); WaitFor( Reply(req, rsp));
, , , . , , .
6.4。 Coroutine TS
? — . Coroutine TS, , WaitFor CoroutineWait, CoroutineTS — - . , - . , Waiter Co, , .
7. ?
. , , , . , , , .
— . , . . , . , , , , .
- , , . , . , , .
, ? , .
. , , , , . . , , , , .
nginx, , , , , . , , , future promises.
, , , , , , , .
futures, promises actors. . , .
: , , , . , , , , . ? , .
分钟的广告。 19-20 C++ Russia 2019. , , Grimm Rainer «Concurrency and parallelism in C++17 and C++20/23» , C++ . , . , , - .