（轉） Twisted ：第十九部分 改變以前的想法

簡介

Twisted是一個正在進展的項目,它的開發者會按期添加新的特性而且擴展舊的特性.

隨着Twisted 10.1.0發佈,開發者向 Deferred 類添加了一個新的特性—— cancellation ——這正是咱們今天要研究的.

異步編程將請求和響應解耦了,如此又帶來一個新的可能性:在請求結果和返回結果之間,你可能決定再也不須要這個結果了.考慮一下 :doc:`p14` 中的詩歌代理服務器.下面是這個如何工做的,至少對於詩歌的第一次請求:

1. 一個對詩歌的請求來了.

2. 這個代理聯繫實際服務器以獲得這首詩

3. 一旦這首詩完成,將其發送給原發出請求的代理

看起來很是完美,可是若是客戶端在得到詩歌以前掛了怎麼辦?也許它們先前請求 Paradise Lost 的所有內容,隨後它們決定實際想要的是 Kojo 的俳句.咱們的代理將陷入下載前者,而且那個慢服務器會等好一會.最好的策略即是關閉鏈接,讓慢服務器回去順覺.

回憶一下 第十五部分,展現了同步程序控制流的概念.在那張圖中咱們能夠看到函數調用自上而下,異常是自下而上.若是咱們但願取消一個同步調用(這僅是假設),控制流的傳遞方向與函數調用的方向一致,都是從高層傳向底層,如圖38所示:

圖38:同步程序流,含假想取消操做

固然,在同步程序中這是不可能的,由於高層的代碼在底層操做結束前沒有恢復運行,天然也就沒有什麼可取消的.可是在異步程序中,高層代碼在底層代碼完成前具備控制權,至少具備在底層代碼完成以前取消它的請求的可能性.

在Twisted程序中,底層請求被包含在一個 Deferred 對象中,你能夠將其想象爲一個外部異步操做的"句柄". deferred 中正常的信息流是向下的,從底層代碼到高層代碼,與同步程序中返回的信息流方向一致.從Twisted 10.1.0開始,高層代碼能夠反向發送信息 —— 它能夠告訴底層代碼它再也不須要其結果了.如圖39:

圖39： deferred 中的信息流,包含取消

取消 Deferreds

讓咱們看一些例程,來了解下取消 deferreds 的實際工做原理.注:爲了運行這些列子以及本部分中的其餘代碼,你須要安裝Twisted 10.1.0或更高 版本. 考慮 deferred-cancel/defer-cancel-1.py:

from twisted.internet import defer

def callback(res):
    print 'callback got:', res

d = defer.Deferred()
d.addCallback(callback)
d.cancel()
print 'done'

伴隨着新的取消特性, Deferred 類得到一個名爲 cancel 的新方法.上面代碼建立了一個新的 deferred,添加了一個回調,這後取消了這個 deferred 而沒有激發它.輸出以下:

done
Unhandled error in Deferred:
Traceback (most recent call last):
Failure: twisted.internet.defer.CancelledError:

OK,取消一個 deferred 看起來像使錯誤回調鏈運行,常規的回調根本沒有被調用.一樣注意到這個錯誤是: twisted.internet.defer.CancelledError,一個意味着 deferred 被取消的個性化異常(但請繼續閱讀).讓咱們添加一個錯誤回調,如deferred-cancel/defer-cancel-2.py

from twisted.internet import defer

def callback(res):
    print 'callback got:', res

def errback(err):
    print 'errback got:', err

d = defer.Deferred()
d.addCallbacks(callback, errback)
d.cancel()
print 'done'

獲得如下輸出:

errback got: [Failure instance: Traceback (failure with no frames): <class 'twisted.internet.defer.CancelledError'>:
]
done

因此咱們能夠'捕獲'從 cancel 產生的錯誤回調,就像其餘 deferred 錯誤同樣.

OK,讓咱們試試激發 deferred 而後取消它,如 deferred-cancel/defer-cancel-3.py:

from twisted.internet import defer

def callback(res):
    print 'callback got:', res

def errback(err):
    print 'errback got:', err

d = defer.Deferred()
d.addCallbacks(callback, errback)
d.callback('result')
d.cancel()
print 'done'

這裏咱們用常規 callback 方法激發 deferred,以後取消它.輸出結果以下:

callback got: result
done

咱們的回調被調用(正如咱們所預期的)以後程序正常結束,就像 cancel 根本沒有被調用.因此取消一個 deferred 好像根本沒有效果若是它已經被激發(但請繼續閱讀!).

若是咱們在取消 deferred 以後激發它會怎樣?參看 deferred-cancel/defer-cancel-4.py:

from twisted.internet import defer

def callback(res):
    print 'callback got:', res

def errback(err):
    print 'errback got:', err

d = defer.Deferred()
d.addCallbacks(callback, errback)
d.cancel()
d.callback('result')
print 'done'

這種狀況的輸出以下:

errback got: [Failure instance: Traceback (failure with no frames): <class 'twisted.internet.defer.CancelledError'>:
]
done

有意思！與第二個例子的輸出同樣,當時沒有激發 deferred.因此若是 deferred 被取消了,再激發它沒有效果.可是爲何 d.callback('result') 沒有產生錯誤,考慮到不能激發 deferred 大於一次,錯誤回調鏈爲什麼沒有運行?

再次考慮 figure39.用結果或失敗激發一個 deferred 是底層代碼的工做,然而取消 deferred 是高層代碼的行爲.激發deferred 意味着"這是你的結果",然而取消 deferred 意味着"我再也不想要這個結果了".同時記住 cancel 是一個新特性,因此大部分現有的Twisted代碼並無處理取消的操做.可是Twisted的開發者使咱們取消 deferred 的想法變得有可能,甚至包括那些在Twisted 10.1.0以前寫的代碼.

爲了實現以上想法, cancel 方法實際上作兩件事:

1. 告訴 Deferred 對象自己你不想要那個結果,若是它尚未返回(如, deferred 沒有被激發),這樣忽略任何回調或錯誤回調的後續調用.

2. 同時,可選地,告訴正在產生結果的底層代碼須要採起何種步驟來取消操做.

因爲舊版本的Twisted代碼會上前去激發任何已經被取消的 deferred, step#1確保咱們的程序不會垮掉若是咱們取消一箇舊有庫中的 deferred.

這意味着咱們能夠爲所欲爲地取消一個 deferred,同時能夠肯定不會獲得結果若是它尚未到來(甚至那些 將要 到來的).可是取消 deferred 可能並無取消異步操做.終止一個異步操做須要一個上下文的具體行動.你可能須要關閉網絡鏈接,回滾數據庫事務,結束子進程,等等.因爲 deferred 僅僅是通常目的的回調組織者,它怎麼知道具體要作什麼當你取消它時?或者,換種說法,它怎樣將 cancel 請求傳遞給首先已經建立和返回了 deferred 的底層代碼? 和我一塊兒說:

I know, with a callback!

本質上取消 Deferreds

好吧,首先看一下 deferred-cancel/defer-cancel-5.py:

from twisted.internet import defer

def canceller(d):
    print "I need to cancel this deferred:", d

def callback(res):
    print 'callback got:', res

def errback(err):
    print 'errback got:', err

d = defer.Deferred(canceller) # created by lower-level code
d.addCallbacks(callback, errback) # added by higher-level code
d.cancel()
print 'done'

這個例子基本上跟第二個例子相同,除了有第三個回調(canceller).這個回調是咱們在建立 Deferred 的時候傳遞給它的,不是以後添加的.這個回調負責執行終止異步操做時所需的上下文相關的具體操做(固然,僅當 deferred 被實際取消). canceller 回調是返回 deferred 的底層代碼的必要部分,不是接收 deferred 的高層代碼爲其本身添加的回調和錯誤回調.

運行這個例子將產生以下輸出:

I need to cancel this deferred: <Deferred at 0xb7669d2cL>
errback got: [Failure instance: Traceback (failure with no frames): <class 'twisted.internet.defer.CancelledError'>:
]
done

正如你所看到, 不須要返回結果的 deferred 被傳遞給 canceller 回調.在這裏咱們能夠作任何須要作的事情以便完全終止異步操做.注意 canceller 在錯誤回調鏈激發前被調用.其實咱們能夠在取消回調中選擇使用任何結果或錯誤本身激發 deferred(這樣就會優先於 CancelledError 失敗).這兩種狀況在 deferred-cancel/defer-cancel-6.py 和 deferred-cancel/defer-cancel-7.py中進行了說明.

在激發 reactor 以前先作一個簡單的測試.咱們將使用 canceller 回調建立一個 deferred,正常的激發它,以後取消它.你能夠在 deferred-cancel/defer-cancel-8.py 中看到代碼.經過檢查那個腳本的輸出,你將看到取消一個被激發的 deferred 不會調用canceller 回調.這正是咱們所要的,由於沒什麼可取消的.

咱們目前看到的例子都沒有實際的異步操做. 讓咱們構造一個調用異步操做的簡單程序,以後咱們將指出如何使那個操做可取消.

參見代碼 deferred-cancel/defer-cancel-9.py:

from twisted.internet.defer import Deferred

def send_poem(d):
    print 'Sending poem'
    d.callback('Once upon a midnight dreary')

def get_poem():
    """Return a poem 5 seconds later."""
    from twisted.internet import reactor
    d = Deferred()
    reactor.callLater(5, send_poem, d)
    return d

def got_poem(poem):
    print 'I got a poem:', poem

def poem_error(err):
    print 'get_poem failed:', err

def main():
    from twisted.internet import reactor
    reactor.callLater(10, reactor.stop) # stop the reactor in 10 seconds

    d = get_poem()
    d.addCallbacks(got_poem, poem_error)

    reactor.run()

main()

這個例子中包含了一個 get_poem 函數,它使用 reactor 的 callLater 方法在被調用5秒鐘後異步地返回一首詩.主函數調用 get_poem,添加一個回調/錯誤回調對,以後啓動 reactor.咱們(一樣使用 callLater)安排 reactor 在10秒鐘以後中止.一般咱們向 deferred 添加一個回調來實現,但你很快就會知道咱們爲什麼這樣作.

運行程序(適當延遲後)產生以下輸出:

Sending poem
I got a poem: Once upon a midnight dreary

10秒鐘後程序終止.如今來試試在詩歌被髮送前取消 deferred.只需加入如下代碼在2秒鐘後取消(在5秒鐘延遲發送詩歌以前):

reactor.callLater(2, d.cancel) # cancel after 2 seconds

完整的例子參見 deferred-cancel/defer-cancel-10.py,這將產生以下輸出:

get_poem failed: [Failure instance: Traceback (failure with no frames): <class 'twisted.internet.defer.CancelledError'>:
]
Sending poem

這個例子清晰地展現了取消一個 deferred 並無取消它背後的異步請求.2秒鐘後咱們看到了錯誤回調輸出,打印出如咱們所料的 CancelledError 錯誤.可是5秒鐘後咱們看到了 send_poem 的輸出(可是這個 deferred 上的回調並無激發).

這時咱們與 deferred-cancel/defer-cancel-4.py 的狀況同樣."取消" deferred 僅僅是使最終結果被忽略,但實際上並無終止這個操做.正如咱們上面所學,爲了獲得一個真正可取消的 deferred,必須在它被建立時添加一個 cancel 回調.

那麼這個新的回調須要作什麼呢? 參考一下關於 callLater 方法的 文檔. 它的返回值是另外一個實現了 IDelayedCall 的對象,用 cancel 方法咱們能夠阻止延遲的調用被執行.

這很是簡單,更新後的代碼參見 deferred-cancel/defer-cancel-11.py.全部相關變化都在 get_poem 函數中:

def get_poem():
    """Return a poem 5 seconds later."""

def canceler(d):
    # They don't want the poem anymore, so cancel the delayed call
    delayed_call.cancel()

    # At this point we have three choices:
    #   1. Do nothing, and the deferred will fire the errback
    #      chain with CancelledError.
    #   2. Fire the errback chain with a different error.
    #   3. Fire the callback chain with an alternative result.

d = Deferred(canceler)

from twisted.internet import reactor
delayed_call = reactor.callLater(5, send_poem, d)

return d

在這個新版本中,咱們保存 callLater 的返回值以便可以在 cancel 回調中使用. cancel 回調的惟一工做是調用 delayed_call.cancel(). 可是正如以前討論的,咱們能夠選擇激發自定義的 deferred. 最新版本的程序產生以下輸出:

get_poem failed: [Failure instance: Traceback (failure with no frames): <class 'twisted.internet.defer.CancelledError'>:
]

正如你看到的, deferred 被取消了而且異步操做被真正地終止了(咱們看不到 send_poem 的輸出了).

詩歌代理 3.0

正如在簡介中所討論,詩歌代理服務器是實現取消的很好的候選者,由於這可讓咱們取消詩歌下載若是事實證實沒有人想要它(如客戶端已經在咱們發送詩歌前關閉了鏈接).版本 3.0的代理位於 twisted-server-4/poetry-proxy.py,實現了 deferred 取消. 變化首先位於 PoetryProxyProtocol:

class PoetryProxyProtocol(Protocol):

    def connectionMade(self):
        self.deferred = self.factory.service.get_poem()
        self.deferred.addCallback(self.transport.write)
        self.deferred.addBoth(lambda r: self.transport.loseConnection())

   def connectionLost(self, reason):
       if self.deferred is not None:
           deferred, self.deferred = self.deferred, None
           deferred.cancel() # cancel the deferred if it hasn't fired

你能夠與 舊版本 對比一下.兩個主要的變化是:

1. 保存咱們從 get_poem 獲得的 deferred,以便以後在須要時取消它.

2. 當鏈接關閉時取消 deferred.注這個操做一樣會取消 deferred 當咱們實際獲得詩歌以後,但正如前例所發現的,取消一個被激發的 deferred 不會有任何效果.

如今咱們須要確保取消 deferred 將實際終止詩歌的下載. 因此咱們須要改變 ProxyService:

class ProxyService(object):

    poem = None # the cached poem

    def __init__(self, host, port):
        self.host = host
        self.port = port

    def get_poem(self):
        if self.poem is not None:
           print 'Using cached poem.'
           # return an already-fired deferred
           return succeed(self.poem)

        def canceler(d):
            print 'Canceling poem download.'
            factory.deferred = None
            connector.disconnect()

        print 'Fetching poem from server.'
        deferred = Deferred(canceler)
        deferred.addCallback(self.set_poem)
        factory = PoetryClientFactory(deferred)
        from twisted.internet import reactor
        connector = reactor.connectTCP(self.host, self.port, factory)
        return factory.deferred

    def set_poem(self, poem):
        self.poem = poem
        return poem

一樣,能夠與 舊版本 對比一下. 這個類具備一些新的變化:

1. 咱們保存 reactor.connetTCP 的返回值,一個 IConnector 對象.咱們可使用這個對象上的 disconnect 方法關閉鏈接.

2. 咱們建立帶 canceler 回調的 deferred.那個回調是一個閉包,它使用 connector 關閉鏈接. 但首先須設置 factory.deferred 屬性爲 None. 不然,工廠會以 "鏈接關閉"錯誤回調激發 deferred 而不是以 CancelledError 激發. 因爲deferred 已經被取消了, 以 CancelledError 激發更加合適.

你一樣會注意到咱們是在 ProxyService 中建立 deferred 而不是 PoetryClientFactory. 因爲 canceler 回調須要獲取IConnector 對象, ProxyService 成爲最方便建立 deferred 的地方.

同時,就像咱們以前的例子, canceler 回調做爲一個閉包實現.閉包看起來在取消回調的實現上很是有用.

讓咱們試試新的代理.首先啓動一個慢服務器.它須要很慢以便咱們有時間取消:

python blocking-server/slowpoetry.py --port 10001 poetry/fascination.txt

如今能夠啓動代理(記住你須要Twisted 10.1.0):

python twisted-server-4/poetry-proxy.py --port 10000 10001

如今咱們能夠用任何客戶端從代理下載一首詩,或者僅使用 curl:

curl localhost:10000

幾秒鐘後,按 Ctrl-C 中止客戶端或者 curl 進程. 在終端運行代理你將看到以下輸出:

Fetching poem from server.
Canceling poem download.

你應該看到慢服務器已經中止了向輸出打印它所發送詩歌的片斷,由於咱們的代理掛了.

你能夠屢次啓動和中止客戶端來證明每一個下載每次都被取消了.可是若是你讓整首詩運行完,那麼代理將緩存它而且在此以後當即發送它.

另外一個難點

以上咱們曾不止一次說取消一個已經激發的 deferred 是沒有效果的.然而,這不是十分正確.在 :doc:`p13` 中,咱們學習了附加給一個 deferred 的回調和錯誤回調也可能返回另外一個 deferred.在那種狀況下,原始的(外層) deferred 暫停執行它的回調鏈而且等待內層 deferred 激發(參見 `figure28`_).

如此, 即便一個 deferred 激發了發出異步請求的高層代碼,它也不能接收到結果,由於在等待內層 deferred 完成以前回調鏈暫停了. 因此當高層代碼取消這個外部 deferred 時會發生什麼狀況呢? 在這種狀況下,外部 deferred 不只僅是取消它本身(它已經激發了);相反地,這個 deferred 取消內部的 deferred.

因此當你取消一個 deferred 時,你可能不是在取消主異步操做,而是一些其餘的做爲前者結果所觸發的異步操做.呼!

咱們能夠用一個例子來講明.考慮代碼 deferred-cancel/defer-cancel-12.py:

from twisted.internet import defer

def cancel_outer(d):
    print "outer cancel callback."

def cancel_inner(d):
    print "inner cancel callback."

def first_outer_callback(res):
    print 'first outer callback, returning inner deferred'
    return inner_d

def second_outer_callback(res):
    print 'second outer callback got:', res

def outer_errback(err):
    print 'outer errback got:', err

outer_d = defer.Deferred(cancel_outer)
inner_d = defer.Deferred(cancel_inner)

outer_d.addCallback(first_outer_callback)
outer_d.addCallbacks(second_outer_callback, outer_errback)

outer_d.callback('result')

# at this point the outer deferred has fired, but is paused
# on the inner deferred.

print 'canceling outer deferred.'
outer_d.cancel()

print 'done'

在這個例子中,咱們建立了兩個 deferred, outer 和 inner,而且有一個外部回調返回內部 deferred. 首先,咱們激發外部deferred,而後取消它. 輸出結果以下:

first outer callback, returning inner deferred
canceling outer deferred.
inner cancel callback.
outer errback got: [Failure instance: Traceback (failure with no frames): <class 'twisted.internet.defer.CancelledError'>:
]
done

正如你看到的,取消外部 deferred 並無使外部 cancel 回調被激發. 相反,它取消了內部 deferred,因此內部 cancel 回調被激發了,以後外部錯誤回調收到 CancelledError (來自內部 deferred).

你可能須要仔細看一看那些代碼,而且作些變化看看如何影響結果.

討論

取消 deferred 是很是有用的操做,使咱們的程序避免去作不須要的工做. 然而正如咱們看到的,它可能有一點點棘手.

須要明白的一個重要事實是取消一個 deferred 並不意味着取消了它後面的異步操做.事實上,當寫這篇文章時,不少 deferreds並不會被真的"取消",由於大部分Twisted代碼寫於Twisted 10.1.0以前而且尚未被升級.這包括不少Twisted自己的APIs！檢查文檔或源代碼去發現"取消 deferred"是否真的取消了背後的請求,仍是僅僅忽略它.

第二個重要事實是從你的異步APIs返回的 deferred 並不必定在完整意義上可取消. 若是你但願在本身的程序中實現取消,你應該先研究一下Twisted源代碼中的許多例子. Cancellation 是一個暫新的特性,因此它的模式和最好實踐還在制定當中.

展望將來

如今咱們已經學習了關於 Deferreds 的方方面面以及Twisted背後的核心概念. 這意味着咱們沒什麼須要介紹的了,由於Twisted的其他部分主要包括一些特定的應用,如網絡編程或異步數據庫處理.故而,在 接下來 的部分中,咱們想走點彎路,看看其餘兩個使用異步I/O的系統跟Twisted有何理念類似之處.以後,在尾聲中,咱們會打個包而且建議一些幫助你繼續學習Twisted的方法.

參考練習

1. 你知道你能夠用多種方式拼寫"cancelled"嗎? 真的. 這取決於你的心情.

2. 細讀 Deferred 類的源代碼,關注 cancellation 的實現.

3. 在Twisted 10.1.0的 源碼 中找具備取消回調的 deferred 的例子.研究它們的實現.

4. 修改咱們詩歌客戶端中 get_poetry 方法返回的 deferred, 使其可取消.

5. 作一個基於 reactor 的例子展現取消外部 deferred,它被內層 deferred 暫停了.若是使用 callLater 你須要當心選擇延遲時間,以確保外層 deferred 在正確的時刻被取消.

6. 找一個 Twisted 中還不支持"本質上取消操做"的異步API,爲它實現本質取消. 並向 Twisted項目 提交一個 補丁.不要忘記單元測試!