AI/rocm-systems
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Support for tracing mutex locking (#52)

Browse Source 
* Parallel overhead example with locks

* Support tracing mutex locking + more

- support wrapping pthread_mutex_lock
- support wrapping pthread_mutex_unlock
- support wrapping pthread_mutex_trylock
- get_perfetto_combined_traces setting
- OMNITRACE_TRACE_THREAD_LOCKS option
- ThreadState
- critical trace includes queue id
- enabled/disabled settings in timemory
- fix OMNITRACE_TIMEMORY_COMPONENTS
- fix reading config
- fix setting categories
- applied ThreadState::Internal in various places
- utility::get_filled_array
- utility::get_reserved_vector
- utility::get_thread_index
- fork_gotcha messages about forks
- split out some pthread_gotcha functionality into pthread_create_gotcha
- handle queue id in roctracer callbacks

* Update timemory and PTL submodules

* Misc CMake updates

- Includes fix to omnitrace-static-lib{gcc,stdcxx}

* Misc cleanup to pthread_mutex_gotcha and backtrace

* Fix to duplicate field in module_function json

* Improvement to debug messages

* omnitrace-dl and common improvements

- tweak to delimit
- common::ignore message
- common::join quoting of strings
- omnitrace_set_env ignores if inited and active
- omnitrace_set_mpi ignores if inited and active

* nsync for transpose example

* Fix to thread_deleter<void> functor invoke

* Fix thread state and HIP stream enums
This commit is contained in:
Jonathan R. Madsen
2022-05-08 04:40:10 -05:00
committed by GitHub
parent bab90baf0b
commit b208047741
50 changed files with 1736 additions and 483 deletions
Download Patch File Download Diff File Expand all files Collapse all files
source/lib/omnitrace/library/components/pthread_gotcha.cpp
+29 -232
View File
@@ -22,224 +22,62 @@
#include "library/components/pthread_gotcha.hpp"
#include "library/components/omnitrace.hpp"
#include "library/components/pthread_create_gotcha.hpp"
#include "library/components/pthread_mutex_gotcha.hpp"
#include "library/components/roctracer.hpp"
#include "library/config.hpp"
#include "library/debug.hpp"
#include "library/runtime.hpp"
#include "library/sampling.hpp"
#include "library/thread_data.hpp"
#include "library/utility.hpp"
#include <timemory/backends/threading.hpp>
#include <timemory/sampling/allocator.hpp>
#include <timemory/utility/types.hpp>
#include <ostream>
#include <pthread.h>
#include <array>
#include <vector>
namespace omnitrace
{
namespace sampling
{
std::set<int>
setup();
std::set<int>
shutdown();
} // namespace sampling
namespace mpl = tim::mpl;
using bundle_t = tim::lightweight_tuple<comp::wall_clock, comp::roctracer_data>;
using wall_pw_t = mpl::piecewise_select<comp::wall_clock>; // only wall-clock
using main_pw_t = mpl::piecewise_ignore<comp::wall_clock>; // exclude wall-clock
namespace
{
auto* is_shutdown = new bool{ false }; // intentional data leak
auto* bundles = new std::map<int64_t, std::shared_ptr<bundle_t>>{};
auto* bundles_mutex = new std::mutex{};
auto bundles_dtor = scope::destructor{ []() {
omnitrace::pthread_gotcha::shutdown();
delete bundles;
delete bundles_mutex;
bundles = nullptr;
bundles_mutex = nullptr;
} };
using bundle_t = tim::lightweight_tuple<pthread_create_gotcha_t, pthread_mutex_gotcha_t>;
inline void
start_bundle(bundle_t& _bundle)
auto&
get_sampling_on_child_threads_history(int64_t _idx = utility::get_thread_index())
{
if(!get_use_timemory()) return;
OMNITRACE_BASIC_VERBOSE_F(3, "starting bundle '%s'...\n", _bundle.key().c_str());
if(comp::roctracer::is_setup())
{
_bundle.push();
_bundle.start();
}
else
{
_bundle.push(wall_pw_t{});
_bundle.start(wall_pw_t{});
}
}
inline void
stop_bundle(bundle_t& _bundle, int64_t _tid)
{
if(!get_use_timemory()) return;
OMNITRACE_BASIC_VERBOSE_F(3, "stopping bundle '%s' in thread %li...\n",
_bundle.key().c_str(), _tid);
_bundle.stop(wall_pw_t{}); // stop wall-clock so we can get the value
// update roctracer_data
_bundle.store(std::plus<double>{},
_bundle.get<comp::wall_clock>()->get() * units::sec);
// stop all other components including roctracer_data after update
_bundle.stop(main_pw_t{});
// exclude popping wall-clock
_bundle.pop(_tid);
}
auto
get_thread_index()
{
static std::atomic<int64_t> _c{ 0 };
static thread_local int64_t _v = _c++;
return _v;
static auto _v = utility::get_filled_array<OMNITRACE_MAX_THREADS>(
[]() { return utility::get_reserved_vector<bool>(32); });
return _v.at(_idx);
}
auto&
get_sampling_on_child_threads_history(int64_t _idx = get_thread_index())
get_bundle()
{
static auto _v = std::array<std::vector<bool>, OMNITRACE_MAX_THREADS>{};
return _v.at(_idx);
static auto _v = std::unique_ptr<bundle_t>{};
if(!_v) _v = std::make_unique<bundle_t>("pthread_gotcha");
return _v;
}
} // namespace
//--------------------------------------------------------------------------------------//
pthread_gotcha::wrapper::wrapper(routine_t _routine, void* _arg, bool _enable_sampling,
int64_t _parent, promise_t* _p)
: m_enable_sampling{ _enable_sampling }
, m_parent_tid{ _parent }
, m_routine{ _routine }
, m_arg{ _arg }
, m_promise{ _p }
{}
void*
pthread_gotcha::wrapper::operator()() const
{
if(is_shutdown && *is_shutdown)
{
if(m_promise) m_promise->set_value();
// execute the original function
return m_routine(m_arg);
}
int64_t _tid = -1;
auto _is_sampling = false;
auto _bundle = std::shared_ptr<bundle_t>{};
auto _signals = std::set<int>{};
auto _coverage = (get_mode() == omnitrace::Mode::Coverage);
auto _dtor = scope::destructor{ [&]() {
if(_is_sampling)
{
sampling::block_signals(_signals);
sampling::shutdown();
}
if(!bundles || !bundles_mutex) return;
if(_bundle && get_state() < omnitrace::State::Finalized)
{
std::unique_lock<std::mutex> _lk{ *bundles_mutex };
stop_bundle(*_bundle, _tid);
_bundle.reset();
bundles->erase(_tid);
}
} };
auto _active = (get_state() == omnitrace::State::Active && bundles && bundles_mutex);
if(_active && !_coverage)
{
_tid = threading::get_id();
threading::set_thread_name(TIMEMORY_JOIN(" ", "Thread", _tid).c_str());
if(bundles && bundles_mutex)
{
std::unique_lock<std::mutex> _lk{ *bundles_mutex };
if(comp::roctracer::is_setup())
_bundle =
bundles->emplace(_tid, std::make_shared<bundle_t>("start_thread"))
.first->second;
}
if(_bundle) start_bundle(*_bundle);
get_cpu_cid_stack(threading::get_id(), m_parent_tid);
if(m_enable_sampling)
{
// initialize thread-local statics
(void) tim::get_unw_backtrace<12, 1, false>();
_is_sampling = true;
push_enable_sampling_on_child_threads(false);
_signals = sampling::setup();
pop_enable_sampling_on_child_threads();
sampling::unblock_signals();
}
}
if(m_promise) m_promise->set_value();
// execute the original function
return m_routine(m_arg);
}
void*
pthread_gotcha::wrapper::wrap(void* _arg)
{
if(_arg == nullptr) return nullptr;
// convert the argument
wrapper* _wrapper = static_cast<wrapper*>(_arg);
// execute the original function
return (*_wrapper)();
}
void
pthread_gotcha::configure()
{
pthread_gotcha_t::get_initializer() = []() {
pthread_gotcha_t::template configure<0, int, pthread_t*, const pthread_attr_t*,
void* (*) (void*), void*>("pthread_create");
};
pthread_create_gotcha::configure();
pthread_mutex_gotcha::configure();
}
void
pthread_gotcha::shutdown()
{
if(is_shutdown)
{
if(*is_shutdown) return;
*is_shutdown = true;
}
if(!bundles_mutex || !bundles) return;
std::unique_lock<std::mutex> _lk{ *bundles_mutex };
unsigned long _ndangling = 0;
for(auto itr : *bundles)
{
if(itr.second)
{
stop_bundle(*itr.second, itr.first);
++_ndangling;
}
itr.second.reset();
}
bundles->clear();
OMNITRACE_CONDITIONAL_BASIC_PRINT(
(get_verbose_env() >= 2 || get_debug_env()) && _ndangling > 0,
"[pthread_gotcha::shutdown] cleaned up %lu dangling bundles\n", _ndangling);
pthread_create_gotcha::shutdown();
pthread_mutex_gotcha::shutdown();
}
bool
@@ -287,57 +125,16 @@ pthread_gotcha::sampling_on_child_threads()
return _v;
}
// pthread_create
int
pthread_gotcha::operator()(pthread_t* thread, const pthread_attr_t* attr,
void* (*start_routine)(void*), void* arg) const
void
pthread_gotcha::start()
{
bundle_t _bundle{ "pthread_create" };
auto _enable_sampling = sampling_enabled_on_child_threads();
auto _coverage = (get_mode() == omnitrace::Mode::Coverage);
auto _active = (get_state() == omnitrace::State::Active);
int64_t _tid = (_active) ? threading::get_id() : 0;
// ensure that cpu cid stack exists on the parent thread if active
if(!_coverage && _active) get_cpu_cid_stack();
if(!get_use_sampling() || !_enable_sampling)
{
auto* _obj = new wrapper(start_routine, arg, _enable_sampling, _tid, nullptr);
// create the thread
auto _ret =
::pthread_create(thread, attr, &wrapper::wrap, static_cast<void*>(_obj));
return _ret;
}
// block the signals in entire process
OMNITRACE_DEBUG("blocking signals...\n");
tim::sampling::block_signals({ SIGALRM, SIGPROF },
tim::sampling::sigmask_scope::process);
start_bundle(_bundle);
// promise set by thread when signal handler is configured
auto _promise = std::promise<void>{};
auto _fut = _promise.get_future();
auto* _wrap = new wrapper(start_routine, arg, _enable_sampling, _tid, &_promise);
// create the thread
auto _ret = ::pthread_create(thread, attr, &wrapper::wrap, static_cast<void*>(_wrap));
// wait for thread to set promise
OMNITRACE_DEBUG("waiting for child to signal it is setup...\n");
_fut.wait();
stop_bundle(_bundle, threading::get_id());
// unblock the signals in the entire process
OMNITRACE_DEBUG("unblocking signals...\n");
tim::sampling::unblock_signals({ SIGALRM, SIGPROF },
tim::sampling::sigmask_scope::process);
OMNITRACE_DEBUG("returning success...\n");
return _ret;
get_bundle()->start();
}
void
pthread_gotcha::stop()
{
get_bundle()->stop();
get_bundle().reset();
}
} // namespace omnitrace