python multiprocessing pool chấm dứt

Question

Tôi đang làm việc trên renderfarm và tôi cần khách hàng của mình có thể khởi chạy nhiều phiên bản của trình kết xuất, không chặn để khách hàng có thể nhận lệnh mới. Tôi đã nhận được rằng làm việc một cách chính xác, tuy nhiên tôi đang gặp khó khăn chấm dứt quá trình tạo ra.

Ở cấp độ toàn cầu, tôi xác định hồ bơi của tôi (vì vậy mà tôi có thể truy cập nó từ bất kỳ chức năng):

p = Pool(2) 

sau đó tôi gọi renderer của tôi với apply_async:

for i in range(totalInstances): 
    p.apply_async(render, (allRenderArgs[i],args[2]), callback=renderFinished) 
p.close() 

Đó hoạt động kết thúc , khởi chạy các tiến trình trong nền và chờ các lệnh mới. Tôi đã thực hiện một lệnh đơn giản sẽ giết khách hàng và ngừng hiển thị:

def close(): 
'close this client instance' 
tn.write ("say "+USER+" is leaving the farm\r\n") 
try: 
    p.terminate() 
except Exception,e: 
    print str(e) 
    sys.exit() 
sys.exit() 

Nó dường như không đưa ra lỗi (nó sẽ in lỗi), con trăn chấm dứt nhưng các quy trình nền vẫn còn đang chạy. Bất cứ ai có thể giới thiệu một cách tốt hơn để kiểm soát các chương trình đưa ra?

Answer 1

Tìm câu trả lời cho câu hỏi của riêng tôi. Vấn đề chính là tôi đã gọi một ứng dụng của bên thứ ba hơn là một chức năng. Khi tôi gọi subprocess [hoặc sử dụng call() hoặc Popen()] nó tạo ra một thể hiện mới của python mà mục đích duy nhất của nó là gọi ứng dụng mới. Tuy nhiên khi python thoát, nó sẽ giết trường hợp python mới này và để ứng dụng chạy.

Giải pháp là làm điều đó một cách khó khăn, bằng cách tìm pid của quá trình python được tạo, nhận con của pid đó và giết chúng. Mã này là dành riêng cho osx; có mã đơn giản hơn (không phụ thuộc vào grep) có sẵn cho Linux.

for process in pool: 
    processId = process.pid 
    print "attempting to terminate "+str(processId) 
    command = " ps -o pid,ppid -ax | grep "+str(processId)+" | cut -f 1 -d \" \" | tail -1" 
    ps_command = Popen(command, shell=True, stdout=PIPE) 
    ps_output = ps_command.stdout.read() 
    retcode = ps_command.wait() 
    assert retcode == 0, "ps command returned %d" % retcode 
    print "child process pid: "+ str(ps_output) 
    os.kill(int(ps_output), signal.SIGTERM) 
    os.kill(int(processId), signal.SIGTERM) 

Answer 2

Nếu bạn vẫn gặp phải sự cố này, bạn có thể thử mô phỏng Pool với daemonic processes (giả sử bạn đang bắt đầu nhóm/quy trình từ quy trình không phải daemon). Tôi nghi ngờ đây là giải pháp tốt nhất vì có vẻ như các quy trình Pool của bạn nên thoát, nhưng đây là tất cả những gì tôi có thể đưa ra. Tôi không biết những gì gọi lại của bạn làm như vậy tôi không chắc chắn nơi để đặt nó trong ví dụ của tôi dưới đây.

Tôi cũng khuyên bạn nên tạo Pool trong __main__ do trải nghiệm của tôi (và tài liệu) với sự kỳ lạ xảy ra khi các quá trình được sinh ra trên toàn cầu. Điều này đặc biệt đúng nếu bạn đang sử dụng Windows: http://docs.python.org/2/library/multiprocessing.html#windows

from multiprocessing import Process, JoinableQueue 

# the function for each process in our pool 
def pool_func(q): 
    while True: 
     allRenderArg, otherArg = q.get() # blocks until the queue has an item 
     try: 
      render(allRenderArg, otherArg) 
     finally: q.task_done() 

# best practice to go through main for multiprocessing 
if __name__=='__main__': 
    # create the pool 
    pool_size = 2 
    pool = [] 
    q = JoinableQueue() 
    for x in range(pool_size): 
     pool.append(Process(target=pool_func, args=(q,))) 

    # start the pool, making it "daemonic" (the pool should exit when this proc exits) 
    for p in pool: 
     p.daemon = True 
     p.start() 

    # submit jobs to the queue 
    for i in range(totalInstances): 
     q.put((allRenderArgs[i], args[2])) 

    # wait for all tasks to complete, then exit 
    q.join() 

Answer 3

tôi tìm thấy giải pháp: ngừng hồ bơi trong chủ đề riêng biệt, như thế này:

def close_pool(): 
    global pool 
    pool.close() 
    pool.terminate() 
    pool.join() 

def term(*args,**kwargs): 
    sys.stderr.write('\nStopping...') 
    # httpd.shutdown() 
    stophttp = threading.Thread(target=httpd.shutdown) 
    stophttp.start() 
    stoppool=threading.Thread(target=close_pool) 
    stoppool.daemon=True 
    stoppool.start() 


signal.signal(signal.SIGTERM, term) 
signal.signal(signal.SIGINT, term) 
signal.signal(signal.SIGQUIT, term) 

Hoạt động tốt và luôn luôn tôi kiểm tra.