ジェネレーターでmultiprocessingに送るデータを順次用意する

時間のかかるバッチ処理はマルチプロセスで短縮すべきです。でも、マルチプロセスってなんかわかりづらいです。

imageprocessing.py

import glob
import os.path
import cv2

dir = r'c:\data\image\'
fl = glob.glob(os.path.join(dir,'**/*'),recursive = True)
for i in fl:
    im = cv2.imread(i)
    im = imageprocess(im)
    cv2.imwrite(i, im)

こういう処理があったとして、マルチプロセスにするにはどうすればよいでしょうか。

imageprocessingmp1.py

import glob
import os.path
import cv2
from multiprocessing import Pool

def func(file):
    im = cv2.imread(file)
    im = imageprocess(im)
    cv2.imwrite(file, im)

dir = r'c:\data\image\'
fl = glob.glob(os.path.join(dir,'**/*'),recursive = True)

with Pool(4) as p:
    for _ in p.imap_unordered(func,fl):
        pass

こうなりますね。
ですが、これには少し問題があるかもしれません。
と、言うのも、これは計算結果をそのままディスクに書き出しているのですが、この部分はマルチプロセス部分なので、同時に複数の読み込みと書き出しが起きています。これはあまりよろしくはありません。出来ればそういう所はワーカープロセスでなく制御する部分に置きたいです。

imageprocessingmp2.py

import glob
import os.path
import cv2
from multiprocessing import Pool

def func(args):
    return args[0],imageprocess(args[1])

dir = r'c:\data\image\'
fl = glob.glob(os.path.join(dir,'**/*'),recursive = True)

with Pool(4) as p:
    for result in p.imap_unordered(func,[i,cv2.imread(i) for i in fl]):
         cv2.imwrite(result[0], result[1])

でもこれではダメです。最初に全部のデータをディスクから読み込むので、メモリがアホみたいに必要になります。
どうすればいいでしょう

Queueで制御する？、そう考えましたがはっきり言って面倒です。

imageprocessingmp3.py

import glob
import os.path
import cv2
from multiprocessing import Pool

def func(args):
    return args[0],imageprocess(args[1])

def files(fl):
    for i in fl:
         im = cv2.imread(i)
         yield [i,im]

dir = r'c:\data\image\'
fl = glob.glob(os.path.join(dir,'**/*'),recursive = True)

with Pool(4) as p:
    for result in p.imap_unordered(func,files(fl)):
         cv2.imwrite(result[0], result[1])

そこで、ジェネレータを使用すると、必要な時に読み込むようになります。
(送るデータが少ない場合は有る程度まとめて実行されるようです)