Skip to content

CuPy:IPC

CuPy 에서 IPC 하는 방법에 대한 내용.

CUDA:IPC 항목을 참조하여 Low-Level API 를 사용해야 한다. (CuPy에서는 High-Level API 지원이 없다)

Low-level CUDA support

CUDA:Memory 관련 IPC Wrapping:

  • cupy.cuda.runtime.ipcGetMemHandle(intptr_t devPtr)
  • cupy.cuda.runtime.ipcOpenMemHandle(bytes handle, unsigned int flags=cudaIpcMemLazyEnablePeerAccess)
  • cupy.cuda.runtime.ipcCloseMemHandle(intptr_t devPtr)

CUDA:Event 관련 IPC Wrapping:

  • cupy.cuda.runtime.ipcGetEventHandle(intptr_t event)
  • cupy.cuda.runtime.ipcOpenEventHandle(bytes handle)
  • 이벤트 에서는 Close API 가 없다.

Memory Handle IPC Example

메모리를 소유하고 있는 프로세스 01: 

import numpy as np
import cupy as cp

def pinned_array(array):
    # first constructing pinned memory
    mem = cp.cuda.alloc_pinned_memory(array.nbytes)
    src = np.frombuffer(mem, array.dtype, array.size).reshape(array.shape)
    src[...] = array
    return src

a_cpu_ori = np.ones((10000, 10000), dtype=np.float32)
a_cpu = pinned_array(a_cpu_ori)

a_stream = cp.cuda.Stream(non_blocking=True)

a_gpu = cp.empty_like(a_cpu)  # numpy.empty_like 와 동일, 단 GPU 를 곁들인
print(a_gpu)  # 모두 0 으로 출력됨
a_gpu.set(a_cpu, stream=a_stream)
a_stream.synchronize()
print(a_gpu)  # 모두 1로 출력됨. (출력시 자동 동기화 되는듯?)

a_gpu *= 2
print(a_gpu)  # 모두 2로 출력됨. (출력시 자동 동기화 되는듯?)

a_gpu_ipc_h = cp.cuda.runtime.ipcGetMemHandle(a_gpu.data.ptr)
assert isinstance(a_gpu_ipc_h, bytes)  # 값이 꽤 긴 bytes 값이다.

## 대략 다음과 같이 출력된다:
## >>> a_gpu.data
##     <MemoryPointer 0x7fb67a000000 device=0 mem=<cupy.cuda.memory.PooledMemory object at 0x7fb698cd3030>>
## >>> a_gpu.device
##     <CUDA Device 0>
## >>> a_gpu.data.ptr
##     140421707595776
## >>> a_gpu.data.mem
##     <cupy.cuda.memory.PooledMemory object at 0x7fb698cd3030>
## >>> a_gpu.data.ptr
##     140421707595776
## >>> a_gpu_ipc_h
##     b'0o\xf1}\x81U\x00\x00O\x12-\x00 .... \x00'
## >>> a_gpu.shape
##     (10000, 10000)
## >>> a_gpu.size
##     100000000

# 이 값을 복사하여 통신을 원하는 프로세스에 전달하자!!
a_send_data = (a_gpu_ipc_h, a_gpu.shape, a_gpu.size)

print(a_send_data)  # 이 값을 복사하여 통신을 원하는 프로세스에 전달하자!!

메모리를 공유 받고자 하는 프로세스 02: 

import numpy as np
import cupy as cp

# 이전 프로세스에서 전달 받은 데이터:
a_gpu_ipc_h, shape, size = a_send_data

mem = cp.cuda.runtime.ipcOpenMemHandle(a_gpu_ipc_h)
assert isinstance(mem, int)
print(mem)  # 140389159796736

um = cp.cuda.UnownedMemory(mem, size, 0)
mp = cp.cuda.MemoryPointer(um, 0)

## >>> um
##     <cupy.cuda.memory.UnownedMemory object at 0x7faf381ce3c0>
## >>> mp
##     <MemoryPointer 0x7faee6000000 device=0 mem=<cupy.cuda.memory.UnownedMemory object at 0x7faf381ce3c0>>

array = cp.ndarray(shape, dtype=cp.float32, memptr=mp)
print(array)  # 모두 2로 출력됨. (출력시 자동 동기화 되는듯?)

cp.cuda.runtime.ipcCloseMemHandle(mem)  # 열었으면 닫자.

Event Handle IPC Example

이벤트를 소유하고 있는 프로세스 01: 

import numpy as np
import cupy as cp

e1 = cp.cuda.Event(block=False, disable_timing=True, interprocess=True)
## `interprocess=True` 일 때 `disable_timing=False` 이면 "ValueError: Timing must be disabled for interprocess events" 에러가 발생된다.
## 즉 `disable_timing=True, interprocess=True` 는 IPC 할 때 고정 값 이다.

e1_h = cp.cuda.runtime.ipcGetEventHandle(e1.ptr)
## 만약 `interprocess=False` 로 설정하면 "cupy_backends.cuda.api.runtime.CUDARuntimeError: cudaErrorInvalidResourceHandle: invalid resource handle" 에러가 발생된다.

assert isinstance(e1_h, bytes)

a_send_data = (e1_h,)
print(a_send_data)  # 이 값을 복사하여 통신을 원하는 프로세스에 전달하자!!

이벤트를 공유 받고자 하는 프로세스 02: 

import numpy as np
import cupy as cp

# 이전 프로세스에서 전달 받은 데이터:
e1_h, = a_send_data

e1 = cp.cuda.runtime.ipcOpenEventHandle(e1_h)
assert isinstance(e1, int)
## cupy.cuda.Event 객체로 Wrapping 하는 방법이 제공되지 않으므로 Low-Level API 를 사용해야 한다.

## -----------------------------------
cp.cuda.runtime.eventRecord(e1, 0)  # 두 번째 인자는 stream ptr 인데, NULL 이면 현재(기본)의 스트림으로 적용된다.
## 즉, 다음과 같다:
s = cp.cuda.get_current_stream()
cp.cuda.runtime.eventRecord(e1, s.ptr)
## -----------------------------------

호환성 확인

리눅스 Only, 64비트, 커널 버전 2.6.18 이상 이여야 한다. 

import platform
import sys

if platform.system() != "Linux":
    print("The application only supported on Linux OS.")
    exit(1)

# https://docs.python.org/3/library/platform.html#platform.architecture
if sys.maxsize <= 2**32:
    print("The application must be built as a 64-bit target.")
    exit(1)

kernel_version_text = platform.release()
print(kernel_version_text)  # e.g. "5.15.0-107-generic"

kernel_version_tuple = tuple(map(lambda x: int(x), kernel_version_text.split('-')[0].split('.')))

# https://github.com/ndd314/cuda_examples/blob/master/0_Simple/simpleIPC/simpleIPC.cu
if kernel_version_tuple < (2, 6, 18):
    print("IPC is only supported with Linux OS kernel version 2.6.18 and higher. Test is being waived.")
    exit(1)

CUDA는 compute capability 2.0 이상이여야 하고,

메모리 복사 Event 동기화를 위한 CUDA Stream 을 사용해야 하는데 Device Overlap 속성이 지원되어야 한다. 

import cupy as cp

num_devices = cp.cuda.runtime.getDeviceCount()
for i in range(num_devices):
    prop = cp.cuda.runtime.getDeviceProperties(i)
    major = prop['major']
    minor = prop['minor']
    computeCapability = major, minor
    if computeCapability < (2, 0):
        print(f"Computing capability must be 2.0 or higher in Device({i})")
        exit(1)

    deviceOverlap = prop['deviceOverlap']
    if not deviceOverlap:
        print(f"Unsupported deviceOverlap in Device({i})")
        exit(1)

Example 

import numpy as np
import cupy as cp

See also

Favorite site