Kernel Serialization Support (#379)

* Serialization-rebased with main branch * Removing client_id from queue completion callbacks * removing debugging code * source formatting (clang-format v11) (#449) Co-authored-by: SrirakshaNag <SrirakshaNag@users.noreply.github.com> * moving ready signal handler to anonymous namespace * source formatting (clang-format v11) (#450) Co-authored-by: SrirakshaNag <SrirakshaNag@users.noreply.github.com> * Handling deque search better in queue destructor * source formatting (clang-format v11) (#451) Co-authored-by: SrirakshaNag <SrirakshaNag@users.noreply.github.com> * disabling test_total_runtime test in code coverage --------- Co-authored-by: Benjamin Welton <bewelton@amd.com> Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Co-authored-by: SrirakshaNag <SrirakshaNag@users.noreply.github.com> [ROCm/rocprofiler-sdk commit: f6198f226a]
2024-02-05 12:42:59 -06:00
commit 64b06a643e
@@ -49,3 +49,4 @@ add_subdirectory(async-copy-tracing)

 # rocprofv3 validation tests
 add_subdirectory(rocprofv3)
+add_subdirectory(counter-collection)
@@ -9,6 +9,7 @@ set(CMAKE_BUILD_RPATH "\$ORIGIN:\$ORIGIN/../lib")

 # applications used by integration tests which DO NOT link to rocprofiler-sdk-roctx
 add_subdirectory(simple-transpose)
+add_subdirectory(multistream)

 set(CMAKE_BUILD_RPATH
    "\$ORIGIN:\$ORIGIN/../lib:$<TARGET_FILE_DIR:rocprofiler-sdk-roctx::rocprofiler-sdk-roctx-shared-library>"
@@ -0,0 +1,46 @@
+#
+#
+#
+cmake_minimum_required(VERSION 3.21.0 FATAL_ERROR)
+
+if(NOT CMAKE_HIP_COMPILER)
+    find_program(
+        amdclangpp_EXECUTABLE
+        NAMES amdclang++
+        HINTS ${ROCM_PATH} ENV ROCM_PATH /opt/rocm
+        PATHS ${ROCM_PATH} ENV ROCM_PATH /opt/rocm
+        PATH_SUFFIXES bin llvm/bin NO_CACHE)
+    mark_as_advanced(amdclangpp_EXECUTABLE)
+
+    if(amdclangpp_EXECUTABLE)
+        set(CMAKE_HIP_COMPILER "${amdclangpp_EXECUTABLE}")
+    endif()
+endif()
+
+project(rocprofiler-test-app-multistream LANGUAGES CXX HIP)
+
+foreach(_TYPE DEBUG MINSIZEREL RELEASE RELWITHDEBINFO)
+    if("${CMAKE_HIP_FLAGS_${_TYPE}}" STREQUAL "")
+        set(CMAKE_HIP_FLAGS_${_TYPE} "${CMAKE_CXX_FLAGS_${_TYPE}}")
+    endif()
+endforeach()
+
+set(CMAKE_CXX_STANDARD 17)
+set(CMAKE_CXX_EXTENSIONS OFF)
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
+set(CMAKE_HIP_STANDARD 17)
+set(CMAKE_HIP_EXTENSIONS OFF)
+set(CMAKE_HIP_STANDARD_REQUIRED ON)
+
+set_source_files_properties(multistream_app.cpp PROPERTIES LANGUAGE HIP)
+add_executable(multistream)
+target_sources(multistream PRIVATE multistream_app.cpp)
+target_compile_options(multistream PRIVATE -W -Wall -Wextra -Wpedantic -Wshadow -Werror)
+
+find_package(Threads REQUIRED)
+target_link_libraries(multistream PRIVATE Threads::Threads)
+
+install(
+    TARGETS multistream
+    DESTINATION bin
+    COMPONENT tests)
@@ -0,0 +1,116 @@
+// MIT License
+//
+// Copyright (c) 2023 Advanced Micro Devices, Inc. All rights reserved.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+#include <hip/hip_runtime.h>
+#include <vector>
+#define HIP_ASSERT(call)                                                                           \
+    do                                                                                             \
+    {                                                                                              \
+        hipError_t err = call;                                                                     \
+        if(err != hipSuccess)                                                                      \
+        {                                                                                          \
+            fprintf(stderr, "%s\n", hipGetErrorString(err));                                       \
+            abort();                                                                               \
+        }                                                                                          \
+    } while(0)
+
+__device__ int counter = 0;
+__global__ void
+add(int n, float* x, float* y)
+{
+    if(__hip_atomic_load(&counter, __ATOMIC_ACQUIRE, __HIP_MEMORY_SCOPE_AGENT) != 0)
+    {
+        abort();
+    }
+    __hip_atomic_fetch_add(&counter, 1, __ATOMIC_RELEASE, __HIP_MEMORY_SCOPE_SYSTEM);
+
+    int index  = blockIdx.x * blockDim.x + threadIdx.x;
+    int stride = blockDim.x * gridDim.x;
+    for(int i = index; i < n; i += stride)
+        y[i] = x[i] + y[i];
+    __hip_atomic_fetch_add(&counter, -1, __ATOMIC_RELEASE, __HIP_MEMORY_SCOPE_SYSTEM);
+}
+
+void
+LaunchMultiStreamKernels()
+{
+    int    N = 1 << 4;
+    float* x = new float[N];
+    float* y = new float[N];
+    float* d_x;
+    float* d_y;
+    //   Allocate Unified Memory -- accessible from CPU or GPU
+    HIP_ASSERT(hipMallocManaged(&d_x, N * sizeof(float)));
+    HIP_ASSERT(hipMallocManaged(&d_y, N * sizeof(float)));
+
+    //   initialize x and y arrays on the host
+    for(int i = 0; i < N; i++)
+    {
+        x[i] = 1.0f;
+        y[i] = 2.0f;
+    }
+    std::vector<hipStream_t> hip_streams;
+    for(int i = 0; i < 100; i++)
+    {
+        hipStream_t stream;
+        HIP_ASSERT(hipStreamCreate(&stream));
+        hip_streams.push_back(stream);
+    }
+    HIP_ASSERT(hipMemcpy(d_x, x, N * sizeof(float), hipMemcpyHostToDevice));
+    HIP_ASSERT(hipMemcpy(d_y, y, N * sizeof(float), hipMemcpyHostToDevice));
+
+    // Launch kernel on 1M elements on the GPU
+    int blockSize = 64;
+    // This Kernel will always be launched with one wave
+    int numBlocks = 1;
+    for(int i = 0; i < 100; i++)
+    {
+        for(size_t j = 0; j < hip_streams.size(); j++)
+        {
+            hipLaunchKernelGGL(add, numBlocks, blockSize, 0, hip_streams[j], N, d_x, d_y);
+        }
+    }
+
+    // Wait for GPU to finish before accessing on host
+    HIP_ASSERT(hipDeviceSynchronize());
+
+    HIP_ASSERT(hipMemcpy(x, d_x, N * sizeof(float), hipMemcpyDeviceToHost));
+    HIP_ASSERT(hipMemcpy(y, d_y, N * sizeof(float), hipMemcpyDeviceToHost));
+
+    //   Free memory
+    HIP_ASSERT(hipFree(d_x));
+    HIP_ASSERT(hipFree(d_y));
+
+    delete[] x;
+    delete[] y;
+
+    for(size_t i = 0; i < hip_streams.size(); i++)
+    {
+        HIP_ASSERT(hipStreamDestroy(hip_streams[i]));
+    }
+}
+
+int
+main()
+{
+    LaunchMultiStreamKernels();
+}
@@ -231,6 +231,15 @@ save(ArchiveT& ar, rocprofiler_buffer_tracing_hsa_api_record_t data)
    save_buffer_tracing_api_record(ar, data);
 }

+template <typename ArchiveT>
+void
+save(ArchiveT& ar, rocprofiler_record_counter_t data)
+{
+    SAVE_DATA_FIELD(id);
+    SAVE_DATA_FIELD(counter_value);
+    SAVE_DATA_FIELD(corr_id);
+}
+
 template <typename ArchiveT>
 void
 save(ArchiveT& ar, rocprofiler_buffer_tracing_hip_api_record_t data)
@@ -0,0 +1,47 @@
+#
+#
+#
+cmake_minimum_required(VERSION 3.21.0 FATAL_ERROR)
+
+project(
+    rocprofiler-tests-counter-collection
+    LANGUAGES CXX
+    VERSION 0.0.0)
+
+find_package(rocprofiler-sdk REQUIRED)
+
+if(ROCPROFILER_MEMCHECK_PRELOAD_ENV)
+    set(PRELOAD_ENV
+        "${ROCPROFILER_MEMCHECK_PRELOAD_ENV}:$<TARGET_FILE:rocprofiler-sdk-json-tool>")
+else()
+    set(PRELOAD_ENV "LD_PRELOAD=$<TARGET_FILE:rocprofiler-sdk-json-tool>")
+endif()
+
+add_test(NAME test-counter-collection-execute COMMAND $<TARGET_FILE:multistream>)
+
+set_tests_properties(
+    test-counter-collection-execute
+    PROPERTIES
+        TIMEOUT
+        45
+        LABELS
+        "integration-tests"
+        ENVIRONMENT
+        "${PRELOAD_ENV};HSA_TOOLS_LIB=$<TARGET_FILE:rocprofiler::rocprofiler-shared-library>;ROCPROFILER_TOOL_OUTPUT_FILE=counter-collection-test.json;ROCPROFILER_TOOL_CONTEXTS=COUNTER_COLLECTION;ROCPROF_COUNTERS=SQ_WAVES_sum"
+        FAIL_REGULAR_EXPRESSION
+        "threw an exception")
+
+foreach(FILENAME validate.py pytest.ini conftest.py)
+    configure_file(${CMAKE_CURRENT_SOURCE_DIR}/${FILENAME}
+                   ${CMAKE_CURRENT_BINARY_DIR}/${FILENAME} COPYONLY)
+endforeach()
+
+add_test(NAME test-counter-collection-validate
+         COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_BINARY_DIR}/validate.py --input
+                 ${CMAKE_CURRENT_BINARY_DIR}/counter-collection-test.json)
+
+set_tests_properties(
+    test-counter-collection-validate
+    PROPERTIES TIMEOUT 45 LABELS "integration-tests" DEPENDS
+               test-counter-collection-execute FAIL_REGULAR_EXPRESSION
+               "threw an exception")
@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+
+import json
+import pytest
+
+
+def pytest_addoption(parser):
+    parser.addoption(
+        "--input",
+        action="store",
+        default="counter-collection-test.json",
+        help="Input JSON",
+    )
+
+
+@pytest.fixture
+def input_data(request):
+    filename = request.config.getoption("--input")
+    with open(filename, "r") as inp:
+        return json.load(inp)
@@ -0,0 +1,4 @@
+
+[pytest]
+addopts = --durations=20 -ras -vv
+testpaths = validate.py
@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+
+import sys
+import pytest
+
+
+# helper function
+def node_exists(name, data, min_len=1):
+    assert name in data
+    assert data[name] is not None
+    assert len(data[name]) >= min_len
+
+
+def test_data_structure(input_data):
+    """verify minimum amount of expected data is present"""
+    node_exists("rocprofiler-sdk-json-tool", input_data)
+    rocp_data = input_data
+    node_exists("names", rocp_data["rocprofiler-sdk-json-tool"]["buffer_records"])
+    node_exists(
+        "counter_collection", rocp_data["rocprofiler-sdk-json-tool"]["buffer_records"]
+    )
+
+
+def test_counter_values(input_data):
+    data = input_data
+    counter_info = {}
+    for itr in data["rocprofiler-sdk-json-tool"]["buffer_records"]["counter_collection"]:
+        value = itr["counter_value"]
+        assert value == 1
+
+
+if __name__ == "__main__":
+    exit_code = pytest.main(["-x", __file__] + sys.argv[1:])
+    sys.exit(exit_code)
@@ -11,7 +11,7 @@ project(
 find_package(rocprofiler-sdk REQUIRED)

 set(PYTEST_ARGS)
-if(ROCPROFILER_MEMCHECK MATCHES "(Address|Thread)Sanitizer")
+if(ROCPROFILER_MEMCHECK MATCHES "(Address|Thread)Sanitizer" OR ROCPROFILER_BUILD_CODECOV)
    set(PYTEST_ARGS -k "not test_total_runtime")
 endif()

@@ -43,6 +43,8 @@
 #include <rocprofiler-sdk/registration.h>
 #include <rocprofiler-sdk/rocprofiler.h>

+#include <glog/logging.h>
+
 #include <unistd.h>
 #include <atomic>
 #include <cassert>
@@ -57,6 +59,7 @@
 #include <iostream>
 #include <map>
 #include <mutex>
+#include <shared_mutex>
 #include <string>
 #include <string_view>
 #include <thread>
@@ -361,15 +364,123 @@ struct marker_api_callback_record_t
    }
 };

-auto code_object_records   = std::deque<code_object_callback_record_t>{};
-auto kernel_symbol_records = std::deque<kernel_symbol_callback_record_t>{};
-auto hsa_api_cb_records    = std::deque<hsa_api_callback_record_t>{};
-auto marker_api_cb_records = std::deque<marker_api_callback_record_t>{};
-auto hip_api_cb_records    = std::deque<hip_api_callback_record_t>{};
+auto code_object_records           = std::deque<code_object_callback_record_t>{};
+auto kernel_symbol_records         = std::deque<kernel_symbol_callback_record_t>{};
+auto hsa_api_cb_records            = std::deque<hsa_api_callback_record_t>{};
+auto marker_api_cb_records         = std::deque<marker_api_callback_record_t>{};
+auto counter_collection_bf_records = std::deque<rocprofiler_record_counter_t>{};
+auto hip_api_cb_records            = std::deque<hip_api_callback_record_t>{};

 rocprofiler_thread_id_t
 push_external_correlation();

+void
+counter_collection_buffered(rocprofiler_context_id_t, /*context*/
+                            rocprofiler_buffer_id_t,  /*buffer_id*/
+                            rocprofiler_record_header_t** headers,
+                            size_t                        num_headers,
+                            void*, /*user_data*/
+                            uint64_t /*drop_count*/)
+{
+    if(num_headers == 0)
+        throw std::runtime_error{"rocprofiler invoked a buffer callback with no headers "
+                                 "this should never happen"};
+
+    else if(headers == nullptr)
+        throw std::runtime_error{"rocprofiler invoked a buffer callback with a null pointer to the "
+                                 "array of headers. this should never happen"};
+    for(size_t i = 0; i < num_headers; ++i)
+    {
+        auto* header = headers[i];
+        if(header->category == ROCPROFILER_BUFFER_CATEGORY_COUNTERS && header->kind == 0)
+        {
+            auto* profiler_record = static_cast<rocprofiler_record_counter_t*>(header->payload);
+            counter_collection_bf_records.emplace_back(*profiler_record);
+        }
+    }
+}
+
+void
+dispatch_callback(rocprofiler_queue_id_t, /*queue_id*/
+                  const rocprofiler_agent_t* agent,
+                  rocprofiler_correlation_id_t,        /*correlation_id*/
+                  const hsa_kernel_dispatch_packet_t*, /*dispatch_packet*/
+                  uint64_t,                            /*kernel_id*/
+                  void* /*callback_data_args*/,
+                  rocprofiler_profile_config_id_t* config)
+{
+    static std::shared_mutex                                             m_mutex       = {};
+    static std::unordered_map<uint64_t, rocprofiler_profile_config_id_t> profile_cache = {};
+
+    auto search_cache = [&]() {
+        if(auto pos = profile_cache.find(agent->id.handle); pos != profile_cache.end())
+        {
+            *config = pos->second;
+            return true;
+        }
+        return false;
+    };
+
+    {
+        auto rlock = std::shared_lock{m_mutex};
+        if(search_cache()) return;
+    }
+
+    auto wlock = std::unique_lock{m_mutex};
+    if(search_cache()) return;
+
+    // Counters we want to collect (here its SQ_WAVES_sum)
+    auto* counters_env = getenv("ROCPROF_COUNTERS");
+    if(std::string(counters_env) != "SQ_WAVES_sum")
+        LOG(FATAL) << "Counter not supported in the test tool";
+
+    std::set<std::string> counters_to_collect = {"SQ_WAVES_sum"};
+    // GPU Counter IDs
+    std::vector<rocprofiler_counter_id_t> gpu_counters;
+
+    // Iterate through the agents and get the counters available on that agent
+    ROCPROFILER_CALL(rocprofiler_iterate_agent_supported_counters(
+                         agent->id,
+                         []([[maybe_unused]] rocprofiler_agent_id_t id,
+                            rocprofiler_counter_id_t*               counters,
+                            size_t                                  num_counters,
+                            void*                                   user_data) {
+                             std::vector<rocprofiler_counter_id_t>* vec =
+                                 static_cast<std::vector<rocprofiler_counter_id_t>*>(user_data);
+                             for(size_t i = 0; i < num_counters; i++)
+                             {
+                                 vec->push_back(counters[i]);
+                             }
+                             return ROCPROFILER_STATUS_SUCCESS;
+                         },
+                         static_cast<void*>(&gpu_counters)),
+                     "Could not fetch supported counters");
+
+    std::vector<rocprofiler_counter_id_t> collect_counters;
+    // Look for the counters contained in counters_to_collect in gpu_counters
+    for(auto& counter : gpu_counters)
+    {
+        const char* name;
+        size_t      size;
+        ROCPROFILER_CALL(rocprofiler_query_counter_name(counter, &name, &size),
+                         "Could not query name");
+        if(counters_to_collect.count(std::string(name)) > 0)
+        {
+            collect_counters.push_back(counter);
+        }
+    }
+
+    // Create a colleciton profile for the counters
+    rocprofiler_profile_config_id_t profile;
+    ROCPROFILER_CALL(rocprofiler_create_profile_config(
+                         agent->id, collect_counters.data(), collect_counters.size(), &profile),
+                     "Could not construct profile cfg");
+
+    profile_cache.emplace(agent->id.handle, profile);
+    // Return the profile to collect those counters for this dispatch
+    *config = profile;
+}
+
 void
 tool_tracing_callback(rocprofiler_callback_tracing_record_t record,
                      rocprofiler_user_data_t* /*user_data*/,
@@ -584,12 +695,14 @@ rocprofiler_context_id_t hip_api_buffered_ctx    = {};
 rocprofiler_context_id_t marker_api_buffered_ctx = {};
 rocprofiler_context_id_t kernel_dispatch_ctx     = {};
 rocprofiler_context_id_t memory_copy_ctx         = {};
+rocprofiler_context_id_t counter_collection_ctx  = {};
 // buffers
 rocprofiler_buffer_id_t hsa_api_buffered_buffer    = {};
 rocprofiler_buffer_id_t hip_api_buffered_buffer    = {};
 rocprofiler_buffer_id_t marker_api_buffered_buffer = {};
 rocprofiler_buffer_id_t kernel_dispatch_buffer     = {};
 rocprofiler_buffer_id_t memory_copy_buffer         = {};
+rocprofiler_buffer_id_t counter_collection_buffer  = {};

 auto contexts = std::unordered_map<std::string_view, rocprofiler_context_id_t*>{
    {"HSA_API_CALLBACK", &hsa_api_callback_ctx},
@@ -600,13 +713,15 @@ auto contexts = std::unordered_map<std::string_view, rocprofiler_context_id_t*>{
    {"HIP_API_BUFFERED", &hip_api_buffered_ctx},
    {"MARKER_API_BUFFERED", &marker_api_buffered_ctx},
    {"KERNEL_DISPATCH", &kernel_dispatch_ctx},
-    {"MEMORY_COPY", &memory_copy_ctx}};
+    {"MEMORY_COPY", &memory_copy_ctx},
+    {"COUNTER_COLLECTION", &counter_collection_ctx}};

-auto buffers = std::array<rocprofiler_buffer_id_t*, 5>{&hsa_api_buffered_buffer,
+auto buffers = std::array<rocprofiler_buffer_id_t*, 6>{&hsa_api_buffered_buffer,
                                                       &hip_api_buffered_buffer,
                                                       &marker_api_buffered_buffer,
                                                       &kernel_dispatch_buffer,
-                                                       &memory_copy_buffer};
+                                                       &memory_copy_buffer,
+                                                       &counter_collection_buffer};

 auto agents = std::vector<rocprofiler_agent_t>{};

@@ -785,6 +900,20 @@ tool_init(rocprofiler_client_finalize_t fini_func, void* tool_data)
                                                     marker_api_buffered_buffer),
        "buffer tracing service configure");

+    ROCPROFILER_CALL(rocprofiler_create_buffer(counter_collection_ctx,
+                                               4096,
+                                               2048,
+                                               ROCPROFILER_BUFFER_POLICY_LOSSLESS,
+                                               counter_collection_buffered,
+                                               nullptr,
+                                               &counter_collection_buffer),
+                     "buffer creation");
+
+    ROCPROFILER_CALL(
+        rocprofiler_configure_buffered_dispatch_profile_counting_service(
+            counter_collection_ctx, counter_collection_buffer, dispatch_callback, nullptr),
+        "setup buffered service");
+
    ROCPROFILER_CALL(
        rocprofiler_configure_buffer_tracing_service(kernel_dispatch_ctx,
                                                     ROCPROFILER_BUFFER_TRACING_KERNEL_DISPATCH,
@@ -885,7 +1014,8 @@ tool_fini(void* tool_data)
              << ", memory_copy_records=" << memory_copy_records.size()
              << ", hsa_api_bf_records=" << hsa_api_bf_records.size()
              << ", hip_api_bf_records=" << hip_api_bf_records.size()
-              << ", marker_api_bf_records=" << marker_api_bf_records.size() << " ...\n"
+              << ", marker_api_bf_records=" << marker_api_bf_records.size()
+              << ", counter_collection_records" << counter_collection_bf_records.size() << "...\n"
              << std::flush;

    auto* _call_stack = static_cast<call_stack_t*>(tool_data);
@@ -966,6 +1096,7 @@ tool_fini(void* tool_data)
            json_ar(cereal::make_nvp("hsa_api_traces", hsa_api_bf_records));
            json_ar(cereal::make_nvp("hip_api_traces", hip_api_bf_records));
            json_ar(cereal::make_nvp("marker_api_traces", marker_api_bf_records));
+            json_ar(cereal::make_nvp("counter_collection", counter_collection_bf_records));
        } catch(std::exception& e)
        {
            std::cerr << "[" << getpid() << "][" << __FUNCTION__
@@ -994,6 +1125,11 @@ start()
    {
        if(itr.second && !is_active(*itr.second))
        {
+            if(itr.first == "COUNTER_COLLECTION")
+            {
+                auto* counters = getenv("ROCPROF_COUNTERS");
+                if(!counters) continue;
+            }
            ROCPROFILER_CALL(rocprofiler_start_context(*itr.second), "context start");
        }
    }