Python 多处理在连接处挂起

Question

我正在读取一个视频文件，每 20 帧我就会将第一帧存储在输入队列中。一旦我在输入队列中获得了所有必需的帧，我就会运行多个进程来对这些帧执行一些操作并将结果存储在输出队列中。但代码总是卡在 join 处，我尝试了针对此类问题提出的不同解决方案，但似乎都不起作用。

import numpy as np
import cv2
import timeit
import face_recognition
from multiprocessing import Process, Queue, Pool
import multiprocessing
import os

s = timeit.default_timer()

def alternative_process_target_func(input_queue, output_queue):

    while not output_queue.full():
        frame_no, small_frame, face_loc = input_queue.get()
        print('Frame_no: ', frame_no, 'Process ID: ', os.getpid(), '----', multiprocessing.current_process())
        #canny_frame(frame_no, small_frame, face_loc)

        #I am just storing frame no for now but will perform something else later
        output_queue.put((frame_no, frame_no)) 

        if output_queue.full():
            print('Its Full ---------------------------------------------------------------------------------------')
        else:
            print('Not Full')

    print(timeit.default_timer() - s, ' seconds.')
    print('I m not reading anymore. . .', os.getpid())


def alternative_process(file_name):
    start = timeit.default_timer()
    cap = cv2.VideoCapture(file_name)
    frame_no = 1
    fps = cap.get(cv2.CAP_PROP_FPS)
    length = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    print('Frames Per Second: ', fps)
    print('Total Number of frames: ', length)
    print('Duration of file: ', int(length / fps))
    processed_frames = 1
    not_processed = 1
    frames = []
    process_this_frame = True
    frame_no = 1
    Input_Queue = Queue()
    while (cap.isOpened()):
        ret, frame = cap.read()
        if not ret:
            print('Size of input Queue: ', Input_Queue.qsize())
            print('Total no of frames read: ', frame_no)
            end1 = timeit.default_timer()
            print('Time taken to fetch useful frames: ', end1 - start)
            threadn = cv2.getNumberOfCPUs()
            Output_Queue = Queue(maxsize=Input_Queue.qsize())
            process_list = []
            #quit = multiprocessing.Event()
            #foundit = multiprocessing.Event()

            for x in range((threadn - 1)):
                # print('Process No : ', x)
                p = Process(target=alternative_process_target_func, args=(Input_Queue, Output_Queue))#, quit, foundit
                #p.daemon = True
                p.start()
                process_list.append(p)
                #p.join()

            # for proc in process_list:
            #     print('---------------------------------------------------------------', proc.p)

            i = 1
            for proc in process_list:
                print('I am hanged here')
                proc.join()
                print('I am done')
                i += 1

            end = timeit.default_timer()
            print('Time taken by face verification: ', end - start)

            break

        if process_this_frame:
            print(frame_no)
            small_frame = cv2.resize(frame, (0, 0), fx=0.25, fy=0.25)
            rgb_small_frame = small_frame[:, :, ::-1]
            face_locations = face_recognition.face_locations(rgb_small_frame)
            # frames.append((rgb_small_frame, face_locations))
            Input_Queue.put((frame_no, rgb_small_frame, face_locations))
            frame_no += 1

        if processed_frames < 5:
            processed_frames += 1
            not_processed = 1

        else:
            if not_processed < 15:
                process_this_frame = False
                not_processed += 1
            else:

                processed_frames = 1
                process_this_frame = True
                print('-----------------------------------------------------------------------------------------------')

    cap.release()
    cv2.destroyAllWindows()

alternative_process('user_verification_2.avi')

Answer 1

如 Process.join() 上的文档所示说，挂起(或“阻塞”)正是预期会发生的情况:

Block the calling thread until the process whose join() method is called terminates or until the optional timeout occurs.

join() 停止当前线程，直到目标进程完成。目标进程正在调用alternative_process_target_func，因此问题显然出在该函数中。它永远不会结束。造成这种情况的原因可能不止一个。

问题 1

alternative_process_target_func 运行直到 output_queue.full()。如果它永远不满怎么办？它永远不会结束？最好以其他方式确定结局，例如运行直到输入队列为空。

问题2

如果输入队列为空，

input_queue.get() 将阻塞。作为documentation说:

Remove and return an item from the queue. If optional args block is true and timeout is None (the default), block if necessary until an item is available.

您正在运行多个进程，因此不要仅仅因为 output_queue.full() 刚才为 False 并且因为输入大小与输出大小相同而期望输入中有内容。在此期间可能发生很多事情。

你想做的是:

try:
    input_queue.get(False)  # or input_queue.get_nowait()
except Empty:
    break  # stop when there is nothing more to read from the input

问题3

如果输出中没有空间来存储数据，

output_queue.put((frame_no, frame_no)) 将阻塞。

再次，您假设输出中有空间，只是因为您刚才检查了 output_queue.full()，并且输入大小等于输出大小。永远不要依赖这样的东西。

您想要执行与输入相同的操作:

try:
    output_queue.put((frame_no, frame_no), False)
    # or output_queue.put_nowait((frame_no, frame_no))
except Empty:
    # deal with this somehow, e.g.
    raise Exception("There is no room in the output queue to write to.")