multiprocessing

experiment.multiprocessing

MultiJobHandler

MultiJobHandler(gpu_ids: Union[int, List[int]], job_per_gpu: int)

여러 실험을 여러 GPU에서 parallel하게 돌려주는 class 입니다.

Example

100가지의 서로 다른 실험을 0, 1 GPU에, GPU당 실험 2개씩 순차적으로 실행

multi_job_handler = MultiJobHandler(gpu_ids=[0, 1], job_per_gpu=2) multi_job_handler.launch(func=some_function, kwargs_list=[ config1, config2, config3, ... , config 100 ])

Note

1. launch 함수를 실행하기 전에 pytorch의 GPU 기능을 사용할 수 없습니다. 1-1. fork 대신 spawn을 사용하면 해결할 수 있으나, spawn을 사용할 경우 code변경 이슈가 생길 수 있음.
2. CUDA_VISIBLE_DEVICES를 사용할 경우 주의가 필요합니다. 현재 구현 방식은 CUDA_VISIBLE_DEVICES=1,3,5,7 이고 gpu_ids=[1, 2]라면 실제 GPU는 3, 5를 사용합니다.

Parameters:

• gpu_ids (Union[int, List[int]]) –

  실험에 사용할 GPU id list.

• job_per_gpu (int) –

  각 GPU device에 최대로 할당할 수 있는 job의 개 수.

Source code in ResearchToolkit/experiment/multiprocessing.py

def __init__(
    self,
    gpu_ids: Union[int, List[int]],
    job_per_gpu: int,
) -> None:
    """
    Args:
        gpu_ids (Union[int, List[int]]): 실험에 사용할 GPU id list.
        job_per_gpu (int): 각 GPU device에 최대로 할당할 수 있는 job의 개 수.
    """
    assert job_per_gpu > 0
    self.job_per_gpu = job_per_gpu
    if isinstance(gpu_ids, int):
        gpu_ids = [gpu_ids]
    self.gpu_ids = list(set(gpu_ids))
    for gpu_id in gpu_ids:
        assert isinstance(gpu_id, int)
    self.proc_list_per_gpu = {gpu_id: [] for gpu_id in self.gpu_ids}
    self.cuda_visible_devices = {gpu_id: str(gpu_id) for gpu_id in self.gpu_ids}
    if os.environ.get("CUDA_VISIBLE_DEVICES") is not None:
        cuda_visible_devices = os.environ["CUDA_VISIBLE_DEVICES"].split(",")
        for gpu_id in self.gpu_ids:
            self.cuda_visible_devices[gpu_id] = cuda_visible_devices[gpu_id]

job_per_gpu instance-attribute

job_per_gpu = job_per_gpu

gpu_ids instance-attribute

gpu_ids = list(set(gpu_ids))

proc_list_per_gpu instance-attribute

proc_list_per_gpu = {gpu_id: [] for gpu_id in (gpu_ids)}

cuda_visible_devices instance-attribute

cuda_visible_devices = {gpu_id: (str(gpu_id)) for gpu_id in (gpu_ids)}

launch

launch(func: Callable, kwargs_list: List) -> None

여러 실험을 여러 GPU에서 parallel하게 실행합니다.

Parameters:

• func (Callable) –

  Function to be executed by the child process.

• kwargs_list (List) –

  List of Keyword arguments to be input to the function.

Source code in ResearchToolkit/experiment/multiprocessing.py

def launch(self, func: Callable, kwargs_list: List) -> None:
    """여러 실험을 여러 GPU에서 parallel하게 실행합니다.

    Args:
        func (Callable): Function to be executed by the child process.
        kwargs_list (List): List of Keyword arguments to be input to the function.
    """
    for kwargs in kwargs_list:
        while self._is_busy():
            self._join()

        cur_gpu_id = self._get_gpu_id()

        context_type = "spawn" if platform.system() == "Windows" else "fork"
        proc = multiprocessing.get_context(context_type).Process(
            target=self.run, kwargs={"gpu_id": cur_gpu_id, "func": func, "kwargs": kwargs}
        )
        proc.start()
        self.proc_list_per_gpu[cur_gpu_id].append(proc)

    self._join(timeout=None)

_is_busy

_is_busy() -> bool

Check devices are busy

Returns:

• bool ( bool ) –

  Returns True if all devices are working else False

Source code in ResearchToolkit/experiment/multiprocessing.py

def _is_busy(self) -> bool:
    """Check devices are busy

    Returns:
        bool: Returns True if all devices are working else False
    """
    is_busy = True
    for gpu_id in self.gpu_ids:
        is_busy = is_busy and (len(self.proc_list_per_gpu[gpu_id]) >= self.job_per_gpu)

    return is_busy

_join

_join(timeout: float = 0.1) -> None

Waits until the child process is finished or as long as the timeout time.

Parameters:

• timeout (float, default: 0.1 ) –

  Waiting time for each child process. Defaults to 0.1.

Source code in ResearchToolkit/experiment/multiprocessing.py

def _join(self, timeout: float = 0.1) -> None:
    """Waits until the child process is finished or as long as the timeout time.

    Args:
        timeout (float, optional): Waiting time for each child process. Defaults to 0.1.
    """
    for gpu_id in self.gpu_ids:
        proc_idx = 0
        while proc_idx < len(self.proc_list_per_gpu[gpu_id]):
            self.proc_list_per_gpu[gpu_id][proc_idx].join(timeout=timeout)
            if self.proc_list_per_gpu[gpu_id][proc_idx].is_alive():
                proc_idx = proc_idx + 1
            else:
                self.proc_list_per_gpu[gpu_id].pop(proc_idx)

_get_gpu_id

_get_gpu_id() -> int

Returns the available gpu id.

Returns:

• int ( int ) –

  gpu id

Source code in ResearchToolkit/experiment/multiprocessing.py

def _get_gpu_id(self) -> int:
    """Returns the available gpu id.

    Returns:
        int: gpu id
    """
    min_job_per_gpu = self.job_per_gpu
    min_gpu_id = -1

    for gpu_id in self.gpu_ids:
        if len(self.proc_list_per_gpu[gpu_id]) < min_job_per_gpu:
            min_job_per_gpu = len(self.proc_list_per_gpu[gpu_id])
            min_gpu_id = gpu_id

    return min_gpu_id

run

run(gpu_id: int, func: Callable, kwargs) -> None

Set the GPU id and execute the function.

Parameters:

• gpu_id (int) –

  GPU id

• func (Callable) –

  Function to be executed by the child process.

• kwargs (_type_) –

  Keyword arguments to be input to the function.

Source code in ResearchToolkit/experiment/multiprocessing.py

def run(self, gpu_id: int, func: Callable, kwargs) -> None:
    """Set the GPU id and execute the function.

    Args:
        gpu_id (int): GPU id
        func (Callable): Function to be executed by the child process.
        kwargs (_type_): Keyword arguments to be input to the function.
    """
    os.environ["CUDA_VISIBLE_DEVICES"] = self.cuda_visible_devices[gpu_id]
    func(**kwargs)