ツイスト:遅延コールバックを遅延スレッドに渡すと、スレッドが突然ブロックされるのはなぜですか?

Question

私が取り組んでいるScrapyプロジェクトで設定しようとしている永続メッセージキューにtxredis（redis用のノンブロッキングツイストAPI）を使用しようとしましたが失敗しました。クライアントはブロックしていませんでしたが、リアクター ループ内の 1 つのイベントであるはずのものが数千のステップに分割されていたため、予想よりもはるかに遅くなっていることがわかりました。

そこで代わりに、redis-py (通常のブロッキング ツイスト API) を利用して、呼び出しを遅延スレッドでラップしてみました。これはうまく機能しますが、さらに高速化するために接続プーリングを設定したいので、redis を呼び出すときに内部遅延を実行したいと考えています。

以下は、私のユースケースを説明するために、遅延スレッドのツイストドキュメントから抜粋したサンプルコードの私の解釈です。

#!/usr/bin/env python
from twisted.internet import reactor,threads
from twisted.internet.task import LoopingCall
import time

def main_loop():
    print 'doing stuff in main loop.. do not block me!'


def aBlockingRedisCall():
    print 'doing lookup... this may take a while'
    time.sleep(10)
    return 'results from redis'

def result(res):
    print res

def main():
    lc = LoopingCall(main_loop)
    lc.start(2)
    d = threads.deferToThread(aBlockingRedisCall)
    d.addCallback(result)
    reactor.run()

if __name__=='__main__':
    main()

そして、これが遅延スレッドのコードをブロックする接続プーリングの変更です。

#!/usr/bin/env python
from twisted.internet import reactor,defer
from twisted.internet.task import LoopingCall
import time

def main_loop():
    print 'doing stuff in main loop.. do not block me!'

def aBlockingRedisCall(x):
    if x<5: #all connections are busy, try later
        print '%s is less than 5, get a redis client later' % x
        x+=1
        d = defer.Deferred()
        d.addCallback(aBlockingRedisCall)
        reactor.callLater(1.0,d.callback,x)
        return d

    else: 
        print 'got a redis client; doing lookup.. this may take a while'
        time.sleep(10) # this is now blocking.. any ideas?
        d = defer.Deferred()
        d.addCallback(gotFinalResult)
        d.callback(x)
        return d

def gotFinalResult(x):
    return 'final result is %s' % x

def result(res):
    print res

def aBlockingMethod():
    print 'going to sleep...'
    time.sleep(10)
    print 'woke up'

def main():
    lc = LoopingCall(main_loop)
    lc.start(2)


    d = defer.Deferred()
    d.addCallback(aBlockingRedisCall)
    d.addCallback(result)
    reactor.callInThread(d.callback, 1)
    reactor.run()

if __name__=='__main__':
    main()

そこで私の質問は、私の変更により遅延スレッドがブロックされる理由を知っている人、および/またはより良い解決策を提案できる人はいますか?

Answer 1

まあ、言うのツイストドキュメントとして

  

のDeferredはコードをしません   魔法のようにブロックしていない

あなたは、このようなsleepとして、ブロックコードを使用しているときはいつでも、

は、新しいスレッドにそれを延期する必要があります。

#!/usr/bin/env python
from twisted.internet import reactor,defer, threads
from twisted.internet.task import LoopingCall
import time

def main_loop():
    print 'doing stuff in main loop.. do not block me!'

def aBlockingRedisCall(x):
    if x<5: #all connections are busy, try later
        print '%s is less than 5, get a redis client later' % x
        x+=1
        d = defer.Deferred()
        d.addCallback(aBlockingRedisCall)
        reactor.callLater(1.0,d.callback,x)
        return d

    else: 
        print 'got a redis client; doing lookup.. this may take a while'
        def getstuff( x ):
            time.sleep(3)
            return "stuff is %s" % x

        # getstuff is blocking, so you need to push it to a new thread
        d = threads.deferToThread(getstuff, x)
        d.addCallback(gotFinalResult)
        return d

def gotFinalResult(x):
    return 'final result is %s' % x

def result(res):
    print res

def aBlockingMethod():
    print 'going to sleep...'
    time.sleep(10)
    print 'woke up'

def main():
    lc = LoopingCall(main_loop)
    lc.start(2)


    d = defer.Deferred()
    d.addCallback(aBlockingRedisCall)
    d.addCallback(result)
    reactor.callInThread(d.callback, 1)
    reactor.run()

if __name__=='__main__':
    main()

の場合にはRedisのAPIは、単に多くのスレッドでブロッキングAPIを呼び出すのではなく、twisted.webを使用して、それを書き換えることがより自然であるかもしれない非常に複雑ではありません。

Answer 2

すでにRedisの2.xの新しいプロトコルと機能をサポートしていツイストための最新のRedisのクライアントもありますあなたはdefinetelyそれを試してみる必要があります。それはtxredisapiと呼ばれています。

永続メッセージキューのために、私はRestMQをお勧めします。サイクロンとtxredisapiの上に構築されたのRedisベースのメッセージキューシステムます。

http://github.com/gleicon/restmqする

乾杯

Answer 3

これに関連して、次のような Twisted 専用に作成された Redis クライアントを使用すると、おそらく多くのことが得られるでしょう。 http://github.com/deldotdr/txRedis