import sys
from array import array
import numpy as np
from timeit import timeit
from multiprocessing import shared_memory
def console(*msg):
print(*msg)
sys.stdout.flush()
def test(size):
data = [1 for i in range(size)]
dataA = array('I', [1 for i in range(size)])
byteSize = 4 * size
shmA1 = shared_memory.SharedMemory(create=True, name=f"dataA{size}", size=byteSize)
bufA1 = shmA1.buf
bufA1[0:byteSize] = array("I", data).tobytes()
dataA1 = bufA1.cast("I")
shmA2 = shared_memory.SharedMemory(name=f"dataA{size}")
bufA2 = shmA2.buf
dataA2 = bufA2.cast('I')
dataN = np.array([1 for i in range(size)])
byteSize = dataN.nbytes
shmN1 = shared_memory.SharedMemory(create=True, name=f"dataN{size}", size=byteSize)
bufN1 = shmN1.buf
dataN1 = np.ndarray(dataN.shape, dtype=dataN.dtype, buffer=bufN1)
dataN1[:] = dataN[:]
shmN2 = shared_memory.SharedMemory(name=f"dataN{size}")
bufN2 = shmN2.buf
dataN2 = np.ndarray(dataN.shape, dtype=dataN.dtype, buffer=bufN2)
half = size // 2
dataA1[half] = 1000
dataN1[half] = 1000
console(f"{dataA1[half]} == {dataA2[half]} ? {dataA1[half] == dataA2[half]}")
console(f"{dataN1[half]} == {dataN2[half]} ? {dataN1[half] == dataN2[half]}")
console(f'''list : {timeit("max(data)", globals=locals(), number=1):>.3e}''')
console(f'''array : {timeit("max(dataA)", globals=locals(), number=1):>.3e}''')
console(f'''array1 : {timeit("max(dataA1)", globals=locals(), number=1):>.3e}''')
console(f'''array2 : {timeit("max(dataA2)", globals=locals(), number=1):>.3e}''')
console(f'''numpy1 : {timeit("max(dataN1)", globals=locals(), number=1):>.3e}''')
console(f'''numpy2 : {timeit("max(dataN2)", globals=locals(), number=1):>.3e}''')
dataA1.release()
dataA2.release()
bufA1.release()
shmA1.close()
shmA2.close()
shmA1.unlink()
bufN1.release()
bufN2.release()
shmN1.close()
shmN2.close()
shmN1.unlink()
test(1000000)
1000 == 1000 ? True 1000 == 1000 ? True list : 1.505e-02 array : 1.789e-02 array1 : 2.230e-02 array2 : 2.371e-02 numpy1 : 1.044e-01 numpy2 : 1.078e-01