daq2raw::NOvASocketInputDriver Class Reference

#include "/cvmfs/nova-development.opensciencegrid.org/novasoft/releases/N20-09-26/DAQ2RawDigit/NOvASocketInputDriver.h"

Inheritance diagram for daq2raw::NOvASocketInputDriver:

Public Member Functions
	NOvASocketInputDriver (fhicl::ParameterSet const &pset, art::ProductRegistryHelper &help, art::SourceHelper const &pm)
void	closeCurrentFile ()
void	readFile (std::string const &name, art::FileBlock *&fb)
bool	readNext (art::RunPrincipal *const &inR, art::SubRunPrincipal *const &inSR, art::RunPrincipal *&outR, art::SubRunPrincipal *&outSR, art::EventPrincipal *&outE)

Protected Member Functions
std::string	getDetectorGDML (const int detid) const

Private Types
typedef std::vector< uint32_t >	eb_t
	Event buffer type. More...
typedef eb_t::value_type	ebe_t
	Event buffer element type. More...
typedef std::vector< uint8_t >	cb_t
	Command buffer type. More...

Static Private Member Functions
static size_t	in_bytes (size_t nElements)
static size_t	in_elements (size_t nBytes)

Private Attributes
art::SourceHelper const &	fSourceHelper
	Class to help with {Run, SubRun, Event}Principal construction. More...
std::string	fCurrentFilename
	Hang on to the current filename. More...
art::SubRunID	fCurrentSubRunID
	Keep track of the current subRun details. More...
boost::asio::io_service	fIoService
	Object representing IO services of the OS. More...
boost::asio::ip::tcp::socket	fSocket
	Object representing the socket from which we read. More...
size_t	fCurrentEventCapacity
	Current event capacity in buffer elements (current buffer size is 1 more than this);. More...
size_t	fMaxEventSizeBytes
	Will not try to read an event bigger than this. More...
std::vector< uint32_t >	fSocketBuffer
	Raw data buffer for socket communication. More...
bool	fFilterCorruptedNanoslices
	Filter the corrupted nanaslices in the unpacking? More...
bool	fDoFillFlatDAQData
	Do we need to fill FlatDAQData object? More...
int	fMaxReadAttempts
	Maximum number of read attempts before we throw. More...
int	fCorruptEventCount
	Counter for the total number of corrupt events encountered. More...
int	fConsecutiveCorruptEventCount
	Counter for the number of consecutive corrupt events encountered. More...
int	fBadReadCount
	Counter for the total number of corrupt events encountered. More...
int	fConsecutiveBadReadCount
	Counter for the number of consecutive corrupt events encountered. More...
uint32_t	trigmask_word1
	Lowest word in the trigger mask only the lower 24 bit are used. More...
uint32_t	trigmask_word2
	Middle word (bits 31-63) in the trigger mask. More...
uint32_t	trigmask_word3
	High word (bits 64-96) in the trigger mask. More...
int	fTransmitDelay
	Delay in microseconds to wait between command transmission. More...

Detailed Description

Definition at line 49 of file NOvASocketInputDriver.h.

Member Typedef Documentation

typedef std::vector<uint8_t> daq2raw::NOvASocketInputDriver::cb_t

private

Command buffer type.

Definition at line 71 of file NOvASocketInputDriver.h.

typedef std::vector<uint32_t> daq2raw::NOvASocketInputDriver::eb_t

private

Event buffer type.

Definition at line 69 of file NOvASocketInputDriver.h.

typedef eb_t::value_type daq2raw::NOvASocketInputDriver::ebe_t

private

Event buffer element type.

Definition at line 70 of file NOvASocketInputDriver.h.

Constructor & Destructor Documentation

daq2raw::NOvASocketInputDriver::NOvASocketInputDriver	(	fhicl::ParameterSet const &	pset,
		art::ProductRegistryHelper &	help,
		art::SourceHelper const &	pm
	)

Class to fill the constraints on a template argument to the class, art::Source

Definition at line 68 of file NOvASocketInputDriver.cxx.

References fDoFillFlatDAQData, art::InEvent, art::InRun, and art::ProductRegistryHelper::reconstitutes().

    : 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");
  }

Member Function Documentation

void daq2raw::NOvASocketInputDriver::closeCurrentFile

(

)

Definition at line 109 of file NOvASocketInputDriver.cxx.

References fSocket.

  {
    // Our "file" is a socket.
    fSocket.close();
  }

std::string daq2raw::NOvAInputDriverBase::getDetectorGDML ( const int  detid ) const

protectedinherited

Definition at line 22 of file NOvAInputDriverBase.cxx.

References daq2raw::NOvAInputDriverBase::fGDMLBase, novadaq::cnv::kFARDET, novadaq::cnv::kNDOS, novadaq::cnv::kNEARDET, and novadaq::cnv::kTESTBEAM.

Referenced by daq2raw::DAQHit2Raw::beginRun(), daq2raw::NOvARawInputDriver::readNext(), readNext(), and daq2raw::TestBeamRawInputDriver::readNext().

                                                                     {
    
    // get the base form of the geometry file for the current detector
    if(!fGDMLBase.empty()) return fGDMLBase;

    //\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("");
    if     (detid == novadaq::cnv::kNDOS)
      detFileBase = "ndos-2x3-4block_nomc-xtru-vacuum";
    else if     (detid == novadaq::cnv::kNEARDET)
      detFileBase = "neardet-3x3-8block-xtru-vacuum-stagger";
    else if(detid == novadaq::cnv::kFARDET)
      detFileBase = "fardet-12x12-28block-xtru-vacuum-stagger-pivoter";
    else if(detid == novadaq::cnv::kTESTBEAM)
      detFileBase = "testbeam-2x2-2block-xtru-vacuum-stagger";

    return detFileBase;
  }

static size_t daq2raw::NOvASocketInputDriver::in_bytes ( size_t  nElements )

inlinestaticprivate

Definition at line 73 of file NOvASocketInputDriver.h.

Referenced by readNext().

                                           {
    return nElements * sizeof(ebe_t);
  }

static size_t daq2raw::NOvASocketInputDriver::in_elements ( size_t  nBytes )

inlinestaticprivate

Definition at line 77 of file NOvASocketInputDriver.h.

References fillBadChanDBTables::result.

                                           {
    size_t result = nBytes / sizeof(ebe_t);
    return (in_bytes(result) == nBytes)?
      result:
      result+1;
  }

void daq2raw::NOvASocketInputDriver::readFile	(	std::string const &	name,
		art::FileBlock *&	fb
	)

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.

Definition at line 123 of file NOvASocketInputDriver.cxx.

References om::cerr, run_hadd::cmd, DispatcherCMD::DSPCMD_SET_EVENTSTREAM_SINGLE, DispatcherCMD::DSPCMD_SET_TRIGGER_MASK_1, DispatcherCMD::DSPCMD_SET_TRIGGER_MASK_2, DispatcherCMD::DSPCMD_SET_TRIGGER_MASK_3, e, febshutoff_auto::end, fCurrentFilename, art::errors::FileReadError, fIoService, fSocket, fTransmitDelay, MECModelEnuComparisons::i, ip, daq2raw::parseFilename(), port, febshutoff_auto::query, stan::math::system_error(), trigmask_word1, trigmask_word2, trigmask_word3, DispatcherCMD::write(), and submit_syst::x.

  {
    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.
      }
    }
}

bool daq2raw::NOvASocketInputDriver::readNext	(	art::RunPrincipal *const &	inR,
		art::SubRunPrincipal *const &	inSR,
		art::RunPrincipal *&	outR,
		art::SubRunPrincipal *&	outSR,
		art::EventPrincipal *&	outE
	)

Required function: read and fill Run, SubRun and/or Event as might be appropriate.

Define data structures, which will be filled by OnlineUnpack::ProcessRawEvent

Put FlatDAQData object

Definition at line 293 of file NOvASocketInputDriver.cxx.

References om::cerr, submit_syst::const, novadaq::timeutils::convertNovaTimeToUnixTime(), DispatcherCMD::DSPCMD_NEXT_EVENT, allTimeWatchdog::endl, fBadReadCount, fConsecutiveBadReadCount, fConsecutiveCorruptEventCount, fCorruptEventCount, fCurrentEventCapacity, fCurrentFilename, fCurrentSubRunID, fDoFillFlatDAQData, fFilterCorruptedNanoslices, daq2raw::OnlineUnpack::fFilterCorruptedNanoslices, art::errors::FileReadError, fMaxEventSizeBytes, fMaxReadAttempts, fSocket, fSocketBuffer, fSourceHelper, rawdata::RawTrigger::fTriggerTimingMarker_TimeStart, daq2raw::NOvAInputDriverBase::getDetectorGDML(), daq2raw::OnlineUnpack::getDetId(), daq2raw::OnlineUnpack::getEventNumber(), daq2raw::OnlineUnpack::getRunNumber(), daq2raw::OnlineUnpack::getSubrunNumber(), gsABORT_ON, gsMAGIC_WORD, gsUNPACK_EXCEPTION, gsUNPACK_RETURN, in_bytes(), art::SourceHelper::makeEventPrincipal(), art::SourceHelper::makeRunPrincipal(), art::SourceHelper::makeSubRunPrincipal(), printf(), daq2raw::OnlineUnpack::ProcessRawEvent(), art::put_product_in_principal(), art::SubRunID::runID(), stan::math::system_error(), msf_helper::timespec, DispatcherCMD::write(), and submit_syst::x.

                                                                 {
    

    // 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;
  }

Member Data Documentation

int daq2raw::NOvASocketInputDriver::fBadReadCount

private

Counter for the total number of corrupt events encountered.

Definition at line 100 of file NOvASocketInputDriver.h.

Referenced by readNext().

int daq2raw::NOvASocketInputDriver::fConsecutiveBadReadCount

private

Counter for the number of consecutive corrupt events encountered.

Definition at line 101 of file NOvASocketInputDriver.h.

Referenced by readNext().

int daq2raw::NOvASocketInputDriver::fConsecutiveCorruptEventCount

private

Counter for the number of consecutive corrupt events encountered.

Definition at line 99 of file NOvASocketInputDriver.h.

Referenced by readNext().

int daq2raw::NOvASocketInputDriver::fCorruptEventCount

private

Counter for the total number of corrupt events encountered.

Definition at line 98 of file NOvASocketInputDriver.h.

Referenced by readNext().

size_t daq2raw::NOvASocketInputDriver::fCurrentEventCapacity

private

Current event capacity in buffer elements (current buffer size is 1 more than this);.

Definition at line 89 of file NOvASocketInputDriver.h.

Referenced by readNext().

std::string daq2raw::NOvASocketInputDriver::fCurrentFilename

private

Hang on to the current filename.

Definition at line 85 of file NOvASocketInputDriver.h.

Referenced by readFile(), and readNext().

art::SubRunID daq2raw::NOvASocketInputDriver::fCurrentSubRunID

private

Keep track of the current subRun details.

Definition at line 86 of file NOvASocketInputDriver.h.

Referenced by readNext().

bool daq2raw::NOvASocketInputDriver::fDoFillFlatDAQData

private

Do we need to fill FlatDAQData object?

Definition at line 94 of file NOvASocketInputDriver.h.

Referenced by NOvASocketInputDriver(), and readNext().

bool daq2raw::NOvASocketInputDriver::fFilterCorruptedNanoslices

private

Filter the corrupted nanaslices in the unpacking?

Definition at line 93 of file NOvASocketInputDriver.h.

Referenced by readNext().

boost::asio::io_service daq2raw::NOvASocketInputDriver::fIoService

private

Object representing IO services of the OS.

Definition at line 87 of file NOvASocketInputDriver.h.

Referenced by readFile().

size_t daq2raw::NOvASocketInputDriver::fMaxEventSizeBytes

private

Will not try to read an event bigger than this.

Definition at line 90 of file NOvASocketInputDriver.h.

Referenced by readNext().

int daq2raw::NOvASocketInputDriver::fMaxReadAttempts

private

Maximum number of read attempts before we throw.

Definition at line 96 of file NOvASocketInputDriver.h.

Referenced by readNext().

boost::asio::ip::tcp::socket daq2raw::NOvASocketInputDriver::fSocket

private

Object representing the socket from which we read.

Definition at line 88 of file NOvASocketInputDriver.h.

Referenced by closeCurrentFile(), readFile(), and readNext().

std::vector<uint32_t> daq2raw::NOvASocketInputDriver::fSocketBuffer

private

Raw data buffer for socket communication.

Definition at line 91 of file NOvASocketInputDriver.h.

Referenced by readNext().

art::SourceHelper const& daq2raw::NOvASocketInputDriver::fSourceHelper

private

Class to help with {Run, SubRun, Event}Principal construction.

Definition at line 84 of file NOvASocketInputDriver.h.

Referenced by readNext().

int daq2raw::NOvASocketInputDriver::fTransmitDelay

private

Delay in microseconds to wait between command transmission.

Definition at line 107 of file NOvASocketInputDriver.h.

Referenced by readFile().

uint32_t daq2raw::NOvASocketInputDriver::trigmask_word1

private

Lowest word in the trigger mask only the lower 24 bit are used.

Definition at line 103 of file NOvASocketInputDriver.h.

Referenced by readFile().

uint32_t daq2raw::NOvASocketInputDriver::trigmask_word2

private

Middle word (bits 31-63) in the trigger mask.

Definition at line 104 of file NOvASocketInputDriver.h.

Referenced by readFile().

uint32_t daq2raw::NOvASocketInputDriver::trigmask_word3

private

High word (bits 64-96) in the trigger mask.

Definition at line 105 of file NOvASocketInputDriver.h.

Referenced by readFile().

---

The documentation for this class was generated from the following files:

• /cvmfs/nova-development.opensciencegrid.org/novasoft/releases/N20-09-26/DAQ2RawDigit/NOvASocketInputDriver.h
• /cvmfs/nova-development.opensciencegrid.org/novasoft/releases/N20-09-26/DAQ2RawDigit/NOvASocketInputDriver.cxx