Riceball's profileRiceball LEE personal we...PhotosBlogLists Tools Help

Blog


    January 22

    Continuation 概念与协程(CoRoutine)

    所谓Continuation就是保存接下来要做的事情的内容(the rest of the computation)。举个简单例子,我在写文档,突然接到电话要外出,这时我存档,存档的数据就是Continuation(继续即将的写作),然后等会儿回来,调入存档,继续写作。Continuation这个概念就协程来说就是协程保护的现场。而对于函数来说就是保存函数调用现场——Stack Frame值和寄存器,以供以后调用继续从Continuation处执行。换一个角度看,它也可以看作是非结构化Goto语句的函数表达。当我们执行Yield从协程返回的时候,需要保存的就是Continuation了。从理论研究的角度上来说Continuation即是对程序"接下来要做的事情"所进行的一种建模,从而能对之作进一步的分析。Continuation是对未来的完整描述,这对于理论分析而言是有很多方便的地方。实际上任何程序都可以通过CPS(Continuation Passing Style)类型转换为使用Continuation的形式。如:

    function foo(x: integer); begin Result := x + 1; end; //foo 函数的CPS形式: procedure foo(x: integer; c: Continuation); begin //使用continuation的函数不"返回"值,而是把值作为一个参数传递给continuation从而"继续"处理值 c.continueWith(x+1); end;

    对于我们熟悉的Pascal,C++等传统高级编译语言而言,程序本身在运行期是固定不变的,我们只需要记录下执行点(excution point)的信息(例如指针位置和堆栈内容)即足以完整的描述程序未来的运行情况,一般都是使用堆栈保存函数上下文的——采用Activation record或者叫Stack frame来记录从最顶层函数到当前函数的所有函数的上下文,当函数进入时候将局部变量等作为StackFrame压入堆栈,退出函数的时候它的StackFrame就被删掉(具体压入弹出的时机视调用约定而定)。

    而在许多的函数式语言中(如 Scheme 和SmallTalk),它们并不采用堆栈来保存上下文,而是将这些信息保存在continuation记录中。这些continuation记录和堆栈的Frame的区别在于,它不采用后入先出的线性方式,所有continuation记录被组成一棵树(或者图),从一个函数调用另一个函数就等于给当前节点生成一个子节点,然后把系统寄存器移动到这个子节点。一个函数的退出等于从当前节点退回到父节点。这些节点的空间回收是由垃圾回收器(garbage collection)来管理。如果没有引用这个continuation记录,则它就是可以被删除的。这样的调用方式和堆栈方式相比,它可以在一个函数内的任何位置储存自己的上下文信息,然后,在以后某个适当的时刻,从其它的任何一个函数里面返回到自己现在的位置。实际上它使得一个函数可以拥有了多个不同的入口点(知道为何可以看作是Goto语句的函数表达式了吧,不过我们在软件工程中极度忌讳使用GOTO模式避免面条式代码成形。)

    在函数式语言中, continuation的引入是非常自然的过程, 考察如下函数调用: f(h(k(arg))),根据函数的结合律,我们可以有复合函数 g = f(h(.)), 它自然就是函数k()的continuation,在理论上我们有可能利用泛函分析的一些技术实现对于continuation(复合函数)的化简,但实践已经证明这是极为艰难的,主要是我们的程序不可避免的要涉及到程序与数据的纠缠。不过,我们在引入continuation概念之后,程序运行的表述是非常简单的:

    Continuation.Proceed();

    针对顺序执行的程序,我们可以建立更加精细的运行模型。

    while(Continuation.HasNextStep()) do begin Continuation.ProceedOneStep(); end;

    只要以某种方式构造出Continuation 子句(closure),就意味着我们能够通过单一变量来表示程序未来的运行结构。这样,我们就有可能在某个层面上实现对程序的一种更简洁的描述。

    在上一次对Delphi的CoRoutine(uMeYield.pas)实现的基础上,我实现在对Delphi语言函数的Continuation。通过为CoRoutine类添加了MarkContinuation方法保存Continuation记录以及CallCC方法调用进入保存的Continuation记录所在位置执行CoRoutine。

    var vContinuationRec: TContinuationRec; procedure YieldProc(const YieldObj: TMeCoroutine); var i: integer; begin i := 0; while true do begin inc(i); writeln('i= ', i); YieldObj.Yield(i); if i >= 3 then break; end; YieldObj.MarkContinuation(vContinuationRec); inc(i); writeln('i++ end: ', i); end; with TYieldInteger.Create(YieldProc) do try Reset; while MoveNext do begin inc(i); Writeln('Current:', Current); end; Writeln('---CallCC---'); CallCC(vContinuationRec); CallCC(vContinuationRec); finally Free; end;

    程序输出:

    i=1 Current=1 i=2 Current=2 i=3 Current=3 i++ end: 4 '---CallCC--- i++ end: 5 i++ end: 6

    以上实现只是纯学术上的研究,如果要在实际上应用这种函数重入,一来是不符合软件工程规范(当然如果你要作为反破解的扰乱另当别论),二来是你必须很清楚它的限制条件:在需要重入的函数体,不能使用动态分配内存的局部变量(如字符串)。

    当然,Continuation不仅仅只是适用于程序语言理论分析领域,如果我们的眼界开阔一些,不拘泥于构造语言级别通用的continuation结构(这需要以抽象的方式定义并保存任意程序的完整运行状态),而是考察“对程序未来运行的整体结构进行建模”这一更宽广的命题,我们很快就能发现大量对于continuation概念的应用,如:异常处理,回退跟踪(back-tracking),AOP的拦截器,动态工作流,Web Continuation。

    异常处理
    应用continuation进行异常处理是很显而易见的,只要在可能抛出异常的函数外面try的地方做一个continuation 记录,那么这个函数就可以在需要的时候直接返回到try语句的地方。在出现了异常之后,当异常处理模块修复了错误后,需要返回道发生错误的地方继续执行的时候,就是continuation大显神威的时候。

    回退跟踪(back-tracking)
    回退跟踪(back-tracking)算法也可以应用continuation,在某些地方保存当前的continuation,然后以后回退的时候,就可以从其它的函数直接跳回。

    AOP(Aspect Oriented Programming)拦截器
    对原来函数拦截后,我们需要在拦截器中处理函数调用前事件,然后继续执行原函数。

    TMethodInterceptor = class procedure Invoke(const Invocation: TMethodInvocation); begin doSthBeforeInvoke(); Invocation.Proceed(); end; end;

    动态工作流
    在一些比较复杂的网络协议中,我们通过注册监听器(listener)来处理接收到的网络指令,网络指令之间往往存在一定的关联,我们可以通过针对不同的指令动态调整关联的命令监听器, 或者建立复杂而庞大的有限自动机来描述所有指令之间的关联规则。

    //动态调整关联的命令监听器的方式 TCommandListener = class procedure OnEvent(aEvent: TEvent, aFutureListeners: TListeners) begin HandleEvent(aEvent); aFutureListeners.clear(); aFutureListeners.add("ACommand", new ACommandListener()); aFutureListeners.add("BCommand", new BCommandListener()); end end;

    而动态调整关联的命令监听器的方式可以看作是对程序未来运行结构的一种动态调整. 沿着这种方式深入下去, 我们可以建立一种完整的动态工作流机制.

    Web Continuation

    传统Web开发,一般都是以客户端作为主动的,客户端发请求,然后接收响应,然后再发请求...,整个流程都是以客户端为推动源。这样的一个结果就是,一般的web框架都是把他们的控制器分成一个个的方法调用,客户端的请求就对应到这些方法调用当中。而Web Continuation Server 通过引入Continuation机制将逻辑反转了过来,并以此实现了对于page flow的完整描述。

    Continuation Server 以服务器作为主动方,服务器发送页面,然后等待客户端输入之后,继续执行,然后在发送页面并等待回应...,整个流程是服务器通过发送页面和等待回((SendPageAndWait))应进行推动。整个过程就像是函数调用那样,服务器发送页面回应(SendPageAndWait)就是函数调用开始,而用户发送请求就是函数的返回。要实现这个效果,就需要服务器端可以在收到请求之后能返回到之前的发送响应的后一语句。这里的核心就是服务器端需要能够动态的获取运行栈,在发送响应前,先对当前的运行栈作一个快照,然后在响应到达时,重新从快照那里执行,这样就相当于实现了刚才所说的函数调用效果。使用continuation server之后服务器端就只需要一个方法调用,对应初始请求。

    procedure OnRequest() begin funcA(); input = SendPageAndWait("GetInfoFromUser.php"); HandleInput(input); end;

    在调用SendPageAndWait的时候,web框架会保存当前函数调用的continuation,向用户返回页面GetInfoFromUser.php,等待用户提交表单之后,web框架重新激活我们所保存的continuation,继续执行我们的函数。这种做法与系统调用和线程调度等机制是非常类似的。虽然这种基于continuation的方式可以自然的解决在session中保存并清理变量的问题,但是使用通用的continuation 实现很有可能会在无意中保存了过多的临时变量,从而对系统性能造成极大的损害,另外在集群环境下,continuation状态如何复制也是需要思考如何解决的问题。

    那么为什么需要Continuation呢?利用 Continuation 理念,可以很好的解决了服务器更新客户端的问题——服务器从而不用再为每一个等待响应的客户端单独建立一个线程,借助Continuations和新IO,可以在有限的线程内支持更多用户。根据Greg Wilkins的测试,使用Continuation的Jetty Cometd服务在10000个并发用户和875个线程下,只用了57M内存。

    January 03

    Coroutine(协程)与复用

    概念

    协程(Coroutine)这个概念最早是Melvin Conway在1963年提出的,用它可以实现协作式多任务,协程(coroutine)技术本质上是一种程序控制机制。比如,消费者/生产者,你走几步,我走几步;下棋对弈,你一步我一步。
    Coroutine(协程)可以分为:

    • 非对称式(asymmetric)协程,或称半对称式(semi-asymmetric)协程,又或干脆就叫半协程(semi-coroutine)
    • 对称式(symmetric)协程。

    非对称式(asymmetric)协程之所以被称为非对称的,是因为它提供了两种传递程序控制权的操作:一种是(重)调用协程(通过coroutine.resume);另一种是挂起协程并将程序控制权返回给协程的调用者(通过coroutine.yield)。一个非对称协程可以看做是从属于它的调用者的,二者的关系非常类似于例程(routine)与其调用者之间的关系。对称式(symmetric)协程的特点是只有一种传递程序控制权的操作(coroutine.transfer),即将控制权直接传递给指定的协程。曾经有这么一种说法,对称式和非对称式协程机制的能力并不等价,但事实上很容易根据前者来实现后者。在不少动态脚本语言(Python、Perl,Lua,Ruby)都提供了协程或与之相似的机制。
    对称式协程机制可以直接指定控制权传递的目标,拥有极大的自由,但得到这种自由的代价却是牺牲程序结构。如果程序稍微复杂一点,那么即使是非常有经验的程序员也很难对程序流程有全面而清晰的把握。这非常类似goto语句,它能让程序跳转到任何想去的地方,但人们却很难理解充斥着goto的程序。非对称式协程具有良好的层次化结构关系,(重)启动这些协程与调用一个函数非常类似:被(重)启动的协程得到控制权开始执行,然后挂起(或结束)并将控制权返回给协程调用者。这与结构化编程风格是完全一致的。

    线程和协程的异同

    协程(Coroutine)类似于线程(Thread)的地方是:每个协程都有有自己的堆栈,自己的局部变量。

    线程和协程的主要区别在于:
     1. 线程可以并发运行,线程之间是不能共享全局变量。
     2. 协程不能并发运行,协程之间可以共享全局变量。

     

    协程的实现和使用

    创建协程

    type TMeCoRoutineFunc = procedure(const aCoRoutine: TMeCoRoutine); TMeCoRoutineMethod = procedure() of object; var func: TMeCoroutineFunc; co = TMeCoRoutine.create(func)

    参数是一个函数,返回值是创建的协程对象。

    协程的状态

    协程有三种状态:挂起(suspended),运行(running),停止(dead)。
    当我们创建一个协程时他开始的状态为挂起态,也就是说我们创建协程的时候不会自动运行。

    st = co.status;

     

    激活协程

    IsSucessful := co.resume();

    激活挂起的协程,使协程继续运行。参数co是一个协程对象。

    如果协程是挂起状态,则继续运行,resume函数返回true。如果协程已经停止或者遇到其他错误,resume函数返回false。

     

    挂起协程

    co.yield([...]);

    挂起当前协程。直到协程被外部协程使用CoRoutine.Resume再次激活,将返回到执行CoRoutine.Yield函数后的地方继续执行。
    CoRoutine.yield的参数将传递给SaveYieldedValue虚方法,你需要重载该方法处理。

    当一个协程正在运行时,不能在外部终止它.只能在协程内部调用coroutine.yield挂起当前协程。
    不需要考虑协程安全、协程同步的问题。协程的代码比线程的代码更容易编写。

     

    === “控制”和“行为”的复用 ===

    在很多时候,我们需要对数据结构(如:List,Stack)中的元素按某种要求进行遍历,我们称之为“控制”;然后对目标元素进行某个操作,我们称之为“行为”。许多情况下,这种“控制”或行为的代码本来是可以被复用的,但是因为难以将这其中的“控制”和“行为”分离,造成了我们不得不一遍又一遍的书写这些类似的代码(虽然利用回调可以实现在一定程度上的“控制”和行为的分离,但是并不优雅,也不无法实现彻底重用)。

    ==== 生产和消费 ====
    让我们先看下面一段代码,producer过程(生产者)产生一些数值(根据要求进行遍历),而Consumer过程(消费者)则处理值(对目标元素进行操作):

    procedure producer(); var i: integer; begin for i := 0 to 100 do if i mod 5 = 0 then consumer(i); end; procedure Consumer(const value: integer); begin writeln(value); end;

    请注意在生产者(producer)调用消费过程(consumer)这里出现了耦合,该生产者只能为这个Cosumer过程服务。我们希望生产者过程(producer)能增强通用性,降低耦合度,能为不同的消费者服务。

    ok,我们想到了回调:

    type TComsumerCallback: procedure(const value:integer); procedure producer(const aCallBack: TComsumerCallback); var i: integer; begin for i := 0 to 100 do //循环枚举控制 if i mod 5 = 0 then aCallBack(i); end;

    好了producer可以为不同的消费者服务了。但是,新的问题又出来。如果我们现在还希望仅当消费者要求值的时候才去调用生产者取得值呢?实现控制和行为的彻底的分离。象这样:

    procedure MainConsumer(); begin //控制在MyProducer中,控制复用 for i in MyProducer do //当消费者要求值的时候才去调用 begin Consumer(i); .... //随时可以停止从生产者取值 end; end;

    ==== Visitor模式与Iterator模式 ====

    回调的实质是一种简单Visitor模式,说它简单是因为它只有Visitor,没有Visited部分。利用回调很难做到:消费者要,生产者才给,如果消费者不问不要,生产者就不答不给,这是由visitor模式的特性所决定的:回调的使用者把Callback函数扔到Traversal算法里面,然后运行算法,同时祈祷并等候算法的完成(Push and Wait),使用者完全失去了控制权,只能等待算法整个完成或者中止,才能重新拿到控制权。而尽管使用Iterator模式能很容易的做到这一点(Iterator本质上属于问答模式,或者说消费者/生产者模式,Iterator的用法本身就是Lazy的,一问一答,遍历算法停在那里恭候Iterator使用者的调遣),但是如果放弃回调方式却又无法复用消费者了。要想同时做到既要复用“控制”,又要复用“行为”,这几乎是不可能的(当然如果是仅仅想实现“消费者要,生产者才给”那是可以的,不过难度比回调大,而且并不能通用各种数据结构),因为visitor模式和Iterator模式的特性恰恰是完全相反的:
     * Iterator是一种主动模型,Pull模型,Ask and Get。Iterator听候用户的调遣。
     * Vistor是一种被动模型,Push模型,Plugin / callback模型,Push and Pray and Wait。Visitor听候算法的调遣。

    //利用Iterator模式实现按消费者需要取值:简单的数组、链表只要保存当前调用步骤(数组索引,或者当前指针)和调用环境(内部数据集)的结构,返回给用户就可以了。用户每次调用iterator.next,iterator就把索引或指针向后移动一下。 如果是内部数据复杂的Tree, Graph结构,就相当复杂了。比如是遍历一棵树,而且这棵树的Node里面没有Parent引用,那么Iterator必须自己维护一个栈把前面的所有的Parent Node都保存起来。当用户调用iterator.next的时候,iterator就必须检查自己当前的状态,如果所有的Child Node都走完了,那么就要返回到上面的Parent,继续检查。 type TProducer = class private i: integer; public Current: Integer; function MoveNext: boolean; end; procedure TProducer.MoveNext; var i: integer; begin Result := False; if i <= 100 then if i mod 5 = 0 then Current := i; i := i + 1; Result := True; end; end; procedure MainConsumer(); begin with TProducer.Create do try while MoveNext do begin Consumer(Current); .... //随时可以停止从生产者取值 end; finally Free; end; end;

    这时候,有聪明人就将目光转向了非对称式(asymmetric-coroutine)协程。不难看出这里面的Iterator就是Coroutine里面的生产者(数据提供者,所以有时我们也称之为Generator)。
    每次Iterator程序就是等在那里,一旦用户(消费者Consumer角色)调用了iterator.next(coroutine.Resume), Iterator就继续向下执行一步,然后把当前遇到的内部数据的Node放到一个消费者用户能够看到的公用的缓冲区(比如,直接放到消费者线程栈里面的局部变量)里面,然后自己就停下来(coroutine.Yield)。然后消费者用户就从缓冲区里面获得了那个Node。这样Iterator就可以自顾自地进行递归运算,不需要自己管理堆栈上下文,而是协程机制帮助它分配和管理运行栈。从而实现将“控制”和“行为”的彻底解藕。请看如下C#程序:

    using System.Collections.Generic; public class MyProducer : IEnumerable<string> {a //iterator block,实现枚举元素控制 public IEnumerator<string> GetEnumerator() { for(int i = 0; i<elements.Length; i++) yield elements[i]; } ... } foreach (string item in new MyProducer()) { //实现终端上打印元素的行为 Console.WriteLine(item); }

    在这段代码执行过程中,foreach 的循环体和 GetEnumerator 函数体实际上是在同一个线程中交替执行的。这是一种介于线程和顺序执行之间的协同执行模式,之所以称之为协同(Coroutine),是因为同时执行的多个代码块之间的调度是由逻辑隐式协同完成的。就协同执行而言,从功能上可以分为行为、控制两部分,控制又可进一步细分为控制逻辑和控制状态。行为对应着如何处理目标对象,如上述代码中:行为就是将目标对象打印到终端;控制则是如何遍历这个 elements 数组,可进一步细分为控制逻辑(顺序遍历)和控制状态(当前遍历到哪个元素)。其中心思想在于通过Coroutine程序控制机制和Yield将其行为与控制彻底分离,以此来进一步降低代码的耦合度,增强通用性,提高代码的复用率。