(self, name_prefix, process_dispatch_scheme, use_gpu, engine_type, *args, **kwargs)
| 40 | setattr(self, method_name, func_generator(method_name)) |
| 41 | |
| 42 | def __init__(self, name_prefix, process_dispatch_scheme, use_gpu, engine_type, *args, **kwargs) -> None: |
| 43 | from torch import nn |
| 44 | # print(f"in DataParallelEngineWrapper, name_prefix = {name_prefix}") |
| 45 | if isinstance(process_dispatch_scheme, RayResourcePool): |
| 46 | rpdc = process_dispatch_scheme |
| 47 | else: |
| 48 | rpdc = RayResourcePool(process_on_nodes=process_dispatch_scheme, |
| 49 | use_gpu=use_gpu, |
| 50 | name_prefix=name_prefix, |
| 51 | max_colocate_count=1) |
| 52 | rcia = RayClassWithInitArgs(cls=engine_type, *args, **kwargs) |
| 53 | |
| 54 | self._engine_type = engine_type |
| 55 | |
| 56 | super().__init__(rpdc, rcia) |
| 57 | |
| 58 | nn_module_methods = [ |
| 59 | method_name for method_name in dir(nn.Module) |
| 60 | if callable(getattr(nn.Module, method_name)) and not method_name.startswith("__") |
| 61 | ] |
| 62 | nn_module_methods += ["__call__"] |
| 63 | |
| 64 | def func_generator(method_name): |
| 65 | |
| 66 | def func(*args, **kwargs): |
| 67 | return self.execute_all_async(method_name, *args, **kwargs) |
| 68 | |
| 69 | return func |
| 70 | |
| 71 | print(f"{engine_type} has methods: {dir(engine_type)}") |
| 72 | for method_name in dir(engine_type): |
| 73 | try: |
| 74 | is_callable = callable(getattr(engine_type, method_name)) |
| 75 | except Exception as _: |
| 76 | pass |
| 77 | else: |
| 78 | if is_callable and method_name not in dir(RefBasicRayActor): |
| 79 | print(f"register method: {method_name}") |
| 80 | setattr(self, method_name, func_generator(method_name)) |
| 81 | |
| 82 | self.module = DPEngineRayWorkerGroup.DummyModule(self, nn_module_methods) |
| 83 | |
| 84 | @property |
| 85 | def engine(self): |
nothing calls this directly
no test coverage detected