原文

概述

无栈协程是函数支持依赖函数而确认和阻止操作的状态机表示.当与IOCP等高级事件循环配对时,它允许高吞吐量系统处理事件处理,同时对较少知识且有意愿可正确编写异步代码的用户友好.

理由

此处介绍的无栈协程考虑了一组有限的用例:
1,你期望在事件间等待.
2,你需要支持在线程间移动.
3,(可能因为规模)你的内存有限.
4,你的工作不需要挂起线程.
5,目标听众并不完全由经验丰富的C(或同等人员)组成.

如果满足所有这些条件,则适用协程.否则,可能不是.

这些标准既反映了内核的事件循环设计,也反映了对套接字,大规模处理事件的需求.如果试以每秒100k请求为目标,则这是合适的,但如果你的目标是350k或5k,则可能不合适.

或用有栈协程支持低每秒请求计数,在d运行时中有个叫纤程的实现.这与无栈协程相比,有一组非常不同的权衡.

因为实现需要多个页,因此它们的数目可能更加有限(因为保护页的原因,限制为大约32k).它们没有编译器的已知调用栈,因此如果不使用传递属性,就不可能保护线本存储变量和其他此类数据的竞争.

因为缺乏临时保护,它们无法在线程间安全地移动,这使得在不引入额外的成本,即使数量不是问题,它们也不适合最先进的事件循环类型(如IOCP).

先前工作

许多语言对无栈协程的支持程度各不相同.它们一般以异步和等待关键字为标志.使用等待关键字等待条件触发后,再继续的概念可跟踪到1973年的论文并行编程概念.

1968年,Dijkstra发布了协同序列处理,其中引入了并行开始/结束的概念,后来的作者用关键字cobegin和coend来表示该概念.

在C族中,在大括号中标签函数中代码块的开始和结束,对当时的语言,在函数级没有如此清晰的划分,而是按域语句对待它们.

在Rust中,异步关键字以必须可在线程间移动的无栈协程来生成状态机.由它们定义的名叫未来(一个特征)的库类型表示每个协程.

从那里,你显式选择如何等待未来完成.常见的选项是执行器提供的库功能,即等待或toblock_on(阻塞).

执行器是一个提供执行协程的线程池的库概念.它执行用户模式分发,或是系统事件循环的一部分.

当协程语言功能与事件循环交互时,并非所有设计都是相同.Rust使用轮询来确定是否已完成未来.这在基于Posix的系统上运行得非常好,这些系统确实有不依赖每次轮询时,直接上传事件处理器的轮询方法.

但是,窗口未公开此功能.这导致Rust使用内部易中断的未记录的行为,以使他们的等待(await)功能可工作.

该行为可通过D语言中已知的条件变量来实现.

在某些语言中,协程与其标准库密切相关,而对C#,它与运行时的任务类关联.这是协程对象的名义表示.

只要适当地注解,就可取消,并会自动为你创建.不支持多个返回值,等待任务会导致返回值.语言中内置了避免逃逸内存.

或在变量声明上使用与Swift5.5引入的Swift的async let绑定功能一样的变量级隐藏任务.

描述

除了前面提到的使用要求外,还有一些技术要求:
1,语言功能不得要求使用特定的库.
2,它必须在dmd的前端易于实现.
3,较低的成本.
4,如果导致错误,则在它的多线程应用中,保证此错误会出错.
5,必须考虑框架的适用性,以最小化用户需要了解协程正在工作(如果框架作者需要).

状态

为了开始设计,语言会生成一个,在任意给定时间点,描述协程状态的结构,如下:

static struct __generatedName {
    alias ReturnType = ...;
    alias Parameters = ...;
    alias VarTypes = ...;
    //如果`和类型`不在语言中,则可能由自定义标签联提供它
    sumtype ExceptionTypes = ...;
    sumtype WaitingOn = ...;
    //执行的阶段
    int tag;
    //输入
    Parameters parameters;
    
    //如果因为异常而结束,则在此处存储它
    ExceptionTypes exception;
    
    //如果在协程上`产生`,则在此存储
    WaitingOn waitingOnCoroutine;
    
    //如果返回后有一个值,则在此存储
    bool haveValue;
    ReturnType value;
    //在调用函数后,剩余的所有内容
    VarTypes vars;
    
    void execute() @safe nothrow;
}

在协程函数中,无法访问你的状态.由编译器自动处理访问,或按库的一部分在外部完成.

协程无权访问其他环境,见闭包的多个环境,了解会导致的问题.

标签

标签字段,指的是接着要执行的阶段,除了负数,由用户提供的函数定义有问题的值.
标签值有:
1,0..int.max保留给状态机中的阶段.
2,-1保留给已完成的协程.
3,-2保留给错误完成的协程.

这反映了协程的三个阶段之一.它可能可以执行,且可能已按依赖或返回一个值产生到另一个协程.也可能以可选值完成.否则,它可能会遇见错误并无法继续.

如果标签超出阶段边界或规定的负值,则为编译器错误.

构造库表示

如果有如上的状态描述符结构,则你拥有构造实例期望的所有可用信息,执行它直到完成以取结果,并处理遇见的错误.

但仅凭它还不能使用它,你还需要可在一个库中绑定它,来取得库可理解的库类型.
给定库结构的以下潜在壳:

struct InstantiableCoroutine(Return, Args...) {
    static InstiableCoroutine opConstructCo(CoroutineDescriptor : __descriptorco)();
}

然后,可按函数的参数用它:

struct ListenSocket {
    static ListenSocket create(InstantiableCoroutine!(void, Socket) co);
}

由此,可在库类型中传入协程函数的语言描述:

ListenSocket ls = ListenSocket.create((Socket socket) {
    ...
});

这是自动的,你不需要处理函数字面.对自由函数,你还需要额外工作.

上例与如下AST类似工作:

 //只要编译时可访问的仍可用,该结构的位置是不重要的
struct __generatedName {
}
ListenSocket ls = ListenSocket.create(
    InstantiableCoroutine!(__generatedName.ReturnType, __generatedName.Parameters)
        .opConstructCo!__generatedName);
);

也可以给库类型赋值:

InstantiableCoroutine!(int, int) co = (int param) {
    return param;
};

降级为:

 //只要编译时可访问的仍可用,该结构的位置是不重要的
struct __generatedName {
}
InstantiableCoroutine!(int, int) co = InstantiableCoroutine!(int, int)
    .opConstructCo!__generatedName;

自由函数

前面提及,自由函数需要一些额外的工作才能变成协程.它是通过查看给它的用户定义属性来完成的.
如果按@异步标记它,则它是个协程.

int myCo(int param) @async {
    return param;
}

但是,为了方便框架使用,作者可能不希望用户必须输入@异步属性,或更重要的是知道他们在协程中.
为此,将core.attributes中的@isasync属性应用至网络服务框架使用的路由构中:

import core.attributes : isasync;

@isasync
struct Route {
    string path;
}

@Route("/")
void myRoute(Request, Response) {
    //这是个协程!
}

如果框架作者选择不在每个路由上都需要它,则只在特定别名上可用也是公平的:

struct Route {
    string path;
}

@isasync
alias AsyncRoute = Route;

@Route("/")
void myRoute1(Request, Response) {
    //这是一个普通的`自由函数`而不是协程!
}
@AsyncRoute("/")
void myRoute2(Request, Response) {
    //这是个协程!
}

方法

方法有个额外的必需参数,即是参数列表的第一个成员的本指针参数.

struct MyThing {
    void myCo(int param) @async {
    }
}

当它是一个结构而不是一个类时,类型将是指向MyThing的指针,而不是MyThing.因此,上面示例中的参数为:
MyThing*this,int param

可通过内省来应用:

Context theContext;
static foreach(m; __traits(allMembers, Context)) {
    static if (is(__traits(getMember, Context, m) : __descriptorco)) {
        pragma(msg, "Member ", m, " is a coroutine!");
        ListenSocket ls = ListenSocket.create(&__traits(getMember, theContext, m));
    }
}

对注册结构或类中定义的一组路由,这是个有用的功能.与静态无关.

完成

当发生以下三个操作之一时,协程完成:

抛未抓的异常

void myCo() @async {
    throw new Exception("uncaught!");
}

函数结束

void myCo() @async {
    //一些工作
    int var;
    //还有一些工作
    //到达了这里,然后隐式返回,就这样完成了!
}

有不带@异步的返回:

int myCo() @async {
    //一些工作
    return 0;
}

产生

当发生以下三种情况之一时,协程会生成其当前阶段并按下一阶段更改标签:

1,有带@异步的返回:

int myCo() @async {
    @async return 0; //产生
    return 1; //返回
}

等待语句时:

int myCo() @async {
    ACoroutine co = ...;
    await co;
    return 0;
}

等待语句为调用它的协程创建依赖以继续执行.依赖不需要完成该依赖,但它确实要求如果尚未完成,则有一个值.

如果调用的方法有core.attributes中定义的@waitrequired(要求等待)属性,则会在前面注入等待:

struct AnotherCo {
    int result() @safe @waitrequired {
        return 2;
    }
}
int myCo() @async {
    AnotherCo co = ...;
    //await co;
    int v = co.result;
    return 0;
}

如果考虑可实现性,则如果没有显式等待,则允许实现出错.但是,如果可以支持它,则建议它.它使框架作者可定位不需要知道正在使用协程的知识较少的用户.

安全

协程专为经验不足用户的多线程环境而设计,其目标是让他们可在短时间内不会出错的编写高效的基于事件的代码.

为了帮助实现它,该语言必须避免常见问题:
1,在状态间,线本存储不能引用它.

int* tlsVar;
void myCo() @async {
    ACo co = ...;
    int* var = tlsVar;
    await co;
    *var = 2; //错误:在`生成`后可能无法访问`"var"`中的`TLS`变量`"tlsVar"`.
}

2,所有协程函数都默认是@安全.你可显式按@trusted更改它,但,不能按@系统更改.如果是@trusted则允许此提案中描述的保护措施,如在阶段间允许不保留线本存储内存.

3,所有协程函数都默认有不抛属性.编译器抓所有异常,并在更新标签时,自动把它放进状态对象中.

4,协程中的参数不能是域(但可有空的逃逸集),引用或出.
5,同步语句,不能跨状态.一个内部的产生可能会导致死锁.

目的是,如果代码正常工作,协程就无法访问线程不安全的内存.但是,对进参数,或执行协程时取得的(协程拥有的)任意对象,这并不能清除所产生的边界问题,也无法触发异步返回它.
这需要进一步的提案来解决.

同步函数

当同步函数有协程对象时,它可能想阻塞等待.为此,库协程对象可能会提供带从core.attributes的@willwait属性的方法.

内部,可用与会阻止线程的互斥锁配对的系统条件变量来实现.

协程可能不会调用按@willwait标记的函数,因为这是个阻塞操作.在对象上用等待.

语法

语法如下更改:

AtAttribute:
+    '@' "async"
TypeSpecialization:
+    "__descriptorco"
TemplateTypeParmeterSpecialization:
+    ':' "__descriptorco"
ReturnStatement:
+    '@' "async" "return" Expression|opt ';'
Keyword:
+    "await"
NonEmptyStatementNoCaseNoDefault:
+    AwaitStatement
AwaitStatement:
+    "await" Expression ';'

此外,core.attributes中还引入了三个新属性:
1,isasync
2,waitrequired
3,willwait

这些标准针对库和框架作者,而不是协程的一般用户.它们帮助D语言经验不足的用户更容易使用协程.同时还帮助有经验的用户通过省略期望等待语句来避免犯错误.

实现

这是精心设计,对编译器的实现者友好.编辑器无需关心协程,只需匹配__descriptorco与函数.

对编译器实现者,如果在分析语义时无法改变以创建其他阶段,则允许犯@waitrequired错误,而不是注入最好产生.

否则,在对体分析语义时,在协程对象中切片函数,只需要关注按状态结构的方法转换自身,并生成基于分支表的猜语句以高效执行.

只有一个函数的好处是猜语句可添加case语句,且因为多线程性质,整数可能不是连续的.

示例

以下示例假设整个提案中,使用的InstantiableCoroutine库类型,包括按Future!ReturnType类型在堆上构造可执行实例中的叫makeInstance的方法.假设返回的未来有阻塞线程,直到返回值或完成的叫阻塞的方法.

质数筛

此例来自编程基准仓库.

void main(string[] args) {
    int n = args.length < 2   100 : to!int(args[1]);
    InstantiableCoroutine!(int) ico = &generate;
    Future!int ch = ico.makeInstance();
    ch.block; //`await ch;`
    foreach(i; 0 .. n) {
        int prime = ch.result;
        writeln(prime);
        filter(ch, prime);
    }
}
int generate() @async {
    int i = 2;
    for(;;) {
        @async return i;
        i++;
    }
}
void filter(ref Future!int ch, int prime) {
    ch.block;
    while(ch.result && ch.result % prime == 0) {
        ch.block;
    }
}

降级

为了完整,此处提供了协程的可能降级,实现者可能会限制和改进它.

该示例是简单超传客户,为了提高可读性,简化它,并忽略了套接字端的一些类型.

void clientCO(Socket socket) @async {
    writeln("Connection has been made");
    socket.write("GET / HTTP/1.1\r\n");
    socket.write("AcceptEncoding: identity\r\n");
    socket.write("\r\n");
    while(Future!string readLine = socket.readUntil("\n")) {
        await readLine;
        if (!readLine.isComplete) {
            writeln("Not alive and did not get a result");
            return;
        }
        string result = readLine.result;
        writeln(result);
        if (result == "</html>") {
            writeln("Saw end of expected input");
            return;
        }
    }
}

描述符:

static struct State {
    alias ReturnType = void;
    alias Parameters = (Socket);
    alias VarTypes = (readLine: Future!string);
    sumtype ExceptionTypes = :None;
    sumtype WaitingOn = Future!string;
    int tag;
    Parameters parameters;
    ExceptionTypes exception;
    WaitingOn waitingOnCoroutine;
    bool haveValue;
    ReturnType value;
    VarTypes vars;
    void execute() @safe nothrow {
        try {
            switch(this.tag) {
                case 0:
                    writeln("Connection has been made");
                    this.parameters.socket.write("GET / HTTP/1.1\r\n");
                    this.parameters.socket.write("AcceptEncoding: identity\r\n");
                    this.parameters.socket.write("\r\n");
                    this.vars.readLine = this.parameters.socket.readUntil("\n");
                    this.waitingOnCoroutine = this.vars.readLine;
                    this.tag = 1;
                    return;
                case 1:
                    if (!this.vars.readLine.isComplete) {
                        writeln("Not alive and did not get a result");
                        this.tag = -1; //完成!
                        return;
                    }
                    string result = this.vars.readLine.result;
                    writeln(result);
                    if (result == "</html>") {
                        writeln("Saw end of expected input");
                        this.tag = -1;
                        return;
                    }
                    this.vars.readLine = this.parameters.socket.readUntil("\n");
                    this.waitingOnCoroutine = this.vars.readLine;
                    this.tag = 1;
                    return;
                default:
                    assert(0); //编译器错误!
            }
        } catch (Exception e) {
            this.exception = e;
            this.tag = -2;
        }
    }
}

参考

放大镜下的纤程
纤程不再有用

并发编程
并发编程概念
函数的颜色
异步Rust在窗口上工作
Rustasync/.await入门
异步挂起函数的C#流
Swift异步let绑定