Files
geant4/source/analysis/g4tools/include/tools/wroot/branch
T
2019-12-06 15:12:28 +01:00

597 lines
18 KiB
Plaintext

// Copyright (C) 2010, Guy Barrand. All rights reserved.
// See the file tools.license for terms.
#ifndef tools_wroot_branch
#define tools_wroot_branch
#include "leaf"
#include "basket"
#include "../forit"
#include "imutex"
namespace tools {
namespace wroot {
class branch : public virtual ibo {
//static uint32 kDoNotProcess() {return (1<<10);} // Active bit for branches
#ifdef TOOLS_MEM
static const std::string& s_class() {
static const std::string s_v("tools::wroot::branch");
return s_v;
}
#endif
public: //ibo
virtual const std::string& store_cls() const {
static const std::string s_v("TBranch");
return s_v;
}
virtual bool stream(buffer& a_buffer) const {
unsigned int c;
if(!a_buffer.write_version(8,c)) return false;
if(!Named_stream(a_buffer,m_name,m_title)) return false;
if(!AttFill_stream(a_buffer)) return false;
int fEntryOffsetLen = 1000;
int fOffset = 0;
int fSplitLevel = 0;
if(!a_buffer.write(fCompress)) return false;
if(!a_buffer.write(m_basket_size)) return false;
if(!a_buffer.write(fEntryOffsetLen)) return false;
if(!a_buffer.write(m_write_basket)) return false;
int fEntryNumber = (int)m_entry_number;
if(!a_buffer.write(fEntryNumber)) return false;
if(!a_buffer.write(fOffset)) return false;
if(!a_buffer.write(m_max_baskets)) return false;
if(!a_buffer.write(fSplitLevel)) return false;
double fEntries = (double)m_entries;
if(!a_buffer.write(fEntries)) return false;
double fTotBytes = (double)m_tot_bytes;
double fZipBytes = (double)m_zip_bytes;
if(!a_buffer.write(fTotBytes)) return false;
if(!a_buffer.write(fZipBytes)) return false;
if(!m_branches.stream(a_buffer)) return false;
if(!m_leaves.stream(a_buffer)) return false;
if(!m_baskets.stream(a_buffer)) return false;
/*
{uint32 remaining_baskets = 0;
for(uint32 i=0;i<m_max_baskets;i++) {
if(m_baskets[i]) {
m_out << "debug : remaining basket : index " << i << ", bytes = " << m_baskets[i]->number_of_bytes()
<< std::endl;
remaining_baskets++;
}
}
m_out << "tools::wroot::branch::stream :"
<< " write_basket = " << m_write_basket
<< ", max_baskets = " << m_max_baskets
<< ", remaining_baskets = " << remaining_baskets << "."
<< std::endl;}
*/
// See TStreamerInfo::ReadBuffer::WriteBasicPointer
if(!a_buffer.write((char)1)) return false;
if(!a_buffer.write_fast_array(fBasketBytes,m_max_baskets)) return false;
if(!a_buffer.write((char)1)) return false;
if(!a_buffer.write_fast_array(fBasketEntry,m_max_baskets)) return false;
char isBigFile = 1;
//GB : begin
//if(fTree.directory().file().end()>RIO_START_BIG_FILE()) isBigFile = 2;
{for(uint32 i=0;i<m_max_baskets;i++) {
if(fBasketSeek[i]>START_BIG_FILE()) {
isBigFile = 2;
break;
}
}}
//GB : end
if(!a_buffer.write(isBigFile)) return false;
if(isBigFile==2) {
if(!a_buffer.write_fast_array(fBasketSeek,m_max_baskets)) return false;
} else {
for(uint32 i=0;i<m_max_baskets;i++) {
if(fBasketSeek[i]>START_BIG_FILE()) { //G.Barrand : add this test.
m_out << "tools::wroot::branch::stream :"
<< " attempt to write big Seek "
<< fBasketSeek[i] << " on 32 bits."
<< std::endl;
return false;
}
if(!a_buffer.write((seek32)fBasketSeek[i])) return false;
}
}
// fFileName
if(!a_buffer.write(std::string(""))) return false;
if(!a_buffer.set_byte_count(c)) return false;
return true;
}
public:
branch(std::ostream& a_out,bool a_byte_swap,uint32 a_compression,
seek a_seek_directory,const std::string& a_name,const std::string& a_title,bool a_verbose)
:m_out(a_out)
,m_byte_swap(a_byte_swap)
,m_verbose(a_verbose)
,m_seek_directory(a_seek_directory)
,m_name(a_name)
,m_title(a_title)
,fAutoDelete(false)
//,m_branches(true)
//,m_leaves(true)
,fCompress(a_compression)
,m_basket_size(32000)
,m_write_basket(0)
,m_entry_number(0)
,m_entries(0)
,m_tot_bytes(0)
,m_zip_bytes(0)
,m_max_baskets(10)
,fBasketBytes(0)
,fBasketEntry(0)
,fBasketSeek(0)
{
#ifdef TOOLS_MEM
mem::increment(s_class().c_str());
#endif
m_baskets.resize(m_max_baskets,0);
fBasketBytes = new uint32[m_max_baskets];
fBasketEntry = new uint32[m_max_baskets];
fBasketSeek = new seek[m_max_baskets];
{for(uint32 i=0;i<m_max_baskets;i++) {
m_baskets[i] = 0;
fBasketBytes[i] = 0;
fBasketEntry[i] = 0;
fBasketSeek[i] = 0;
}}
m_baskets[m_write_basket] = new basket(m_out,m_byte_swap,a_seek_directory,m_name,m_title,"TBasket",m_basket_size,m_verbose);
fBasketEntry[m_write_basket] = (uint32)m_entry_number;
}
virtual ~branch(){
// {for(uint32 i=0;i<=m_write_basket;i++) {
// m_out << " " << i << ", " << fBasketEntry[i] << std::endl;
// }}
delete [] fBasketBytes;
delete [] fBasketEntry;
delete [] fBasketSeek;
fBasketBytes = 0;
fBasketEntry = 0;
fBasketSeek = 0;
#ifdef TOOLS_MEM
mem::decrement(s_class().c_str());
#endif
}
protected:
branch(const branch& a_from)
:ibo(a_from)
,m_out(a_from.m_out)
,m_seek_directory(a_from.m_seek_directory)
{}
branch& operator=(const branch&){return *this;}
public:
const std::string& name() const {return m_name;}
const std::string& title() const {return m_title;}
uint64 entries() const {return m_entries;}
uint64 tot_bytes() const {return m_tot_bytes;}
void set_tot_bytes(uint64 a_value) {m_tot_bytes = a_value;}
uint64 zip_bytes() const {return m_zip_bytes;}
void set_zip_bytes(uint64 a_value) {m_zip_bytes = a_value;}
void set_basket_size(uint32 a_size) {m_basket_size = a_size;}
uint32 basket_size() const {return m_basket_size;}
////////////////////////////////////////////////
////////////////////////////////////////////////
////////////////////////////////////////////////
template <class T>
leaf_ref<T>* create_leaf_ref(const std::string& a_name,const T& a_ref){
leaf_ref<T>* lf = new leaf_ref<T>(m_out,a_name,a_ref);
m_leaves.push_back(lf);
return lf;
}
leaf_string_ref* create_leaf_string_ref(const std::string& a_name,const std::string& a_ref){
leaf_string_ref* lf = new leaf_string_ref(m_out,a_name,a_ref);
m_leaves.push_back(lf);
return lf;
}
////////////////////////////////////////////////
////////////////////////////////////////////////
////////////////////////////////////////////////
template <class T>
leaf<T>* create_leaf(const std::string& a_name){
leaf<T>* lf = new leaf<T>(m_out,a_name);
m_leaves.push_back(lf);
return lf;
}
leaf_string* create_leaf_string(const std::string& a_name){
leaf_string* lf = new leaf_string(m_out,a_name);
m_leaves.push_back(lf);
return lf;
}
leaf_element* create_leaf_element(const std::string& a_name,int a_id,int a_type){
leaf_element* lf = new leaf_element(m_out,a_name,a_id,a_type);
m_leaves.push_back(lf);
return lf;
}
// for row_wise ntuple :
template <class T>
leaf_std_vector_ref<T>* create_leaf_std_vector_ref(const std::string& a_name,
base_leaf& a_leaf_count,const std::vector<T>& a_ref) {
leaf_std_vector_ref<T>* lf = new leaf_std_vector_ref<T>(m_out,a_name,a_leaf_count,a_ref);
m_leaves.push_back(lf);
return lf;
}
const std::vector<base_leaf*>& leaves() const {return m_leaves;}
void reset() {
m_baskets.clear_objs();
delete [] fBasketBytes;
delete [] fBasketEntry;
delete [] fBasketSeek;
fBasketBytes = 0;
fBasketEntry = 0;
fBasketSeek = 0;
m_max_baskets = 10;
m_write_basket = 0;
m_entry_number = 0;
m_entries = 0;
m_tot_bytes = 0;
m_zip_bytes = 0;
m_baskets.resize(m_max_baskets,0);
fBasketBytes = new uint32[m_max_baskets];
fBasketEntry = new uint32[m_max_baskets];
fBasketSeek = new seek[m_max_baskets];
{for(uint32 i=0;i<m_max_baskets;i++) {
m_baskets[i] = 0;
fBasketBytes[i] = 0;
fBasketEntry[i] = 0;
fBasketSeek[i] = 0;
}}
fBasketEntry[m_write_basket] = (uint32)m_entry_number;
m_baskets[m_write_basket] = new basket(m_out,m_byte_swap,m_seek_directory,m_name,m_title,"TBasket",m_basket_size,m_verbose);
{tools_vforit(branch*,m_branches,it) (*it)->reset();}
}
bool fill(ifile& a_file,uint32& a_nbytes,uint32& a_add_bytes,uint32& a_nout) {
a_nbytes = 0;
a_add_bytes = 0;
a_nout = 0;
if(m_write_basket>=m_max_baskets) {
m_out << "tools::wroot::branch::fill :"
<< " potential overflow : m_write_basket (" << m_write_basket << ")"
<< " >= m_max_baskets (" << m_max_baskets << ")."
<< std::endl;
return false;
}
//FIXME if (TestBit(kDoNotProcess)) return 0;
basket* bk = m_baskets[m_write_basket];
if(!bk) {
m_out << "tools::wroot::branch::fill :"
<< " m_baskets[m_write_basket] should not be null."
<< std::endl;
return false;
}
buffer& buf = bk->datbuf();
buf.reset_objs_map();
uint32 lold = buf.length();
bk->update(bk->key_length()+lold);
m_entries++;
m_entry_number++;
if(!fill_leaves(buf)) {
m_out << "tools::wroot::branch::fill :"
<< " fill_leaves() failed."
<< std::endl;
return false;
}
uint32 lnew = buf.length();
uint32 nbytes = lnew - lold;
uint32 nsize = 0;
uint32 add_bytes = 0;
uint32 nout = 0;
// Should we create a new basket?
// Compare expected next size with m_basket_size.
if((lnew+2*nsize+nbytes)>=m_basket_size) {
if(!bk->write_on_file(a_file,m_write_basket,nout)) {//it set bk m_seek_key, set a_file.m_END.
//Does a a_file.set_END().
m_out << "tools::wroot::branch::fill :"
<< " basket.write_on_file() failed."
<< std::endl;
return false;
}
fBasketBytes[m_write_basket] = bk->number_of_bytes();
//fBasketEntry[m_write_basket] = // not done here since we do not know here the first entry.
fBasketSeek[m_write_basket] = bk->seek_key();
add_bytes = bk->object_size() + bk->key_length();
delete bk;
m_baskets[m_write_basket] = 0;
m_tot_bytes += add_bytes;
m_zip_bytes += nout;
m_write_basket++;
if(!check_alloc_fBasketXxx()) return false;
m_baskets[m_write_basket] = new basket(m_out,m_byte_swap,
m_seek_directory,m_name,m_title,"TBasket",m_basket_size,
m_verbose);
fBasketEntry[m_write_basket] = (uint32)m_entry_number;
}
a_nbytes = nbytes;
a_add_bytes = add_bytes;
a_nout = nout;
return true;
}
///////////////////////////////////////////////////////////////////////////
/// for parallelization : /////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////
bool add_basket(ifile& a_file,basket& a_basket,uint32& a_add_bytes,uint32& a_nout) {
//done on a (main) ntuple/branch.
if(m_write_basket>=m_max_baskets) {
m_out << "tools::wroot::branch::add_basket :"
<< " potential overflow : m_write_basket (" << m_write_basket << ")"
<< " >= m_max_baskets (" << m_max_baskets << ")."
<< std::endl;
return false;
}
uint32 nout;
if(!a_basket.write_on_file(a_file,m_write_basket,nout)) { //a_file is the main file (a true file).
//the upper set a_basket.m_seek_key, set a_file.m_END. Does a a_file.set_END().
m_out << "tools::wroot::branch::add_basket :"
<< " basket.write_on_file() failed."
<< std::endl;
return false;
}
fBasketBytes[m_write_basket] = a_basket.number_of_bytes();
fBasketEntry[m_write_basket] = (uint32)m_entry_number;
fBasketSeek[m_write_basket] = a_basket.seek_key();
m_entries += a_basket.nev();
m_entry_number += a_basket.nev();
uint32 add_bytes = a_basket.object_size() + a_basket.key_length();
delete m_baskets[m_write_basket];
m_baskets[m_write_basket] = 0;
m_write_basket++;
if(!check_alloc_fBasketXxx()) return false;
m_baskets[m_write_basket] = new basket(m_out,m_byte_swap,
m_seek_directory,m_name,m_title,"TBasket",m_basket_size,
m_verbose);
fBasketEntry[m_write_basket] = (uint32)m_entry_number;
a_add_bytes = add_bytes;
a_nout = nout;
return true;
}
class iadd_basket {
public:
virtual ~iadd_basket() {}
public:
virtual bool add_basket(basket*) = 0;
};
bool pfill(iadd_basket& a_badd,uint32 a_nev) {
//done on a pntuple/branch.
//a_nbytes = 0;
//a_add_bytes = 0;
//a_nout = 0;
// this method uses always m_baskets[0].
basket* bk = m_baskets[m_write_basket];
if(!bk) {
m_out << "tools::wroot::branch::parallel_fill :"
<< " get_basket failed."
<< std::endl;
return false;
}
buffer& buf = bk->datbuf();
uint32 lold = buf.length();
bk->update(bk->key_length()+lold);
//m_entries++; //not used by parallel branches.
//m_entry_number++; //idem.
if(!fill_leaves(buf)) {
m_out << "tools::wroot::branch::parallel_fill :"
<< " fill_leaves() failed."
<< std::endl;
return false;
}
uint32 lnew = buf.length();
uint32 nbytes = lnew - lold;
// Should we create a new basket?
bool store_basket = false;
if(a_nev) {
store_basket = (bk->nev()>=a_nev); // trigger add_basket per number of entries reached.
} else {
store_basket = ((lnew+nbytes)>=m_basket_size); // trigger add_basket when reaching m_basket_size.
}
if(store_basket) {
if(!a_badd.add_basket(bk)) { //we give ownership of bk.
m_out << "tools::wroot::branch::parallel_fill :"
<< " main_branch.add_basket() failed."
<< std::endl;
return false;
}
//delete bk; //no, it is deleted by the a_badd.add_basket().
bool main_branch_byte_swap = m_byte_swap; //NOTE : it assumes that pntuple/branch and main/branch have same byte_swap.
bool main_branch_verbose = m_verbose;
m_baskets[m_write_basket] = new basket(m_out,main_branch_byte_swap,
m_seek_directory,m_name,m_title,"TBasket",m_basket_size,
main_branch_verbose);
}
return true;
}
bool end_pfill(iadd_basket& a_badd) {
//done on a pntuple/branch.
basket* bk = m_baskets[m_write_basket];
if(!bk) {
m_out << "tools::wroot::branch::end_pfill :"
<< " m_baskets[m_write_basket] should not be null."
<< std::endl;
return false;
}
buffer& buf = bk->datbuf();
uint32 lold = buf.length();
if(lold) {
if(!a_badd.add_basket(bk)) { //we give ownership of bk.
m_out << "tools::wroot::branch::parallel_fill :"
<< " main_branch.add_basket() failed."
<< std::endl;
return false;
}
} else {
delete bk;
}
m_baskets[m_write_basket] = 0; // no need to recreate a basket since it is end of fill.
return true;
}
///////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////
protected:
bool check_alloc_fBasketXxx() {
if(m_write_basket>=m_max_baskets) {
//Increase BasketEntry buffer of a minimum of 10 locations
// and a maximum of 50 per cent of current size
uint32 newsize = mx<uint32>(10,uint32(1.5*m_max_baskets));
if(newsize>=START_BIG_FILE()) {
//we are going to have pb with uint32[] indexing.
m_out << "tools::wroot::branch::add_basket :"
<< " new size for fBasket[Bytes,Entry,Seek] arrays"
<< " is too close of 32 bits limit."
<< std::endl;
m_out << "tools::wroot::branch::add_basket :"
<< " you have to work with larger basket size."
<< std::endl;
return false;
}
m_baskets.resize(newsize,0);
if(!realloc<uint32>(fBasketBytes,newsize,m_max_baskets,true)) {
m_out << "tools::wroot::branch::add_basket : realloc failed." << std::endl;
return false;
}
if(!realloc<uint32>(fBasketEntry,newsize,m_max_baskets,true)){
m_out << "tools::wroot::branch::add_basket : realloc failed." << std::endl;
return false;
}
if(!realloc<seek>(fBasketSeek,newsize,m_max_baskets,true)){
m_out << "tools::wroot::branch::add_basket : realloc failed." << std::endl;
return false;
}
m_max_baskets = newsize;
}
m_baskets[m_write_basket] = 0;
fBasketBytes[m_write_basket] = 0;
fBasketEntry[m_write_basket] = 0;
fBasketSeek[m_write_basket] = 0;
return true;
}
virtual bool fill_leaves(buffer& a_buffer) { //virtual for branch_element.
tools_vforit(base_leaf*,m_leaves,it) {
if(!(*it)->fill_buffer(a_buffer)) return false;
}
return true;
}
protected:
std::ostream& m_out;
bool m_byte_swap;
bool m_verbose;
seek m_seek_directory;
obj_array<basket> m_baskets;
/// for parallelization :
public:
std::vector<basket*> m_parallel_baskets;
protected:
//Object
//uint32 m_bits;
//Named
std::string m_name;
std::string m_title;
bool fAutoDelete;
private: //be sure that sub branches are not used for the moment.
obj_array<branch> m_branches;
protected:
obj_array<base_leaf> m_leaves;
uint32 fCompress; // Compression level and algorithm
uint32 m_basket_size; // Initial Size of Basket Buffer
uint32 m_write_basket; // Last basket number written
uint64 m_entry_number; // Current entry number (last one filled in this branch)
uint64 m_entries; // Number of entries
uint64 m_tot_bytes;
uint64 m_zip_bytes;
uint32 m_max_baskets;
uint32* fBasketBytes; //[m_max_baskets] Lenght of baskets on file
uint32* fBasketEntry; //[m_max_baskets] Table of first entry in eack basket
seek* fBasketSeek; //[m_max_baskets] Addresses of baskets on file
};
}}
#endif