Warum erhöht multiprocessing.pool.map einen PicklingError (Encoding)?

Question

Warum unten laufen wird der Code bei der Verwendung von threads aber eine Ausnahme aus, wenn multiprocessing verwendet wird?

from multiprocessing import Pool 
from multiprocessing.dummy import Pool as ThreadsPool 
import urllib2 

urls = [ 
    'http://www.python.org', 
    'http://www.python.org/about/', 
    'http://www.python.org/doc/', 
    'http://www.python.org/download/'] 

def use_threads(): 

    pool = ThreadsPool(4) 
    results = pool.map(urllib2.urlopen, urls) 
    pool.close() 
    pool.join() 

    print [len(x.read()) for x in results] 

def use_procs(): 

    p_pool = Pool(4) 
    p_results = p_pool.map(urllib2.urlopen, urls) 
    p_pool.close() 
    p_pool.join() 

    print 'using procs instead of threads' 
    print [len(x.read()) for x in p_results] 

if __name__ == '__main__': 
    use_procs() 

Die Ausnahme ist

Traceback (most recent call last): 
    File "pools.py", line 39, in <module> 
    use_procs() 
    File "pools.py", line 31, in use_procs 
    p_results = p_pool.map(urllib2.urlopen, urls) 
    File "/usr/lib64/python2.7/multiprocessing/pool.py", line 250, in map 
    return self.map_async(func, iterable, chunksize).get() 
    File "/usr/lib64/python2.7/multiprocessing/pool.py", line 554, in get 
    raise self._value 
multiprocessing.pool.MaybeEncodingError: Error sending result: '[<addinfourl at 35286624 whose fp = <socket._fileobject object at 0x2198ad0>>]'. Reason: 'PicklingError("Can't pickle <type 'instancemethod'>: attribute lookup __builtin__.instancemethod failed",)' 

Ich weiß, es ist der Unterschied zwischen dem, wie Prozesse und Threads miteinander kommunizieren. Warum schlägt der Inhalt der Website pickle fehl? Wie kann ich die Codierung einstellen, um das zu beheben?

Answer 1

Das Problem isn 't ein Kodierungsfehler, es ist wegen eines Beizfehlers, da das Ergebnis urllib2.urlopen() returns ein nicht pickbares Objekt ist (ein _ssl._SSLSocket entsprechend dem etwas anderen Grund, der in der Fehlermeldung angezeigt wird, die ich mit deinem Code erhalte). Um dies zu umgehen, können Sie die Verwendung des zurückgegebenen Objekts auf den Subprozess selbst beschränken, indem Sie die Daten nach dem Öffnen der URL wie unten gezeigt lesen. Dies bedeutet wahrscheinlich, dass viel mehr Daten zwischen den Prozessen ausgetauscht werden müssen.

# Added. 
def get_data(url): 

    soc = urllib2.urlopen(url) 
    return soc.read() 

def use_procs(): 

    p_pool = Pool(4) 
# p_results = p_pool.map(urllib2.urlopen, urls) 
    p_results = p_pool.map(get_data, urls) 
    p_pool.close() 
    p_pool.join() 

    print 'using procs instead of threads' 
# print [len(x.read()) for x in results] 
    print [len(x) for x in p_results] 

Ausgang:

using procs instead of threads 
[49062, 41616, 40086, 101224] 

Answer 2

Wie bereits erwähnt - dieser Fehler wird ausgelöst, weil Sie versucht haben, Socket-Objekt zwischen Prozessen übergeben. Sie müssen Skriptlogik in etwa wie folgt ändern:

from multiprocessing.pool import Pool 
from multiprocessing.pool import ThreadPool 
import urllib2 

urls = [ 
    'http://www.python.org', 
    'http://www.python.org/about/', 
    'http://www.python.org/doc/', 
    'http://www.python.org/download/' 
] 

def worker(url): 
    return urllib2.urlopen(url).read() # string returned 

def use_threads(): 

    pool = ThreadPool(4) 
    results = pool.map(worker, urls) 
    pool.close() 
    pool.join() 

    print([len(x) for x in results]) 

def use_procs(): 

    p_pool = Pool(4) 
    p_results = p_pool.map(worker, urls) 
    p_pool.close() 
    p_pool.join() 

    print('using procs instead of threads') 
    print([len(x) for x in p_results]) 

if __name__ == '__main__': 
    use_procs() 

Durch die Art und Weise: Sie könnten Pool Fabrik tun und es holen Pool anstatt Code zu duplizieren in use_threads und use_procs:

from multiprocessing.pool import Pool 
from multiprocessing.pool import ThreadPool 
import urllib2 

urls = [ 
    'http://www.python.org', 
    'http://www.python.org/about/', 
    'http://www.python.org/doc/', 
    'http://www.python.org/download/' 
] 


def worker(url): 
    return urllib2.urlopen(url).read() 


def pool_factory(key, n): 
    if key == 'proc': 
     print('using procs instead of threads') 
     return Pool(n) 
    else: 
     return ThreadPool(n) 


def main(): 

    pool = pool_factory('proc', 4) # change `proc` to anything for using ThreadPool 
    results = pool.map(worker, urls) 
    pool.close() 
    pool.join() 
    print([len(x) for x in results]) 


if __name__ == '__main__': 
    main()