Introduction

twisted 是一個正在發展的項目,twisted 的開發者們會添加一些新的特色或者擴展舊的.隨著twisted 10.1.0 的發布,開發者們增加了一個新的功能–取消,這個就是我們今天要講的內容.
異步的程序把request 和response 進行了解耦,于是增加了一個新的可能:在發送一個請求和得到返回結果之間你可以決定是否你還要返回結果.細想一下第十四部分的poetry proxy server.下面是這個代理怎么工作的(第一次沒有緩存的時候):

????一個請求進來
????這個proxy 連接真正的server去獲取詩
????一但獲取到完整的詩,發送給client


一切工作的很好,但是假如client在得到詩之前掛起了怎么辦?這樣的話我們的proxy 就會被卡住.在出現這種情況的時候我們最好的辦法就是關掉連接.

回想一下圖片十五,一個描述了同步程序中概念上的流程控制的圖表.在這個圖片中我們可以看到函數調用往下走,異常往上走.假如我們想取消一個同步的調用,程序的流程會和函數調用的方向是一樣的,從高層代碼到底層代碼.圖片三十八描述了這個過程:

圖片三十八

當然,在一個同步的程序中這個是不可能的因為高層的代碼不會恢復運行直到底層的代碼運行完,在這時就沒有什么可以取消的了.但是在一個異步的程序中,高層的代碼在底層的代碼運行完之前控制著程序,這就讓我們有能力取消底層次的請求.

在一個twisted 程序中,底層的請求是被deferred對象所體現的.在deferred 中信息流是向下的,從底層的代碼到高層的代碼,這個和同步程序中return 信息流是相同的.從twsited 10.1.0 開始,高層的代碼可以向底層的代碼傳送信息–它可以告訴底層的代碼它不再想要返回的結果了.看圖片三十九:

圖片三十九

Canceling Deferreds

先讓我們看一些例子來搞明白取消deferred 是怎樣工作的.記住,運行這一部分的代碼你需要將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,增加一個callback,然后在觸發這個deferred之前取消它.下面是輸出:

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

ok,取消一個deferred 觸發了errback,我們正常的callback 沒有被調用.并且出現的錯誤是twisted.internet.defer.CancelledError.讓我們在deferred-cancel/defer-cancel-2.py增加一個errback:

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): :
]
done

我們可以捕捉到這個CancelledError 就像捕捉其他的deferred 的錯誤一樣.
下面讓我們先觸發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 got: result
done

我們的callback 被正常的觸發了,然后我們的程序正常的結束.就好像cancel 沒有被調用一樣.看起來取消一個cancel 對一個已經觸發的deferred 沒有什么作用一樣.(繼續往下看).
如果我們取消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): :
]
done

很有意思,輸出和第二個例子中的一樣,就像根本沒有觸發deferred一樣.所以假如deferred 已經被取消了,再去觸發這個deferred 會沒有什么用.但是為什么d.callback(‘result’) 會拋出一個異常,因為你不能觸發一個已經被觸發過的deferred?

再看一下圖片三十九.觸發一個deferred 是底層代碼的任務,而取消一個deferred是高層代碼的的動作.觸發這個deferred 意味著”這里是你的結果”,而取消一個deferred 意味著”我不在需要它”.

cancel 方法主要做了兩件事情:

????告訴deferred 對象如果這個結果還沒有來的話,我們就不要了.并忽略后面的callback或者errback
????告訴底層的正在產生結果的代碼去做一些取消操作需要做的事情

?

我們取消deferred 的行為是自由的,在取消之后,我們就不會得到那些還沒有返回的結果(即使返回了我們也收不到).但是取消deferred 并不會最終取消異步的操作.取消一個異步的操作需要一系列的操作.你可能需要關閉一個連接,回滾一些事物,殺掉一個自進程,等等. 因為deferred 只是一個callback 的組織者,它怎么知道當你取消它的時候會采取什么操作? 或者,它怎樣把取消請求轉發給底層的代碼? 大聲跟著我喊:

用Callback

Canceling Deferreds, Really

好的,讓我們看一下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 callback,這個callback 是deferred創建的時候傳入的,不是后面加入的.這個callback 負責執行要取消這個deferred 所要進行的一系列操作(僅僅這個deferred 被取消的時候).這個canceller callback 是底層代碼返回deferred 不可缺少的一部分,不是用來接收deferred并添加callback 和errback 的高層代碼.

運行這個代碼,你會看到如下的輸出:

I need to cancel this deferred:
errback got: [Failure instance: Traceback (failure with no frames): :
]
done

你可以看到,這個canceller callback被傳入一個我們想取消的deferred.在canceller callback 中我們可以進行一些取消deferred要完成的一些附屬操作.canceller callback 是在errback之前觸發的.實際上,這時我們可以選擇用正常的結果還是錯誤的結果觸發deferred.所有的情況在 deferred-cancel/defer-cancel-6.py 和deferred-cancel/defer-cancel-7.py.

在把reactor引入進來之前,讓我們再做一個簡單的測試.我們創建了一個帶有canceller 的deferred,正常的觸發,并取消它,你可以在deferred-cancel/defer-cancel-8.py看到代碼.通過檢查這個例子的輸出,你可以看到在取消一個已經觸發的deferred之后并沒有觸發canceller callback.

我們上面看的例子都沒有做些實際的異步操作.下面讓我們創建一個簡單的觸發異步操作的程序,然后我們會弄明白怎樣讓這個異步操作變為可以取消的.看一下代碼 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 函數,get_poem利用callLater 方法五秒鐘之后返回一首詩.main 函數調用get_poem,并添加了一個callback/errback 對,并啟動reactor.
運行這個例子會有如下的輸出:

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

讓我們試著在這首詩返回之前取消deferred.我們加上一些代碼在兩秒鐘之后取消deferred.:

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): :
]
Sending poem

這個例子清晰的描述了取消一個deferred并不一定會取消底層的異步請求.在兩秒之后我們從errback看到了輸出了CancelledError,在五秒之后我們仍舊會看到send_poem 的輸出(但是接下來的callback沒有觸發).

這和例子四deferred-cancel/defer-cancel-4.py.取消deferred 導致最后的結果被忽略,但是真正意義上說并沒有完全終止操作.根據我們上面講的,要想徹底取消一個deferred你必須在deferred 創建的時候向其傳入一個cancel callback.

這個被傳入的callback需要做些什么呢?看一下callLater 的手冊.callLater 返回的值是一個實現了IDelayedCall,并帶有一個cancel 方法的對象,這個cancel 方法可以用來組織延遲的call 被執行.

這樣就比較簡單了,更新后的代碼在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 返回的值從而讓我們可以在callback中用到.我們的callback唯一需要做的是觸發delayed_call.cancel().根據我們上面講的,我們也可以選擇我們自己去觸發這個callback.最新版的例子的輸出為:

get_poem failed: [Failure instance: Traceback (failure with no frames): :
]

你可以看到,deferred 被取消了,異步的操作確實被拋棄了.

Poetry Proxy 3.0

根據我們所講的,poetry proxy server 是一個應用cancel deferred 的很好的場景,因為它可以允許我們放棄詩的下載假如沒有人想要詩的話. proxy Version 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

和老一版本的相比.主要的變化是:

????保存我們從get_poem 得到的deferred,以便我們可以以后用于取消deferred
????當連接關閉的時候取消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

和老版本的ProxyService,有如下變化:

????我們保存了reactor.connectTCP 的返回值,它是一個IConnector 對象,我們可以用它提供的disconnect方法來關閉連接
????我們創建了一個帶有canceler callback 的deferred.這個callback 用connector 來關閉連接.但首先要先設置factory.deferred 屬性為None.否則的話,這個factory可能會在這個deferred 被CancelledError觸發之前觸發這個deferred(發生連接錯誤的時候)


你可能已經注意到我們現在在ProxyService 創建這個deferred,而不是在PoetryClientFactory中.因為我們的canceler callback 需要操作IConnector 對象,ProxyService 就成了創建deferred 最方便的額地方.

在我們之前的的例子中,我們的canceler callback 都是作為一個閉包來實現的.閉包是非常有用的當我們要取消callback的時候.

讓我們試一下我們的proxy.首先開啟一個slow server.它需要足夠慢讓我們有時間來取消:

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

現在我們開啟proxy:

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

現在我們可以從proxy 下載詩了,可以用curl:

curl localhost:10000

幾秒之后,按下Ctrl-C停止client,在運行proxy 的終端中你可以看到如下的輸出:

Fetching poem from server.
Canceling poem download.

你可以看到我們的server 已經停止輸出了,因為我們的proxy已經掛起了.

One More Wrinkle

我們上面已經說過了取消一個已經觸發過的deferred 會沒有作用.這并不是十分準確.在第十三部分我們了解到被添加到deferred 上的callback 和errback 會返回deferred.在這種情況下,外部的deferred 暫停執行它的cllback 鏈并等待內部的deferred 觸發.

因此,即使一個deferred 觸發了發起異步請求的高層的代碼也不會接收到結果.因為這個callback 正在等待內部的deferred 實行完畢.如果高層的代碼取消了外部的deferred會發生什么? 在這種情況下外部的deffered 不會取消自己,相反的它會取消內部的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,一個內部的一個外部的.首先我們觸發外部的deferred,然后取消它.這個例子會有如下輸出:

first outer callback, returning inner deferred
canceling outer deferred.
inner cancel callback.
outer errback got: [Failure instance: Traceback (failure with no frames): :
]
done

你可以看到,取消外部的deferred 不會引起外部的cancel callback 觸發.相反的,它取消內部的deferred,所以內部deferred 的cancel callback會觸發,然后外部的errback 會接收到一個CancelledError.

Discussion

取消一個deferred 是一個非常有用的操作,允許我們的程序去避免無謂的工作.但是也有一些取巧.

一個非常重要的要記住的事實是取消一個異步操作并不會一定取消根本的異步操作.實際上,大多數deferred 并不會真正的取消,因為大多數的twisted 代碼還沒有更新.檢查文檔或者源代碼去看看是否取消deferred 會不會真正取消請求,或者只是忽略它.

第二個重要的事實是僅僅從異步的api 中返回一個deferred 并不會真正的取消deferred.假如你想在你的代碼中實現取消deferred,你需要在在源代碼中找更多的例子.

總結

以上是生活随笔為你收集整理的twisted系列教程十九–cancel deferred的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。