diff --git a/projects/rocprofiler-systems/CHANGELOG.md b/projects/rocprofiler-systems/CHANGELOG.md index ef8207e0aa3b..28855e7b3f27 100644 --- a/projects/rocprofiler-systems/CHANGELOG.md +++ b/projects/rocprofiler-systems/CHANGELOG.md @@ -34,6 +34,19 @@ replaces Perfetto as the primary trace output. `ROCPROFSYS_CONFIG_FILE` is preserved, and the one given on the command line is appended to it. +- Pausing sampling now stops the underlying per-thread timers instead of only + discarding the samples they produce. Previously a paused sampler kept delivering + timer signals, so the profiled application's sleeps were still interrupted + throughout a window in which no data was being collected. + +### Fixed + +- `ROCPROFSYS_TRACE_DELAY`/`ROCPROFSYS_TRACE_DURATION` now actually gate GPU context + startup, producing a real gap in cached GPU/RocPD data. Previously they only + suppressed downstream category emission. For GPU-only tracing with no marker + domain or trace region configured, the configured delay had no effect at all on + when GPU data collection actually began. + ### Removed - Removed the `ROCPROFSYS_BUILD_SQLITE3` CMake option and the in-tree SQLite3/rocpd diff --git a/projects/rocprofiler-systems/docs/reference/development-guide.rst b/projects/rocprofiler-systems/docs/reference/development-guide.rst index ea703c202481..5ab151c2476e 100644 --- a/projects/rocprofiler-systems/docs/reference/development-guide.rst +++ b/projects/rocprofiler-systems/docs/reference/development-guide.rst @@ -419,26 +419,45 @@ a new internal thread is created to handle the new samplers. Time-window constraint model ======================================== -With the recent introduction of tracing delay and duration, the -`constraint namespace `_ -was introduced to improve the management of delays and duration limits for -data collection. The ``spec`` class accepts a clock identifier, a delay value, a duration value, and an -integer indicating how many times to repeat the delay and duration cycle. It is therefore -possible to perform tasks such as periodically enabling tracing for brief periods -of time in between long periods without data collection while the application runs. The -syntax follows the format ``clock_identifier:delay:capture_duration:cycles``, so a value of -``10:1:3`` for the last three parameters represents the following sequence of operations: +The time-window constraint model manages delays and duration limits for data collection. +Each window entry specifies a delay before collection begins, a duration for how long +collection is active, and an optional repeat count. The syntax for ``ROCPROFSYS_TRACE_PERIODS`` +entries follows the format ``delay:duration`` or ``delay:duration:repeat``, so a value of +``10:1:3`` represents the following sequence of operations: * Ten seconds where no data is collected, then one second where it is * Ten seconds where no data is collected, then one second where it is * Ten seconds where no data is collected, then one second where it is * Stop -As another example, ``ROCPROFSYS_TRACE_PERIODS = realtime:10:1:5 process_cputime:10:2:20`` translates -to this sequence: +Clock selection +--------------- -* Five cycles of: no data collection for ten seconds of real-time followed by one second of data collection -* Twenty cycles of: no data collection for ten seconds of process CPU time followed by two CPU-time seconds of data collection +``ROCPROFSYS_TRACE_PERIOD_CLOCK_ID`` controls what the delay and duration values mean. +It accepts exactly two values: + +``realtime`` (default) + Delays and durations are measured in wall-clock time using a monotonic + steady clock. A ten-second delay waits for ten real seconds regardless of + how many CPU cores are active or how busy they are. + +``cputime`` + Delays and durations are measured in process CPU time + (``CLOCK_PROCESS_CPUTIME_ID``). A ten-second delay waits until the process + has consumed ten CPU-seconds in total across all threads. This is useful + when you want to skip a fixed amount of computation rather than a fixed + elapsed time — for example, to profile only after a workload's warm-up + phase, measured in CPU work rather than calendar seconds. + + Be aware that rocprof-sys instrumentation itself contributes to the + CPU-time counter, so the effective delay will be slightly shorter than the + configured value in heavily instrumented code. + +Example — profile only the steady-state CPU work, skipping the first ten +CPU-seconds and capturing the next two, repeated twenty times:: + + ROCPROFSYS_TRACE_PERIOD_CLOCK_ID=cputime + ROCPROFSYS_TRACE_PERIODS=10:2:20 Eventually, the goal is to migrate all subsets of data collection which currently support more rudimentary models of time window constraints, such as process sampling and causal profiling, diff --git a/projects/rocprofiler-systems/source/lib/core/CMakeLists.txt b/projects/rocprofiler-systems/source/lib/core/CMakeLists.txt index 83d1341623d6..04e5faa2cdbb 100644 --- a/projects/rocprofiler-systems/source/lib/core/CMakeLists.txt +++ b/projects/rocprofiler-systems/source/lib/core/CMakeLists.txt @@ -7,7 +7,6 @@ set(core_sources ${CMAKE_CURRENT_LIST_DIR}/argparse.cpp ${CMAKE_CURRENT_LIST_DIR}/categories.cpp ${CMAKE_CURRENT_LIST_DIR}/config.cpp - ${CMAKE_CURRENT_LIST_DIR}/constraint.cpp ${CMAKE_CURRENT_LIST_DIR}/cpu.cpp ${CMAKE_CURRENT_LIST_DIR}/dynamic_library.cpp ${CMAKE_CURRENT_LIST_DIR}/exception.cpp @@ -34,7 +33,6 @@ set(core_headers ${CMAKE_CURRENT_LIST_DIR}/common_types.hpp ${CMAKE_CURRENT_LIST_DIR}/concepts.hpp ${CMAKE_CURRENT_LIST_DIR}/config.hpp - ${CMAKE_CURRENT_LIST_DIR}/constraint.hpp ${CMAKE_CURRENT_LIST_DIR}/cpu.hpp ${CMAKE_CURRENT_LIST_DIR}/demangler.hpp ${CMAKE_CURRENT_LIST_DIR}/dynamic_library.hpp @@ -63,6 +61,7 @@ add_library( target_sources(rocprofiler-systems-core-library PRIVATE ${core_sources} ${core_headers}) add_subdirectory(binary) add_subdirectory(components) +add_subdirectory(config) add_subdirectory(containers) add_subdirectory(control) add_subdirectory(perfetto) diff --git a/projects/rocprofiler-systems/source/lib/core/categories.cpp b/projects/rocprofiler-systems/source/lib/core/categories.cpp index 3df4f02a3805..4cb1582fe1df 100644 --- a/projects/rocprofiler-systems/source/lib/core/categories.cpp +++ b/projects/rocprofiler-systems/source/lib/core/categories.cpp @@ -4,8 +4,6 @@ #include "core/categories.hpp" #include "core/common.hpp" #include "core/config.hpp" -#include "core/constraint.hpp" -#include "core/state.hpp" #include "core/timemory.hpp" #include "core/utility.hpp" @@ -72,46 +70,6 @@ setup() { // disable specified categories disable_categories(); - - auto _trace_specs = constraint::get_trace_specs(); - - if(!_trace_specs.empty()) - { - auto _trace_stages = constraint::get_trace_stages(); - - _trace_stages.init = [](const constraint::spec& _spec) { - if(_spec.delay > 1.0e-3) disable_categories(config::get_enabled_categories()); - return state::process::get() < state::process::Finalized; - }; - - _trace_stages.start = [](const constraint::spec&) { - enable_categories(config::get_enabled_categories()); - return state::process::get() < state::process::Finalized; - }; - - _trace_stages.stop = [](const constraint::spec&) { - // only disable categories if not finalized since this might run in background - // during finalization and disable output of data in those categories - if(state::process::get() < state::process::Finalized) - disable_categories(config::get_enabled_categories()); - return state::process::get() < state::process::Finalized; - }; - - auto _promise = std::promise(); - std::thread{ [_trace_specs, _trace_stages](std::promise* _prom) { - // ensure all categories are disabled before proceeding - // if a delay is requested - if(_trace_specs.front().delay > 1.0e-3) - disable_categories(config::get_enabled_categories()); - _prom->set_value(); - for(const auto& itr : _trace_specs) - itr(_trace_stages); - }, - &_promise } - .detach(); - - _promise.get_future().wait_for(std::chrono::seconds{ 1 }); - } } void diff --git a/projects/rocprofiler-systems/source/lib/core/config.cpp b/projects/rocprofiler-systems/source/lib/core/config.cpp index 9fe207ff0b75..5e24d6c93044 100644 --- a/projects/rocprofiler-systems/source/lib/core/config.cpp +++ b/projects/rocprofiler-systems/source/lib/core/config.cpp @@ -11,7 +11,6 @@ #include "common/environment.hpp" #include "common/path.hpp" #include "common/static_object.hpp" -#include "constraint.hpp" #include "gpu.hpp" #include "logger/logger.hpp" #include "mproc.hpp" @@ -399,8 +398,8 @@ install_strict_config_value_callbacks(const std::shared_ptr& _config) // for ENV_NAME -- std::string{} can be constructed from either. #define ROCPROFSYS_CONFIG_SETTING(TYPE, ENV_NAME, DESCRIPTION, INITIAL_VALUE, ...) \ [&]() { \ - auto _env_name = std::string{ ENV_NAME }; \ - auto _ret = _config->insert( \ + auto _env_name = std::string{ ENV_NAME }; \ + const auto _ret = _config->insert( \ _env_name, get_setting_name(_env_name), DESCRIPTION, TYPE{ INITIAL_VALUE }, \ std::set{ "custom", "rocprofsys", "librocprof-sys", \ __VA_ARGS__ }); \ @@ -839,29 +838,26 @@ configure_settings(bool _init) "roctxRangeStart/roctxRangeStop markers.", std::string{}, "trace", "profile", "perfetto", "rocpd", "timemory", "rocm"); - auto _clock_choices = std::vector{}; - for(const auto& itr : constraint::get_valid_clock_ids()) - { - _clock_choices.emplace_back( - fmt::format("({}|{}|{})", itr.name, itr.value, itr.raw_name)); - } - - ROCPROFSYS_CONFIG_SETTING(std::string, env_vars::TRACE_PERIODS, - "Similar to specify trace delay and/or duration except in " - "the form :, ::, " - "and/or :::", - std::string{}, "trace", "profile", "perfetto", "timemory"); + ROCPROFSYS_CONFIG_SETTING( + std::string, env_vars::TRACE_PERIODS, + "Space- or semicolon-separated list of trace windows. Each entry has the form " + ": or ::. The clock used for all " + "entries is controlled by ROCPROFSYS_TRACE_PERIOD_CLOCK_ID.", + std::string{}, "trace", "profile", "perfetto", "timemory"); ROCPROFSYS_CONFIG_SETTING( std::string, env_vars::TRACE_PERIOD_CLOCK_ID, - "Set the default clock ID for ROCPROFSYS_TRACE_DELAY, ROCPROFSYS_TRACE_DURATION, " - "and/or ROCPROFSYS_TRACE_PERIODS. E.g. \"realtime\" == the delay/duration is " - "governed by the elapsed realtime, \"cputime\" == the delay/duration is governed " - "by the elapsed CPU-time within the process, etc. Note: when using CPU-based " - "timing, it is recommened to scale the value by the number of threads and be " - "aware that rocprof-sys may contribute to advancing the process CPU-time", - "CLOCK_REALTIME", "trace", "profile", "perfetto", "timemory") - ->set_choices(_clock_choices); + "Clock used to measure delay and duration for ROCPROFSYS_TRACE_DELAY, " + "ROCPROFSYS_TRACE_DURATION, and ROCPROFSYS_TRACE_PERIODS.\n" + " realtime (default) : wall-clock time. A 10-second delay waits for\n" + " 10 real seconds regardless of how many CPU cores are active.\n" + " cputime : process CPU time (CLOCK_PROCESS_CPUTIME_ID). A 10-second\n" + " delay waits until the process has consumed 10 CPU-seconds in total\n" + " across all threads. Useful when you want to skip a fixed amount of\n" + " computation rather than a fixed elapsed time. Note: rocprof-sys\n" + " instrumentation itself contributes to the CPU-time counter.", + "realtime", "trace", "profile", "perfetto", "timemory") + ->set_choices({ "realtime", "cputime" }); ROCPROFSYS_CONFIG_SETTING( double, env_vars::SAMPLING_FREQ, diff --git a/projects/rocprofiler-systems/source/lib/core/config.hpp b/projects/rocprofiler-systems/source/lib/core/config.hpp index b13f1b880042..69832c698c70 100644 --- a/projects/rocprofiler-systems/source/lib/core/config.hpp +++ b/projects/rocprofiler-systems/source/lib/core/config.hpp @@ -274,12 +274,6 @@ get_perfetto_backend(); std::string get_perfetto_output_filename(); -double -get_trace_delay(); - -double -get_trace_duration(); - std::string get_trace_region(); diff --git a/projects/rocprofiler-systems/source/lib/core/config/CMakeLists.txt b/projects/rocprofiler-systems/source/lib/core/config/CMakeLists.txt new file mode 100644 index 000000000000..c3f7b6c4ef56 --- /dev/null +++ b/projects/rocprofiler-systems/source/lib/core/config/CMakeLists.txt @@ -0,0 +1,10 @@ +# Copyright (c) Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +set(config_headers ${CMAKE_CURRENT_LIST_DIR}/trace_period_config.hpp) + +target_sources(rocprofiler-systems-core-library PRIVATE ${config_headers}) + +if(ROCPROFSYS_BUILD_TESTING) + add_subdirectory(tests) +endif() diff --git a/projects/rocprofiler-systems/source/lib/core/config/tests/CMakeLists.txt b/projects/rocprofiler-systems/source/lib/core/config/tests/CMakeLists.txt new file mode 100644 index 000000000000..2e0ebfb81475 --- /dev/null +++ b/projects/rocprofiler-systems/source/lib/core/config/tests/CMakeLists.txt @@ -0,0 +1,19 @@ +# Copyright (c) Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +add_library(trace-period-config-tests OBJECT test_trace_period_config.cpp) + +target_link_libraries( + trace-period-config-tests + PRIVATE + rocprofiler-systems-googletest-library + rocprofiler-systems-core-library + rocprofiler-systems-logger + rocprofiler-systems-headers + rocprofiler-systems-timemory +) + +target_include_directories( + trace-period-config-tests + PRIVATE ${PROJECT_SOURCE_DIR}/source/lib ${CMAKE_BINARY_DIR}/source/lib/ +) diff --git a/projects/rocprofiler-systems/source/lib/core/config/tests/test_trace_period_config.cpp b/projects/rocprofiler-systems/source/lib/core/config/tests/test_trace_period_config.cpp new file mode 100644 index 000000000000..0df6eb57a029 --- /dev/null +++ b/projects/rocprofiler-systems/source/lib/core/config/tests/test_trace_period_config.cpp @@ -0,0 +1,180 @@ +// Copyright (c) Advanced Micro Devices, Inc. +// SPDX-License-Identifier: MIT + +#include "core/config/trace_period_config.hpp" + +#include + +#include +#include + +using rocprofsys::config::trace_config; +using rocprofsys::config::trace_period_settings; + +namespace +{ +constexpr double k_sample_duration_seconds = 2.0; +constexpr double k_sample_delay_only_seconds = 3.0; + +// Function-local static so initialization happens on first use rather than +// before main - avoids static initialization order concerns for a +// namespace-scope trace_period_settings. +trace_period_settings& +mock_settings() +{ + static trace_period_settings s_settings{}; + return s_settings; +} + +enum class externals_tag : int +{ + nothing_configured = 1, + multiple_entries = 2, + empty_periods = 3, + delay_only = 4, + clock_realtime = 5, + clock_cputime = 6, + clock_default = 7, + clock_invalid = 8, + both_delay_and_periods = 9, + negative_delay = 10, + negative_duration = 11, + period_entry_with_extra_field = 12, +}; + +// Distinct Tag values give each test its own instantiation, so no static state +// inside trace_config can leak between tests. +template +struct mock_externals +{ + static trace_period_settings get_trace_period_settings() { return mock_settings(); } +}; + +template +using config_for = trace_config>; + +class trace_period_config_test_interface : public ::testing::Test +{ +protected: + void SetUp() override { mock_settings() = trace_period_settings{}; } + void TearDown() override { mock_settings() = trace_period_settings{}; } +}; +} // namespace + +TEST_F(trace_period_config_test_interface, + get_trace_specs_without_configuration_yields_no_periods) +{ + EXPECT_TRUE(config_for::get_trace_specs().empty()); +} + +TEST_F(trace_period_config_test_interface, get_trace_specs_splits_multiple_period_entries) +{ + mock_settings().periods = "1:2 3:4;5:6"; + + const auto parsed = config_for::get_trace_specs(); + + ASSERT_EQ(parsed.size(), 3u); + EXPECT_DOUBLE_EQ(parsed[0].delay, 1.0); + EXPECT_DOUBLE_EQ(parsed[0].duration, 2.0); + EXPECT_DOUBLE_EQ(parsed[1].delay, 3.0); + EXPECT_DOUBLE_EQ(parsed[1].duration, 4.0); + EXPECT_DOUBLE_EQ(parsed[2].delay, 5.0); + EXPECT_DOUBLE_EQ(parsed[2].duration, 6.0); +} + +TEST_F(trace_period_config_test_interface, + get_trace_specs_empty_periods_yields_only_delay_duration_window) +{ + mock_settings().delay = 1.0; + mock_settings().duration = k_sample_duration_seconds; + + const auto parsed = config_for::get_trace_specs(); + + ASSERT_EQ(parsed.size(), 1u); + EXPECT_DOUBLE_EQ(parsed[0].delay, 1.0); + EXPECT_DOUBLE_EQ(parsed[0].duration, k_sample_duration_seconds); + EXPECT_EQ(parsed[0].repeat, 1u); +} + +TEST_F(trace_period_config_test_interface, + get_trace_specs_emits_window_when_only_delay_is_set) +{ + mock_settings().delay = k_sample_delay_only_seconds; + + const auto parsed = config_for::get_trace_specs(); + + ASSERT_EQ(parsed.size(), 1u); + EXPECT_DOUBLE_EQ(parsed[0].delay, k_sample_delay_only_seconds); + EXPECT_DOUBLE_EQ(parsed[0].duration, 0.0); +} + +TEST_F(trace_period_config_test_interface, + get_trace_specs_throws_when_delay_duration_and_periods_are_both_set) +{ + mock_settings().delay = 1.0; + mock_settings().duration = k_sample_duration_seconds; + mock_settings().periods = "5"; + + EXPECT_THROW( + (void) config_for::get_trace_specs(), + std::runtime_error); +} + +TEST_F(trace_period_config_test_interface, get_trace_specs_throws_on_negative_delay) +{ + mock_settings().delay = -1.0; + + EXPECT_THROW((void) config_for::get_trace_specs(), + std::runtime_error); +} + +TEST_F(trace_period_config_test_interface, get_trace_specs_throws_on_negative_duration) +{ + mock_settings().duration = -1.0; + + EXPECT_THROW((void) config_for::get_trace_specs(), + std::runtime_error); +} + +TEST_F(trace_period_config_test_interface, + get_trace_specs_ignores_period_fields_past_the_third) +{ + mock_settings().periods = "5:10:3:99"; + + const auto parsed = + config_for::get_trace_specs(); + + ASSERT_EQ(parsed.size(), 1u); + EXPECT_DOUBLE_EQ(parsed[0].delay, 5.0); + EXPECT_DOUBLE_EQ(parsed[0].duration, 10.0); + EXPECT_EQ(parsed[0].repeat, 3u); +} + +TEST_F(trace_period_config_test_interface, get_trace_period_clock_id_maps_realtime) +{ + mock_settings().period_clock = "realtime"; + EXPECT_EQ(config_for::get_trace_period_clock_id(), + CLOCK_REALTIME); // NOLINT(misc-include-cleaner) +} + +TEST_F(trace_period_config_test_interface, get_trace_period_clock_id_maps_cputime) +{ + mock_settings().period_clock = "cputime"; + EXPECT_EQ(config_for::get_trace_period_clock_id(), + CLOCK_PROCESS_CPUTIME_ID); // NOLINT(misc-include-cleaner) +} + +TEST_F(trace_period_config_test_interface, get_trace_period_clock_id_defaults_to_realtime) +{ + EXPECT_EQ(config_for::get_trace_period_clock_id(), + CLOCK_REALTIME); +} + +TEST_F(trace_period_config_test_interface, + get_trace_period_clock_id_throws_on_unrecognized_value) +{ + mock_settings().period_clock = "bogus"; + EXPECT_THROW( + (void) config_for::get_trace_period_clock_id(), + std::runtime_error); +} diff --git a/projects/rocprofiler-systems/source/lib/core/config/trace_period_config.hpp b/projects/rocprofiler-systems/source/lib/core/config/trace_period_config.hpp new file mode 100644 index 000000000000..80878b2d0fa1 --- /dev/null +++ b/projects/rocprofiler-systems/source/lib/core/config/trace_period_config.hpp @@ -0,0 +1,219 @@ +// Copyright (c) Advanced Micro Devices, Inc. +// SPDX-License-Identifier: MIT + +#pragma once + +/// @file +/// Constrains data collection to configurable time windows. +/// Each window is defined by a delay before collection begins and a +/// duration for how long it runs. `trace_period::repeat` is parsed from +/// ROCPROFSYS_TRACE_PERIODS but not yet consumed - only the first +/// configured window is currently wired into a running time_window trigger; +/// repeating/multi-window scheduling is not yet implemented. +/// +/// ROCPROFSYS_TRACE_DELAY/ROCPROFSYS_TRACE_DURATION and ROCPROFSYS_TRACE_PERIODS +/// are mutually exclusive: get_trace_specs() throws if both are configured, and +/// also throws if either delay or duration is negative. +/// +/// The clock governing all windows is set once via +/// ROCPROFSYS_TRACE_PERIOD_CLOCK_ID: +/// "realtime" (default) — wall-clock time (std::chrono::steady_clock) +/// "cputime" — process CPU time (CLOCK_PROCESS_CPUTIME_ID) +/// +/// Configuration is read through an Externals policy - a single +/// get_trace_period_settings() call - so the translation from raw settings to +/// trace_periods can be exercised without the global config singleton; +/// default_trace_config_externals below is the production policy. + +#include "common/delimit.hpp" +#include "common/env_vars.hpp" +#include "core/config.hpp" +#include "core/utility.hpp" +#include "logger/debug.hpp" + +#include // NOLINT(misc-include-cleaner) + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace rocprofsys::inline config +{ +struct trace_period +{ + double delay = 0.0; + double duration = 0.0; + std::uint64_t repeat = 1; +}; + +/// Raw trace-window settings as read from configuration, before parsing +/// ROCPROFSYS_TRACE_PERIODS into trace_period entries. +struct trace_period_settings +{ + double delay = 0.0; + double duration = 0.0; + std::string periods; + std::string period_clock = "realtime"; +}; + +template +concept trace_config_externals = requires { + { + Externals::get_trace_period_settings() + } -> std::convertible_to; +}; + +/// Production Externals policy: reads settings directly. Tests inject a mock instead. +struct default_trace_config_externals +{ + static trace_period_settings get_trace_period_settings() + { + return trace_period_settings{ + .delay = get_setting_value(std::string{ env_vars::TRACE_DELAY }) + .value_or(0.0), + .duration = get_setting_value(std::string{ env_vars::TRACE_DURATION }) + .value_or(0.0), + .periods = + get_setting_value(std::string{ env_vars::TRACE_PERIODS }) + .value_or(std::string{}), + .period_clock = get_setting_value( + std::string{ env_vars::TRACE_PERIOD_CLOCK_ID }) + .value_or(std::string{ "realtime" }) + }; + } +}; + +/// Translates the trace-window settings supplied by @p Externals into the +/// value types the control plane consumes. +template + requires trace_config_externals +class trace_config +{ +public: + [[nodiscard]] static std::vector get_trace_specs(); + + // NOLINTNEXTLINE(misc-include-cleaner) + [[nodiscard]] static clockid_t get_trace_period_clock_id(); + +private: + [[nodiscard]] static trace_period parse_period_entry(std::string_view entry); + + [[nodiscard]] static std::vector parse_periods( + std::string_view periods); +}; + +template + requires trace_config_externals +trace_period +trace_config::parse_period_entry(std::string_view entry) +{ + const auto fields = rocprofsys::delimit(std::string{ entry }, ":"); + trace_period period{}; + + if(fields.size() > 3) + { + LOG_WARNING("{} entry '{}' has {} colon-separated fields; only the first 3 " + "(delay:duration:repeat) are used, the rest are ignored", + env_vars::TRACE_PERIODS, entry, fields.size()); + } + + switch(std::min(fields.size(), 3)) + { + case 3: + period.repeat = utility::convert(fields[2]); + [[fallthrough]]; + + case 2: period.duration = utility::convert(fields[1]); [[fallthrough]]; + + case 1: period.delay = utility::convert(fields[0]); [[fallthrough]]; + + case 0: + default: break; + } + return period; +} + +template + requires trace_config_externals +std::vector +trace_config::parse_periods(std::string_view periods) +{ + const auto entries = rocprofsys::delimit(std::string{ periods }, " ;\t\n"); + + auto parsed_periods = std::vector{}; + parsed_periods.reserve(entries.size()); + std::transform(entries.begin(), entries.end(), std::back_inserter(parsed_periods), + [](const auto& entry) { return parse_period_entry(entry); }); + return parsed_periods; +} + +template + requires trace_config_externals +std::vector +trace_config::get_trace_specs() +{ + const auto trace_control_settings = Externals::get_trace_period_settings(); + + if(trace_control_settings.delay < 0.0 || trace_control_settings.duration < 0.0) + { + throw std::runtime_error( + fmt::format("{}/{} must not be negative (delay={}, duration={})", + env_vars::TRACE_DELAY, env_vars::TRACE_DURATION, + trace_control_settings.delay, trace_control_settings.duration)); + } + + const bool has_delay_duration = + trace_control_settings.delay > 0.0 || trace_control_settings.duration > 0.0; + const bool has_periods = !trace_control_settings.periods.empty(); + + if(has_delay_duration && has_periods) + { + // NOLINTNEXTLINE(misc-include-cleaner) + throw std::runtime_error(fmt::format( + "{}/{} and {} are mutually exclusive; configure one or the other, not both", + env_vars::TRACE_DELAY, env_vars::TRACE_DURATION, env_vars::TRACE_PERIODS)); + } + + if(has_periods) + { + return parse_periods(trace_control_settings.periods); + } + if(has_delay_duration) + { + return { trace_period{ trace_control_settings.delay, + trace_control_settings.duration, 1 } }; + } + return {}; +} + +template + requires trace_config_externals +clockid_t +trace_config::get_trace_period_clock_id() +{ + const auto clock_id_str = Externals::get_trace_period_settings().period_clock; + + clockid_t result; + if(clock_id_str == "cputime") + { + result = CLOCK_PROCESS_CPUTIME_ID; // NOLINT(misc-include-cleaner) + } + else if(clock_id_str == "realtime") + { + result = CLOCK_REALTIME; // NOLINT(misc-include-cleaner) + } + else + { + throw std::runtime_error( + fmt::format("Unknown {}: '{}'. Valid choices: realtime, cputime", + env_vars::TRACE_PERIOD_CLOCK_ID, clock_id_str)); + } + return result; +} +} // namespace rocprofsys::inline config diff --git a/projects/rocprofiler-systems/source/lib/core/constraint.cpp b/projects/rocprofiler-systems/source/lib/core/constraint.cpp deleted file mode 100644 index 948d13adaf8a..000000000000 --- a/projects/rocprofiler-systems/source/lib/core/constraint.cpp +++ /dev/null @@ -1,372 +0,0 @@ -// Copyright (c) Advanced Micro Devices, Inc. -// SPDX-License-Identifier: MIT - -#include "constraint.hpp" -#include "common/delimit.hpp" -#include "common/env_vars.hpp" -#include "config.hpp" -#include "state.hpp" -#include "utility.hpp" - -#include "logger/debug.hpp" - -#include - -#include -#include -#include -#include -#include -#include -#include -#include - -using namespace std::chrono_literals; - -namespace rocprofsys::constraint -{ -namespace -{ -using clock_type = std::chrono::high_resolution_clock; -using duration_type = std::chrono::duration; - -constexpr auto k_max_poll_interval = 100ms; - -#define ROCPROFSYS_CLOCK_IDENTIFIER(VAL) \ - clock_identifier { #VAL, VAL } - -auto -clock_name(std::string _v) -{ - constexpr auto _clock_prefix = std::string_view{ "clock_" }; - for(auto& itr : _v) - itr = tolower(itr); - auto _pos = _v.find(_clock_prefix); - if(_pos == 0) _v = _v.substr(_pos + _clock_prefix.length()); - if(_v == "process_cputime_id") _v = "cputime"; - return _v; -} - -const std::set& -accepted_clock_ids() -{ - // NOLINTBEGIN(misc-include-cleaner) - static const auto k_instance = - std::set{ ROCPROFSYS_CLOCK_IDENTIFIER(CLOCK_REALTIME), - ROCPROFSYS_CLOCK_IDENTIFIER(CLOCK_MONOTONIC), - ROCPROFSYS_CLOCK_IDENTIFIER(CLOCK_PROCESS_CPUTIME_ID), - ROCPROFSYS_CLOCK_IDENTIFIER(CLOCK_MONOTONIC_RAW), - ROCPROFSYS_CLOCK_IDENTIFIER(CLOCK_REALTIME_COARSE), - ROCPROFSYS_CLOCK_IDENTIFIER(CLOCK_MONOTONIC_COARSE), - ROCPROFSYS_CLOCK_IDENTIFIER(CLOCK_BOOTTIME) }; - // NOLINTEND(misc-include-cleaner) - return k_instance; -} - -template -clock_identifier -find_clock_identifier(const Tp& _v) -{ - const auto& accepted = accepted_clock_ids(); - - const char* _descript = ""; - if constexpr(std::is_integral::value) - { - _descript = "value"; - for(const auto& itr : accepted) - { - if(itr.value == _v) - { - return itr; - } - } - } - else - { - _descript = "name"; - auto _clock_name = clock_name(_v); - for(const auto& itr : accepted) - { - if(itr.name == _clock_name || itr.raw_name == _v || - std::to_string(itr.value) == _v) - { - return itr; - } - } - } - - auto _choices = std::vector{}; - _choices.reserve(accepted.size()); - for(const auto& itr : accepted) - { - _choices.emplace_back(itr.as_string()); - } - - throw std::runtime_error(fmt::format("Unknown clock id {}: {}. Valid choices: {}", - _descript, _v, fmt::join(_choices, ""))); -} - -timespec -get_timespec(clockid_t clock_id) noexcept -{ - struct timespec _ts; - clock_gettime(clock_id, &_ts); - return _ts; -} - -template -Tp -get_clock_now(clockid_t clock_id) noexcept -{ - constexpr Tp factor = (Precision::den == std::nano::den) - ? 1 - : (Precision::den / static_cast(std::nano::den)); - auto _ts = get_timespec(clock_id); - return (_ts.tv_sec * std::nano::den + _ts.tv_nsec) * factor; -} -} // namespace - -//--------------------------------------------------------------------------------------// -// -// stages implementation -// -//--------------------------------------------------------------------------------------// - -stages::stages() -: init{ [](const spec&) { return state::process::get() < state::process::Finalized; } } -, wait{ [](const spec& _spec) { - std::this_thread::sleep_for(std::min>( - k_max_poll_interval, std::chrono::duration{ _spec.delay })); - return state::process::get() < state::process::Finalized; -} } -, start{ [](const spec&) { return state::process::get() < state::process::Finalized; } } -, collect{ [](const spec& _spec) { - std::this_thread::sleep_for(std::min>( - k_max_poll_interval, std::chrono::duration{ _spec.duration })); - return state::process::get() < state::process::Finalized; -} } -, stop{ [](const spec&) { return state::process::get() < state::process::Finalized; } } -{} - -//--------------------------------------------------------------------------------------// -// -// clock identifier implementation -// -//--------------------------------------------------------------------------------------// - -clock_identifier::clock_identifier(std::string_view _name, int _val) -: value{ _val } -, raw_name{ _name } -, name{ clock_name(std::string{ _name }) } -{} - -bool -clock_identifier::operator<(const clock_identifier& _rhs) const -{ - return value < _rhs.value; -} - -bool -clock_identifier::operator==(const clock_identifier& _rhs) const -{ - return std::tie(raw_name, value) == std::tie(_rhs.raw_name, _rhs.value); -} - -bool -clock_identifier::operator==(int _rhs) const -{ - return (value == _rhs); -} - -bool -clock_identifier::operator==(std::string _rhs) const -{ - return (raw_name == std::string_view{ _rhs }) || - (name == clock_name(std::move(_rhs))); -} - -std::string -clock_identifier::as_string() const -{ - auto _name = name; - for(auto& itr : _name) - itr = tolower(itr); - auto _ss = std::stringstream{}; - _ss << _name << "(id=" << raw_name << ", value=" << value << ")"; - return _ss.str(); -} - -//--------------------------------------------------------------------------------------// -// -// spec implementation -// -//--------------------------------------------------------------------------------------// - -spec::spec(clock_identifier _id, double _delay, double _dur, std::uint64_t _n, - std::uint64_t _rep) -: delay{ _delay } -, duration{ _dur } -, count{ _n } -, repeat{ _rep } -, clock_id{ std::move(_id) } -{} - -spec::spec(int _clock_id, double _delay, double _dur, std::uint64_t _n, - std::uint64_t _rep) -: delay{ _delay } -, duration{ _dur } -, count{ _n } -, repeat{ _rep } -, clock_id{ find_clock_identifier(_clock_id) } -{} - -spec::spec(const std::string& _clock_id, double _delay, double _dur, std::uint64_t _n, - std::uint64_t _rep) -: delay{ _delay } -, duration{ _dur } -, count{ _n } -, repeat{ _rep } -, clock_id{ find_clock_identifier(_clock_id) } -{} - -spec::spec(const std::string& _line) -: spec{ - config::get_setting_value(std::string{ env_vars::TRACE_PERIOD_CLOCK_ID }) - .value_or("CLOCK_REALTIME"), - config::get_setting_value(std::string{ env_vars::TRACE_DELAY }).value_or(0.0), - config::get_setting_value(std::string{ env_vars::TRACE_DURATION }) - .value_or(0.0) -} -{ - auto _delim = rocprofsys::delimit(_line, ":"); - if(!_delim.empty()) delay = utility::convert(_delim.at(0)); - if(_delim.size() > 1) duration = utility::convert(_delim.at(1)); - if(_delim.size() > 2) repeat = utility::convert(_delim.at(2)); - if(_delim.size() > 3) clock_id = find_clock_identifier(_delim.at(3)); -} - -void -spec::operator()(const stages& _stages) const -{ - auto _n = repeat; - if(_n < 1) _n = std::numeric_limits::max(); - - while(state::process::get() < state::process::Active) - { - std::this_thread::sleep_for(std::chrono::microseconds{ 1 }); - } - - for(std::uint64_t i = 0; i < _n; ++i) - { - auto specifications = spec{ clock_id, delay, duration, i, repeat }; - const auto wait = [specifications](const auto& func, auto dur) { - auto ret = true; - const auto now = get_clock_now(specifications.clock_id.value); - const auto del = std::chrono::duration_cast( - std::chrono::duration{ dur }) - .count(); - const auto end = now + del; - while(get_clock_now(specifications.clock_id.value) < end) - { - ret = func(specifications); - if(!ret) - { - break; - } - } - return ret; - }; - - LOG_DEBUG("Executing constraint spec {} of {} :: delay: {:.3f}, " - "duration: {:.3f}, clock: {}", - i, specifications.repeat, specifications.delay, specifications.duration, - specifications.clock_id.as_string()); - if(_stages.init(specifications) && wait(_stages.wait, specifications.delay) && - _stages.start(specifications) && - wait(_stages.collect, specifications.duration) && _stages.stop(specifications)) - { - } - else - { - break; - } - } -} - -//--------------------------------------------------------------------------------------// -// -// global usage functions -// -//--------------------------------------------------------------------------------------// - -const std::set& -get_valid_clock_ids() -{ - return accepted_clock_ids(); -} - -std::vector -get_trace_specs() -{ - auto _v = std::vector{}; - - { - auto _delay_v = - config::get_setting_value(std::string{ env_vars::TRACE_DELAY }) - .value_or(0.0); - auto _duration_v = - config::get_setting_value(std::string{ env_vars::TRACE_DURATION }) - .value_or(0.0); - auto _clock_v = - find_clock_identifier(config::get_setting_value( - std::string{ env_vars::TRACE_PERIOD_CLOCK_ID }) - .value_or("CLOCK_REALTIME")); - - if(_delay_v > 0.0 || _duration_v > 0.0) - { - _v.emplace_back(_clock_v, _delay_v, _duration_v); - } - } - - { - auto _periods_v = - config::get_setting_value(std::string{ env_vars::TRACE_PERIODS }) - .value_or(""); - if(!_periods_v.empty()) - { - for(auto itr : rocprofsys::delimit(_periods_v, " ;\t\n")) - _v.emplace_back(itr); - } - } - - return _v; -} - -stages -get_trace_stages() -{ - auto _v = stages{}; - - _v.init = [](const spec&) { - return state::process::get() < state::process::Finalized; - }; - _v.wait = [](const spec& _spec) { - std::this_thread::sleep_for(std::min>( - k_max_poll_interval, std::chrono::duration{ _spec.delay })); - return state::process::get() < state::process::Finalized; - }; - _v.start = [](const spec&) { - return state::process::get() < state::process::Finalized; - }; - _v.collect = [](const spec& _spec) { - std::this_thread::sleep_for(std::min>( - k_max_poll_interval, std::chrono::duration{ _spec.duration })); - return state::process::get() < state::process::Finalized; - }; - _v.stop = [](const spec&) { - return state::process::get() < state::process::Finalized; - }; - - return _v; -} -} // namespace rocprofsys::constraint diff --git a/projects/rocprofiler-systems/source/lib/core/constraint.hpp b/projects/rocprofiler-systems/source/lib/core/constraint.hpp deleted file mode 100644 index cf2f7fab2275..000000000000 --- a/projects/rocprofiler-systems/source/lib/core/constraint.hpp +++ /dev/null @@ -1,104 +0,0 @@ -// Copyright (c) Advanced Micro Devices, Inc. -// SPDX-License-Identifier: MIT - -#pragma once - -/// @file -/// This provides generic functionality for constraining data collection within -/// a windows of time. E.g., delay, delay + duration, (delay + duration) * nrepeat -/// -/// @todo Migrate delay/duration for sampling, process sampling, and causal profiling -/// to use this -/// - -#include "common/defines.h" - -#include -#include -#include -#include -#include -#include - -namespace rocprofsys -{ -namespace constraint -{ -struct spec; - -struct stages -{ - using functor_t = std::function; - - stages(); - - stages(const stages&) = default; - stages(stages&&) noexcept = default; - stages& operator=(const stages&) = default; - stages& operator=(stages&&) noexcept = default; - - functor_t init = [](const spec&) { return true; }; - functor_t wait = [](const spec&) { return true; }; - functor_t start = [](const spec&) { return true; }; - functor_t collect = [](const spec&) { return true; }; - functor_t stop = [](const spec&) { return true; }; -}; - -struct clock_identifier -{ - int value = -1; - std::string_view raw_name = {}; - std::string name = {}; - - clock_identifier(); - clock_identifier(std::string_view, int); - - clock_identifier(const clock_identifier&) = default; - clock_identifier(clock_identifier&&) noexcept = default; - clock_identifier& operator=(const clock_identifier&) = default; - clock_identifier& operator=(clock_identifier&&) noexcept = default; - - std::string as_string() const; - - bool operator<(const clock_identifier& _rhs) const; - bool operator==(const clock_identifier& _rhs) const; - bool operator==(int _rhs) const; - bool operator==(std::string _rhs) const; - - friend std::ostream& operator<<(std::ostream& _os, const clock_identifier& _v) - { - return (_os << _v.as_string()); - } -}; - -struct spec -{ - spec(int, double, double, std::uint64_t = 0, std::uint64_t = 1); - spec(clock_identifier, double, double, std::uint64_t = 0, std::uint64_t = 1); - spec(const std::string&, double, double, std::uint64_t = 0, std::uint64_t = 1); - spec(const std::string&); - - spec(const spec&) = default; - spec(spec&&) noexcept = default; - spec& operator=(const spec&) = default; - spec& operator=(spec&&) noexcept = default; - - void operator()(const stages&) const; - - double delay = 0.0; - double duration = 0.0; - std::uint64_t count = 0; - std::uint64_t repeat = 1; - clock_identifier clock_id = {}; -}; - -const std::set& -get_valid_clock_ids(); - -std::vector -get_trace_specs(); - -stages -get_trace_stages(); -} // namespace constraint -} // namespace rocprofsys diff --git a/projects/rocprofiler-systems/source/lib/core/control/CMakeLists.txt b/projects/rocprofiler-systems/source/lib/core/control/CMakeLists.txt index 05611b9312e4..42976df94231 100644 --- a/projects/rocprofiler-systems/source/lib/core/control/CMakeLists.txt +++ b/projects/rocprofiler-systems/source/lib/core/control/CMakeLists.txt @@ -9,6 +9,7 @@ set(control_sources set(control_headers ${CMAKE_CURRENT_LIST_DIR}/clock.hpp + ${CMAKE_CURRENT_LIST_DIR}/clocks/posix.hpp ${CMAKE_CURRENT_LIST_DIR}/clocks/steady.hpp ${CMAKE_CURRENT_LIST_DIR}/session.hpp ${CMAKE_CURRENT_LIST_DIR}/triggers/roctx.hpp @@ -19,3 +20,7 @@ target_sources( rocprofiler-systems-core-library PRIVATE ${control_sources} ${control_headers} ) + +if(ROCPROFSYS_BUILD_TESTING) + add_subdirectory(tests) +endif() diff --git a/projects/rocprofiler-systems/source/lib/core/control/clocks/posix.hpp b/projects/rocprofiler-systems/source/lib/core/control/clocks/posix.hpp new file mode 100644 index 000000000000..c7fef5f17c7a --- /dev/null +++ b/projects/rocprofiler-systems/source/lib/core/control/clocks/posix.hpp @@ -0,0 +1,138 @@ +// Copyright (c) Advanced Micro Devices, Inc. +// SPDX-License-Identifier: MIT + +#pragma once + +#include "core/control/clock.hpp" +#include "core/state.hpp" + +#include "logger/debug.hpp" + +#include +#include +#include +#include +#include +#include +#include + +namespace rocprofsys::control::clocks +{ +/// Clock backed by a POSIX clock (clock_gettime / clock_nanosleep). +class posix +{ +public: + explicit posix(clockid_t clock_id = CLOCK_REALTIME) noexcept // NOLINT + : m_clock_id{ clock_id } + { + struct timespec specs = {}; + if(clock_gettime(m_clock_id, &specs) != 0) + { + m_usable = false; + LOG_WARNING("unusable clock id {}: {}", m_clock_id, + std::system_category().message(errno)); + } + } + + ~posix() = default; + + posix(const posix&) = delete; + posix& operator=(const posix&) = delete; + posix(posix&&) = delete; + posix& operator=(posix&&) = delete; + + [[nodiscard]] clock_time_point now() const noexcept + { + struct timespec specs = {}; + if(clock_gettime(m_clock_id, &specs) != 0) + { + LOG_WARNING("clock_gettime failed for clock id {}: {}", m_clock_id, + std::system_category().message(errno)); + } + return clock_time_point{ clock_duration{ to_nanoseconds(specs) } }; + } + + [[nodiscard]] bool sleep_until(clock_time_point deadline) + { + if(!m_usable) + { + return false; + } + + for(auto current = now(); current < deadline; current = now()) + { + if(is_interrupted()) + { + return false; + } + + const auto remaining_ns = (deadline - current).count(); + if(remaining_ns <= 0) + { + break; + } + + sleep_one_chunk(current, remaining_ns); + } + + return !is_interrupted(); + } + + void interrupt() + { + const auto thread_state_guard = state::thread::scoped(state::thread::Internal); + const std::scoped_lock notify_lk{ m_mutex }; + m_interrupted = true; + } + + void reset() + { + const auto thread_state_guard = state::thread::scoped(state::thread::Internal); + const std::scoped_lock notify_lk{ m_mutex }; + m_interrupted = false; + } + +private: + static constexpr std::int64_t k_nsec_per_sec = 1'000'000'000; + static constexpr std::int64_t k_chunk_ns = 1'000'000; // 1 ms + + [[nodiscard]] bool is_interrupted() + { + const std::scoped_lock notify_lk{ m_mutex }; + return m_interrupted; + } + + void sleep_one_chunk(clock_time_point current, std::int64_t remaining_ns) + { + const auto this_chunk_ns = std::min(remaining_ns, k_chunk_ns); + const auto next_ns = current.time_since_epoch().count() + this_chunk_ns; + const struct timespec specs = to_timespec(next_ns); + if(const auto sleep_status = + clock_nanosleep(m_clock_id, TIMER_ABSTIME, &specs, nullptr); // NOLINT + sleep_status != 0 && sleep_status != EINTR) + { + LOG_WARNING("clock_nanosleep failed for clock id {}: {}", m_clock_id, + std::system_category().message(sleep_status)); + } + } + + [[nodiscard]] static std::int64_t to_nanoseconds( + const struct timespec& specs) noexcept + { + return (static_cast(specs.tv_sec) * k_nsec_per_sec) + specs.tv_nsec; + } + + [[nodiscard]] static struct timespec to_timespec(std::int64_t nanoseconds) noexcept + { + struct timespec specs = {}; + specs.tv_sec = static_cast(nanoseconds / k_nsec_per_sec); + specs.tv_nsec = static_cast(nanoseconds % k_nsec_per_sec); + return specs; + } + + clockid_t m_clock_id; + bool m_usable = true; + std::mutex m_mutex; + bool m_interrupted{ false }; +}; +} // namespace rocprofsys::control::clocks diff --git a/projects/rocprofiler-systems/source/lib/core/control/clocks/steady.hpp b/projects/rocprofiler-systems/source/lib/core/control/clocks/steady.hpp index 7635fdf0beb4..176ba98e033f 100644 --- a/projects/rocprofiler-systems/source/lib/core/control/clocks/steady.hpp +++ b/projects/rocprofiler-systems/source/lib/core/control/clocks/steady.hpp @@ -10,11 +10,6 @@ namespace rocprofsys::control::clocks { -/// Production clock backed by std::chrono::steady_clock. sleep_until uses -/// std::condition_variable::wait_until so interrupt() can wake any waiter -/// (used during shutdown to cancel a sleeping time_window trigger). -/// -/// Satisfies the clock concept (see core/control/clock.hpp). class steady { public: diff --git a/projects/rocprofiler-systems/source/lib/core/control/session.cpp b/projects/rocprofiler-systems/source/lib/core/control/session.cpp index 918cfa969a6c..70081180fde5 100644 --- a/projects/rocprofiler-systems/source/lib/core/control/session.cpp +++ b/projects/rocprofiler-systems/source/lib/core/control/session.cpp @@ -8,6 +8,7 @@ #include #include +#include #include #include #include @@ -16,6 +17,14 @@ namespace rocprofsys::control { +session::session() noexcept +{ + for(auto& tracing_flag : m_scope_tracing) + { + tracing_flag.store(true, std::memory_order_relaxed); + } +} + void session::shutdown() { @@ -26,8 +35,14 @@ session::shutdown() } { const std::scoped_lock action_lock{ m_actions_mutex }; - m_actions.clear(); - m_active.store(true, std::memory_order_relaxed); + for(auto& scoped : m_actions) + { + scoped.clear(); + } + for(auto& tracing_flag : m_scope_tracing) + { + tracing_flag.store(true, std::memory_order_relaxed); + } } } @@ -40,39 +55,58 @@ session::subscribe(subscriber sub) } void -session::register_trigger(std::string_view name, action initial) +session::register_trigger(std::string_view name, action initial, scope event_scope) { - apply_locked_transition([&] { m_actions[std::string{ name }] = initial; }, name); + apply_locked_transition( + [&] { + m_actions[static_cast(event_scope)][std::string{ name }] = + initial; + }, + name, event_scope); } void -session::unregister_trigger(std::string_view name) +session::unregister_trigger(std::string_view name, scope event_scope) { - apply_locked_transition([&] { m_actions.erase(std::string{ name }); }, name); + apply_locked_transition( + [&] { + m_actions[static_cast(event_scope)].erase(std::string{ name }); + }, + name, event_scope); } void -session::set_action(std::string_view name, action act) +session::set_action(std::string_view name, action act, scope event_scope) { - apply_locked_transition([&] { m_actions[std::string{ name }] = act; }, name); + apply_locked_transition( + [&] { + m_actions[static_cast(event_scope)][std::string{ name }] = act; + }, + name, event_scope); } void session::apply_locked_transition(const std::function& mutate, - std::string_view name) + std::string_view name, scope event_scope) { const auto thread_state_guard = state::thread::scoped(state::thread::Internal); + // Serializes compute-then-notify across concurrent callers so subscribers + // observe transitions in the same order they were computed. Deliberately + // a separate mutex from m_actions_mutex (released below, before + // notify_pause()/notify_resume() run) so a subscriber callback that + // re-enters is_active()/is_active_without() cannot deadlock. const std::scoped_lock notify_lk{ m_notify_mutex }; - bool was_active = false; - bool now_active = false; + const auto scope_idx = static_cast(event_scope); + bool was_active = false; + bool now_active = false; { const std::scoped_lock action_lk{ m_actions_mutex }; - was_active = m_active.load(std::memory_order_relaxed); + was_active = m_scope_tracing[scope_idx].load(std::memory_order_relaxed); mutate(); - now_active = resolve_locked(); - m_active.store(now_active, std::memory_order_relaxed); + now_active = resolve_locked(event_scope); + m_scope_tracing[scope_idx].store(now_active, std::memory_order_relaxed); } if(was_active == now_active) @@ -80,35 +114,59 @@ session::apply_locked_transition(const std::function& mutate, return; } - LOG_DEBUG("session: trigger '{}' {} the session", name, + LOG_DEBUG("session: trigger '{}' {} its scope", name, now_active ? "resumed" : "paused"); if(now_active) { - notify_resume(); + notify_resume(event_scope); } else { - notify_pause(); + notify_pause(event_scope); } } -// Any pause action pauses the session. Skip is ignored. -// With no actions the session is active by default. +// Any pause action within the given scope pauses that scope. Skip is +// ignored. With no actions for the scope, the scope is active by default. bool -session::resolve_locked() const noexcept +session::resolve_locked(scope event_scope) const noexcept { + const auto& scoped = m_actions[static_cast(event_scope)]; return std::ranges::none_of( - m_actions, [](const auto& entry) { return entry.second == action::pause; }); + scoped, [](const auto& entry) { return entry.second == action::pause; }); +} + +bool +session::is_active_without(std::string_view name, scope event_scope) const noexcept +{ + const std::scoped_lock actions_lk{ m_actions_mutex }; + const auto& scoped = m_actions[static_cast(event_scope)]; + return std::ranges::none_of(scoped, [name](const auto& entry) { + return entry.first != name && entry.second == action::pause; + }); +} + +namespace +{ +bool +listens_to(const subscriber& sub, scope event_scope) +{ + return sub.scopes.contains(event_scope); } +} // namespace void -session::notify_pause() +session::notify_pause(scope event_scope) { const auto thread_state_guard = state::thread::scoped(state::thread::Internal); const std::scoped_lock notify_lk{ m_subscribers_mutex }; for(const auto& sub : m_subscribers) { + if(!listens_to(sub, event_scope)) + { + continue; + } LOG_DEBUG("session: pausing subscriber '{}'", sub.name); if(sub.on_pause) { @@ -118,12 +176,22 @@ session::notify_pause() } void -session::notify_resume() +session::notify_resume(scope event_scope) { const auto thread_state_guard = state::thread::scoped(state::thread::Internal); const std::scoped_lock notify_lk{ m_subscribers_mutex }; for(const auto& sub : m_subscribers) { + if(!listens_to(sub, event_scope)) + { + continue; + } + const bool all_active = + sub.scopes.all_of([this](scope listened) { return is_active(listened); }); + if(!all_active) + { + continue; + } LOG_DEBUG("session: resuming subscriber '{}'", sub.name); if(sub.on_resume) { diff --git a/projects/rocprofiler-systems/source/lib/core/control/session.hpp b/projects/rocprofiler-systems/source/lib/core/control/session.hpp index 447e12a8d24d..1891bc40002c 100644 --- a/projects/rocprofiler-systems/source/lib/core/control/session.hpp +++ b/projects/rocprofiler-systems/source/lib/core/control/session.hpp @@ -3,8 +3,13 @@ #pragma once +#include #include +#include +#include +#include #include +#include #include #include #include @@ -22,17 +27,76 @@ enum class action pause }; +/// A trigger's blast radius: which subscribers its actions can reach. +enum class scope : std::size_t +{ + global = 0, + sampling, + count, // sentinel: number of scopes +}; + +inline constexpr std::size_t k_scope_count = static_cast(scope::count); + +class scope_set +{ +public: + scope_set() noexcept { m_bits.set(static_cast(scope::global)); } + + scope_set(std::initializer_list scopes) noexcept + { + for(const auto event_scope : scopes) + { + m_bits.set(static_cast(event_scope)); + } + } + + [[nodiscard]] bool contains(scope event_scope) const noexcept + { + return m_bits.test(static_cast(event_scope)); + } + + template + [[nodiscard]] bool all_of(Predicate&& pred) const + { + for(std::size_t i = 0; i < k_scope_count; ++i) + { + if(m_bits.test(i) && !pred(static_cast(i))) + { + return false; + } + } + return true; + } + + template + [[nodiscard]] bool any_of(Predicate&& pred) const + { + for(std::size_t i = 0; i < k_scope_count; ++i) + { + if(m_bits.test(i) && pred(static_cast(i))) + { + return true; + } + } + return false; + } + +private: + std::bitset m_bits; +}; + struct subscriber { std::function on_pause; std::function on_resume; std::string name; + scope_set scopes; }; class session { public: - session() = default; + session() noexcept; ~session() = default; session(const session&) = delete; @@ -42,38 +106,56 @@ class session void shutdown(); + /// Register a subscriber. Callers MUST subscribe every subscriber + /// before constructing any trigger: register_trigger()/unregister_trigger() + /// broadcast on a pause/resume transition at the moment they're called, so + /// a subscriber that joins after a trigger's initial vote already flipped + /// the session will miss that broadcast. void subscribe(subscriber sub); /// Seed a trigger's action. @p name identifies the trigger for the - /// lifetime of its registration. Broadcasts to subscribers if this - /// registration changes the session's active/paused state. - void register_trigger(std::string_view name, action initial); + /// lifetime of its registration; @p event_scope fixes which subscribers + /// its later actions can reach. Broadcasts to subscribers of that scope + /// if this registration changes the scope's active/paused state. + void register_trigger(std::string_view name, action initial, + scope event_scope = scope::global); - void unregister_trigger(std::string_view name); + void unregister_trigger(std::string_view name, scope event_scope = scope::global); - void set_action(std::string_view name, action act); + void set_action(std::string_view name, action act, scope event_scope = scope::global); - [[nodiscard]] bool is_active() const noexcept + [[nodiscard]] bool is_active(scope event_scope = scope::global) const noexcept { - return m_active.load(std::memory_order_relaxed); + assert(static_cast(event_scope) < k_scope_count); + return m_scope_tracing[static_cast(event_scope)].load( + std::memory_order_relaxed); } + /// True iff every trigger of @p event_scope except @p name currently has + /// a trace/skip action. Used where a trigger's own write decision must + /// also respect other triggers' actions without double-counting its own. + [[nodiscard]] bool is_active_without( + std::string_view name, scope event_scope = scope::global) const noexcept; + private: - std::unordered_map m_actions; - std::vector m_subscribers; - std::atomic m_active{ true }; + using scoped_actions = std::unordered_map; + + std::array m_actions; + std::vector m_subscribers; + std::array, k_scope_count> m_scope_tracing{}; mutable std::mutex m_actions_mutex; std::mutex m_subscribers_mutex; std::mutex m_notify_mutex; - [[nodiscard]] bool resolve_locked() const noexcept; - void notify_pause(); - void notify_resume(); + [[nodiscard]] bool resolve_locked(scope event_scope) const noexcept; + void notify_pause(scope event_scope); + void notify_resume(scope event_scope); - /// Applies @p mutate to m_actions under lock, recomputes the active - /// state, and broadcasts to subscribers only if that state changed. + /// Applies @p mutate to the scoped action map under lock, recomputes + /// that scope's active state, and broadcasts if it changed. Shared by + /// register_trigger(), unregister_trigger(), and set_action(). void apply_locked_transition(const std::function& mutate, - std::string_view name); + std::string_view name, scope event_scope); }; } // namespace rocprofsys::control diff --git a/projects/rocprofiler-systems/source/lib/core/control/tests/CMakeLists.txt b/projects/rocprofiler-systems/source/lib/core/control/tests/CMakeLists.txt new file mode 100644 index 000000000000..12ad565e76ef --- /dev/null +++ b/projects/rocprofiler-systems/source/lib/core/control/tests/CMakeLists.txt @@ -0,0 +1,34 @@ +# Copyright (c) Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT + +add_library(control-posix-clock-tests OBJECT test_posix_clock.cpp) + +target_link_libraries( + control-posix-clock-tests + PRIVATE + rocprofiler-systems-googletest-library + rocprofiler-systems-core-library + rocprofiler-systems-common-library + rocprofiler-systems-logger +) + +target_include_directories( + control-posix-clock-tests + PRIVATE ${PROJECT_SOURCE_DIR}/source/lib ${CMAKE_BINARY_DIR}/source/lib +) + +add_library(control-session-scope-tests OBJECT test_session_scope.cpp) + +target_link_libraries( + control-session-scope-tests + PRIVATE + rocprofiler-systems-googletest-library + rocprofiler-systems-core-library + rocprofiler-systems-common-library + rocprofiler-systems-logger +) + +target_include_directories( + control-session-scope-tests + PRIVATE ${PROJECT_SOURCE_DIR}/source/lib ${CMAKE_BINARY_DIR}/source/lib +) diff --git a/projects/rocprofiler-systems/source/lib/core/control/tests/test_posix_clock.cpp b/projects/rocprofiler-systems/source/lib/core/control/tests/test_posix_clock.cpp new file mode 100644 index 000000000000..40f95b0d7192 --- /dev/null +++ b/projects/rocprofiler-systems/source/lib/core/control/tests/test_posix_clock.cpp @@ -0,0 +1,106 @@ +// Copyright (c) Advanced Micro Devices, Inc. +// SPDX-License-Identifier: MIT + +#include "core/control/clocks/posix.hpp" +#include "core/control/session.hpp" + +#include "core/control/clock.hpp" +#include "core/control/triggers/time_window.hpp" +#include + +#include +#include +#include +#include +#include + +namespace +{ +using rocprofsys::control::clock_duration; +using rocprofsys::control::session; +using rocprofsys::control::clocks::posix; +using time_window_t = rocprofsys::control::triggers::time_window; + +// Polls the session's active state until it reaches @p expected or a +// wall-clock timeout elapses. posix uses real time, so events cannot be +// deterministically stepped - poll instead. +bool +wait_until_active(session& sess, bool expected) +{ + constexpr auto k_timeout = std::chrono::seconds{ 2 }; + const auto deadline = std::chrono::steady_clock::now() + k_timeout; + while(sess.is_active() != expected) + { + if(std::chrono::steady_clock::now() > deadline) + { + return false; + } + std::this_thread::sleep_for(std::chrono::milliseconds{ 1 }); + } + return true; +} +} // namespace + +TEST(posix_clock_test, now_returns_advancing_time_points) +{ + using namespace std::chrono_literals; + const posix clk{ CLOCK_REALTIME }; // NOLINT + + const auto earlier = clk.now(); + std::this_thread::sleep_for(5ms); // NOLINT(misc-include-cleaner) + const auto later = clk.now(); + + EXPECT_GT(later, earlier) << "now() should advance over time"; +} + +TEST(posix_clock_test, interrupt_makes_sleep_until_return_false) +{ + using namespace std::chrono_literals; + posix clk{ CLOCK_REALTIME }; + + // Schedule a deadline far in the future; interrupt from another thread. + const auto deadline = clk.now() + clock_duration{ 30'000'000'000LL }; // 30s + + std::atomic result{ true }; + std::thread sleeper{ [&] { result.store(clk.sleep_until(deadline)); } }; + + std::this_thread::sleep_for(10ms); // NOLINT(misc-include-cleaner) + clk.interrupt(); + sleeper.join(); + + EXPECT_FALSE(result.load()) << "sleep_until should return false when interrupted"; +} + +TEST(posix_clock_test, reset_allows_reuse_after_interrupt) +{ + posix clk{ CLOCK_REALTIME }; + clk.interrupt(); + clk.reset(); + + // After reset, a deadline already in the past should return true (no interrupt). + const auto past = clk.now() - clock_duration{ 1 }; + EXPECT_TRUE(clk.sleep_until(past)); +} + +TEST(posix_clock_test, time_window_with_posix_clock_drives_session) +{ + const auto s_ptr = std::make_shared(); + auto& sess = *s_ptr; + + // 20ms/40ms windows polled against a 2s timeout (100x/50x margin) - + // deliberately generous so the test doesn't flake under CI scheduler + // jitter. Don't shrink the timeout to "tighten" this; shrink the + // windows if faster tests are needed instead. + posix clk{ CLOCK_REALTIME }; + constexpr auto k_delay = clock_duration{ 20'000'000 }; // 20 ms + constexpr auto k_dur = clock_duration{ 40'000'000 }; // 40 ms + time_window_t window{ s_ptr, clk, { .delay = k_delay, .duration = k_dur } }; + + EXPECT_FALSE(sess.is_active()) << "initial action should be paused when delay > 0"; + + window.start(); + ASSERT_TRUE(wait_until_active(sess, true)) + << "session should become active once the delay elapses"; + ASSERT_TRUE(wait_until_active(sess, false)) + << "session should pause again once the duration elapses"; +} diff --git a/projects/rocprofiler-systems/source/lib/core/control/tests/test_session_scope.cpp b/projects/rocprofiler-systems/source/lib/core/control/tests/test_session_scope.cpp new file mode 100644 index 000000000000..c2ae24947aed --- /dev/null +++ b/projects/rocprofiler-systems/source/lib/core/control/tests/test_session_scope.cpp @@ -0,0 +1,180 @@ +// Copyright (c) Advanced Micro Devices, Inc. +// SPDX-License-Identifier: MIT + +#include "core/control/session.hpp" + +#include +#include +#include + +namespace +{ +using rocprofsys::control::action; +using rocprofsys::control::scope; +using rocprofsys::control::session; + +// Minimal trigger stub: registers under a fixed name and scope, and exposes +// set_action() so a test can drive a later action change. +class mock_trigger +{ +public: + mock_trigger(session& sess, std::string name, scope event_scope, action initial) + : m_session{ sess } + , m_name{ std::move(name) } + , m_scope{ event_scope } + { + m_session.register_trigger(m_name, initial, m_scope); + } + + ~mock_trigger() { m_session.unregister_trigger(m_name, m_scope); } + + mock_trigger(const mock_trigger&) = delete; + mock_trigger& operator=(const mock_trigger&) = delete; + mock_trigger(mock_trigger&&) = delete; + mock_trigger& operator=(mock_trigger&&) = delete; + + void set_action(action act) const { m_session.set_action(m_name, act, m_scope); } + +private: + session& m_session; + std::string m_name; + scope m_scope; +}; + +class session_scope_test_interface : public ::testing::Test +{ +protected: + session m_ses; +}; +} // namespace + +TEST_F(session_scope_test_interface, + subscriber_not_resumed_while_a_listened_scope_is_still_paused) +{ + // Reproduces the scenario a scoped subscriber (e.g. "sampling", which + // listens to both global and sampling) must not be resumed just + // because ONE of its scopes cleared - it must wait for ALL of them. + const mock_trigger global_trigger{ m_ses, "global_trigger", scope::global, + action::pause }; + const mock_trigger sampling_trigger{ m_ses, "sampling_trigger", scope::sampling, + action::pause }; + + int resume_count = 0; + int pause_count = 0; + m_ses.subscribe({ .on_pause = [&pause_count]() { ++pause_count; }, + .on_resume = [&resume_count]() { ++resume_count; }, + .name = "scoped_sub", + .scopes = { scope::global, scope::sampling } }); + + // Both scopes paused initially - resuming just the global scope must not + // fire the subscriber's on_resume, since sampling is still paused. + global_trigger.set_action(action::trace); + EXPECT_EQ(resume_count, 0); + + // Once the remaining scope (sampling) also clears, the subscriber + // should resume exactly once. + sampling_trigger.set_action(action::trace); + EXPECT_EQ(resume_count, 1); +} + +TEST_F(session_scope_test_interface, + subscriber_paused_immediately_when_any_listened_scope_pauses) +{ + const mock_trigger global_trigger{ m_ses, "global_trigger", scope::global, + action::trace }; + const mock_trigger sampling_trigger{ m_ses, "sampling_trigger", scope::sampling, + action::trace }; + + int pause_count = 0; + m_ses.subscribe({ .on_pause = [&pause_count]() { ++pause_count; }, + .on_resume = []() {}, + .name = "scoped_sub", + .scopes = { scope::global, scope::sampling } }); + + sampling_trigger.set_action(action::pause); + EXPECT_EQ(pause_count, 1) << "any listened scope pausing must pause the subscriber"; +} + +TEST_F(session_scope_test_interface, single_scope_subscriber_unaffected_by_other_scope) +{ + // A subscriber listening only to Scope::global (the default) must be + // unaffected by a Scope::Sampling trigger's transitions. + const mock_trigger sampling_trigger{ m_ses, "sampling_trigger", scope::sampling, + action::trace }; + + int pause_count = 0; + int resume_count = 0; + m_ses.subscribe({ .on_pause = [&pause_count]() { ++pause_count; }, + .on_resume = [&resume_count]() { ++resume_count; }, + .name = "global_only_sub", + .scopes = { scope::global } }); + + sampling_trigger.set_action(action::pause); + sampling_trigger.set_action(action::trace); + + EXPECT_EQ(pause_count, 0); + EXPECT_EQ(resume_count, 0); +} + +TEST_F(session_scope_test_interface, is_active_is_tracked_independently_per_scope) +{ + const mock_trigger global_trigger{ m_ses, "global_trigger", scope::global, + action::trace }; + const mock_trigger sampling_trigger{ m_ses, "sampling_trigger", scope::sampling, + action::trace }; + + sampling_trigger.set_action(action::pause); + + EXPECT_TRUE(m_ses.is_active(scope::global)); + EXPECT_FALSE(m_ses.is_active(scope::sampling)); +} + +TEST_F(session_scope_test_interface, + same_named_triggers_in_different_scopes_do_not_clobber_each_other) +{ + // The global TRACE_DELAY window and the sampling-scoped SAMPLING_DURATION + // window are both time_window instances, so both report the same trigger + // name. Storing them under that name alone lets the second registration + // overwrite the first. + const mock_trigger global_window{ m_ses, "time_window", scope::global, + action::pause }; + const mock_trigger sampling_window{ m_ses, "time_window", scope::sampling, + action::pause }; + + EXPECT_FALSE(m_ses.is_active(scope::global)); + EXPECT_FALSE(m_ses.is_active(scope::sampling)); + + sampling_window.set_action(action::trace); + EXPECT_TRUE(m_ses.is_active(scope::sampling)); + + // Force a fresh resolve of the global scope without touching + // global_window's own setter: if its entry had been overwritten, the + // global scope would now wrongly resolve active. + { + const mock_trigger probe{ m_ses, "probe", scope::global, action::trace }; + } + EXPECT_FALSE(m_ses.is_active(scope::global)) + << "global_window's pause must survive the sampling window's registration"; +} + +TEST_F(session_scope_test_interface, unregister_removes_only_the_matching_scope) +{ + const mock_trigger sampling_window{ m_ses, "time_window", scope::sampling, + action::pause }; + { + const mock_trigger global_window{ m_ses, "time_window", scope::global, + action::pause }; + EXPECT_FALSE(m_ses.is_active(scope::global)); + } + + EXPECT_TRUE(m_ses.is_active(scope::global)) + << "the global window's entry should be gone once it unregisters"; + + // Force a fresh resolve of sampling: a shared entry erased by the + // global window's destructor would leave nothing to keep this scope paused. + { + const mock_trigger probe{ m_ses, "probe", scope::sampling, action::trace }; + } + EXPECT_FALSE(m_ses.is_active(scope::sampling)) + << "unregistering the global window must not remove the same-named sampling one"; +} diff --git a/projects/rocprofiler-systems/source/lib/core/control/triggers/roctx.hpp b/projects/rocprofiler-systems/source/lib/core/control/triggers/roctx.hpp index 3e5421783d26..f9345228bf51 100644 --- a/projects/rocprofiler-systems/source/lib/core/control/triggers/roctx.hpp +++ b/projects/rocprofiler-systems/source/lib/core/control/triggers/roctx.hpp @@ -20,6 +20,8 @@ namespace rocprofsys::control::triggers class roctx { public: + static constexpr std::string_view k_trigger_name = "roctx"; + roctx(std::shared_ptr sess, std::string_view trace_regions); ~roctx(); @@ -51,8 +53,6 @@ class roctx std::mutex m_mutex; std::shared_ptr m_session; - static constexpr std::string_view k_trigger_name = "roctx"; - [[nodiscard]] action compute_action() const noexcept; [[nodiscard]] bool compute_should_write() const noexcept; void refresh_state(); diff --git a/projects/rocprofiler-systems/source/lib/core/control/triggers/time_window.hpp b/projects/rocprofiler-systems/source/lib/core/control/triggers/time_window.hpp index d5748d390769..756d659fed32 100644 --- a/projects/rocprofiler-systems/source/lib/core/control/triggers/time_window.hpp +++ b/projects/rocprofiler-systems/source/lib/core/control/triggers/time_window.hpp @@ -15,24 +15,32 @@ namespace rocprofsys::control::triggers { +/// A time_window's delay/duration schedule. Clock-agnostic - the same specs +/// can be handed to a time_window for any Clock. +struct time_window_specs +{ + clock_duration delay{}; + clock_duration duration{}; +}; + template class time_window { public: - time_window(std::shared_ptr sess, Clock& clk, clock_duration delay, - clock_duration duration) + time_window(std::shared_ptr sess, Clock& clk, time_window_specs specs, + scope event_scope = scope::global) : m_session{ std::move(sess) } , m_clock{ clk } - , m_delay{ delay } - , m_duration{ duration } + , m_specs{ specs } + , m_scope{ event_scope } { - m_session->register_trigger(k_trigger_name, initial_action(delay, duration)); + m_session->register_trigger(k_trigger_name, initial_action(m_specs), m_scope); } ~time_window() { stop(); - m_session->unregister_trigger(k_trigger_name); + m_session->unregister_trigger(k_trigger_name, m_scope); } time_window(const time_window&) = delete; @@ -59,11 +67,6 @@ class time_window m_thread = std::thread{ [this]() { worker(); } }; } - /// Interrupt the clock and join the worker thread. Idempotent. - /// m_thread.join() can only throw if joinable() is false (guarded above) - /// or if called from the worker thread itself, which never happens - - /// stop() is only ever invoked from the owning thread (including via - /// the destructor), never from worker(). void stop() noexcept { const auto thread_state_guard = state::thread::scoped(state::thread::Internal); @@ -81,19 +84,18 @@ class time_window std::shared_ptr m_session; Clock& m_clock; - const clock_duration m_delay; - const clock_duration m_duration; + const time_window_specs m_specs; + const scope m_scope; std::thread m_thread; std::mutex m_lifecycle_mutex; - [[nodiscard]] static action initial_action(clock_duration delay, - clock_duration duration) noexcept + [[nodiscard]] static action initial_action(const time_window_specs& specs) noexcept { - if(delay > clock_duration::zero()) + if(specs.delay > clock_duration::zero()) { return action::pause; } - if(duration > clock_duration::zero()) + if(specs.duration > clock_duration::zero()) { return action::trace; } @@ -102,7 +104,8 @@ class time_window [[nodiscard]] bool has_window() const noexcept { - return m_delay > clock_duration::zero() || m_duration > clock_duration::zero(); + return m_specs.delay > clock_duration::zero() || + m_specs.duration > clock_duration::zero(); } void worker() @@ -110,26 +113,26 @@ class time_window const auto thread_state_guard = state::thread::scoped(state::thread::Internal); const auto current_ts = m_clock.now(); - const bool has_delay = m_delay > clock_duration::zero(); - const bool has_duration = m_duration > clock_duration::zero(); + const bool has_delay = m_specs.delay > clock_duration::zero(); + const bool has_duration = m_specs.duration > clock_duration::zero(); if(has_delay) { - if(!m_clock.sleep_until(current_ts + m_delay)) + if(!m_clock.sleep_until(current_ts + m_specs.delay)) { return; // interrupted } - m_session->set_action(k_trigger_name, action::trace); + m_session->set_action(k_trigger_name, action::trace, m_scope); } if(has_duration) { - const auto end = current_ts + m_delay + m_duration; + const auto end = current_ts + m_specs.delay + m_specs.duration; if(!m_clock.sleep_until(end)) { return; // interrupted } - m_session->set_action(k_trigger_name, action::pause); // terminal + m_session->set_action(k_trigger_name, action::pause, m_scope); // terminal } } }; diff --git a/projects/rocprofiler-systems/source/lib/rocprof-sys/library.cpp b/projects/rocprofiler-systems/source/lib/rocprof-sys/library.cpp index 2d04e58d712f..28a903f60bf4 100644 --- a/projects/rocprofiler-systems/source/lib/rocprof-sys/library.cpp +++ b/projects/rocprofiler-systems/source/lib/rocprof-sys/library.cpp @@ -21,8 +21,11 @@ #include "core/components/fwd.hpp" #include "core/concepts.hpp" #include "core/config.hpp" -#include "core/constraint.hpp" +#include "core/config/trace_period_config.hpp" +#include "core/control/clocks/posix.hpp" +#include "core/control/clocks/steady.hpp" #include "core/control/session.hpp" +#include "core/control/triggers/time_window.hpp" #include "core/cpu.hpp" #include "core/gpu.hpp" #include "core/locking.hpp" @@ -87,9 +90,11 @@ #include #include #include +#include #include #include #include +#include #include #include #include @@ -431,6 +436,51 @@ invoke_external_resume_callbacks() _fn(); } +using trace_window_t = + rocprofsys::control::triggers::time_window; +using posix_trace_window_t = + rocprofsys::control::triggers::time_window; +using trace_config_t = + rocprofsys::config::trace_config; + +// Constructs a time_window, which self-registers with the session +// (and its initial action is reflected) as soon as it's constructed. Does +// NOT start the worker thread - that must happen after State::Active, since +// the worker captures its delay/duration reference point (t0) at start() +// time, and starting any earlier lets init-time overhead eat into the +// window. +template +std::unique_ptr> +make_time_window( + std::shared_ptr session, Clock& clock, + rocprofsys::control::triggers::time_window_specs specs, + rocprofsys::control::scope event_scope = rocprofsys::control::scope::global) +{ + using window_t = rocprofsys::control::triggers::time_window; + return std::make_unique(std::move(session), clock, specs, event_scope); +} + +rocprofsys::control::clocks::steady g_trace_window_clock; +rocprofsys::control::clocks::steady g_sampling_dur_window_clock; +std::optional g_posix_window_clock; +std::unique_ptr g_trace_window; +std::unique_ptr g_posix_trace_window; +std::unique_ptr g_sampling_dur_window; + +void +stop_time_windows() +{ + const auto stop_and_reset = [](auto& window) { + if(window) + { + window->stop(); + } + window.reset(); + }; + stop_and_reset(g_sampling_dur_window); + stop_and_reset(g_trace_window); + stop_and_reset(g_posix_trace_window); +} } // namespace extern "C" void @@ -690,17 +740,18 @@ rocprofsys_init_tooling_hidden(void) // clang-format off auto subscribers = std::to_array({ - { .on_pause = &rocprofiler_sdk::pause, .on_resume = &rocprofiler_sdk::resume, .name = "rocm" }, - { .on_pause = &sampling::pause, .on_resume = &sampling::resume, .name = "sampling" }, - { .on_pause = &component::mpi_gotcha::pause, .on_resume = &component::mpi_gotcha::resume, .name = "mpi" }, - { .on_pause = &ucx_t::pause, .on_resume = &ucx_t::resume, .name = "ucx" }, - { .on_pause = &shmem_t::pause, .on_resume = &shmem_t::resume, .name = "shmem" }, - { .on_pause = &component::vaapi_gotcha::pause, .on_resume = &component::vaapi_gotcha::resume, .name = "vaapi" }, - { .on_pause = &::rocprofsys::pthread_gotcha::pause, .on_resume = &::rocprofsys::pthread_gotcha::resume, .name = "pthread" }, - { .on_pause = &component::numa_gotcha::pause, .on_resume = &component::numa_gotcha::resume, .name = "numa" }, - { .on_pause = &rocprofsys::kokkosp::pause, .on_resume = &rocprofsys::kokkosp::resume, .name = "kokkos" }, - { .on_pause = &process_sampler::pause, .on_resume = &process_sampler::resume, .name = "process_sampler" }, - { .on_pause = &invoke_external_pause_callbacks, .on_resume = &invoke_external_resume_callbacks, .name = "external" }, + { .on_pause = &rocprofiler_sdk::pause, .on_resume = &rocprofiler_sdk::resume, .name = "rocm", .scopes = { control::scope::global } }, + { .on_pause = &sampling::pause, .on_resume = &sampling::resume, .name = "sampling", + .scopes = { control::scope::global, control::scope::sampling } }, + { .on_pause = &component::mpi_gotcha::pause, .on_resume = &component::mpi_gotcha::resume, .name = "mpi", .scopes = { control::scope::global } }, + { .on_pause = &ucx_t::pause, .on_resume = &ucx_t::resume, .name = "ucx", .scopes = { control::scope::global } }, + { .on_pause = &shmem_t::pause, .on_resume = &shmem_t::resume, .name = "shmem", .scopes = { control::scope::global } }, + { .on_pause = &component::vaapi_gotcha::pause, .on_resume = &component::vaapi_gotcha::resume, .name = "vaapi", .scopes = { control::scope::global } }, + { .on_pause = &::rocprofsys::pthread_gotcha::pause, .on_resume = &::rocprofsys::pthread_gotcha::resume, .name = "pthread", .scopes = { control::scope::global } }, + { .on_pause = &component::numa_gotcha::pause, .on_resume = &component::numa_gotcha::resume, .name = "numa", .scopes = { control::scope::global } }, + { .on_pause = &rocprofsys::kokkosp::pause, .on_resume = &rocprofsys::kokkosp::resume, .name = "kokkos", .scopes = { control::scope::global } }, + { .on_pause = &process_sampler::pause, .on_resume = &process_sampler::resume, .name = "process_sampler", .scopes = { control::scope::global } }, + { .on_pause = &invoke_external_pause_callbacks, .on_resume = &invoke_external_resume_callbacks, .name = "external", .scopes = { control::scope::global } }, }); // clang-format on for(auto& sub : subscribers) @@ -708,14 +759,78 @@ rocprofsys_init_tooling_hidden(void) get_control_session()->subscribe(std::move(sub)); } - // Every subscriber above must be registered before this call: a - // trigger's registration broadcasts a pause/resume transition - // immediately, so any subscriber added afterward would miss it. + // Every subscriber above must be registered before any trigger + // constructed below: a trigger's registration broadcasts a + // pause/resume transition immediately, so a subscriber added + // afterward would miss it. + if(auto trace_specs = trace_config_t::get_trace_specs(); !trace_specs.empty()) + { + const auto& spec = trace_specs.front(); + const auto delay = std::chrono::duration_cast( + std::chrono::duration{ spec.delay }); + const auto dur = std::chrono::duration_cast( + std::chrono::duration{ spec.duration }); + + // Safety-net subscriber for category-traited recording paths + // (timemory storage, perfetto trace_events from callbacks not + // covered by a subsystem pause subscriber). + get_control_session()->subscribe( + { .on_pause = + []() { + categories::disable_categories( + config::get_enabled_categories()); + }, + .on_resume = + []() { + categories::enable_categories( + config::get_enabled_categories()); + }, + .name = "trace_categories", + .scopes = { control::scope::global } }); + + if(trace_config_t::get_trace_period_clock_id() == + CLOCK_PROCESS_CPUTIME_ID) + { + g_posix_window_clock.emplace(CLOCK_PROCESS_CPUTIME_ID); + g_posix_trace_window = + make_time_window(get_control_session(), *g_posix_window_clock, + { .delay = delay, .duration = dur }); + } + else + { + g_trace_window = + make_time_window(get_control_session(), g_trace_window_clock, + { .delay = delay, .duration = dur }); + } + } + + if(const auto samp_dur = config::get_sampling_duration(); samp_dur > 0.0) + { + g_sampling_dur_window = make_time_window( + get_control_session(), g_sampling_dur_window_clock, + { .duration = std::chrono::duration_cast( + std::chrono::duration{ samp_dur }) }, + control::scope::sampling); + } + rocprofiler_sdk::create_roctx_client(); } state::process::set( state::process::Active); // set to active as very last operation + + if(g_trace_window) + { + g_trace_window->start(); + } + if(g_posix_trace_window) + { + g_posix_trace_window->start(); + } + if(g_sampling_dur_window) + { + g_sampling_dur_window->start(); + } } }; ROCPROFSYS_SCOPED_SAMPLING_ON_CHILD_THREADS(false); @@ -962,6 +1077,8 @@ rocprofsys_finalize_hidden(void) LOG_INFO("Finalizing rocprof-sys..."); + stop_time_windows(); + sampling::block_samples(); thread_info::set_stop(comp::wall_clock::record()); diff --git a/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk.cpp b/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk.cpp index 02cd015ed4f7..52afd7db9e83 100644 --- a/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk.cpp +++ b/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk.cpp @@ -2442,6 +2442,15 @@ tool_hip_stream_callback(rocprofiler_callback_tracing_record_t record, } #endif +// True when tool_init must skip starting the main (primary/counter) contexts, +// leaving them for the "rocm" subscriber's on_resume to start once the session +// goes active. +bool +should_defer_main_contexts() +{ + return !g_session->is_active(control::scope::global); +} + int tool_init(rocprofiler_client_finalize_t fini_func, void* user_data) { @@ -2772,23 +2781,12 @@ tool_init(rocprofiler_client_finalize_t fini_func, void* user_data) assert(g_session); create_roctx_client(); - if(g_roctx_client) - { - g_roctx_client->configure_services(_data->get_control_context()); + if(g_roctx_client) g_roctx_client->configure_services(_data->get_control_context()); - const auto filtering_active = g_roctx_client->get_trigger().filter_active(); - if(!filtering_active) - { - start(); - } - else - { - if(_data != nullptr) - { - start_context(_data->get_code_obj_context()); - start_context(_data->get_control_context()); - } - } + if(should_defer_main_contexts()) + { + start_context(_data->get_code_obj_context()); + start_context(_data->get_control_context()); } else { diff --git a/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk/roctx_client.cpp b/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk/roctx_client.cpp index 749b0e1cd435..b106fd98bba3 100644 --- a/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk/roctx_client.cpp +++ b/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk/roctx_client.cpp @@ -115,7 +115,6 @@ roctx_client::handle_marker_core_enter( { auto* data = static_cast(record.payload); - const bool write_enabled = m_trigger->should_write_markers(); switch(record.operation) { @@ -125,7 +124,7 @@ roctx_client::handle_marker_core_enter( const std::uint64_t range_id = s_push_range_id.fetch_sub(1, std::memory_order_relaxed); m_trigger->on_range_start(range_id, name); - const bool pushed_write_enabled = m_trigger->should_write_markers(); + const bool pushed_write_enabled = should_write(); m_pushed_ranges.push_back( { tim::add_hash_id(name), ts, pushed_write_enabled, range_id }); if(pushed_write_enabled) @@ -143,7 +142,7 @@ roctx_client::handle_marker_core_enter( { const char* name = data->args.roctxMarkA.message; tim::add_hash_id(name); - if(write_enabled) + if(should_write()) { m_writer.write_begin(name); } @@ -161,7 +160,7 @@ roctx_client::handle_marker_core_enter( } default: { - if(write_enabled) + if(should_write()) { const auto& name = trace_cache::get_metadata_registry().get_callback_tracing_info().at( @@ -228,7 +227,7 @@ roctx_client::handle_marker_core_exit( } case ROCPROFILER_MARKER_CORE_API_ID_roctxMarkA: { - if(m_trigger->should_write_markers()) + if(should_write()) { m_writer.write_end(data->args.roctxMarkA.message, begin_ts, ts, args_str, record); @@ -246,7 +245,7 @@ roctx_client::handle_marker_core_exit( m_trigger->on_range_start(range_id, name); - const bool write_enabled = m_trigger->should_write_markers(); + const bool write_enabled = should_write(); m_started_ranges.push_back( { tim::get_hash_id(name), begin_ts, write_enabled }); if(write_enabled) @@ -257,7 +256,7 @@ roctx_client::handle_marker_core_exit( } default: { - if(m_trigger->should_write_markers()) + if(should_write()) { const auto& name = trace_cache::get_metadata_registry().get_callback_tracing_info().at( diff --git a/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk/roctx_client.hpp b/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk/roctx_client.hpp index a04c401bacb1..65aade0ac947 100644 --- a/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk/roctx_client.hpp +++ b/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk/roctx_client.hpp @@ -73,6 +73,12 @@ class roctx_client static thread_local marker_range_stack_t m_pushed_ranges; static thread_local marker_range_stack_t m_started_ranges; + [[nodiscard]] bool should_write() const + { + return m_trigger->should_write_markers() && + m_session->is_active_without(control::triggers::roctx::k_trigger_name); + } + void handle_marker_core_enter(rocprofiler_callback_tracing_record_t record, rocprofiler_user_data_t* user_data, rocprofiler_timestamp_t ts); diff --git a/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk/tests/test_roctx_client.cpp b/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk/tests/test_roctx_client.cpp index 9fb6f6fe7cec..5d996d3e9821 100644 --- a/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk/tests/test_roctx_client.cpp +++ b/projects/rocprofiler-systems/source/lib/rocprof-sys/library/rocprofiler-sdk/tests/test_roctx_client.cpp @@ -213,7 +213,8 @@ class roctx_client_control_test : public mock_cleanup_base const auto& ctrl = client->get_session(); ctrl->subscribe({ .on_pause = [this]() { stop_count++; }, .on_resume = [this]() { start_count++; }, - .name = "test_counters" }); + .name = "test_counters", + .scopes = { rocprofsys::control::scope::global } }); return client; } @@ -227,7 +228,8 @@ class roctx_client_control_test : public mock_cleanup_base { m_session->subscribe({ .on_pause = [this]() { stop_count++; }, .on_resume = [this]() { start_count++; }, - .name = "test_counters" }); + .name = "test_counters", + .scopes = { rocprofsys::control::scope::global } }); const roctx_config_t config{ .pause_resume_enabled = true, .use_perfetto = false, @@ -850,7 +852,8 @@ class roctx_push_pop_region_test : public mock_cleanup_base const auto& ctrl = client->get_session(); ctrl->subscribe({ .on_pause = [this]() { stop_count++; }, .on_resume = [this]() { start_count++; }, - .name = "test_counters" }); + .name = "test_counters", + .scopes = { rocprofsys::control::scope::global } }); return client; } @@ -862,7 +865,8 @@ class roctx_push_pop_region_test : public mock_cleanup_base { m_session->subscribe({ .on_pause = [this]() { stop_count++; }, .on_resume = [this]() { start_count++; }, - .name = "test_counters" }); + .name = "test_counters", + .scopes = { rocprofsys::control::scope::global } }); const roctx_config_t config{ .pause_resume_enabled = true, .use_perfetto = false, @@ -988,3 +992,96 @@ TEST_F(roctx_push_pop_region_test, push_pop_no_filter_always_active) client->get_trigger().on_range_stop(k_push_id); EXPECT_TRUE(client->get_trigger().should_write_markers()); } + +class roctx_marker_gating_test_interface : public mock_cleanup_base +{ +protected: + using roctx_client_t = rocprofsys::rocprofiler_sdk::roctx_client; + using roctx_config_t = rocprofsys::rocprofiler_sdk::roctx_client_config; + + // Independent trigger used to pause the session without touching the + // roctx trigger itself — reproduces "some other trigger (e.g. the + // time_window trigger) has paused the global scope". + class other_trigger + { + public: + static constexpr std::string_view k_trigger_name = "other"; + + explicit other_trigger(rocprofsys::control::session& sess) + : m_session{ sess } + { + m_session.register_trigger(k_trigger_name, + rocprofsys::control::action::trace); + } + + ~other_trigger() { m_session.unregister_trigger(k_trigger_name); } + + other_trigger(const other_trigger&) = delete; + other_trigger& operator=(const other_trigger&) = delete; + other_trigger(other_trigger&&) = delete; + other_trigger& operator=(other_trigger&&) = delete; + + void set_action(rocprofsys::control::action act) const + { + m_session.set_action(k_trigger_name, act); + } + + private: + rocprofsys::control::session& m_session; + }; + + std::shared_ptr m_session = + std::make_shared(); + + std::unique_ptr make_client(const std::string& regions = "") + { + const roctx_config_t config{ .pause_resume_enabled = false, + .use_perfetto = false, + .use_timemory = false, + .perfetto_annotations = false, + .selected_trace_regions = regions }; + return std::make_unique(m_session, config); + } + + static bool observed_should_write(const roctx_client_t& client) + { + return client.get_trigger().should_write_markers() && + client.get_session()->is_active_without( + rocprofsys::control::triggers::roctx::k_trigger_name); + } +}; + +TEST_F(roctx_marker_gating_test_interface, should_write_true_when_session_fully_active) +{ + const auto client = make_client(); + EXPECT_TRUE(observed_should_write(*client)); +} + +TEST_F(roctx_marker_gating_test_interface, + should_write_false_when_session_paused_by_other_trigger) +{ + auto client = make_client(); + const auto& session = client->get_session(); + + const other_trigger other{ *session }; + other.set_action(rocprofsys::control::action::pause); + + ASSERT_TRUE(client->get_trigger().should_write_markers()) + << "the roctx trigger itself has no filter and is not paused - only the " + "unrelated 'other' trigger is pausing the session"; + + EXPECT_FALSE(observed_should_write(*client)) + << "should_write() must respect other triggers' votes, not just its own"; +} + +TEST_F(roctx_marker_gating_test_interface, + should_write_false_when_own_trigger_has_no_matching_region_active) +{ + auto client = make_client("Region1"); + + ASSERT_FALSE(client->get_trigger().should_write_markers()) + << "a region filter is configured but no matching region is active yet"; + + EXPECT_FALSE(observed_should_write(*client)) + << "should_write() must also respect the trigger's own filtering decision"; +} diff --git a/projects/rocprofiler-systems/source/lib/rocprof-sys/library/sampling.cpp b/projects/rocprofiler-systems/source/lib/rocprof-sys/library/sampling.cpp index c44191af3207..e400d98478f6 100644 --- a/projects/rocprofiler-systems/source/lib/rocprof-sys/library/sampling.cpp +++ b/projects/rocprofiler-systems/source/lib/rocprof-sys/library/sampling.cpp @@ -59,8 +59,7 @@ #include #include -#include -#include +#include #include #include #include @@ -476,136 +475,10 @@ get_sampler_running(std::int64_t _tid) return sampler_running_instances::instance(construct_on_thread{ _tid }, false); } -auto& -get_duration_disabled() -{ - static auto _v = std::atomic{ false }; - return _v; -} - -auto& -get_is_duration_thread() -{ - static thread_local auto _v = false; - return _v; -} - -auto& -get_duration_cv() -{ - static auto _v = std::condition_variable{}; - return _v; -} - -auto& -get_duration_mutex() -{ - static auto _v = std::mutex{}; - return _v; -} - -auto& -get_duration_thread() -{ - static auto _v = std::unique_ptr{}; - return _v; -} - -auto -notify_duration_thread() -{ - if(get_duration_thread() && !get_is_duration_thread()) - { - std::unique_lock _lk{ get_duration_mutex(), std::defer_lock }; - if(!_lk.owns_lock()) _lk.lock(); - get_duration_cv().notify_all(); - } -} - -void -stop_duration_thread() -{ - if(get_duration_thread() && !get_is_duration_thread()) - { - notify_duration_thread(); - get_duration_thread()->join(); - get_duration_thread().reset(); - } -} - -void -start_duration_thread() -{ - static std::mutex _start_mutex{}; - std::unique_lock _start_lk{ _start_mutex, std::defer_lock }; - if(!_start_lk.owns_lock()) _start_lk.lock(); - - if(!get_duration_thread() && config::get_sampling_duration() > 0.0) - { - // we may need to protect against recursion bc of pthread wrapper - static bool _protect = false; - if(_protect) return; - _protect = true; - const auto now = std::chrono::steady_clock::now(); - const auto end = - now + std::chrono::duration_cast( - std::chrono::duration{ config::get_sampling_duration() }); - const auto func = [end]() { - thread_info::init(true); - threading::set_thread_name("omni.samp.dur"); - get_is_duration_thread() = true; - bool _wait = true; - while(_wait) - { - _wait = false; - std::unique_lock _lk{ get_duration_mutex(), std::defer_lock }; - if(!_lk.owns_lock()) _lk.lock(); - get_duration_cv().wait_until(_lk, end); - const auto premature = (std::chrono::steady_clock::now() < end); - const auto finalized = - (state::process::get() >= state::process::Finalized); - if(premature && !finalized) - { - // protect against spurious wakeups - LOG_WARNING("Spurious wakeup of sampling duration thread..."); - _wait = true; - } - else if(finalized) - { - if(premature) - { - LOG_INFO("Sampling duration of {:.6f} seconds was " - "interrupted by finalization. Shutting down " - "sampling...", - config::get_sampling_duration()); - } - else - { - LOG_INFO("Sampling duration of {:.6f} seconds has " - "elapsed. Shutting down sampling...", - config::get_sampling_duration()); - } - break; - } - else - { - get_duration_disabled().store(true); - LOG_INFO("Sampling duration of {:.6f} seconds has elapsed. " - "Shutting down sampling...", - config::get_sampling_duration()); - configure(false, 0); - } - } - }; - - LOG_INFO("Sampling will be disabled after {:.6f} seconds", - config::get_sampling_duration()); - - ROCPROFSYS_SCOPED_SAMPLING_ON_CHILD_THREADS(false); - get_duration_thread() = std::make_unique(func); - _protect = false; - } -} +// Hoisted ahead of configure() so the new-sampler guard can consult it. +// pause_intervals / pause_mutex / pending_pause_ts stay near the pause/resume +// implementation below since only those touch them. +auto g_sampling_paused = std::atomic{ false }; auto& get_offload_file() @@ -792,7 +665,10 @@ configure(bool _setup, std::int64_t _tid) if(_setup && !_sampler && !_is_running && !_signal_types->empty()) { - if(get_duration_disabled()) return std::set{}; + if(g_sampling_paused.load(std::memory_order_relaxed)) + { + return std::set{}; + } // if this thread has an offset ID, that means it was created internally // and is probably here bc it called a function which was instrumented. @@ -971,7 +847,6 @@ configure(bool _setup, std::int64_t _tid) *_running = true; sampling::get_sampler_init(_tid)->sample(); - start_duration_thread(); _sampler->start(); } else if(!_setup && _sampler && _is_running) @@ -984,8 +859,6 @@ configure(bool _setup, std::int64_t _tid) sampling::block_signals(*_signal_types); } - notify_duration_thread(); - if(_tid == 0) { // this propagates to all threads @@ -1018,8 +891,6 @@ configure(bool _setup, std::int64_t _tid) // wait for the samples to finish for(auto& itr : get_sampler_allocators()) if(itr) itr->flush(); - - stop_duration_thread(); } if(trait::runtime_enabled::get()) @@ -1058,7 +929,6 @@ struct pause_interval_t std::uint64_t resume_ts = 0; }; -auto sampling_paused = std::atomic{ false }; auto pause_mutex = std::mutex{}; auto pause_intervals = std::vector{}; auto pending_pause_ts = std::atomic{ 0 }; @@ -1135,9 +1005,7 @@ shutdown() return std::set{}; } - auto configured_signals = configure(false, tid); - if(utility::get_thread_index() == 0) stop_duration_thread(); - return configured_signals; + return configure(false, tid); } void @@ -1154,6 +1022,38 @@ unblock_samples() trait::runtime_enabled::set(true); } +enum class timer_state +{ + stopped, + running +}; + +// Blocking samples only makes the handler discard them; the OS timer keeps +// firing and its signal keeps interrupting the target's sleeps. Stop the +// timers themselves so a paused sampler is unobservable to the application. +void +set_sampler_timers(timer_state state) +{ + for(std::int64_t i = 0; i < ROCPROFSYS_MAX_THREADS; ++i) + { + auto& sampler = get_sampler(i); + const auto& running = get_sampler_running(i); + if(!sampler || !running || !*running) + { + continue; + } + + if(state == timer_state::running) + { + sampler->start(); + } + else + { + sampler->stop(); + } + } +} + void block_signals(std::set _signals) { @@ -2026,7 +1926,7 @@ void pause() { bool _expected = false; - if(!sampling_paused.compare_exchange_strong(_expected, true)) + if(!g_sampling_paused.compare_exchange_strong(_expected, true)) { LOG_WARNING("sampling::pause() called but sampling is already paused"); return; @@ -2035,13 +1935,14 @@ pause() LOG_DEBUG("Pausing sampling..."); pending_pause_ts.store(tim::get_clock_real_now()); block_samples(); + set_sampler_timers(timer_state::stopped); } void resume() { bool _expected = true; - if(!sampling_paused.compare_exchange_strong(_expected, false)) + if(!g_sampling_paused.compare_exchange_strong(_expected, false)) { LOG_WARNING("sampling::resume() called but sampling is not paused"); return; @@ -2056,6 +1957,7 @@ resume() pause_intervals.push_back(pause_interval_t{ _pause_ts, _resume_ts }); } + set_sampler_timers(timer_state::running); unblock_samples(); } diff --git a/projects/rocprofiler-systems/source/tests/CMakeLists.txt b/projects/rocprofiler-systems/source/tests/CMakeLists.txt index 22ab72d594bf..0661b10b968b 100644 --- a/projects/rocprofiler-systems/source/tests/CMakeLists.txt +++ b/projects/rocprofiler-systems/source/tests/CMakeLists.txt @@ -7,6 +7,9 @@ set(UNIT_TEST_OBJECTS $ $ $ + $ + $ + $ $ $ $ diff --git a/projects/rocprofiler-systems/tests/pytest/test_openmp.py b/projects/rocprofiler-systems/tests/pytest/test_openmp.py index d7ea0be04770..c96db7db8321 100644 --- a/projects/rocprofiler-systems/tests/pytest/test_openmp.py +++ b/projects/rocprofiler-systems/tests/pytest/test_openmp.py @@ -6,9 +6,10 @@ """ from __future__ import annotations +import os import pytest from pathlib import Path -from conftest import RocprofsysTest +from conftest import RocprofsysTest, check_use_perfetto pytestmark = [ pytest.mark.openmp, @@ -17,6 +18,119 @@ ), # Requires SDK version >= 600, 6.3 ships with 500 ] +# ============================================================================ +# Sampling Duration Validation +# ============================================================================ + +SAMPLING_DURATION_SEC = 0.25 + +# The window is closed by a timer, so the last in-flight sample and the +# finalization behind it can land after the deadline. +SAMPLING_DURATION_UPPER_SLACK_SEC = 0.35 + +# A window that collapses to nothing would satisfy the upper bound trivially. +SAMPLING_DURATION_MIN_SEC = 0.02 + +# The run has to outlive the window, otherwise a workload that simply finished +# first is indistinguishable from a duration that actually cut sampling off. +SAMPLING_TRUNCATION_MARGIN_SEC = 0.25 + +_SAMPLING_WINDOW_QUERY = """ + SELECT COUNT(*) AS track_count, + MAX(track_end_ns - track_start_ns) AS max_track_span_ns, + MAX(track_end_ns) - MIN(track_start_ns) AS window_span_ns + FROM ( + SELECT track_id, + MIN(ts) AS track_start_ns, + MAX(ts + COALESCE(dur, 0)) AS track_end_ns + FROM slice + WHERE category = 'timer_sampling' + GROUP BY track_id + ) +""" + + +def _query_sampling_window(perfetto_file: Path): + """Return the aggregated timer_sampling extents from a Perfetto trace.""" + # check_use_perfetto() only proves that `import perfetto` succeeds, which a + # partial install satisfies without providing the trace_processor bindings. + try: + from perfetto.common.exceptions import PerfettoException + from perfetto.trace_processor import TraceProcessor, TraceProcessorConfig + except ImportError as exc: + pytest.skip(f"Perfetto trace processor bindings unavailable: {exc}") + + trace_processor_path = os.environ.get("ROCPROFSYS_TRACE_PROC_SHELL") + config = None + if trace_processor_path and Path(trace_processor_path).is_file(): + config = TraceProcessorConfig(bin_path=trace_processor_path) + + try: + trace_processor = ( + TraceProcessor(trace=str(perfetto_file), config=config) + if config is not None + else TraceProcessor(trace=str(perfetto_file)) + ) + except PerfettoException as exc: + pytest.skip(f"Perfetto trace processor unavailable on this system: {exc}") + + try: + rows = list(trace_processor.query(_SAMPLING_WINDOW_QUERY)) + finally: + close = getattr(trace_processor, "close", None) + if close is not None: + close() + + return rows[0] if rows else None + + +def _assert_sampling_duration_window(result) -> None: + """Assert ROCPROFSYS_SAMPLING_DURATION actually bounds timer sampling. + + The window is enforced by control::triggers::time_window, which emits no log + output, so the trace is the only place the contract is observable. + """ + if not check_use_perfetto(): + pytest.skip("Perfetto is disabled") + + perfetto_file = result.perfetto_file + if perfetto_file is None or not perfetto_file.exists(): + pytest.fail("No Perfetto trace file was produced by the run") + + row = _query_sampling_window(perfetto_file) + if row is None or not row.track_count or row.max_track_span_ns is None: + pytest.fail("No timer_sampling rows found in Perfetto output") + + max_track_span_sec = row.max_track_span_ns / 1.0e9 + window_span_sec = row.window_span_ns / 1.0e9 + max_allowed_sec = SAMPLING_DURATION_SEC + SAMPLING_DURATION_UPPER_SLACK_SEC + + if max_track_span_sec > max_allowed_sec: + pytest.fail( + "Sampling outlived the configured duration: max per-track span " + f"{max_track_span_sec:.3f}s > allowed {max_allowed_sec:.3f}s " + f"(ROCPROFSYS_SAMPLING_DURATION={SAMPLING_DURATION_SEC}s)" + ) + + if window_span_sec < SAMPLING_DURATION_MIN_SEC: + pytest.fail( + f"Sampling window collapsed to {window_span_sec:.3f}s; expected at " + f"least {SAMPLING_DURATION_MIN_SEC:.3f}s of samples before the " + "duration elapsed" + ) + + run_duration_sec = result.duration + if run_duration_sec is None: + return + + if window_span_sec + SAMPLING_TRUNCATION_MARGIN_SEC > run_duration_sec: + pytest.fail( + f"Run lasted {run_duration_sec:.3f}s but sampling covered " + f"{window_span_sec:.3f}s of it, so the duration cutoff is unproven " + "- the workload may simply have ended before the window closed" + ) + + # ============================================================================ # OpenMP Fixtures # ============================================================================ @@ -106,8 +220,6 @@ class TestOpenMPCG(RocprofsysTest): DURATION_SAMPLING_PASS_REGEX = [ r"Sampler for thread 0 will be triggered 1000\.0x per second of CPU-time", r"Sampler for thread 0 will be triggered 500\.0x per second of wall-time", - r"Sampling will be disabled after 0\.250000 seconds", - r"Sampling duration of 0\.250000 seconds (has elapsed|was interrupted by finalization)\. Shutting down sampling", r"sampling_percent\.(json|txt)", r"sampling_cpu_clock\.(json|txt)", r"sampling_wall_clock\.(json|txt)", @@ -141,6 +253,7 @@ def test_sampling_duration(self, ompt_sampling_env): env=ompt_sampling_env, ) self.assert_regex(result, pass_regex=self.DURATION_SAMPLING_PASS_REGEX) + _assert_sampling_duration_window(result) @pytest.mark.no_tmp_files def test_no_tmp_files(self, ompt_no_tmp_env): @@ -165,8 +278,6 @@ class TestOpenMPLU(RocprofsysTest): DURATION_SAMPLING_PASS_REGEX = [ r"Sampler for thread 0 will be triggered 1000\.0x per second of CPU-time", r"Sampler for thread 0 will be triggered 500\.0x per second of wall-time", - r"Sampling will be disabled after 0\.250000 seconds", - r"Sampling duration of 0\.250000 seconds (has elapsed|was interrupted by finalization)\. Shutting down sampling", r"sampling_percent\.(json|txt)", r"sampling_cpu_clock\.(json|txt)", r"sampling_wall_clock\.(json|txt)", @@ -206,6 +317,7 @@ def test_sampling_duration(self, ompt_sampling_env): env=ompt_sampling_env, ) self.assert_regex(result, pass_regex=self.DURATION_SAMPLING_PASS_REGEX) + _assert_sampling_duration_window(result) # ============================================================================ diff --git a/projects/rocprofiler-systems/tests/pytest/test_time_window.py b/projects/rocprofiler-systems/tests/pytest/test_time_window.py index fc27c920fde3..c2721ebbabc2 100644 --- a/projects/rocprofiler-systems/tests/pytest/test_time_window.py +++ b/projects/rocprofiler-systems/tests/pytest/test_time_window.py @@ -80,7 +80,7 @@ def test(self, mode, time_window_env): def test_delay(self, mode, time_window_env): env = time_window_env.copy() env.update( - {"ROCPROFSYS_TRACE_DELAY": "0.75", "ROCPROFSYS_TRACE_DURATION": "0.75"} + {"ROCPROFSYS_TRACE_DELAY": "0.75", "ROCPROFSYS_TRACE_DURATION": "0.85"} ) result = self.run_test( mode, diff --git a/projects/rocprofiler-systems/tests/pytest/test_transpose.py b/projects/rocprofiler-systems/tests/pytest/test_transpose.py index 961849fde391..6db892f58240 100644 --- a/projects/rocprofiler-systems/tests/pytest/test_transpose.py +++ b/projects/rocprofiler-systems/tests/pytest/test_transpose.py @@ -90,6 +90,17 @@ def rocprofiler_rules(validation_rules_dir: Path) -> list[Path]: ] +@pytest.fixture +def trace_delay_rules(validation_rules_dir: Path) -> list[Path]: + """Get validation rules asserting zero kernels were captured. + + Deliberately does NOT include default-rules.json/transpose_rules: those + require at least one kernel row, which is the opposite of what this + rule set checks. + """ + return [validation_rules_dir / "transpose" / "trace-delay-rules.json"] + + # ============================================================================ # Test Class: Basic Transpose Tests # ============================================================================ @@ -209,6 +220,43 @@ def test_two_kernels(self, mode, transpose_env): "(self-deadlock on synchronized<>'s rwlock)", } + @pytest.mark.timeout(120) + @pytest.mark.rocpd("transpose_env") + @pytest.mark.parametrize("mode", ["binary_rewrite"]) + def test_trace_delay_gates_gpu_contexts(self, mode, transpose_env, trace_delay_rules): + """ + Regression test: ROCPROFSYS_TRACE_DELAY must gate the actual startup of + the "main" rocprofiler-sdk contexts (primary_ctx, counter_ctx), not just + suppress downstream category emission. Before the fix, tool_init() + unconditionally started every context whenever no roctx marker/trace + region client existed - the common case for plain GPU tracing with just + TRACE_DELAY set - so a configured delay never produced a real gap in + cached GPU/RocPD data. + + transpose with TWO_KERNELS_RUN_ARGS (1 thread, 2 iterations, sync every + 2) completes in well under a second end-to-end. A 60s TRACE_DELAY is + two orders of magnitude larger than that, so the delay window is + guaranteed to never close before the process exits: if the fix works, + the GPU contexts never start and zero kernels are ever captured. + """ + env = transpose_env.copy() + env["ROCPROFSYS_TRACE_DELAY"] = "60" + result = self.run_test( + mode, + "transpose", + env=env, + binary_rewrite_args=self.BINARY_REWRITE_ARGS, + runtime_instrument_args=self.RUNTIME_INSTRUMENT_ARGS, + run_args=self.TWO_KERNELS_RUN_ARGS, + check_target_arch=True, + ) + self.assert_regex(result) + self.assert_rocpd( + result, + subtest_name="ROCpd TRACE_DELAY GPU context gating validation", + rules_files=trace_delay_rules, + ) + @pytest.mark.locks @pytest.mark.timeout(60) @pytest.mark.parametrize( diff --git a/projects/rocprofiler-systems/tests/rocpd-validation-rules/transpose/trace-delay-rules.json b/projects/rocprofiler-systems/tests/rocpd-validation-rules/transpose/trace-delay-rules.json new file mode 100644 index 000000000000..24914c95978e --- /dev/null +++ b/projects/rocprofiler-systems/tests/rocpd-validation-rules/transpose/trace-delay-rules.json @@ -0,0 +1,46 @@ +{ + "required_tables": [ + { + "min_rows": 0, + "name": "kernels", + "required_columns": [ + "id", + "category", + "name", + "start", + "end", + "queue", + "stream" + ], + "validation_queries": [ + { + "comparison": "equals", + "description": "TRACE_DELAY exceeding the whole process runtime must gate GPU context startup - zero kernels should ever be captured", + "error_message": "Kernels were captured despite ROCPROFSYS_TRACE_DELAY exceeding the process runtime; the delay window never closed, so the GPU contexts must never have started", + "expected_result": 0, + "query": "SELECT COUNT(*) as count FROM kernels" + } + ] + }, + { + "min_rows": 0, + "name": "top_kernels", + "required_columns": [ + "name", + "total_calls", + "total_duration", + "average", + "percentage" + ], + "validation_queries": [ + { + "comparison": "equals", + "description": "Kernel summary view must also be empty when TRACE_DELAY gates GPU context startup", + "error_message": "top_kernels has entries despite ROCPROFSYS_TRACE_DELAY gating GPU context startup", + "expected_result": 0, + "query": "SELECT COUNT(*) as count FROM top_kernels" + } + ] + } + ] +}