import logging
import os
from yr import log
from yr.config_manager import ConfigManager
data_system_import = True
_import_error = None
try:
import torch
import torch_npu
from datasystem import DsTensorClient
from torch.npu import current_device
except ImportError as e:
data_system_import = False
_import_error = e
_global_tensor_client = None
_logger = logging.getLogger(__name__)
def _inspect_data_system_address():
address = os.getenv("YR_DS_ADDRESS")
if address and len(address) != 0:
return address
address = ConfigManager().ds_address
if len(address) != 0:
return address
raise RuntimeError("cannot inspect data system address")
def get_tensor_client():
if not data_system_import:
raise RuntimeError(f"import err: {_import_error}")
global _global_tensor_client
if _global_tensor_client:
return _global_tensor_client
ds_worker_addr = _inspect_data_system_address()
_logger.info(
f"start get ds tensor client, ds address is {ds_worker_addr}, current device id is {current_device()}")
if ds_worker_addr.find(":") == -1:
raise ValueError(
"Invalid address of data system worker: {}, expect 'ip:port'".format(ds_worker_addr))
host, port = ds_worker_addr.split(":")
port = int(port)
_global_tensor_client = DsTensorClient(host, port, current_device())
_global_tensor_client.init()
return _global_tensor_client