MatacqProducer.cc

Go to the documentation of this file.

#include "EventFilter/EcalRawToDigi/interface/MatacqProducer.h"
#include "EventFilter/EcalRawToDigi/src/Majority.h"
#include "FWCore/Framework/interface/Event.h"
#include "FWCore/Framework/interface/EventSetup.h"
#include "Geometry/Records/interface/IdealGeometryRecord.h"
#include "FWCore/Framework/interface/ESHandle.h"
#include "FWCore/ParameterSet/interface/ParameterSet.h"
#include "DataFormats/FEDRawData/interface/FEDRawData.h"
#include "DataFormats/FEDRawData/interface/FEDNumbering.h"
#include "DataFormats/EcalDigi/interface/EcalMatacqDigi.h"
#include "DataFormats/EcalDigi/interface/EcalDigiCollections.h"
#include <boost/algorithm/string.hpp>
#include <boost/format.hpp>

#include <iostream>
#include <memory>

#include <cstdio>

#include <fstream>
#include <iomanip>

#include "DataFormats/EcalDigi/interface/EcalMatacqDigi.h"

#include <sys/types.h>
#include <unistd.h>
#include <csignal>
#include <sys/stat.h>
#include <glob.h>

using namespace std;
using namespace boost;
using namespace edm;

// #undef LogInfo
// #define LogInfo(a) cout << "INFO " << a << ": "
// #undef LogWarning
// #define LogWarning(a) cout << "WARN " << a << ": "
// #undef LogDebug
// #define LogDebug(a) cout << "DBG " << a << ": "


//verbose mode for matacq event retrieval debugging:
//static const bool searchDbg = false;

//laser freq is 1 every 112 orbit => >80 orbit
const int MatacqProducer::orbitTolerance_ = 80; 

const MatacqProducer::stats_t MatacqProducer::stats_init = {0,0,0};

static std::string now(){
  struct timeval t;
  gettimeofday(&t, nullptr);
 
  char buf[256];
  strftime(buf, sizeof(buf), "%F %R %S s", localtime(&t.tv_sec));
  buf[sizeof(buf)-1] = 0;

  stringstream buf2;
  buf2 << buf << " " << ((t.tv_usec+500)/1000)  << " ms";

  return buf2.str();
}


MatacqProducer::MatacqProducer(const edm::ParameterSet& params):
  fileNames_(params.getParameter<std::vector<std::string> >("fileNames")),
  digiInstanceName_(params.getParameter<string>("digiInstanceName")),
  rawInstanceName_(params.getParameter<string>("rawInstanceName")),
  timing_(params.getUntrackedParameter<bool>("timing", false)),
  disabled_(params.getParameter<bool>("disabled")),
  verbosity_(params.getUntrackedParameter<int>("verbosity", 0)),
  produceDigis_(params.getParameter<bool>("produceDigis")),
  produceRaw_(params.getParameter<bool>("produceRaw")),
  inputRawCollection_(params.getParameter<edm::InputTag>("inputRawCollection")),
  mergeRaw_(params.getParameter<bool>("mergeRaw")),
  ignoreTriggerType_(params.getParameter<bool>("ignoreTriggerType")),
  matacq_(nullptr, 0),
  inFile_(nullptr),
  data_(bufferSize),
  openedFileRunNumber_(0),
  lastOrb_(0),
  fastRetrievalThresh_(0),
  orbitOffsetFile_(params.getUntrackedParameter<std::string>("orbitOffsetFile",
                                 "")),
  inFileName_(""),
  stats_(stats_init),
  logFileName_(params.getUntrackedParameter<std::string>("logFileName",
                             "matacqProducer.log")),
  eventSkipCounter_(0),
  onErrorDisablingEvtCnt_(params.getParameter<int>("onErrorDisablingEvtCnt")),
  timeLogFile_(params.getUntrackedParameter<std::string>("timeLogFile", "")),
  runNumber_(0)
{
  if(verbosity_>=4) cout << "[Matacq " << now() << "] in MatacqProducer ctor"  << endl;
  
  gettimeofday(&timer_, nullptr);

  if(!timeLogFile_.empty()){
    timeLog_.open(timeLogFile_.c_str());
    if(timeLog_.fail()){
      cout << "[LaserSorter " << now() << "] "
           << "Failed to open file " << timeLogFile_ << " to log timing.\n";
      logTiming_ = false;
    } else{
      logTiming_ = true;
    }
  }
  
  posEstim_.verbosity(verbosity_);

  logFile_.open(logFileName_.c_str(), ios::app | ios::out);
  
  if(logFile_.bad()){
    throw cms::Exception("FileOpen") << "Failed to open file "
                     << logFileName_ << " for logging.\n";
  }

  inputRawCollectionToken_ = consumes<FEDRawDataCollection>(params.getParameter<InputTag>("inputRawCollection"));


  if(produceDigis_){
    if(verbosity_>0) cout << "[Matacq " << now() << "] registering new "
                       "EcalMatacqDigiCollection product with instance name '"
                          << digiInstanceName_ << "'\n";
    produces<EcalMatacqDigiCollection>(digiInstanceName_);
  }
  
  if(produceRaw_){
    if(verbosity_>0) cout << "[Matacq " << now() << "] registering new FEDRawDataCollection "
                       "product with instance name '"
                          << rawInstanceName_ << "'\n";
    produces<FEDRawDataCollection>(rawInstanceName_);
  }
  
  startTime_.tv_sec = startTime_.tv_usec = 0;
  if(!orbitOffsetFile_.empty()){
    doOrbitOffset_ = true;
    loadOrbitOffset();
  } else{
    doOrbitOffset_ = false;
  }
  if(verbosity_>=4) cout << "[Matacq " << now() << "] exiting MatacqProducer ctor"  << endl;
}


void
MatacqProducer::produce(edm::Event& event, const edm::EventSetup& eventSetup){
  if(verbosity_>=4) cout << "[Matacq " << now() << "] in MatacqProducer::produce"  << endl;
  if(logTiming_){
    timeval t;
    gettimeofday(&t, nullptr);

    timeLog_ << t.tv_sec << "."
             << setfill('0') << setw(3) << (t.tv_usec+500)/1000 << setfill(' ')<< "\t"
             << (t.tv_usec - timer_.tv_usec)*1. 
      + (t.tv_sec - timer_.tv_sec)*1.e6 << "\t";
    timer_ = t;
  } 
 
  if(startTime_.tv_sec==0) gettimeofday(&startTime_, nullptr);
  ++stats_.nEvents;  
  if(disabled_) return;
  const uint32_t runNumber = getRunNumber(event);
  if(runNumber!=runNumber_){
    newRun(runNumber_, runNumber);
  }
  addMatacqData(event);

  if(logTiming_){
    timeval t;
      gettimeofday(&t, nullptr);
      timeLog_ << (t.tv_usec - timer_.tv_usec)*1. 
        + (t.tv_sec - timer_.tv_sec)*1.e6 << "\n";
      timer_ = t;
  }
}

void
MatacqProducer::addMatacqData(edm::Event& event){

  edm::Handle<FEDRawDataCollection> sourceColl;
  event.getByToken(inputRawCollectionToken_, sourceColl);

  std::unique_ptr<FEDRawDataCollection> rawColl;
  if(produceRaw_){
    if(mergeRaw_){
      rawColl = std::make_unique<FEDRawDataCollection>(*sourceColl);
    } else{
      rawColl = std::make_unique<FEDRawDataCollection>();
    }
  }
  
  auto digiColl = std::make_unique<EcalMatacqDigiCollection>();
  
  if(eventSkipCounter_==0){
    if(sourceColl->FEDData(matacqFedId_).size()>4 && !produceRaw_){
      //input raw data collection already contains matacqData
      formatter_.interpretRawData(sourceColl->FEDData(matacqFedId_),
                  *digiColl);             
    } else{
      bool isLaserEvent = (getCalibTriggerType(event) == laserType);


      //      cout << "---> " << (ignoreTriggerType_?"yes":"no") << " " << getCalibTriggerType(event) << endl;
      
      if(isLaserEvent || ignoreTriggerType_){
    
    const uint32_t runNumber = getRunNumber(event);
    const uint32_t orbitId   = getOrbitId(event);
      
    LogInfo("Matacq") << "Run " << runNumber << "\t Orbit " << orbitId << "\n";
    
    bool fileChange;
    if(doOrbitOffset_){
      map<uint32_t,uint32_t>::iterator it = orbitOffset_.find(runNumber);
      if(it == orbitOffset_.end()){
        LogWarning("Matacq") << "Orbit offset not found for run "
                 << runNumber
                 << ". No orbit correction will be applied.";
      }
    }  
      
    if(getMatacqFile(runNumber, orbitId, &fileChange)){
      //matacq file retrieval succeeded
      LogInfo("Matacq") << "Matacq data file found for "
                << "run " << runNumber << " orbit " << orbitId;
      if(getMatacqEvent(runNumber, orbitId, fileChange)){
        if(produceDigis_){
          formatter_.interpretRawData(matacq_, *digiColl);
        }
        if(produceRaw_){
          uint32_t dataLen64 = matacq_.getParsedLen();
          if(dataLen64 > bufferSize*8 || matacq_.getDccLen()!= dataLen64){
        LogWarning("Matacq") << " Error in Matacq event fragment length! "
                     << "DCC len: " << matacq_.getDccLen()
                     << "*8 Bytes, Parsed len: "
                     << matacq_.getParsedLen() << "*8 Bytes.  "
                     << "Matacq data will not be included for this event.\n";
          } else{
        rawColl->FEDData(matacqFedId_).resize(dataLen64*8);
        copy(data_.begin(), data_.begin() + dataLen64*8,
             rawColl->FEDData(matacqFedId_).data());
          }
        }
        LogInfo("Matacq") << "Associating matacq data with orbit id "
                  << matacq_.getOrbitId()
                  << " to dcc event with orbit id "
                  << orbitId << std::endl;
        if(isLaserEvent){
          ++stats_.nLaserEventsWithMatacq;
        } else{
          ++stats_.nNonLaserEventsWithMatacq;
        }
      } else{
        if(isLaserEvent){
          LogWarning("Matacq") << "No matacq data found for laser event "
                   << "of run " << runNumber << " orbit "
                   << orbitId;
        }
      }
    } else{
      LogWarning("Matacq") << "No matacq file found for event "
                   << event.id();
    }
      }
    }
    if(eventSkipCounter_>0){ //error occured for this events
      //                       and some events will be skipped following
      //                       to this error.
      LogInfo("Matacq") << " [" << now() << "] "
            << eventSkipCounter_
            << " next events will be skipped, following to an "
            << "error on the last processed event, "
            << "which is expected to be persistant.";
    }
  } else{
    --eventSkipCounter_;
  }
  
  if(produceRaw_){
    if(verbosity_>1) cout << "[Matacq " << now() << "] "
                          << "Adding FEDRawDataCollection collection "
                          << " to event.\n";
    event.put(std::move(rawColl), rawInstanceName_);
  }

  if(produceDigis_){
    if(verbosity_>1) cout << "[Matacq " << now() << "] "
                          << "Adding EcalMatacqDigiCollection collection "
                          << " to event.\n";
    event.put(std::move(digiColl), digiInstanceName_);
  }
}

// #if 0
// bool
// MatacqProducer::getMatacqEvent(std::ifstream& f,
//                 uint32_t runNumber,
//                 uint32_t orbitId,
//                 bool doWrap,
//                 std::streamoff maxPos){
//   bool found = false;
//   streampos startPos = f.tellg();

//   while(!f.eof()
//  && !found
//  && (maxPos<0 || f.tellg()<=maxPos)){
//     const streamsize headerSize = 8*8;
//     f.read((char*)&data_[0], headerSize);
//     if(f.eof()) break;
//     int32_t orb = MatacqRawEvent::getOrbitId(&data_[0], headerSize);
//     uint32_t len = MatacqRawEvent::getDccLen(&data_[0], headerSize);
//     uint32_t run = MatacqRawEvent::getRunNum(&data_[0], headerSize);
//     //     cout << "Matacq: orbit = " << orb
//     //    << " len = " << len
//     //    << " run = " << run << endl;
//     if((abs(orb-(int32_t)orbitId) < orbitTolerance_)
//        && (runNumber==0 || runNumber==run)){
//       found = true;
//       //reads the rest of the event:
//       if(data_.size() < len*8){
//  throw cms::Exception("Matacq") << "Buffer overflow";
//       }
//       f.read((char*)&data_[0]+headerSize, len*8-headerSize);
//       matacq_ = MatacqRawEvent((unsigned char*)&data_[0], len*8);
//     } else{
//       //moves to next event:
//       f.seekg(len*8 - headerSize, ios::cur);
//     }
//   }
  
//   f.clear(); //clears eof error to allow seekg  
//   if(doWrap && !found){
//     f.seekg(0, ios::beg);
//     found =  getMatacqEvent(f, runNumber, orbitId, false, startPos);
//   }
//   return found;
// }
//#endif

bool
MatacqProducer::getMatacqEvent(uint32_t runNumber,
                   int32_t orbitId,
                   bool fileChange){
  filepos_t startPos;
  if(!mtell(startPos)) return false;
  
  int32_t startOrb = -1;
  const size_t headerSize = 8*8;
  if(mread((char*)&data_[0], headerSize, "Reading matacq header", true)){
    startOrb = MatacqRawEvent::getOrbitId(&data_[0], headerSize);
    if(startOrb<0) startOrb = 0;
  } else{
    if(verbosity_>2){
      cout << "[Matacq " << now() << "] Failed to read matacq header. Moved to start of "
    " the file.\n";
    }
    mrewind();
    if(mread((char*)&data_[0], headerSize, "Reading matacq header", true)){
      startPos = 0;
      startOrb = MatacqRawEvent::getOrbitId(&data_[0], headerSize);
    } else{
      if(verbosity_>2) cout << "[Matacq " << now() << "] Looks like matacq file is empty"
                            << "\n";
      return false;
    }
  }
  
  if(verbosity_>2) cout << "[Matacq " << now() << "] Last read orbit: " << lastOrb_
                        << " looking for orbit " << orbitId
                        << ". Current file position: " << startPos
                        << " Orbit at current position: " << startOrb << "\n";
  
  //  f.clear();
  bool didCoarseMove = false;

  //FIXME: case where posEtim_.invalid() is false
  if(!posEstim_.invalid()
     && (abs(lastOrb_-orbitId) > fastRetrievalThresh_)){
    filepos_t pos = posEstim_.pos(orbitId);

    //    struct stat st;
    filepos_t fsize;
    //    if(0==stat(inFileName_.c_str(), &st)){
    if(msize(fsize)){
      //      const int64_t fsize = st.st_size;
      if(0!=posEstim_.eventLength() && pos > fsize){
    //estimated position is beyong end of file
    //-> move to beginning of last event:
    int64_t evtSize = posEstim_.eventLength()*sizeof(uint64_t);
    pos = ((int64_t)fsize/evtSize-1)*evtSize;
    if(verbosity_>2){
      cout << "[Matacq " << now() << "] Estimated position was beyond end of file. "
        "Changed to " << pos << "\n";
    }
      }
    } else{
      LogWarning("Matacq") << "Failed to access file " << inFileName_ << ".";
    }
    if(pos>=0){
      if(verbosity_>2) cout << "[Matacq " << now() << "] jumping to estimated position "
                << pos << "\n";
      mseek(pos, SEEK_SET, "Jumping to estimated event position");
      if(mread((char*)&data_[0], headerSize, "Reading matacq header", true)){
    didCoarseMove = true;
      } else{
    //estimated position might have been beyond the end of the file,
    //try, with original position:
    didCoarseMove = false;
    if(!mread((char*)&data_[0], headerSize, "Reading event header", true)){
      return false;
    }
      }
    } else{
      if(verbosity_) cout << "[Matacq " << now() << "] Event orbit outside of orbit range "
               "of matacq data file events\n";
      return false;
    }
  }
  
  int32_t orb = MatacqRawEvent::getOrbitId(&data_[0], headerSize);

  if(didCoarseMove){
    //autoadjustement of threshold for coarse move:
    if(abs(orb-orbitId) > fastRetrievalThresh_){
      if(verbosity_>2) cout << "[Matacq " << now() << "] Fast retrieval threshold increased from "
                            << fastRetrievalThresh_;
      fastRetrievalThresh_ = 2*abs(orb-orbitId);
      if(verbosity_>2) cout << " to " << fastRetrievalThresh_ << "\n";
    }

    //if coarse move did not improve situation, rolls back:
    if(startOrb > 0
       && (abs(orb-orbitId) > abs(startOrb-orbitId))){
      if(verbosity_>2) cout << "[Matacq " << now() << "] Estimation (-> orbit " << orb << ") "
                         "was worst than original position (-> orbit "
                            << startOrb
                            << "). Restoring position (" << startPos << ").\n";
      mseek(startPos, SEEK_SET);
      mread((char*)&data_[0], headerSize, "Reading event header", true);
      orb = MatacqRawEvent::getOrbitId(&data_[0], headerSize);
    }
  }
  
  bool searchBackward = (orb>orbitId)?true:false;
  //BEWARE: len must be signed, because we are using latter in the code (-len)
  //expression
  int len = (int)MatacqRawEvent::getDccLen(&data_[0], headerSize);

  if(len==0){
    cout << "[Matacq " << now() << "] read DCC length is null! Cancels matacq event search "
     << " and move matacq file pointer to beginning of the file. "
     << "(" << __FILE__ << ":" << __LINE__ << ")."
     << "\n";
    //rewind(f);
    mrewind();
    return false;
  }
  
  enum state_t { searching, found, failed } state = searching;
  
  while(state == searching){
    orb = MatacqRawEvent::getOrbitId(&data_[0], headerSize);
    len = (int)MatacqRawEvent::getDccLen(&data_[0], headerSize);
    uint32_t run = MatacqRawEvent::getRunNum(&data_[0], headerSize);
    if(verbosity_>3){
      filepos_t pos = -1;
      mtell(pos);
      cout << "[Matacq " << now() << "] Header read at file position "
       << pos
       << ":  orbit = " << orb
       << " len = " << len << "x8 Byte"
       << " run = " << run << "\n";
    }
    if((abs(orb-orbitId) < orbitTolerance_)
       && (runNumber==0 || runNumber==run)){
      state = found;
      lastOrb_ = orb;
      //reads the rest of the event:
      if((int)data_.size() < len*8){
    throw cms::Exception("Matacq") << "Buffer overflow";
      }
      if(verbosity_>2) cout << "[Matacq " << now() << "] Event found. Reading "
                         " matacq event." << "\n";
      if(!mread((char*)&data_[0], len*8, "Reading matacq event")){
    if(verbosity_>2) cout << "[Matacq " << now() << "] Failed to read matacq event."
                              << "\n";
    state = failed;
      }
      matacq_ = MatacqRawEvent((unsigned char*)&data_[0], len*8);
    } else {
      if((searchBackward && (orb < orbitId))
     || (!searchBackward && (orb > orbitId))){ //search ended
    lastOrb_ = orb;      
    state = failed;
    if(verbosity_>2) cout << "[Matacq " << now()
                << "] No matacq data found for run " << run
                << ", orbit ID " << orbitId << "." << "\n";
      } else{
    off_t offset = (searchBackward?-len:len)*8;
    lastOrb_ = orb; 
    if(verbosity_>3){
      cout << "[Matacq " << now() << "] In matacq file, moving "
           << abs(offset) << " byte " << (offset>0?"forward":"backward")
           << ".\n";
    }   

    if(mseek(offset, SEEK_CUR,
         (searchBackward?"Moving to previous event":
          "Moving to next event"))
       && mread((char*)&data_[0], headerSize, "Reading event header",
            true)){
    } else{
      if(!searchBackward) mseek(-len*8, SEEK_CUR,
                    "Moving to start of last complete event");
      state = failed;
    }
      }
    }
  }
  
  if(state==found){
    filepos_t pos = -1;
    filepos_t fsize = -1;
    mtell(pos);
    msize(fsize);
    if(pos==fsize-1){ //last byte.
      if(verbosity_>2){
    cout << "[Matacq " << now() << "] Event found was at the end of the file. Moving "
      "stream position to beginning of this event."
         << "\n";
      }
      mseek(-(int)len*8-1, SEEK_CUR,
        "Moving to beginning of last matacq event");
    }
  }
  return (state==found);
}


bool
MatacqProducer::getMatacqFile(uint32_t runNumber, uint32_t orbitId,
                  bool* fileChange){
  if(openedFileRunNumber_ != 0
     && openedFileRunNumber_ == runNumber){
    uint32_t firstOrb, lastOrb;
    bool goodRange = getOrbitRange(firstOrb, lastOrb);
    //    if(orbitId < firstOrb || orbitId > lastOrb) continue;
    if(goodRange && firstOrb <= orbitId && orbitId <= lastOrb){
      if(fileChange!=nullptr) *fileChange = false;
      return misOpened();
    }
  }

  if(fileNames_.empty()) return false;

  const string runNumberFormat = "%08d{,_*}";
  string sRunNumber = str(boost::format(runNumberFormat) % runNumber);
  //cout << "Run number string: " << sRunNumber << "\n";
  bool found = false;
  string fname;
  uint32_t maxOrb = 0;
  //we make two iterations to handle the case where the event is procesed
  //before the matacq data are available. In such case we would have
  //orbitId > maxOrb (maxOrb: orbit of last written matacq event)
  for(int itry = 0; itry < 2 && (orbitId > maxOrb); ++itry){
    if(itry > 0){
      int n_sec = 1;
      std::cout << "[Matacq " << now() << "] Event orbit id (" << orbitId << ") goes "
    "beyound the range of available one. Waiting for " << n_sec << " seconds in case "
    "it was not written yet to disk.";
      sleep(n_sec);
    }
    
    for(unsigned i=0; i < fileNames_.size() && !found; ++i){
      fname = fileNames_[i];
      boost::algorithm::replace_all(fname, "%run_subdir%",
                    runSubDir(runNumber));
      boost::algorithm::replace_all(fname, "%run_number%", sRunNumber);

      glob_t g;
      int rc  = glob(fname.c_str(), GLOB_BRACE, nullptr, &g);
      if(rc){
    if(verbosity_ > 1){
      switch(rc){
      case GLOB_NOSPACE:
        std::cout << "[Matacq " << now() << "] Running out of memory while calling glob function to look for matacq file paths\n";
        break;
      case GLOB_ABORTED:
        std::cout << "[Matacq " << now() << "] Read error while calling glob function to look for matacq file paths\n";
        break;
      case GLOB_NOMATCH:
        //ok. No message to report.
        break;
      }
      continue;
    }
      } //rc
      for(unsigned iglob = 0; iglob < g.gl_pathc; ++iglob){
    char* thePath = g.gl_pathv[iglob];
    //FIXME: add sanity check on the path
    static std::atomic<int> nOpenErrors {0};
    const int maxOpenErrors = 50;
    if(!mopen(thePath) && nOpenErrors < maxOpenErrors){
      std::cout << "[Matacq " << now() << "] Failed to open file " << thePath;
      ++nOpenErrors;
      if(nOpenErrors == maxOpenErrors){
        std::cout << nOpenErrors << "This is the " << maxOpenErrors
              << "th occurence of this error. Report of this error is now disabled.\n";
      } else{
        std::cout << "\n";
      }
    }
    uint32_t firstOrb;
    uint32_t lastOrb;
    bool goodRange = getOrbitRange(firstOrb, lastOrb);
    if(goodRange && lastOrb > maxOrb) maxOrb = lastOrb;
    if(goodRange && firstOrb <= orbitId && orbitId <= lastOrb){
      found = true;
      //continue;
      fname = thePath;
      if(verbosity_ > 1) std::cout << "[Matacq " << now() << "] Switching to file " << fname << "\n";
      break;
    }
      } //next iglob
      globfree(&g);
    }//next filenames
  } //next itry
  
  if(found){
    LogInfo("Matacq") << "Uses matacq data file: '" << fname << "'\n";
  } else{
    if(verbosity_>=0) cout << "[Matacq " << now() << "] no matacq file found "
            "for run " << runNumber << "\n";
    eventSkipCounter_ = onErrorDisablingEvtCnt_;
    openedFileRunNumber_ = 0;
    if(fileChange!=nullptr) *fileChange = false;
    return false;
  }

   if(found){
    openedFileRunNumber_ = runNumber;
    lastOrb_ = 0;
    posEstim_.init(this);
    if(fileChange!=nullptr) *fileChange = true;
    return true;
  } else{
    return false;
  }
}


uint32_t MatacqProducer::getRunNumber(edm::Event& ev) const{
  return ev.run();
}

uint32_t MatacqProducer::getOrbitId(edm::Event& ev) const{
  //on CVS HEAD (June 4, 08), class Event has a method orbitNumber()
  //we could use here. The code would be shorten to:
  //return ev.orbitNumber();
  //we have to deal with what we have in current CMSSW releases:
  edm::Handle<FEDRawDataCollection> rawdata;
  ev.getByToken(inputRawCollectionToken_, rawdata);
  if(!(rawdata.isValid())){
    throw cms::Exception("NotFound")
      << "No FED raw data collection found. ECAL raw data are "
      "required to retrieve the orbit ID";
  }
  
  int orbit = 0;
  for(int id=601; id<=654; ++id){
    if(!FEDNumbering::inRange(id)) continue;
    const FEDRawData& data = rawdata->FEDData(id);
    const int orbitIdOffset64 = 3;
    if(data.size()>=8*(orbitIdOffset64+1)){//orbit id is in 4th 64-bit word
      const unsigned char* pOrbit = data.data() + orbitIdOffset64*8;
      int thisOrbit = pOrbit[0]
    | (pOrbit[1] <<8)
    | (pOrbit[2] <<16)
    | (pOrbit[3] <<24);
      if(orbit!=0 && thisOrbit!=0 && abs(orbit-thisOrbit)>orbitTolerance_){
    //throw cms::Exception("EventCorruption")
    //  << "Orbit ID inconsitency in DCC headers";
    LogWarning("EventCorruption")
      << "Orbit ID inconsitency in DCC headers";
    orbit = 0;
    break;
      }
      if(thisOrbit!=0) orbit = thisOrbit;
    }
  }
  
  if(orbit==0){
    //    throw cms::Exception("NotFound")
    //  << "Failed to retrieve orbit ID of event "<< ev.id();
    LogWarning("NotFound") << "Failed to retrieve orbit ID of event "
               << ev.id();
  }
  return orbit;
}

int MatacqProducer::getCalibTriggerType(edm::Event& ev) const{  
  edm::Handle<FEDRawDataCollection> rawdata;
  ev.getByToken(inputRawCollectionToken_, rawdata);
  if(!(rawdata.isValid())){
    throw cms::Exception("NotFound")
      << "No FED raw data collection found. ECAL raw data are "
      "required to retrieve the trigger type";
  }
  
  Majority<int> stat;
  for(int id=601; id<=654; ++id){
    if(!FEDNumbering::inRange(id)) continue;
    const FEDRawData& data = rawdata->FEDData(id);
    const int detailedTrigger32 = 5;
    if(data.size()>=4*(detailedTrigger32+1)){
      const unsigned char* pTType = data.data() + detailedTrigger32*4;
      int tType = pTType[1] & 0x7;
      stat.add(tType);
    }
  }
  double p;
  int tType = stat.result(&p);
  if(p<0){
    //throw cms::Exception("NotFound") << "No ECAL DCC data found\n";
    LogWarning("NotFound")  << "No ECAL DCC data found\n";
    tType = -1;
  }
  if(p<.8){
    //throw cms::Exception("EventCorruption") << "Inconsitency in detailed trigger type indicated in ECAL DCC data headers\n";
    LogWarning("EventCorruption") << "Inconsitency in detailed trigger type indicated in ECAL DCC data headers\n";
    tType = -1;
  }
  return tType;
}
  
void MatacqProducer::PosEstimator::init(MatacqProducer* mp){
  mp->mrewind();

  const size_t headerSize = 8*8;
  unsigned char data[headerSize];
  if(!mp->mread((char*)data, headerSize)){
    if(verbosity_) cout << "[Matacq " << now() << "] reached end of file!\n"; 
    firstOrbit_ = eventLength_ = orbitStepMean_ = 0;
    return;
  } else{
    firstOrbit_ = MatacqRawEvent::getOrbitId(data, headerSize);
    eventLength_ = MatacqRawEvent::getDccLen(data, headerSize);
    if(verbosity_>1) cout << "[Matacq " << now() << "] First event orbit: " << firstOrbit_
                          << " event length: " << eventLength_
                          << "*8 byte\n";
  }

  mp->mrewind();
  
  if(eventLength_==0){    
    if(verbosity_) cout << "[Matacq " << now() << "] event length is null!" << endl; 
    return;
  }

  filepos_t s;
  mp->msize(s);

  //number of complete events:
  const unsigned nEvents = s/eventLength_/8;

  if(nEvents==0){
    if(verbosity_) cout << "[Matacq " << now() << "] File is empty!" << endl;
    orbitStepMean_ = 0;
    return;
  }

  if(verbosity_>1) cout << "[Matacq " << now() << "] File size: " << s
                        << " Number of events: " << nEvents << endl;
  
  //position of last complete events:
  off_t last = (nEvents-1)*(off_t)eventLength_*8;
  mp->mseek(last, SEEK_SET, "Moving to beginning of last complete "
        "matacq event");
  if(!mp->mread((char*) data, headerSize, "Reading matacq header", true)){
    LogWarning("Matacq") << "Fast matacq event retrieval failure. "
      "Falling back to safe retrieval mode.";
    orbitStepMean_ = 0;
  }
  
  int32_t lastOrb = MatacqRawEvent::getOrbitId(data, headerSize);
  int32_t lastLen = MatacqRawEvent::getDccLen(data, headerSize);

  if(verbosity_>1) cout << "[Matacq " << now() << "] Last event orbit: " << lastOrb
                        << " last event length: " << lastLen << endl;
  
  //some consistency check
  if(lastLen!=eventLength_){
    LogWarning("Matacq")
      //throw cms::Exception("Matacq")
      << "Fast matacq event retrieval failure: it looks like "
      "the matacq file contains events of different sizes.";
      //      " Falling back to safe retrieval mode.";
    invalid_ = false; //true;
    orbitStepMean_ = 112; //0;
    return;
  }

  orbitStepMean_ = (lastOrb - firstOrbit_)/nEvents;
  
  if(verbosity_>1) cout << "[Matacq " << now() << "] Orbit step mean: " << orbitStepMean_
                        << "\n";

  invalid_ = false;
}

int64_t MatacqProducer::PosEstimator::pos(int orb) const{
  if(orb<firstOrbit_) return -1;
  uint64_t r = orbitStepMean_!=0?
    (((uint64_t)(orb-firstOrbit_))/orbitStepMean_)*eventLength_*8
    :0;
  if(verbosity_>2) cout << "[Matacq " << now() << "] Estimated Position for orbit  " << orb
                        << ": " << r << endl;
  return r;
}

MatacqProducer::~MatacqProducer(){
  mclose();
  timeval t;
  gettimeofday(&t, nullptr);
  if(logTiming_ && startTime_.tv_sec!=0){
    //not using logger, to allow timing with different logging options
    cout << "[Matacq " << now() << "] Time elapsed between first event and "
      "destruction of MatacqProducer: "
     << ((t.tv_sec-startTime_.tv_sec)*1.
         + (t.tv_usec-startTime_.tv_usec)*1.e-6) << "s\n";
  }
}

void MatacqProducer::loadOrbitOffset(){
  std::ifstream f(orbitOffsetFile_.c_str());
  if(f.bad()){
    throw cms::Exception("Matacq")
      << "Failed to open orbit ID correction file '"
      << orbitOffsetFile_ << "'\n";
  }

  cout << "[Matacq " << now() << "] "
       << "Offset to substract to Matacq events Orbit ID: \n"
       << "#Run Number\t Offset\n";

  int iline = 0;
  string s;
  stringstream buf;
  while(f.eof()){
    getline(f, s);
    ++iline;
    if(s[0]=='#'){//comment
      //skip line:
      f.ignore(numeric_limits<streamsize>::max(), '\n');
      continue;
    }
    buf.str("");
    buf << s;
    int run;
    int orbit;
    buf >> run;
    buf >> orbit;
    if(buf.bad()){
      throw cms::Exception("Matacq")
    << "Syntax error in Orbit offset file '"
    << orbitOffsetFile_ << "'";
    }
    cout << run << "\t" << orbit << "\n";
    orbitOffset_.insert(pair<int, int>(run, orbit));
  }
}

#ifdef USE_STORAGE_MANAGER
bool MatacqProducer::mseek(filepos_t offset, int whence, const char* mess){
  if(0==inFile_.get()) return false;
  try{
    Storage::Relative wh;
    if(whence==SEEK_SET) wh = Storage::SET;
    else if(whence==SEEK_CUR) wh = Storage::CURRENT;
    else if(whence==SEEK_END) wh = Storage::END;
    else throw cms::Exception("Bug") << "Bug found in "
                                     << __FILE__ << ": "<< __LINE__ << "\n";
                   
    inFile_->position(offset, wh);
  } catch(cms::Exception& e){
    if(verbosity_){
      cout << "[Matacq " << now() << "] ";
      if(mess) cout << mess << ". ";
      cout << "Random access error on input matacq file. ";
      if(whence==SEEK_SET) cout << "Failed to seek absolute position " << offset;
      else if(whence==SEEK_CUR) cout << "Failed to move " << offset << " bytes forward";
      else if(whence==SEEK_END) cout << "Failed to seek position at " << offset << " bytes before end of file";
    cout << ". Reopening file. " << e.what() << "\n";
      mopen(inFileName_);
      return false;
    }
  }
  return true;
}

bool MatacqProducer::mtell(filepos_t& pos){
  if(0==inFile_.get()) return false;
  pos = inFile_->position();
  return true;
}

bool MatacqProducer::mread(char* buf, size_t n, const char* mess, bool peek){
  if(0==inFile_.get()) return false;
  
  filepos_t pos = -1;
  if(!mtell(pos)) return false;

  bool rc = false;
  try{
    rc =  (n==inFile_->xread(buf, n));
  } catch(cms::Exception& e){
    if(verbosity_){
      cout << "[Matacq " << now() << "] ";
      if(mess) cout << mess << ". ";
      cout << "Read failure from input matacq file: "
           << e.what() << "\n";
    }
    //recovering from error:
    mopen(inFileName_);
    mseek(pos);
    return false;
  }
  if(peek){//asked to restore original file position
    mseek(pos);
  }
  return rc;
}

bool MatacqProducer::msize(filepos_t& s){
  if(inFile_.get()==0) return false;
  s = inFile_.get()->size();
  return true;
}

bool MatacqProducer::mrewind(){
  Storage* file = inFile_.get();
  if(file==0) return false;
  try{
    file->rewind();
  } catch(cms::Exception e){
    if(verbosity_) cout << "Exception cautgh while rewinding file "
                        << inFileName_ << ": " << e.what() << ". "
                        << "File will be reopened.";
    return mopen(inFileName_);
  }
  return true;
}

bool MatacqProducer::mcheck(const std::string& name){
  return StorageFactory::get()->check(name);
}

bool MatacqProducer::mopen(const std::string& name){
  //close already opened file if any:
  mclose();
  
  try{
    inFile_
      = auto_ptr<Storage>(StorageFactory::get()->open(name,
                                                      IOFlags::OpenRead));
    inFileName_ = name;
  } catch(cms::Exception& e){
    LogWarning("Matacq") << e.what();
    inFile_.reset();
    inFileName_ = "";
    return false;
  }
  return true;
}

void MatacqProducer::mclose(){
  if(inFile_.get()!=0){
    inFile_->close();
    inFile_.reset();
  }
}

bool MatacqProducer::misOpened(){
  return inFile_.get()!=0;
}

bool MatacqProducer::meof(){
  if(inFile_.get()==0) return true;
  return inFile_->eof();
}

#else //USE_STORAGE_MANAGER not defined
bool MatacqProducer::mseek(off_t offset, int whence, const char* mess){
  if(nullptr==inFile_) return false;
  const int rc = fseeko(inFile_, offset, whence);
  if(rc!=0 && verbosity_){
    cout << "[Matacq " << now() << "] ";
    if(mess) cout << mess << ". ";
    cout << "Random access error on input matacq file. "
      "Rewind file.\n";
    mrewind();
  }
  return rc==0;
}

bool MatacqProducer::mtell(filepos_t& pos){
  if(nullptr==inFile_) return false;
  pos = ftello(inFile_);
  return pos != -1;
    
}

bool MatacqProducer::mread(char* buf, size_t n, const char* mess, bool peek){
  if(nullptr==inFile_) return false;
  off_t pos = ftello(inFile_);
  bool rc = (pos!=-1) && (1==fread(buf, n, 1, inFile_));
  if(!rc){
    if(verbosity_){
      cout << "[Matacq " << now() << "] ";
      if(mess) cout << mess << ". ";
      cout << "Read failure from input matacq file.\n";
    }
    clearerr(inFile_);
  }
  if(peek || !rc){//need to restore file position
    if(0!=fseeko(inFile_, pos, SEEK_SET)){
      if(verbosity_){
    cout << "[Matacq " << now() << "] ";
    if(mess) cout << mess << ". ";
    cout << "Failed to restore file position of "
      "before read error. Rewind file.\n";
      }
      //rewind(inFile_.get());
      mrewind();
      lastOrb_ = 0;
    }
  }
  return rc;
}

bool MatacqProducer::msize(filepos_t& s){
  if(nullptr==inFile_) return false;
  struct stat buf;
  if(0!=fstat(fileno(inFile_), &buf)){
    s = 0;
    return false;
  } else{
    s = buf.st_size;
    return true;
  }
}

bool MatacqProducer::mrewind(){
  if(nullptr==inFile_) return false;
  clearerr(inFile_);
  return fseeko(inFile_, 0, SEEK_SET)!=0; 
}

bool MatacqProducer::mcheck(const std::string& name){
  struct stat dummy;
  return 0==stat(name.c_str(), &dummy);
//   if(stat(name.c_str(), &dummy)==0){
//     return true;
//   } else{
//     cout << "[Matacq " << now() << "] Failed to stat file '"
//   << name.c_str() << "'. " 
//   << "Error " << errno << ": " << strerror(errno) << "\n";
//     return false;
//   }
}

bool MatacqProducer::mopen(const std::string& name){
  if(inFile_!=nullptr) mclose();
  inFile_ = fopen(name.c_str(), "r");
  if(inFile_!=nullptr){
    inFileName_ = name;
    return true;
  } else{
    inFileName_ = "";
    return false;
  }
}

void MatacqProducer::mclose(){
  if(inFile_!=nullptr) fclose(inFile_);
  inFile_ = nullptr;
}

bool MatacqProducer::misOpened(){
  return inFile_!=nullptr;
}

bool MatacqProducer::meof(){
  if(nullptr==inFile_) return true;
  return feof(inFile_)==0;
}

#endif //USE_STORAGE_MANAGER defined

std::string MatacqProducer::runSubDir(uint32_t runNumber){
  int millions = runNumber / (1000*1000);
  int thousands = (runNumber-millions*1000*1000) / 1000;
  int units = runNumber-millions*1000*1000 - thousands*1000;
  return str(boost::format("%03d/%03d/%03d") % millions % thousands % units);
}

void MatacqProducer::newRun(int prevRun, int newRun){
    runNumber_ = newRun;
    eventSkipCounter_ = 0;
    logFile_ << "[" << now() << "] Event count for run "
         << runNumber_ << ": "
         << "total: " << stats_.nEvents << ", "
         << "Laser event with Matacq data: "
         << stats_.nLaserEventsWithMatacq << ", "
         << "Non laser event (according to DCC header) with Matacq data: "
         << stats_.nNonLaserEventsWithMatacq << "\n" << flush;

    stats_.nEvents = 0;
    stats_.nLaserEventsWithMatacq = 0;
    stats_.nNonLaserEventsWithMatacq = 0;

    
}

bool MatacqProducer::getOrbitRange(uint32_t& firstOrb, uint32_t& lastOrb){
  filepos_t pos = -1;
  filepos_t fsize = -1;
  mtell(pos);
  msize(fsize);
  const unsigned headerSize = 8*8;
  unsigned char header[headerSize];
  //FIXME: Don't we need here to rewind?
  mseek(0);
  if(!mread((char*)header, headerSize, nullptr, false)) return false;
  firstOrb = MatacqRawEvent::getOrbitId(header, headerSize);
  int len = (int)MatacqRawEvent::getDccLen(header, headerSize);
  //number of complete events. If last event is partially written,
  //it won't be included in the count.
  unsigned nEvts = fsize / (len*64);
  //Position of last complete event:
  filepos_t lastEvtPos = (nEvts - 1) * len * 64;
  mseek(lastEvtPos);
  mread((char*)header, headerSize, nullptr, false);
  lastOrb = MatacqRawEvent::getOrbitId(header, headerSize);
  
  //restore file position:
  mseek(pos);

  return true;
}