AI/rocm-systems
2
0
Forkuj 0
Kod Zgłoszenia Oczekujące zmiany Actions Packages Projekty Wydania Wiki Aktywność

Buffering: initial implementation and tests (#20)

Przeglądaj źródła 
* Update source/lib/common

- CMakeLists.txt
  - less verbose
  - rocprofiler-common-library uses rocprofiler-headers target
- mpl.hpp
  - metaprogramming header with type_list, size_of, index_of, and is_one_of
- record_header_buffer.{hpp,cpp}
  - wrapper class around atomic_ring_buffer and vector of rocprofiler_record_header_t
- atomic_ring_buffer.{hpp,cpp}
  - request function accepts wrap param when overwritting is not desirable
  - can_clear member function
  - clear member function for rewinding write pointer to start of buffer
- containers/CMakeLists.txt
  - include record_header_buffer.{hpp,cpp} in build target

* Update source/lib/tests: Buffering tests

- Added buffering tests. See comments in code for description

* atomic_ring_buffer -> ring_buffer

- remove ring_buffer implementation
- rename atomic_ring_buffer to ring_buffer

* atomic_ring_buffer -> ring_buffer

- remove ring_buffer implementation
- rename atomic_ring_buffer to ring_buffer

* Update record_header_buffer

- lock, unlock, is_locked, clear, save, and load member functions

* Buffering tests

- add buffer test for save/load capability

* Update rocprofiler_memcheck.cmake

- fix erroneous spaces causing incorrect string evaluation

* Update ring_buffer

- fix exception message

* undef HIP_PROF_API

- make sure HIP_PROF_API is undefined before including hip_runtime.h
- avoid directly including hip/hip_runtime.h

* Update rocprofiler_config_interfaces

- remove stale preprocessor defines that are from old rocprofiler/roctracer
  - HIP_PROF_HIP_API_STRING=1
  - PROF_API_IMPL=1

* Update run-ci.py

- fix paths to suppression files
- improve printing logs to console in github actions

* Update buffering implementation

- remove support for using malloc instead of mmap in ring_buffer
- provide some info functions in record_header_buffer
- improve the testing of the save-load buffer test

* Update run-ci.py

- fix CTEST_CUSTOM_COVERAGE_EXCLUDE

* Update hip/api_args.h

- remove undef HIP_PROF_API

* Update buffering-save-load.cpp

- updated comments

* Update record_header_buffer

- default ctor
- allocate member function
- is_allocated member function

* Update buffering-save-load.cpp

- tweaked usage of record_header_buffer to delay allocation

[ROCm/rocprofiler-sdk commit: b12ef4a75e]
This commit is contained in:
Jonathan R. Madsen
2023-08-30 11:34:03 -05:00
zatwierdzone przez GitHub
rodzic d4a977349c
commit ccd154b74c
20 zmienionych plików z 1525 dodań i 1063 usunięć
Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files
projects/rocprofiler-sdk/source/lib/common/CMakeLists.txt
+8 -8
Wyświetl plik
@@ -1,10 +1,9 @@
set(common_sources ${CMAKE_CURRENT_LIST_DIR}/config.cpp
${CMAKE_CURRENT_LIST_DIR}/helper.cpp)
set(common_headers
${CMAKE_CURRENT_LIST_DIR}/config.hpp ${CMAKE_CURRENT_LIST_DIR}/defines.hpp
${CMAKE_CURRENT_LIST_DIR}/environment.hpp ${CMAKE_CURRENT_LIST_DIR}/join.hpp
${CMAKE_CURRENT_LIST_DIR}/log.hpp ${CMAKE_CURRENT_LIST_DIR}/helper.hpp)
#
# Builds common utilities into a static library
#
set(common_sources config.cpp helper.cpp)
set(common_headers config.hpp defines.hpp environment.hpp join.hpp log.hpp helper.hpp
mpl.hpp)
add_library(rocprofiler-common-library STATIC)
add_library(rocprofiler::rocprofiler-common-library ALIAS rocprofiler-common-library)
@@ -17,7 +16,8 @@ target_include_directories(rocprofiler-common-library
target_link_libraries(
rocprofiler-common-library
PUBLIC $<BUILD_INTERFACE:rocprofiler::rocprofiler-build-flags>
PUBLIC $<BUILD_INTERFACE:rocprofiler::rocprofiler-headers>
$<BUILD_INTERFACE:rocprofiler::rocprofiler-build-flags>
$<BUILD_INTERFACE:rocprofiler::rocprofiler-threading>
$<BUILD_INTERFACE:rocprofiler::rocprofiler-memcheck>
$<BUILD_INTERFACE:rocprofiler::rocprofiler-stdcxxfs>
projects/rocprofiler-sdk/source/lib/common/container/CMakeLists.txt
+5 -6
Wyświetl plik
@@ -1,10 +1,9 @@
#
set(containers_sources)
set(containers_headers atomic_ring_buffer.hpp c_array.hpp operators.hpp ring_buffer.hpp
stable_vector.hpp static_vector.hpp)
set(containers_sources atomic_ring_buffer.cpp ring_buffer.cpp)
# add container sources and headers to common library target
#
set(containers_headers ring_buffer.hpp c_array.hpp operators.hpp record_header_buffer.hpp
ring_buffer.hpp stable_vector.hpp static_vector.hpp)
set(containers_sources ring_buffer.cpp record_header_buffer.cpp ring_buffer.cpp)
target_sources(rocprofiler-common-library PRIVATE ${containers_sources}
${containers_headers})
projects/rocprofiler-sdk/source/lib/common/container/atomic_ring_buffer.cpp
-295
Wyświetl plik
@@ -1,295 +0,0 @@
// MIT License
//
// Copyright (c) 2023 ROCm Developer Tools
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
#include "atomic_ring_buffer.hpp"
#include "lib/common/environment.hpp"
#include "lib/common/units.hpp"
#include <sys/mman.h>
#include <cerrno>
#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
namespace rocprofiler
{
namespace common
{
namespace container
{
namespace base
{
atomic_ring_buffer::atomic_ring_buffer(size_t _size, bool _use_mmap)
{
set_use_mmap(_use_mmap);
init(_size);
}
atomic_ring_buffer::~atomic_ring_buffer() { destroy(); }
atomic_ring_buffer::atomic_ring_buffer(const atomic_ring_buffer& rhs)
: m_use_mmap{rhs.m_use_mmap}
, m_use_mmap_explicit{rhs.m_use_mmap_explicit}
{
init(rhs.m_size);
}
atomic_ring_buffer::atomic_ring_buffer(atomic_ring_buffer&& rhs) noexcept
: m_init{rhs.m_init}
, m_use_mmap{rhs.m_use_mmap}
, m_use_mmap_explicit{rhs.m_use_mmap_explicit}
, m_ptr{rhs.m_ptr}
, m_size{rhs.m_size}
, m_read_count{rhs.m_read_count.load()}
, m_write_count{rhs.m_write_count.load()}
{
rhs.reset();
}
atomic_ring_buffer&
atomic_ring_buffer::operator=(const atomic_ring_buffer& rhs)
{
if(this == &rhs) return *this;
destroy();
m_use_mmap = rhs.m_use_mmap;
m_use_mmap_explicit = rhs.m_use_mmap_explicit;
init(rhs.m_size);
return *this;
}
atomic_ring_buffer&
atomic_ring_buffer::operator=(atomic_ring_buffer&& rhs) noexcept
{
if(this == &rhs) return *this;
destroy();
m_init = rhs.m_init;
m_use_mmap = rhs.m_use_mmap;
m_use_mmap_explicit = rhs.m_use_mmap_explicit;
m_ptr = rhs.m_ptr;
m_size = rhs.m_size;
m_read_count = rhs.m_read_count.load();
m_write_count = rhs.m_write_count.load();
rhs.reset();
return *this;
}
void
atomic_ring_buffer::init(size_t _size)
{
if(m_init)
throw std::runtime_error(
"tim::base::atomic_ring_buffer::init(size_t) :: already initialized");
m_init = true;
// Round up to multiple of page size.
_size += units::get_page_size() - ((_size % units::get_page_size() > 0)
? (_size % units::get_page_size())
: units::get_page_size());
if((_size % units::get_page_size()) > 0)
{
std::ostringstream _oss{};
_oss << "Error! size is not a multiple of page size: " << _size << " % "
<< units::get_page_size() << " = " << (_size % units::get_page_size());
throw std::runtime_error(_oss.str());
}
m_size = _size;
m_read_count = 0;
m_write_count = 0;
if(!m_use_mmap_explicit) m_use_mmap = get_env("ROCPROFILER_USE_MMAP", m_use_mmap);
if(!m_use_mmap)
{
m_ptr = malloc(m_size * sizeof(char));
return;
}
// Map twice the buffer size.
if((m_ptr =
mmap(nullptr, m_size, PROT_READ | PROT_WRITE, MAP_ANONYMOUS | MAP_PRIVATE, -1, 0)) ==
MAP_FAILED)
{
destroy();
auto _err = errno;
// TIMEMORY_PRINTF_FATAL(stderr, "Error using mmap: %s\n", strerror(_err));
throw std::runtime_error(strerror(_err));
}
}
void
atomic_ring_buffer::destroy()
{
if(m_ptr && m_init)
{
if(!m_use_mmap)
{
::free(m_ptr);
}
else
{
// Unmap the mapped virtual memmory.
auto ret = munmap(m_ptr, m_size);
if(ret != 0) perror("munmap");
}
}
m_init = false;
m_size = 0;
m_read_count = 0;
m_write_count = 0;
m_ptr = nullptr;
}
void
atomic_ring_buffer::set_use_mmap(bool _v)
{
if(m_init)
throw std::runtime_error("tim::base::atomic_ring_buffer::set_use_mmap(bool) cannot be "
"called after initialization");
m_use_mmap = _v;
m_use_mmap_explicit = true;
}
std::string
atomic_ring_buffer::as_string() const
{
std::ostringstream ss{};
ss << std::boolalpha << "is_initialized: " << is_initialized() << ", capacity: " << capacity()
<< ", count: " << count() << ", free: " << free() << ", is_empty: " << is_empty()
<< ", is_full: " << is_full() << ", pointer: " << m_ptr << ", read count: " << m_read_count
<< ", write count: " << m_write_count;
return ss.str();
}
//
void*
atomic_ring_buffer::request(size_t _length)
{
if(m_ptr == nullptr || m_size == 0) return nullptr;
if(is_full()) return retrieve(_length);
// if write count is at the tail of buffer, bump to the end of buffer
size_t _write_count = 0;
size_t _offset = 0;
do
{
// Make sure we don't put in more than there's room for, by writing no
// more than there is free.
if(_length > free()) return nullptr;
_offset = 0;
_write_count = m_write_count.load();
auto _modulo = m_size - (_write_count % m_size);
if(_modulo < _length) _offset = _modulo;
} while(!m_write_count.compare_exchange_strong(
_write_count, _write_count + _length + _offset, std::memory_order_seq_cst));
// pointer in buffer
void* _out = write_ptr(_write_count);
return _out;
}
//
void*
atomic_ring_buffer::retrieve(size_t _length) const
{
if(m_ptr == nullptr || m_size == 0) return nullptr;
// Make sure we don't put in more than there's room for, by writing no
// more than there is free.
// if read count is at the tail of buffer, bump to the end of buffer
size_t _read_count = 0;
size_t _offset = 0;
do
{
if(_length > count()) return nullptr;
_offset = 0;
_read_count = m_read_count.load();
auto _modulo = m_size - (_read_count % m_size);
if(_modulo < _length) _offset = _modulo;
} while(!m_read_count.compare_exchange_strong(
_read_count, _read_count + _length + _offset, std::memory_order_seq_cst));
// pointer in buffer
void* _out = read_ptr(_read_count);
return _out;
}
//
void
atomic_ring_buffer::reset()
{
m_init = false;
m_size = 0;
m_ptr = nullptr;
m_read_count.store(0);
m_write_count.store(0);
}
//
void
atomic_ring_buffer::save(std::fstream& _fs)
{
auto _read_count = m_read_count.load();
auto _write_count = m_write_count.load();
_fs.write(reinterpret_cast<char*>(&m_use_mmap), sizeof(m_use_mmap));
_fs.write(reinterpret_cast<char*>(&m_use_mmap_explicit), sizeof(m_use_mmap_explicit));
_fs.write(reinterpret_cast<char*>(&m_size), sizeof(m_size));
_fs.write(reinterpret_cast<char*>(&_read_count), sizeof(_read_count));
_fs.write(reinterpret_cast<char*>(&_write_count), sizeof(_write_count));
_fs.write(reinterpret_cast<char*>(m_ptr), m_size * sizeof(char));
}
//
void
atomic_ring_buffer::load(std::fstream& _fs)
{
destroy();
size_t _read_count = 0;
size_t _write_count = 0;
_fs.read(reinterpret_cast<char*>(&m_use_mmap), sizeof(m_use_mmap));
_fs.read(reinterpret_cast<char*>(&m_use_mmap_explicit), sizeof(m_use_mmap_explicit));
_fs.read(reinterpret_cast<char*>(&m_size), sizeof(m_size));
init(m_size);
if(!m_ptr) m_ptr = malloc(m_size);
_fs.read(reinterpret_cast<char*>(&_read_count), sizeof(_read_count));
_fs.read(reinterpret_cast<char*>(&_write_count), sizeof(_write_count));
_fs.read(reinterpret_cast<char*>(m_ptr), m_size * sizeof(char));
m_read_count.store(_read_count);
m_write_count.store(_write_count);
}
} // namespace base
} // namespace container
} // namespace common
} // namespace rocprofiler
projects/rocprofiler-sdk/source/lib/common/container/atomic_ring_buffer.hpp
-424
Wyświetl plik
@@ -1,424 +0,0 @@
// MIT License
//
// Copyright (c) 2023 ROCm Developer Tools
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
#pragma once
#include "lib/common/environment.hpp"
#include "lib/common/units.hpp"
#include <algorithm>
#include <atomic>
#include <cmath>
#include <cstddef>
#include <cstdlib>
#include <fstream>
#include <functional>
#include <iomanip>
#include <iostream>
#include <sstream>
#include <stdexcept>
#include <utility>
#include <vector>
namespace rocprofiler
{
namespace common
{
namespace container
{
template <typename Tp>
struct atomic_ring_buffer;
//
namespace base
{
/// \struct tim::base::atomic_ring_buffer
/// \brief Ring buffer implementation, with support for mmap as backend (Linux only).
struct atomic_ring_buffer
{
template <typename Tp>
friend struct container::atomic_ring_buffer;
atomic_ring_buffer() = default;
explicit atomic_ring_buffer(bool _use_mmap) { set_use_mmap(_use_mmap); }
explicit atomic_ring_buffer(size_t _size) { init(_size); }
atomic_ring_buffer(size_t _size, bool _use_mmap);
~atomic_ring_buffer();
atomic_ring_buffer(const atomic_ring_buffer&);
atomic_ring_buffer& operator=(const atomic_ring_buffer&);
atomic_ring_buffer(atomic_ring_buffer&&) noexcept;
atomic_ring_buffer& operator=(atomic_ring_buffer&&) noexcept;
/// Returns whether the buffer has been allocated
bool is_initialized() const { return m_init; }
/// Get the total number of bytes supported
size_t capacity() const { return m_size; }
/// Creates new ring buffer.
void init(size_t size);
/// Destroy ring buffer.
void destroy();
/// Request a pointer for writing at least \param n bytes.
void* request(size_t n);
/// Retrieve a pointer for reading at least \param n bytes.
void* retrieve(size_t n) const;
/// Write class-type data to buffer (uses placement new).
template <typename Tp>
std::pair<size_t, Tp*> write(Tp* in, std::enable_if_t<std::is_class<Tp>::value, int> = 0);
/// Write non-class-type data to buffer (uses memcpy).
template <typename Tp>
std::pair<size_t, Tp*> write(Tp* in, std::enable_if_t<!std::is_class<Tp>::value, int> = 0);
/// Request a pointer to an allocation. This is similar to a "write" except the
/// memory is uninitialized. Typically used by allocators. If Tp is a class type,
/// be sure to use a placement new instead of a memcpy.
template <typename Tp>
Tp* request();
/// Read class-type data from buffer (uses placement new).
template <typename Tp>
std::pair<size_t, Tp*> read(Tp* _dest,
std::enable_if_t<std::is_class<Tp>::value, int> = 0) const;
/// Read non-class-type data from buffer (uses memcpy).
template <typename Tp>
std::pair<size_t, Tp*> read(Tp* _dest,
std::enable_if_t<!std::is_class<Tp>::value, int> = 0) const;
/// Retrieve a pointer to the head allocation (read).
template <typename Tp>
Tp* retrieve() const;
/// Returns number of bytes currently held by the buffer.
size_t count() const { return (m_write_count - m_read_count); }
/// Returns how many bytes are availiable in the buffer.
size_t free() const { return (m_size - count()); }
/// Returns if the buffer is empty.
bool is_empty() const { return (count() == 0); }
/// Returns if the buffer is full.
bool is_full() const { return (count() == m_size); }
/// explicitly configure to use mmap if avail
void set_use_mmap(bool);
/// query whether using mmap
bool get_use_mmap() const { return m_use_mmap; }
std::string as_string() const;
void save(std::fstream& _fs);
void load(std::fstream& _fs);
private:
/// Returns the current write pointer.
void* write_ptr(size_t _write_count) const
{
return static_cast<char*>(m_ptr) + (_write_count % m_size);
}
/// Returns the current read pointer.
void* read_ptr(size_t _read_count) const
{
return static_cast<char*>(m_ptr) + (_read_count % m_size);
}
void reset();
private:
bool m_init = false;
bool m_use_mmap = true;
bool m_use_mmap_explicit = false;
void* m_ptr = nullptr;
size_t m_size = 0;
mutable std::atomic<size_t> m_read_count = 0;
std::atomic<size_t> m_write_count = 0;
};
//
template <typename Tp>
std::pair<size_t, Tp*>
atomic_ring_buffer::write(Tp* in, std::enable_if_t<std::is_class<Tp>::value, int>)
{
if(in == nullptr || m_ptr == nullptr) return {0, nullptr};
auto _length = sizeof(Tp);
void* _out_p = request(_length);
if(_out_p == nullptr) return {0, nullptr};
// Copy in.
new(_out_p) Tp{std::move(*in)};
// pointer in buffer
Tp* _out = reinterpret_cast<Tp*>(_out_p);
return {_length, _out};
}
//
template <typename Tp>
std::pair<size_t, Tp*>
atomic_ring_buffer::write(Tp* in, std::enable_if_t<!std::is_class<Tp>::value, int>)
{
if(in == nullptr || m_ptr == nullptr) return {0, nullptr};
auto _length = sizeof(Tp);
void* _out_p = request(_length);
if(_out_p == nullptr) return {0, nullptr};
// Copy in.
memcpy(_out_p, in, _length);
// pointer in buffer
Tp* _out = reinterpret_cast<Tp*>(_out_p);
return {_length, _out};
}
//
template <typename Tp>
Tp*
atomic_ring_buffer::request()
{
if(m_ptr == nullptr) return nullptr;
return request(sizeof(Tp));
}
//
template <typename Tp>
std::pair<size_t, Tp*>
atomic_ring_buffer::read(Tp* _dest, std::enable_if_t<std::is_class<Tp>::value, int>) const
{
if(is_empty() || _dest == nullptr) return {0, nullptr};
auto _length = sizeof(Tp);
void* _out_p = retrieve(_length);
if(_out_p == nullptr) return {0, nullptr};
// pointer in buffer
Tp* in = reinterpret_cast<Tp*>(_out_p);
// Copy out for BYTE, nothing magic here.
*_dest = *in;
return {_length, in};
}
//
template <typename Tp>
std::pair<size_t, Tp*>
atomic_ring_buffer::read(Tp* _dest, std::enable_if_t<!std::is_class<Tp>::value, int>) const
{
if(is_empty() || _dest == nullptr) return {0, nullptr};
auto _length = sizeof(Tp);
void* _out_p = retrieve(_length);
if(_out_p == nullptr) return {0, nullptr};
// pointer in buffer
Tp* in = reinterpret_cast<Tp*>(_out_p);
using Up = typename std::remove_const<Tp>::type;
// Copy out for BYTE, nothing magic here.
Up* _out = const_cast<Up*>(_dest);
memcpy(_out, in, _length);
return {_length, in};
}
//
template <typename Tp>
Tp*
atomic_ring_buffer::retrieve() const
{
if(m_ptr == nullptr) return nullptr;
return retrieve(sizeof(Tp));
}
//
} // namespace base
//
/// \struct tim::data_storage::atomic_ring_buffer
/// \brief Ring buffer wrapper around \ref tim::base::atomic_ring_buffer for data of type
/// Tp. If the data object size is larger than the page size (typically 4KB), behavior is
/// undefined. During initialization, one requests a minimum number of objects and the
/// buffer will support that number of object + the remainder of the page, e.g. if a page
/// is 1000 bytes, the object is 1 byte, and the buffer is requested to support 1500
/// objects, then an allocation supporting 2000 objects (i.e. 2 pages) will be created.
template <typename Tp>
struct atomic_ring_buffer : private base::atomic_ring_buffer
{
using base_type = base::atomic_ring_buffer;
static size_t get_items_per_page();
atomic_ring_buffer() = default;
~atomic_ring_buffer() = default;
explicit atomic_ring_buffer(bool _use_mmap)
: base_type{_use_mmap}
{}
explicit atomic_ring_buffer(size_t _size)
: base_type{_size * sizeof(Tp)}
{}
atomic_ring_buffer(size_t _size, bool _use_mmap)
: base_type{_size * sizeof(Tp), _use_mmap}
{}
atomic_ring_buffer(const atomic_ring_buffer&);
atomic_ring_buffer(atomic_ring_buffer&&) noexcept = default;
atomic_ring_buffer& operator=(const atomic_ring_buffer&);
atomic_ring_buffer& operator=(atomic_ring_buffer&&) noexcept = default;
/// Returns whether the buffer has been allocated
bool is_initialized() const { return base_type::is_initialized(); }
/// Get the total number of Tp instances supported
size_t capacity() const { return (base_type::capacity()) / sizeof(Tp); }
/// Creates new ring buffer.
void init(size_t _size) { base_type::init(_size * sizeof(Tp)); }
/// Destroy ring buffer.
void destroy() { base_type::destroy(); }
/// Write data to buffer.
size_t data_size() const { return sizeof(Tp); }
/// Write data to buffer. Return pointer to location of write
Tp* write(Tp* in) { return base_type::write<Tp>(in).second; }
/// Read data from buffer. Return pointer to location of read
Tp* read(Tp* _dest) const { return base_type::read<Tp>(_dest).second; }
/// Get an uninitialized address at tail of buffer.
Tp* request() { return base_type::request<Tp>(); }
/// Read data from head of buffer.
Tp* retrieve() { return base_type::retrieve<Tp>(); }
/// Returns number of Tp instances currently held by the buffer.
size_t count() const { return (base_type::count()) / sizeof(Tp); }
/// Returns how many Tp instances are availiable in the buffer.
size_t free() const { return (base_type::free()) / sizeof(Tp); }
/// Returns if the buffer is empty.
bool is_empty() const { return base_type::is_empty(); }
/// Returns if the buffer is full.
bool is_full() const { return (base_type::free() < sizeof(Tp)); }
template <typename... Args>
auto emplace(Args&&... args)
{
Tp _obj{std::forward<Args>(args)...};
return write(&_obj);
}
using base_type::get_use_mmap;
using base_type::load;
using base_type::save;
using base_type::set_use_mmap;
std::string as_string() const
{
std::ostringstream ss{};
size_t _w = std::log10(base_type::capacity()) + 1;
ss << std::boolalpha << std::right << "data size: " << std::setw(_w) << data_size()
<< " B, is_initialized: " << std::setw(5) << is_initialized()
<< ", is_empty: " << std::setw(5) << is_empty() << ", is_full: " << std::setw(5)
<< is_full() << ", capacity: " << std::setw(_w) << capacity()
<< ", count: " << std::setw(_w) << count() << ", free: " << std::setw(_w) << free()
<< ", raw capacity: " << std::setw(_w) << base_type::capacity()
<< " B, raw count: " << std::setw(_w) << base_type::count()
<< " B, raw free: " << std::setw(_w) << base_type::free()
<< " B, pointer: " << std::setw(15) << base_type::m_ptr
<< ", raw read count: " << std::setw(_w) << base_type::m_read_count
<< ", raw write count: " << std::setw(_w) << base_type::m_write_count;
return ss.str();
}
friend std::ostream& operator<<(std::ostream& os, const atomic_ring_buffer& obj)
{
return os << obj.as_string();
}
};
//
template <typename Tp>
size_t
atomic_ring_buffer<Tp>::get_items_per_page()
{
return std::max<size_t>(units::get_page_size() / sizeof(Tp), 1);
}
//
template <typename Tp>
atomic_ring_buffer<Tp>::atomic_ring_buffer(const atomic_ring_buffer<Tp>& rhs)
: base_type{rhs}
{
size_t _n = rhs.count();
char* _end = static_cast<char*>(rhs.m_ptr) + rhs.m_size;
for(size_t i = 0; i < _n; ++i)
{
char* _addr = static_cast<char*>(rhs.read_ptr(m_read_count)) + (i * sizeof(Tp));
if((_addr + sizeof(Tp)) > _end) _addr = static_cast<char*>(rhs.m_ptr);
Tp* _in = static_cast<Tp*>(static_cast<void*>(_addr));
write(_in);
}
}
//
template <typename Tp>
atomic_ring_buffer<Tp>&
atomic_ring_buffer<Tp>::operator=(const atomic_ring_buffer<Tp>& rhs)
{
if(this == &rhs) return *this;
base_type::operator=(rhs);
size_t _n = rhs.count();
char* _end = static_cast<char*>(rhs.m_ptr) + rhs.m_size;
for(size_t i = 0; i < _n; ++i)
{
char* _addr = static_cast<char*>(rhs.read_ptr(m_read_count)) + (i * sizeof(Tp));
if((_addr + sizeof(Tp)) > _end) _addr = static_cast<char*>(rhs.m_ptr);
Tp* _in = static_cast<Tp*>(static_cast<void*>(_addr));
write(_in);
}
return *this;
}
//
} // namespace container
} // namespace common
} // namespace rocprofiler
projects/rocprofiler-sdk/source/lib/common/container/record_header_buffer.cpp
+165
Wyświetl plik
@@ -0,0 +1,165 @@
// MIT License
//
// Copyright (c) 2023 ROCm Developer Tools
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
#include "lib/common/container/record_header_buffer.hpp"
#include <rocprofiler/rocprofiler.h>
#include <algorithm>
#include <atomic>
#include <new>
namespace rocprofiler::common::container
{
namespace
{
// record_header_buffer RAII locker
struct rhb_raii_lock
{
explicit rhb_raii_lock(record_header_buffer& _rhb)
: m_rhb{_rhb}
{
m_rhb.lock();
}
~rhb_raii_lock() { m_rhb.unlock(); }
record_header_buffer& m_rhb;
};
} // namespace
record_header_buffer::record_header_buffer(size_t num_bytes) { allocate(num_bytes); }
record_header_buffer::record_header_buffer(record_header_buffer&& _rhs) noexcept
{
this->operator=(std::move(_rhs));
}
record_header_buffer&
record_header_buffer::operator=(record_header_buffer&& _rhs) noexcept
{
if(this != &_rhs)
{
auto _lk = rhb_raii_lock{_rhs};
m_index = _rhs.m_index.load(std::memory_order_relaxed);
m_buffer = std::move(_rhs.m_buffer);
m_headers = std::move(_rhs.m_headers);
_rhs.reset();
}
return *this;
}
bool
record_header_buffer::allocate(size_t num_bytes)
{
if(m_buffer.is_initialized()) return false;
auto _lk = rhb_raii_lock{*this};
m_buffer.init(num_bytes);
m_headers.resize(m_buffer.capacity(), rocprofiler_record_header_t{0, nullptr});
return true;
}
record_header_buffer::record_ptr_vec_t
record_header_buffer::get_record_headers(size_t _n)
{
auto _lk = rhb_raii_lock{*this};
auto _sz = m_index.load(std::memory_order_relaxed);
if(_n > _sz) _n = _sz;
auto _ret = record_ptr_vec_t{};
_ret.reserve(_n);
for(size_t i = 0; i < _n; ++i)
{
if(auto& itr = m_headers.at(i); itr.kind > 0 && itr.payload != nullptr)
_ret.emplace_back(&itr);
}
return _ret;
}
size_t
record_header_buffer::clear()
{
auto _lk = rhb_raii_lock{*this};
auto _n = m_index.load(std::memory_order_acquire);
{
auto _sz = m_buffer.capacity();
if(!m_buffer.clear(std::nothrow_t{})) return 0;
std::for_each(m_headers.begin(), m_headers.end(), [](auto& itr) {
itr = rocprofiler_record_header_t{0, nullptr};
});
m_headers.resize(_sz, rocprofiler_record_header_t{0, nullptr});
m_index.store(0, std::memory_order_release);
}
return _n;
}
size_t
record_header_buffer::reset()
{
auto _lk = rhb_raii_lock{*this};
auto _n = m_index.load(std::memory_order_acquire);
m_buffer.destroy();
m_buffer.clear();
m_headers.clear();
m_index.store(0, std::memory_order_release);
return _n;
}
void
record_header_buffer::save(std::fstream& _fs)
{
auto _lk = rhb_raii_lock{*this};
auto _idx = m_index.load(std::memory_order_acquire);
auto _sz = m_headers.size();
_fs.write(reinterpret_cast<char*>(&_idx), sizeof(_idx));
_fs.write(reinterpret_cast<char*>(&_sz), sizeof(_sz));
_fs.write(reinterpret_cast<char*>(m_headers.data()), sizeof(rocprofiler_record_header_t) * _sz);
m_buffer.save(_fs);
}
void
record_header_buffer::load(std::fstream& _fs)
{
auto _lk = rhb_raii_lock{*this};
{
auto _idx = size_t{0};
_fs.read(reinterpret_cast<char*>(&_idx), sizeof(_idx));
m_index.store(_idx, std::memory_order_release);
}
{
auto _sz = size_t{0};
_fs.read(reinterpret_cast<char*>(&_sz), sizeof(_sz));
m_headers.resize(_sz);
_fs.read(reinterpret_cast<char*>(m_headers.data()),
sizeof(rocprofiler_record_header_t) * _sz);
}
m_buffer.load(_fs);
}
} // namespace rocprofiler::common::container
projects/rocprofiler-sdk/source/lib/common/container/record_header_buffer.hpp
+220
Wyświetl plik
@@ -0,0 +1,220 @@
// MIT License
//
// Copyright (c) 2023 ROCm Developer Tools
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
#pragma once
#include <rocprofiler/rocprofiler.h>
#include "lib/common/container/ring_buffer.hpp"
#include <atomic>
#include <limits>
#include <mutex>
#include <vector>
namespace rocprofiler
{
namespace common
{
namespace container
{
/// @brief this struct stores all the record information in an ring_buffer.
/// It is thread-safe to have multiple threads emplace records into the buffer.
struct record_header_buffer
{
using base_buffer_t = base::ring_buffer;
using record_vec_t = std::vector<rocprofiler_record_header_t>;
using record_ptr_vec_t = std::vector<rocprofiler_record_header_t*>;
record_header_buffer() = default;
explicit record_header_buffer(size_t nbytes);
~record_header_buffer() = default;
record_header_buffer(const record_header_buffer&) = delete;
record_header_buffer(record_header_buffer&&) noexcept;
record_header_buffer& operator=(const record_header_buffer&) = delete;
record_header_buffer& operator =(record_header_buffer&&) noexcept;
// allocate the buffer if it is not already allocated. Will return false if buffer is already
// allocated
bool allocate(size_t nbytes);
// return whether the buffer has been allocated
bool is_allocated() const;
/// place an object in the buffer using its typeid hash code
template <typename Tp>
bool emplace(Tp&);
/// place an object in the buffer using the specified numerical identifier
template <typename Tp>
bool emplace(uint64_t, Tp&);
/// this function will return a vector of pointers to the record headers
/// at the time of invocation.
record_ptr_vec_t get_record_headers(size_t _n = std::numeric_limits<size_t>::max());
/// prevent emplace
void lock();
/// try to re-enable emplace
void unlock();
/// check if emplace is available
bool is_locked() const;
/// restores to original empty state
size_t clear();
/// binary save to file
void save(std::fstream& _fs);
/// binary load from file
void load(std::fstream& _fs);
/// full deallocation
size_t reset();
/// the number of header entries
auto size() const;
/// the number of bytes in the buffer
auto capacity() const;
/// the number of used bytes in the buffer
auto count() const;
/// the number of free bytes in the buffer
auto free() const;
/// true if no bytes are used in the buffer
auto is_empty() const;
/// true if all the bytes are used in the buffer or there is no buffer allocation
auto is_full() const;
private:
std::atomic<int32_t> m_locked = {0};
std::atomic<size_t> m_index = {};
base_buffer_t m_buffer = {};
record_vec_t m_headers = {};
};
inline bool
record_header_buffer::is_locked() const
{
return m_locked.load(std::memory_order_acquire) > 0;
}
inline void
record_header_buffer::lock()
{
m_locked.fetch_add(1, std::memory_order_release);
}
inline void
record_header_buffer::unlock()
{
m_locked.fetch_add(-1, std::memory_order_release);
}
inline bool
record_header_buffer::is_allocated() const
{
return m_buffer.is_initialized();
}
inline auto
record_header_buffer::size() const
{
return m_index.load(std::memory_order_acquire);
}
inline auto
record_header_buffer::capacity() const
{
return std::min<size_t>(m_headers.size(), m_buffer.capacity());
}
inline auto
record_header_buffer::count() const
{
return m_buffer.count();
}
inline auto
record_header_buffer::free() const
{
return m_buffer.free();
}
inline auto
record_header_buffer::is_empty() const
{
return m_buffer.is_empty() || m_headers.empty();
}
inline auto
record_header_buffer::is_full() const
{
return m_buffer.is_full() || size() == m_headers.size();
}
template <typename Tp>
bool
record_header_buffer::emplace(uint64_t _kind, Tp& _v)
{
if(is_locked() || m_headers.empty()) return false;
// request N bytes in the buffer (where N=sizeof(Tp)) and if
// available, copy _v into the buffer region
auto _create_record = [](auto& _buf, auto& _data) {
constexpr auto buffer_sz = sizeof(Tp);
void* _ptr = _buf.request(buffer_sz, false);
if(_ptr) new(_ptr) Tp{_data};
return _ptr;
};
auto _addr = _create_record(m_buffer, _v);
if(_addr)
{
// if there is space in the buffer, atomically get an index
// for where the header record should be placed.
// NOTE: m_headers was resized to be large enough to accomodate
// sizeof(Tp) == 1 for every entry in buffer
auto _idx = m_index++;
m_headers.at(_idx) = rocprofiler_record_header_t{_kind, _addr};
}
return (_addr != nullptr);
}
template <typename Tp>
bool
record_header_buffer::emplace(Tp& _v)
{
// if enumerations are not used, use the typeid hash code
return emplace(typeid(Tp).hash_code(), _v);
}
} // namespace container
} // namespace common
} // namespace rocprofiler
projects/rocprofiler-sdk/source/lib/common/container/ring_buffer.cpp
+99 -117
Wyświetl plik
@@ -21,12 +21,18 @@
// SOFTWARE.
#include "ring_buffer.hpp"
#include "lib/common/environment.hpp"
#include "lib/common/units.hpp"
#include <sys/mman.h>
#include <atomic>
#include <cerrno>
#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <new>
namespace rocprofiler
{
@@ -36,56 +42,28 @@ namespace container
{
namespace base
{
ring_buffer::ring_buffer(size_t _size, bool _use_mmap)
{
set_use_mmap(_use_mmap);
init(_size);
}
ring_buffer::~ring_buffer() { destroy(); }
ring_buffer::ring_buffer(const ring_buffer& rhs)
: m_use_mmap{rhs.m_use_mmap}
, m_use_mmap_explicit{rhs.m_use_mmap_explicit}
{
init(rhs.m_size);
}
ring_buffer::ring_buffer(ring_buffer&& rhs) noexcept
: m_init{rhs.m_init}
, m_use_mmap{rhs.m_use_mmap}
, m_use_mmap_explicit{rhs.m_use_mmap_explicit}
, m_ptr{rhs.m_ptr}
, m_size{rhs.m_size}
, m_read_count{rhs.m_read_count}
, m_write_count{rhs.m_write_count}
, m_read_count{rhs.m_read_count.load()}
, m_write_count{rhs.m_write_count.load()}
{
rhs.reset();
}
ring_buffer&
ring_buffer::operator=(const ring_buffer& rhs)
{
if(this == &rhs) return *this;
destroy();
m_use_mmap = rhs.m_use_mmap;
m_use_mmap_explicit = rhs.m_use_mmap_explicit;
init(rhs.m_size);
return *this;
}
ring_buffer&
ring_buffer::operator=(ring_buffer&& rhs) noexcept
{
if(this == &rhs) return *this;
destroy();
m_init = rhs.m_init;
m_use_mmap = rhs.m_use_mmap;
m_use_mmap_explicit = rhs.m_use_mmap_explicit;
m_ptr = rhs.m_ptr;
m_size = rhs.m_size;
m_read_count = rhs.m_read_count;
m_write_count = rhs.m_write_count;
m_init = rhs.m_init;
m_ptr = rhs.m_ptr;
m_size = rhs.m_size;
m_read_count = rhs.m_read_count.load();
m_write_count = rhs.m_write_count.load();
rhs.reset();
return *this;
}
@@ -94,7 +72,8 @@ void
ring_buffer::init(size_t _size)
{
if(m_init)
throw std::runtime_error("tim::base::ring_buffer::init(size_t) :: already initialized");
throw std::runtime_error("rocprofiler::common::container::base::ring_buffer::init(size_t) "
":: already initialized");
m_init = true;
@@ -115,14 +94,6 @@ ring_buffer::init(size_t _size)
m_read_count = 0;
m_write_count = 0;
if(!m_use_mmap_explicit) m_use_mmap = get_env("ROCPROFILER_USE_MMAP", m_use_mmap);
if(!m_use_mmap)
{
m_ptr = malloc(m_size * sizeof(char));
return;
}
// Map twice the buffer size.
if((m_ptr =
mmap(nullptr, m_size, PROT_READ | PROT_WRITE, MAP_ANONYMOUS | MAP_PRIVATE, -1, 0)) ==
@@ -130,7 +101,6 @@ ring_buffer::init(size_t _size)
{
destroy();
auto _err = errno;
// TIMEMORY_PRINTF_FATAL(stderr, "Error using mmap: %s\n", strerror(_err));
throw std::runtime_error(strerror(_err));
}
}
@@ -140,16 +110,9 @@ ring_buffer::destroy()
{
if(m_ptr && m_init)
{
if(!m_use_mmap)
{
::free(m_ptr);
}
else
{
// Unmap the mapped virtual memmory.
auto ret = munmap(m_ptr, m_size);
if(ret != 0) perror("munmap");
}
// Unmap the mapped virtual memmory.
auto ret = munmap(m_ptr, m_size);
if(ret != 0) perror("ring_buffer: munmap failed");
}
m_init = false;
m_size = 0;
@@ -158,21 +121,6 @@ ring_buffer::destroy()
m_ptr = nullptr;
}
void
ring_buffer::set_use_mmap(bool _v)
{
if(!m_init)
{
m_use_mmap = _v;
m_use_mmap_explicit = true;
}
else
{
throw std::runtime_error("tim::base::ring_buffer::set_use_mmap(bool) cannot be "
"called after initialization");
}
}
std::string
ring_buffer::as_string() const
{
@@ -186,81 +134,82 @@ ring_buffer::as_string() const
//
void*
ring_buffer::request(size_t _length)
ring_buffer::request(size_t _length, bool _wrap)
{
if(m_ptr == nullptr) return nullptr;
if(m_ptr == nullptr || m_size == 0) return nullptr;
// Make sure we don't put in more than there's room for, by writing no
// more than there is free.
if(_length > free())
throw std::runtime_error("heap-buffer-overflow :: ring buffer is full. read data "
"to avoid data corruption");
if(is_full()) return (_wrap) ? retrieve(_length) : nullptr;
// if write count is at the tail of buffer, bump to the end of buffer
auto _modulo = m_size - (m_write_count % m_size);
if(_modulo < _length) m_write_count += _modulo;
size_t _write_count = 0;
size_t _offset = 0;
do
{
// Make sure we don't put in more than there's room for, by writing no
// more than there is free.
if(_length > free()) return nullptr;
_offset = 0;
_write_count = m_write_count.load(std::memory_order_acquire);
auto _modulo = m_size - (_write_count % m_size);
if(_modulo < _length) _offset = _modulo;
} while(!m_write_count.compare_exchange_strong(
_write_count, _write_count + _length + _offset, std::memory_order_seq_cst));
// pointer in buffer
void* _out = write_ptr();
// Update write count
m_write_count += _length;
void* _out = write_ptr(_write_count);
return _out;
}
//
void*
ring_buffer::retrieve(size_t _length)
ring_buffer::retrieve(size_t _length) const
{
if(m_ptr == nullptr) return nullptr;
if(m_ptr == nullptr || m_size == 0) return nullptr;
// Make sure we don't put in more than there's room for, by writing no
// more than there is free.
if(_length > count()) throw std::runtime_error("ring buffer is empty");
// if read count is at the tail of buffer, bump to the end of buffer
auto _modulo = m_size - (m_read_count % m_size);
if(_modulo < _length) m_read_count += _modulo;
size_t _read_count = 0;
size_t _offset = 0;
do
{
if(_length > count()) return nullptr;
_offset = 0;
_read_count = m_read_count.load(std::memory_order_acquire);
auto _modulo = m_size - (_read_count % m_size);
if(_modulo < _length) _offset = _modulo;
} while(!m_read_count.compare_exchange_strong(
_read_count, _read_count + _length + _offset, std::memory_order_seq_cst));
// pointer in buffer
void* _out = read_ptr();
// Update write count
m_read_count += _length;
void* _out = read_ptr(_read_count);
return _out;
}
//
size_t
ring_buffer::rewind(size_t n) const
{
if(n > m_read_count) n = m_read_count;
m_read_count -= n;
return n;
}
//
void
ring_buffer::reset()
{
m_init = false;
m_ptr = nullptr;
m_size = 0;
m_read_count = 0;
m_write_count = 0;
m_init = false;
m_size = 0;
m_ptr = nullptr;
m_read_count.store(0);
m_write_count.store(0);
}
//
void
ring_buffer::save(std::fstream& _fs)
{
_fs.write(reinterpret_cast<char*>(&m_use_mmap), sizeof(m_use_mmap));
_fs.write(reinterpret_cast<char*>(&m_use_mmap_explicit), sizeof(m_use_mmap_explicit));
auto _read_count = m_read_count.load();
auto _write_count = m_write_count.load();
_fs.write(reinterpret_cast<char*>(&m_size), sizeof(m_size));
_fs.write(reinterpret_cast<char*>(&m_read_count), sizeof(m_read_count));
_fs.write(reinterpret_cast<char*>(&m_write_count), sizeof(m_write_count));
_fs.write(reinterpret_cast<char*>(&_read_count), sizeof(_read_count));
_fs.write(reinterpret_cast<char*>(&_write_count), sizeof(_write_count));
_fs.write(reinterpret_cast<char*>(m_ptr), m_size * sizeof(char));
}
//
@@ -270,16 +219,49 @@ ring_buffer::load(std::fstream& _fs)
{
destroy();
_fs.read(reinterpret_cast<char*>(&m_use_mmap), sizeof(m_use_mmap));
_fs.read(reinterpret_cast<char*>(&m_use_mmap_explicit), sizeof(m_use_mmap_explicit));
_fs.read(reinterpret_cast<char*>(&m_size), sizeof(m_size));
size_t _read_count = 0;
size_t _write_count = 0;
size_t _size = 0;
init(m_size);
if(!m_ptr) m_ptr = malloc(m_size);
_fs.read(reinterpret_cast<char*>(&_size), sizeof(_size));
_fs.read(reinterpret_cast<char*>(&m_read_count), sizeof(m_read_count));
_fs.read(reinterpret_cast<char*>(&m_write_count), sizeof(m_write_count));
init(_size);
if(!m_ptr) throw std::bad_alloc{};
_fs.read(reinterpret_cast<char*>(&_read_count), sizeof(_read_count));
_fs.read(reinterpret_cast<char*>(&_write_count), sizeof(_write_count));
_fs.read(reinterpret_cast<char*>(m_ptr), m_size * sizeof(char));
m_read_count.store(_read_count, std::memory_order_release);
m_write_count.store(_write_count, std::memory_order_release);
}
bool
ring_buffer::can_clear() const
{
auto _read_count = m_read_count.load(std::memory_order_acquire);
return (_read_count == 0);
}
bool
ring_buffer::clear()
{
if(!can_clear())
throw std::runtime_error(
"ring_buffer does not permit invoking clear() member function when the read "
"pointer is non-zero because this introduces thread-safety issues");
m_write_count.store(0, std::memory_order_release);
return true;
}
bool ring_buffer::clear(std::nothrow_t)
{
if(!can_clear()) return false;
m_write_count.store(0, std::memory_order_release);
return true;
}
} // namespace base
} // namespace container
projects/rocprofiler-sdk/source/lib/common/container/ring_buffer.hpp
+92 -167
Wyświetl plik
@@ -22,19 +22,19 @@
#pragma once
#include "lib/common/environment.hpp"
#include "lib/common/units.hpp"
#include <algorithm>
#include <atomic>
#include <cmath>
#include <cstddef>
#include <cstdlib>
#include <fstream>
#include <functional>
#include <iomanip>
#include <iostream>
#include <new>
#include <sstream>
#include <stdexcept>
#include <utility>
#include <vector>
namespace rocprofiler
{
@@ -47,7 +47,7 @@ struct ring_buffer;
//
namespace base
{
/// \struct tim::base::ring_buffer
/// \struct rocprofiler::common::container::base::ring_buffer
/// \brief Ring buffer implementation, with support for mmap as backend (Linux only).
struct ring_buffer
{
@@ -55,15 +55,10 @@ struct ring_buffer
friend struct container::ring_buffer;
ring_buffer() = default;
explicit ring_buffer(bool _use_mmap) { set_use_mmap(_use_mmap); }
explicit ring_buffer(size_t _size) { init(_size); }
ring_buffer(size_t _size, bool _use_mmap);
~ring_buffer();
ring_buffer(const ring_buffer&);
ring_buffer& operator=(const ring_buffer&);
ring_buffer(ring_buffer&&) noexcept;
ring_buffer& operator=(ring_buffer&&) noexcept;
@@ -79,6 +74,12 @@ struct ring_buffer
/// Destroy ring buffer.
void destroy();
/// Request a pointer for writing at least \param n bytes.
void* request(size_t n, bool wrap = true);
/// Retrieve a pointer for reading at least \param n bytes.
void* retrieve(size_t n) const;
/// Write class-type data to buffer (uses placement new).
template <typename Tp>
std::pair<size_t, Tp*> write(Tp* in, std::enable_if_t<std::is_class<Tp>::value, int> = 0);
@@ -93,24 +94,19 @@ struct ring_buffer
template <typename Tp>
Tp* request();
/// Request a pointer to an allocation for at least \param n bytes.
void* request(size_t n);
/// Read class-type data from buffer (uses placement new).
template <typename Tp>
std::pair<size_t, Tp*> read(Tp* out, std::enable_if_t<std::is_class<Tp>::value, int> = 0) const;
std::pair<size_t, Tp*> read(Tp* _dest,
std::enable_if_t<std::is_class<Tp>::value, int> = 0) const;
/// Read non-class-type data from buffer (uses memcpy).
template <typename Tp>
std::pair<size_t, Tp*> read(Tp* out,
std::pair<size_t, Tp*> read(Tp* _dest,
std::enable_if_t<!std::is_class<Tp>::value, int> = 0) const;
/// Retrieve a pointer to the head allocation (read).
template <typename Tp>
Tp* retrieve();
/// Retrieve a pointer to the head allocation of at least \param n bytes (read).
void* retrieve(size_t n);
Tp* retrieve() const;
/// Returns number of bytes currently held by the buffer.
size_t count() const { return (m_write_count - m_read_count); }
@@ -124,42 +120,52 @@ struct ring_buffer
/// Returns if the buffer is full.
bool is_full() const { return (count() == m_size); }
/// Rewind the read position n bytes
size_t rewind(size_t n) const;
/// explicitly configure to use mmap if avail
void set_use_mmap(bool);
/// query whether using mmap
bool get_use_mmap() const { return m_use_mmap; }
/// Display info about buffer
std::string as_string() const;
/// save the entire buffer to a filestream
void save(std::fstream& _fs);
/// load the entire buffer from a filestream
void load(std::fstream& _fs);
friend std::ostream& operator<<(std::ostream& os, const ring_buffer& obj)
{
return os << obj.as_string();
}
/// query whether the read pointer is zero and thus clearing is supported
bool can_clear() const;
/// reset the read and write pointer to their initial values.
/// effectively, wiping and existing memory. Please note,
/// this should be used with care in a double buffer system
/// where you are not actually using the read pointer.
/// If the read pointer is non-zero, this will throw an exception
bool clear();
/// reset the read and write pointer to their initial values.
/// effectively, wiping and existing memory. Please note,
/// this should be used with care in a double buffer system
/// where you are not actually using the read pointer.
bool clear(std::nothrow_t);
private:
/// Returns the current write pointer.
void* write_ptr() const { return static_cast<char*>(m_ptr) + (m_write_count % m_size); }
void* write_ptr(size_t _write_count) const
{
return static_cast<char*>(m_ptr) + (_write_count % m_size);
}
/// Returns the current read pointer.
void* read_ptr() const { return static_cast<char*>(m_ptr) + (m_read_count % m_size); }
void* read_ptr(size_t _read_count) const
{
return static_cast<char*>(m_ptr) + (_read_count % m_size);
}
void reset();
private:
bool m_init = false;
bool m_use_mmap = true;
bool m_use_mmap_explicit = false;
void* m_ptr = nullptr;
size_t m_size = 0;
mutable size_t m_read_count = 0;
size_t m_write_count = 0;
bool m_init = false;
void* m_ptr = nullptr;
size_t m_size = 0;
mutable std::atomic<size_t> m_read_count = 0;
std::atomic<size_t> m_write_count = 0;
};
//
template <typename Tp>
@@ -168,28 +174,18 @@ ring_buffer::write(Tp* in, std::enable_if_t<std::is_class<Tp>::value, int>)
{
if(in == nullptr || m_ptr == nullptr) return {0, nullptr};
auto _length = sizeof(Tp);
auto _length = sizeof(Tp);
void* _out_p = request(_length);
// Make sure we don't put in more than there's room for, by writing no
// more than there is free.
if(_length > free())
throw std::runtime_error("heap-buffer-overflow :: ring buffer is full. read data "
"to avoid data corruption");
// if write count is at the tail of buffer, bump to the end of buffer
auto _modulo = m_size - (m_write_count % m_size);
if(_modulo < _length) m_write_count += _modulo;
// pointer in buffer
Tp* out = reinterpret_cast<Tp*>(write_ptr());
if(_out_p == nullptr) return {0, nullptr};
// Copy in.
new((void*) out) Tp{std::move(*in)};
new(_out_p) Tp{std::move(*in)};
// Update write count
m_write_count += _length;
// pointer in buffer
Tp* _out = reinterpret_cast<Tp*>(_out_p);
return {_length, out};
return {_length, _out};
}
//
template <typename Tp>
@@ -198,28 +194,18 @@ ring_buffer::write(Tp* in, std::enable_if_t<!std::is_class<Tp>::value, int>)
{
if(in == nullptr || m_ptr == nullptr) return {0, nullptr};
auto _length = sizeof(Tp);
auto _length = sizeof(Tp);
void* _out_p = request(_length);
// Make sure we don't put in more than there's room for, by writing no
// more than there is free.
if(_length > free())
throw std::runtime_error("heap-buffer-overflow :: ring buffer is full. read data "
"to avoid data corruption");
// if write count is at the tail of buffer, bump to the end of buffer
auto _modulo = m_size - (m_write_count % m_size);
if(_modulo < _length) m_write_count += _modulo;
// pointer in buffer
Tp* out = reinterpret_cast<Tp*>(write_ptr());
if(_out_p == nullptr) return {0, nullptr};
// Copy in.
memcpy((void*) out, in, _length);
memcpy(_out_p, in, _length);
// Update write count
m_write_count += _length;
// pointer in buffer
Tp* _out = reinterpret_cast<Tp*>(_out_p);
return {_length, out};
return {_length, _out};
}
//
template <typename Tp>
@@ -228,118 +214,70 @@ ring_buffer::request()
{
if(m_ptr == nullptr) return nullptr;
auto _length = sizeof(Tp);
// Make sure we don't put in more than there's room for, by writing no
// more than there is free.
if(_length > free())
throw std::runtime_error("heap-buffer-overflow :: ring buffer is full. read data "
"to avoid data corruption");
// if write count is at the tail of buffer, bump to the end of buffer
auto _modulo = m_size - (m_write_count % m_size);
if(_modulo < _length) m_write_count += _modulo;
// pointer in buffer
Tp* _out = reinterpret_cast<Tp*>(write_ptr());
// Update write count
m_write_count += _length;
return _out;
return request(sizeof(Tp));
}
//
template <typename Tp>
std::pair<size_t, Tp*>
ring_buffer::read(Tp* out, std::enable_if_t<std::is_class<Tp>::value, int>) const
ring_buffer::read(Tp* _dest, std::enable_if_t<std::is_class<Tp>::value, int>) const
{
if(is_empty() || out == nullptr) return {0, nullptr};
if(is_empty() || _dest == nullptr) return {0, nullptr};
auto _length = sizeof(Tp);
auto _length = sizeof(Tp);
void* _out_p = retrieve(_length);
// Make sure we do not read out more than there is actually in the buffer.
if(_length > count()) throw std::runtime_error("ring buffer is empty");
// if read count is at the tail of buffer, bump to the end of buffer
auto _modulo = m_size - (m_read_count % m_size);
if(_modulo < _length) m_read_count += _modulo;
if(_out_p == nullptr) return {0, nullptr};
// pointer in buffer
Tp* in = reinterpret_cast<Tp*>(read_ptr());
Tp* in = reinterpret_cast<Tp*>(_out_p);
// Copy out for BYTE, nothing magic here.
*out = *in;
// Update read count.
m_read_count += _length;
*_dest = *in;
return {_length, in};
}
//
template <typename Tp>
std::pair<size_t, Tp*>
ring_buffer::read(Tp* out, std::enable_if_t<!std::is_class<Tp>::value, int>) const
ring_buffer::read(Tp* _dest, std::enable_if_t<!std::is_class<Tp>::value, int>) const
{
if(is_empty() || out == nullptr) return {0, nullptr};
if(is_empty() || _dest == nullptr) return {0, nullptr};
auto _length = sizeof(Tp);
auto _length = sizeof(Tp);
void* _out_p = retrieve(_length);
if(_out_p == nullptr) return {0, nullptr};
// pointer in buffer
Tp* in = reinterpret_cast<Tp*>(_out_p);
using Up = typename std::remove_const<Tp>::type;
// Make sure we do not read out more than there is actually in the buffer.
if(_length > count()) throw std::runtime_error("ring buffer is empty");
// if read count is at the tail of buffer, bump to the end of buffer
auto _modulo = m_size - (m_read_count % m_size);
if(_modulo < _length) m_read_count += _modulo;
// pointer in buffer
Tp* in = reinterpret_cast<Tp*>(read_ptr());
// Copy out for BYTE, nothing magic here.
Up* _out = const_cast<Up*>(out);
Up* _out = const_cast<Up*>(_dest);
memcpy(_out, in, _length);
// Update read count.
m_read_count += _length;
return {_length, in};
}
//
template <typename Tp>
Tp*
ring_buffer::retrieve()
ring_buffer::retrieve() const
{
if(m_ptr == nullptr) return nullptr;
auto _length = sizeof(Tp);
// Make sure we don't put in more than there's room for, by writing no
// more than there is free.
if(_length > count()) throw std::runtime_error("ring buffer is empty");
// if read count is at the tail of buffer, bump to the end of buffer
auto _modulo = m_size - (m_read_count % m_size);
if(_modulo < _length) m_read_count += _modulo;
// pointer in buffer
Tp* _out = reinterpret_cast<Tp*>(read_ptr());
// Update write count
m_read_count += _length;
return _out;
return retrieve(sizeof(Tp));
}
//
} // namespace base
///
/// \struct rocprofiler::container::ring_buffer
/// \brief Ring buffer wrapper around \ref tim::base::ring_buffer for data of type Tp. If
/// the data object size is larger than the page size (typically 4KB), behavior is
/// undefined. During initialization, one requests a minimum number of objects and the
/// buffer will support that number of object + the remainder of the page, e.g. if a page
/// is 1000 bytes, the object is 1 byte, and the buffer is requested to support 1500
/// objects, then an allocation supporting 2000 objects (i.e. 2 pages) will be created.
//
/// \struct rocprofiler::common::container::ring_buffer
/// \brief Ring buffer wrapper around \ref rocprofiler::common::container::base::ring_buffer for
/// data of type Tp. If the data object size is larger than the page size (typically 4KB), behavior
/// is undefined. During initialization, one requests a minimum number of objects and the buffer
/// will support that number of object + the remainder of the page, e.g. if a page is 1000 bytes,
/// the object is 1 byte, and the buffer is requested to support 1500 objects, then an allocation
/// supporting 2000 objects (i.e. 2 pages) will be created.
template <typename Tp>
struct ring_buffer : private base::ring_buffer
{
@@ -350,18 +288,10 @@ struct ring_buffer : private base::ring_buffer
ring_buffer() = default;
~ring_buffer() = default;
explicit ring_buffer(bool _use_mmap)
: base_type{_use_mmap}
{}
explicit ring_buffer(size_t _size)
: base_type{_size * sizeof(Tp)}
{}
ring_buffer(size_t _size, bool _use_mmap)
: base_type{_size * sizeof(Tp), _use_mmap}
{}
ring_buffer(const ring_buffer&);
ring_buffer(ring_buffer&&) noexcept = default;
@@ -387,7 +317,7 @@ struct ring_buffer : private base::ring_buffer
Tp* write(Tp* in) { return base_type::write<Tp>(in).second; }
/// Read data from buffer. Return pointer to location of read
Tp* read(Tp* out) const { return base_type::read<Tp>(out).second; }
Tp* read(Tp* _dest) const { return base_type::read<Tp>(_dest).second; }
/// Get an uninitialized address at tail of buffer.
Tp* request() { return base_type::request<Tp>(); }
@@ -407,9 +337,6 @@ struct ring_buffer : private base::ring_buffer
/// Returns if the buffer is full.
bool is_full() const { return (base_type::free() < sizeof(Tp)); }
/// Rewinds the read pointer
size_t rewind(size_t n) const { return base_type::rewind(n); }
template <typename... Args>
auto emplace(Args&&... args)
{
@@ -417,10 +344,8 @@ struct ring_buffer : private base::ring_buffer
return write(&_obj);
}
using base_type::get_use_mmap;
using base_type::load;
using base_type::save;
using base_type::set_use_mmap;
std::string as_string() const
{
@@ -461,7 +386,7 @@ ring_buffer<Tp>::ring_buffer(const ring_buffer<Tp>& rhs)
char* _end = static_cast<char*>(rhs.m_ptr) + rhs.m_size;
for(size_t i = 0; i < _n; ++i)
{
char* _addr = static_cast<char*>(rhs.read_ptr()) + (i * sizeof(Tp));
char* _addr = static_cast<char*>(rhs.read_ptr(m_read_count)) + (i * sizeof(Tp));
if((_addr + sizeof(Tp)) > _end) _addr = static_cast<char*>(rhs.m_ptr);
Tp* _in = static_cast<Tp*>(static_cast<void*>(_addr));
write(_in);
@@ -479,7 +404,7 @@ ring_buffer<Tp>::operator=(const ring_buffer<Tp>& rhs)
char* _end = static_cast<char*>(rhs.m_ptr) + rhs.m_size;
for(size_t i = 0; i < _n; ++i)
{
char* _addr = static_cast<char*>(rhs.read_ptr()) + (i * sizeof(Tp));
char* _addr = static_cast<char*>(rhs.read_ptr(m_read_count)) + (i * sizeof(Tp));
if((_addr + sizeof(Tp)) > _end) _addr = static_cast<char*>(rhs.m_ptr);
Tp* _in = static_cast<Tp*>(static_cast<void*>(_addr));
write(_in);
projects/rocprofiler-sdk/source/lib/common/mpl.hpp
+95
Wyświetl plik
@@ -0,0 +1,95 @@
// MIT License
//
// Copyright (c) 2023 ROCm Developer Tools
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
#pragma once
#include <cstddef>
#include <tuple>
#include <type_traits>
namespace rocprofiler
{
namespace common
{
namespace mpl
{
// dummy tuple with low instantiation cost
template <typename... Tp>
struct type_list
{
static constexpr auto size() { return sizeof...(Tp); }
};
/// get the index of a type in expansion
template <typename Tp, typename Type>
struct index_of;
template <typename Tp, template <typename...> class Tuple, typename... Types>
struct index_of<Tp, Tuple<Tp, Types...>>
{
static constexpr size_t value = 0;
};
template <typename Tp, typename Head, template <typename...> class Tuple, typename... Tail>
struct index_of<Tp, Tuple<Head, Tail...>>
{
static constexpr size_t value = 1 + index_of<Tp, Tuple<Tail...>>::value;
};
/// get the index of a type in expansion
template <typename Tp>
struct size_of;
template <typename... Tp>
struct size_of<type_list<Tp...>>
{
static constexpr size_t value = sizeof...(Tp);
};
template <typename... Tp>
struct size_of<std::tuple<Tp...>>
{
static constexpr size_t value = sizeof...(Tp);
};
// check if type is in expansion
//
template <typename... Tp>
struct is_one_of
{
static constexpr bool value = false;
};
template <typename F, typename S, template <typename...> class Tuple, typename... T>
struct is_one_of<F, S, Tuple<T...>>
{
static constexpr bool value = std::is_same<F, S>::value || is_one_of<F, Tuple<T...>>::value;
};
template <typename F, typename S, template <typename...> class Tuple, typename... T>
struct is_one_of<F, Tuple<S, T...>>
{
static constexpr bool value = is_one_of<F, S, Tuple<T...>>::value;
};
} // namespace mpl
} // namespace common
} // namespace rocprofiler