TPC/groot/prdfoStream.cc

0001 #include <prdfoStream.h>
0002
0003 #include <unistd.h>
0004 #include <sys/types.h>
0005 #include <sys/stat.h>
0006 #include <fcntl.h>
0007
0008 #include "EvtConstants.h"
0009
0010 using namespace std;
0011
0012 // the constructor first ----------------
0013 prdfoStream::prdfoStream (const char *filename, const int length)
0014 {
0015   buffer_sequence = 1;
0016
0017   _filename = filename;
0018
0019   int *b = new int [length];
0020   bptr = ( buffer_ptr ) b;
0021
0022   data_ptr = &(bptr->data[0]);
0023   max_length = length;   // in 32bit units
0024   max_size = max_length;
0025   bptr->ID = -64;
0026   bptr->Bufseq = buffer_sequence;
0027
0028   bptr->Runnr = 0;
0029   has_buffer = 0;
0030   defunct = 0;
0031
0032   // test if the file exists, do not overwrite
0033   fd =  open(_filename.c_str(), O_WRONLY | O_CREAT | O_EXCL | O_LARGEFILE ,
0034           S_IRWXU | S_IROTH | S_IRGRP );
0035   if (fd < 0)
0036     {
0037       cout << " error opening file " << _filename << endl;
0038       perror ( _filename.c_str());
0039       fd = 0;
0040       defunct = 1;
0041     }
0042
0043 }
0044
0045
0046 // the destructor... ----------------
0047 prdfoStream::~prdfoStream ()
0048 {
0049   writeout();
0050   int *b = (int *)  bptr;
0051   delete [] b;
0052   if (fd) close (fd);
0053 }
0054
0055
0056
0057 int prdfoStream::prepare_next( const int iseq
0058               , const int irun)
0059 {
0060   if (defunct)
0061     {
0062       cout << "defunct oStream object " << endl;
0063       return -1;
0064     }
0065
0066   // re-initialize the event header length
0067   bptr->Length =  BUFFERHEADERLENGTH*4;
0068   bptr->ID = -64;
0069   bptr->Bufseq = iseq;
0070   if (irun>0) bptr->Runnr = irun;
0071
0072   current_index = 0;
0073   left = max_size - BUFFERHEADERLENGTH - EOBLENGTH;
0074   has_end = 0;
0075   has_buffer =1;
0076   return 0;
0077 }
0078
0079 // ---------------------------------------------------------
0080 int prdfoStream::addEvent(Event * evt)
0081 {
0082   if (defunct)
0083     {
0084       cout << "defunct oStream object " << endl;
0085       return -1;
0086     }
0087
0088   if ( ! has_buffer) prepare_next (buffer_sequence++, evt->getRunNumber() );
0089
0090   int evtsize = evt->getEvtLength();
0091
0092   if (evtsize*4 > left-EOBLENGTH)
0093     {
0094       //cout << "++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++Prdf buffer being written" << endl;
0095       writeout();
0096     }
0097
0098   int nw;
0099   evt->Copy( &(bptr->data[current_index]), evtsize, &nw);
0100   left -= evtsize;
0101   current_index += evtsize;
0102   bptr->Length  += evtsize*4;
0103   return 0;
0104 }
0105
0106
0107 // ----------------------------------------------------------
0108 int prdfoStream::addEoB()
0109 {
0110   if (has_end) return -1;
0111   bptr->data[current_index++] = 2;
0112   bptr->data[current_index++] = 0;
0113   bptr->Length  += 2*4;
0114
0115   has_end = 1;
0116   //  cout << "addind EOB" << endl;
0117   return 0;
0118 }
0119
0120 unsigned int prdfoStream::writeout ()
0121 {
0122   if ( bptr->Length ==  BUFFERHEADERLENGTH*4) return 0;
0123   if (!has_end) addEoB();
0124
0125   unsigned int bytes;
0126
0127   int blockcount = ( bptr->Length + 8192 -1)/8192;
0128   int bytecount = blockcount*8192;
0129   bytes = writen ( fd, (char *) bptr , bytecount );
0130   has_buffer = 0;
0131   return bytes;
0132 }
0133
0134
0135 unsigned int prdfoStream::writen (int fd, char *ptr, const unsigned int nbytes)
0136 {
0137
0138   unsigned int nleft, nwritten;
0139   nleft = nbytes;
0140   while ( nleft>0 )
0141     {
0142       nwritten = write (fd, ptr, nleft);
0143       if ( nwritten < 0 )
0144     return nwritten;
0145
0146       nleft -= nwritten;
0147       ptr += nwritten;
0148     }
0149   return (nbytes-nleft);
0150 }