Pythonの同時プログラミングを理解するだけで十分です|プロセス編
前節ではGIL(グローバル解釈ロック)の問題でマルチスレッドはマルチコアプロセッサを十分に利用できないと述べ,CPU計算型のタスクであればマルチプロセスモジュールmultiprocessingを用いるべきである.スレッド・ライブラリとはまったく異なりますが、2つのライブラリの構文とインタフェースは非常に似ています.Multiprocessingは各プロセスに個別のPython解釈器を与え,グローバル解釈ロックによる問題を回避する.しかし、あまり早く喜ばないでください.これからお話しするマルチプロセス間でどのように通信するかという問題に直面するからです.
まず、前節の例を単一プロセスとマルチプロセスに変更して、パフォーマンスを比較します.
import time
import multiprocessing
def profile(func):
def wrapper(*args, **kwargs):
import time
start = time.time()
func(*args, **kwargs)
end = time.time()
print 'COST: {}'.format(end - start)
return wrapper
def fib(n):
if n<= 2:
return 1
return fib(n-1) + fib(n-2)
@profile
def nomultiprocess():
fib(35)
fib(35)
@profile
def hasmultiprocess():
jobs = []
for i in range(2):
p = multiprocessing.Process(target=fib, args=(35,))
p.start()
jobs.append(p)
for p in jobs:
p.join()
nomultiprocess()
hasmultiprocess()
：
❯ python profile_process.py
COST: 4.66861510277
COST: 2.5424861908
， ， 2 fib(35)， 。 ，GIL 。
： CPU 。 hasmultiprocess ， Pool， :
from multiprocessing import Pool
pool = Pool(2)
pool.map(fib, [35] * 2)
map map ， 。
PS: ， multiprocessing.Pool Python MapReduce。 。
dummy
/ ， ， ， 。 ， multiprocessing.dummy ，「dummy」 「 」 ， ， 。 ：
multiprocessing.dummy replicates the API of multiprocess ing but is no more than a wrapper around the threading module.
！！！ CPU IO ， 2 ：
from multiprocessing import Pool
from multiprocessing.dummy import Pool
。 ， / 。
： CPU I/O ， ， 。
Pipe parmap
（IPC） rpc、socket、pipe（ ） （queue）。 3 。 ， ：
from multiprocessing import Process, Pipe
def f(conn):
conn.send(['hello'])
conn.close()
parent_conn, child_conn = Pipe()
p = Process(target=f, args=(child_conn,))
p.start()
print parent_conn.recv()
p.join()
Pipe 2 ：「 」 「 」。 hello ， ， parent_conn.recv()が されます.これにより、Python のデータ をマルチプロセス で することが に できます.しかし、xmlrpclibにシーケンス できないオブジェクトは、このように できないことを に する.
の で べたhasmultiprocess はPoolのmapメソッドを いており,まあまあだ.しかし、 のビジネスでは、 の のように です.
class CalculateFib(object):
@classmethod
def fib(cls, n):
if n<= 2:
return 1
return cls.fib(n-1) + cls.fib(n-2)
def map_run(self):
pool = Pool(2)
print pool.map(self.fib, [35] * 2)
cl = CalculateFib()
cl.map_run()
fib ， ：
❯ python parmap.py
Exception in thread Thread-1:
Traceback (most recent call last):
File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/threading.py", line 810, in __bootstrap_inner
self.run()
File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/threading.py", line 763, in run
self.__target(*self.__args, **self.__kwargs)
File "/Library/Python/2.7/site-packages/multiprocessing-2.6.2.1-py2.7-macosx-10.9-intel.egg/multiprocessing/pool.py", line 225, in _handle_tasks
put(task)
PicklingError: Can't pickle instancemethod'>: attribute lookup __builtin__.instancemethod failed
， 。 。 ：
from multiprocessing import Pool, Process, Pipe
from itertools import izip
def spawn(f):
def func(pipe, item):
pipe.send(f(item))
pipe.close()
return func
def parmap(f, items):
pipe = [Pipe() for _ in items]
proc = [Process(target=spawn(f),
args=(child, item))
for item, (parent, child) in izip(items, pipe)]
[p.start() for p in proc]
[p.join() for p in proc]
return [parent.recv() for (parent, child) in pipe]
class CalculateFib(object):
...
def parmap_run(self):
print parmap(self.fib, [35] * 2)
cl = CalculateFib()
cl.parmap_run()
parmap （ fib(35)） ， ， 。
。 ， ， ， 。
Queue ， Queue ， 。 / ， 2 ： ， ：
import time
from multiprocessing import Process, JoinableQueue, Queue
from random import random
tasks_queue = JoinableQueue()
results_queue = Queue()
def double(n):
return n * 2
def producer(in_queue):
while 1:
wt = random()
time.sleep(wt)
in_queue.put((double, wt))
if wt > 0.9:
in_queue.put(None)
print 'stop producer'
break
def consumer(in_queue, out_queue):
while 1:
task = in_queue.get()
if task is None:
break
func, arg = task
result = func(arg)
in_queue.task_done()
out_queue.put(result)
processes = []
p = Process(target=producer, args=(tasks_queue,))
p.start()
processes.append(p)
p = Process(target=consumer, args=(tasks_queue, results_queue))
p.start()
processes.append(p)
tasks_queue.join()
for p in processes:
p.join()
while 1:
if results_queue.empty():
break
result = results_queue.get()
print 'Result:', result
， ：
-
， 0.9 tasks_queue put None，
-
None ， ， tasks_queue ， put results_queue
-
（ join ）， results_queue
Queue task_done join ， JoinableQueue， results_queue Queue 。
CalculateFib ， parmap ， ：
from multiprocessing import Queue, Process, cpu_count
def apply_func(f, q_in, q_out):
while not q_in.empty():
i, item = q_in.get()
q_out.put((i, f(item)))
def parmap(f, items, nprocs = cpu_count()):
q_in, q_out = Queue(), Queue()
proc = [Process(target=apply_func, args=(f, q_in, q_out))
for _ in range(nprocs)]
sent = [q_in.put((i, item)) for i, item in enumerate(items)]
[p.start() for p in proc]
res = [q_out.get() for _ in sent]
[p.join() for p in proc]
return [item for _, item in sorted(res)]
enumerate ， 。
multiprocessing Lock、Condition、Event、RLock、Semaphore threading ， ， ， 。
multiprocessing 2 ：
Value Array 。 ：
In : from multiprocessing.sharedctypes import typecode_to_type
In : typecode_to_type
Out:
{'B': ctypes.c_ubyte,
'H': ctypes.c_ushort,
'I': ctypes.c_uint,
'L': ctypes.c_ulong,
'b': ctypes.c_byte,
'c': ctypes.c_char,
'd': ctypes.c_double,
'f': ctypes.c_float,
'h': ctypes.c_short,
'i': ctypes.c_int,
'l': ctypes.c_long,
'u': ctypes.c_wchar}
Value Array lock ， RLock。
：
from multiprocessing import Process, Lock
from multiprocessing.sharedctypes import Value, Array
from ctypes import Structure, c_bool, c_double
lock = Lock()
class Point(Structure):
_fields_ = [('x', c_double), ('y', c_double)]
def modify(n, b, s, arr, A):
n.value **= 2
b.value = True
s.value = s.value.upper()
arr[0] = 10
for a in A:
a.x **= 2
a.y **= 2
n = Value('i', 7)
b = Value(c_bool, False, lock=False)
s = Array('c', 'hello world', lock=lock)
arr = Array('i', range(5), lock=True)
A = Array(Point, [(1.875, -6.25), (-5.75, 2.0)], lock=lock)
p = Process(target=modify, args=(n, b, s, arr, A))
p.start()
p.join()
print n.value
print b.value
print s.value
print arr[:]
print [(a.x, a.y) for a in A]
。 2 ：
-
typecode_to_type ， ctypes 。
-
arr int ， array list ， SynchronizedArray ， ， append/extend 。
：
❯ python shared_memory.py
49
True
HELLO WORLD
[10, 1, 2, 3, 4]
[(3.515625, 39.0625), (33.0625, 4.0)]
multiprocessing.Manager ， 。 ：
-
Namespace。 。
-
Value/Array。 ctypes 。
-
dict/list。 dict/list， 。
-
Condition/Event/Lock/Queue/Semaphore。 。
：
from multiprocessing import Manager, Process
def modify(ns, lproxy, dproxy):
ns.a **= 2
lproxy.extend(['b', 'c'])
dproxy['b'] = 0
manager = Manager()
ns = manager.Namespace()
ns.a = 1
lproxy = manager.list()
lproxy.append('a')
dproxy = manager.dict()
dproxy['b'] = 2
p = Process(target=modify, args=(ns, lproxy, dproxy))
p.start()
print 'PID:', p.pid
p.join()
print ns.a
print lproxy
print dproxy
id 8341 ：
❯ python manager.py
PID: 8341
1
['a', 'b', 'c']
{'b': 0}
， Manager Queue （C/S ）。
：
from multiprocessing.managers import BaseManager
host = '127.0.0.1'
port = 9030
authkey = 'secret'
shared_list = []
class RemoteManager(BaseManager):
pass
RemoteManager.register('get_list', callable=lambda: shared_list)
mgr = RemoteManager(address=(host, port), authkey=authkey)
server = mgr.get_server()
server.serve_forever()
shared_list ， ：
from multiprocessing.managers import BaseManager
host = '127.0.0.1'
port = 9030
authkey = 'secret'
class RemoteManager(BaseManager):
pass
RemoteManager.register('get_list')
mgr = RemoteManager(address=(host, port), authkey=authkey)
mgr.connect()
l = mgr.get_list()
print l
l.append(1)
print mgr.get_list()
， client callable 。
PS