Blog

Python multiprocessing.shared_memory로 프로세스 간 메모리 공유하기

numpy 배열을 multiprocessing.Queue/Pipe로 주고받으면 매번 pickle 직렬화·역직렬화가 일어나 배열이 클수록 병목이 된다. Python 3.8부터 표준 라이브러리에 들어온 multiprocessing.shared_memory는 pickle 없이 OS의 진짜 공유 메모리(POSIX /dev/shm)를 이름으로 attach해 여러 프로세스가 같은 버퍼를 직접 읽고 쓰게 한다. 이 글에서는 SharedMemory 기본 사용법, numpy zero-copy 공유, ShareableList, resource tracker 주의사항을 실제 실행 결과로 정리한다.

핵심 개념

SharedMemory(create=True, size=N)로 블록을 만들거나 name만으로 attach한다. buf(memoryview)에 쓴 값은 같은 이름으로 attach한 다른 프로세스에서 즉시 보인다 — 값을 주고받는 게 아니라 같은 물리 메모리를 보는 것이다.

특징
multiprocessing.Value/Arrayctypes 기반 단일 타입, numpy 연결하려면 변환 필요
shared_memoryraw 바이트 버퍼 — numpy ndarray(shape, dtype, buffer=...)에 그대로 사용
ShareableListint/float/bool/str/bytes/None 혼합 가능, 길이 고정

블록을 쓰지 않을 때 각 프로세스는 close()를 호출하고, 커널에서 실제로 삭제하는 unlink()는 전체 중 정확히 한 곳에서 한 번만 호출한다.

기본 블록 생성과 attach

from multiprocessing import shared_memory

shm_a = shared_memory.SharedMemory(create=True, size=10)
print("생성된 블록 이름:", shm_a.name)
shm_a.buf[:4] = bytearray([22, 33, 44, 55])

# 다른 프로세스라고 가정, 이름만으로 attach
shm_b = shared_memory.SharedMemory(name=shm_a.name)
print("attach한 프로세스가 본 값:", bytes(shm_b.buf[:4]))

shm_b.buf[0] = 99
print("shm_a에서 본 값(shm_b가 수정):", shm_a.buf[0])

shm_b.close()
shm_a.close()
shm_a.unlink()
$ python3 basic_shm.py
생성된 블록 이름: psm_aea0d3eb
attach한 프로세스가 본 값: b'\x16!,7'
shm_a에서 본 값(shm_b가 수정): 99

numpy 배열을 shared memory 위에 얹기

공유 메모리를 배열 nbytes만큼 만들고 같은 shape/dtype의 ndarray를 얹는다. 자식은 배열 전체가 아니라 블록 이름과 shape/dtype만 받는다.

import numpy as np
from multiprocessing import Process, shared_memory


def worker(shm_name, shape, dtype_name):
    existing_shm = shared_memory.SharedMemory(name=shm_name)
    arr = np.ndarray(shape, dtype=np.dtype(dtype_name), buffer=existing_shm.buf)
    arr *= 2  # 복사본이 아니라 공유 버퍼 위에서 바로 연산
    existing_shm.close()


if __name__ == "__main__":
    data = np.arange(10, dtype=np.float64)
    shm = shared_memory.SharedMemory(create=True, size=data.nbytes)
    shared_arr = np.ndarray(data.shape, dtype=data.dtype, buffer=shm.buf)
    shared_arr[:] = data[:]

    print("worker 실행 전:", shared_arr)
    p = Process(target=worker, args=(shm.name, data.shape, data.dtype.name))
    p.start()
    p.join()
    print("worker 실행 후:", shared_arr)

    shm.close()
    shm.unlink()
$ python3 numpy_shm.py
worker 실행 전: [0. 1. 2. 3. 4. 5. 6. 7. 8. 9.]
worker 실행 후: [ 0.  2.  4.  6.  8. 10. 12. 14. 16. 18.]

자식이 부모에게 통신으로 결과를 돌려준 게 아니라 arr *= 2가 공유 버퍼를 직접 고친 것이다. 배열이 클수록 이 차이가 성능 이득으로 이어진다.

pickle 대비 얼마나 빠른가

float64 2천만 개(약 160MB) 배열 합계를 자식 프로세스에서 구해 돌려받는 작업을 Queue와 shared_memory로 각각 구현해 비교.

import time
import numpy as np
from multiprocessing import Process, Queue

SIZE = 20_000_000  # float64 8바이트 * 2천만개 = 약 160MB


def worker(q_in, q_out):
    arr = q_in.get()          # pickle 역직렬화
    total = float(arr.sum())
    q_out.put(total)


if __name__ == "__main__":
    data = np.arange(SIZE, dtype=np.float64)
    q_in, q_out = Queue(), Queue()
    p = Process(target=worker, args=(q_in, q_out))
    p.start()

    start = time.perf_counter()
    q_in.put(data)             # pickle 직렬화 + 파이프 전송
    result = q_out.get()
    elapsed = time.perf_counter() - start

    p.join()
    print(f"Queue(pickle) 방식: {elapsed:.3f}초, 합계={result}")
import time
import numpy as np
from multiprocessing import Process, Queue, shared_memory

SIZE = 20_000_000  # float64 8바이트 * 2천만개 = 약 160MB


def worker(shm_name, shape, dtype_name, q_out):
    shm = shared_memory.SharedMemory(name=shm_name)
    arr = np.ndarray(shape, dtype=np.dtype(dtype_name), buffer=shm.buf)
    total = float(arr.sum())
    shm.close()
    q_out.put(total)


if __name__ == "__main__":
    data = np.arange(SIZE, dtype=np.float64)
    shm = shared_memory.SharedMemory(create=True, size=data.nbytes)
    shared_arr = np.ndarray(data.shape, dtype=data.dtype, buffer=shm.buf)
    shared_arr[:] = data[:]

    q_out = Queue()
    p = Process(target=worker, args=(shm.name, data.shape, data.dtype.name, q_out))

    start = time.perf_counter()
    p.start()
    result = q_out.get()
    elapsed = time.perf_counter() - start

    p.join()
    shm.close()
    shm.unlink()
    print(f"shared_memory 방식: {elapsed:.3f}초, 합계={result}")
방식소요 시간합계
Queue(pickle)0.564초199999990000000.0
shared_memory0.017초199999990000000.0

같은 환경 기준 30배 이상 차이. Queue는 배열 전체를 직렬화하는 반면 shared_memory는 블록 이름과 shape/dtype 메타데이터만 주고받고 데이터는 아예 복사되지 않는다.

ShareableList

from multiprocessing import shared_memory

sl = shared_memory.ShareableList(["cpu", 4, 3.5, True, None])
print("이름:", sl.shm.name)
print("초기값:", list(sl))

sl2 = shared_memory.ShareableList(name=sl.shm.name)
sl2[1] = 8
print("sl2에서 수정 후 sl에서 본 값:", list(sl))
print("index(8):", sl.index(8))

sl2.shm.close()
sl.shm.close()
sl.shm.unlink()
$ python3 shareablelist_demo.py
이름: psm_d2167fe7
초기값: ['cpu', 4, 3.5, True, None]
sl2에서 수정 후 sl에서 본 값: ['cpu', 8, 3.5, True, None]
index(8): 1

ShareableList도 내부는 SharedMemory 하나라 sl.shm.close()/unlink()로 정리한다. 문자열/바이트 끝의 널 바이트(\x00)는 조회 시 잘려나가는 알려진 동작이다.

주의사항

  • close()는 attach한 프로세스 각자가, unlink()는 만든 쪽에서 딱 한 번만. unlink() 후 접근하면 POSIX에서는 메모리 오류 위험 (Windows는 unlink() 자체가 no-op, 모든 핸들이 닫히면 자동 정리).
  • POSIX에서는 기본 track=True로 resource tracker가 관리한다. 정리 누락 시 UserWarning: resource_tracker: There appear to be 1 leaked shared_memory objects to clean up at shutdown 경고 후 강제 정리된다 — 나타나면 close()/unlink() 호출을 점검할 것. try/finally나 SharedMemoryManager로 보장하는 게 안전하다.
  • subprocess로 띄운 별개 프로세스끼리 공유하면 서로 다른 resource tracker를 가져, 만든 쪽이 먼저 끝나면 아직 쓰고 있는 블록을 “leaked”로 오인해 지울 수 있다. Python 3.13+는 track=False로 자동 추적을 끄고 직접 정리할 수 있다.
  • size는 OS 페이지 크기 단위로 정렬돼 요청보다 크게 잡힐 수 있다 — numpy로 얹을 땐 항상 원본 shape/dtype 기준으로 ndarray를 만들 것.

마무리

대용량 numpy 배열을 여러 프로세스가 다뤄야 할 때 shared_memory는 pickle 비용을 걷어내는 선택지다. 데이터가 클수록 이득이 커지지만 자원 정리 책임이 개발자에게 넘어오므로, close()/unlink() 규칙을 지키고 SharedMemoryManager나 try/finally로 정리 누락을 방지할 것.

참고

Tags: , , , , ,

답글 남기기