22 #include "tbb/concurrent_queue.h"
24 #include "ThreadLocalStorage.h"
26 #include "boost/algorithm/string/predicate.hpp"
34 #include <type_traits>
35 #include <unordered_map>
50 pool_string m_identifierStorage;
51 mutable std::optional<std::string> m_identifier;
55 using allocator_type = LocalAlloc<char>;
59 allocator_type alloc ) noexcept
60 : m_data{
std::move( data )}, m_addr{
std::move( addr )}, m_identifierStorage{id, alloc} {
61 if ( m_data ) m_data->setRegistry(
this );
64 Entry(
const Entry& ) =
delete;
65 Entry& operator=(
const Entry& rhs ) =
delete;
66 Entry( Entry&& rhs ) =
delete;
67 Entry& operator=( Entry&& rhs ) =
delete;
70 unsigned long addRef()
override {
return -1; }
71 unsigned long release()
override {
return -1; }
72 const name_type&
name()
const override {
74 if ( !m_identifier ) m_identifier.emplace( m_identifierStorage );
78 if ( !m_identifier ) m_identifier.emplace( m_identifierStorage );
81 std::string_view identifierView()
const {
return m_identifierStorage; }
86 m_addr.
reset( iAddr );
94 LocalAlloc<std::pair<const std::string_view, Entry>>>;
97 template <
typename Map = UnorderedMap>
99 LocalArena m_resource;
102 std::optional<Map> m_store{std::in_place, &m_resource};
103 static_assert( std::is_same_v<typename Map::key_type, std::string_view> );
108 auto [i, b] = m_store->try_emplace( k, k,
std::move( d ),
std::move( a ), &m_resource );
110 auto nh = m_store->extract( i );
111 nh.key() = nh.mapped().identifierView();
114 return r.position->second;
119 [[nodiscard]]
bool empty()
const {
return m_store->empty(); }
121 [[nodiscard]]
std::size_t used_bytes() const noexcept {
return m_resource.size(); }
122 [[nodiscard]]
std::size_t used_blocks() const noexcept {
return m_resource.num_blocks(); }
123 [[nodiscard]]
std::size_t used_buckets()
const {
return m_store->bucket_count(); }
124 [[nodiscard]]
std::size_t num_allocations() const noexcept {
return m_resource.num_allocations(); }
129 m_store.emplace( m_est_size, &m_resource );
137 const Entry* d =
find( k );
138 return d ? d->object() :
nullptr;
140 const Entry*
find( std::string_view k )
const noexcept {
141 auto i = m_store->find( k );
142 return i != m_store->end() ? &( i->second ) :
nullptr;
145 [[nodiscard]]
auto begin() const noexcept {
return m_store->begin(); }
146 [[nodiscard]]
auto end() const noexcept {
return m_store->end(); }
147 void clear() noexcept { m_store->clear(); }
148 auto erase( std::string_view k ) {
return m_store->erase( k ); }
149 template <
typename Predicate>
150 void erase_if( Predicate p ) {
151 auto i = m_store->begin();
152 auto end = m_store->end();
154 if ( std::invoke( p, std::as_const( *i ) ) )
155 i = m_store->erase( i );
169 std::string_view normalize_path( std::string_view
path, std::string_view
prefix ) {
172 if ( !
path.empty() &&
path.front() ==
'/' )
path.remove_prefix( 1 );
178 auto status = cnv.
createObj( &addr, pObject );
180 if ( status.isFailure() )
object.reset();
185 struct Partition final {
189 std::optional<Store<>> store;
190 int eventNumber = -1;
193 template <
typename T,
typename Mutex = std::recursive_mutex,
typename ReadLock = std::scoped_lock<Mutex>,
194 typename WriteLock = ReadLock>
200 template <
typename F>
201 decltype(
auto ) with_lock( F&& f ) {
202 WriteLock
lock{m_mtx};
205 template <
typename F>
206 decltype(
auto ) with_lock( F&& f )
const {
207 ReadLock
lock{m_mtx};
212 template <
typename Fun>
213 auto with_lock( Fun&& f ) {
214 return [f = std::forward<Fun>( f )](
auto& p ) -> decltype(
auto ) {
return p.with_lock( f ); };
217 TTHREAD_TLS( Synced<Partition>* ) s_current =
nullptr;
219 template <
typename Fun>
221 return s_current ? s_current->with_lock( std::forward<Fun>( f ) )
240 Gaudi::Property<bool> m_forceLeaves{
this,
"ForceLeaves",
false,
"force creation of default leaves on registerObject"};
253 if (
LIKELY( !m_printPoolStats ) )
return;
254 auto n_allocs = p.store->num_allocations();
256 m_storeEntries += p.store->size();
257 m_usedPoolSize += p.store->used_bytes();
258 m_storeBuckets += p.store->used_buckets();
259 m_usedPoolAllocations += p.store->used_blocks();
260 m_servedPoolAllocations += n_allocs;
269 p.store.emplace( m_estStoreBuckets, poolSize() );
285 "InhibitedPathPrefixes",
287 "Prefixes of TES locations that will not be loaded by the persistency service "};
289 this,
"FollowLinksToAncestors",
true,
290 "Load objects which reside in files other than the one corresponding to the root of the event store"};
294 using extends::extends;
296 CLID rootCLID()
const override;
300 size_t allocateStore(
int evtnumber )
override;
301 StatusCode freeStore(
size_t partition )
override;
302 size_t freeSlots()
override {
return m_freeSlots.unsafe_size(); }
303 StatusCode selectStore(
size_t partition )
override;
305 StatusCode clearStore(
size_t partition )
override;
306 StatusCode setNumberOfStores(
size_t slots )
override;
308 size_t getPartitionNumber(
int eventnumber )
const override;
311 return findObject(
id.fullKey(), pObject ).isSuccess();
319 StatusCode clearSubTree( std::string_view )
override;
338 StatusCode registerObject( std::string_view parentPath, std::string_view objectPath,
DataObject* pObject )
override;
341 StatusCode unregisterObject( std::string_view )
override;
346 return !obj ? unregisterObject(
sr )
374 Entry::setDataProviderSvc(
this );
375 extends::initialize().ignore();
376 if ( !setNumberOfStores( m_slots ).isSuccess() ) {
377 error() <<
"Cannot set number of slots" <<
endmsg;
382 for (
auto& synced_p : m_partitions ) {
383 synced_p.with_lock( [
this]( Partition& p ) { initStore( p ); } );
385 for (
size_t i = 0; i < m_slots; i++ ) { m_freeSlots.push( i ); }
386 selectStore( 0 ).ignore();
390 error() <<
"Cannot get IConversionSvc " << m_loader.value() <<
endmsg;
393 return setDataLoader(
loader,
nullptr );
396 if ( m_printPoolStats ) {
397 info() <<
"Mean memory pool usage: " << float( 1e-3f * m_usedPoolSize.
mean() ) <<
" KiB serving "
398 << float( m_servedPoolAllocations.
mean() ) <<
" allocations from " << float( m_usedPoolAllocations.mean() )
399 <<
" to produce " << float( m_storeEntries.mean() ) <<
" entries in " << float( m_storeBuckets.mean() )
402 setDataLoader(
nullptr,
nullptr ).ignore();
403 return extends::finalize();
421 size_t slot = std::string::npos;
423 assert( slot != std::string::npos );
425 [[maybe_unused]]
auto prev =
m_partitions[slot].with_lock(
426 [evtnumber]( Partition& p ) {
return std::exchange( p.eventNumber, evtnumber ); } );
427 assert( prev == -1 );
433 if ( slots <
size_t{1} ) {
434 error() <<
"Invalid number of slots (" << slots <<
")" <<
endmsg;
438 error() <<
"Too late to change the number of slots!" <<
endmsg;
448 with_lock( [eventnumber](
const Partition& p ) {
return p.eventNumber == eventnumber; } ) );
459 auto prev =
m_partitions[partition].with_lock( []( Partition& p ) {
return std::exchange( p.eventNumber, -1 ); } );
467 return m_partitions[partition].with_lock( [
this]( Partition& p ) {
475 return fwd( [&]( Partition& p ) {
476 p.store->erase_if( [
top](
const auto& value ) {
return boost::algorithm::starts_with( value.first,
top ); } );
482 return fwd( [
this]( Partition& p ) {
489 return fwd( [&]( Partition& p ) {
491 auto cmp = [](
const Entry* lhs,
const Entry* rhs ) {
return lhs->identifier() < rhs->identifier(); };
493 for (
const auto&
v : *p.store ) {
494 if ( boost::algorithm::starts_with(
v.second.identifier(),
top ) )
keys.insert( &
v.second );
496 auto k =
keys.begin();
497 while ( k !=
keys.end() ) {
498 const auto&
id = ( *k )->identifier();
500 bool accept = pAgent->
analyse(
const_cast<Entry*
>( *( k++ ) ),
level );
503 [&
id](
const auto& e ) { return boost::algorithm::starts_with( e->identifier(), id ); } );
511 debug() <<
"setRoot( " << root_path <<
", (DataObject*)" << (
void*)pObject <<
" )" <<
endmsg;
513 if (
UNLIKELY( !fwd( []( Partition& p ) {
515 } ).isSuccess() ) ) {
523 debug() <<
"setRoot( " << root_path <<
", (IOpaqueAddress*)" << rootAddr.get();
524 if ( rootAddr ) debug() <<
"[ " << rootAddr->par()[0] <<
", " << rootAddr->par()[1] <<
" ]";
525 debug() <<
" )" <<
endmsg;
527 if (
UNLIKELY( !fwd( []( Partition& p ) {
529 } ).isSuccess() ) ) {
532 if ( !rootAddr )
return Status::INVALID_OBJ_ADDR;
535 if ( !
object )
return Status::INVALID_OBJECT;
537 LocalArena dummy_arena{root_path.
size() + 1};
538 auto dummy = Entry{root_path, {}, {}, &dummy_arena};
539 object->setRegistry( &dummy );
540 rootAddr->setRegistry( &dummy );
541 auto status =
m_dataLoader->fillObjRefs( rootAddr.get(),
object.get() );
542 if ( status.isSuccess() ) {
543 auto pObject =
object.get();
554 if ( !addr )
return Status::INVALID_OBJ_ADDR;
556 debug() <<
"registerAddress( (IRegistry*)" << (
void*)pReg <<
", " <<
path <<
", (IOpaqueAddress*)" << addr.get()
557 <<
"[ " << addr->
par()[0] <<
", " << addr->
par()[1] <<
" ]"
563 debug() <<
"Attempt to load " << addr->
par()[1] <<
" from file " << addr->
par()[0] <<
" blocked -- different file"
568 [addrPath = addr->
par()[1]]( std::string_view
prefix ) {
569 return boost::algorithm::starts_with( addrPath, prefix );
572 debug() <<
"Attempt to load " << addr->
par()[1] <<
" from file " << addr->
par()[0] <<
" blocked -- path inhibited"
578 if ( !
object )
return Status::INVALID_OBJECT;
581 LocalArena dummy_arena{fullpath.size() + 1};
582 auto dummy = Entry{fullpath, {}, {}, &dummy_arena};
583 object->setRegistry( &dummy );
585 auto status =
m_dataLoader->fillObjRefs( addr.get(),
object.get() );
586 if ( !status.isSuccess() )
return status;
591 auto ptr =
object.
get();
592 debug() <<
"registerAddress: " << std::quoted( normalize_path( fullpath,
rootName() ) ) <<
" (DataObject*)"
596 fwd( [&]( Partition& p ) {
604 return parentPath.empty()
614 for (
auto i = dir.rfind(
'/' ); i != std::string_view::npos; i = dir.rfind(
'/' ) ) {
615 dir = dir.substr( 0, i );
616 if ( !p.store->find( dir ) ) {
617 if ( msgLevel( MSG::DEBUG ) ) {
618 debug() <<
"registerObject: adding directory " << std::quoted( dir ) << endmsg;
620 p.store->put( dir, std::unique_ptr<DataObject>{} );
625 auto ptr =
object.get();
626 debug() <<
"registerObject: " << std::quoted(
path ) <<
" (DataObject*)" <<
static_cast<void*
>( ptr )
635 return fwd( [&]( Partition& p ) {
639 debug() <<
"retrieveObject: " << std::quoted(
path ) <<
" (DataObject*)" << (
void*)pObject