Parallel processing in (regular) python is a bit of a pain: in other languages we'd just use threads but the GIL makes that problematic, and using multiprocessing has a big overhead in moving data around. I've found that fine-grained parallelism is (relatively) hard to do, whilst processing 'chunks' of work that take 10's of seconds (or more) to process in a single process can be much more straight-forward.

An easier path to parallel processing your problem - if you're on a UNIXy system - would be to make a python program which processes a segment of video specified on the command-line (i.e. a frame number to start with, and a number of frames to process), and then use the GNU parallel tool to process multiple segments at once. A second python program can consolidate the results from a collection of files, or reading from stdin, piped from parallel. This way means that the processing code doesn't need to do it's own parallelism, but it does require the input file to be multiply accessed and to extract frames starting from mid-points. (This might also be extendable to work across multiple machines without changing the python...)

Using multiprocessing.Pool.map could be used in a similar way if you need a pure-python solution: map over a list of tuples (say, (file, startframe, endframe)) and then open the file in the function and process that segment.

There is indeed a mistake in the code : the use of the same VideoCapture object across processes. Obviously there's a conflict on the position currently being read in the file.

This being said, when trying to instantiate one VideoCapture per process, my interpreter crashes (tested with python3.4.2 + opencv3.0.0-beta, and python2.7.6 + opencv2.4.8). Here's my try so far if you want to check it / go further.

import cv2
import multiprocessing as mp

def getFrame(queue, startFrame, endFrame):
    cap = cv2.VideoCapture(file)  # crashes here
    print("opened capture {}".format(mp.current_process()))
    for frame in range(startFrame, endFrame):
        # cap.set(cv2.CAP_PROP_POS_FRAMES, frame)  # opencv3            
        cap.set(cv2.cv.CV_CAP_PROP_POS_FRAMES, frame)
        # frameNo = int(cap.get(cv2.CAP_PROP_POS_FRAMES))  # opencv3
        frameNo = int(cap.get(cv2.cv.CV_CAP_PROP_POS_FRAMES))
        ret, f = cap.read()
        if ret:
            print("{} - put ({})".format(mp.current_process(), frameNo))
            queue.put((frameNo, f))
    cap.release()

file = "video.mov"
capture_temp = cv2.VideoCapture(file)
# fileLen = int((capture_temp).get(cv2.CAP_PROP_FRAME_COUNT))  # opencv3
fileLen = int((capture_temp).get(cv2.cv.CV_CAP_PROP_FRAME_COUNT))
capture_temp.release()

# get cpuCount for processCount
# processCount = mp.cpu_count() / 3
processCount = 2

inQ1 = mp.JoinableQueue()  # not sure if this is right queue type, but I also tried mp.Queue()
inQ2 = mp.JoinableQueue()
qList = [inQ1, inQ2]

# set up bunches
bunches = []
for startFrame in range(0, fileLen, int(fileLen / processCount)):
    endFrame = startFrame + int(fileLen / processCount)
    bunches.append((startFrame, endFrame))

getFrames = []
for i in range(processCount):
    getFrames.append(mp.Process(target=getFrame, args=(qList[i], bunches[i][0], bunches[i][1])))

for process in getFrames:
    process.start()

results1 = [inQ1.get() for p in range(bunches[0][0], bunches[0][1])]
results2 = [inQ2.get() for p in range(bunches[1][0], bunches[1][1])]

inQ1.close()
inQ2.close()

for process in getFrames:
    process.terminate()
    process.join()

OpenCV has an example for multithreaded video processing in their github repo.

https://github.com/opencv/opencv/blob/master/samples/python/video_threaded.py

Import ThreadPool from multiprocessing.pool and kick off a new thread for every cpu core. OpenCV has a function called getNumberOfCPUs()

Example:

from __future__ import print_function

import numpy as np
import cv2 as cv

from multiprocessing.pool import ThreadPool
from collections import deque

from common import clock, draw_str, StatValue
import video


class DummyTask:
    def __init__(self, data):
    self.data = data
def ready(self):
    return True
def get(self):
    return self.data

if __name__ == '__main__':
    import sys

    print(__doc__)

    try:
        fn = sys.argv[1]
    except:
        fn = 0
    cap = video.create_capture(fn)


    def process_frame(frame, t0):
    # some intensive computation...
        frame = cv.medianBlur(frame, 19)
        frame = cv.medianBlur(frame, 19)
        return frame, t0

    threadn = cv.getNumberOfCPUs()
    pool = ThreadPool(processes = threadn)
    pending = deque()

    threaded_mode = True

    latency = StatValue()
    frame_interval = StatValue()
    last_frame_time = clock()
    while True:
        while len(pending) > 0 and pending[0].ready():
            res, t0 = pending.popleft().get()
            latency.update(clock() - t0)
            draw_str(res, (20, 20), "threaded      :  " +             str(threaded_mode))
            draw_str(res, (20, 40), "latency        :  %.1f ms" %     (latency.value*1000))
            draw_str(res, (20, 60), "frame interval :  %.1f ms" % (frame_interval.value*1000))
            cv.imshow('threaded video', res)
        if len(pending) < threadn:
            ret, frame = cap.read()
            t = clock()
            frame_interval.update(t - last_frame_time)
            last_frame_time = t
            if threaded_mode:
                task = pool.apply_async(process_frame, (frame.copy(), t))
            else:
                task = DummyTask(process_frame(frame, t))
            pending.append(task)
        ch = cv.waitKey(1)
        if ch == ord(' '):
            threaded_mode = not threaded_mode
        if ch == 27:
            break
    cv.destroyAllWindows()

You should be able to use that example code and place your image processing in the process_frame function.

Add a counter to in your loop and call cv2.imshow when count == X

There are too many aspects of making things faster, I'll only focus on the multiprocessing part.

As you don't want to read the whole video at a time, we have to read the video frame by frame.

I'll be using opencv (cv2), numpy for reading the frames, calculating mse, and saving the mse to disk.

First, we can start without any multiprocessing so we can benchmark our results. I'm using a video of 1920 by 1080 dimension, 60 FPS, duration: 1:29, size: 100 MB.

import cv2
import sys
import time

import numpy as np
import subprocess as sp
import multiprocessing as mp

filename = '2.mp4'

def process_video():    
    cap = cv2.VideoCapture(filename)

    proc_frames = 0

    mse = []
    prev_frame = None
    ret = True
    while ret:
        ret, frame = cap.read() # reading frames sequentially
        if ret == False:
            break

        if not (prev_frame is None):
            c_mse = np.mean(np.square(prev_frame-frame))
            mse.append(c_mse)

        prev_frame = frame

        proc_frames += 1

    np.save('data/' + 'sp' + '.npy', np.array(mse))

    cap.release()
    return


if __name__ == "__main__":

    t1 = time.time()

    process_video()

    t2 = time.time()

    print(t2-t1)

In my system, it runs for 142 secs.

Now, we can take the multiprocessing approach. The idea can be summarized in the following illustration.

GIF credit: Google

We make some segments (based on how many cpu cores we have) and process those segmented frames in parallel.

import cv2
import sys
import time

import numpy as np
import subprocess as sp
import multiprocessing as mp

filename = '2.mp4'

def process_video(group_number):    
    cap = cv2.VideoCapture(filename)
    num_processes = mp.cpu_count()
    frame_jump_unit = cap.get(cv2.CAP_PROP_FRAME_COUNT) // num_processes
    cap.set(cv2.CAP_PROP_POS_FRAMES, frame_jump_unit * group_number)
    proc_frames = 0

    mse = []
    prev_frame = None
    while proc_frames < frame_jump_unit:
        ret, frame = cap.read()
        if ret == False:
            break

        if not (prev_frame is None):
            c_mse = np.mean(np.square(prev_frame-frame))
            mse.append(c_mse)

        prev_frame = frame

        proc_frames += 1

    np.save('data/' + str(group_number) + '.npy', np.array(mse))

    cap.release()
    return


if __name__ == "__main__":

    t1 = time.time()

    num_processes =  mp.cpu_count()
    print(f'CPU: {num_processes}')

    # only meta-data
    cap = cv2.VideoCapture(filename)

    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    fps = cap.get(cv2.CAP_PROP_FPS)
    frame_jump_unit = cap.get(cv2.CAP_PROP_FRAME_COUNT) // num_processes
    cap.release()

    p = mp.Pool(num_processes)
    p.map(process_video, range(num_processes))

    # merging



    # the missing mse will be 

    final_mse = []
    for i in range(num_processes):
        na = np.load(f'data/{i}.npy')
        final_mse.extend(na)


        try:
            cap = cv2.VideoCapture(filename) # you could also take it outside the loop to reduce some overhead
            frame_no = (frame_jump_unit) * (i+1) - 1
            print(frame_no)
            cap.set(1, frame_no)
            _, frame1 = cap.read()
            #cap.set(1, ((frame_jump_unit) * (i+1)))
            _, frame2 = cap.read()
            c_mse = np.mean(np.square(frame1-frame2))
            final_mse.append(c_mse)
            cap.release()
        except:
            print('failed in 1 case')
            # in the last few frames, nothing left
            pass




    t2 = time.time()

    print(t2-t1)

    np.save(f'data/final_mse.npy', np.array(final_mse))

I'm using just numpy save to save the partial results, you can try something better.

This one runs for 49.56 secs with my cpu_count = 12. There are definitely some bottlenecks that can be avoided to make it run faster.

The only issue with my implementation is, it's missing the mse for regions where the video was segmented, it's pretty easy to add. As we can index individual frames at any location with OpenCV in O(1), we can just go to those locations and calculate mse separately and merge to the final solution. [Check the updated code it fixes the merging part]

You can write a simple sanity check to ensure, both provide the same result.

import numpy as np

a = np.load('data/sp.npy')

b = np.load('data/final_mse.npy')

print(a.shape)

print(b.shape)

print(a[:10])

print(b[:10])

for i in range(len(a)):
    if a[i] != b[i]:
        print(i)

Now, some additional speedups can come from using a CUDA-compiled opencv, ffmpeg, adding queuing mechanism plus multiprocessing, etc.

Interesting application! Re the error, I run your code and the message is that it can't pickle VideoCapture object, see the links below, that's probably why the receiving pipe is empty. There are two errors from two threads: the first is the pickle, then comes the EOF.

EDIT #2: I managed to run it with one process per video etc.:

Regarding performance, I first did it without merging the images (I had to fix some details) to see whether it received, and for 3 and 4 frames, displayed in separated windows from the receiving threads it played very fast, faster than real time (tested with 3-4 streams). I think the merging and resizing for display are slow, the picture gets 2560x1440 for 4 streams (4x1280x720). In my case it's resized to fit the screen.

Thanks for sharing that question and that library etc!

(BTW, I tried with a lock as well, initially but it happened that it is not necessary. The code needs cleaning of some experiments. Also, the current implementation might not be synchronized per frame for each stream, because it doesn't join per frame as your original example, which created new processes to grab one frame from each and then merge them.)

The CPU load is mainly in the main process (a 4-core CPU, thus max=25% per instance):

Some times:

0.06684677699999853 0.030737616999999773 1.2829999995744856e-06 LEN(FRAMES)= 9 0.06703700200000284 0.030708104000002123 6.409999997458726e-07 LEN(FRAMES)= 9

The waitKey in the main loop can be tweaked.

Code

https://github.com/Twenkid/Twenkid-FX-Studio/blob/master/Py/YoutubeAggregatorPafy/y6.py

# Merging Youtube streams with pafy, opencv and multithreading
# Base code by Fraser Langton - Thanks!
# Refactored and debugged by Twenkid
# version y6 - more cleaning of unused code, properly close VideoCapture in the processes

import multiprocessing #Process, Lock
from multiprocessing import Lock # Not needed
import cv2
import numpy as np
import pafy
import typing
import timeit
import time

urls = [
    "https://www.youtube.com/watch?v=tT0ob3cHPmE",
    "https://www.youtube.com/watch?v=XmjKODQYYfg",
    "https://www.youtube.com/watch?v=E2zrqzvtWio",

    "https://www.youtube.com/watch?v=6cQLNXELdtw",
    "https://www.youtube.com/watch?v=s_rmsH0wQ3g",
    "https://www.youtube.com/watch?v=QfhpNe6pOqU",

    "https://www.youtube.com/watch?v=C_9x0P0ebNc",
    "https://www.youtube.com/watch?v=Ger6gU_9v9A", 
    "https://www.youtube.com/watch?v=39dZ5WhDlLE"
]

# Merging seems to require equal number of sides, so 2x2, 3x3 etc. The  resolutions should be the same.
'''
[    
    "https://www.youtube.com/watch?v=C_9x0P0ebNc",
    "https://www.youtube.com/watch?v=Ger6gU_9v9A",
    "https://www.youtube.com/watch?v=39dZ5WhDlLE",   
    "https://www.youtube.com/watch?v=QfhpNe6pOqU",
]
'''

width = np.math.ceil(np.sqrt(len(urls)))
dim = 1920, 1080

streams = []
#bestStreams = []

def main():
    global bestStreams
    streams = [pafy.new(url).getbest() for url in urls]
    print(streams)
    #[bestStreams for best in streams]
    #print(bestStreams)
    cv2.waitKey(0)
    videos = [cv2.VideoCapture() for streams in streams]
    bestURLS = [] 
    #[video.open(best.url) for video, best in zip(videos, streams)]  # Opened per process
    [bestURLS.append(best.url) for best in streams]
    
    #[ for video, best in zip(videos, streams)]
    print(bestURLS)
    cv2.waitKey(0)
    cv2.namedWindow('Video', cv2.WINDOW_FREERATIO)
    cv2.setWindowProperty('Video', cv2.WND_PROP_FULLSCREEN, cv2.WINDOW_FULLSCREEN)
    LOCK = Lock()
    #proc = get_framesUL(bestStreams, LOCK)
    #proc, pipes = get_framesULJ(bestStreams, LOCK)
    proc, pipes = get_framesULJ(bestURLS, LOCK)     
    print("PROC, PIPES", proc, pipes)
    #cv2.waitKey(0)
    frames = []
    numStreams = len(streams)
    while True:
        start_time = timeit.default_timer()
        # frames = [cv2.resize(video.read()[-1], (dim[0] // width, dim[1] // width)) for video in videos]
        #frames = get_frames(videos, LOCK)
        #frames = get_framesUL(streams, LOCK)
        
        
        print(timeit.default_timer() - start_time)

        start_time = timeit.default_timer()
        
        frames = [x.recv() for x in pipes]
        lf = len(frames)
        print("LEN(FRAMES)=", lf);
        #if lf<3: time.sleep(3); print("LEN(FRAMES)=", lf); #continue #Else merge and show
        #proc.join()
        #elif lf==3: frames = [x.recv() for x in pipes]
                
        dst = merge_frames(frames)
        print(timeit.default_timer() - start_time)
         
        start_time = timeit.default_timer()      
        #if cv2!=None:
        try:
          cv2.imshow('Video', dst)
        except: print("Skip")
        #cv2.waitKey(1)  

        if cv2.waitKey(20) & 0xFF == ord('e'):
            break
        print(timeit.default_timer() - start_time)

        continue
        
    for proc in jobs:
        proc.join()
        
    # [video.release() for video in videos] # Per process
    cv2.destroyAllWindows()



def get_framesULJ(videosURL, L): #return the processes, join in main and read the frames there
    # frames = [video.read()[-1] for video in videos]
    print("get_framesULJ:",videosURL)    
    jobs = []
    pipe_list = []
    #print("VIDEOS:",videosURL)    
    #for video in videos:
    for videoURL in videosURL: #urls:
        recv_end, send_end = multiprocessing.Pipe(False)
        print(recv_end, send_end)
        p = multiprocessing.Process(target=get_frame2L, args=(videoURL, send_end, L))
        #p = multiprocessing.Process(target=get_frame, args=(video, send_end, L))
        #if (p==None): continue
        print("P = ", p)
        #time.sleep(0.001)
        jobs.append(p)
        print("JOBS, len", jobs, len(jobs))                
        pipe_list.append(recv_end)
        print("pipe_list", pipe_list)               
        p.start()
        #cv2.waitKey(0)

    #for proc in jobs:
    #    proc.join()

    #frames = [x.recv() for x in pipe_list]
    #return frames
    #cv2.waitKey(0)
    return jobs, pipe_list

def get_frame2L(videoURL, send_end, L):
    v = cv2.VideoCapture()
    #[video.open(best.url)
    #L.acquire()
    v.open(videoURL)
    print("get_frame2", videoURL, v, send_end)
    #cv2.waitKey(0)
    while True:      
      ret, frame = v.read()
      if ret: send_end.send(frame); #cv2.imshow("FRAME", frame); cv2.waitKey(1)   
      else: print("NOT READ!"); break
    #send_end.send(v.read()[1])
    #L.release()
    
def get_framesUL(videosURL, L):
    # frames = [video.read()[-1] for video in videos]

    jobs = []
    pipe_list = []
    print("VIDEOS:",videosURL)    
    #for video in videos:
    for videoURL in videosURL: #urls:
        recv_end, send_end = multiprocessing.Pipe(False)
        print(recv_end, send_end)
        p = multiprocessing.Process(target=get_frame2L, args=(videoURL, send_end, L))
        #p = multiprocessing.Process(target=get_frame, args=(video, send_end, L))
        #if (p==None): continue
        print("P = ", p)
        #time.sleep(0.001)
        jobs.append(p)
        print("JOBS, len", jobs, len(jobs))                
        pipe_list.append(recv_end)
        print("pipe_list", pipe_list)               
        p.start()

    for proc in jobs:
        proc.join()

    frames = [x.recv() for x in pipe_list]
    return frames


def get_frames(videos, L):
    # frames = [video.read()[-1] for video in videos]

    jobs = []
    pipe_list = []
    print("VIDEOS:",videos)    
    for video in videos:
        recv_end, send_end = multiprocessing.Pipe(False)
        print(recv_end, send_end)
        p = multiprocessing.Process(target=get_frame, args=(video, send_end, L))
        #p = multiprocessing.Process(target=get_frame, args=(video, send_end, L))
        #if (p==None): continue
        print("P = ", p)
        #time.sleep(0.001)
        jobs.append(p)
        print("JOBS, len", jobs, len(jobs))                
        pipe_list.append(recv_end)
        print("pipe_list", pipe_list)               
        p.start()

    for proc in jobs:
        proc.join()

    frames = [x.recv() for x in pipe_list]
    return frames
    
def get_frame(video, send_end, L):
    L.acquire()
    print("get_frame", video, send_end)
    send_end.send(video.read()[1])
    L.release()
    # send_end.send(cv2.resize(video.read()[1], (dim[0] // width, dim[1] // width)))

    
def get_frame2(videoURL, send_end):
    v = video.open(videoURL)       
    while True:
      ret, frame = v.read()
      if ret: send_end.send(frame)
      else: break
      
    
def merge_frames(frames: typing.List[np.ndarray]):
    #cv2.imshow("FRAME0", frames[0]) ########## not images/small
    #cv2.imshow("FRAME1", frames[1]) ##########
    #cv2.imshow("FRAME2", frames[2]) ##########
    #cv2.imshow("FRAME3", frames[3]) ##########
    #cv2.waitKey(1)
    width = np.math.ceil(np.sqrt(len(frames)))
    rows = []
    for row in range(width):
        i1, i2 = width * row, width * row + width
        rows.append(np.hstack(frames[i1: i2]))
    
    
    return np.vstack(rows)


if __name__ == '__main__':
    main()

EDIT #1 IDEA: Creating one process per video stream and reading it in a loop (pumping in a pipe), instead of a new process for each frame, and/thus opening the videos/VideoCapture objects with a videoURL through the pipe, instead of sending a VideoCapture object. (I don't know if it has the same pickle issue in this form)

...
in main:
bestURLS = []
proc, pipes = get_framesULJ(bestURLS, LOCK) 
[bestURLS.append(best.url) for best in streams]



def get_frame2(videoURL, send_end):
    v = video.open(videoURL)       
    while True:
      ret, frame = v.read()
      if ret: send_end.send(video)
      else: break

 def get_framesULJ(videosURL, L): #return the processes, join in main and read the frames there
print("get_framesULJ:",videosURL)    
jobs = []
pipe_list = []
for videoURL in videosURL:
    recv_end, send_end = multiprocessing.Pipe(False)
    print(recv_end, send_end)
    p = multiprocessing.Process(target=get_frame2L, args=(videoURL, send_end, L))       
    print("P = ", p)
    jobs.append(p)
    print("JOBS, len", jobs, len(jobs))                
    pipe_list.append(recv_end)
    print("pipe_list", pipe_list)               
    p.start()

return jobs, pipe_list

Original answer:

<multiprocessing.connection.PipeConnection object at 0x000000000D3C7D90> <multip
rocessing.connection.PipeConnection object at 0x000000000D3BD2E0>
Traceback (most recent call last):
  File "y.py", line 104, in <module>
    main()
  File "y.py", line 48, in main
    frames = get_frames(videos)
  File "y.py", line 80, in get_frames
    p.start()
  File "C:\Program Files\Python38\lib\multiprocessing\process.py", line 121, in
start
    self._popen = self._Popen(self)
  File "C:\Program Files\Python38\lib\multiprocessing\context.py", line 224, in
_Popen
    return _default_context.get_context().Process._Popen(process_obj)
  File "C:\Program Files\Python38\lib\multiprocessing\context.py", line 326, in
_Popen
    return Popen(process_obj)
  File "C:\Program Files\Python38\lib\multiprocessing\popen_spawn_win32.py", lin
e 93, in __init__
    reduction.dump(process_obj, to_child)
  File "C:\Program Files\Python38\lib\multiprocessing\reduction.py", line 60, in
 dump
    ForkingPickler(file, protocol).dump(obj)
TypeError: cannot pickle 'cv2.VideoCapture' object

Z:\>Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\Program Files\Python38\lib\multiprocessing\spawn.py", line 116, in sp
awn_main
    exitcode = _main(fd, parent_sentinel)
  File "C:\Program Files\Python38\lib\multiprocessing\spawn.py", line 126, in _m
ain
    self = reduction.pickle.load(from_parent)
EOFError: Ran out of input

It fails before p.start. The instances are created and the structures seems OK:

VIDEOS: [<VideoCapture 000000000D418710>, <VideoCapture 000000000D4186F0>, <Vide
oCapture 000000000D418B70>]
<multiprocessing.connection.PipeConnection object at 0x000000000D3C3D90> <multip
rocessing.connection.PipeConnection object at 0x000000000D3B62E0>
P =  <Process name='Process-1' parent=8892 initial>
JOBS, len [<Process name='Process-1' parent=8892 initial>] 1
RECV_END <multiprocessing.connection.PipeConnection object at 0x000000000D3C3D90
>

See the module pickle:

https://docs.python.org/3/library/pickle.html

It seems not everything can be "pickled".

What can be pickled and unpickled?

The following types can be pickled:

None, True, and False

integers, floating point numbers, complex numbers

strings, bytes, bytearrays

tuples, lists, sets, and dictionaries containing only picklable objects

functions defined at the top level of a module (using def, not lambda)

built-in functions defined at the top level of a module

classes that are defined at the top level of a module

instances of such classes whose __dict__ or the result of calling __getstate__() is picklable (see section Pickling Class Instances for details).

Besides it seems there was a bug in opencv causing that. One of the solution given is to turn the multiprocessing off...

Python multiprocess can't pickle opencv videocapture object

https://github.com/MVIG-SJTU/AlphaPose/issues/164

Fang-Haoshu commented on 17 Oct 2018

This bug is due to multi-processing in opencv. --sp disable multi-processing. BTW, can you tell me the version of opencv that you are using?

I guess something about locked memory or something.

A workaround I would try would be to first dump the pixels of the object as plain data or raw something, maybe with a header about the size etc.

Also, in general, for smoother play I think some buffering needs to be added.

BTW, what version is your openCV? Mine is 4.2.0

You may stick to the ProcessPoolExecutor Example.

The example is as follows:

def main():
    with concurrent.futures.ProcessPoolExecutor() as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            print('%d is prime: %s' % (number, prime))

if __name__ == '__main__':
    main()

• You must use the if __name__ == '__main__': syntax.
  Place the preparation code as cap_list = [cv2.VideoCapture(k) for k in names] in the main() method.
  Without it, the entire script is executed multiple times, and gets stuck.
• The example uses a method, and not an object.
  I suppose you can map object methods, but I couldn't make it work.
• When using Python 3.6, I am getting an error: "TypeError: can't pickle cv2.VideoCapture objects".
  I am passing the file name to display_frames, and execute cv2.VideoCapture at the beginning of display_frames method.
• The code you have posted is not using class VideoCapture.

For testing:

• My example generates synthetic video files (in order to make the example "self contained").
• The example shows the video frames for testing.

Here is the code:

import cv2
import numpy as np
import concurrent.futures

def display_frames(name):
    cap = cv2.VideoCapture(name)

    while True:
        ret, frames = cap.read()

        if ret is False:
            break

        # Show frames for testing:
        cv2.imshow(str(cap), frames)
        cv2.waitKey(100)

    cap.release()

    return name


def main():
    names = ['test2.mp4', 'test3.mp4']

    # Generate two synthetic video files to be used as input:
    ###############################################################################
    width, height, n_frames = 640, 480, 30  # 30 frames, resolution 640x480

    intput_filename1 = names[0]
    intput_filename2 = names[1]

    # Use MPEG4 codec (for testing)
    synthetic_out = cv2.VideoWriter(intput_filename1, cv2.VideoWriter_fourcc(*'mp4v'), 25, (width, height))

    for i in range(n_frames):
        img = np.full((height, width, 3), 60, np.uint8)
        cv2.putText(img, str(i+1), (width//2-100*len(str(i+1)), height//2+100), cv2.FONT_HERSHEY_DUPLEX, 10, (30, 255, 30), 20)  # Green number
        synthetic_out.write(img)

    synthetic_out.release()

    width, height, n_frames = 320, 240, 20 # 20 frames, resolution 320x240
    synthetic_out = cv2.VideoWriter(intput_filename2, cv2.VideoWriter_fourcc(*'mp4v'), 25, (width, height))

    for i in range(n_frames):
        img = np.full((height, width, 3), 60, np.uint8)
        cv2.putText(img, str(i+1), (width//2-50*len(str(i+1)), height//2+50), cv2.FONT_HERSHEY_DUPLEX, 5, (255, 30, 30), 10)  # Blue number
        synthetic_out.write(img)

    synthetic_out.release()
    ###############################################################################


    with concurrent.futures.ProcessPoolExecutor(max_workers=3) as executor:
        for name in executor.map(display_frames, names):
            print(name)

    cv2.destroyAllWindows() # For testing


# Using Python 3.6 there is an error: "TypeError: can't pickle cv2.VideoCapture objects"
if __name__ == '__main__':
    main()

I don't really know if it answers you question, because not much is left from your original code.
I hope it helps...

Here's a minimal working example with optional FPS control. If you need to extract frames from your process back to the main program, you can use multiprocessing.Queue() to transfer frames since multiprocesses have an independent memory stack.

import multiprocessing as mp
import cv2, time

def capture_frames():
    src = 'test.mp4'
    capture = cv2.VideoCapture(src)
    capture.set(cv2.CAP_PROP_BUFFERSIZE, 2)

    # FPS = 1/X, X = desired FPS
    FPS = 1/120
    FPS_MS = int(FPS * 1000)

    while True:
        # Ensure camera is connected
        if capture.isOpened():
            (status, frame) = capture.read()
            
            # Ensure valid frame
            if status:
                cv2.imshow('frame', frame)
            else:
                break
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
            time.sleep(FPS)

    capture.release()
    cv2.destroyAllWindows()

if __name__ == '__main__':
    print('Starting video stream')
    capture_process = mp.Process(target=capture_frames, args=())
    capture_process.start()

Related camera/IP/RTSP/streaming, FPS, video, threading, and multiprocessing posts

1. Python OpenCV streaming from camera - multithreading, timestamps

2. Video Streaming from IP Camera in Python Using OpenCV cv2.VideoCapture

3. How to capture multiple camera streams with OpenCV?

4. OpenCV real time streaming video capture is slow. How to drop frames or get synced with real time?

5. Storing RTSP stream as video file with OpenCV VideoWriter

6. OpenCV video saving

7. Python OpenCV multiprocessing cv2.VideoCapture mp4

Although it is not exactly what you are asking for, I was curious my-self about decomposing the image into sub-images for further processing so I put together this example. It shows how to thread or multi-process. You should be able to figure out how to decompose in time as opposed to space.

Basically, fill the FRAME_QUEUE with frames, not chunks.

import cv2
import numpy as np
import multiprocessing
import threading
import time
import argparse

FRAME_QUEUE = multiprocessing.JoinableQueue()
RESULT_QUEUE = multiprocessing.Queue()

# defaults
NWORKERS = 2
PARALLEL_TYPE = 'thread'
FONT = cv2.FONT_HERSHEY_SIMPLEX


def process_frame(frame):
    # Convert BGR to HSV
    hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)

    lower_red = np.array([150, 150, 50], np.uint8)
    upper_red = np.array([180, 255, 150], np.uint8)

    # Threshold the HSV image to get only blue colors
    # finding the range of red
    mask = cv2.inRange(hsv, lower_red, upper_red)
    res = cv2.bitwise_and(frame, frame, mask=mask)

    kernel = np.ones((5, 5), np.uint8)
    mask = cv2.dilate(mask, kernel)
    res = cv2.bitwise_and(frame, frame, mask=mask)

    _, contours, hierarchy = cv2.findContours(mask, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)

    for pic, contour in enumerate(contours):
        area = cv2.contourArea(contour)
        if(area > 1000):
            x, y, w, h = cv2.boundingRect(contour)
            frame = cv2.rectangle(frame, (x, y), (x+w, y+h), (0, 0, 255), 2)

    return frame, mask, res


def parallel_process_frame(cls):
    """the main function that perfroms the image processing"""
    while True:
        next_frame = FRAME_QUEUE.get()
        if next_frame is None:
            # Poison pill means shutdown
            break

        # process the frame
        i, frame = next_frame
        frame, mask, res = process_frame(frame)

        # draw worker on frame
        frame = cv2.putText(frame, '{}'.format(i),
                            (0, 20), FONT,
                            0.5, (255, 0, 0), 1, cv2.LINE_AA)

        # tell the queue that we are done
        FRAME_QUEUE.task_done()

        # place the results in the result queue
        RESULT_QUEUE.put((i, (frame, mask, res)))


class ParallelProcessWorker(multiprocessing.Process):
    """multiprocess worker"""
    def __init__(self):
        multiprocessing.Process.__init__(self)

    def run(self):
        parallel_process_frame(self)


class ParallelThreadWorker(threading.Thread):
    """multithread worker"""
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        parallel_process_frame(self)


def serial(frame):
    """serial process the frame"""
    return process_frame(frame)


def multi(frame):
    """multithread/multiprocess process the frame"""
    # split the frame and place in the queue
    for i, chunk in enumerate(np.split(frame, NWORKERS)):
        FRAME_QUEUE.put((i, chunk))

    # wait for the chunks to finish
    FRAME_QUEUE.join()

    # collect the results
    results = []
    for i in range(NWORKERS):
        results.append(RESULT_QUEUE.get())

    # sort, because they can come back in any order
    results.sort(key=lambda x: x[0])

    # combine chunks
    frame = np.vstack((r[1][0] for r in results))
    mask = np.vstack((r[1][1] for r in results))
    res = np.vstack((r[1][2] for r in results))

    return frame, mask, res


if __name__ == "__main__":

    parser = argparse.ArgumentParser(
        prog='Parallel',
        description='Program parallel video processing')

    ARG = parser.add_argument
    ARG('-m', '--method', metavar='PAR', action='store', default=None,
        choices=['serial', 'thread', 'process'])
    ARG('-w', '--workers', metavar='N', action='store', default=2,
        help='specify the number of workers')

    args = parser.parse_args()

    # set flags
    PARALLEL_TYPE = args.method
    NWORKERS = int(args.workers)

    # setup parallel scheme
    if PARALLEL_TYPE != 'serial':
        method = multi
        if PARALLEL_TYPE == 'thread':
            worker = ParallelThreadWorker
        else:
            worker = ParallelProcessWorker
        # start the workers
        for i in range(NWORKERS):
            w = worker()
            w.start()
    else:
        method = serial

    # variables

    frame_count = 0
    start = time.time()

    # start capture
    cap = cv2.VideoCapture(0)

    # start the loop
    while(True):
        frame_count += 1

        # Take each frame
        _, frame = cap.read()

        frame, mask, res = method(frame)

        FPS = frame_count/(time.time()-start)

        frame = cv2.putText(frame, 'FPS: {0:.1f}'.format(FPS),
                            (0, 20), FONT,
                            0.5, (0, 255, 0), 1, cv2.LINE_AA)

        cv2.imshow('frame', frame)
        # cv2.imshow('mask', mask)
        # cv2.imshow('res', res)

        k = cv2.waitKey(5) & 0xFF
        if k == 27:
            break
    cap.release()
    cv2.destroyAllWindows()

    # place a bunch of Nones to shutdown the threads/process
    [FRAME_QUEUE.put(None) for i in range(NWORKERS*2)]

    # report
    print('Using a {0} approach with {1} workers resulted in an average FPS of {2:.1f}'.format(PARALLEL_TYPE, NWORKERS, FPS))

You can specify the parallel scheme and workers like this:

python camera_parallel.py -m serial
python camera_parallel.py -m thread -w 4
python camera_parallel.py -m process -w 2

On my machine (MacBook Pro, 2009, Core2Duo), serial gives me the highest frame rate.