Python3.8並列処理 データ受け渡し方法によるデータ送受信速度の計測

bench.py

import ctypes
import queue
import threading
import multiprocessing
from multiprocessing.managers import SharedMemoryManager
from time import time


def main():
    test_configs = [
        # test name, test func, args
        ('Thread + Queue', test_queue, (threading.Thread, queue.Queue)),
        ('Thread + mp.Queue', test_queue, (threading.Thread, multiprocessing.Queue)),
        ('Process + mp.Queue', test_queue, (multiprocessing.Process, multiprocessing.Queue)),
        ('Process + mp.Value', test_mp_value, None),
        ('Process + SharedMemory', test_shared_mem, None),
    ]

    header_row = ['data size'] + [c[0] for c in test_configs]
    print(','.join(header_row))

    for data_name, data in generate_data(64 * 1024**2, 100, 1024**3):
        results = [data_name]
        for test_name, test_func, args in test_configs:
            if args:
                results.append(str(test_func(data, *args)))
            else:
                results.append(str(test_func(data)))
        print(','.join(results))
        del(data)
        del(data_name)


def generate_data(max_size=1024**3, n_data=1000, max_total_bytes=1024**3):
    data_bytes = 1
    while True:
        if data_bytes < 1024:
            data_name = '{}B'.format(data_bytes)
        elif data_bytes < 1024**2:
            data_name = '{}KiB'.format(int(data_bytes/1024))
        elif data_bytes < 1024**3:
            data_name = '{}MiB'.format(int(data_bytes/1024**2))
        num_data = min(int(max_total_bytes / data_bytes), n_data)
        yield data_name, [b'1' * data_bytes for i in range(num_data)]
        if num_data < 20 or data_bytes > max_size:
            break
        data_bytes *= 2


def test_queue(data, worker_class, queue_class):
    """ queueを使ってworkerにデータを渡す速度を計測する """
    q = queue_class()
    worker = worker_class(target=read_queue, args=(q, ))
    worker.start()

    s = time()
    for d in data:
        q.put(d)
    q.put(None)
    worker.join()
    return (time() - s) / len(data) * 1000


def read_queue(q):
    while True:
        if (d := q.get()) is None:
            break
        else:
            v = len(d)


def test_multi_queue(data, worker_class, queue_class):
    """ queueを使ってworkerにデータを渡す速度を計測する """
    qs = [queue_class(), queue_class()]
    worker = worker_class(target=read_multi_queue, args=(qs,))
    worker.start()

    s = time()
    for i, d in enumerate(data):
        qs[i % 2].put(d)
    for q in qs:
        q.put(None)
    worker.join()
    return time() - s


def read_multi_queue(qs):
    valid_q_idx = list(range(len(qs)))
    while valid_q_idx:
        for i, q_idx in enumerate(valid_q_idx):
            if (d := qs[q_idx].get()) is None:
                del(valid_q_idx[i])
                break
            else:
                v = len(d)


def test_shared_mem(data):
    sm = multiprocessing.shared_memory.SharedMemory(create=True, size=len(data[0]))
    input_q = multiprocessing.Queue()
    output_q = multiprocessing.Queue()
    worker = multiprocessing.Process(target=read_shared_mem, args=(sm, input_q, output_q))
    worker.start()

    s = time()
    for d in data:
        sm.buf[:] = d
        input_q.put(1)
        output_q.get()
    input_q.put(None)
    worker.join()
    e = time()
    sm.close()
    sm.unlink()
    return (e - s) / len(data) * 1000


def read_shared_mem(sm, input_q, output_q):
    while input_q.get() is not None:
        v = len(sm.buf)
        output_q.put(1)


def test_mp_value(data):
    shared_val = multiprocessing.Value(ctypes.c_char_p, data[0])
    input_q = multiprocessing.Queue()
    output_q = multiprocessing.Queue()
    worker = multiprocessing.Process(target=read_shared_val, args=(shared_val, input_q, output_q))
    worker.start()

    s = time()
    for d in data:
        shared_val.value = d
        input_q.put(1)
        output_q.get()
    input_q.put(None)
    worker.join()
    return (time() - s) / len(data) * 1000


def read_shared_val(shared_val, input_q, output_q):
    while input_q.get() is not None:
        v = len(shared_val.value)
        output_q.put(1)


if __name__ == '__main__':
    main()

result_laptop_core_i5_7y54.csv

result_raspberry_pi_3b.csv

result_server_ryzen7_1700.csv