CMS 3D CMS Logo

 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Properties Friends Macros Pages
StreamerInputSource.cc
Go to the documentation of this file.
2 
6 
17 
18 #include "zlib.h"
19 
28 
32 
33 #include <string>
34 #include <iostream>
35 #include <set>
36 
37 namespace edm {
38  namespace {
39  int const init_size = 1024*1024;
40  }
41 
43  ParameterSet const& pset,
44  InputSourceDescription const& desc):
45  RawInputSource(pset, desc),
46  tc_(getTClass(typeid(SendEvent))),
47  dest_(init_size),
48  xbuf_(TBuffer::kRead, init_size),
49  sendEvent_(),
50  eventPrincipalHolder_(),
51  adjustEventToNewProductRegistry_(false),
52  processName_(),
53  protocolVersion_(0U) {
54  }
55 
57 
58  // ---------------------------------------
59  std::unique_ptr<FileBlock>
61  return std::unique_ptr<FileBlock>(new FileBlock);
62  }
63 
64  void
65  StreamerInputSource::mergeIntoRegistry(SendJobHeader const& header, ProductRegistry& reg, BranchIDListHelper& branchIDListHelper, bool subsequent) {
66 
67  SendDescs const& descs = header.descs();
68 
69  FDEBUG(6) << "mergeIntoRegistry: Product List: " << std::endl;
70 
71  if (subsequent) {
72  ProductRegistry pReg;
73  pReg.updateFromInput(descs);
75  if (!mergeInfo.empty()) {
76  throw cms::Exception("MismatchedInput","RootInputFileSequence::previousEvent()") << mergeInfo;
77  }
78  branchIDListHelper.updateFromInput(header.branchIDLists());
79  } else {
80  declareStreamers(descs);
81  buildClassCache(descs);
83  if(!reg.frozen()) {
84  reg.updateFromInput(descs);
85  }
86  branchIDListHelper.updateFromInput(header.branchIDLists());
87  }
88  }
89 
90  void
92  for(auto const& item : descs) {
93  //pi->init();
94  std::string const real_name = wrappedClassName(item.className());
95  FDEBUG(6) << "declare: " << real_name << std::endl;
96  loadCap(real_name);
97  }
98  }
99 
100 
101  void
103  for(auto const& item : descs) {
104  //pi->init();
105  std::string const real_name = wrappedClassName(item.className());
106  FDEBUG(6) << "BuildReadData: " << real_name << std::endl;
107  doBuildRealData(real_name);
108  }
109  }
110 
115  std::auto_ptr<SendJobHeader>
117  if(initView.code() != Header::INIT)
118  throw cms::Exception("StreamTranslation","Registry deserialization error")
119  << "received wrong message type: expected INIT, got "
120  << initView.code() << "\n";
121 
122  //Get the process name and store if for Protocol version 4 and above.
123  if (initView.protocolVersion() > 3) {
124 
125  processName_ = initView.processName();
126  protocolVersion_ = initView.protocolVersion();
127 
128  FDEBUG(10) << "StreamerInputSource::deserializeRegistry processName = "<< processName_<< std::endl;
129  FDEBUG(10) << "StreamerInputSource::deserializeRegistry protocolVersion_= "<< protocolVersion_<< std::endl;
130  }
131 
132  // calculate the adler32 checksum
133  uint32_t adler32_chksum = cms::Adler32((char const*)initView.descData(),initView.descLength());
134  //std::cout << "Adler32 checksum of init message = " << adler32_chksum << std::endl;
135  //std::cout << "Adler32 checksum of init messsage from header = " << initView.adler32_chksum() << " "
136  // << "host name = " << initView.hostName() << " len = " << initView.hostName_len() << std::endl;
137  if((uint32)adler32_chksum != initView.adler32_chksum()) {
138  std::cerr << "Error from StreamerInputSource: checksum of Init registry blob failed "
139  << " chksum from registry data = " << adler32_chksum << " from header = "
140  << initView.adler32_chksum() << " host name = " << initView.hostName() << std::endl;
141  // skip event (based on option?) or throw exception?
142  }
143 
144  TClass* desc = getTClass(typeid(SendJobHeader));
145 
146  TBufferFile xbuf(TBuffer::kRead, initView.descLength(),
147  const_cast<char*>((char const*)initView.descData()),kFALSE);
148  RootDebug tracer(10,10);
149  std::auto_ptr<SendJobHeader> sd((SendJobHeader*)xbuf.ReadObjectAny(desc));
150 
151  if(sd.get() == nullptr) {
152  throw cms::Exception("StreamTranslation","Registry deserialization error")
153  << "Could not read the initial product registry list\n";
154  }
155 
156  sd->initializeTransients();
157  return sd;
158  }
159 
164  void
166  std::auto_ptr<SendJobHeader> sd = deserializeRegistry(initView);
168  if (subsequent) {
170  }
171  SendJobHeader::ParameterSetMap const& psetMap = sd->processParameterSet();
172  pset::Registry& psetRegistry = *pset::Registry::instance();
173  for (auto const& item : psetMap) {
174  ParameterSet pset(item.second.pset());
175  pset.setID(item.first);
176  psetRegistry.insertMapped(pset);
177  }
178  }
179 
183  void
185  if(eventView.code() != Header::EVENT)
186  throw cms::Exception("StreamTranslation","Event deserialization error")
187  << "received wrong message type: expected EVENT, got "
188  << eventView.code() << "\n";
189  FDEBUG(9) << "Decode event: "
190  << eventView.event() << " "
191  << eventView.run() << " "
192  << eventView.size() << " "
193  << eventView.adler32_chksum() << " "
194  << eventView.eventLength() << " "
195  << eventView.eventData()
196  << std::endl;
197  // uncompress if we need to
198  // 78 was a dummy value (for no uncompressed) - should be 0 for uncompressed
199  // need to get rid of this when 090 MTCC streamers are gotten rid of
200  unsigned long origsize = eventView.origDataSize();
201  unsigned long dest_size; //(should be >= eventView.origDataSize())
202 
203  uint32_t adler32_chksum = cms::Adler32((char const*)eventView.eventData(), eventView.eventLength());
204  //std::cout << "Adler32 checksum of event = " << adler32_chksum << std::endl;
205  //std::cout << "Adler32 checksum from header = " << eventView.adler32_chksum() << " "
206  // << "host name = " << eventView.hostName() << " len = " << eventView.hostName_len() << std::endl;
207  if((uint32)adler32_chksum != eventView.adler32_chksum()) {
208  std::cerr << "Error from StreamerInputSource: checksum of event data blob failed "
209  << " chksum from event = " << adler32_chksum << " from header = "
210  << eventView.adler32_chksum() << " host name = " << eventView.hostName() << std::endl;
211  // skip event (based on option?) or throw exception?
212  }
213  if(origsize != 78 && origsize != 0) {
214  // compressed
215  dest_size = uncompressBuffer(const_cast<unsigned char*>((unsigned char const*)eventView.eventData()),
216  eventView.eventLength(), dest_, origsize);
217  } else { // not compressed
218  // we need to copy anyway the buffer as we are using dest in xbuf
219  dest_size = eventView.eventLength();
220  dest_.resize(dest_size);
221  unsigned char* pos = (unsigned char*) &dest_[0];
222  unsigned char const* from = (unsigned char const*) eventView.eventData();
223  std::copy(from,from+dest_size,pos);
224  }
225  //TBuffer xbuf(TBuffer::kRead, dest_size,
226  // (char const*) &dest[0],kFALSE);
227  //TBuffer xbuf(TBuffer::kRead, eventView.eventLength(),
228  // (char const*) eventView.eventData(),kFALSE);
229  xbuf_.Reset();
230  xbuf_.SetBuffer(&dest_[0],dest_size,kFALSE);
231  RootDebug tracer(10,10);
232 
234  sendEvent_ = std::unique_ptr<SendEvent>((SendEvent*)xbuf_.ReadObjectAny(tc_));
236 
237  if(sendEvent_.get() == nullptr) {
238  throw cms::Exception("StreamTranslation","Event deserialization error")
239  << "got a null event from input stream\n";
240  }
242 
243  FDEBUG(5) << "Got event: " << sendEvent_->aux().id() << " " << sendEvent_->products().size() << std::endl;
244  if(runAuxiliary().get() == nullptr || runAuxiliary()->run() != sendEvent_->aux().run()) {
246  runAuxiliary->setProcessHistoryID(sendEvent_->processHistory().id());
247  setRunAuxiliary(runAuxiliary);
249  }
252  new LuminosityBlockAuxiliary(runAuxiliary()->run(), eventView.lumi(), sendEvent_->aux().time(), Timestamp::invalidTimestamp());
253  luminosityBlockAuxiliary->setProcessHistoryID(sendEvent_->processHistory().id());
254  setLuminosityBlockAuxiliary(luminosityBlockAuxiliary);
255  }
256  setEventCached();
257  }
258 
259  void
263  bool eventOK = eventPrincipal.adjustToNewProductRegistry(*productRegistry());
264  assert(eventOK);
266  }
267  EventSelectionIDVector ids(sendEvent_->eventSelectionIDs());
268  BranchListIndexes indexes(sendEvent_->branchListIndexes());
269  branchIDListHelper()->fixBranchListIndexes(indexes);
270  eventPrincipal.fillEventPrincipal(sendEvent_->aux(), processHistoryRegistry(), std::move(ids), std::move(indexes));
271  eventPrincipalHolder_.setEventPrincipal(&eventPrincipal);
272 
273  // no process name list handling
274 
275  SendProds& sps = sendEvent_->products();
276  for(auto& spitem : sps) {
277  FDEBUG(10) << "check prodpair" << std::endl;
278  if(spitem.desc() == nullptr)
279  throw cms::Exception("StreamTranslation","Empty Provenance");
280  FDEBUG(5) << "Prov:"
281  << " " << spitem.desc()->className()
282  << " " << spitem.desc()->productInstanceName()
283  << " " << spitem.desc()->branchID()
284  << std::endl;
285 
286  BranchDescription const branchDesc(*spitem.desc());
287  // This ProductProvenance constructor inserts into the entry description registry
288  ProductProvenance productProvenance(spitem.branchID(), *spitem.parents());
289 
290  if(spitem.prod() != nullptr) {
291  FDEBUG(10) << "addproduct next " << spitem.branchID() << std::endl;
292  eventPrincipal.putOnRead(branchDesc, spitem.prod(), productProvenance);
293  FDEBUG(10) << "addproduct done" << std::endl;
294  } else {
295  FDEBUG(10) << "addproduct empty next " << spitem.branchID() << std::endl;
296  eventPrincipal.putOnRead(branchDesc, spitem.prod(), productProvenance);
297  FDEBUG(10) << "addproduct empty done" << std::endl;
298  }
299  spitem.clear();
300  }
301 
302  FDEBUG(10) << "Size = " << eventPrincipal.size() << std::endl;
303  }
304 
313  unsigned int
314  StreamerInputSource::uncompressBuffer(unsigned char* inputBuffer,
315  unsigned int inputSize,
316  std::vector<unsigned char>& outputBuffer,
317  unsigned int expectedFullSize) {
318  unsigned long origSize = expectedFullSize;
319  unsigned long uncompressedSize = expectedFullSize*1.1;
320  FDEBUG(1) << "Uncompress: original size = " << origSize
321  << ", compressed size = " << inputSize
322  << std::endl;
323  outputBuffer.resize(uncompressedSize);
324  int ret = uncompress(&outputBuffer[0], &uncompressedSize,
325  inputBuffer, inputSize); // do not need compression level
326  //std::cout << "unCompress Return value: " << ret << " Okay = " << Z_OK << std::endl;
327  if(ret == Z_OK) {
328  // check the length against original uncompressed length
329  FDEBUG(10) << " original size = " << origSize << " final size = "
330  << uncompressedSize << std::endl;
331  if(origSize != uncompressedSize) {
332  std::cerr << "deserializeEvent: Problem with uncompress, original size = "
333  << origSize << " uncompress size = " << uncompressedSize << std::endl;
334  // we throw an error and return without event! null pointer
335  throw cms::Exception("StreamDeserialization","Uncompression error")
336  << "mismatch event lengths should be" << origSize << " got "
337  << uncompressedSize << "\n";
338  }
339  } else {
340  // we throw an error and return without event! null pointer
341  std::cerr << "deserializeEvent: Problem with uncompress, return value = "
342  << ret << std::endl;
343  throw cms::Exception("StreamDeserialization","Uncompression error")
344  << "Error code = " << ret << "\n ";
345  }
346  return (unsigned int) uncompressedSize;
347  }
348 
350  // called from an online streamer source to reset after a stop command
351  // so an enable command will work
354  assert(!eventCached());
355  reset();
356  }
357 
359  // Need to define a dummy setRun here or else the InputSource::setRun is called
360  // if we have a source inheriting from this and wants to define a setRun method
362  << "StreamerInputSource::setRun()\n"
363  << "Run number cannot be modified for this type of Input Source\n"
364  << "Contact a Storage Manager Developer\n";
365  }
366 
368 
370 
373  return eventPrincipal_ ? eventPrincipal_->getIt(id) : WrapperHolder();
374  }
375 
376  unsigned int
378  assert(eventPrincipal_ != nullptr);
379  return eventPrincipal_->transitionIndex();
380  }
381 
382  void
384  eventPrincipal_ = ep;
385  }
386 
387  void
390  }
391 }
void setID(ParameterSetID const &id)
ProcessHistoryRegistry const & processHistoryRegistry() const
Const accessor for process history registry.
Definition: InputSource.h:171
static void fillDescription(ParameterSetDescription &description)
size_t size() const
Definition: Principal.cc:238
uint32 lumi() const
Definition: EventMessage.cc:85
static void mergeIntoRegistry(SendJobHeader const &header, ProductRegistry &, BranchIDListHelper &, bool subsequent)
static Timestamp invalidTimestamp()
Definition: Timestamp.h:101
const uint8 * eventData() const
Definition: EventMessage.h:80
const uint8 * descData() const
Definition: InitMessage.h:91
std::string hostName() const
std::vector< BranchDescription > SendDescs
void doBuildRealData(const std::string &name)
Definition: ClassFiller.cc:31
std::map< ParameterSetID, ParameterSetBlob > ParameterSetMap
#define nullptr
bool registerProcessHistory(ProcessHistory const &processHistory)
void resetRunAuxiliary(bool isNewRun=true) const
Definition: InputSource.h:358
void setRefCoreStreamer(bool resetAll=false)
boost::shared_ptr< LuminosityBlockAuxiliary > luminosityBlockAuxiliary() const
Called by the framework to merge or insert lumi in principal cache.
Definition: InputSource.h:262
RunNumber_t run() const
Accessor for current run number.
Definition: InputSource.cc:595
StreamerInputSource(ParameterSet const &pset, InputSourceDescription const &desc)
#define FDEBUG(lev)
Definition: DebugMacros.h:18
virtual void setRun(RunNumber_t r)
TClass * getTClass(const std::type_info &ti)
Definition: ClassFiller.cc:78
const int init_size
std::vector< EventSelectionID > EventSelectionIDVector
uint32 run() const
Definition: EventMessage.cc:73
uint32 adler32_chksum() const
Definition: EventMessage.h:97
uint32 eventLength() const
Definition: EventMessage.h:82
static void declareStreamers(SendDescs const &descs)
uint32 adler32_chksum() const
Definition: InitMessage.h:94
std::vector< BranchListIndex > BranchListIndexes
static void buildClassCache(SendDescs const &descs)
boost::shared_ptr< BranchIDListHelper > branchIDListHelper() const
Accessor for branchIDListHelper.
Definition: InputSource.h:177
void setLuminosityBlockAuxiliary(LuminosityBlockAuxiliary *lbp)
Definition: InputSource.h:354
virtual WrapperHolder getIt(edm::ProductID const &id) const override
uint32 code() const
Definition: EventMessage.h:77
std::unique_ptr< SendEvent > sendEvent_
void deserializeEvent(EventMsgView const &eventView)
std::string hostName() const
Definition: InitMessage.cc:190
std::string merge(ProductRegistry const &other, std::string const &fileName, BranchDescription::MatchMode branchesMustMatch=BranchDescription::Permissive)
void setProcessHistoryID(ProcessHistoryID const &phid)
EventPrincipalHolder eventPrincipalHolder_
uint32 event() const
Definition: EventMessage.cc:79
std::vector< StreamedProduct > SendProds
void fillEventPrincipal(EventAuxiliary const &aux, ProcessHistoryRegistry const &processHistoryRegistry, DelayedReader *reader=nullptr)
unsigned int uint32
Definition: MsgTools.h:13
void setEventCached()
Called by the framework to merge or ached() const {return eventCached_;}.
Definition: InputSource.h:381
virtual unsigned int transitionIndex_() const override
void Adler32(char const *data, size_t len, uint32_t &a, uint32_t &b)
LuminosityBlockNumber_t luminosityBlock() const
Accessor for current luminosity block number.
Definition: InputSource.cc:601
uint32 protocolVersion() const
Definition: InitMessage.cc:110
double sd
ProductRegistry & productRegistryUpdate() const
Definition: InputSource.h:347
std::string wrappedClassName(std::string const &iFullName)
uint32 size() const
Definition: EventMessage.h:78
BranchIDLists const & branchIDLists() const
uint32 origDataSize() const
Definition: EventMessage.cc:91
bool updateFromInput(BranchIDLists const &bidlists)
void resetLuminosityBlockAuxiliary(bool isNewLumi=true) const
Definition: InputSource.h:362
boost::shared_ptr< RunAuxiliary > runAuxiliary() const
Called by the framework to merge or insert run in principal cache.
Definition: InputSource.h:259
boost::shared_ptr< ProductRegistry const > productRegistry() const
Accessor for product registry.
Definition: InputSource.h:168
std::auto_ptr< SendJobHeader > deserializeRegistry(InitMsgView const &initView)
void reset() const
Definition: InputSource.h:366
uint32 code() const
Definition: InitMessage.h:70
void setRunAuxiliary(RunAuxiliary *rp)
Definition: InputSource.h:350
bool adjustToNewProductRegistry(ProductRegistry const &reg)
Definition: Principal.cc:253
ProcessHistoryRegistry & processHistoryRegistryUpdate() const
Definition: InputSource.h:348
std::string processName() const
Definition: InitMessage.cc:127
void updateFromInput(ProductList const &other)
void putOnRead(BranchDescription const &bd, void const *product, ProductProvenance const &productProvenance)
void setProcessHistoryID(ProcessHistoryID const &phid)
Definition: RunAuxiliary.h:36
unsigned int RunNumber_t
Definition: EventRange.h:32
void adjustIndexesAfterProductRegistryAddition()
Definition: Principal.cc:831
uint32 descLength() const
Definition: InitMessage.h:90
volatile std::atomic< bool > shutdown_flag false
bool insertMapped(value_type const &v)
Definition: Registry.cc:37
void deserializeAndMergeWithRegistry(InitMsgView const &initView, bool subsequent=false)
SendDescs const & descs() const
static void fillDescription(ParameterSetDescription &description)
static unsigned int uncompressBuffer(unsigned char *inputBuffer, unsigned int inputSize, std::vector< unsigned char > &outputBuffer, unsigned int expectedFullSize)
static Registry * instance()
Definition: Registry.cc:14
virtual void read(EventPrincipal &eventPrincipal)
void loadExtraClasses()
Definition: ClassFiller.cc:43
virtual std::unique_ptr< FileBlock > readFile_()
bool eventCached() const
Definition: InputSource.h:379
std::vector< unsigned char > dest_
void loadCap(const std::string &name)
Definition: ClassFiller.cc:20