DDDigi/src/DigiContainerCombine.cpp

0001 //==========================================================================
0002 //  AIDA Detector description implementation
0003 //--------------------------------------------------------------------------
0004 // Copyright (C) Organisation europeenne pour la Recherche nucleaire (CERN)
0005 // All rights reserved.
0006 //
0007 // For the licensing terms see $DD4hepINSTALL/LICENSE.
0008 // For the list of contributors see $DD4hepINSTALL/doc/CREDITS.
0009 //
0010 // Author     : M.Frank
0011 //
0012 //==========================================================================
0013
0014 // Framework include files
0015 #include <DD4hep/InstanceCount.h>
0016 #include <DDDigi/DigiData.h>
0017 #include <DDDigi/DigiKernel.h>
0018 #include <DDDigi/DigiContext.h>
0019 #include <DDDigi/DigiContainerCombine.h>
0020
0021 /// C/C++ include files
0022 #include <set>
0023
0024 using namespace dd4hep::digi;
0025
0026 /// Worker class to combine items of identical types as given by the input definition
0027 /**
0028  *  This is a utility class.
0029  *
0030  *  \author  M.Frank
0031  *  \version 1.0
0032  *  \ingroup DD4HEP_DIGITIZATION
0033  */
0034 class DigiContainerCombine::work_definition_t  {
0035 public:
0036   /// reference to parent
0037   const DigiContainerCombine* combine;
0038   /// Printout format string
0039   char        format[128];
0040   /// Local counters
0041   std::size_t cnt_conts    = 0;
0042   std::size_t cnt_depos    = 0;
0043   std::size_t cnt_parts    = 0;
0044   std::size_t cnt_hist     = 0;
0045   std::size_t cnt_response = 0;
0046   /// Work definition
0047   std::vector<Key>            keys;
0048   std::vector<std::any*>      work;
0049   std::set<Key::itemkey_type> items;
0050   /// Work done
0051   std::vector<Key>            used_keys;
0052
0053   /// Input arguments
0054   DigiEvent&                  event;
0055   DataSegment&                inputs;
0056   DataSegment&                outputs;
0057   std::mutex&                 used_keys_lock;
0058
0059   /// Initializing constructor
0060   work_definition_t(const DigiContainerCombine* c, DigiEvent& ev, DataSegment& in, DataSegment& out, std::mutex& used_lock)
0061     : combine(c), event(ev), inputs(in), outputs(out), used_keys_lock(used_lock)
0062   {
0063     keys.reserve(inputs.size());
0064     work.reserve(inputs.size());
0065     for( auto& i : inputs )   {
0066       Key  key(i.first);
0067       if ( combine->use_key(key) )   {
0068         keys.emplace_back(key);
0069         work.emplace_back(&i.second);
0070         items.insert(key.item());
0071       }
0072     }
0073     ::snprintf(format, sizeof(format),
0074                "%s Thread:%%2d+++ %%-32s Out-Mask: $%04X In-Mask: $%%04X Merged %%6ld %%s",
0075                event.id(), combine->m_deposit_mask);
0076     format[sizeof(format)-1] = 0;
0077   }
0078
0079   void used_keys_insert(Key key)   {
0080     std::lock_guard<std::mutex> lock(used_keys_lock);
0081     used_keys.emplace_back(key);
0082   }
0083
0084   /// Specialized deposit merger: implicitly assume identical item types are mapped sequentially
0085   template<typename OUT, typename IN> void merge_depos(OUT& output, IN& input, int thr)  {
0086     std::size_t cnt = 0;
0087     const auto& nam = input.name;
0088     Key::mask_type mask = input.key.mask();
0089     if ( output.data_type == SegmentEntry::UNKNOWN )
0090       output.data_type = input.data_type;
0091     else if ( output.data_type != input.data_type )
0092       combine->except("+++ Digitization does not allow to mix data of different type!");
0093     if ( combine->m_erase_combined )   {
0094       cnt = output.merge(std::move(input));
0095     }
0096     else   {
0097       cnt = output.insert(input);
0098     }
0099     combine->info(this->format, thr, nam.c_str(), mask, cnt, "deposits");
0100     this->cnt_depos += cnt;
0101     this->cnt_conts++;
0102   }
0103
0104   /// Generic deposit merger: implicitly assume identical item types are mapped sequentially
0105   void merge(const std::string& nam, size_t start, int thr)  {
0106     Key key = keys[start];
0107     DepositVector out(nam, combine->m_deposit_mask, SegmentEntry::UNKNOWN);
0108     for( std::size_t j = start; j < keys.size(); ++j )   {
0109       if ( keys[j].item() == key.item() )   {
0110         if ( DepositMapping* m = std::any_cast<DepositMapping>(work[j]) )
0111           merge_depos(out, *m, thr);
0112         else if ( DepositVector* v = std::any_cast<DepositVector>(work[j]) )
0113           merge_depos(out, *v, thr);
0114         else
0115           break;
0116         used_keys_insert(keys[j]);
0117       }
0118     }
0119     key.set_mask(combine->m_deposit_mask);
0120     outputs.emplace(std::move(key), std::move(out));
0121   }
0122
0123   /// Merge history records: implicitly assume identical item types are mapped sequentially
0124   void merge_hist(const std::string& nam, size_t start, int thr)  {
0125     std::size_t cnt;
0126     Key key = keys[start];
0127     DetectorHistory out(nam, combine->m_deposit_mask);
0128     for( std::size_t j=start; j < keys.size(); ++j )   {
0129       if ( keys[j].item() == key.item() )   {
0130         DetectorHistory* next = std::any_cast<DetectorHistory>(work[j]);
0131         if ( next )   {
0132           std::string next_name = next->name;
0133           cnt = (combine->m_erase_combined) ? out.merge(std::move(*next)) : out.insert(*next);
0134           combine->info(format, thr, next_name.c_str(), keys[j].mask(), cnt, "histories");
0135           used_keys_insert(keys[j]);
0136           cnt_hist += cnt;
0137           cnt_conts++;
0138         }
0139       }
0140     }
0141     key.set_mask(combine->m_deposit_mask);
0142     outputs.emplace(std::move(key), std::move(out));
0143   }
0144
0145   /// Merge detector rsponse records: implicitly assume identical item types are mapped sequentially
0146   void merge_response(const std::string& nam, size_t start, int thr)  {
0147     std::size_t cnt;
0148     Key key = keys[start];
0149     DetectorResponse out(nam, combine->m_deposit_mask);
0150     for( std::size_t j=start; j < keys.size(); ++j )   {
0151       if ( keys[j].item() == key.item() )   {
0152         DetectorResponse* next = std::any_cast<DetectorResponse>(work[j]);
0153         if ( next )   {
0154           std::string next_name = next->name;
0155           cnt = (combine->m_erase_combined) ? out.merge(std::move(*next)) : out.insert(*next);
0156           combine->info(format, thr, next_name.c_str(), keys[j].mask(), cnt, "responses");
0157           used_keys_insert(keys[j]);
0158           cnt_response += cnt;
0159           cnt_conts++;
0160         }
0161       }
0162     }
0163     key.set_mask(combine->m_deposit_mask);
0164     outputs.emplace(std::move(key), std::move(out));
0165   }
0166
0167   /// Merge particle objects: implicitly assume identical item types are mapped sequentially
0168   void merge_parts(const std::string& nam, size_t start, int thr)  {
0169     std::size_t cnt;
0170     Key key = keys[start];
0171     ParticleMapping out(nam, combine->m_deposit_mask);
0172     for( std::size_t j=start; j < keys.size(); ++j )   {
0173       if ( keys[j].item() == key.item() )   {
0174         ParticleMapping* next = std::any_cast<ParticleMapping>(work[j]);
0175         if ( next )   {
0176           std::string next_name = next->name;
0177           cnt = (combine->m_erase_combined) ? out.merge(std::move(*next)) : out.insert(*next);
0178           combine->info(format, thr, next_name.c_str(), keys[j].mask(), cnt, "particles");
0179           used_keys_insert(keys[j]);
0180           cnt_parts += cnt;
0181           cnt_conts++;
0182         }
0183       }
0184     }
0185     key.set_mask(combine->m_deposit_mask);
0186     outputs.emplace(std::move(key), std::move(out));
0187   }
0188
0189   /// Merge single item type
0190   void merge_one(Key::itemkey_type itm, int thr)   {
0191     const std::string& opt = combine->m_output_name_flag;
0192     for( std::size_t i=0; i < keys.size(); ++i )   {
0193       if ( keys[i].item() != itm )
0194         continue;
0195       /// Merge deposit mapping
0196       if ( DepositMapping* depom = std::any_cast<DepositMapping>(work[i]) )   {
0197         if ( combine->m_merge_deposits  ) merge(depom->name+opt, i, thr);
0198       }
0199       /// Merge deposit vector
0200       else if ( DepositVector* depov = std::any_cast<DepositVector>(work[i]) )   {
0201         if ( combine->m_merge_deposits  ) merge(depov->name+opt, i, thr);
0202       }
0203       /// Merge detector response
0204       else if ( DetectorResponse* resp = std::any_cast<DetectorResponse>(work[i]) )   {
0205         if ( combine->m_merge_response  ) merge_response(resp->name+opt, i, thr);
0206       }
0207       /// Merge response history
0208       else if ( DetectorHistory* hist = std::any_cast<DetectorHistory>(work[i]) )   {
0209         if ( combine->m_merge_history   ) merge_hist(hist->name+opt, i, thr);
0210       }
0211       /// Merge particle container
0212       else if ( ParticleMapping* parts = std::any_cast<ParticleMapping>(work[i]) )   {
0213         if ( combine->m_merge_particles ) merge_parts(parts->name+opt, i, thr);
0214       }
0215       break;
0216     }
0217   }
0218
0219   void merge_all()   {
0220     for( auto itm : items )
0221       merge_one(itm, 0);
0222   }
0223 };
0224
0225 template <> void DigiParallelWorker<DigiContainerCombine,
0226                                     DigiContainerCombine::work_definition_t,
0227                                     std::size_t>::execute(void* data) const  {
0228   calldata_t* args = reinterpret_cast<calldata_t*>(data);
0229   std::size_t cnt = 0;
0230   for( auto itm : args->items )  {
0231     if ( cnt == this->options )   {
0232       args->merge_one(itm, this->options);
0233       return;
0234     }
0235     ++cnt;
0236   }
0237 }
0238
0239 /// Standard constructor
0240 DigiContainerCombine::DigiContainerCombine(const DigiKernel& krnl, const std::string& nam)
0241   : DigiEventAction(krnl, nam)
0242 {
0243   declareProperty("containers",       m_containers);
0244   declareProperty("input_masks",      m_input_masks);
0245   declareProperty("input_segment",    m_input  = "inputs");
0246   declareProperty("output_segment",   m_output = "deposits");
0247   declareProperty("output_mask",      m_deposit_mask);
0248   declareProperty("output_name_flag", m_output_name_flag);
0249   declareProperty("erase_combined",   m_erase_combined  = false);
0250   declareProperty("merge_deposits",   m_merge_deposits  = true);
0251   declareProperty("merge_response",   m_merge_response  = true);
0252   declareProperty("merge_history",    m_merge_history   = true);
0253   declareProperty("merge_particles",  m_merge_particles = false);
0254   m_kernel.register_initialize(std::bind(&DigiContainerCombine::initialize,this));
0255   InstanceCount::increment(this);
0256 }
0257
0258 /// Default destructor
0259 DigiContainerCombine::~DigiContainerCombine() {
0260   InstanceCount::decrement(this);
0261 }
0262
0263 /// Initializing function: compute values which depend on properties
0264 void DigiContainerCombine::initialize()    {
0265   for ( const auto& cont : m_containers )   {
0266     Key key(cont, 0x0);
0267     m_cont_keys.emplace(key.item());
0268     if ( m_input_masks.empty() )   {
0269       m_keys.emplace(key.value());
0270       continue;
0271     }
0272     for ( int m : m_input_masks )    {
0273       key.set_mask(m);
0274       m_keys.emplace(key.value());
0275     }
0276   }
0277   if ( !m_output_name_flag.empty() )
0278     m_output_name_flag += '/';
0279 }
0280
0281 /// Initializing function: compute values which depend on properties
0282 void DigiContainerCombine::have_workers(size_t count)  const   {
0283   if ( m_workers.size() < count )   {
0284     auto group = m_workers.get_group(); // Lock worker group
0285     for(size_t i=m_workers.size(); i <= count; ++i)
0286       m_workers.insert(new Worker(nullptr, i));
0287   }
0288 }
0289
0290 /// Decide if a continer is to merged based on the properties
0291 bool DigiContainerCombine::use_key(Key key)  const   {
0292   const auto& m = m_input_masks;
0293   bool use = m.empty() || m_keys.empty();
0294   if ( !use )  {
0295     if ( !m_cont_keys.empty() )  {
0296       key.set_mask(0);
0297       return m_cont_keys.find(key.value()) != m_cont_keys.end();
0298     }
0299     return std::find(m.begin(), m.end(), key.mask()) != m.end();
0300   }
0301   return true;
0302 }
0303
0304 /// Combine selected containers to one single deposit container
0305 std::size_t DigiContainerCombine::combine_containers(DigiContext& context,
0306                                                      DigiEvent&   event,
0307                                                      DataSegment& inputs,
0308                                                      DataSegment& outputs)  const
0309 {
0310   work_definition_t def(this, event, inputs, outputs, m_used_keys_lock);
0311   if ( m_parallel )  {
0312     have_workers(def.items.size());
0313     m_kernel.submit(context, m_workers.get_group(), def.items.size(), &def);
0314   }
0315   else  {
0316     def.merge_all();
0317   }
0318   if ( m_erase_combined )   {
0319     inputs.erase(def.used_keys);
0320   }
0321   info("%s+++ Merged %ld particles and %ld deposits from segment '%s' to segment '%s'",
0322        event.id(), def.cnt_parts, def.cnt_depos, m_input.c_str(), m_output.c_str());
0323   return def.cnt_depos;
0324 }
0325
0326 /// Main functional callback
0327 void DigiContainerCombine::execute(DigiContext& context)  const    {
0328   auto& event    = *context.event;
0329   auto& inputs   = event.get_segment(m_input);
0330   auto& outputs  = event.get_segment(m_output);
0331   combine_containers(context, event, inputs, outputs);
0332 }