前回に続き、ラズベリーパイ4B (8G, bookworm(x64)) で、multiprocessingのManagerを使い、ndarrayを共有します。ただ今回は、500MBという巨大なndarrayを共有します。

最初に、core1で4000w x 3000h の画像ndarrayを、vconcatとhconcatで結合し、巨大ndarrayを作成します。同じcore1で、この巨大ndarrayを「Managerのlist」に書き込みます。次にcore4で、「Managerのlist」から巨大ndarrayを読み出します。最後に、同じcore4で、巨大ndarrayを縮小して表示します。以下がコードです。

動かしてみると、「Managerのlistへの書き込み」と「Managerのlistからの読み出し」部分は、共に7秒です。mod==1のndarrayサイズは 24000x8000x3 = 576,000,000 = 576MBです。576/7 = 82MB/sです。1GbpsのLANのMAXスピードに近い。ストリーミングのProxyListとして納得できるスピードです。

さらに今回は、前設定の簡単なPythonのmultiprocessingのManagerを使って、ラズベリーパイ4B (8G, bookworm(x64))で、500MBの巨大ndarrayを共有できたことが、大きな収穫です。

# manager-ndarray---bookworm-x64-500MB.py
import time
from multiprocessing import Manager, Value, Process
import numpy as np
import cv2 as cv


# Raspberry Pi 4B, 8G   under cooling   on Thonny
# ----------------------------------------------------------
# Bookworm(x64) OS:  Linux raspberrypi 6.1.0-rpi7-rpi-v8 #1 SMP PREEMPT Debian 1:6.1.63-1+rpt1 (2023-11-24) aarch64 GNU/Linux
# Python 3.11.2 (main, Mar 13 2023, 12:18:29)
# numpy  1.24.2
# OpenCV  4.6.0 (sudo apt install python3-opencv)
# ----------------------------------------------------------
# https://qiita.com/like_ndarray/items/d39dbde96d5cdc81421d
# ref
# https://qiita.com/like_ndarray/items/b6accb010bde24c477d3


def process1(vcount00, f_plist0, plist0, f_plist1, plist1):

    img00 = cv.imread("d00-4000.png")   #4000w x 3000h (100K)
    img01 = cv.imread("d01-4000.png")
    img02 = cv.imread("d02-4000.png")
    img03 = cv.imread("d03-4000.png")   #warning for bookworm

    try:
        while(True):

            count00_old = vcount00.value
            q,  mod = divmod(count00_old, 2)

            if mod == 0:
                img480 = cv.vconcat([img00, img01, img02, img03, img00, img01, img02, img03])
                img480_4d  = np.reshape(img480, [1, 24000, 4000, 3])
                while True:
                    time.sleep(0.01)
                    if f_plist0.value == 0:
                        f_plist0.value = 9
                    if f_plist0.value == 9:
                        f_plist0.value = 1
                        break

                plist0[:]= []
                for i in img480_4d:
                    plist0.append(i)
                    
                if len(plist0)==0:
                    print("plist0==0")
                f_plist0.value = 4

            elif mod == 1:
                img480v = cv.vconcat([img00, img01, img02, img03, img00, img01, img02, img03])
                img480  = cv.hconcat([img480v, img480v])
                img480_4d  = np.reshape(img480, [1, 24000, 8000, 3])
                while True:
                    time.sleep(0.01)
                    if f_plist1.value == 0:
                        f_plist1.value = 9
                    if f_plist1.value == 9:
                        f_plist1.value = 1
                        break

                plist1[:]= []
                pretime=time.time()
                for i in img480_4d:
                    plist1.append(i)
                print("Writing")
                print(time.time()-pretime) ## 7sec for writing
                
                if len(plist1)==0:
                    print("plist1==0")
                f_plist1.value = 4


            vcount00.value  += 1
            

    except KeyboardInterrupt:
        print('process1  excepted')


def process4(vcount00, f_plist0, plist0, f_plist1, plist1):

    try:
        while(True):
            time.sleep(0.1)
            count0_old = vcount00.value - 1
            q,  mod = divmod(count0_old, 2)

            if f_plist0.value == 4 and mod == 0: # 5sec

                i_retry = 0
                while True:
                    l_ndarray00=[]
                    for i in plist0:
                        l_ndarray00.append(i)

                    i_retry += 1
                    if i_retry > 100:
                        print("i_retry >= 100  error")
                        f_plist0.value = 9
                        break
                    if len(l_ndarray00) == 0:
                        time.sleep(0.0001)
                        continue
                    else:
                        print(i_retry)
                        a_ndarray00 = np.array(l_ndarray00, dtype="uint8")
                        a_ndarray00c = a_ndarray00.copy()
                        f_plist0.value = 9
                        break

                a_ndarray00c333  = np.reshape(a_ndarray00c, [24000, 4000, 3])
                img480 = cv.resize(a_ndarray00c333, (160, 960))
                cv.imshow("Image", img480)
                key = cv.waitKey(1000)
                
                # for check
                a_ndarray00c333aa  = cv.hconcat([a_ndarray00c333, a_ndarray00c333])

            elif f_plist1.value == 4 and mod == 1:  # 10sec
                i_retry = 0
                while True:
                    l_ndarray00=[]
                    pretime=time.time()
                    for i in plist1:
                        l_ndarray00.append(i)
                    print("Reading")
                    print(time.time()-pretime) ## 7sec for reading 

                    i_retry += 1
                    if i_retry > 100:
                        print("i_retry >= 100  error")
                        f_plist1.value = 9
                        break
                    if len(l_ndarray00) == 0:
                        time.sleep(0.0001)
                        continue
                    else:
                        print(i_retry)
                        a_ndarray00 = np.array(l_ndarray00, dtype="uint8")
                        a_ndarray00c = a_ndarray00.copy()
                        f_plist1.value = 9
                        break

                a_ndarray00c333bb  = np.reshape(a_ndarray00c, [24000, 8000, 3])
                img480 = cv.resize(a_ndarray00c333bb, (320, 960))
                cv.imshow("Image", img480)
                key = cv.waitKey(1000)
                
                # for check
                print(np.array_equal(a_ndarray00c333aa, a_ndarray00c333bb))

    except KeyboardInterrupt:
        cv.destroyAllWindows()
        print('process4  excepted')



if __name__ == "__main__":

    with Manager() as manager:

        vcount00 = manager.Value('i', 0)
        vcount00.value  = 0

        f_plist0 = manager.Value('i', 0)
        f_plist0.value  = 9
        plist0  =  manager.list()

        f_plist1 = manager.Value('i', 0)
        f_plist1.value  = 9
        plist1  =  manager.list()

        process1 = Process(target=process1, args=[vcount00, f_plist0, plist0, f_plist1, plist1])
        process4 = Process(target=process4, args=[vcount00, f_plist0, plist0, f_plist1, plist1])

        process1.start()
        process4.start()

        process1.join()
        process4.join()

今回のプログラムより、100GBの巨大なcsvファイルも、あらかじめ500MB毎に分割すれば、ラズベリーパイ4B (8G, bookworm(x64))で、multiprocessingのManagerを使って共有できます。

※ csvファイルは、UTF-8のテキストファイルです。「テキストファイル分割」などで検索すれば、ヒットすると思います。Windowsであれば、「PowerShell」を使い、1行のコマンドで分割してくれます。

※ 仮に、ラズベリーパイ(bookworm(x64))のMainメモリー(8G)が不足した場合は、｢Swapの設定」をすれば、Swapメモリー増設が可能です。｢Swapメモリーの動作確認」もされています。ただしSwapメモリーとして、SSDの追加購入が必要です。▶2024FEB22追記：ラズベリーパイ5では、PCIeのM.2,NVMeのSSDが使えるようです。ただbookworm(x64)のSwapメモリーは100MB毎のシーケンシャルRead･Writeなので、USB型SSDでもSwapメモリーとして充分に実用に耐えると思います。
　
お忙しい中、ご覧いただきありがとうございました。
なお公開プログラムは、コピー・改変も自由です。著作権に関する問題も発生しません。
　
以下はプログラム中で使用した画像です。

d00-4000.png

d01-4000.png

d02-4000.png

d03-4000.png