2308C++简单异步懒

Lazy

Lazy由C++20无栈协程实现.一个Lazy闭包一个懒求值的计算任务.

使用Lazy

想用Lazy,需要先#inlude<async_simple/coro/Lazy.h>,再实现返回类型为Lazy<T>的协程函数即可.如:

# 包含<简单异步/协程/懒.h>懒<整>任务1(整 x){协中 x;//带有`协中`的函数是协程函数.
}

在Lazy中也可协待其他可等待对象:

# 包含<简单异步/协程/懒.h>
懒<整>任务2(整 x){协待 标::总是挂起{};协中 x;
}

启动方式

一个Lazy应该以协待,同步等待及.start(回调)方式启动.

协待启动

如:

# 包含<简单异步/协程/懒.h>
懒<整>任务1(整 x){协中 x;
}
懒<>任务2(){动 t=任务1(10);动 x=协待 t;//开始执行t协程断定(x==10);
}

协待的Lazy时会触发对称变换
,它会挂起当前协程并立即执行协待的Lazy.执行完协待的Lazy后,会使用对称变换来唤醒当前挂起的协程.
并按协待式的值返回Lazy<T>中所包含的值.

需要注意,协待一个Lazy时并不能假设一定会执行协待后的语句.原因如:
1,未完成等待的任务.
2,分发器实现有bug,最后不一定会执行了提交到分发器的任务.
3,执行等待任务过程中出现了异常,此时协待会直接返回该异常到当前正在等待的Lazy,而不会执行之后的语句了.

同时需要注意,使用协待启动Lazy需要当前函数也为C++20无栈协程.

.start(回调)启动

如:

# 包含<简单异步/协程/懒.h>
# 包含<io流>
懒<整>任务1(整 x){协中 x;
}
懒<>任务2(){动 t=任务1(10);动 x=协待 t;断定(x==10);
}
空 函数(){任务2().开始([](试<空>结果){如(结果.有错误())标::输出<<"任务2错误.\n";异标::输出<<"任务2成功.\n";});
}

Lazy<T>::start(回调)中的回调需要是一个接受Try<T>类型参数的callable对象.

调用Lazy<T>::start(回调)方法后,会立即开始执行Lazy,并在Lazy执行完成后,把Lazy的结果传入回调中执行.
在设计上,start是非阻塞异步调用接口.语义上,用户可认为调用start后立即返回.用户不应假设调用start后会何时返回.这是由Lazy的执行情况决定的.

对不需要回调的情况,用户可写:

任务().开始([](动&&){});

同步等待启动

如:

# 包含<简单异步/协程/懒.h>
懒<整>任务1(整 x){协中 x;
}
懒<>任务2(){动 t=任务1(10);动 x=协待 t;断定(x==10);
}
空 函数(){动 值=同步等待(任务2());//同步等待,执行完任务2
}

在启动Lazy的同时,同步等待会阻塞当前进程直到执行完同步等待的Lazy.同步等待属于同步阻塞.

取值与异常处理

对Lazy<T>类型的任务对象,协待任务的返回类型为T.如果执行任务过程中有异常,则协待任务会直接抛该异常.

如:

懒<整>福(){抛 标::运行时错误("测试");协中 1;
}
懒<整>条(){整 资源;试{资源=协待 福();}抓(...){标::输出<<"福有错.置结果为-1.\n";资源=-1;}协中 资源;
}
空 栏(){动 资源=同步等待(条());标::输出<<"结果:"<<资源<<"\n";//资源 的值会是-1.
}

需要注意,虽然协待任务时有可能会抛异常,但总是使用try...catch包装协待并不是推荐的做法.一方面这样写确实很麻烦,另一方面,当协待抛异常时,协程的框架保证了会由协待所在的协程处理抛的异常.

如:

懒<整>福(){抛 标::运行时错误("测试");协中1;
}
懒<整>条(){整 资源;资源=协待 福();断定(假);//不会执行,因为上一行会抛异常.协中 资源;
}
懒<整>栏(){协中 协待 条();
}
空 正常(){试{同步等待(栏());}抓(...){//可在此抓到异常.}
}
空 正常2(){栏().开始([](试<整>结果){如(结果.有错误())标::输出<<"栏有错误!!\n";});
}

当由协待组成的任务链的某一层出现异常时,异常会一层层地往上抛.对同步等待形式的启动方式,可在同步等待处加上try..catch语句.而对实际中更常用的.start方式,可在回调处通过Try对象取是否出现异常.

如果不想让异常直接往上抛,而想直接处理,则可用coAwaitTry接口.如:

懒<整>福(){抛 标::运行时错误("测试");协中 1;
}
懒<整>条(){试<整>资源=协待 福().试协待();如(资源.有错误()){标::异常针 错误=资源.取异常();//计算错误}协中 资源.值();
}

对Lazy<T>类型的对象任务,协待任务.coAwaitTry()的返回值类型为Try<T>.

调度懒

调度懒在语义上是绑定了Executor的Lazy.调度懒在协待 Lazy时并不会使用对称变换立即执行Lazy,而是唤醒Lazy的任务,并提交它到调度懒绑定的Executor中.因为语义上调度懒也是Lazy,所以调度懒也能使用Lazy的协待,.start和同步等待接口.

创建调度懒

不能直接创建调度懒,也不能像Lazy一样作为协程的返回类型.只能通过Lazy的via接口创建调度懒,如:

空 福(){执行器::简单执行器 e1(1);动 加一=[&](整 x)->懒<整>{动 临=协待 取值(x);协中 临+2;};再安排懒 分发=加一().通过(&e1);同步等待(分发);//由分发器决定何时开始`加一`的执行
}

分发器的传递

在编写计算任务时,都只使用Lazy.当想要为一个计算任务指定分发器时,只需要在任务开始阶段指定分发器即可.
指定的分发器会随着协待一路传递下去.
如:

# 包含<简单异步/协程/懒.h>
# 包含<io流>
懒<整>任务1(整 x){协中 协待 计算(x);
}
懒<整>任务2(整 x){协中 协待 任务1(x);
}
懒<整>任务3(整 x){协中 协待 任务2(x);
}
懒<整>任务4(整 x){协中 协待 任务3(x);
}
空 函数(整 x,执行器*e){任务4(x).通过(e).开始([](动&&结果){标::输出<<"完成计算x的任务.\n结果是"<<结果<<"\n";});
}

在上例中,任务1...任务4转发计算任务的调用链,其结果都是以Lazy表示的.
在func中,在创建了任务4的Lazy后,为其指定了分发器e得到了一个调度懒.
之后调用.start时,不仅任务4具体何时执行是由分发器决定的.
包括任务3,任务2,任务1何时执行,也都是由分发器决定的.

总之,当需要为多个Lazy组成的任务链指定分发器时,只需要在任务链的开头指定分发器就好了.

Collect

collectAll

当有多个计算任务时,一个很常见需求是等待所有计算任务完成以取所有任务的结果后再进一步的计算.
可用collectAll接口完成该需求.如:

懒<整>福(){标::向量<懒<整>>输入;输入.压后(计算任务(1));输入.压后(计算任务(2));向量<试<整>>出=协待 收集所有(标::移动(输入));协中 出[0].值()+出[1].值();
}

collectAll表示开始执行参数中所有Lazy的任务.collectAll是一个协程,要用协待以取结果.

参数要求

collectAll接受两类的参数:
1,参数类型:std::vector<Lazy<T>>.返回类型:std::vector<Try<T>>.
2,参数类型:Lazy<T1>,Lazy<T2>,Lazy<T3>,....返回类型:std::tuple<Try<T1>,Try<T2>,Try<T3>,...>.

第二个参数类型的示例为:

懒<整>计算整();
懒<双精>计算双精();
懒<标::串>计算串();
懒<>福(){标::元组<试<整>,试<双精>,试<标::串>>资源=协待 收集所有(计算整(),计算双精(),计算串());试<整>整资源=标::取<0>(资源);如(整资源.有错误())标::输出<<"计算整()时,错误\n";异标::输出<<"计算整结果为:"<<整资源.值()<<"\n";//...
}

其他接口

collectAllPara

如果colletAll的所有参数都是Lazy而非调度懒,则collectAll会单线程地执行每个Lazy.
有两个办法:
1,为所以参数绑定Lazy使其成为调度懒.
2,使用collectAllPara.

使用collectAllPara需要注意其所在协程必须要绑定分发器,否则依然是单线程执行所有Lazy.

如:

懒<整>福(){标::向量<懒<整>>输入;输入.压后(计算任务(1));输入.压后(计算任务(2));向量<试<整>>出=协待 收集所有段(标::移动(输入));协中 出[0].值()+出[1].值();
}
空 条(){//动 t=同步等待(福());//错误!福()没有绑定分发器!依然会串行执行.执行器::简单执行器 e1(1);动 t=同步等待(福().通过(&e1));//正确,提前为 福()绑定分发器
}

collectAllPara的参数与返回类型和collectAll相同.

collectAllWindowed

当需要分批次的执行并发任务时,可用collectAllWindowed.
collectAllWindowed的参数列表及语义为为:
1,size_t maxConcurrency.每个批次所规定的任务数.
2,bool yield.在执行完当前批次后,是否需要yield使当前协程退出,把控制权交给其他协程.
3,std::vector<Lazy<T>>lazys.所有需要执行的任务.

如:

懒<整>和(标::向量<试<整>>输入);
懒<整>批量求和(大小型 总数,大小型 批大小){标::向量<懒<整>>输入;对(动 i=0;i<总数;i++)输入.压后(计算任务());动 出=协待 收集所有窗口(批大小,真,标::移动(输入));协中 协待 和(标::移动(出));
}

collectAny

只需要等待所有计算任务中的任意一个任务完成时,可用collectAny来取第一个完成的任务结果.此时会忽略其他尚未完成任务的结果.
collectAny接受的参数为std::vector<Lazy<T>>.返回类型为Lazy<CollectAnyResult<T>>.

CollectAnyResult的数据结构为:

元<型名 T>
构 收集任一结果<空>{大小型 _索引;试<T> _值;
};

其中_idx表示第一个完成的任务的编号,_value表示第一个完成的任务的值.

如:

懒<空>福(){标::向量<懒<整>>输入;输入.压后(计算任务(1));输入.压后(计算任务(2));动 结果=协待 收集任一(标::移动(输入));标::输出<<"第1任务索引为"<<结果._索引<<"\n";如(结果._值.有错误())标::输出<<"失败了.\n";异标::输出<<"结果:"<<结果._值.值()<<"\n";
}