Skip to content

experiment

Module diagram

classDiagram
  class experiment {
  }
  class importlib {
  }
  class multiprocessing {
  }
  class sweep {
  }

experiment

importlib

get_module_from_path 

get_module_from_path(file_path: str) -> ModuleType

입력으로 받은 파이썬 파일을 모듈화하여 반환합니다.

Parameters:

  • file_path (str) –

    path to python file

Returns:

  • ModuleType ( ModuleType ) –

    loaded module

Source code in ResearchToolkit/experiment/importlib.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
def get_module_from_path(file_path: str) -> ModuleType:
    """입력으로 받은 파이썬 파일을 모듈화하여 반환합니다.

    Args:
        file_path (str): path to python file

    Returns:
        ModuleType: loaded module
    """
    module_name = os.path.basename(file_path).rstrip(".py")
    spec = importlib.util.spec_from_file_location(module_name, file_path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    return module

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 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
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 
71
72
73
74
75
76
77
78
79
80
81
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 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
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 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
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 
114
115
116
117
118
119
120
121
122
123
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)

sweep

SweepList 

SweepList(sweep_list: List[Any], tag: Optional[str] = None, descriptions: Optional[List[Any]] = None, rank: Union[int, float] = float('inf'))

sweep 하고 싶은 config를 나타낼 때 사용하는 class 입니다.

Parameters:

  • sweep_list (List) –

    sweep할 element들이 들어있는 List입니다.

  • tag (str, default: None ) –

    SweepList를 group으로 묶고 싶을 때 사용하는 tag 입니다. tag가 같은 SweepList는 Group으로 묶여서 동시에 sweep 됩니다. Defaults to None.

  • descriptions (List[Any], default: None ) –

    sweep할 element들을 설명해주는 description이 들어있는 List입니다. sweep_list와 길이가 일치해야 합니다. Defaults to None.

  • rank (Union[int, float], default: float('inf') ) –

    생성된 config의 반복 되는 순서를 정해주는 값입니다. rank가 낮은 SweepList가 먼저 반복되어 생성됩니다. Defaults to float("inf").

Source code in ResearchToolkit/experiment/sweep.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(
    self,
    sweep_list: List[Any],
    tag: Optional[str] = None,
    descriptions: Optional[List[Any]] = None,
    rank: Union[int, float] = float("inf"),
) -> None:
    self.sweep_list = sweep_list
    self.tag = tag
    self.descriptions = descriptions
    self.rank = rank
    if self.descriptions is None:
        self.descriptions = [_NoDescription() for _ in range(len(self.sweep_list))]
    assert len(self.sweep_list) == len(
        self.descriptions
    ), "sweep_list and descriptions must be the same length."
sweep_list instance-attribute 
sweep_list = sweep_list
tag instance-attribute 
tag = tag
descriptions instance-attribute 
descriptions = descriptions
rank instance-attribute 
rank = rank

_NoDescription

Description이 없는지를 나타내기 위한 class

_SweepList 

_SweepList(sweeplist: SweepList, key_list: List)

Bases: SweepList

sweep 함수의 쉬운 구현을 위해 SweepList에 약간의 구현이 추가된 class

Source code in ResearchToolkit/experiment/sweep.py 
50
51
52
53
54
55
56
57
def __init__(self, sweeplist: SweepList, key_list: List) -> None:
    super().__init__(
        sweep_list=sweeplist.sweep_list,
        tag=sweeplist.tag,
        descriptions=sweeplist.descriptions,
        rank=sweeplist.rank,
    )
    self.key_list = key_list
key_list instance-attribute 
key_list = key_list
__iter__ 
__iter__() -> Iterator
Source code in ResearchToolkit/experiment/sweep.py 
59
60
61
62
63
64
65
66
67
def __iter__(self) -> Iterator:
    return [
        {
            "config_value": config_value,
            "description": description,
            "key_list": self.key_list,
        }
        for config_value, description in zip(self.sweep_list, self.descriptions)
    ].__iter__()
__len__ 
__len__() -> int
Source code in ResearchToolkit/experiment/sweep.py 
69
70
def __len__(self) -> int:
    return len(self.sweep_list)

_SweepGroup 

_SweepGroup(sweeplist_list: Optional[List[_SweepList]] = None)

tag가 같은 _SweepList를 Group으로 묶어서 관리하는 class

Source code in ResearchToolkit/experiment/sweep.py 
76
77
78
79
80
def __init__(self, sweeplist_list: Optional[List[_SweepList]] = None) -> None:
    if sweeplist_list is None:
        self.sweep_group = []
    else:
        self.sweep_group = sweeplist_list
sweep_group instance-attribute 
sweep_group = []
rank property 
rank: int
append 
append(sweeplist: _SweepList) -> None
Source code in ResearchToolkit/experiment/sweep.py 
82
83
def append(self, sweeplist: _SweepList) -> None:
    self.sweep_group.append(sweeplist)
__iter__ 
__iter__() -> zip
Source code in ResearchToolkit/experiment/sweep.py 
85
86
87
88
89
90
def __iter__(self) -> zip:
    for _sweeplist in self.sweep_group:
        assert len(_sweeplist) == len(
            self.sweep_group[0]
        ), "All SweepLists with the same tag must have the same length."
    return zip(*self.sweep_group)

_SweepListManager 

_SweepListManager(sweeplist_list: List[_SweepList])

_SweepList를 관리하고 데카르트 곱(cartesian product)을 iterator로 만들어주는 class

Source code in ResearchToolkit/experiment/sweep.py 
100
101
102
103
104
105
106
107
108
109
110
111
def __init__(self, sweeplist_list: List[_SweepList]) -> None:
    self.tagged_sweepgroup_dict = {}
    self.non_tagged_sweepgroup_list = []

    for sweeplist in sweeplist_list:
        tag = sweeplist.tag
        if tag is None:
            self.non_tagged_sweepgroup_list.append(_SweepGroup([sweeplist]))
        else:
            if tag not in self.tagged_sweepgroup_dict.keys():
                self.tagged_sweepgroup_dict[tag] = _SweepGroup()
            self.tagged_sweepgroup_dict[tag].append(sweeplist)
tagged_sweepgroup_dict instance-attribute 
tagged_sweepgroup_dict = {}
non_tagged_sweepgroup_list instance-attribute 
non_tagged_sweepgroup_list = []
__iter__ 
__iter__() -> product
Source code in ResearchToolkit/experiment/sweep.py 
113
114
115
116
117
118
119
def __iter__(self) -> itertools.product:
    iter_list = [(sweepgroup.rank, sweepgroup) for sweepgroup in self.non_tagged_sweepgroup_list] + [
        (sweepgroup.rank, sweepgroup) for _, sweepgroup in self.tagged_sweepgroup_dict.items()
    ]
    iter_list.sort(key=lambda sweepgroup: sweepgroup[0], reverse=True)
    iter_list = [sweepgroup for rank, sweepgroup in iter_list]
    return itertools.product(*iter_list)

_find_sweeplist 

_find_sweeplist(config: Dict, key_list: List) -> List[_SweepList]

Find the SweepList by recursively traversing the config dictionary. For subsequent operations, SweepList is changed to _SweepList.

Parameters:

  • config (Dict) –

    The config dictionary in which to find the SweepList.

  • key_list (List) –

    List containing the keys of the dictionary sequentially.

Returns:

  • List[_SweepList] –

    List[_SweepList]: List of _SweepList

Source code in ResearchToolkit/experiment/sweep.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def _find_sweeplist(config: Dict, key_list: List) -> List[_SweepList]:
    """
    Find the SweepList by recursively traversing the config dictionary.
    For subsequent operations, SweepList is changed to _SweepList.

    Args:
        config (Dict): The config dictionary in which to find the SweepList.
        key_list (List): List containing the keys of the dictionary sequentially.

    Returns:
        List[_SweepList]: List of _SweepList
    """
    sweeplist_list = []
    for name, values in config.items():
        if isinstance(values, SweepList):
            cur_key = key_list + [name]
            sweeplist_list.append(_SweepList(values, cur_key))
        elif isinstance(values, Dict):
            sweeplist_list = sweeplist_list + _find_sweeplist(values, key_list + [name])

    return sweeplist_list

_set_value 

_set_value(config: Dict, key_list: List, value: Any, make_sub_dict: bool = False) -> None

Set the value in the config dict using the keys in key_list.

Parameters:

  • config (Dict) –

    The dictionary in which you want to set the value.

  • key_list (List) –

    List containing the keys of the dictionary sequentially.

  • value (Any) –

    The value you want to set.

Example

config = { "A": { "B": { "C" = 3 } } }

_set_value(config=config, key_list=["A", "B", "C"], value=5)

results: { "A": { "B": { "C" = 5 } } }

Source code in ResearchToolkit/experiment/sweep.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def _set_value(config: Dict, key_list: List, value: Any, make_sub_dict: bool = False) -> None:
    """
    Set the value in the config dict using the keys in key_list.

    Args:
        config (Dict): The dictionary in which you want to set the value.
        key_list (List): List containing the keys of the dictionary sequentially.
        value (Any): The value you want to set.

    Example:
        config = {
            "A": {
                "B": {
                    "C" = 3
                }
            }
        }

        _set_value(config=config, key_list=["A", "B", "C"], value=5)

        results:
        {
            "A": {
                "B": {
                    "C" = 5
                }
            }
        }
    """
    for idx, key in enumerate(key_list):
        if idx == len(key_list) - 1:
            config[key] = value
        else:
            if make_sub_dict and key not in config:
                config[key] = {}
            config = config[key]

sweep 

sweep(config: Dict) -> List[Dict]

config 내부에 있는 SweepList를 찾아 데카르트 곱(cartesian product)을 만들어줍니다.

Parameters:

  • config (Dict) –

    sweep을 진행하고자 하는 config dictionary.

Returns:

  • List[Dict] –

    List[Dict]: 만들어진 (description, config)들이 들어있는 List.

  • Dict ( List[Dict] ) –

    { "description": Dict, 해당 config에 대해 설정한 description. config와 같은 구조를 가집니다. "config": Dict, 만들어진 config

  • List[Dict] –

    }

Example

config = { "A": 3, "B": SweepList([1, 2], description=["1", "2"]), "C": SweepList([3, 4]), }

configs = sweep(config)

results: configs => [ { "description": {"B": "1",}, "config": {"A": 3, "B": 1, "C": 3}, }, { "description": {"B": "1"}, "config": {"A": 3, "B": 1, "C": 4}, } { "description": {"B": "2"}, "config": {"A": 3, "B": 2, "C": 3}, } { "description": {"B": "2"}, "config": {"A": 3, "B": 2, "C": 4}, } ]

Source code in ResearchToolkit/experiment/sweep.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def sweep(config: Dict) -> List[Dict]:
    """
    config 내부에 있는 SweepList를 찾아 데카르트 곱(cartesian product)을 만들어줍니다.

    Args:
        config (Dict): sweep을 진행하고자 하는 config dictionary.

    Returns:
        List[Dict]: 만들어진 (description, config)들이 들어있는 List.
        Dict: {
            "description": Dict, 해당 config에 대해 설정한 description.
                                 config와 같은 구조를 가집니다.
            "config": Dict, 만들어진 config
        }

    Example:
        config = {
            "A": 3,
            "B": SweepList([1, 2], description=["1", "2"]),
            "C": SweepList([3, 4]),
        }

        configs = sweep(config)

        results:
        configs => [
            {
                "description": {"B": "1",},
                "config": {"A": 3, "B": 1, "C": 3},
            },
            {
                "description": {"B": "1"},
                "config": {"A": 3, "B": 1, "C": 4},
            }
            {
                "description": {"B": "2"},
                "config": {"A": 3, "B": 2, "C": 3},
            }
            {
                "description": {"B": "2"},
                "config": {"A": 3, "B": 2, "C": 4},
            }
        ]
    """
    config = copy.deepcopy(config)

    sweeplist_list = _find_sweeplist(config, [])
    sweeplist_manager = _SweepListManager(sweeplist_list)

    config_list = []
    for sweepgroup_list in sweeplist_manager:
        config_description_dict = {}
        for sweepgroup in sweepgroup_list:
            for config_info in sweepgroup:
                config_value = config_info["config_value"]
                description = config_info["description"]
                key_list = config_info["key_list"]

                _set_value(config, key_list, config_value)
                if not isinstance(description, _NoDescription):
                    _set_value(config_description_dict, key_list, description, make_sub_dict=True)

        config_list.append({"description": config_description_dict, "config": copy.deepcopy(config)})

    return config_list

sweep_native 

sweep_native(options: Dict[str, List[Any]]) -> List[Dict]

SweepList 없이 native list 를 이용해 sweep을 수행합니다.

Example

config = { "A": [3], "B": [1, 2], "C": [3, 4], }

configs = sweep_native(config)

results: configs => [ {"A": 3, "B": 1, "C": 3}, {"A": 3, "B": 1, "C": 4}, {"A": 3, "B": 2, "C": 3}, {"A": 3, "B": 2, "C": 4}, ]

Source code in ResearchToolkit/experiment/sweep.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
def sweep_native(options: Dict[str, List[Any]]) -> List[Dict]:
    """SweepList 없이 native list 를 이용해 sweep을 수행합니다.

    Example:
        config = {
            "A": [3],
            "B": [1, 2],
            "C": [3, 4],
        }

        configs = sweep_native(config)

        results:
        configs => [
            {"A": 3, "B": 1, "C": 3},
            {"A": 3, "B": 1, "C": 4},
            {"A": 3, "B": 2, "C": 3},
            {"A": 3, "B": 2, "C": 4},
        ]
    """
    option_tuples = [[(name, value) for value in values] for name, values in options.items()]
    return [dict(option) for option in itertools.product(*option_tuples)]