/afs/cern.ch/work/a/aaltunda/public/www/CMSSW_5_3_13_patch3/src/EventFilter/ShmBuffer/src/FUShmBuffer.cc

Go to the documentation of this file.

//
// FUShmBuffer
// -----------
//
//            15/11/2006 Philipp Schieferdecker <philipp.schieferdecker@cern.ch>


#include "DataFormats/FEDRawData/interface/FEDNumbering.h"
#include "EventFilter/FEDInterface/interface/GlobalEventNumber.h"
#include "EventFilter/ShmBuffer/interface/FUShmBuffer.h"
#include "EventFilter/Utilities/interface/Exception.h"

#include <unistd.h>
#include <iostream>
#include <string>
#include <cassert>

#include <sstream>
#include <fstream>

#include <cstdlib>
#include <cstring>
#include <stdint.h>

// the shmem keys are henceforth going to be FIXED for a give userid
// prior to creation, the application will attempt to get ownership
// of existing segments by the same key and destroy them

#define SHM_DESCRIPTOR_KEYID           1 /* Id used on ftok for 1. shmget key */
#define SHM_KEYID                      2 /* Id used on ftok for 2. shmget key */
#define SEM_KEYID                      1 /* Id used on ftok for semget key    */

#define NSKIP_MAX                    100

using namespace std;
using namespace evf;

//obsolete!!!
const char* FUShmBuffer::shmKeyPath_ =
                (getenv("FUSHM_KEYFILE") == NULL ? "/dev/null"
                                : getenv("FUSHM_KEYFILE"));
const char* FUShmBuffer::semKeyPath_ =
                (getenv("FUSEM_KEYFILE") == NULL ? "/dev/null"
                                : getenv("FUSEM_KEYFILE"));

// construction/destruction

//______________________________________________________________________________
FUShmBuffer::FUShmBuffer(bool segmentationMode, unsigned int nRawCells,
                unsigned int nRecoCells, unsigned int nDqmCells,
                unsigned int rawCellSize, unsigned int recoCellSize,
                unsigned int dqmCellSize) :
        segmentationMode_(segmentationMode), nClientsMax_(128),
                        nRawCells_(nRawCells), rawCellPayloadSize_(rawCellSize),
                        nRecoCells_(nRecoCells), recoCellPayloadSize_(recoCellSize),
                        nDqmCells_(nDqmCells), dqmCellPayloadSize_(dqmCellSize) {
        rawCellTotalSize_ = FUShmRawCell::size(rawCellPayloadSize_);
        recoCellTotalSize_ = FUShmRecoCell::size(recoCellPayloadSize_);
        dqmCellTotalSize_ = FUShmDqmCell::size(dqmCellPayloadSize_);

        void* addr;

        rawWriteOffset_ = sizeof(FUShmBuffer);
        addr = (void*) ((unsigned long) this + rawWriteOffset_);
        new (addr) unsigned int[nRawCells_];

        rawReadOffset_ = rawWriteOffset_ + nRawCells_ * sizeof(unsigned int);
        addr = (void*) ((unsigned long) this + rawReadOffset_);
        new (addr) unsigned int[nRawCells_];

        recoWriteOffset_ = rawReadOffset_ + nRawCells_ * sizeof(unsigned int);
        addr = (void*) ((unsigned long) this + recoWriteOffset_);
        new (addr) unsigned int[nRecoCells_];

        recoReadOffset_ = recoWriteOffset_ + nRecoCells_ * sizeof(unsigned int);
        addr = (void*) ((unsigned long) this + recoReadOffset_);
        new (addr) unsigned int[nRecoCells_];

        dqmWriteOffset_ = recoReadOffset_ + nRecoCells_ * sizeof(unsigned int);
        addr = (void*) ((unsigned long) this + dqmWriteOffset_);
        new (addr) unsigned int[nDqmCells_];

        dqmReadOffset_ = dqmWriteOffset_ + nDqmCells_ * sizeof(unsigned int);
        addr = (void*) ((unsigned long) this + dqmReadOffset_);
        new (addr) unsigned int[nDqmCells_];

        evtStateOffset_ = dqmReadOffset_ + nDqmCells_ * sizeof(unsigned int);
        addr = (void*) ((unsigned long) this + evtStateOffset_);
        new (addr) evt::State_t[nRawCells_];

        evtDiscardOffset_ = evtStateOffset_ + nRawCells_ * sizeof(evt::State_t);
        addr = (void*) ((unsigned long) this + evtDiscardOffset_);
        new (addr) unsigned int[nRawCells_];

        evtNumberOffset_ = evtDiscardOffset_ + nRawCells_ * sizeof(unsigned int);
        addr = (void*) ((unsigned long) this + evtNumberOffset_);
        new (addr) unsigned int[nRawCells_];

        evtPrcIdOffset_ = evtNumberOffset_ + nRawCells_ * sizeof(unsigned int);
        addr = (void*) ((unsigned long) this + evtPrcIdOffset_);
        new (addr) pid_t[nRawCells_];

        evtTimeStampOffset_ = evtPrcIdOffset_ + nRawCells_ * sizeof(pid_t);
        addr = (void*) ((unsigned long) this + evtTimeStampOffset_);
        new (addr) time_t[nRawCells_];

        dqmStateOffset_ = evtTimeStampOffset_ + nRawCells_ * sizeof(time_t);
        addr = (void*) ((unsigned long) this + dqmStateOffset_);
        new (addr) dqm::State_t[nDqmCells_];

        clientPrcIdOffset_ = dqmStateOffset_ + nDqmCells_ * sizeof(dqm::State_t);
        addr = (void*) ((unsigned long) this + clientPrcIdOffset_);
        new (addr) pid_t[nClientsMax_];

        rawCellOffset_ = clientPrcIdOffset_ + nClientsMax_ * sizeof(pid_t);

        if (segmentationMode_) {
                recoCellOffset_ = rawCellOffset_ + nRawCells_ * sizeof(key_t);
                dqmCellOffset_ = recoCellOffset_ + nRecoCells_ * sizeof(key_t);
                addr = (void*) ((unsigned long) this + rawCellOffset_);
                new (addr) key_t[nRawCells_];
                addr = (void*) ((unsigned long) this + recoCellOffset_);
                new (addr) key_t[nRecoCells_];
                addr = (void*) ((unsigned long) this + dqmCellOffset_);
                new (addr) key_t[nDqmCells_];
        } else {
                recoCellOffset_ = rawCellOffset_ + nRawCells_ * rawCellTotalSize_;
                dqmCellOffset_ = recoCellOffset_ + nRecoCells_ * recoCellTotalSize_;
                for (unsigned int i = 0; i < nRawCells_; i++) {
                        addr = (void*) ((unsigned long) this + rawCellOffset_ + i
                                        * rawCellTotalSize_);
                        new (addr) FUShmRawCell(rawCellSize);
                }
                for (unsigned int i = 0; i < nRecoCells_; i++) {
                        addr = (void*) ((unsigned long) this + recoCellOffset_ + i
                                        * recoCellTotalSize_);
                        new (addr) FUShmRecoCell(recoCellSize);
                }
                for (unsigned int i = 0; i < nDqmCells_; i++) {
                        addr = (void*) ((unsigned long) this + dqmCellOffset_ + i
                                        * dqmCellTotalSize_);
                        new (addr) FUShmDqmCell(dqmCellSize);
                }
        }
}

//______________________________________________________________________________
FUShmBuffer::~FUShmBuffer() {

}

// implementation of member functions


//______________________________________________________________________________
void FUShmBuffer::initialize(unsigned int shmid, unsigned int semid) {
        shmid_ = shmid;
        semid_ = semid;

        if (segmentationMode_) {
                int shmKeyId = 666;
                key_t* keyAddr = (key_t*) ((unsigned long) this + rawCellOffset_);
                for (unsigned int i = 0; i < nRawCells_; i++) {
                        *keyAddr = ftok(shmKeyPath_, shmKeyId++);
                        int shmid = shm_create(*keyAddr, rawCellTotalSize_);
                        void* shmAddr = shm_attach(shmid);
                        new (shmAddr) FUShmRawCell(rawCellPayloadSize_);
                        shmdt(shmAddr);
                        ++keyAddr;
                }
                keyAddr = (key_t*) ((unsigned long) this + recoCellOffset_);
                for (unsigned int i = 0; i < nRecoCells_; i++) {
                        *keyAddr = ftok(shmKeyPath_, shmKeyId++);
                        int shmid = shm_create(*keyAddr, recoCellTotalSize_);
                        void* shmAddr = shm_attach(shmid);
                        new (shmAddr) FUShmRecoCell(recoCellPayloadSize_);
                        shmdt(shmAddr);
                        ++keyAddr;
                }
                keyAddr = (key_t*) ((unsigned long) this + dqmCellOffset_);
                for (unsigned int i = 0; i < nDqmCells_; i++) {
                        *keyAddr = ftok(shmKeyPath_, shmKeyId++);
                        int shmid = shm_create(*keyAddr, dqmCellTotalSize_);
                        void* shmAddr = shm_attach(shmid);
                        new (shmAddr) FUShmDqmCell(dqmCellPayloadSize_);
                        shmdt(shmAddr);
                        ++keyAddr;
                }
        }

        reset(true);
}

//______________________________________________________________________________
void FUShmBuffer::reset(bool shm_detach) {
        nClients_ = 0;


        for (unsigned int i = 0; i < nRawCells_; i++) {
                FUShmRawCell* cell = rawCell(i);
                cell->initialize(i);
                if (segmentationMode_ && shm_detach)
                        shmdt(cell);
        }

        for (unsigned int i = 0; i < nRecoCells_; i++) {
                FUShmRecoCell* cell = recoCell(i);
                cell->initialize(i);
                if (segmentationMode_ && shm_detach)
                        shmdt(cell);
        }

        for (unsigned int i = 0; i < nDqmCells_; i++) {
                FUShmDqmCell* cell = dqmCell(i);
                cell->initialize(i);
                if (segmentationMode_ && shm_detach)
                        shmdt(cell);
        }


        // setup ipc semaphores
        sem_init(0, 1); // lock (binary)
        sem_init(1, nRawCells_); // raw  write semaphore
        sem_init(2, 0); // raw  read  semaphore
        sem_init(3, 1); // binary semaphore to schedule raw event for discard
        sem_init(4, 0); // binary semaphore to discard raw event
        sem_init(5, nRecoCells_);// reco write semaphore
        sem_init(6, 0); // reco send (read) semaphore
        sem_init(7, nDqmCells_); // dqm  write semaphore
        sem_init(8, 0); // dqm  send (read) semaphore

        sem_print();

        unsigned int *iWrite, *iRead;

        rawWriteNext_ = 0;
        rawWriteLast_ = 0;
        rawReadNext_ = 0;
        rawReadLast_ = 0;
        iWrite = (unsigned int*) ((unsigned long) this + rawWriteOffset_);
        iRead = (unsigned int*) ((unsigned long) this + rawReadOffset_);
        for (unsigned int i = 0; i < nRawCells_; i++) {
                *iWrite++ = i;
                *iRead++ = 0xffffffff;
        }

        recoWriteNext_ = 0;
        recoWriteLast_ = 0;
        recoReadNext_ = 0;
        recoReadLast_ = 0;
        iWrite = (unsigned int*) ((unsigned long) this + recoWriteOffset_);
        iRead = (unsigned int*) ((unsigned long) this + recoReadOffset_);
        for (unsigned int i = 0; i < nRecoCells_; i++) {
                *iWrite++ = i;
                *iRead++ = 0xffffffff;
        }

        dqmWriteNext_ = 0;
        dqmWriteLast_ = 0;
        dqmReadNext_ = 0;
        dqmReadLast_ = 0;
        iWrite = (unsigned int*) ((unsigned long) this + dqmWriteOffset_);
        iRead = (unsigned int*) ((unsigned long) this + dqmReadOffset_);
        for (unsigned int i = 0; i < nDqmCells_; i++) {
                *iWrite++ = i;
                *iRead++ = 0xffffffff;
        }

        for (unsigned int i = 0; i < nRawCells_; i++) {
                setEvtState(i, evt::EMPTY);
                setEvtDiscard(i, 0);
                setEvtNumber(i, 0xffffffff);
                setEvtPrcId(i, 0);
                setEvtTimeStamp(i, 0);
        }

        for (unsigned int i = 0; i < nDqmCells_; i++)
                setDqmState(i, dqm::EMPTY);
}

//______________________________________________________________________________
unsigned int FUShmBuffer::nbRawCellsToWrite() const {
        return semctl(semid(), 1, GETVAL);
}

//______________________________________________________________________________
int FUShmBuffer::nbRawCellsToRead() const {
        return semctl(semid(), 2, GETVAL);
}

//______________________________________________________________________________
FUShmRawCell* FUShmBuffer::rawCellToWrite() {
        if (waitRawWrite() != 0)
                return 0;
        unsigned int iCell = nextRawWriteIndex();
        FUShmRawCell* cell = rawCell(iCell);
        evt::State_t state = evtState(iCell);
        if(!(state == evt::EMPTY)) {
          stringstream details;
          details << "state==evt::EMPTY assertion failed! Actual state is " << state << ", iCell = " << iCell;
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        lock();
        setEvtState(iCell, evt::RAWWRITING,false);
        setEvtDiscard(iCell, 1,false,false);
        unlock();
        return cell;
}

//______________________________________________________________________________
FUShmRawCell* FUShmBuffer::rawCellToRead() {
        waitRawRead();
        unsigned int iCell = nextRawReadIndex();
        FUShmRawCell* cell = rawCell(iCell);
        evt::State_t state = evtState(iCell);
        if(!(state == evt::RAWWRITTEN || state == evt::EMPTY || 
             state == evt::STOP || state == evt::LUMISECTION))
        {
          stringstream details;
          details
                << "state==evt::RAWWRITTEN ||state==evt::EMPTY ||state==evt::STOP"
                << "||state==evt::LUMISECTION assertion failed! Actual state is "
                << state << ", iCell = " << iCell;
          XCEPT_ASSERT(false,evf::Exception, details.str());
        }
        if (state == evt::RAWWRITTEN) {
                setEvtPrcId(iCell, getpid());
                setEvtState(iCell, evt::RAWREADING);
        }
        return cell;
}

//______________________________________________________________________________
FUShmRecoCell* FUShmBuffer::recoCellToRead() {
        waitRecoRead();
        unsigned int iCell = nextRecoReadIndex();
        FUShmRecoCell* cell = recoCell(iCell);
        unsigned int iRawCell = cell->rawCellIndex();
        if (iRawCell < nRawCells_) {
                //evt::State_t   state=evtState(iRawCell);
                //XCEPT_ASSERT(state==evt::RECOWRITTEN, evf::Exception, "state==evt::RECOWRITTEN assertion failed!");
                setEvtState(iRawCell, evt::SENDING);
        }
        return cell;
}

//______________________________________________________________________________
FUShmDqmCell* FUShmBuffer::dqmCellToRead() {
        waitDqmRead();
        unsigned int iCell = nextDqmReadIndex();
        FUShmDqmCell* cell = dqmCell(iCell);
        dqm::State_t state = dqmState(iCell);
        
        if(!(state == dqm::WRITTEN || state == dqm::EMPTY)) {
          stringstream details;
          details << "state==dqm::WRITTEN || state==dqm::EMPTY assertion failed! Actual state is "
                  << state << ", iCell = " << iCell;
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }

        if (state == dqm::WRITTEN)
                setDqmState(iCell, dqm::SENDING);
        return cell;
}

//______________________________________________________________________________
FUShmRawCell* FUShmBuffer::rawCellToDiscard() {
        waitRawDiscarded();
        FUShmRawCell* cell = rawCell(rawDiscardIndex_);
        evt::State_t state = evtState(cell->index());
        if(!(state == evt::PROCESSED || state == evt::SENT || 
             state == evt::EMPTY || state == evt::STOP ||
             state == evt::USEDLS)) {
          stringstream details;
          details << "state==evt::PROCESSED || state==evt::SENT || "
                  << "state==evt::EMPTY || state==evt::STOP || "
                  << "state==evt::USEDLS assertion failed! Actual state is "
                  << state << ", index = " << cell->index();
          XCEPT_ASSERT(false,evf::Exception,details.str());
        }
        if (state != evt::EMPTY && state != evt::USEDLS && state != evt::STOP)
                setEvtState(cell->index(), evt::DISCARDING);
        return cell;
}

//______________________________________________________________________________
void FUShmBuffer::finishWritingRawCell(FUShmRawCell* cell) {
        evt::State_t state = evtState(cell->index());
        if(!(state == evt::RAWWRITING)) {
          stringstream details;
          details << "state==evt::RAWWRITING assertion failed! Actual state is "
          << state << ", index = " << cell->index();
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        setEvtState(cell->index(), evt::RAWWRITTEN);
        setEvtNumber(cell->index(), cell->evtNumber());
        postRawIndexToRead(cell->index());
        if (segmentationMode_)
                shmdt(cell);
        postRawRead();
}

//______________________________________________________________________________
void FUShmBuffer::finishReadingRawCell(FUShmRawCell* cell) {
        evt::State_t state = evtState(cell->index());
        if(!(state == evt::RAWREADING)) {
          stringstream details;
          details << "state==evt::RAWREADING assertion failed! Actual state is "
                  << state << ", index = " << cell->index();
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        setEvtState(cell->index(), evt::RAWREAD);
        setEvtState(cell->index(), evt::PROCESSING);
        setEvtTimeStamp(cell->index(), time(0));
        if (segmentationMode_)
                shmdt(cell);
}

//______________________________________________________________________________
void FUShmBuffer::finishReadingRecoCell(FUShmRecoCell* cell) {
        unsigned int iRawCell = cell->rawCellIndex();
        if (iRawCell < nRawCells_) {
                //evt::State_t state=evtState(cell->rawCellIndex());
                //XCEPT_ASSERT(state==evt::SENDING, evf::Exception, "state==evt::SENDING assertion failed!");
                setEvtState(cell->rawCellIndex(), evt::SENT);
        }
        if (segmentationMode_)
                shmdt(cell);
}

//______________________________________________________________________________
void FUShmBuffer::finishReadingDqmCell(FUShmDqmCell* cell) {
        dqm::State_t state = dqmState(cell->index());
        if(!(state == dqm::SENDING || state == dqm::EMPTY)) {
        stringstream details;
        details << "state==dqm::SENDING||state==dqm::EMPTY assertion failed! Actual state is "
                << state << ", index = " << cell->index();
          XCEPT_ASSERT(false, evf::Exception, details.str());

        }
        if (state == dqm::SENDING)
                setDqmState(cell->index(), dqm::SENT);
        if (segmentationMode_)
                shmdt(cell);
}

//______________________________________________________________________________
void FUShmBuffer::scheduleRawCellForDiscard(unsigned int iCell) {
        waitRawDiscard();
        if (rawCellReadyForDiscard(iCell)) {
                rawDiscardIndex_ = iCell;
                evt::State_t state = evtState(iCell);
                if(!(state == evt::PROCESSING || state == evt::SENT ||
                     state == evt::EMPTY || state == evt::STOP ||
                     state == evt::LUMISECTION || state == evt::RECOWRITTEN)) {
                  stringstream details;
                  details << "state==evt::PROCESSING||state==evt::SENT||state==evt::EMPTY||"
                          <<"state==evt::STOP||state==evt::LUMISECTION||state==evt::RECOWRITTEN assertion failed! Actual state is "
                          << state << ", iCell = " << iCell;
                  XCEPT_ASSERT( false,evf::Exception,details.str()); 
                }
                if (state == evt::PROCESSING)
                        setEvtState(iCell, evt::PROCESSED);
                if (state == evt::LUMISECTION)
                        setEvtState(iCell, evt::USEDLS);
                postRawDiscarded();
        } else
                postRawDiscard();
}

//______________________________________________________________________________
void FUShmBuffer::scheduleRawCellForDiscardServerSide(unsigned int iCell) {
        waitRawDiscard();
        if (rawCellReadyForDiscard(iCell)) {
                rawDiscardIndex_ = iCell;
                evt::State_t state = evtState(iCell);
                // UPDATE: aspataru
                if (state != evt::LUMISECTION && state != evt::EMPTY 
                    && state != evt::USEDLS && state != evt::STOP)
                        setEvtState(iCell, evt::PROCESSED);
                if (state == evt::LUMISECTION)
                        setEvtState(iCell, evt::USEDLS);
                postRawDiscarded();
        } else
                postRawDiscard();
}

//______________________________________________________________________________
void FUShmBuffer::discardRawCell(FUShmRawCell* cell) {
        releaseRawCell(cell);
        postRawDiscard();
}

//______________________________________________________________________________
void FUShmBuffer::discardRecoCell(unsigned int iCell) {
        FUShmRecoCell* cell = recoCell(iCell);
        unsigned int iRawCell = cell->rawCellIndex();
        if (iRawCell < nRawCells_) {
                //evt::State_t state=evtState(iRawCell);
                //XCEPT_ASSERT(state==evt::SENT, evf::Exception, "state==evt::SENT assertion failed!");
                scheduleRawCellForDiscard(iRawCell);
        }
        cell->clear();
        if (segmentationMode_)
                shmdt(cell);
        postRecoIndexToWrite(iCell);
        postRecoWrite();
}

//______________________________________________________________________________
void FUShmBuffer::discardOrphanedRecoCell(unsigned int iCell) {
        FUShmRecoCell* cell = recoCell(iCell);
        cell->clear();
        if (segmentationMode_)
                shmdt(cell);
        postRecoIndexToWrite(iCell);
        postRecoWrite();
}

//______________________________________________________________________________
void FUShmBuffer::discardDqmCell(unsigned int iCell) {
        dqm::State_t state = dqmState(iCell);
        if(!(state == dqm::EMPTY || state == dqm::SENT)) {
          stringstream details;
          details << "state==dqm::EMPTY||state==dqm::SENT assertion failed! Actual state is "
                  << state << ", iCell = " << iCell;
          XCEPT_ASSERT(false, evf::Exception,details.str());
        }
        setDqmState(iCell, dqm::DISCARDING);
        FUShmDqmCell* cell = dqmCell(iCell);
        cell->clear();
        if (segmentationMode_)
                shmdt(cell);
        setDqmState(iCell, dqm::EMPTY);
        postDqmIndexToWrite(iCell);
        postDqmWrite();
}

//______________________________________________________________________________
void FUShmBuffer::releaseRawCell(FUShmRawCell* cell) {
        evt::State_t state = evtState(cell->index());
        if(!( state == evt::DISCARDING || state == evt::RAWWRITING ||
              state== evt::EMPTY || state == evt::STOP ||
              /*||state==evt::LUMISECTION*/
              state == evt::USEDLS)) {
          std::cout << "=================releaseRawCell state " << state << std::endl;
          stringstream details;
          details << "state==evt::DISCARDING||state==evt::RAWWRITING||"
                  << "state==evt::EMPTY||state==evt::STOP||state==evt::USEDLS"
                  << " assertion failed! Actual state is " << state << ", index = " << cell->index();
          XCEPT_ASSERT( false, evf::Exception, details.str());
        }
        setEvtState(cell->index(), evt::EMPTY);
        setEvtDiscard(cell->index(), 0);
        setEvtNumber(cell->index(), 0xffffffff);
        setEvtPrcId(cell->index(), 0);
        setEvtTimeStamp(cell->index(), 0);
        cell->clear();
        postRawIndexToWrite(cell->index());
        if (segmentationMode_)
                shmdt(cell);
        postRawWrite();
}

//______________________________________________________________________________
void FUShmBuffer::writeRawEmptyEvent() {
        FUShmRawCell* cell = rawCellToWrite();
        if (cell == 0)
                return;
        evt::State_t state = evtState(cell->index());
        if(!(state == evt::RAWWRITING)) {
          stringstream details;
          details << "state==evt::RAWWRITING assertion failed! Actual state is "
                  << state << ", index = " << cell->index();
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        setEvtState(cell->index(), evt::STOP);
        cell->setEventTypeStopper();
        postRawIndexToRead(cell->index());
        if (segmentationMode_)
                shmdt(cell);
        postRawRead();
}

//______________________________________________________________________________
void FUShmBuffer::writeRawLumiSectionEvent(unsigned int ls) {
        FUShmRawCell* cell = rawCellToWrite();
        if (cell == 0)
                return;
        cell->setLumiSection(ls);
        evt::State_t state = evtState(cell->index());
        if (!(state == evt::RAWWRITING)) {
          stringstream details;
          details << "state==evt::RAWWRITING assertion failed! Actual state is "
                  << state << ", index = " << cell->index();
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        setEvtNumber(cell->index(),0xfffffffe);
        setEvtState(cell->index(), evt::LUMISECTION);
        cell->setEventTypeEol();
        postRawIndexToRead(cell->index());
        if (segmentationMode_)
                shmdt(cell);
        postRawRead();
}

//______________________________________________________________________________
void FUShmBuffer::writeRecoEmptyEvent() {
        waitRecoWrite();
        unsigned int iCell = nextRecoWriteIndex();
        FUShmRecoCell* cell = recoCell(iCell);
        cell->clear();
        postRecoIndexToRead(iCell);
        if (segmentationMode_)
                shmdt(cell);
        postRecoRead();
}

//______________________________________________________________________________
void FUShmBuffer::writeDqmEmptyEvent() {
        waitDqmWrite();
        unsigned int iCell = nextDqmWriteIndex();
        FUShmDqmCell* cell = dqmCell(iCell);
        cell->clear();
        postDqmIndexToRead(iCell);
        if (segmentationMode_)
                shmdt(cell);
        postDqmRead();
}

//______________________________________________________________________________
void FUShmBuffer::scheduleRawEmptyCellForDiscard() {
        FUShmRawCell* cell = rawCellToWrite();
        if (cell == 0)
                return;
        rawDiscardIndex_ = cell->index();
        setEvtState(cell->index(), evt::STOP);
        cell->setEventTypeStopper();
        setEvtNumber(cell->index(), 0xffffffff);
        setEvtPrcId(cell->index(), 0);
        setEvtTimeStamp(cell->index(), 0);
        postRawDiscarded();
}

//______________________________________________________________________________
bool FUShmBuffer::scheduleRawEmptyCellForDiscard(FUShmRawCell* cell, bool & pidstatus) {
        waitRawDiscard();
        if (rawCellReadyForDiscard(cell->index())) {
                rawDiscardIndex_ = cell->index();
                // as this function is called by the reader, the state and type should
                // already be correct
                //    setEvtState(cell->index(),evt::STOP);
                //    cell->setEventType(evt::STOPPER);
                //    setEvtNumber(cell->index(),0xffffffff);
                //    setEvtPrcId(cell->index(),0);
                //    setEvtTimeStamp(cell->index(),0);
                pidstatus = removeClientPrcId(getpid());
                if (segmentationMode_)
                        shmdt(cell);
                postRawDiscarded();
                return true;
        } else {
                postRawDiscard();
                return false;
        }
}

//______________________________________________________________________________
void FUShmBuffer::scheduleRawEmptyCellForDiscardServerSide(FUShmRawCell* cell) {
        //  waitRawDiscard();
        if (rawCellReadyForDiscard(cell->index())) {
                rawDiscardIndex_ = cell->index();
                //    setEvtState(cell->index(),evt::STOP);
                //    cell->setEventType(evt::STOPPER);
                //    setEvtNumber(cell->index(),0xffffffff);
                //    setEvtPrcId(cell->index(),0);
                //    setEvtTimeStamp(cell->index(),0);
                //    removeClientPrcId(getpid());
                if (segmentationMode_)
                        shmdt(cell);
                postRawDiscarded();
        } else
                postRawDiscard();
}

//______________________________________________________________________________
bool FUShmBuffer::writeRecoInitMsg(unsigned int outModId,
                unsigned int fuProcessId, unsigned int fuGuid, unsigned char *data,
                unsigned int dataSize, unsigned int nExpectedEPs) {
        if (dataSize > recoCellPayloadSize_) {
                cout << "FUShmBuffer::writeRecoInitMsg() ERROR: buffer overflow."
                                << endl;
                return false;
        }

        waitRecoWrite();
        unsigned int iCell = nextRecoWriteIndex();
        FUShmRecoCell* cell = recoCell(iCell);
        cell->writeInitMsg(outModId, fuProcessId, fuGuid, data, dataSize,nExpectedEPs);
        postRecoIndexToRead(iCell);
        if (segmentationMode_)
                shmdt(cell);
        postRecoRead();
        return true;
}

//______________________________________________________________________________
bool FUShmBuffer::writeRecoEventData(unsigned int runNumber,
                unsigned int evtNumber, unsigned int outModId,
                unsigned int fuProcessId, unsigned int fuGuid, unsigned char *data,
                unsigned int dataSize) {
        if (dataSize > recoCellPayloadSize_) {
                cout << "FUShmBuffer::writeRecoEventData() ERROR: buffer overflow."
                                << endl;
                return false;
        }

        waitRecoWrite();
        unsigned int rawCellIndex = indexForEvtNumber(evtNumber);
        unsigned int iCell = nextRecoWriteIndex();
        FUShmRecoCell* cell = recoCell(iCell);
        //evt::State_t state=evtState(rawCellIndex);
        //XCEPT_ASSERT(state==evt::PROCESSING||state==evt::RECOWRITING||state==evt::SENT,
        //evf::Exception, "state==evt::PROCESSING||state==evt::RECOWRITING||state==evt::SENT assertion failed!");
        lock();
        setEvtState(rawCellIndex, evt::RECOWRITING,false);
        incEvtDiscard(rawCellIndex,false);
        unlock();
        cell->writeEventData(rawCellIndex, runNumber, evtNumber, outModId,
                        fuProcessId, fuGuid, data, dataSize);
        setEvtState(rawCellIndex, evt::RECOWRITTEN);
        postRecoIndexToRead(iCell);
        if (segmentationMode_)
                shmdt(cell);
        postRecoRead();
        return true;
}

//______________________________________________________________________________
bool FUShmBuffer::writeErrorEventData(unsigned int runNumber,
                unsigned int fuProcessId, unsigned int iRawCell,bool checkValue) {
        FUShmRawCell *raw = rawCell(iRawCell);

        unsigned int dataSize = sizeof(uint32_t) * (4 + 1024) + raw->eventSize();
        unsigned char *data = new unsigned char[dataSize];
        uint32_t *pos = (uint32_t*) data;
        // 06-Oct-2008, KAB - added a version number for the error event format.
        //
        // Version 1 had no version number, so the run number appeared in the
        // first value.  So, a reasonable test for version 1 is whether the
        // first value is larger than some relatively small cutoff (say 32).
        // Version 2 added the lumi block number.
        //
        *pos++ = (uint32_t) 2; // protocol version number
        *pos++ = (uint32_t) runNumber;
        *pos++ = (uint32_t) evf::evtn::getlbn(
                        raw->fedAddr(FEDNumbering::MINTriggerGTPFEDID)) + 1;
        *pos++ = (uint32_t) raw->evtNumber();
        for (unsigned int i = 0; i < 1024; i++)
                *pos++ = (uint32_t) raw->fedSize(i);
        memcpy(pos, raw->payloadAddr(), raw->eventSize());

        // DEBUG
        /*
         if (1) {
         stringstream ss;
         ss<<"/tmp/run"<<runNumber<<"_evt"<<raw->evtNumber()<<".err";
         ofstream fout;
         fout.open(ss.str().c_str(),ios::out|ios::binary);
         if (!fout.write((char*)data,dataSize))
         cout<<"Failed to write error event to "<<ss.str()<<endl;
         fout.close();

         stringstream ss2;
         ss2<<"/tmp/run"<<runNumber<<"_evt"<<raw->evtNumber()<<".info";
         ofstream fout2;
         fout2.open(ss2.str().c_str());
         fout2<<"dataSize = "<<dataSize<<endl;
         fout2<<"runNumber = "<<runNumber<<endl;
         fout2<<"evtNumber = "<<raw->evtNumber()<<endl;
         fout2<<"eventSize = "<<raw->eventSize()<<endl;
         unsigned int totalSize(0);
         for (unsigned int i=0;i<1024;i++) {
         unsigned int fedSize = raw->fedSize(i);
         totalSize += fedSize;
         if (fedSize>0) fout2<<i<<": "<<fedSize<<endl;
         }
         fout2<<"totalSize = "<<totalSize<<endl;
         fout2.close();
         }
         */// END DEBUG

        waitRecoWrite();
        unsigned int iRecoCell = nextRecoWriteIndex();
        FUShmRecoCell* reco = recoCell(iRecoCell);
        lock();
        setEvtState(iRawCell, evt::RECOWRITING,false);
        setEvtDiscard(iRawCell, 1, true,false);
        unlock();
        reco->writeErrorEvent(iRawCell, runNumber, raw->evtNumber(), fuProcessId,
                        data, dataSize);
        delete[] data;
        setEvtState(iRawCell, evt::RECOWRITTEN);
        postRecoIndexToRead(iRecoCell);
        if (segmentationMode_) {
                shmdt(raw);
                shmdt(reco);
        }
        postRecoRead();
        return true;
}

//______________________________________________________________________________
bool FUShmBuffer::writeDqmEventData(unsigned int runNumber,
                unsigned int evtAtUpdate, unsigned int folderId,
                unsigned int fuProcessId, unsigned int fuGuid, unsigned char *data,
                unsigned int dataSize) {
        if (dataSize > dqmCellPayloadSize_) {
                cout << "FUShmBuffer::writeDqmEventData() ERROR: buffer overflow."
                                << endl;
                return false;
        }

        waitDqmWrite();
        unsigned int iCell = nextDqmWriteIndex();
        FUShmDqmCell* cell = dqmCell(iCell);
        dqm::State_t state = dqmState(iCell);
        if (!(state == dqm::EMPTY)) {
          stringstream details;
          details << "state==dqm::EMPTY assertion failed! Actual state is " << state
                  << ", iCell = " << iCell;
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        setDqmState(iCell, dqm::WRITING);
        cell->writeData(runNumber, evtAtUpdate, folderId, fuProcessId, fuGuid,
                        data, dataSize);
        setDqmState(iCell, dqm::WRITTEN);
        postDqmIndexToRead(iCell);
        if (segmentationMode_)
                shmdt(cell);
        postDqmRead();
        return true;
}

//______________________________________________________________________________
void FUShmBuffer::sem_print() {
        cout << "--> current sem values:" << endl << " lock=" << semctl(semid(), 0,
                        GETVAL) << endl << " wraw=" << semctl(semid(), 1, GETVAL)
                        << " rraw=" << semctl(semid(), 2, GETVAL) << endl << " wdsc="
                        << semctl(semid(), 3, GETVAL) << " rdsc=" << semctl(semid(), 4,
                        GETVAL) << endl << " wrec=" << semctl(semid(), 5, GETVAL)
                        << " rrec=" << semctl(semid(), 6, GETVAL) << endl << " wdqm="
                        << semctl(semid(), 7, GETVAL) << " rdqm=" << semctl(semid(), 8,
                        GETVAL) << endl;
}

std::string FUShmBuffer::sem_print_s() {
        ostringstream ostr;
        ostr    << "--> current sem values:" << endl << " lock=" << semctl(semid(), 0,
                        GETVAL) << endl << " wraw=" << semctl(semid(), 1, GETVAL)
                << " rraw=" << semctl(semid(), 2, GETVAL) << endl << " wdsc="
                << semctl(semid(), 3, GETVAL) << " rdsc=" << semctl(semid(), 4,
                                GETVAL) << endl << " wrec=" << semctl(semid(), 5, GETVAL)
                << " rrec=" << semctl(semid(), 6, GETVAL) << endl << " wdqm="
                << semctl(semid(), 7, GETVAL) << " rdqm=" << semctl(semid(), 8,
                                GETVAL) << endl;
        return ostr.str();

}

//______________________________________________________________________________
void FUShmBuffer::printEvtState(unsigned int index) {
        evt::State_t state = evtState(index);
        std::string stateName;
        if (state == evt::EMPTY)
                stateName = "EMPTY";
        else if (state == evt::STOP)
                stateName = "STOP";
        else if (state == evt::RAWWRITING)
                stateName = "RAWWRITING";
        else if (state == evt::RAWWRITTEN)
                stateName = "RAWRITTEN";
        else if (state == evt::RAWREADING)
                stateName = "RAWREADING";
        else if (state == evt::RAWREAD)
                stateName = "RAWREAD";
        else if (state == evt::PROCESSING)
                stateName = "PROCESSING";
        else if (state == evt::PROCESSED)
                stateName = "PROCESSED";
        else if (state == evt::RECOWRITING)
                stateName = "RECOWRITING";
        else if (state == evt::RECOWRITTEN)
                stateName = "RECOWRITTEN";
        else if (state == evt::SENDING)
                stateName = "SENDING";
        else if (state == evt::SENT)
                stateName = "SENT";
        else if (state == evt::DISCARDING)
                stateName = "DISCARDING";
        cout << "evt " << index << " in state '" << stateName << "'." << endl;
}

//______________________________________________________________________________
void FUShmBuffer::printDqmState(unsigned int index) {
        dqm::State_t state = dqmState(index);
        cout << "dqm evt " << index << " in state '" << state << "'." << endl;
}

//______________________________________________________________________________
FUShmBuffer* FUShmBuffer::createShmBuffer(bool segmentationMode,
                unsigned int nRawCells, unsigned int nRecoCells,
                unsigned int nDqmCells, unsigned int rawCellSize,
                unsigned int recoCellSize, unsigned int dqmCellSize) {
        // if necessary, release shared memory first!
        if (FUShmBuffer::releaseSharedMemory())
                cout << "FUShmBuffer::createShmBuffer: "
                                << "REMOVAL OF OLD SHARED MEM SEGMENTS SUCCESSFULL." << endl;

        // create bookkeeping shared memory segment
        int size = sizeof(unsigned int) * 7;
        int shmid = shm_create(FUShmBuffer::getShmDescriptorKey(), size);
        if (shmid < 0)
                return 0;
        void*shmAddr = shm_attach(shmid);
        if (0 == shmAddr)
                return 0;

        if (1 != shm_nattch(shmid)) {
                cout << "FUShmBuffer::createShmBuffer() FAILED: nattch=" << shm_nattch(
                                shmid) << endl;
                shmdt(shmAddr);
                return 0;
        }

        unsigned int* p = (unsigned int*) shmAddr;
        *p++ = segmentationMode;
        *p++ = nRawCells;
        *p++ = nRecoCells;
        *p++ = nDqmCells;
        *p++ = rawCellSize;
        *p++ = recoCellSize;
        *p++ = dqmCellSize;
        shmdt(shmAddr);

        // create the 'real' shared memory buffer
        size = FUShmBuffer::size(segmentationMode, nRawCells, nRecoCells,
                        nDqmCells, rawCellSize, recoCellSize, dqmCellSize);
        shmid = shm_create(FUShmBuffer::getShmKey(), size);
        if (shmid < 0)
                return 0;
        int semid = sem_create(FUShmBuffer::getSemKey(), 9);
        if (semid < 0)
                return 0;
        shmAddr = shm_attach(shmid);
        if (0 == shmAddr)
                return 0;

        if (1 != shm_nattch(shmid)) {
                cout << "FUShmBuffer::createShmBuffer FAILED: nattch=" << shm_nattch(
                                shmid) << endl;
                shmdt(shmAddr);
                return 0;
        }
        FUShmBuffer* buffer = new (shmAddr) FUShmBuffer(segmentationMode,
                        nRawCells, nRecoCells, nDqmCells, rawCellSize, recoCellSize,
                        dqmCellSize);

        cout << "FUShmBuffer::createShmBuffer(): CREATED shared memory buffer."
                        << endl;
        cout << "                                segmentationMode="
                        << segmentationMode << endl;

        buffer->initialize(shmid, semid);

        return buffer;
}

//______________________________________________________________________________
FUShmBuffer* FUShmBuffer::getShmBuffer() {
        // get bookkeeping shared memory segment
        int size = sizeof(unsigned int) * 7;
        int shmid = shm_get(FUShmBuffer::getShmDescriptorKey(), size);
        if (shmid < 0)
                return 0;
        void* shmAddr = shm_attach(shmid);
        if (0 == shmAddr)
                return 0;

        unsigned int *p = (unsigned int*) shmAddr;
        bool segmentationMode = *p++;
        unsigned int nRawCells = *p++;
        unsigned int nRecoCells = *p++;
        unsigned int nDqmCells = *p++;
        unsigned int rawCellSize = *p++;
        unsigned int recoCellSize = *p++;
        unsigned int dqmCellSize = *p++;
        shmdt(shmAddr);

        cout << "FUShmBuffer::getShmBuffer():" << " segmentationMode="
                        << segmentationMode << " nRawCells=" << nRawCells << " nRecoCells="
                        << nRecoCells << " nDqmCells=" << nDqmCells << " rawCellSize="
                        << rawCellSize << " recoCellSize=" << recoCellSize
                        << " dqmCellSize=" << dqmCellSize << endl;

        // get the 'real' shared memory buffer
        size = FUShmBuffer::size(segmentationMode, nRawCells, nRecoCells,
                        nDqmCells, rawCellSize, recoCellSize, dqmCellSize);
        shmid = shm_get(FUShmBuffer::getShmKey(), size);
        if (shmid < 0)
                return 0;
        int semid = sem_get(FUShmBuffer::getSemKey(), 9);
        if (semid < 0)
                return 0;
        shmAddr = shm_attach(shmid);
        if (0 == shmAddr)
                return 0;

        if (0 == shm_nattch(shmid)) {
                cout << "FUShmBuffer::getShmBuffer() FAILED: nattch=" << shm_nattch(
                                shmid) << endl;
                return 0;
        }
        FUShmBuffer* buffer = new (shmAddr) FUShmBuffer(segmentationMode,
                        nRawCells, nRecoCells, nDqmCells, rawCellSize, recoCellSize,
                        dqmCellSize);

        cout << "FUShmBuffer::getShmBuffer(): shared memory buffer RETRIEVED. PID:" << getpid()
                        << endl;
        cout << "                             segmentationMode="
                        << segmentationMode << endl;

        buffer->setClientPrcId(getpid());

        return buffer;
}

//______________________________________________________________________________
bool FUShmBuffer::releaseSharedMemory() {
        // get bookkeeping shared memory segment
        int size = sizeof(unsigned int) * 7;
        int shmidd = shm_get(FUShmBuffer::getShmDescriptorKey(), size);
        if (shmidd < 0)
                return false;
        void* shmAddr = shm_attach(shmidd);
        if (0 == shmAddr)
                return false;

        unsigned int*p = (unsigned int*) shmAddr;
        bool segmentationMode = *p++;
        unsigned int nRawCells = *p++;
        unsigned int nRecoCells = *p++;
        unsigned int nDqmCells = *p++;
        unsigned int rawCellSize = *p++;
        unsigned int recoCellSize = *p++;
        unsigned int dqmCellSize = *p++;
        shmdt(shmAddr);

        // get the 'real' shared memory segment
        size = FUShmBuffer::size(segmentationMode, nRawCells, nRecoCells,
                        nDqmCells, rawCellSize, recoCellSize, dqmCellSize);
        int shmid = shm_get(FUShmBuffer::getShmKey(), size);
        if (shmid < 0)
                return false;
        int semid = sem_get(FUShmBuffer::getSemKey(), 9);
        if (semid < 0)
                return false;
        shmAddr = shm_attach(shmid);
        if (0 == shmAddr)
                return false;

        int att = 0;
        for (; att < 10; att++) {
                if (shm_nattch(shmid) > 1) {
                        cout << att << " FUShmBuffer::releaseSharedMemory(): nattch="
                                        << shm_nattch(shmid)
                                        << ", failed attempt to release shared memory." << endl;
			::sleep(1);
                } else
                        break;
        }

        if (att >= 10)
                return false;

        if (segmentationMode) {
                FUShmBuffer* buffer = new (shmAddr) FUShmBuffer(segmentationMode,
                                nRawCells, nRecoCells, nDqmCells, rawCellSize, recoCellSize,
                                dqmCellSize);
                int cellid;
                for (unsigned int i = 0; i < nRawCells; i++) {
                        cellid = shm_get(buffer->rawCellShmKey(i),
                                        FUShmRawCell::size(rawCellSize));
                        if ((shm_destroy(cellid) == -1))
                                return false;
                }
                for (unsigned int i = 0; i < nRecoCells; i++) {
                        cellid = shm_get(buffer->recoCellShmKey(i),
                                        FUShmRecoCell::size(recoCellSize));
                        if ((shm_destroy(cellid) == -1))
                                return false;
                }
                for (unsigned int i = 0; i < nDqmCells; i++) {
                        cellid = shm_get(buffer->dqmCellShmKey(i),
                                        FUShmDqmCell::size(dqmCellSize));
                        if ((shm_destroy(cellid) == -1))
                                return false;
                }
        }
        shmdt(shmAddr);
        if (sem_destroy(semid) == -1)
                return false;
        if (shm_destroy(shmid) == -1)
                return false;
        if (shm_destroy(shmidd) == -1)
                return false;

        return true;
}

//______________________________________________________________________________
unsigned int FUShmBuffer::size(bool segmentationMode, unsigned int nRawCells,
                unsigned int nRecoCells, unsigned int nDqmCells,
                unsigned int rawCellSize, unsigned int recoCellSize,
                unsigned int dqmCellSize) {
        unsigned int offset = sizeof(FUShmBuffer) + sizeof(unsigned int) * 4
                        * nRawCells + sizeof(evt::State_t) * nRawCells
                        + sizeof(dqm::State_t) * nDqmCells;

        unsigned int rawCellTotalSize = FUShmRawCell::size(rawCellSize);
        unsigned int recoCellTotalSize = FUShmRecoCell::size(recoCellSize);
        unsigned int dqmCellTotalSize = FUShmDqmCell::size(dqmCellSize);

        unsigned int realSize = (segmentationMode) ? offset + sizeof(key_t)
                        * (nRawCells + nRecoCells + nDqmCells) : offset + rawCellTotalSize
                        * nRawCells + recoCellTotalSize * nRecoCells + dqmCellTotalSize
                        * nDqmCells;

        unsigned int result = realSize / 0x10 * 0x10 + (realSize % 0x10 > 0) * 0x10;

        return result;
}

//______________________________________________________________________________
key_t FUShmBuffer::getShmDescriptorKey() {
        key_t result = getuid() * 1000 + SHM_DESCRIPTOR_KEYID;
        if (result == (key_t) -1)
                cout << "FUShmBuffer::getShmDescriptorKey: failed " << "for file "
                                << shmKeyPath_ << "!" << endl;
        return result;
}

//______________________________________________________________________________
key_t FUShmBuffer::getShmKey() {
        key_t result = getuid() * 1000 + SHM_KEYID;
        if (result == (key_t) -1)
                cout << "FUShmBuffer::getShmKey: ftok() failed " << "for file "
                                << shmKeyPath_ << "!" << endl;
        return result;
}

//______________________________________________________________________________
key_t FUShmBuffer::getSemKey() {
        key_t result = getuid() * 1000 + SEM_KEYID;
        if (result == (key_t) -1)
                cout << "FUShmBuffer::getSemKey: ftok() failed " << "for file "
                                << semKeyPath_ << "!" << endl;
        return result;
}

//______________________________________________________________________________
int FUShmBuffer::shm_create(key_t key, int size) {
        // first check and possibly remove existing segment with same id
        int shmid = shmget(key, 1, 0644);//using minimal size any segment with key "key" will be connected
        if (shmid != -1) {
                // an existing segment was found, remove it
                shmid_ds shmstat;
                shmctl(shmid, IPC_STAT, &shmstat);
                cout << "FUShmBuffer found segment for key 0x " << hex << key << dec
                                << " created by process " << shmstat.shm_cpid << " owned by "
                                << shmstat.shm_perm.uid << " permissions " << hex
                                << shmstat.shm_perm.mode << dec << endl;
                shmctl(shmid, IPC_RMID, &shmstat);
        }
        shmid = shmget(key, size, IPC_CREAT | 0644);
        if (shmid == -1) {
                int err = errno;
                cout << "FUShmBuffer::shm_create(" << key << "," << size
                                << ") failed: " << strerror(err) << endl;
        }
        return shmid;
}

//______________________________________________________________________________
int FUShmBuffer::shm_get(key_t key, int size) {
        int shmid = shmget(key, size, 0644);
        if (shmid == -1) {
                int err = errno;
                cout << "FUShmBuffer::shm_get(" << key << "," << size << ") failed: "
                                << strerror(err) << endl;
        }
        return shmid;
}

//______________________________________________________________________________
void* FUShmBuffer::shm_attach(int shmid) {
        void* result = shmat(shmid, NULL, 0);
        if (0 == result) {
                int err = errno;
                cout << "FUShmBuffer::shm_attach(" << shmid << ") failed: "
                                << strerror(err) << endl;
        }
        return result;
}

//______________________________________________________________________________
int FUShmBuffer::shm_nattch(int shmid) {
        shmid_ds shmstat;
        shmctl(shmid, IPC_STAT, &shmstat);
        return shmstat.shm_nattch;
}

//______________________________________________________________________________
int FUShmBuffer::shm_destroy(int shmid) {
        shmid_ds shmstat;
        int result = shmctl(shmid, IPC_RMID, &shmstat);
        if (result == -1)
                cout << "FUShmBuffer::shm_destroy(shmid=" << shmid << ") failed."
                                << endl;
        return result;
}

//______________________________________________________________________________
int FUShmBuffer::sem_create(key_t key, int nsem) {
        int semid = semget(key, nsem, IPC_CREAT | 0666);
        if (semid < 0) {
                int err = errno;
                cout << "FUShmBuffer::sem_create(key=" << key << ",nsem=" << nsem
                                << ") failed: " << strerror(err) << endl;
        }
        return semid;
}

//______________________________________________________________________________
int FUShmBuffer::sem_get(key_t key, int nsem) {
        int semid = semget(key, nsem, 0666);
        if (semid < 0) {
                int err = errno;
                cout << "FUShmBuffer::sem_get(key=" << key << ",nsem=" << nsem
                                << ") failed: " << strerror(err) << endl;
        }
        return semid;
}

//______________________________________________________________________________
int FUShmBuffer::sem_destroy(int semid) {
        int result = semctl(semid, 0, IPC_RMID);
        if (result == -1)
                cout << "FUShmBuffer::sem_destroy(semid=" << semid << ") failed."
                                << endl;
        return result;
}

// implementation of private member functions

//______________________________________________________________________________
unsigned int FUShmBuffer::nextIndex(unsigned int offset, unsigned int nCells,
                unsigned int& iNext) {
        lock();
        unsigned int* pindex = (unsigned int*) ((unsigned long) this + offset);
        pindex += iNext;
        iNext = (iNext + 1) % nCells;
        unsigned int result = *pindex;
        unlock();
        return result;
}

//______________________________________________________________________________
void FUShmBuffer::postIndex(unsigned int index, unsigned int offset,
                unsigned int nCells, unsigned int& iLast) {
        lock();
        unsigned int* pindex = (unsigned int*) ((unsigned long) this + offset);
        pindex += iLast;
        *pindex = index;
        iLast = (iLast + 1) % nCells;
        unlock();
}

//______________________________________________________________________________
unsigned int FUShmBuffer::nextRawWriteIndex() {
        return nextIndex(rawWriteOffset_, nRawCells_, rawWriteNext_);
}

//______________________________________________________________________________
unsigned int FUShmBuffer::nextRawReadIndex() {
        return nextIndex(rawReadOffset_, nRawCells_, rawReadNext_);
}

//______________________________________________________________________________
void FUShmBuffer::postRawIndexToWrite(unsigned int index) {
        postIndex(index, rawWriteOffset_, nRawCells_, rawWriteLast_);
}

//______________________________________________________________________________
void FUShmBuffer::postRawIndexToRead(unsigned int index) {
        postIndex(index, rawReadOffset_, nRawCells_, rawReadLast_);
}

//______________________________________________________________________________
unsigned int FUShmBuffer::nextRecoWriteIndex() {
        return nextIndex(recoWriteOffset_, nRecoCells_, recoWriteNext_);
}

//______________________________________________________________________________
unsigned int FUShmBuffer::nextRecoReadIndex() {
        return nextIndex(recoReadOffset_, nRecoCells_, recoReadNext_);
}

//______________________________________________________________________________
void FUShmBuffer::postRecoIndexToWrite(unsigned int index) {
        postIndex(index, recoWriteOffset_, nRecoCells_, recoWriteLast_);
}

//______________________________________________________________________________
void FUShmBuffer::postRecoIndexToRead(unsigned int index) {
        postIndex(index, recoReadOffset_, nRecoCells_, recoReadLast_);
}

//______________________________________________________________________________
unsigned int FUShmBuffer::nextDqmWriteIndex() {
        return nextIndex(dqmWriteOffset_, nDqmCells_, dqmWriteNext_);
}

//______________________________________________________________________________
unsigned int FUShmBuffer::nextDqmReadIndex() {
        return nextIndex(dqmReadOffset_, nDqmCells_, dqmReadNext_);
}

//______________________________________________________________________________
void FUShmBuffer::postDqmIndexToWrite(unsigned int index) {
        postIndex(index, dqmWriteOffset_, nDqmCells_, dqmWriteLast_);
}

//______________________________________________________________________________
void FUShmBuffer::postDqmIndexToRead(unsigned int index) {
        postIndex(index, dqmReadOffset_, nDqmCells_, dqmReadLast_);
}

//______________________________________________________________________________
unsigned int FUShmBuffer::indexForEvtNumber(unsigned int evtNumber) {
        unsigned int *pevt = (unsigned int*) ((unsigned long) this
                        + evtNumberOffset_);
        for (unsigned int i = 0; i < nRawCells_; i++) {
                if ((*pevt++) == evtNumber)
                        return i;
        }
        XCEPT_ASSERT(false, evf::Exception, "This point should not be reached!");
        return 0xffffffff;
}

//______________________________________________________________________________
unsigned int FUShmBuffer::indexForEvtPrcId(pid_t prcid) {
        pid_t *pevt = (pid_t*) ((unsigned long) this + evtPrcIdOffset_);
        for (unsigned int i = 0; i < nRawCells_; i++) {
                if ((*pevt++) == prcid)
                        return i;
        }
        XCEPT_ASSERT(false, evf::Exception, "This point should not be reached!");
        return 0xffffffff;
}

//______________________________________________________________________________
evt::State_t FUShmBuffer::evtState(unsigned int index) {
        if (!(index < nRawCells_)) {
          stringstream details;
          details << "index<nRawCells_ assertion failed! Actual index is " << index;
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        evt::State_t *pstate = (evt::State_t*) ((unsigned long) this
                        + evtStateOffset_);
        pstate += index;
        return *pstate;
}

//______________________________________________________________________________
dqm::State_t FUShmBuffer::dqmState(unsigned int index) {
        if (!(index < nDqmCells_)) {
          stringstream details;
          details << "index<nDqmCells_ assertion failed! Actual index is " << index;
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        dqm::State_t *pstate = (dqm::State_t*) ((unsigned long) this
                        + dqmStateOffset_);
        pstate += index;
        return *pstate;
}

//______________________________________________________________________________
unsigned int FUShmBuffer::evtNumber(unsigned int index) {
        if (!(index < nRawCells_)) {
          stringstream details;
          details << "index<nRawCells_ assertion failed! Actual index is " << index;
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        unsigned int *pevt = (unsigned int*) ((unsigned long) this
                        + evtNumberOffset_);
        pevt += index;
        return *pevt;
}

//______________________________________________________________________________
pid_t FUShmBuffer::evtPrcId(unsigned int index) {
        if (!(index < nRawCells_)) {
          stringstream details;
          details << "index<nRawCells_ assertion failed! Actual index is " << index;
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        pid_t *prcid = (pid_t*) ((unsigned long) this + evtPrcIdOffset_);
        prcid += index;
        return *prcid;
}

//______________________________________________________________________________
time_t FUShmBuffer::evtTimeStamp(unsigned int index) {
        if (!(index < nRawCells_)) {
          stringstream details;
          details << "index<nRawCells_ assertion failed! Actual index is " << index;
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        time_t *ptstmp = (time_t*) ((unsigned long) this + evtTimeStampOffset_);
        ptstmp += index;
        return *ptstmp;
}

//______________________________________________________________________________
pid_t FUShmBuffer::clientPrcId(unsigned int index) {
        if (!(index < nClientsMax_)) {
          stringstream details;
          details << "index<nClientsMax_ assertion failed! Actual index is " << index;
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        pid_t *prcid = (pid_t*) ((unsigned long) this + clientPrcIdOffset_);
        prcid += index;
        return *prcid;
}

//______________________________________________________________________________
bool FUShmBuffer::setEvtState(unsigned int index, evt::State_t state, bool lockShm) {
        if(!(index < nRawCells_)) {
          stringstream details;
          details << "index<nRawCells_ assertion failed! Actual index is " << index;
          //unlock here as it will skip external unlock
          if (!lockShm) unlock();
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        evt::State_t *pstate = (evt::State_t*) ((unsigned long) this
                        + evtStateOffset_);
        pstate += index;
        if (lockShm) lock();
        *pstate = state;
        if (lockShm) unlock();
        return true;
}

//______________________________________________________________________________
bool FUShmBuffer::setDqmState(unsigned int index, dqm::State_t state) {
        if(!(index < nDqmCells_)) {
          stringstream details;
          details << "index<nDqmCells_ assertion failed! Actual index is " << index;
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        dqm::State_t *pstate = (dqm::State_t*) ((unsigned long) this
                        + dqmStateOffset_);
        pstate += index;
        lock();
        *pstate = state;
        unlock();
        return true;
}

//______________________________________________________________________________
bool FUShmBuffer::setEvtDiscard(unsigned int index, unsigned int discard, bool checkValue, bool lockShm) {
        if(!(index < nRawCells_)) {
          stringstream details;
          details << "index<nRawCells_ assertion failed! Actual index is " << index;
          //unlock here as it will skip external unlock
          if (!lockShm) unlock();
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        unsigned int *pcount = (unsigned int*) ((unsigned long) this
                        + evtDiscardOffset_);
        pcount += index;
        if (lockShm) lock();
        if (checkValue) {
                if (*pcount < discard)
                        *pcount = discard;
        } else
                *pcount = discard;
        if (lockShm) unlock();
        return true;
}

//______________________________________________________________________________
int FUShmBuffer::incEvtDiscard(unsigned int index, bool lockShm) {
        int result = 0;
        if(!(index < nRawCells_)) {
          stringstream details;
          details << "index<nRawCells_ assertion failed! Actual index is " << index;
          //unlock here as it will skip external unlock
          if (!lockShm) unlock();
          XCEPT_ASSERT(false,evf::Exception,details.str());
        }
        unsigned int *pcount = (unsigned int*) ((unsigned long) this
                        + evtDiscardOffset_);
        pcount += index;
        if (lockShm) lock();
        (*pcount)++;
        result = *pcount;
        if (lockShm) unlock();
        return result;
}

//______________________________________________________________________________
bool FUShmBuffer::setEvtNumber(unsigned int index, unsigned int evtNumber) {
        if(!(index < nRawCells_)) {
          stringstream details;
          details << "index<nRawCells_ assertion failed! Actual index is " << index;
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        unsigned int *pevt = (unsigned int*) ((unsigned long) this
                        + evtNumberOffset_);
        pevt += index;
        lock();
        *pevt = evtNumber;
        unlock();
        return true;
}

//______________________________________________________________________________
bool FUShmBuffer::setEvtPrcId(unsigned int index, pid_t prcId) {
        if(!(index < nRawCells_)) {
          stringstream details;
          details << "index<nRawCells_ assertion failed! Actual index is " << index;
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        pid_t* prcid = (pid_t*) ((unsigned long) this + evtPrcIdOffset_);
        prcid += index;
        lock();
        *prcid = prcId;
        unlock();
        return true;
}

//______________________________________________________________________________
bool FUShmBuffer::setEvtTimeStamp(unsigned int index, time_t timeStamp) {
        if(!(index < nRawCells_)) {
          stringstream details;
          details << "index<nRawCells_ assertion failed! Actual index is " << index;
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        time_t *ptstmp = (time_t*) ((unsigned long) this + evtTimeStampOffset_);
        ptstmp += index;
        lock();
        *ptstmp = timeStamp;
        unlock();
        return true;
}

//______________________________________________________________________________
bool FUShmBuffer::setClientPrcId(pid_t prcId) {
        lock();
        if(!(nClients_ < nClientsMax_)) {
          stringstream details;
          details << "nClients_<nClientsMax_ assertion failed! Actual nClients is "
                  << nClients_ << " and nClientsMax is " << nClientsMax_;
          unlock();
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        pid_t *prcid = (pid_t*) ((unsigned long) this + clientPrcIdOffset_);
        for (unsigned int i = 0; i < nClients_; i++) {
                if ((*prcid) == prcId) {
                        unlock();
                        return false;
                }
                prcid++;
        }
        nClients_++;
        *prcid = prcId;
        unlock();
        return true;
}

//______________________________________________________________________________
bool FUShmBuffer::removeClientPrcId(pid_t prcId) {
        lock();
        pid_t *prcid = (pid_t*) ((unsigned long) this + clientPrcIdOffset_);
        unsigned int iClient(0);
        while (iClient < nClients_ && (*prcid) != prcId) {
                prcid++;
                iClient++;
        }
        if (iClient==nClients_) {
                unlock();
                return false;
        }
        //stringstream details;
        //details << "iClient!=nClients_ assertion failed! Actual iClient is "
        //              << iClient;
        //XCEPT_ASSERT(iClient != nClients_, evf::Exception, details.str());
        pid_t* next = prcid;
        next++;
        while (iClient < nClients_ - 1) {
                *prcid = *next;
                prcid++;
                next++;
                iClient++;
        }
        nClients_--;
        unlock();
        return true;
}

//______________________________________________________________________________
FUShmRawCell* FUShmBuffer::rawCell(unsigned int iCell) {
        FUShmRawCell* result(0);

        if (iCell >= nRawCells_) {
                cout << "FUShmBuffer::rawCell(" << iCell << ") ERROR: " << "iCell="
                                << iCell << " >= nRawCells()=" << nRawCells_ << endl;
                return result;
        }

        if (segmentationMode_) {
                key_t shmkey = rawCellShmKey(iCell);
                int shmid = shm_get(shmkey, rawCellTotalSize_);
                void* cellAddr = shm_attach(shmid);
                result = new (cellAddr) FUShmRawCell(rawCellPayloadSize_);
        } else {
                result = (FUShmRawCell*) ((unsigned long) this + rawCellOffset_ + iCell
                                * rawCellTotalSize_);
        }

        return result;
}

//______________________________________________________________________________
FUShmRecoCell* FUShmBuffer::recoCell(unsigned int iCell) {
        FUShmRecoCell* result(0);

        if (iCell >= nRecoCells_) {
                cout << "FUShmBuffer::recoCell(" << iCell << ") ERROR: " << "iCell="
                                << iCell << " >= nRecoCells=" << nRecoCells_ << endl;
                return result;
        }

        if (segmentationMode_) {
                key_t shmkey = recoCellShmKey(iCell);
                int shmid = shm_get(shmkey, recoCellTotalSize_);
                void* cellAddr = shm_attach(shmid);
                result = new (cellAddr) FUShmRecoCell(recoCellPayloadSize_);
        } else {
                result = (FUShmRecoCell*) ((unsigned long) this + recoCellOffset_
                                + iCell * recoCellTotalSize_);
        }

        return result;
}

//______________________________________________________________________________
FUShmDqmCell* FUShmBuffer::dqmCell(unsigned int iCell) {
        FUShmDqmCell* result(0);

        if (iCell >= nDqmCells_) {
                cout << "FUShmBuffer::dqmCell(" << iCell << ") ERROR: " << "iCell="
                                << iCell << " >= nDqmCells=" << nDqmCells_ << endl;
                return result;
        }

        if (segmentationMode_) {
                key_t shmkey = dqmCellShmKey(iCell);
                int shmid = shm_get(shmkey, dqmCellTotalSize_);
                void* cellAddr = shm_attach(shmid);
                result = new (cellAddr) FUShmDqmCell(dqmCellPayloadSize_);
        } else {
                result = (FUShmDqmCell*) ((unsigned long) this + dqmCellOffset_ + iCell
                                * dqmCellTotalSize_);
        }

        return result;
}

//______________________________________________________________________________
bool FUShmBuffer::rawCellReadyForDiscard(unsigned int index) {
        if(!(index < nRawCells_)) {
          stringstream details;
          details << "index<nRawCells_ assertion failed! Actual index is " << index;
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        unsigned int *pcount = (unsigned int*) ((unsigned long) this
                        + evtDiscardOffset_);
        pcount += index;
        lock();
        if(!(*pcount > 0)) {
          stringstream details;
          details << "*pcount>0 assertion failed! Value at pcount is " << *pcount << " for cell index " << index;
          unlock();
          XCEPT_ASSERT(false, evf::Exception, details.str());
        }
        --(*pcount);
        bool result = (*pcount == 0);
        unlock();
        return result;
}

//______________________________________________________________________________
key_t FUShmBuffer::shmKey(unsigned int iCell, unsigned int offset) {
        if (!segmentationMode_) {
                cout << "FUShmBuffer::shmKey() ERROR: only valid in segmentationMode!"
                                << endl;
                return -1;
        }
        key_t* addr = (key_t*) ((unsigned long) this + offset);
        for (unsigned int i = 0; i < iCell; i++)
                ++addr;
        return *addr;
}

//______________________________________________________________________________
key_t FUShmBuffer::rawCellShmKey(unsigned int iCell) {
        if (iCell >= nRawCells_) {
                cout << "FUShmBuffer::rawCellShmKey() ERROR: " << "iCell>=nRawCells: "
                                << iCell << ">=" << nRawCells_ << endl;
                return -1;
        }
        return shmKey(iCell, rawCellOffset_);
}

//______________________________________________________________________________
key_t FUShmBuffer::recoCellShmKey(unsigned int iCell) {
        if (iCell >= nRecoCells_) {
                cout << "FUShmBuffer::recoCellShmKey() ERROR: "
                                << "iCell>=nRecoCells: " << iCell << ">=" << nRecoCells_
                                << endl;
                return -1;
        }
        return shmKey(iCell, recoCellOffset_);
}

//______________________________________________________________________________
key_t FUShmBuffer::dqmCellShmKey(unsigned int iCell) {
        if (iCell >= nDqmCells_) {
                cout << "FUShmBuffer::dqmCellShmKey() ERROR: " << "iCell>=nDqmCells: "
                                << iCell << ">=" << nDqmCells_ << endl;
                return -1;
        }
        return shmKey(iCell, dqmCellOffset_);
}

//______________________________________________________________________________
void FUShmBuffer::sem_init(int isem, int value) {
        if (semctl(semid(), isem, SETVAL, value) < 0) {
                cout << "FUShmBuffer: FATAL ERROR in semaphore initialization." << endl;
        }
}

//______________________________________________________________________________
int FUShmBuffer::sem_wait(int isem) {
        struct sembuf sops[1];
        sops[0].sem_num = isem;
        sops[0].sem_op = -1;
        sops[0].sem_flg = 0;
        if (semop(semid(), sops, 1) == -1) {
                cout << "FUShmBuffer: ERROR in semaphore operation sem_wait." << endl;
                return -1;
        }
        return 0;
}

//______________________________________________________________________________
void FUShmBuffer::sem_post(int isem) {
        struct sembuf sops[1];
        sops[0].sem_num = isem;
        sops[0].sem_op = 1;
        sops[0].sem_flg = 0;
        if (semop(semid(), sops, 1) == -1) {
                cout << "FUShmBuffer: ERROR in semaphore operation sem_post." << endl;
        }
}