Python multiprocessing returns queue is empty when actually it is not

Question

In this program, after some iterations all the processes terminate which means that input_queue is empty as per the condition in target function. But after returning to the main function when I print the input_queue there are still items left in that queue, then why those multiple processes terminated at first place?

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'

[how-to-use-multiprocessing-queue-get-method](https://stackoverflow.com/q/53132498/9059420) — Darkonaut, Nov 29 '18 at 15:48

score 0 · Answer 1 · answered Nov 29 '18 at 13:40

0

Your code contains while not input_queue.empty(). I guess during the work input_queue becomes empty, while loop stops and then you add something else to input_queue to process this something else. But it's too late at that moment.

Usually you work with queues like this:

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

To stop this loop you may count number of treated elements, use timeout argument or kill process under some condition. Another option is to use multiprocessing.Pool or concurrent.futures.ProcessPoolExecutor.

answered Nov 29 '18 at 13:40

No I'm not adding anything extra into Input Queue after that while loop in the target function. However I was to able to get that right by using Manager.queue but I still don't know why the earlier one was not working and why using manager queues resolves the error. – Muhammad Hassan Nov 29 '18 at 14:01
@Muhammadhassan Looking at the code I see that you do `p.start()` before `Input_Queue.put((frame_no, rgb_small_frame, face_locations))`. – Nov 29 '18 at 14:17
No, the "if not ret" condition makes sure that, no items are added into the input queue after processes start. – Muhammad Hassan Nov 30 '18 at 05:57
@Muhammadhassan The logic behind the code is obscured to me. What I see is the following - if `ret` becomes `False` after some time then you may get unlimited amount of frames in the `input_queue` because once every 20 steps `process_this_frame` is `True` (or 5 out of 20?). Please understand the logic of your own code. I guess you should re-organize the code. – Nov 30 '18 at 06:52

Python multiprocessing returns queue is empty when actually it is not

1 Answers1