| 32 | ) |
| 33 | @pytest.mark.require_ngpu(1) |
| 34 | def test_profiler(format, trace_mode): |
| 35 | tempdir = tempfile.TemporaryDirectory() |
| 36 | profile_prefix = tempdir.name |
| 37 | profile_path = os.path.join(profile_prefix, "{}.{}".format(os.getpid(), format)) |
| 38 | |
| 39 | def infer(): |
| 40 | with scope("my_scope"): |
| 41 | oup = Simple()(tensor([1.23], dtype="float32")) |
| 42 | return oup |
| 43 | |
| 44 | if trace_mode: |
| 45 | infer = trace(symbolic=trace_mode)(infer) |
| 46 | |
| 47 | with Profiler(profile_prefix, format=format): |
| 48 | infer() |
| 49 | |
| 50 | assert os.path.exists(profile_path), "profiling results not found" |
| 51 | |
| 52 | if format == "chrome_timeline.json": |
| 53 | with open(profile_path, "r") as f: |
| 54 | events = json.load(f) |
| 55 | if isinstance(events, dict): |
| 56 | assert "traceEvents" in events |
| 57 | events = events["traceEvents"] |
| 58 | prev_ts = {} |
| 59 | scope_count = 0 |
| 60 | for event in events: |
| 61 | if "dur" in event: |
| 62 | assert event["dur"] >= 0 |
| 63 | elif "ts" in event and "tid" in event: |
| 64 | ts = event["ts"] |
| 65 | tid = event["tid"] |
| 66 | if ts != 0: |
| 67 | assert (tid not in prev_ts) or prev_ts[tid] <= ts |
| 68 | prev_ts[tid] = ts |
| 69 | if "name" in event and event["name"] == "my_scope": |
| 70 | scope_count += 1 |
| 71 | assert scope_count > 0 and scope_count % 2 == 0 |
| 72 | |
| 73 | |
| 74 | @pytest.mark.parametrize("format", ["chrome_timeline.json", "memory_flow.svg"]) |