[rocprofiler-sdk] Handle validate tests script when Blit kernels are used for mem copies

projects/rocprofiler-sdk/tests/rocprofv3/attachment/attach-once/CMakeLists.txt

@@ -19,6 +19,9 @@ else()
     set(IS_DISABLED OFF)
 endif()
 
+# Until this test is fixed in PSDB-tests
+set(IS_DISABLED ON)
+
 if(ROCPROFILER_MEMCHECK STREQUAL "LeakSanitizer")
     set(LOG_LEVEL "warning") # info produces memory leak
 else()

projects/rocprofiler-sdk/tests/rocprofv3/tracing/validate.py

@@ -125,9 +125,13 @@ def test_kernel_trace(kernel_input_data):
         "_Z15matrixTransposePfS_i.kd",
         "matrixTranspose(float*, float*, int)",
     )
+    found_valid_kernel = False
 
-    assert len(kernel_input_data) == 1
+    # Blit kernel maybe recorded, when greater than 1.
+    assert len(kernel_input_data) >= 1
     for row in kernel_input_data:
+        if re.search(r"__amd_rocclr_.*", row["Kernel_Name"]):
+            continue
         assert row["Kind"] == "KERNEL_DISPATCH"
         assert int(row["Agent_Id"].split(" ")[-1]) >= 0
         assert int(row["Queue_Id"]) > 0
@@ -141,6 +145,8 @@ def test_kernel_trace(kernel_input_data):
         assert int(row["Grid_Size_Y"]) == 1024
         assert int(row["Grid_Size_Z"]) == 1
         assert int(row["End_Timestamp"]) >= int(row["Start_Timestamp"])
+        found_valid_kernel = True
+    assert found_valid_kernel
 
 
 def test_host_functions_json(json_data):
@@ -184,8 +190,10 @@ def get_kind_name(kind_id):
         "_Z15matrixTransposePfS_i.kd",
         "matrixTranspose(float*, float*, int)",
     )
+    found_valid_kernel = False
     kernel_dispatch_data = data["buffer_records"]["kernel_dispatch"]
-    assert len(kernel_dispatch_data) == 1
+    # Blit kernel maybe recorded, when greater than 1.
+    assert len(kernel_dispatch_data) >= 1
     for dispatch in kernel_dispatch_data:
         dispatch_info = dispatch["dispatch_info"]
         kernel_name = get_kernel_name(dispatch_info["kernel_id"])
@@ -195,34 +203,45 @@ def get_kind_name(kind_id):
         assert dispatch_info["agent_id"]["handle"] > 0
         assert dispatch_info["queue_id"]["handle"] > 0
         assert dispatch_info["kernel_id"] > 0
-        if not re.search(r"__amd_rocclr_.*", kernel_name):
-            assert kernel_name in valid_kernel_names
-
+        if re.search(r"__amd_rocclr_.*", kernel_name):
+            continue
+        assert kernel_name in valid_kernel_names
         assert dispatch_info["workgroup_size"]["x"] == 4
         assert dispatch_info["workgroup_size"]["y"] == 4
         assert dispatch_info["workgroup_size"]["z"] == 1
         assert dispatch_info["grid_size"]["x"] == 1024
         assert dispatch_info["grid_size"]["y"] == 1024
         assert dispatch_info["grid_size"]["z"] == 1
         assert dispatch["end_timestamp"] >= dispatch["start_timestamp"]
+        found_valid_kernel = True
+    assert found_valid_kernel
 
 
-def test_memory_copy_trace(agent_info_input_data, memory_copy_input_data):
+def test_memory_copy_trace(
+    agent_info_input_data,
+    memory_copy_input_data,
+    kernel_input_data,
+):
     def get_agent(node_id):
         for row in agent_info_input_data:
             if row["Logical_Node_Id"] == node_id:
                 return row
         return None
 
+    has_blit_kernel = any(
+        re.search(r"__amd_rocclr_.*", row["Kernel_Name"]) for row in kernel_input_data
+    )
+    # memory copies may be missing if rocr decides to use blit kernels
+    if not has_blit_kernel:
+        assert len(memory_copy_input_data) > 0
+
+    directions = set()
+    valid_directions = ("MEMORY_COPY_HOST_TO_DEVICE", "MEMORY_COPY_DEVICE_TO_HOST")
     for row in memory_copy_input_data:
         assert row["Kind"] == "MEMORY_COPY"
-
-    assert len(memory_copy_input_data) == 2
-
-    def test_row(idx, direction):
-        assert direction in ("MEMORY_COPY_HOST_TO_DEVICE", "MEMORY_COPY_DEVICE_TO_HOST")
-        row = memory_copy_input_data[idx]
-        assert row["Direction"] == direction
+        direction = row["Direction"]
+        assert direction in valid_directions
+        directions.add(direction)
         src_agent = get_agent(row["Source_Agent_Id"].split(" ")[-1])
         dst_agent = get_agent(row["Destination_Agent_Id"].split(" ")[-1])
         assert src_agent is not None and dst_agent is not None, f"{agent_info_input_data}"
@@ -235,8 +254,13 @@ def test_row(idx, direction):
         assert int(row["Correlation_Id"]) > 0
         assert int(row["End_Timestamp"]) >= int(row["Start_Timestamp"])
 
-    test_row(0, "MEMORY_COPY_HOST_TO_DEVICE")
-    test_row(1, "MEMORY_COPY_DEVICE_TO_HOST")
+    expected_directions = set(valid_directions)
+    if has_blit_kernel:
+        assert directions.issubset(expected_directions)
+    else:
+        assert (
+            directions == expected_directions
+        ), f"Expected directions {expected_directions}, got {directions}"
 
 
 def test_memory_copy_json_trace(json_data):
@@ -256,28 +280,46 @@ def get_agent(node_id):
         return None
 
     # one threads * two directions
-    assert len(memory_copy_data) >= 2, f"{memory_copy_data}"
-    assert (len(memory_copy_data) % 2) == 0, f"{memory_copy_data}"
-
-    def test_row(idx, direction):
-        assert direction in ("MEMORY_COPY_HOST_TO_DEVICE", "MEMORY_COPY_DEVICE_TO_HOST")
-        row = memory_copy_data[idx]
+    kernel_dispatch_data = buffer_records.get("kernel_dispatch", [])
+    has_blit_kernel = False
+    for dispatch in kernel_dispatch_data:
+        dispatch_info = dispatch["dispatch_info"]
+        kernel_name = data["kernel_symbols"][dispatch_info["kernel_id"]][
+            "formatted_kernel_name"
+        ]
+        if re.search(r"__amd_rocclr_.*", kernel_name):
+            has_blit_kernel = True
+            break
+    # memory copies may be missing if rocr decides to use blit kernels
+    if not has_blit_kernel:
+        assert len(memory_copy_data) > 0, f"{memory_copy_data}"
+
+    directions = set()
+    valid_directions = ("MEMORY_COPY_HOST_TO_DEVICE", "MEMORY_COPY_DEVICE_TO_HOST")
+    for row in memory_copy_data:
         src_agent = get_agent(row["src_agent_id"])
         dst_agent = get_agent(row["dst_agent_id"])
         assert get_kind_name(row["kind"]) == "MEMORY_COPY"
         assert src_agent is not None, f"{row}"
         assert dst_agent is not None, f"{row}"
-        if direction == "MEMORY_COPY_HOST_TO_DEVICE":
-            assert src_agent["type"] == 1
-            assert dst_agent["type"] == 2
+        if src_agent["type"] == 1 and dst_agent["type"] == 2:
+            directions.add("MEMORY_COPY_HOST_TO_DEVICE")
+        elif src_agent["type"] == 2 and dst_agent["type"] == 1:
+            directions.add("MEMORY_COPY_DEVICE_TO_HOST")
         else:
-            assert src_agent["type"] == 2
-            assert dst_agent["type"] == 1
+            assert (
+                False
+            ), f"Unexpected agent types: {src_agent['type']} -> {dst_agent['type']}"
         assert row["correlation_id"]["internal"] > 0
         assert row["end_timestamp"] >= row["start_timestamp"]
 
-    test_row(0, "MEMORY_COPY_HOST_TO_DEVICE")
-    test_row(1, "MEMORY_COPY_DEVICE_TO_HOST")
+    expected_directions = set(valid_directions)
+    if has_blit_kernel:
+        assert directions.issubset(expected_directions)
+    else:
+        assert (
+            directions == expected_directions
+        ), f"Expected directions {expected_directions}, got {directions}"
 
 
 def test_marker_api_trace(marker_input_data):