Multiprocessing: Bagaimana cara menggunakan Pool.map pada fungsi yang didefinisikan di kelas?

Ketika saya menjalankan sesuatu seperti:

from multiprocessing import Pool

p = Pool(5)
def f(x):
     return x*x

p.map(f, [1,2,3])

ini bekerja dengan baik. Namun, menempatkan ini sebagai fungsi kelas:

class calculate(object):
    def run(self):
        def f(x):
            return x*x

        p = Pool()
        return p.map(f, [1,2,3])

cl = calculate()
print cl.run()

Memberi saya kesalahan berikut:

Exception in thread Thread-1:
Traceback (most recent call last):
  File "/sw/lib/python2.6/threading.py", line 532, in __bootstrap_inner
    self.run()
  File "/sw/lib/python2.6/threading.py", line 484, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/sw/lib/python2.6/multiprocessing/pool.py", line 225, in _handle_tasks
    put(task)
PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed

Saya telah melihat posting dari Alex Martelli berurusan dengan masalah yang sama, tetapi itu tidak cukup eksplisit.

Saya juga terganggu oleh pembatasan fungsi apa yang bisa diterima pool.map. Saya menulis yang berikut untuk menghindari ini. Tampaknya berfungsi, bahkan untuk penggunaan parmap secara rekursif.

from multiprocessing import Process, Pipe
from itertools import izip

def spawn(f):
    def fun(pipe, x):
        pipe.send(f(x))
        pipe.close()
    return fun

def parmap(f, X):
    pipe = [Pipe() for x in X]
    proc = [Process(target=spawn(f), args=(c, x)) for x, (p, c) in izip(X, pipe)]
    [p.start() for p in proc]
    [p.join() for p in proc]
    return [p.recv() for (p, c) in pipe]

if __name__ == '__main__':
    print parmap(lambda x: x**x, range(1, 5))

Saya tidak dapat menggunakan kode yang diposting sejauh ini karena kode yang menggunakan "multiprocessing.Pool" tidak bekerja dengan ekspresi lambda dan kode tidak menggunakan "multiprocessing.Pool" menelurkan banyak proses karena ada item pekerjaan.

Saya mengadaptasi kode st yang memunculkan jumlah pekerja yang telah ditentukan dan hanya beralih melalui daftar input jika ada pekerja yang menganggur. Saya juga mengaktifkan mode "daemon" untuk ctrl-c karya pekerja seperti yang diharapkan.

import multiprocessing


def fun(f, q_in, q_out):
    while True:
        i, x = q_in.get()
        if i is None:
            break
        q_out.put((i, f(x)))


def parmap(f, X, nprocs=multiprocessing.cpu_count()):
    q_in = multiprocessing.Queue(1)
    q_out = multiprocessing.Queue()

    proc = [multiprocessing.Process(target=fun, args=(f, q_in, q_out))
            for _ in range(nprocs)]
    for p in proc:
        p.daemon = True
        p.start()

    sent = [q_in.put((i, x)) for i, x in enumerate(X)]
    [q_in.put((None, None)) for _ in range(nprocs)]
    res = [q_out.get() for _ in range(len(sent))]

    [p.join() for p in proc]

    return [x for i, x in sorted(res)]


if __name__ == '__main__':
    print(parmap(lambda i: i * 2, [1, 2, 3, 4, 6, 7, 8]))

Multiprocessing dan acar rusak dan terbatas kecuali Anda melompat di luar perpustakaan standar.

Jika Anda menggunakan fork yang multiprocessingdipanggil pathos.multiprocesssing, Anda bisa langsung menggunakan kelas dan metode kelas dalam mapfungsi multi-pemrosesan . Ini karena dilldigunakan sebagai ganti pickleatau cPickle, dan dilldapat membuat serialisasi hampir semua hal dengan python.

pathos.multiprocessingjuga menyediakan fungsi peta yang tidak sinkron ... dan dapat mapberfungsi dengan banyak argumen (mis. map(math.pow, [1,2,3], [4,5,6]))

Lihat diskusi: Apa yang dapat dilakukan multiprocessing dan dill bersama?

dan: http://matthewrocklin.com/blog/work/2013/12/05/Parallelism-and-Serialization

Bahkan menangani kode yang Anda tulis pada awalnya, tanpa modifikasi, dan dari penerjemah. Mengapa melakukan hal lain yang lebih rapuh dan spesifik untuk satu kasus?

>>> from pathos.multiprocessing import ProcessingPool as Pool
>>> class calculate(object):
...  def run(self):
...   def f(x):
...    return x*x
...   p = Pool()
...   return p.map(f, [1,2,3])
... 
>>> cl = calculate()
>>> print cl.run()
[1, 4, 9]

Dapatkan kode di sini: https://github.com/uqfoundation/pathos

Dan, hanya untuk memamerkan sedikit tentang apa yang dapat dilakukannya:

>>> from pathos.multiprocessing import ProcessingPool as Pool
>>> 
>>> p = Pool(4)
>>> 
>>> def add(x,y):
...   return x+y
... 
>>> x = [0,1,2,3]
>>> y = [4,5,6,7]
>>> 
>>> p.map(add, x, y)
[4, 6, 8, 10]
>>> 
>>> class Test(object):
...   def plus(self, x, y): 
...     return x+y
... 
>>> t = Test()
>>> 
>>> p.map(Test.plus, [t]*4, x, y)
[4, 6, 8, 10]
>>> 
>>> res = p.amap(t.plus, x, y)
>>> res.get()
[4, 6, 8, 10]

Saat ini tidak ada solusi untuk masalah Anda, sejauh yang saya tahu: fungsi yang Anda berikan map()harus dapat diakses melalui impor modul Anda. Inilah sebabnya mengapa kode robert berfungsi: fungsinya f()dapat diperoleh dengan mengimpor kode berikut:

def f(x):
    return x*x

class Calculate(object):
    def run(self):
        p = Pool()
        return p.map(f, [1,2,3])

if __name__ == '__main__':
    cl = Calculate()
    print cl.run()

Saya sebenarnya menambahkan bagian "utama", karena ini mengikuti rekomendasi untuk platform Windows ("Pastikan bahwa modul utama dapat diimpor dengan aman oleh juru bahasa Python baru tanpa menyebabkan efek samping yang tidak diinginkan").

Saya juga menambahkan huruf besar di depan Calculate, untuk mengikuti PEP 8 . :)

Solusi oleh mrule benar tetapi memiliki bug: jika anak mengirim kembali sejumlah besar data, itu dapat mengisi buffer pipa, menghalangi anak pipe.send(), sementara orang tua menunggu anak untuk keluar pipe.join(). Solusinya adalah dengan membaca data anak sebelum join()memasukkan anak. Lebih jauh lagi, anak harus menutup ujung pipa orang tua untuk mencegah jalan buntu. Kode di bawah ini memperbaikinya. Perlu diketahui juga bahwa ini parmapmenciptakan satu proses per elemen di X. Solusi yang lebih maju adalah menggunakan multiprocessing.cpu_count()untuk membagi Xmenjadi beberapa potongan, dan kemudian menggabungkan hasilnya sebelum kembali. Saya meninggalkan itu sebagai latihan untuk pembaca agar tidak merusak keringkasan jawaban bagus oleh mrule. ;)

from multiprocessing import Process, Pipe
from itertools import izip

def spawn(f):
    def fun(ppipe, cpipe,x):
        ppipe.close()
        cpipe.send(f(x))
        cpipe.close()
    return fun

def parmap(f,X):
    pipe=[Pipe() for x in X]
    proc=[Process(target=spawn(f),args=(p,c,x)) for x,(p,c) in izip(X,pipe)]
    [p.start() for p in proc]
    ret = [p.recv() for (p,c) in pipe]
    [p.join() for p in proc]
    return ret

if __name__ == '__main__':
    print parmap(lambda x:x**x,range(1,5))

Saya juga berjuang dengan ini. Saya memiliki fungsi sebagai anggota data kelas, sebagai contoh sederhana:

from multiprocessing import Pool
import itertools
pool = Pool()
class Example(object):
    def __init__(self, my_add): 
        self.f = my_add  
    def add_lists(self, list1, list2):
        # Needed to do something like this (the following line won't work)
        return pool.map(self.f,list1,list2)  

Saya perlu menggunakan fungsi self.f dalam panggilan Pool.map () dari dalam kelas yang sama dan self.f tidak menganggap tuple sebagai argumen. Karena fungsi ini tertanam dalam kelas, tidak jelas bagi saya bagaimana menulis jenis pembungkus jawaban yang disarankan.

Saya memecahkan masalah ini dengan menggunakan pembungkus berbeda yang mengambil tuple / daftar, di mana elemen pertama adalah fungsi, dan elemen yang tersisa adalah argumen untuk fungsi itu, yang disebut eval_func_tuple (f_args). Dengan menggunakan ini, baris bermasalah dapat diganti dengan return pool.map (eval_func_tuple, itertools.izip (itertools.repeat (self.f), list1, list2)). Ini kode lengkapnya:

File: util.py

def add(a, b): return a+b

def eval_func_tuple(f_args):
    """Takes a tuple of a function and args, evaluates and returns result"""
    return f_args[0](*f_args[1:])  

File: main.py

from multiprocessing import Pool
import itertools
import util  

pool = Pool()
class Example(object):
    def __init__(self, my_add): 
        self.f = my_add  
    def add_lists(self, list1, list2):
        # The following line will now work
        return pool.map(util.eval_func_tuple, 
            itertools.izip(itertools.repeat(self.f), list1, list2)) 

if __name__ == '__main__':
    myExample = Example(util.add)
    list1 = [1, 2, 3]
    list2 = [10, 20, 30]
    print myExample.add_lists(list1, list2)  

Menjalankan main.py akan memberi [11, 22, 33]. Jangan ragu untuk memperbaiki ini, misalnya eval_func_tuple juga dapat dimodifikasi untuk mengambil argumen kata kunci.

Pada catatan lain, pada jawaban lain, fungsi "parmap" dapat dibuat lebih efisien untuk kasus lebih banyak Proses daripada jumlah CPU yang tersedia. Saya menyalin versi yang diedit di bawah ini. Ini adalah posting pertama saya dan saya tidak yakin apakah saya harus langsung mengedit jawaban aslinya. Saya juga mengganti nama beberapa variabel.

from multiprocessing import Process, Pipe  
from itertools import izip  

def spawn(f):  
    def fun(pipe,x):  
        pipe.send(f(x))  
        pipe.close()  
    return fun  

def parmap(f,X):  
    pipe=[Pipe() for x in X]  
    processes=[Process(target=spawn(f),args=(c,x)) for x,(p,c) in izip(X,pipe)]  
    numProcesses = len(processes)  
    processNum = 0  
    outputList = []  
    while processNum < numProcesses:  
        endProcessNum = min(processNum+multiprocessing.cpu_count(), numProcesses)  
        for proc in processes[processNum:endProcessNum]:  
            proc.start()  
        for proc in processes[processNum:endProcessNum]:  
            proc.join()  
        for proc,c in pipe[processNum:endProcessNum]:  
            outputList.append(proc.recv())  
        processNum = endProcessNum  
    return outputList    

if __name__ == '__main__':  
    print parmap(lambda x:x**x,range(1,5))         

Saya mengambil jawaban klaus se dan aganders3, dan membuat modul terdokumentasi yang lebih mudah dibaca dan disimpan dalam satu file. Anda bisa menambahkannya ke proyek Anda. Bahkan memiliki progress bar opsional!

"""
The ``processes`` module provides some convenience functions
for using parallel processes in python.

Adapted from http://stackoverflow.com/a/16071616/287297

Example usage:

    print prll_map(lambda i: i * 2, [1, 2, 3, 4, 6, 7, 8], 32, verbose=True)

Comments:

"It spawns a predefined amount of workers and only iterates through the input list
 if there exists an idle worker. I also enabled the "daemon" mode for the workers so
 that KeyboardInterupt works as expected."

Pitfalls: all the stdouts are sent back to the parent stdout, intertwined.

Alternatively, use this fork of multiprocessing: 
https://github.com/uqfoundation/multiprocess
"""

# Modules #
import multiprocessing
from tqdm import tqdm

################################################################################
def apply_function(func_to_apply, queue_in, queue_out):
    while not queue_in.empty():
        num, obj = queue_in.get()
        queue_out.put((num, func_to_apply(obj)))

################################################################################
def prll_map(func_to_apply, items, cpus=None, verbose=False):
    # Number of processes to use #
    if cpus is None: cpus = min(multiprocessing.cpu_count(), 32)
    # Create queues #
    q_in  = multiprocessing.Queue()
    q_out = multiprocessing.Queue()
    # Process list #
    new_proc  = lambda t,a: multiprocessing.Process(target=t, args=a)
    processes = [new_proc(apply_function, (func_to_apply, q_in, q_out)) for x in range(cpus)]
    # Put all the items (objects) in the queue #
    sent = [q_in.put((i, x)) for i, x in enumerate(items)]
    # Start them all #
    for proc in processes:
        proc.daemon = True
        proc.start()
    # Display progress bar or not #
    if verbose:
        results = [q_out.get() for x in tqdm(range(len(sent)))]
    else:
        results = [q_out.get() for x in range(len(sent))]
    # Wait for them to finish #
    for proc in processes: proc.join()
    # Return results #
    return [x for i, x in sorted(results)]

################################################################################
def test():
    def slow_square(x):
        import time
        time.sleep(2)
        return x**2
    objs    = range(20)
    squares = prll_map(slow_square, objs, 4, verbose=True)
    print "Result: %s" % squares

EDIT : Menambahkan saran @ alexander-mcfarlane dan fungsi tes

Saya tahu ini ditanyakan lebih dari 6 tahun yang lalu sekarang, tetapi hanya ingin menambahkan solusi saya, karena beberapa saran di atas tampak sangat rumit, tetapi solusi saya sebenarnya sangat sederhana.

Yang harus saya lakukan adalah membungkus panggilan pool.map () ke fungsi pembantu. Melewati objek kelas bersama dengan args untuk metode sebagai tuple, yang terlihat sedikit seperti ini.

def run_in_parallel(args):
    return args[0].method(args[1])

myclass = MyClass()
method_args = [1,2,3,4,5,6]
args_map = [ (myclass, arg) for arg in method_args ]
pool = Pool()
pool.map(run_in_parallel, args_map)

Fungsi yang didefinisikan dalam kelas (bahkan di dalam fungsi di dalam kelas) tidak benar-benar acar. Namun, ini berfungsi:

def f(x):
    return x*x

class calculate(object):
    def run(self):
        p = Pool()
    return p.map(f, [1,2,3])

cl = calculate()
print cl.run()

Saya tahu bahwa pertanyaan ini ditanyakan 8 tahun dan 10 bulan yang lalu tetapi saya ingin memberikan solusi kepada Anda:

from multiprocessing import Pool

class Test:

    def __init__(self):
        self.main()

    @staticmethod
    def methodForMultiprocessing(x):
        print(x*x)

    def main(self):
        if __name__ == "__main__":
            p = Pool()
            p.map(Test.methodForMultiprocessing, list(range(1, 11)))
            p.close()

TestObject = Test()

Anda hanya perlu menjadikan fungsi kelas Anda menjadi metode statis. Tetapi itu juga mungkin dengan metode kelas:

from multiprocessing import Pool

class Test:

    def __init__(self):
        self.main()

    @classmethod
    def methodForMultiprocessing(cls, x):
        print(x*x)

    def main(self):
        if __name__ == "__main__":
            p = Pool()
            p.map(Test.methodForMultiprocessing, list(range(1, 11)))
            p.close()

TestObject = Test()

Diuji dalam Python 3.7.3

Saya memodifikasi metode klaus se karena ketika itu bekerja untuk saya dengan daftar kecil, itu akan hang ketika jumlah item ~ 1000 atau lebih besar. Alih-alih mendorong pekerjaan satu per satu dengan Nonekondisi berhenti, saya memuat antrian input sekaligus dan membiarkan proses mengunyah sampai kosong.

from multiprocessing import cpu_count, Queue, Process

def apply_func(f, q_in, q_out):
    while not q_in.empty():
        i, x = q_in.get()
        q_out.put((i, f(x)))

# map a function using a pool of processes
def parmap(f, X, 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, x)) for i, x in enumerate(X)]
    [p.start() for p in proc]
    res = [q_out.get() for _ in sent]
    [p.join() for p in proc]

    return [x for i,x in sorted(res)]

Sunting: sayangnya sekarang saya mengalami kesalahan pada sistem saya: Batas maksimum antrian multiprosesing adalah 32767 , mudah-mudahan solusi di sana akan membantu.

Anda dapat menjalankan kode Anda tanpa masalah jika Anda entah bagaimana mengabaikan Poolobjek secara manual dari daftar objek di kelas karena tidak pickledapat seperti yang dikatakan kesalahan. Anda dapat melakukan ini dengan __getstate__fungsi (lihat di sini juga) sebagai berikut. The Poolobjek akan mencoba untuk menemukan __getstate__dan __setstate__fungsi dan mengeksekusi mereka jika menemukan ketika Anda menjalankan map, map_asyncdll:

class calculate(object):
    def __init__(self):
        self.p = Pool()
    def __getstate__(self):
        self_dict = self.__dict__.copy()
        del self_dict['p']
        return self_dict
    def __setstate__(self, state):
        self.__dict__.update(state)

    def f(self, x):
        return x*x
    def run(self):
        return self.p.map(self.f, [1,2,3])

Kemudian lakukan:

cl = calculate()
cl.run()

akan memberi Anda output:

[1, 4, 9]

Saya sudah menguji kode di atas dengan Python 3.x dan berhasil.

Saya tidak yakin apakah pendekatan ini telah diambil tetapi upaya yang saya gunakan adalah:

from multiprocessing import Pool

t = None

def run(n):
    return t.f(n)

class Test(object):
    def __init__(self, number):
        self.number = number

    def f(self, x):
        print x * self.number

    def pool(self):
        pool = Pool(2)
        pool.map(run, range(10))

if __name__ == '__main__':
    t = Test(9)
    t.pool()
    pool = Pool(2)
    pool.map(run, range(10))

Output harus:

0
9
18
27
36
45
54
63
72
81
0
9
18
27
36
45
54
63
72
81

class Calculate(object):
  # Your instance method to be executed
  def f(self, x, y):
    return x*y

if __name__ == '__main__':
  inp_list = [1,2,3]
  y = 2
  cal_obj = Calculate()
  pool = Pool(2)
  results = pool.map(lambda x: cal_obj.f(x, y), inp_list)

Ada kemungkinan Anda ingin menerapkan fungsi ini untuk setiap instance kelas yang berbeda. Maka di sini adalah solusi untuk itu juga

class Calculate(object):
  # Your instance method to be executed
  def __init__(self, x):
    self.x = x

  def f(self, y):
    return self.x*y

if __name__ == '__main__':
  inp_list = [Calculate(i) for i in range(3)]
  y = 2
  pool = Pool(2)
  results = pool.map(lambda x: x.f(y), inp_list)

Ini solusi saya, yang menurut saya agak kurang hackish daripada kebanyakan orang lain di sini. Ini mirip dengan jawaban nightowl.

someclasses = [MyClass(), MyClass(), MyClass()]

def method_caller(some_object, some_method='the method'):
    return getattr(some_object, some_method)()

othermethod = partial(method_caller, some_method='othermethod')

with Pool(6) as pool:
    result = pool.map(othermethod, someclasses)

Dari http://www.rueckstiess.net/research/snippets/show/ca1d7d90 dan http://qingkaikong.blogspot.com/2016/12/python-parallel-method-in-class.html

Kita dapat membuat fungsi eksternal dan menaburinya dengan objek self class:

from joblib import Parallel, delayed
def unwrap_self(arg, **kwarg):
    return square_class.square_int(*arg, **kwarg)

class square_class:
    def square_int(self, i):
        return i * i

    def run(self, num):
        results = []
        results = Parallel(n_jobs= -1, backend="threading")\
            (delayed(unwrap_self)(i) for i in zip([self]*len(num), num))
        print(results)

ATAU tanpa joblib:

from multiprocessing import Pool
import time

def unwrap_self_f(arg, **kwarg):
    return C.f(*arg, **kwarg)

class C:
    def f(self, name):
        print 'hello %s,'%name
        time.sleep(5)
        print 'nice to meet you.'

    def run(self):
        pool = Pool(processes=2)
        names = ('frank', 'justin', 'osi', 'thomas')
        pool.map(unwrap_self_f, zip([self]*len(names), names))

if __name__ == '__main__':
    c = C()
    c.run()

Ini mungkin bukan solusi yang sangat baik tetapi dalam kasus saya, saya menyelesaikannya seperti ini.

from multiprocessing import Pool

def foo1(data):
    self = data.get('slf')
    lst = data.get('lst')
    return sum(lst) + self.foo2()

class Foo(object):
    def __init__(self, a, b):
        self.a = a
        self.b = b

    def foo2(self):
        return self.a**self.b   

    def foo(self):
        p = Pool(5)
        lst = [1, 2, 3]
        result = p.map(foo1, (dict(slf=self, lst=lst),))
        return result

if __name__ == '__main__':
    print(Foo(2, 4).foo())

Saya harus beralih selfke fungsi saya karena saya harus mengakses atribut dan fungsi kelas saya melalui fungsi itu. Ini bekerja untuk saya. Koreksi dan saran selalu kami terima.

Berikut adalah boilerplate yang saya tulis untuk menggunakan Pool multiprocessing di python3, khususnya python3.7.7 digunakan untuk menjalankan tes. Saya menggunakan lari tercepat saya imap_unordered. Cukup masukkan skenario Anda dan coba. Anda dapat menggunakan timeitatau hanya time.time()untuk mencari tahu mana yang paling cocok untuk Anda.

import multiprocessing
import time

NUMBER_OF_PROCESSES = multiprocessing.cpu_count()
MP_FUNCTION = 'starmap'  # 'imap_unordered' or 'starmap' or 'apply_async'

def process_chunk(a_chunk):
    print(f"processig mp chunk {a_chunk}")
    return a_chunk


map_jobs = [1, 2, 3, 4]

result_sum = 0

s = time.time()
if MP_FUNCTION == 'imap_unordered':
    pool = multiprocessing.Pool(processes=NUMBER_OF_PROCESSES)
    for i in pool.imap_unordered(process_chunk, map_jobs):
        result_sum += i
elif MP_FUNCTION == 'starmap':
    pool = multiprocessing.Pool(processes=NUMBER_OF_PROCESSES)
    try:
        map_jobs = [(i, ) for i in map_jobs]
        result_sum = pool.starmap(process_chunk, map_jobs)
        result_sum = sum(result_sum)
    finally:
        pool.close()
        pool.join()
elif MP_FUNCTION == 'apply_async':
    with multiprocessing.Pool(processes=NUMBER_OF_PROCESSES) as pool:
        result_sum = [pool.apply_async(process_chunk, [i, ]).get() for i in map_jobs]
    result_sum = sum(result_sum)
print(f"result_sum is {result_sum}, took {time.time() - s}s")

Dalam skenario di atas imap_unorderedsebenarnya sepertinya melakukan yang terburuk bagi saya. Cobalah kasing Anda dan berikan tolok ukur pada mesin yang ingin Anda jalankan. Baca juga di Proses Kolam . Bersulang!