延迟对象deferred(ZT)

周末休息了两天，啥都没做，就看了个《大宅门》，自觉自己太堕落。今天上班也不顾老板的催促，看了一天的twisted。用twisted也有几周了，多多少少还是有些感悟，在这里写出来与大家分享，如果什么地方说的不对，还请互相帮助，大家共同进步~~
 
   好了，废话不多说，进入正题。今天我们讨论的是twisted的里面的又一大核心基础--deferred。
   Twisted 官方称，“Twisted is event-based, asynchronous framework ”。这个“异步”功能的代表就是 deferred。

deferred 的作用类似于“多线程”，负责保障多头连接、多项任务的异步执行。
当然，deferred “异步”功能的实现，与多线程完全不同，具有以下特点：

１. deferred 产生的 event，是函数调用返回的对象；

２. deferred 代表一个连接任务，负责报告任务执行的延迟情况和最终结果；

３. 对deferred 的操作，通过预定的“事件响应器”（event handler）进行。

有了deferred，即可对任务的执行进行管理控制。防止程序的运行，由于等待某项任务的完成而陷入阻塞停滞，提高整体运行的效率。

如果编写过非阻塞网络程序的人都知道，C语言中有个函数, 原型是：int connect(int fd, const void *addr, int addrlen); 此函数发起一个套接字fd上的到server addr的连接。
    如果fd是阻塞的，那么connect将一直阻塞直到连接成功建立或者连接失败。这个连接可能立即建立，也可能等待很久后才建立成功返回，那么程序将阻塞在这个地方，不能干其他的事情。这就是阻塞程序的缺点，但阻塞程序编程比较简单，而且在大多时候都是立即返回，所以在实际中并没有被淘汰。
    如果fd被设置成了非阻塞的，那么不论连接建立成功与否，connect会立即返回。如果出错或者连接已发起，但未成功，connect都会返回-1,但系统会设置errno,通过errno的值我们知道具体是出了什么错。这里我们关注的是errno的值[EINPROGRESS]，man手册是这样说的：
   Nonblocking I/O is enabled using O_NONBLOCK, O_NDELAY, or FIOSNBIO, and the connection cannot be completed immediately. This is not a failure. Make the connect() call again a few seconds later. Alternatively, wait for completion by calling select() and selecting for write.
   这不是一个错误，这表明连接已经发起，但还未成功。我们可以在一个时间段以后通过getsockopt检查连接是否成功，也可以通过重新调用connect，或者select的监听可写来判断监听是否成功。 这三种可以根据自己的情况选用。
  
  上面说了一些题外话，但对于我们理解defer确实大大有益。因为在用twisted编写网络服务器的时候，一般都是基于单线程事件处理，那么一旦某一个连接需要大量的时间，就会影响后续的连接，从而影响了系统在单位时间的吞吐量。这些需要花费大量时间的操作一般是指磁盘IO和数据库操作。如果让他们阻塞起，会白白浪费很多cpu时间，那么我们想要的就是它能够在调用后立即返回，我们去处理其他的事情。待到IO和数据库操作完成后有一个东西来通知我们，我们再去接着处理它。这样就能大大得提高了cpu的利用率，从而在资源一定的情况下提高了服务器的性能。而我们需要的这个它，就是 twisted为我提供的defer。

第一节： defer闪亮登场

 

from twisted.internet import reactor, defer

    class Getter:
        def getData(self, x):
            # this won't block
            d = defer.Deferred()
            reactor.callLater(2, d.callback, x * 3)
            return d

    def printData(d):
        print d

if __name__ == '__main__':    g = Getter()    d = g.getData(3)    d.addCallback(printData)        reactor.callLater(4, reactor.stop);     reactor.run()

 

分析下这个程序， 我们单独运行这个程序#python getdata.py
   系统从13行开始执行，在14行，生成一个Getter类的对象g，在15行调用g的方法getData，返回的deferred对象赋给d。 我们单步跟踪进getData函数内部， 也就是从第6行开始，15行的实参3传递给x，
函数首先生成一个defer对象d，用于管理稍后的回调。接着第7行，告诉程序2秒后执行d的回调函数，传进去的参数是x*3，也即9. 然后立即返回。紧接着执行16行，对d注册一个回调函数printData。在18行又注册一个4秒后的回调，这个回调将在4秒后停止整个reactor的运行，退出程序。待一切注册完毕之后，系统进入19行，开始了循环轮询事件。
   在07行， reactor.callLater(2, d.callback, x * 3)这一句有点容易把人给搞糊涂了。 reactor自身提供了callLater可以在一定时间以后调用某个函数，这也是一种回调，但不是defer的回调！
这句话的意思是2秒后，调用d的回调函数，而这个回调函数也就是我们在16行注册的printData。这里模拟了一个延时的环境，相当于我们在进行一个耗时的操作，而这个操作可能2秒后才能完成。
   所以整个程序，先调用getData，传进去3，然后返回defer对象d，reactor在2秒后调用了printData，d在整个过程中保存了getData的上下文环境，为最终调用printData提供了正确的实参。

看完第一个例子，想必大家对defer也有了一定的认识，下面我们更进一步讨论。

第二节： defer的几种使用方式

   上一节，介绍的是最基础的defer的使用，但在使用twisted的过程中，很少这样使用。因为我们一般都是建立自己的serverFactory和 serverProtocol，客户端对应clientFactory和clientProtocol，这样我们一般都是在主程序中创建一个工厂对象和它对应的协议，然后系统就进入了事件循环了，这时候大家可能会想，根据前面的例子套的话，该在什么地方创建defer？又该在什么地方添加 callback？在什么地方调用这些callback？  这些问题一直困扰了我很久，通过反复的看例子，终于在今天有了一些初步的醒悟。
根据我对twisted的了解，我总结了defer在factory和protocol的情况下的三种使用方式：

  1). 在factory中创建一个deferred对象，在Protocol中为这个deferred对象添加callback和errback。但真正的返回deferred的函数却在factory中。示例如下：

 

 

from twisted.internet import protocol, reactor, defer, utils
from twisted.protocols import basic
class FingerProtocol(basic.LineReceiver):
     def lineReceived(self, user):
         self.factory.getUser(user
         ).addErrback(lambda _: "Internal error in server"
         ).addCallback(lambda m:
                       (self.transport.write(str(m)+"/r/n"),
                        self.transport.loseConnection()))
 class FingerFactory(protocol.ServerFactory):
     protocol = FingerProtocol
     def getUser(self, user):
         return utils.getProcessOutput("finger", [user])

 if __name__ == '__main__':
     reactor.listenTCP(1079, FingerFactory())
     reactor.run()

第一点需要说明的是，这个程序要在linux下面运行，并且安装有finger程序。
如果系统没有finger程序，那么也没有问题，得把程序的13行改为：
return utils.getProcessOutput("ls", ['-a'])
这句话表示新建一个进程，这个进程执行” ls -a “命令，当然此处可以用系统的其他可执行命令，
这时候我们运行服务器
#python finger.py
那么程序将监听在1079端口上，我们在另外一台机器上连接这个服务器：

只要telnet成功后，任意输入几个字符，回车，本例中我输入的是‘d’，便能得到在服务器的当前目录下执行"ls -a"的结果。

这个程序比第一个要复杂一些，但此处我们不关注其他细节，只关注defer的使用。对于函数的13行，我们后来改的这句return utils.getProcessOutput("ls", ['-a'])，我们只需要知道它创建一个新进程执行一个命令”ls -a“， 然后返回一个deferred对象，这个deferred对象又将ls -a的结果传递给callback，然后在callback中将这个结果加上行结束符写回客户端。

这里又用到了一个python的特性lambda, lambda是用来生成一个临时函数，这个函数可以有参数也可以无参数。上例的errback就是无参数，用一个下划线_做形参，callback的形参为m，而实参就是通过defer传递进来的新进程执行ls -a的结果。

程序中用到了许多twisted的基本东西，像 basic.LineReceiver, 它是对twisted基本protocol的一个继承。说到这里，插一句题外话，学习twisted，一定得看源代码，因为twisted是一个很高层次的封装，虽然用起来很简单，但如果对底层的实现不清楚，你就不知道怎么用，也不知道在什么时候会出现什么异常。具体源代码在python安装目录下的 C:/Python25/Lib/site-packages/twisted下面，其中我是把python直接装在C盘下面的。

2). 在factory中创建deferred对象，但在调用factory对象之外返回自身的deferred对象，然后再添加callback和errback， 具体示例如下：

from twisted.internet import reactor, defer, protocol
class CallbackAndDisconnectProtocol(protocol.Protocol):
    def connectionMade(self):
        self.factory.deferred.callback("Connected!")

    self.transport.loseConnection()

class ConnectionTestFactory(protocol.ClientFactory):
    protocol = CallbackAndDisconnectProtocol
    def __init__(self):
        self.deferred = defer.Deferred() #report deferred event, prevent program block on this task

    def clientConnectionFailed(self, connector, reason):
        self.deferred.errback(reason)

def testConnect(host, port):
    testFactory = ConnectionTestFactory()
    reactor.connectTCP(host, port, testFactory)
    return testFactory.deferred

def handleSuccess(result, port):
    #deferred "event-responsor": handle finished connection
    print "Connected to port %i" % port
    reactor.stop()

def handleFailure(failure, port):
    print "Error connecting to port %i: %s" % (port, failure.getErrorMessage())
    reactor.stop()

if __name__ == '__main__':
    import sys
    if not len(sys.argv) == 3:
        print "Usage: connectiontest.py host port"
        sys.exit()

    host = sys.argv[1]
    port = int(sys.argv[2])
    connecting = testConnect(host, port)
    connecting.addCallback(handleSuccess, port)
    connecting.addErrback(handleFailure, port)
    reactor.run()

 

这是一个客户端的程序，他是在外部通过函数调用factory对象返回一个deferred对象（line 44），然后在45，46行添加了callback和errback。 我们用上一个例子作为本程序的服务器，然后运行本程序：

在连接建立成功后，系统自动调用connectionMade（line 06），在函数里面调用factory的deferred对象的callback函数handleSuccess，传递实参”connected！“给 25行的result，但这个实参在本程序中并未使用，只是相当于一个标识。参数port是通过45行添加callback的时候传递的。

3). 不显示创建deferred对象， 在函数返回时，我们调用defer.succeed(results)函数。其中result则可以为调用我们延时操作函数返回的结果。示例如下：

 

01 #-*- coding:utf-8 -*-
02 from twisted.internet import protocol, reactor, defer
03 from twisted.protocols import basic
04
05 class FingerProtocol(basic.LineReceiver):
06 def lineReceived(self, user):
07 self.factory.getUser(user
08 ).addErrback(lambda _: "Internal error in server"
09 ).addCallback(lambda m:
10 (self.transport.write(m+"/r/n"),
11 self.transport.loseConnection()))
12
13 class FingerFactory(protocol.ServerFactory):
14 protocol = FingerProtocol
15
16 def __init__(self, **kwargs):
17 self.users = kwargs
18
19 def getUser(self, user):
20 return defer.succeed(self.users.get(user, "No such user"))
21
22 if __name__ == '__main__':
23 reactor.listenTCP(1079, FingerFactory(hd='Hello my python world'))
24 reactor.run()

  这是一个简单的返回用户信息的服务器程序，在23行，系统初始化了一个用户hd, 他的信息为”Hello my python world“, 在20行调用users.get的时候，我们告诉系统，如果不存在这个用户，就返回”No such user“， 17行的self.users是一个字典，{'hd':'Hello my python world'}.
我们运行这个服务器，然后在另一台电脑上telnet登陆，发送hd和其他随机的字符，看看结果如何：

看看结果怎么样，跟你预期的相同么？
这里附上defer.succeed的函数定义

 

01 def succeed(result):
02 """
03 Return a Deferred that has already had '.callback(result)' called.
04
05 This is useful when you're writing synchronous code to an
06 asynchronous interface: i.e., some code is calling you expecting a
07 Deferred result, but you don't actually need to do anything
08 asynchronous. Just return defer.succeed(theResult).
09
10 See L{fail} for a version of this function that uses a failing
11 Deferred rather than a successful one.
12
13 @param result: The result to give to the Deferred's 'callback'
14 method.
15
16 @rtype: L{Deferred}
17 """
18 d = Deferred()
19 d.callback(result)
20 return d

在这段代码里面，自动给我们创建了一个deferred对象，然后将result传递给callback函数。

看了这几个例子，你是不是感觉对于defer稍微有点feel了？ 累了就歇会，等会我们再进入下一节。

第三节：多重回调

在twisted的multi callbacks 中有下面一段话，
Multiple callbacks can be added to a Deferred. The first callback in the Deferred's callback chain will be called with the result, the second with the result of the first callback, and so on. Why do we need this? Well, consider a Deferred returned by twisted.enterprise.adbapi - the result of a SQL query. A web widget might add a callback that converts this result into HTML, and pass the Deferred onwards, where the callback will be used by twisted to return the result to the HTTP client. The callback chain will be bypassed in case of errors or exceptions.
通过这段话我们知道，我们可以给twisted添加多个callback，而且每一个callback都是以上一个的返回结果作为传入参数被调用，这些callback会被依次调用。

图1  callback基本原理

 

图2 callback和errback调用规则

1. When the result is ready, give it to the Deferred object. .callback(result) if the operation succeeded, .errback(failure) if it failed. Note that failure is typically an instance of a twisted.python.failure.Failure instance.
2. Deferred object triggers previously-added (call/err)back with the result or failure. Execution then follows the following rules, going down the chain of callbacks to be processed.
3. • Result of the callback is always passed as the first argument to the next callback, creating a chain of processors.
   • If a callback raises an exception, switch to errback.
   • An unhandled failure gets passed down the line of errbacks, this creating an asynchronous analog to a series to a series of except: statements.
   • If an errback doesn't raise an exception or return a twisted.python.failure.Failure instance, switch to callback.

看了上面的图，大家有了基本的了解，一般来说添加一个callback也需要添加一个errback。