当前位置：首页 > news >正文

2507C++,C++协程的发送者

news 2025/7/26 11:50:25

原文

为什么C++有协程时需要发送者?
这一切都太复杂了!

所以把它具体化.这里展示如何把一个生硬的旧C风格的异步API引入发送者的世界,及你可能想要这样的原因.

`C(经典)`异步`API`

过去,我做了很多Win32编程.Win32有多种异步模型待选,但最简单的是好的回调.异步IOAPI(如ReadFileEx)的形状如下:

    //带回调的旧式异步`CAPI/`(如`Win32`的`ReadFileEx)`
struct overlapped {//...操作系统`内部结构`在此...
};
using overlapped_callback =void(int status, int bytes, overlapped* user);
int read_file(FILE*, char* buffer, int bytes,overlapped* user, overlapped_callback* cb);

协议非常简单:传入一般的参数加上两个额外的参数来调用read_file:"覆盖"结构的指针和回调.操作系统将使用覆盖的结构,但用户也可在其中填充数据,这样回调可稍后使用它.像这样:

struct my_overlapped : overlapped {//...额外`数据`在此......
};
void my_callback(int status, int bytes, overlapped* data) {auto* my_data = static_cast<my_overlapped*>(data);//...使用放入`"my_overlapped"`对象中的额外内容......delete my_data;//清理
}
void enqueue_read(FILE* pfile) {//分配和初化`my_data...`auto* my_data =new my_overlapped{{}, /*`数据`在此*/ };int status =read_file(pfile, buff, bytes, my_data, my_callback);//...
}

原理:read_file导致操作系统排队IO操作,并保存覆盖(overlapped)和overlapped_callback指针.IO完成后,OS传入覆盖结构指针并调用回调.

我已写过数百次此代码了.你可能也有.

这很简单.它有效.为什么要让它更复杂,对吧?

`C(混乱)`异步`API`

回调API没有任何问题.问题是,每个公开异步的库都使用略有不同的回调API.如果想链接两个不同库的两个异步操作,需要编写一堆胶水代码来按另一个异步抽象映射该异步抽象.
这是巴别塔问题.

解决方法是让C++标准认可一个异步抽象.然后,所有公开异步的库都可按标准抽象映射它们的抽象,这样都可相互通信.

这就是C++的标准化委员会对它感兴趣的原因.实际上,只能通过标准来解决的问题.

`C(组合)`异步`API`

因此引入发送者.它们很高效,结构化,且是组合的!,这里展示一些代码,让代码说话.
如果查看上面的read_file,API,可识别出一些不同部分:
1,分配异步操作期望的任何资源.
2,操作时,数据必须在稳定的地址(即覆盖结构),
3,初化将异步IO排入队列的异步操作,及处理异常,
4,异步操作完成后执行用户提供的延续(即回调).
5,回收第1步中分配的任何资源.

发送者也有所有这些部件,但形状统一,因此可通用地使用它们.事实上,发送者和C风格API之间唯一区别是,发送者中不是一个回调,而是三个:成功,失败和取消各一个.

第1步:分配

重新构想的read_fileAPI将如下:

read_file_senderasync_read_file(FILE* file, char* buffer, int size){return {file, buffer, size};}

工作是将参数放入发送器形状的对象中,该对象如下:

namespace stdex = std::execution;
struct read_file_sender
{using sender_concept = stdex::sender_t;//`(1)`using completion_signatures =//`(2)`stdex::completion_signatures<stdex::set_value_t( int, char* ),stdex::set_error_t( int ) >;auto connect( stdex::receiver auto rcvr )//`(3)`{return read_file_operation{{}, {}, pfile, buffer, size, std::move(rcvr)};}FILE* pfile;char* buffer;int size;
};

发送者的工作是描述异步操作.(它也是操作状态的工厂,但这是第2步.在标有"(1)"的行上,按发送者声明此类型.在标有"(2)"的行上,声明了完成此异步操作的方式.
使用函数类型列表来完成:

stdex::set_value_t( int, char* )

…声明此异步操作可通过将整和char*传递给值回调来成功完成.记住,有三个回调.还有该:

stdex::set_error_t( int )

…声明此异步操作可能会通过向错误回调传递整来在错误中完成.如果此异步操作是可取消的,它将使用stdex::set_stopped_t()声明.

第2步:`数据`

在上面标有"(3)"的行上,连接,成员函数接受"接收者"并返回"操作状态".接收者是三个回调的合并:值,错误和停止(大致是已取消).

连接发送者和接收者的结果是操作状态.操作状态,与CAPI中的覆盖结构一样,是异步操作的数据.它必须保存在持续时间内的稳定的地址.

连接函数返回一个read_file_operation对象.连接的调用者负责确保保活此对象,且在执行其中一个回调前不会移动.
read_file_operation类型如下:

struct immovable {immovable() = default;immovable(immovable&&) = delete;
};
template <class Receiver>
struct read_file_operation : overlapped, immovable//`(1)`
{static void _callback(int status, int bytes,//`(2)`overlapped* data){auto* op =static_cast<read_file_operation*>(data);//`(3)`if (status == OK)stdex::set_value(std::move(op->rcvr),//`(4)`bytes, op->buffer);elsestdex::set_error(std::move(op->rcvr), status);}void start() noexcept//`(5)`{int status =read_file(pfile, buffer, size, this, &_callback);if (status != OK)stdex::set_error(std::move(rcvr), status);}FILE* pfile;char* buffer;int size;Receiver rcvr;
};

操作状态存储初化异步操作的期望参数及接收者(三个回调).按行分析一下.

1,"(1)":操作状态从覆盖继承,因此可把它的指针传递进read_file.它还从不可移动(immovable)的结构继承.虽然不是绝对必要的,但这可确保不会意外移动操作状态.

2,"(2)":按类静态函数定义传递给read_file的overlapped_callback.

3,"(3)":在回调中,将覆盖的指针下转至指向read_file_operation对象的指针.

4,"(4)":在回调中,检查状态,看看是否成功完成操作,并适当调用接收者的set_value或set_error.
5,"(5)":在必须要有所有操作状态的start()函数中,实际初化读操作.如果初化失败,因为不会执行回调,会立即将错误传递给接收者.

第3步:初化

注意,当调用发送者版本的async_read_file函数时,只是在构造一个发送者.并未开始实际工作.
然后用接收者调用连接并返回操作状态,但仍没有开始任何工作.

只是刚刚排好队,确保一切都在稳定的地址,这样可开始工作.在操作状态调用.start()之前,不会初化工作.只有这样,才会调用C风格的read_fileAPI,从而排队IO操作.

一旦开始构建发送者的管道和任务图,所有这些就很重要.把初化工作与构建操作状态分开,可按一个包含整个任务图期望的所有数据的状态,聚集大量操作状态,从而在开始任何工作前将所有内容旋转到位.

即可仅使用单个动态分配或有时不分配,就启动大量有复杂依赖的异步工作.

必须说明,当我说连接的调用者直到执行其中一个回调,要在稳定地址保持操作状态时,我有点胡言乱语.

只有在调用.start()后,这才会变成现实.只要尚未调用.start(),将发送者连接到接收者,然后在地板上删除操作状态是完全可接受的.
但是调用.start()后,你就承诺了..start()发射火箭.没有退路了.

好的,已构造了操作状态,并在其上调用了.start().

第4步:延续

操作系统发挥其IO神奇.时间流逝.完成IO后,它将用状态码,覆盖结构指针(read_file_operation),调用_callback函数,如果成功,读取了字节数.

_callback将完成信息传递给连接到发送者的接收者,圆圈完成.

但是等等,"第5步:释放"呢从一开始就没有真正分配过任何东西!连接函数按值返回操作状态.由连接的调用者,无论是谁,来保活它.

可通过在堆上放置它来完成,此时,他们负责清理它.或,如果此异步操作是任务图的一部分,他们可通过在更大的状态中聚集操作状态来完成.

第6步:获利!

此时,你会问,这一切的意义何在.发送者和接收者,麻烦的生命期期望操作状态,连接,初化,三个不同回调,谁想要管理所有这些?CAPI要简单得多.是真!
则,为什么我对这一切如此激动?
async_read_file的调用者不需要关心这些.

终端用户,即async_read_file的调用者,不关心接收者和操作状态.他们只是在协程中等待发送者.如下代码使用stdexec库中的协程任务类型.

exec::task< std::string > process_file( FILE* pfile )
{std::string str;str.resize( MAX_BUFFER );auto [bytes, buff] =co_await async_read_file(pfile, str.data(), str.size());str.resize( bytes );co_return str;
}