EventQueue.cpp 4.25 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48
#include "EventQueue.h"
#include "commonDefs.h"

#include <iostream>

using namespace std;

UInt_t EventQueue::addData(UInt_t * i4dat, UInt_t n4dat, Int_t tstCorr)
{
  if(EodInserted) // no data can be inserted on a closed queue
    return 0;     // in which case the data is simply discarded

  //UInt_t wrong = 0;
  //UInt_t count1 = countEvents();
  if(indNext+n4dat >= validSize) {
    // try to reshuffle
    if(indFirst && indNext) {
      endCount++;
      int mm = 0;
      for(UInt_t nn = indFirst; nn < indNext; nn++) {
        data[mm++] = data[nn];
      }
      indLast -= indFirst;
      indFirst = 0;
      indNext  = mm;
    }
    //UInt_t count2 = countEvents();
    //if(count2 != count1)
    //  wrong++;
    // will try instead to take what is possible
    //if(indNext+n4dat > queueSize)
    //  return false;
  }

  UInt_t i4Add = 0;
  UInt_t added = 0;
  while(n4dat > 0) {
    int i4len = i4dat[0]/4;
    if( i4dat[1] == inpKey ) {      //copy only valid frames
      if(indNext+i4len >= validSize) {
        break;    // input queue full
      }
      memcpy(data+indNext, i4dat, i4len*sizeof(UInt_t));
      indLast  = indNext;
      indNext += i4len;
      i4Add   += i4len;
      if(tstCorr)
        addTstamp(data + indLast + 3, tstCorr);
49 50
      if(sTstDiff)
        addTstDiff(getTstamp(data + indLast + 3));
51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163
      numEvts++;
      added++;
    }
    else {
      // We should check here if this frame is of interest (e.g. conf: or meta:eof)
      // If we allow it to go through (as we probably should) it has to be rechecked in EventBuilder::process_block
      // If it is meta:eof let's change its evnumber and timestamp to maxval 
      //if(i4dat[1] == 0xFA001100 && i4dat[0] == 20) {
      //  if(indNext+i4len >= validSize) {
      //    break;    // input queue full
      //  }
      //  i4dat[2] = 0xFFFFFFFF;
      //  i4dat[3] = 0xFFFFFFFF;
      //  i4dat[4] = 0xFFFFFFFF;
      //  // copy the frame 
      //  memcpy(data+indNext, i4dat, i4len*sizeof(UInt_t));
      //  indLast  = indNext;
      //  indNext += i4len;
      //  i4Add   += i4len;
      //  if(tstCorr)
      //    addTstamp(data + indLast + 3, tstCorr);
      //  numEvts++;
      //  added++;
      //}
      badKeys++;
    }
    i4dat += i4len;
    n4dat -= i4len;
  }

  //UInt_t count3 = countEvents();
  //if(count3 != count1+added)
  //  wrong++;
  maxInputSize = max(maxInputSize, i4Add);
  maxQueueSize = max(maxQueueSize, indNext-indFirst);
 
  tstFirst = getTstamp(data + indFirst + 3);
  evnFirst = getEvnumb(data + indFirst + 2);
  tstLast  = getTstamp(data + indLast  + 3);
  evnLast  = getEvnumb(data + indLast  + 2);

  return n4dat;   // how much is still in the input buffer
}

UInt_t EventQueue::moveOut(UInt_t *dest)
{
  //UInt_t wrong = 0;
  //UInt_t count1 = countEvents();
  UInt_t size = data[indFirst];
  if(dest)
    memcpy(dest, data+indFirst, size);
  numEvts--;
  size /= 4;
  indFirst += size;
  if(indFirst >= indNext) {
    indFirst = indLast = indNext = numEvts = 0;
  }
  else {
    tstFirst = getTstamp(data + indFirst + 3);
    evnFirst = getEvnumb(data + indFirst + 2);
  }
  //UInt_t count2 = countEvents();
  //if(count2 != count1-1) 
  //  wrong++;
  return size;
}

UInt_t EventQueue::countEvents()
{
  UInt_t nevs = 0;
  UInt_t ipos = indFirst;
  while(ipos < indNext) {
    nevs++;
    UInt_t size = data[ipos]/4;
    ipos += size;
  }
  return nevs;
}

bool EventQueue::setEod()
{
  if(EodInserted)   // already closed
    return false;

  UInt_t i4dat[5] = {20, 0, 0xFFFFFFFF, 0xFFFFFFFF, 0xFFFFFFFF};
  i4dat[1] = inpKey;

  validSize = queueSize;
  UInt_t rest = addData(i4dat, 5, 0);
  validSize = queueSize-5;
  //if(rest) {
  //  EodInserted = false; // just to be able to set a breakpoint
  //}
  EodInserted = true;

  return true;
}

void EventQueue::Error(int ich)
{
  errCount++;
  UInt_t evcount = 0;
  UInt_t ind = indFirst;
  while(ind < indNext) {
    evcount++;
    ind += data[ind]/4;
  } 

  LOCK_COUT;
  cout << "   " << myBuilder << "::ERROR  "  << errCount << "  Flushing " 
       << evcount << " events ( " << usedSize() << " words ) in chain " << ich << endl;
  indFirst = indLast = indNext = numEvts = 0;
}