NOvASocketInputDriver.cxx

Go to the documentation of this file.

////////////////////////////////////////////////////////////////////////
/// \brief   Source to read raw Events from a socket, per specifications
///          in NOvA-DocDB-3947 and NOvA-DocDB-4390.
///
/// \author  paterno@fnal.gov
////////////////////////////////////////////////////////////////////////
#include "DAQ2RawDigit/NOvASocketInputDriver.h"

#include "art/Framework/IO/Sources/put_product_in_principal.h"
#include "canvas/Utilities/Exception.h"
#include "art/Framework/Core/FileBlock.h"
#include "art/Framework/Principal/EventPrincipal.h"
#include "art/Framework/Principal/RunPrincipal.h"
#include "art/Framework/Principal/SubRunPrincipal.h"
#include "art/Framework/Core/ProductRegistryHelper.h"
#include "cetlib/split.h"
#include "fhiclcpp/ParameterSet.h"
#include "messagefacility/MessageLogger/MessageLogger.h"

#include "DAQ2RawDigit/OnlineUnpack.h"
#include "EventDispatcher_CommandSet/DispatcherCmd.h"
#include "NovaDAQConventions/DAQConventions.h"
#include "NovaTimingUtilities/TimingUtilities.h"
#include "RawData/DAQHeader.h"
#include "RawData/RawDigit.h"
#include "RawData/RawTrigger.h"
#include "RawData/FlatDAQData.h"
#include "RawData/RawSumDropMB.h"
#include "RawData/RawSumDCM.h"
#include "SummaryData/RunData.h"

//
// A bit mask to tell us which errors to abort on. Use "0" if you want
// to ignore errors and try to keep running. When debugging, you can
// select which errors you want to focus on.
//
static const int gsMAGIC_WORD       = 1<<0;
static const int gsUNPACK_RETURN    = 1<<1;
static const int gsUNPACK_EXCEPTION = 1<<2;
static const int gsABORT_ON         = 0;

namespace daq2raw {
  // Utility functions for this file, which do not need to be class
  // members.
  
  /// Parse a string (the "filename") to yield an address and a port.  The
  /// "filename" is expected to have the form <address>:<port>, where
  /// neither *address* nor *port* contain a colon.
  void parseFilename(std::string const& filename,
                     std::string& address,
                     std::string& port)
  {
    std::vector<std::string> components;
    cet::split(filename, ':', std::back_inserter(components));
    if (components.size() != 2)
      {
        throw art::Exception(art::errors::Configuration)
          << "Bad socket address in NOvASocketDriver: "
          << filename
          << "\n";
      }
    address = components[0];
    port = components[1];
  }
  
  //---------------------------------------------------------------------
  // Required constructor.
  NOvASocketInputDriver::NOvASocketInputDriver(fhicl::ParameterSet const&  pset,
                                               art::ProductRegistryHelper& helper,
                                               art::SourceHelper const&  pm)
    // look to see if the user provided a base file name for the gdml file, if not
    // pass along an emptry string to default to the behavior previously in NOvAInputDriverBase
    : NOvAInputDriverBase(pset.get<std::string>("GDMLFileBase",""))
    , fSourceHelper(pm)
    , fCurrentFilename()
    , fCurrentSubRunID(-1, -1) ///< Has to be distinguished from the ones that are acually used
    , fIoService()
    , fSocket(fIoService)
    , fCurrentEventCapacity(in_elements(65536 * 1024))
    , fMaxEventSizeBytes(pset.get<size_t>("maxEventSizeK", 65536) * 1024)
    , fSocketBuffer(fCurrentEventCapacity + 1) // Allow for data size info..
    , fFilterCorruptedNanoslices(pset.get<bool>("FilterCorruptedNanoslices"))
    , fDoFillFlatDAQData        (pset.get<bool>("DoFillFlatDAQData"))
    , fMaxReadAttempts          (pset.get<int>("MaxReadAttempts",100))
    , fCorruptEventCount(0)
    , fConsecutiveCorruptEventCount(0)
    , fBadReadCount(0)
    , fConsecutiveBadReadCount(0)
    , trigmask_word1(pset.get<uint32_t>("TriggerMask1",0x00ffffff))
    , trigmask_word2(pset.get<uint32_t>("TriggerMask2",0xffffffff))
    , trigmask_word3(pset.get<uint32_t>("TriggerMask3",0xffffffff))
    , fTransmitDelay(pset.get<int>("TransmitDelay",1000))
  {
    // Declare products that *may* be in the event (or run, or subRun)
    // here. Note the use of "reconstitutes" vs "produces": this is what
    // allows us to specify module label, process name and (optionally)
    // instance label for each product.
    helper.reconstitutes<rawdata::DAQHeader,               art::InEvent>("daq");
    helper.reconstitutes<std::vector<rawdata::RawDigit>,   art::InEvent>("daq");
    helper.reconstitutes<std::vector<rawdata::RawTrigger>, art::InEvent>("daq");
    helper.reconstitutes<std::vector<rawdata::RawSumDropMB>, art::InEvent>("daq");
    helper.reconstitutes<std::vector<rawdata::RawSumDCM>, art::InEvent>("daq");
    if(fDoFillFlatDAQData) helper.reconstitutes<std::vector<rawdata::FlatDAQData>, art::InEvent>("daq");
    helper.reconstitutes<sumdata::RunData,                 art::InRun  >("daq");
  }
  
  //---------------------------------------------------------------------
  // Required function: close currently-open file.
  void NOvASocketInputDriver::closeCurrentFile()
  {
    // Our "file" is a socket.
    fSocket.close();
  }
  
  //---------------------------------------------------------------------
  /// Required function: open the file, "name" and construct and return a
  /// new FileBlock object. MUST be successful or throw:
  /// art::Exception(art::errors::FileOpenError) or
  /// art::Exception(art::errors::FileReadError) are good candidates.
  /// In this implementation, the role of the 'file' is played by a socket;
  /// we interpret the "name" as the address and port number identifying
  /// the socket.
  void NOvASocketInputDriver::readFile(std::string const &name,
                                       art::FileBlock* &fb)
  {
    namespace ba = boost::asio;
    namespace bai = boost::asio::ip;
    fCurrentFilename = name;
    std::string port;
    std::string socketAddress;
    parseFilename(fCurrentFilename, socketAddress, port);
    try
      {
        // Create the file block now, since if we don't throw we're expected
        // to return one.
      fb = new art::FileBlock(art::FileFormatVersion(1, "NOvARawInput 2011a"),
                              fCurrentFilename);
      
      bai::tcp::resolver resolver(fIoService);
      bai::tcp::resolver::query query(socketAddress, port);
      bai::tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
      bai::tcp::resolver::iterator end;
      
      boost::system::error_code error = boost::asio::error::host_not_found;
      while (error && endpoint_iterator != end)
        {
          fSocket.close();
          fSocket.connect(*endpoint_iterator++, error);
        }
      if (error) throw boost::system::system_error(error);
      static size_t const cb_size = 32;
      // Put a command on the socket ...
      //uint32_t dummy;
      cb_t cmd_buffer(cb_size);

#define SET_DISPATCHER_TRIGGER_MASKS 1
#ifdef SET_DISPATCHER_TRIGGER_MASKS
      // Internal buffers for trigger mask commands
      cb_t mask1_buffer(cb_size);
      cb_t mask2_buffer(cb_size);
      cb_t mask3_buffer(cb_size);

      // uint32_t trigmask_word1 = 0xFFFFFFFF;
      // uint32_t trigmask_word2 = 0xFFFFFFFF;
      // uint32_t trigmask_word3 = 0xFFFFFFFF;

      // Construct the commands to set the different trigger masks (1-3)
      DispatcherCMD mask1_cmd(DispatcherCMD::DSPCMD_SET_TRIGGER_MASK_1, trigmask_word1  & 0x00ffffff, &mask1_buffer[0]);
      DispatcherCMD mask2_cmd(DispatcherCMD::DSPCMD_SET_TRIGGER_MASK_2, trigmask_word2, &mask2_buffer[0]);
      DispatcherCMD mask3_cmd(DispatcherCMD::DSPCMD_SET_TRIGGER_MASK_3, trigmask_word3, &mask3_buffer[0]);

      // Serialize the set mask commands to their transmit buffers
      ssize_t cmd1_size_bytes = mask1_cmd.write();
      ssize_t cmd2_size_bytes = mask2_cmd.write();
      ssize_t cmd3_size_bytes = mask3_cmd.write();

      // Send the mask commands to the dispatcher
      fSocket.send(boost::asio::buffer(&mask1_buffer[0], cmd1_size_bytes));
      // ... and receive the acknowledgement.
      // fSocket.receive(boost::asio::buffer(&cmd_buffer[0], cb_size));
      usleep(fTransmitDelay);

      fSocket.send(boost::asio::buffer(&mask2_buffer[0], cmd2_size_bytes));
      // ... and receive the acknowledgement.
      // fSocket.receive(boost::asio::buffer(&cmd_buffer[0], cb_size));
      usleep(fTransmitDelay);


      fSocket.send(boost::asio::buffer(&mask3_buffer[0], cmd3_size_bytes));
      // ... and receive the acknowledgement.
      // fSocket.receive(boost::asio::buffer(&cmd_buffer[0], cb_size));
      usleep(fTransmitDelay);

      // If we try to send too many messages to the dispatcher too rapidly, we will get stuck in a non-responsive state.
      // So, after sending the message to the dispatcher to configure the trigger masks, we will pause for one second
      // before sending the next message.
      // usleep(fTransmitDelay);

#endif

      DispatcherCMD single_cmd(DispatcherCMD::DSPCMD_SET_EVENTSTREAM_SINGLE,
                               &cmd_buffer[0]);
      ssize_t cmd_size_bytes = single_cmd.write();
      fSocket.send(boost::asio::buffer(&cmd_buffer[0], cmd_size_bytes));
      usleep(fTransmitDelay);

      // ... and receive the acknowledgement.
      fSocket.receive(boost::asio::buffer(&cmd_buffer[0], cb_size));
      
#ifndef NOVA_DEBUG_STREAM_ACK
      std::cerr << "Received: "
                << boost::numeric_cast<unsigned>(cmd_buffer[0])
                << " bytes.\n";
      for (cb_t::const_iterator
             i = cmd_buffer.begin(),
             e = cmd_buffer.end();
           i != e;
           ++i) {
        std::cerr << boost::numeric_cast<unsigned>(*i) << " ";
      }
      std::cerr << "\n";
      for (size_t i = 0; i + 3 < cb_size; i += 4) {
        std::cerr
          << std::hex
          << std::setfill('0')
          << "0x"
          << std::setw(8)
          << *reinterpret_cast<uint32_t*>(&cmd_buffer[i]) << " ";
      }
      std::cerr << "\n";
#endif // NOVA_DEBUG_STREAM_ACK
      if (cmd_buffer[0] > cb_size) {
        mf::LogWarning("StreamReadFail")
          << "Unable to read entire command acknowledgement of "
          << cmd_buffer[0]
          << " bytes: buffer size is only "
          << cb_size
          << " bytes.\n";
        try {
          if (fSocket.is_open()) fSocket.close();
        } catch (...) {
          // Ignore.
        }
        return; // Can't open is not fatal for this input source.
      }
      static char const ACK = 6;
      unsigned char ack = cmd_buffer[1];
      if (cmd_buffer[1] != ACK) { // No ACK
        throw art::Exception(art::errors::FileReadError)
          << "Did not receive ACK for SINGLE command: expected 0x"
          << std::hex
          << std::setfill('0')
          << std::setw(2)
          << static_cast<unsigned>(ACK)
          << ", got 0x"
          << std::setfill('0')
          << std::setw(2)
          << static_cast<unsigned>(ack)
          << ".\n";
      }
      unsigned char cmd = cmd_buffer[2];
      if (cmd != DispatcherCMD::DSPCMD_SET_EVENTSTREAM_SINGLE) {
        throw art::Exception(art::errors::FileReadError)
          << "Received ACK for unexpected command 0x"
          << std::hex
          << std::setfill('0')
          << std::setw(2)
          << cmd
          << " (expected 0x"
          << DispatcherCMD::DSPCMD_SET_EVENTSTREAM_SINGLE
          << ").\n";
      }
    }
  catch (boost::system::system_error& x)
    {
      mf::LogWarning("StreamOpenFail")
        << "Unable to open socket to address "
        << fCurrentFilename
        << " for NOvASocketInputDriver, due to system error:\n"
        << x.what()
        << "\n";
      try {
        if (fSocket.is_open()) fSocket.close();
      } catch (...) {
        // Ignore.
      }
    }
}

  //---------------------------------------------------------------------
  /// Required function: read and fill Run, SubRun and/or Event as might 
  /// be appropriate.
  bool NOvASocketInputDriver::readNext(art::RunPrincipal* const &inR,
                                       art::SubRunPrincipal* const &inSR,
                                       art::RunPrincipal* &outR,
                                       art::SubRunPrincipal* &outSR,
                                       art::EventPrincipal* &outE) {
    

    // Allocate and initialize here to allow for a proper use
    // of a goto later
    int                       EvtStartIDX    =0;
    size_t                    receivedBytes  =0;
    ssize_t                   cmd_size_bytes =0;
    uint32_t                  eventSize      =0;
    boost::system::error_code error;
    int                       readAttempts   =0;
    ssize_t                   bytesRead      =0;
    
    receivedBytes=0;
    cmd_size_bytes=0;
    
  retryLOOP:
    
    // Check that the socket is open
    if (!fSocket.is_open()){
      // If the socket is closed/has closed return back
      return false;
    }
    
    ////////////////////////////////////
    // Send the 'NEXT_EVENT' command here, then read the resulting event.
    try {
      // Construct the NEXT_EVENT command object
      DispatcherCMD next_event_cmd(DispatcherCMD::DSPCMD_NEXT_EVENT,
                                   &fSocketBuffer[0]);
      
      // Serialize the command out to the socketbuffer
      cmd_size_bytes = next_event_cmd.write();
      
      // Now send the NEXT_EVENT command to the dispatcher source
      fSocket.send(boost::asio::buffer(&fSocketBuffer[0], cmd_size_bytes));
      
      // Cast the buffer to an 8bit byte array from a 32bit word array
      uint8_t*const pSocketBuffer= reinterpret_cast<uint8_t*const>(&fSocketBuffer[0]);
      
      // Reset the number of bytes that have been received
      receivedBytes =0;
      bytesRead     =0;
      
      // Read the first 4 words of the event 
      // this allows us to retrieve the total
      // event size from the response:
      while ( (receivedBytes < 16) ) {
        ++readAttempts;
        // If we reach our maximum number of retrys
        // then we throw 
        if(readAttempts > fMaxReadAttempts){
          throw "Maximum number of retrys reached on byte count read";
        }
        
        // Attempt to read the header
        bytesRead = fSocket.receive(boost::asio::buffer(pSocketBuffer + receivedBytes, 16 - receivedBytes ),
                                    0,
                                    error);
        
        // If we receive zero bytes back from the receive call 
        // this indicates an error.  We go to check the error code
        // in this case and take the appropriate action.
        if(bytesRead == 0){
          // Catch errors in the IO system
          if(error==boost::asio::error::eof){
            return false;
          }else if(error){
            // Some other error occured
            throw boost::system::system_error(error);
          } // endif error
        } // endif bytesRead
        
          // Increment the total number of bytes that have been received
        receivedBytes += bytesRead;
        
        // receive is a blocking IO call.  There is no reason to sleep.
        // usleep(1);
      }
      
      //
      // Debugging - we should always find the magic number 0xe929e929
      // as the second word in the stream. If we don't there may be
      // problems. Take a look at the start of the stream.
      //
      if (fSocketBuffer[1]!=0xe929e929) {
        std::cerr << __FILE__ " : " << __LINE__ 
                  << " Error: Bad magic word at fSocketBuffer[1]. Will try shifting..."
                  << std::endl;
        if (gsABORT_ON & gsMAGIC_WORD) abort();
      }
      
      //
      // Seek in the stream until we find the magic 0xe929e929. This
      // should be the second word so don't look too far.
      //
      int kEvtStartIDXMax = 8;
      for (EvtStartIDX=0; EvtStartIDX<kEvtStartIDXMax; ++EvtStartIDX) {
        if(0xe929e929 == fSocketBuffer[EvtStartIDX+1]) break;
      }
      if (EvtStartIDX>=kEvtStartIDXMax) {
        throw std::string("Unable to determine byte count on network read");
      }
      eventSize = ntohl(fSocketBuffer[EvtStartIDX]);
      
      // Don't modify the data stream in place.  This can result
      // in a mixed endian problem and we don't want to deal with it.
      // fSocketBuffer[0] = ntohl(fSocketBuffer[0]);
      
#ifndef NOVA_DEBUG_STREAM_ACK
      std::cerr << "Event contains " << eventSize << " bytes.\n";
#endif // NOVA_DEBUG_STREAM_ACK
      
      if(eventSize == 0){
        printf("EVT DATA:\n0x%08x 0x%08x 0x%08x 0x%08x\n",
               fSocketBuffer[0], fSocketBuffer[1],
               fSocketBuffer[2], fSocketBuffer[3]);
      }
      
      // Trap on the various bad event values
      
      // An "event size" of 0xFFFFFFFF means END_OF_DATA, so return false.
      switch(eventSize){
      case 0:
        // This is an empty event.  This indicates a problem
        // with the network stream 
        ++fBadReadCount;
        ++fConsecutiveBadReadCount;
        eventSize = ntohl(fSocketBuffer[1]);
        EvtStartIDX = 1;
        break;
      case 0xFFFFFFFF:
        // This is the event corrupt byte count.  This indicates a 
        // problem with the dispatchers ability to buffer the event
        ++fCorruptEventCount;
        ++fConsecutiveCorruptEventCount;
        return false;
        break;
      default:
        // This is a normal event, we reset our 
        // consecutive error counters
        fConsecutiveBadReadCount      =0;
        fConsecutiveCorruptEventCount =0;
        break;
      }
      
      // If we make it to this point then we probably have a good event
      // header in the buffer.  We will use this to get the rest of the event
      // data off of the network.
      
      if( eventSize > (ssize_t)fMaxEventSizeBytes){
        throw art::Exception(art::errors::FileReadError)
          << "Unable to read entire event "
          << eventSize
          << " bytes: buffer size is only "
          << fMaxEventSizeBytes
          << " bytes.\n";
      } else if ( eventSize > fCurrentEventCapacity) {
        fSocketBuffer.resize( eventSize + 1);
        fCurrentEventCapacity = eventSize;
        fSocket.receive(boost::asio::buffer(&fSocketBuffer[1 + fCurrentEventCapacity],
                                            in_bytes(fSocketBuffer[0] - fCurrentEventCapacity)));
      }
      
      // Read the response:

      while ( receivedBytes < (eventSize+4) )
        {
          size_t remainingBytes = eventSize +4  -receivedBytes;
          remainingBytes = remainingBytes > 1400? 1400:remainingBytes;
          
          receivedBytes += fSocket.receive(boost::asio::buffer(pSocketBuffer+receivedBytes, remainingBytes));
          
        }
      
      
    } catch (boost::system::system_error& x) {
      std::cerr  << " A socket read error should be treated as END_OF_DATA.\n";
      return false;
    }
    
    
    ////////////////////////////////////
    // Unpack the data.
    
    // Need to convert the data to host byte order from network byte order
    // (event size has already been done).
    /*
      size_t count_check = 0;
      for (eb_t::iterator
      i = fSocketBuffer.begin() + 1,
      e = i + in_elements(fSocketBuffer[0]);
      i != e;
      ++i, ++count_check) {
      *i = ntohl(*i);
      }
      
      #ifndef NOVA_DEBUG_STREAM_ACK
      std::cerr << "Converted " << in_bytes(count_check) << " bytes.\n";
      #endif // NOVA_DEBUG_STREAM_ACK
    */
    // NOTE: As of OnlineUnpack.cxx r1.9, the histogram pointers required
    // in the constructor are *not* filled. In the event that one *does*
    // want this kind of information, the data to be histogrammed should
    // be put into event products (not necessarily written out to file)
    // and turned into histograms by a downstream analyzer. This will
    // enable easy management of different histograms for diferent runs,
    // subruns, etc, etc and trivial control of whether histograns are
    // actually wanted or not.
    daq2raw::OnlineUnpack up(nullptr, nullptr, nullptr, &fCurrentFilename);
    up.fFilterCorruptedNanoslices = fFilterCorruptedNanoslices;
    
    
    /// Define data structures, which will be filled by OnlineUnpack::ProcessRawEvent
    std::unique_ptr<std::vector<rawdata::RawDigit> > digits(new std::vector<rawdata::RawDigit>);
    rawdata::RawTrigger                            raw_trigger;
    rawdata::RawSumDropMB                            raw_sumDropMB;
    rawdata::RawSumDCM                            raw_sumDCM;
    rawdata::FlatDAQData                           flat_daq_data;
    rawdata::DAQHeader                             daq_header;
    
    // Fill the data structures for this event.
    // MFP: Who is responsible for deleting the vector 'digits'? At what point
    // is the responsibility for doing so accepted?
    // ProcessRawEvent expects the first argument to be the first word of *data* (not the size
    // of the event data), thus the +1 in addressing the vector.
    bool unpackok = false;
    try {
      unpackok = up.ProcessRawEvent(&fSocketBuffer[EvtStartIDX+1], 
                                    digits.get(),
                                    &raw_trigger,
                                    &raw_sumDropMB,
                                    &raw_sumDCM,
                                    &flat_daq_data, &daq_header);
      if (!unpackok) {
        std::cerr << __FILE__ << ":" << __LINE__ 
                  << " Error: up.ProcessRawEvent(...) unpack failed" 
                  << std::endl;
        if (gsABORT_ON & gsUNPACK_RETURN) abort();
      }
    }
    catch (...) {
      std::cerr << __FILE__ << ":" << __LINE__ 
                << " Error: Caught exception thrown from up.ProcessRawEvent(...)" << std::endl;
      unpackok = false;
      if (gsABORT_ON & gsUNPACK_EXCEPTION) abort();
    }
    
    if (!unpackok)
      {
        std::cerr  << " ProcessRawEvent failed.\n";
        try{
          //drain leftovers
          // fSocket.receive(boost::asio::buffer(&fSocketBuffer[EvtStartIDX], fCurrentEventCapacity ));
          //
          // messier@indiana.edu - This "fSocket.receive" never
          // completes. My tests show that maybe its not needed. I've
          // commented it out until its better understood.
        } catch (boost::system::system_error& x) {
          std::cerr  << " A socket read error should be treated as END_OF_DATA.\n";
          return false;
        }
        goto retryLOOP; //FIXME: server code needs to use double buffering
        
        return false;
      }
    
    // Get the actual timestamp here - use the current time as the
    // default, maybe the default should really be a ridiculous value
    struct timespec ts;
    
    novadaq::timeutils::
      convertNovaTimeToUnixTime(raw_trigger.fTriggerTimingMarker_TimeStart, ts);
    
    // art::Timestamp has the seconds since the beginning of the epoch
    // in the upper 32 bits and the fractional seconds in the lower 32
    // bits.
    art::Timestamp tstamp = (ts.tv_sec << 32) | ts.tv_nsec;
    
    ////////////////////////////////////
    // Assemble new principals as required.
    art::SubRunID newID(up.getRunNumber(), up.getSubrunNumber());
    if (inR == nullptr || fCurrentSubRunID.runID() != newID.runID()) { // New Run fragment
      outR = fSourceHelper.makeRunPrincipal(up.getRunNumber(), tstamp);
      // get the base form of the geometry file for the current detector
      //\todo at some point we should query a database to get this, especially
      //\todo when we have instances of only part of a detector working
      std::string detFileBase = getDetectorGDML(up.getDetId());
      std::unique_ptr<sumdata::RunData> rundata(new sumdata::RunData(novadaq::cnv::DetId(up.getDetId()), detFileBase));
      art::put_product_in_principal(std::move(rundata),
                                    *outR,
                                    "daq"); // Module label
    }
    if (inSR == nullptr || fCurrentSubRunID != newID) { // New SubRun fragment
      outSR = fSourceHelper.makeSubRunPrincipal(up.getRunNumber(),
                                                  up.getSubrunNumber(),
                                                  tstamp);
      fCurrentSubRunID = newID;
    }
    outE = fSourceHelper.makeEventPrincipal(up.getRunNumber(),
                                              up.getSubrunNumber(),
                                              up.getEventNumber(),
                                              tstamp);
    
    ////////////////////////////////////
    // Add the event products to the event:
    
    // Digitized data:
    art::put_product_in_principal(std::move(digits), *outE, "daq");
    // Trigger data:
    std::unique_ptr<std::vector<rawdata::RawTrigger> > rtcol(new std::vector<rawdata::RawTrigger>);
    rtcol->push_back(raw_trigger); // Read one, product is a vector.
    art::put_product_in_principal(std::move(rtcol),
                                  *outE,
                                  "daq"); // Module label
    
    // RawSumDropMB data:
    std::unique_ptr<std::vector<rawdata::RawSumDropMB> > rtcol_MB(new std::vector<rawdata::RawSumDropMB>);
    rtcol_MB->push_back(raw_sumDropMB); // Read one, product is a vector.
    art::put_product_in_principal(std::move(rtcol_MB),
                                  *outE,
                                  "daq"); // Module label

    // RawSumDCM data:
    std::unique_ptr<std::vector<rawdata::RawSumDCM> > rtcol_DCM(new std::vector<rawdata::RawSumDCM>);
    rtcol_DCM->push_back(raw_sumDCM); // Read one, product is a vector.
    art::put_product_in_principal(std::move(rtcol_DCM),
                                  *outE,
                                  "daq"); // Module label

    // DAQ header:
    std::unique_ptr<rawdata::DAQHeader > daqcol(new rawdata::DAQHeader(daq_header));
    art::put_product_in_principal(std::move(daqcol),
                                  *outE,
                                  "daq"); // Module label
    
    /// Put FlatDAQData object
    if(fDoFillFlatDAQData) {
      std::unique_ptr<std::vector<rawdata::FlatDAQData> > flatcol(new std::vector<rawdata::FlatDAQData>);
      flatcol->push_back(flat_daq_data); // Read one, product is a vector.
      art::put_product_in_principal(std::move(flatcol),
                                    *outE,
                                    "daq"); // Module label
  }
    
    
    return true;
  }
  
}// end of namespace