公司网站建设设计公司哪家好,每天看七个广告赚40元的app,注册网站法律风险,安徽海外网络推广pytorch单元测试 一.公共模块[common.py]二.普通算子测试[test_clone.py]三.集合通信测试[test_ccl.py]四.测试命令五.测试报告 希望测试pytorch各种算子、block、网络等在不同硬件平台,不同软件版本下的计算误差、耗时、内存占用等指标.
本文基于torch.testing._internal
一… pytorch单元测试 一.公共模块[common.py]二.普通算子测试[test_clone.py]三.集合通信测试[test_ccl.py]四.测试命令五.测试报告 希望测试pytorch各种算子、block、网络等在不同硬件平台,不同软件版本下的计算误差、耗时、内存占用等指标.
本文基于torch.testing._internal
一.公共模块[common.py]
import torch
from torch import nn
import math
import torch.nn.functional as F
import time
import os
import socket
import sys
from datetime import datetime
import numpy as np
import collections
import math
import json
import copy
import traceback
import subprocess
import unittest
import torch
import inspect
from torch.testing._internal.common_utils import TestCase, run_tests,parametrize,instantiate_parametrized_tests
from torch.testing._internal.common_distributed import MultiProcessTestCase
import torch.distributed as distos.environ[MASTER_ADDR] localhost
os.environ[MASTER_PORT] 29500
os.environ[RANDOM_SEED] 0 devicecpu
device_typecpu
device_namecputry:if torch.cuda.is_available(): device_nametorch.cuda.get_device_name().replace( ,)devicecuda:0device_typecudaccl_backendnccl
except:passhost_namesocket.gethostname()
sdk_versionos.getenv(SDK_VERSION,) #从环境变量中获取sdk版本号
metric_data_rootos.getenv(TORCH_UT_METRICS_DATA,./ut_data) #日志存放的目录
device_counttorch.cuda.device_count()if not os.path.exists(metric_data_root):os.makedirs(metric_data_root)def device_warmup(device):设备warmup,确保设备已经正常工作,排除设备初始化的耗时left torch.rand([128,512], dtype torch.float16).to(device)right torch.rand([512,128], dtype torch.float16).to(device)outtorch.matmul(left,right)torch.cuda.synchronize()torch.manual_seed(1)
np.random.seed(1)def loop_decorator(loops,rank0):循环装饰器用于统计函数的执行时间内存占用等def decorator(func):def wrapper(*args,**kwargs):latency[]memory_allocated_t0torch.cuda.memory_allocated(rank)for _ in range(loops):input_copy[x.clone() for x in args]beg datetime.now().timestamp() * 1e6pred func(*input_copy)gtkwargs[golden]torch.cuda.synchronize()enddatetime.now().timestamp() * 1e6mse torch.mean(torch.pow(pred.cpu().float()- gt.cpu().float(), 2)).item()latency.append(end-beg)memory_allocated_t1torch.cuda.memory_allocated(rank)avg_latencynp.mean(latency[len(latency)//2:]).round(3)first_latencylatency[0]return { first_latency:first_latency,avg_latency:avg_latency,memory_allocated:memory_allocated_t1-memory_allocated_t0,mse:mse}return wrapperreturn decoratorclass TorchUtMetrics:用于统计测试结果比较之前的最小值def __init__(self,ut_name,thresold0.2,rank0):self.ut_namef{ut_name}_{rank}self.thresoldthresoldself.rankrankself.data{ut_name:self.ut_name,metrics:[]}self.metrics_pathos.path.join(metric_data_root,f{self.ut_name}_{self.rank}.jon)try:with open(self.metrics_path,r) as f:self.datajson.loads(f.read())except:passdef __enter__(self):self.beg datetime.now().timestamp() * 1e6return selfdef __exit__(self, exc_type, exc_val, exc_tb): self.report()self.save_data()def save_data(self):with open(self.metrics_path,w) as f:f.write(json.dumps(self.data,indent4))def set_metrics(self,metrics):self.enddatetime.now().timestamp() * 1e6itemcollections.OrderedDict()item[time]datetime.now().strftime(%Y-%m-%d %H:%M:%S.%f)item[sdk_version]sdk_versionitem[device_name]device_nameitem[host_name]host_nameitem[metrics]metricsitem[metrics][e2e_time]self.end-self.begself.cur_itemitemself.data[metrics].append(self.cur_item)def get_metric_names(self):return self.data[metrics][0][metrics].keys()def get_min_metric(self,metric_name,devicenameNone):min_value0min_value_index-1for idx,item in enumerate(self.data[metrics]):if devicename and (devicename!item[device_name]): continue valfloat(item[metrics][metric_name])if min_value_index-1 or valmin_value:min_valuevalmin_value_indexidxreturn min_value,min_value_indexdef get_metric_info(self,index):metricsself.data[metrics][index]return f{metrics[device_name]}{metrics[sdk_version]}def report(self):assert len(self.data[metrics])0for metric_name in self.get_metric_names():min_value,min_value_indexself.get_min_metric(metric_name)min_value_same_dev,min_value_index_same_devself.get_min_metric(metric_name,device_name)cur_valuefloat(self.cur_item[metrics][metric_name])print(f-------------------------------{metric_name}-------------------------------)print(f{cur_value}#{device_name}{sdk_version})if min_value_index_same_dev0:print(f{min_value_same_dev}#{self.get_metric_info(min_value_index_same_dev)})if min_value_index0:print(f{min_value}#{self.get_metric_info(min_value_index)})二.普通算子测试[test_clone.py]
from common import *
class TestCaseClone(TestCase):#如果不满足条件,则跳过这个测试unittest.skipIf(device_count1, Not enough devices) def test_todo(self):print(.TODO)#框架会自动遍历以下参数组合parametrize(shape, [(10240,20480),(128,256)])parametrize(dtype, [torch.float16,torch.float32])def test_clone(self,shape,dtype):#让这个函数循环执行loops次,统计第一次执行的耗时、后半段的平均时间、整个执行过程总的GPU内存使用量loop_decorator(loops5)def run(input_dev):outputinput_dev.clone()return output#记录整个测试的总耗时,保存统计量,输出摘要(self._testMethodName:测试方法,result:函数返回值,metrics:统计量)with TorchUtMetrics(ut_nameself._testMethodName,thresold0.2) as m:input_hosttorch.ones(shape,dtypedtype)*np.random.rand()input_devinput_host.to(device)metricsrun(input_dev,goldeninput_host.cpu())m.set_metrics(metrics)assert(metrics[mse]0)instantiate_parametrized_tests(TestCaseClone)if __name__ __main__:run_tests()三.集合通信测试[test_ccl.py]
from common import *
class TestCCL(MultiProcessTestCase):CCL测试用例def _create_process_group_vccl(self, world_size, store):dist.init_process_group(ccl_backend, world_sizeworld_size, rankself.rank, storestore) pg dist.distributed_c10d._get_default_group()return pgdef setUp(self):super().setUp()self._spawn_processes()def tearDown(self):super().tearDown()try:os.remove(self.file_name)except OSError:passpropertydef world_size(self):return 4#框架会自动遍历以下参数组合unittest.skipIf(device_count4, Not enough devices) parametrize(op,[dist.ReduceOp.SUM])parametrize(shape, [(1024,8192)])parametrize(dtype, [torch.int64])def test_allreduce(self,op,shape,dtype):if self.rank self.world_size:returnstore dist.FileStore(self.file_name, self.world_size)pg self._create_process_group_vccl(self.world_size, store)if not torch.distributed.is_initialized():returntorch.cuda.set_device(self.rank)device torch.device(device_type,self.rank)device_warmup(device)#让这个函数循环执行loops次,统计第一次执行的耗时、后半段的平均时间、整个执行过程总的GPU内存使用量loop_decorator(loops5,rankself.rank)def run(input_dev):dist.all_reduce(input_dev, opop)return input_dev#记录整个测试的总耗时,保存统计量,输出摘要(self._testMethodName:测试方法,result:函数返回值,metrics:统计量)with TorchUtMetrics(ut_nameself._testMethodName,thresold0.2,rankself.rank) as m:input_hosttorch.ones(shape,dtypedtype)*(100self.rank)gt[torch.ones(shape,dtypedtype)*(100i) for i in range(self.world_size)]gt_gt[0]for i in range(1,self.world_size):gt_gt_gt[i]input_devinput_host.to(device)metricsrun(input_dev,goldengt_)m.set_metrics(metrics)assert(metrics[mse]0)dist.destroy_process_group(pg)instantiate_parametrized_tests(TestCCL)if __name__ __main__:run_tests()四.测试命令
# 运行所有的测试
pytest -v -s -p no:warnings --htmltorch_report.html --self-contained-html --capturesys ./# 运行某一个测试
python3 test_clone.py -k test_clone_shape_(128, 256)_float32五.测试报告
