Deferred リファレンス

この文書は twisted.internet.defer.Deferred の振舞いと関数から Deferred が返されたときの扱い方の解説です。

なおこの文書は Twisted フレームワークを構成する基本的な手法、プログラムをブロックさせず、スレッドも使わずに関数を即座にリターンさせ、データが到着した時点でコールバックチェーンの実行を開始する非同期処理、コールベースのプログラミングについて理解していることを前提に書かれています。

上記内容に関する解説:

• Twisted による非同期プログラミング

この文書を読むことにより、Twisted の最も基本的な API や Twisted を使って Deferred を返すコードを扱えるようになるはずです。

• 関数が呼び出しの結果として Deferred を返すことにより、どんなことが可能になるのか
• Deferred コードの中で確実なとエラーハンドリングをおこなう方法

非同期プログラミングについて不案内な人は、まず先に挙げた文書の Deferred の項を読むことを強くお薦めします。Deferred がなぜ存在するのか、その理由がわかるはずです。

コールバック

twisted.internet.defer.Deferred はデータがある時点で必ず得られることを確約するものです。Deferred にはコールバック関数を登録でき、データが到着した時点でコールバックが呼び出されます。 Deferred にはエラー発生時用のコールバックも登録可能で、デフォルトではログに出力するようになっています。Deferred という仕組みはアプリケーションプログラマに対し、あらゆる種類のブロッキング、遅延が発生する処理用のインターフェースを提供するものです。

from twisted.internet import reactor, defer

def getDummyData(x):
    """
    この関数は結果の遅延をシミュレートするためのダミーであり、
    Deferred を返します。Deferred は遅延して届く結果と共に
    処理されます。関数の処理自体は特に意味はありません。
    """
    d = defer.Deferred()
    # 結果の遅延シミュレーション。2秒後に Deferred の処理を開始し、そ
    # の際結果データとして x * 3 を渡すよう reactor に指示しています。
    reactor.callLater(2, d.callback, x * 3)
    return d

def printData(d):
    """
    データ・ハンドリング用にコールバックとして登録される関数:
    受け取ったデータを print します。
    """
    print d

d = getDummyData(3)
d.addCallback(printData)

# プロセスの終了処理。reactor に対し4秒後に自分自身を停止するよう
# 指示しています。
reactor.callLater(4, reactor.stop)
# Twisted reactor (イベントループ・ハンドラ)の開始
reactor.run()

複数のコールバック登録

Deferred には複数のコールバックを登録できます。Deferred のコールバックチェーン先頭のコールバックは、結果データを引数にして呼び出されます。二つ目のコールバックは最初のコールバックの結果を引数にして呼び出されます。 三つ目は...(以降同様)。なぜこのような仕様になっているのでしょう？ twisted.enterprise.adbapi が返す Deferred — SQL 問合せに対する結果が返ってくる — 場合を例に見てみましょう。

from twisted.internet import reactor, defer

class Getter:
    def gotResults(self, x):
        """
        Deferred はエラー状態を通知する仕組みを提供しています。
        この場合、奇数をエラーとしています。

        結果が期待したものであったかどうかをチェックし、コールバックと
        エラーバックのどちらを実行すべきかを選択するという、より複雑な
        例がこの関数です。

        """
        if x % 2 == 0:
            self.d.callback(x*3)
        else:
            self.d.errback(ValueError("You used an odd number!"))

    def _toHTML(self, r):
        """
        HTML への変換処理関数。

        getDummyData によってコールバックチェーンに登録されます。これは
        コールバックが自分自身の結果を次のコールバックに渡す例です。
        """
        return "Result: %s" % r

    def getDummyData(self, x):
        """
        Defered はコールバックを数珠つなぎにして登録できるようになって
        います。この例の場合、gotResults の結果は printData に渡される
        前に _toHTML を経由するようになっています。
        
        この関数は結果の遅延をシミュレートするダミーです。本当の非同期
        処理をしているのではなく、callLater を使ってわざと遅延を発生さ
        せています。
        """
        self.d = defer.Deferred()
        # gotResults を2秒後に実行するよう reactor にスケジュール指示
        # をすることにより、結果の遅延をシミュレート。
        reactor.callLater(2, self.gotResults, x)
        self.d.addCallback(self._toHTML)
        return self.d

def printData(d):
    print d

def printError(failure):
    import sys
    sys.stderr.write(str(failure))

# 以下のコードはエラーメッセージを表示します。
g = Getter()
d = g.getDummyData(3)
d.addCallback(printData)
d.addErrback(printError)

# 以下のコードは "Result: 12" を表示します。
g = Getter()
d = g.getData(4)
d.addCallback(printData)
d.addErrback(printError)

reactor.callLater(4, reactor.stop); reactor.run()

図による解説

1. 要求側のメソッド(data sink)はデータを要求し、その結果として Deferred オブジェクトを受け取ります。
2. 要求側のメソッドは Defered にコールバックを登録します。

1. データが到着すると、それは Deferred オブジェクトに渡され、処理が成功したときは .callback(result)、失敗したときは .errback(failure) が実行されます。エラーバックの引数 failure は通常 twisted.python.failure.Failure のインスタンスです。
2. Deferred オブジェクトは予め登録されたコールバック(エラーバック)を result または failure を引数にして実行します。実行の際は以下のルールに従いコールバックのチェーンを辿っていきます。
   • コールバックの実行結果は常に、チェーンを構成する次のコールバックの第1引数として渡されます。
   • コールバック実行時に例外が発生すると、エラーバックへ切り替わります。
   • エラーバックで failure が捕捉されない場合はさらにエラーバックのチェーンを辿ります。つまりこの流れは except: 文の非同期処理版とも言えます。
   • エラーバックが例外を発生させたり twisted.python.failure.Failure を返さなかった場合は、再びコールバックへ切り替わります。

エラーバック

Deferred のエラーハンドリングは Python の 例外ハンドリングに準じたモデルとなっています。例外発生のない場合、前述の通りコールバックだけが順次実行されます。

(データベースのクエリでエラーが発生した場合など)コールバックではな くエラーバックが呼び出される場合、twisted.python.failure.Failure が第1引数として渡されます(エラーバックもコールバック同様、複数登録可能です)。エラーバックは、一般の Python コードにおける except ブロックのようなものと考えてください。

except ブロック内で明示的にエラーを raise しない限り、 Exception は捕捉され例外の伝播は停止、通常の処理が継続されます。エラーバックも同様で、明示的にFailure を return したり例外を発生させない限りエラーの伝播は停止、その時点からコールバックの処理が継続されます(この場合、エラーバックの戻り値がコールバックの引数に用いられます)。エラーバックが Failure を返したり例外を発生させたときは、次のエラーバックへ処理が引き継がれます。チェーンの残りも同様です。

注意: エラーバックが何も返さなかった場合、つまり事実上 None を返したときは、エラーバックの後に再びコールバックの実行が継続されることになります。想定した動作と異なることがないよう注意してください。エラーバックが返すのは(おそらくは最初に渡されたそのままの) Failure または次のコールバックにとってちゃんとした意味のある値でなければなりません。

twisted.python.failure.Failure インスタンスは trap という有用なメソッドを備えています。次の例外処理と同等の内容を実行するものです。

try:
    # 例外が発生するかもしれない処理
    cookSpamAndEggs()
except (SpamException, EggException):
    # SpamExceptions と EggExceptions を捕捉
    ...

同等処理のエラーバック版は次のようになります。

def errorHandler(failure):
    failure.trap(SpamException, EggException)
    # SpamExceptions と EggExceptions を捕捉

d.addCallback(cookSpamAndEggs)
d.addErrback(errorHandler)

Failure にカプセル化されたエラーにマッチさせる引数が省略された状態で failure.trap が呼ばれた場合、エラーが再発生します。

もうひとつ潜在的にハマりやすいネタがあります。 twisted.internet.defer.Deferred.addCallbacks というメソッドがあり、その動作は addCallback に addErrback を続けて書いた場合と似ているのですが、まったく同じではないのです。特に次の二つのケースに注意してください。

# ケース1
d = getDeferredFromSomewhere()
d.addCallback(callback1)       # A
d.addErrback(errback1)         # B
d.addCallback(callback2)       
d.addErrback(errback2)        

# ケース2
d = getDeferredFromSomewhere()
d.addCallbacks(callback1, errback1)  # C
d.addCallbacks(callback2, errback2)

仮に callback1 でエラーが発生したとします。ケース1では errback1 が failure を引数に呼び出されます。しかしケース2では errback2 が呼び出されるのです。コールバックとエラーバックの扱いには注意が必要です。

実際にこれらがどのように解釈されるかといえば、ケース1の場合、"A" はgetDeferredFromSomewhere が成功した場合のデータを処理し、 "B" は チェーン上位のソースや 'A' で発生したエラーすべてを処理します。しかしケース2の "C" にある errback1 は getDeferredFromSomewhere で発生したエラーだけを処理し、callback1 のエラーは処理しないのです。

捕捉されなかったエラーの処理

エラーが捕捉されなかった場合、次のエラーバックがあればそれが呼び出されますが、ない場合は Deferred と共にエラーもガーベジコレクトされ、 Twisted はエラーのトレースバックをログに出力します。つまりエラーバックを一切書かなくてもエラーのログだけは残るわけです。しかし、コードの他の部分に Defered への参照が残っているとガーベジコレクトされないため、エラーを確認できない場合もあり得ます(この場合同時に、呼ばれるはずのコールバックが呼ばれないという奇妙な現象に悩まされることになります)。ガーベジコレクトが確実におこなわれる場面だと断言できない限り、コールバックの後には常にエラーバックを書くべきです。

# エラーを確実にログに残す方法
from twisted.python import log
d.addErrback(log.err)

同期的な結果と非同期的な結果の両方を処理する

アプリケーションによっては、同期的な関数と非同期的な関数が混在している場合があります。ユーザ認証用の関数を考えてみましょう。ユーザが認証済みかどうかをメモリ上の情報をチェックするだけで良いのであればすぐに結果を返せます。しかしネットワークを通じて問い合わせをおこなう場合は Deferred を返し、データが届いた時点で処理が開始されるようにする必要があります。この場合、認証関数を呼び出す側は戻り値としてが通常の値と Deferred 両方を受け入れるように作成しておく必要があります。

次の関数 authenticateUser はユーザ認証をおこなうために別の関数 isValidUser を呼び出しています。

def authenticateUser(isValidUser, user):
    if isValidUser(user):
        print "User is authenticated"
    else:
        print "User is not authenticated"

しかしこの例は isValidUser が同期的に通常の値を返すことを前提にしており、非同期的な認証により Deferred が返ってくることは想定していません。通常の値と Deferred の両方を受け入れるように変更して同期的な isValidUser と 非同期的な isValidUser のどちらでも処理できるようにする方法のほか、同期的な関数にも必ず Deferred を返させるという方法が考えられます。この項ではライブラリ関数(authenticateUser)やアプリケーションコードが同期的なものか非同期的なものかわからない場合の対応方法について説明します。

ライブラリコード内の Deferred の扱い

次は authenticateUser 対し同期的に結果を返すユーザ認証関数の例です

def synchronousIsValidUser(user):
    '''
    Return true if user is a valid user, false otherwise
    '''
    return user in ["Alice", "Angus", "Agnes"]

一方こちらの asynchronousIsValidUser は非同期的な関数で、 Deferred を返します。

from twisted.internet import reactor

def asynchronousIsValidUser(d, user):
    d = Deferred()
    reactor.callLater(2, d.callback, user in ["Alice", "Angus", "Agnes"])
    return d

先の authenticateUser では isValidUser が同期的な関数だという前提になっていました。しかし isValidUser が同期、非同期のどちらでも処理できるように変更しなければなりません。 この場合 isValidUser の呼び出しに maybeDeferred を使うことで、isValidUser の戻り値が Deferred であっても、同期的に値を返しても対応できるようになります。

from twisted.internet import defer

def printResult(result):
    if result:
        print "User is authenticated"
    else:
        print "User is not authenticated"

def authenticateUser(isValidUser, user):
    d = defer.maybeDeferred(isValidUser, user)
    d.addCallback(printResult)

これで isValidUser の実態 が synchronousIsValidUser と asynchronousIsValidUser のどちらでも対応できるようになりました。

synchronousIsValidUser が Deferred を返すように変更する方法については Deferred の作り方を参照してください。

DeferredList

複数のイベント通知を受け取る場合、個別にそれぞれの通知を受けるのではなく、まとめて受け取りたいことがあります。たとえばネットワーク接続のリストがあり、そのすべてが close するのを待っている場合などです。このような処理を可能にするのが twisted.internet.defer.DeferredList です。

複数の Deferred をまとめて DeferredList を作成するには、Deferred のリストを引数として渡します。

# DeferredList の作成
dl = defer.DeferredList([deferred1, deferred2, deferred3])

DeferredList は addCallbacks など、通常のDeferred と同じように扱えます。DeferredList はすべての Deferred が完了した時点でコールバックを実行します。コールバックは Deferred が返した結果のリストを引数にして実行されます。以下をご覧ください。

def printResult(result):
    print result
deferred1 = defer.Deferred()
deferred2 = defer.Deferred()
deferred3 = defer.Deferred()
dl = defer.DeferredList([deferred1, deferred2, deferred3])
dl.addCallback(printResult)
deferred1.callback('one')
deferred2.errback('bang!')
deferred3.callback('three')
# この後 dl がコールバックを実行し以下の内容が出力されます。
#     [(1, 'one'), (0, 'bang!'), (1, 'three')]
# (注: 1と0は それぞれ defer.SUCCESS、defer.FAILURE の値です)

通常の DeferredList はエラーバックを呼び出さないことに注意してください。

def printResult(result):
    print result
def addTen(result):
    return result + " ten"

# DeferredList の作成前に Deferred に コールバックを登録
deferred1 = defer.Deferred()
deferred2 = defer.Deferred()
deferred1.addCallback(addTen)
dl = defer.DeferredList([deferred1, deferred2])
dl.addCallback(printResult)
deferred1.callback("one") # addTen の実行後 DeferredList の完了がチェックされ、 "one ten" が保存される
deferred2.callback("two")
# この後 dl はコールバックを実行、次の内容が表示される
#     [(1, 'one ten'), (1, 'two')]

# DeferredList を作成した後で Deferred に コールバックを登録
deferred1 = defer.Deferred()
deferred2 = defer.Deferred()
dl = defer.DeferredList([deferred1, deferred2])
deferred1.addCallback(addTen) # DeferredList が引数を受け取った *後* に実行される
dl.addCallback(printResult)
deferred1.callback("one") # まず DeferredList の完了がチェックされ、"one" を保存、その後で addTen が実行される
deferred2.callback("two")
# この後 dl はコールバックを実行、次の内容が表示される
#     [(1, 'one), (1, 'two')]

そのほかの挙動について

DeferredList にはその振舞いを変える三つのキーワード引数、 fireOnOneCallback、fireOnOneErrback そして consumeErrors を渡すことができます。 fireOnOneCallback がセットされた場合、 DeferredList の各 Deferred のコールバックが実行される前に、セットされたコールバックが実行されます。 fireOnOneErrback も同様で、各 Deferred のエラーバックが呼び出される前に、セットされたエラーバックが呼び出されます。DeferredList は通常の Deferred と同様一回完結です。実行後はそれ以上何もしません(この場合各 Defered の結果は無視されます)。

fireOnOneErrback オプションは、すべての Deferred の成功を待っていて、もしひとつでもエラーになったときは即座に通知が欲しいケースなどで役に立ちます。

キーワード引数 consumeErrors は DeferredList 内の各 Deferred のコールバックチェーンで発生したエラーの伝播を停止します(ただし DeferredList 作成自体が各 Deferred のコールバックやエラーバックの結果に影響を及ぼすことはありません)。DeferredList でエラー伝播を停止することで、 特別なエラーバックを追加せずに Unhandled error in Deferred 警告の出力を抑制することができます。1のことです—先に述べた通り、DeferredList に登録後の Deferred にコールバックを追加する方法は混乱をまねくため避けるべきです。

クラスの概要

これは関数から返されるオブジェクトとして Deferred を見た場合の API 概要です。 Deferred クラスの docstring の代用品ではなく、使用する上でのガイドラインとなるように意図して書かれたものです。

Deferred を作成する立場から見た概要は Deferred の作り方に書かれています。

コールバック関数の基本

• addCallbacks(self, callback[, errback, callbackArgs, errbackArgs, errbackKeywords, asDefaults])

  Deferred と情報をやり取りするために使うメソッドです。このメソッドの呼び出しにより、Deferred に処理が移ったとき作成されるコールバックのリストに、一組のお互いにパラレルなコールバック/エラーバックが追加されます(上図参照)。addCallbacks で追加するコールバックのシグニチャは myMethod(result, *methodArgs, **methodKeywords) のようにする必要があります。コールバックがコールバックスロットに渡されると、タプルに入ったすべての引数 callbackArgs は *methodArgs の形でコールバック関数に渡されます。

  addCallbacks から派生した有用なメソッドがいくつかあります。ここでそれぞれの詳細は述べませんが、簡潔なコードを書く上でこれらの内容は必ず把握しておく必要があります。

  • addCallback(callback, *callbackArgs, **callbackKeywords)

    処理チェーン中の次の位置にコールバックを追加します。エラーが発生時に後の処理に影響しないよう、エラーバックとして単に第1引数を返すだけの関数が同時に登録されます。

    addCallback (単数形) は コールバックに渡す引数をそのままの形ですべて受け取ることができますが、addCallbacks (複数形) には引数をタプルとして渡す必要があることに注意してください。addCallbacks (複数形)は引数がコールバックのものなのか、エラーバックのものなのか判別する必要があるため、それぞれの引数をタプルとしてもらう必要があるのです。addCallback (単数形) に渡されるのは当然コールバック用の引数だけですから、Python の文法 * と ** を使ってすべての引数をそのまま受け取ることができます。

  • addErrback(errback, *errbackArgs, **errbackKeywords)

    処理チェーン中の次の位置にエラーバックを追加します。成功時の後の処理に影響しないよう、コールバックとして単に第1引数を返すだけの関数が同時に登録されます。

  • addBoth(callbackOrErrback, *callbackOrErrbackArgs, **callbackOrErrbackKeywords)

    処理チェーン中の次に位置するコールバックとエラーバックとして、同じ関数を登録します。このメソッドを使うときは第1引数のタイプが確定できないことに注意してください(訳注: つまりエラーのときは failure が渡されます)。このメソッドは finally: スタイルブロックのようにして使います。

Deferred のチェーン化

Deferred が別の Deferred の結果を待たなければならないことがありますが、この場合は addCallbacks で追加したコールバックが返す Deferred をそのまま返すだけで OK です。もう少し詳しく説明しましょう。Deferred A に A.addCallbacks を使って登録したメソッドが Deferred B を返す場合、B のコールバックが実行されるまで Deferred A の処理チェーンは一時停止されます。そして A のチェーンの次のコールバックには、B のチェーン最後のコールバックの結果が渡されます。

この説明は難しく感じるかもしれませんが、実際にこのような処理が必要な場面になればすぐに理解できて、なぜこのようになっているのかわかるはずです。Deferred のチェーン化を手動でおこなうためのメソッドも用意されています。

• chainDeferred(otherDeferred)

  この Deferred の処理チェーンの末尾に otherDeferred を追加します。self.callback が呼び出されると、この処理チェーンの結果が otherDeferred.callback に渡されます。コールバックチェーンに後からコールバックを追加した場合 otherDeferred には反映されないので注意してください。

  これは self.addCallbacks(otherDeferred.callback, otherDeferred.errback) のように書くこともできます。

関連文書

1. Deferred の作り方: Deferred を返す非同期関数の書き方。

Footnotes

1. もちろん後続のコールバックで新たなエラーが発生しない場合に限ります。