Многопроцессорная обработка Python возвращает очередь пустую, хотя на самом деле это не так - PullRequest
0 голосов
/ 29 ноября 2018

В этой программе после некоторых итераций все процессы завершаются, что означает, что input_queue пуст в соответствии с условием в целевой функции.Но после возврата к основной функции, когда я печатаю input_queue, в этой очереди все еще остаются элементы, тогда почему эти множественные процессы прервались на первом месте?

import cv2
import timeit
import face_recognition
import queue
from multiprocessing import Process, Queue
import multiprocessing
import os

s = timeit.default_timer()

def alternative_process_target_func(input_queue, output_queue):

    while not input_queue.empty():
        try:
            frame_no, small_frame, face_loc = input_queue.get(False)  # or input_queue.get_nowait()
            print('Frame_no: ', frame_no, 'Process ID: ', os.getpid(), '----', multiprocessing.current_process())

        except queue.Empty:
            print('___________________________________ Breaking __________________________________________________')
            break  # stop when there is nothing more to read from the input


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
                #print('I am a new process with process id of: ', os.getpid())
                p.start()
                process_list.append(p)
                #p.join()

            i = 1
            for proc in process_list:
                print('I am hanged here and my process id is : ', os.getpid())
                proc.join()
                print('I have been joined and my process id is : ', os.getpid())
                i += 1

            for value in range(Output_Queue.qsize()):
                print(Output_Queue.get())

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

            #Here I am again printing the Input Queue which should be empty logically.
            for frame in range(Input_Queue.qsize()):
                frame_no, _, _ = Input_Queue.get()
                print(frame_no)

            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()

#chec_queues()
#compare_images()
#fps_finder()
alternative_process('user_verification_2.avi')#'hassan_checking.avi'

1 Ответ

0 голосов
/ 29 ноября 2018

Ваш код содержит while not input_queue.empty().Я предполагаю, что во время работы input_queue становится пустым, в то время как цикл останавливается, и затем вы добавляете что-то еще к input_queue, чтобы обработать это что-то еще.Но сейчас уже слишком поздно.

Обычно вы работаете с такими очередями:

while True:
    element = my_queue.get()
    ...

Чтобы остановить этот цикл, вы можете посчитать количество обработанных элементов, использовать аргумент timeout или killпроцесс при некоторых условиях.Другой вариант - использовать multiprocessing.Pool или concurrent.futures.ProcessPoolExecutor.

Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.
...