4

IPC 通信と HTTP を介した非同期 RESTful 通信の組み合わせにgrequestsとの両方を使用するアプリケーションがあります。multiprocessing.managers

のメソッドをgrequests使用すると、クラスとその派生物で使用されるモジュールが壊れているようです。gevent.monkeypatch_all()multiprocessing.connectionmultiprocessing.manager.SyncManager

これは明らかに孤立した問題ではありませんがmultiprocessing.connetion、たとえばmultiprocessing.pool、 を実装するすべてのユース ケースに影響します。

のコードをgevent/monkey.py詳しく調べたところ、stdlibsocketモジュールを と交換したgevent.socketことが破損の原因であることがわかりました。これは、関数gevent/monkey.pyの下の115 行目にあります。patch_socket()

def patch_socket(dns=True, aggressive=True):
    """Replace the standard socket object with gevent's cooperative sockets.
    ...
    _socket.socket = socket.socket # This line breaks multiprocessing.connection!
    ...

私の質問は、なぜこの swappage breakが発生するのか、stdlib のモジュールの代わりにmultiprocessing.connection使用することでどのような利点が得られるのかということです。つまり、モジュールにパッチを適用しないと、パフォーマンスがどの程度低下するのでしょうか?gevent.socketsocketsocket

トレースバック

Traceback (most recent call last):
  File "clientWithGeventMonkeyPatch.py", line 49, in <module>
    client = GetClient(host, port, authkey)
  File "clientWithGeventMonkeyPatch.py", line 39, in GetClient
    client.connect()
  File "/usr/lib/python2.7/multiprocessing/managers.py", line 500, in connect
    conn = Client(self._address, authkey=self._authkey)
  File "/usr/lib/python2.7/multiprocessing/connection.py", line 175, in Client
    answer_challenge(c, authkey)
  File "/usr/lib/python2.7/multiprocessing/connection.py", line 414, in answer_challenge
    response = connection.recv_bytes(256)        # reject large message
IOError: [Errno 11] Resource temporarily unavailable

エラーを再現するコード

(ubuntu サーバー 11.10、python2.7.3、gevent、greenlet、および grequests がインストールされている場合)

manager.py

## manager.py
import multiprocessing
import multiprocessing.managers
import datetime


class LocalManager(multiprocessing.managers.SyncManager):
    def __init__(self, *args, **kwargs):
        multiprocessing.managers.SyncManager.__init__(self, *args, **kwargs)
        self.__type__ = 'LocalManager'

def GetManager(host, port, authkey):
    def getdatetime():
        return '{}'.format(datetime.datetime.now())

    LocalManager.register('getdatetime', callable = getdatetime)
    manager = LocalManager(address = (host, port), authkey = authkey)
    manager.start()

    return manager

if __name__ == '__main__':
    # define our manager connection parameters
    port = 55555
    host = 'localhost'
    authkey = 'auth1234'

    # start a manager
    man = GetManager(host, port, authkey)

    # wait for user input to shut down
    raw_input('return to shutdown')
    man.shutdown()

client.py

## client.py -- this one works
import time
import multiprocessing.managers

class RemoteClient(multiprocessing.managers.SyncManager):
    def __init__(self, *args, **kwargs):
        multiprocessing.managers.SyncManager.__init__(self, *args, **kwargs)
        self.__type__ = 'RemoteClient'

def GetClient(host, port, authkey):
    RemoteClient.register('getdatetime')
    client = RemoteClient(address = (host, port), authkey = authkey)
    client.connect()
    return client

if __name__ == '__main__':
    # define our client connection parameters
    port = 55555
    host = 'localhost'
    authkey = 'auth1234'

    # start a manager
    client = GetClient(host, port, authkey)
    print 'connected', client
    print 'client.getdatetime()', client.getdatetime()
    # wait a couple of seconds, then do it again
    time.sleep(2)
    print 'client.getdatetime()', client.getdatetime()

    # exit...

clientWithGeventMonkeyPatch.py

## clientWithGeventMonkeyPatch.py -- breaks, depending on patch_all() parameters        
import time
import multiprocessing.managers


# this part is copied from grequests
# bear in mind that it doesn't actually do anything in this module.
try:
    import gevent
    from gevent import monkey as curious_george
    from gevent.pool import Pool
except ImportError:
    raise RuntimeError('Gevent is required for grequests.')

# this line causes breakage of the multiprocessing.manager connection auth method:
# Monkey-patch. 
# patch_all() parameters with default values:  socket=True, dns=True, time=True, select=True, thread=True, os=True, ssl=True, aggressive=True

curious_george.patch_all(thread=False, select=False) # breaks
#~ curious_george.patch_all(thread=False, select=False, socket = False) # works!
#~ curious_george.patch_all(thread=False, select=False, socket = True, aggressive = True, dns = True) # same as (thread=False, select=False); breaks
#~ curious_george.patch_all(thread=False, select=False, socket = True, aggressive = True, dns = False) # breaks
#~ curious_george.patch_all(thread=False, select=False, socket = True, aggressive = False, dns = True) # breaks
#~ curious_george.patch_all(thread=False, select=False, socket = True, aggressive = False, dns = False) # breaks







class RemoteClient(multiprocessing.managers.SyncManager):
    def __init__(self, *args, **kwargs):
        multiprocessing.managers.SyncManager.__init__(self, *args, **kwargs)
        self.__type__ = 'RemoteClient'

def GetClient(host, port, authkey):
    RemoteClient.register('getdatetime')
    client = RemoteClient(address = (host, port), authkey = authkey)
    client.connect()
    return client

if __name__ == '__main__':
    # define our client connection parameters
    port = 55555
    host = 'localhost'
    authkey = 'auth1234'

    # start a manager
    client = GetClient(host, port, authkey)
    print 'connected', client
    print 'client.getdatetime()', client.getdatetime()
    # wait a couple of seconds, then do it again
    time.sleep(2)
    print 'client.getdatetime()', client.getdatetime()

    # exit...
4

1 に答える 1

6

socket モジュールにパッチを適用しないgeventと、ネットワーク操作をブロックしないという の機能が利用できなくなりgevent、最初から使用する利点のほとんどが利用できなくなります。

geventmultiprocessing互いにうまく動作するように実際には設計されていません-ほとんどのgevent場合、それを介してネットワーク接続を行っており、最高レベルのPythonソケットインターフェイスをバイパスしていないことを前提としています(バイパスしますmultiprocessing)。

于 2013-02-06T19:10:12.003 に答える