Quy trình Python Pool non-daemonic?

Question

Có thể tạo một Hồ bơi python không phải là daemon không? Tôi muốn một hồ bơi để có thể gọi một chức năng có một hồ bơi bên trong. Cảm ơn.

Answer 1

Lớp multiprocessing.pool.Pool tạo ra các quá trình lao động trong phương pháp __init__ của nó, làm cho chúng ma quỉ và bắt đầu họ, và nó không thể để tái thiết lập thuộc tính daemon của họ để False trước khi chúng được bắt đầu (và sau đó nó không cho phép nữa). Nhưng bạn có thể tạo một lớp con của riêng bạn là multiprocesing.pool.Pool (multiprocessing.Pool chỉ là một hàm bao bọc) và thay thế lớp con multiprocessing.Process của riêng bạn, mà luôn luôn không phải là daemonic, được sử dụng cho các quy trình công nhân.

Dưới đây là ví dụ đầy đủ về cách thực hiện việc này. Các phần quan trọng là hai lớp NoDaemonProcess và MyPool ở trên cùng và gọi pool.close() và pool.join() trên phiên bản MyPool của bạn ở cuối.

#!/usr/bin/env python 
# -*- coding: UTF-8 -*- 

import multiprocessing 
# We must import this explicitly, it is not imported by the top-level 
# multiprocessing module. 
import multiprocessing.pool 
import time 

from random import randint 


class NoDaemonProcess(multiprocessing.Process): 
    # make 'daemon' attribute always return False 
    def _get_daemon(self): 
     return False 
    def _set_daemon(self, value): 
     pass 
    daemon = property(_get_daemon, _set_daemon) 

# We sub-class multiprocessing.pool.Pool instead of multiprocessing.Pool 
# because the latter is only a wrapper function, not a proper class. 
class MyPool(multiprocessing.pool.Pool): 
    Process = NoDaemonProcess 

def sleepawhile(t): 
    print("Sleeping %i seconds..." % t) 
    time.sleep(t) 
    return t 

def work(num_procs): 
    print("Creating %i (daemon) workers and jobs in child." % num_procs) 
    pool = multiprocessing.Pool(num_procs) 

    result = pool.map(sleepawhile, 
     [randint(1, 5) for x in range(num_procs)]) 

    # The following is not really needed, since the (daemon) workers of the 
    # child's pool are killed when the child is terminated, but it's good 
    # practice to cleanup after ourselves anyway. 
    pool.close() 
    pool.join() 
    return result 

def test(): 
    print("Creating 5 (non-daemon) workers and jobs in main process.") 
    pool = MyPool(5) 

    result = pool.map(work, [randint(1, 5) for x in range(5)]) 

    pool.close() 
    pool.join() 
    print(result) 

if __name__ == '__main__': 
    test() 

Answer 2

Module multiprocessing có một giao diện đẹp để sử dụng hồ bơi với các quá trình hay đề. Tùy thuộc vào trường hợp sử dụng hiện tại của bạn, bạn có thể cân nhắc sử dụng multiprocessing.pool.ThreadPool cho hồ bơi ngoài của mình, điều này sẽ dẫn đến chủ đề (cho phép sinh ra các quy trình từ bên trong) trái với quy trình.

Nó có thể bị giới hạn bởi các GIL, nhưng trong trường hợp cụ thể của tôi (Tôi đã thử nghiệm cả hai), thời gian khởi động cho các quá trình từ bên ngoài Pool như tạo here nằm ngoài những giải pháp với ThreadPool.

---

Thật dễ dàng để hoán đổi Processes cho Threads. Đọc thêm về cách sử dụng giải pháp ThreadPoolhere hoặc here.

Answer 3

Vấn đề tôi gặp phải là cố gắng nhập hình cầu nối giữa các mô-đun, làm cho dòng ProcessPool() được đánh giá nhiều lần.

globals.py

from processing    import Manager, Lock 
from pathos.multiprocessing import ProcessPool 
from pathos.threading  import ThreadPool 

class SingletonMeta(type): 
    def __new__(cls, name, bases, dict): 
     dict['__deepcopy__'] = dict['__copy__'] = lambda self, *args: self 
     return super(SingletonMeta, cls).__new__(cls, name, bases, dict) 

    def __init__(cls, name, bases, dict): 
     super(SingletonMeta, cls).__init__(name, bases, dict) 
     cls.instance = None 

    def __call__(cls,*args,**kw): 
     if cls.instance is None: 
      cls.instance = super(SingletonMeta, cls).__call__(*args, **kw) 
     return cls.instance 

    def __deepcopy__(self, item): 
     return item.__class__.instance 

class Globals(object): 
    __metaclass__ = SingletonMeta 
    """  
    This class is a workaround to the bug: AssertionError: daemonic processes are not allowed to have children 

    The root cause is that importing this file from different modules causes this file to be reevalutated each time, 
    thus ProcessPool() gets reexecuted inside that child thread, thus causing the daemonic processes bug  
    """ 
    def __init__(self): 
     print "%s::__init__()" % (self.__class__.__name__) 
     self.shared_manager  = Manager() 
     self.shared_process_pool = ProcessPool() 
     self.shared_thread_pool = ThreadPool() 
     self.shared_lock   = Lock()  # BUG: Windows: global name 'lock' is not defined | doesn't affect cygwin 

Sau đó, nhập khẩu một cách an toàn từ nơi khác trong mã của bạn

from globals import Globals 
Globals().shared_manager  
Globals().shared_process_pool 
Globals().shared_thread_pool 
Globals().shared_lock