/afs/cern.ch/work/a/aaltunda/public/www/CMSSW_6_2_7/src/EventFilter/ResourceBroker/src/SMProxy.cc

Go to the documentation of this file.

//
// SMProxy
// -------
//
//            03/20/2007 Philipp Schieferdecker <philipp.schieferdecker@cern.ch>


#include "EventFilter/ResourceBroker/interface/SMProxy.h"

#include "xdaq/Application.h"

#include "toolbox/mem/Reference.h"
#include "toolbox/mem/MemoryPoolFactory.h"
#include "toolbox/mem/exception/Exception.h"

#include "i2o/Method.h"
#include "i2o/utils/AddressMap.h"

#include "xcept/tools.h"

#include <iostream>
#include <cmath>

using namespace std;
using namespace evf;

// construction/destruction

//______________________________________________________________________________
SMProxy::SMProxy(xdaq::ApplicationDescriptor *fuAppDesc,
                xdaq::ApplicationDescriptor *smAppDesc,
                xdaq::ApplicationContext *fuAppContext, toolbox::mem::Pool *i2oPool) :
        log_(fuAppContext->getLogger()), fuAppDesc_(fuAppDesc),
                        smAppDesc_(smAppDesc), fuAppContext_(fuAppContext),
                        i2oPool_(i2oPool),
                        initHeaderSize_(sizeof(I2O_SM_PREAMBLE_MESSAGE_FRAME)),
                        dataHeaderSize_(sizeof(I2O_SM_DATA_MESSAGE_FRAME)),
                        dqmHeaderSize_(sizeof(I2O_SM_DQM_MESSAGE_FRAME)) {
        fuUrl_ = fuAppDesc_->getContextDescriptor()->getURL();
        if (fuUrl_.size() >= MAX_I2O_SM_URLCHARS)
                fuUrl_ = fuUrl_.substr(0, MAX_I2O_SM_URLCHARS - 1);

        fuClassName_ = fuAppDesc_->getClassName();
        if (fuClassName_.size() >= MAX_I2O_SM_URLCHARS)
                fuClassName_ = fuClassName_.substr(0, MAX_I2O_SM_URLCHARS - 1);
}

//______________________________________________________________________________
SMProxy::~SMProxy() {

}

// implementation of member functions

//______________________________________________________________________________
UInt_t SMProxy::sendInitMessage(UInt_t fuResourceId, UInt_t outModId,
                UInt_t fuProcessId, UInt_t fuGuid, UChar_t*data, UInt_t dataSize,
                UInt_t nExpectedEPs) throw (evf::Exception) {
        UInt_t totalSize = 0;
        MemRef_t* bufRef = createFragmentChain(I2O_SM_PREAMBLE, initHeaderSize_,
                        data, dataSize, totalSize);

        I2O_SM_PREAMBLE_MESSAGE_FRAME *msg;
        MemRef_t* next = bufRef;
        do {
                msg = (I2O_SM_PREAMBLE_MESSAGE_FRAME*) next->getDataLocation();
                msg->rbBufferID = fuResourceId;
                msg->outModID = outModId;
                msg->fuProcID = fuProcessId;
                msg->fuGUID = fuGuid;
                msg->nExpectedEPs = nExpectedEPs;
        } while ((next = next->getNextReference()));

        try {
                fuAppContext_->postFrame(bufRef, fuAppDesc_, smAppDesc_);
        } catch (xdaq::exception::Exception &e) {
                string msg = "Failed to post INIT Message.";
                XCEPT_RETHROW(evf::Exception, msg, e);
        }

        return totalSize;
}

//______________________________________________________________________________
UInt_t SMProxy::sendDataEvent(UInt_t fuResourceId, UInt_t runNumber,
                UInt_t evtNumber, UInt_t outModId, UInt_t fuProcessId, UInt_t fuGuid,
                UChar_t *data, UInt_t dataSize) throw (evf::Exception) {
        UInt_t totalSize = 0;
        MemRef_t* bufRef = createFragmentChain(I2O_SM_DATA, dataHeaderSize_, data,
                        dataSize, totalSize);

        I2O_SM_DATA_MESSAGE_FRAME *msg;
        MemRef_t* next = bufRef;
        do {
                msg = (I2O_SM_DATA_MESSAGE_FRAME*) next->getDataLocation();
                msg->rbBufferID = fuResourceId;
                msg->runID = runNumber;
                msg->eventID = evtNumber;
                msg->outModID = outModId;
                msg->fuProcID = fuProcessId;
                msg->fuGUID = fuGuid;
        } while ((next = next->getNextReference()));

        try {
                fuAppContext_->postFrame(bufRef, fuAppDesc_, smAppDesc_);
        } catch (xdaq::exception::Exception &e) {
                string errmsg = "Failed to post DATA Message.";
                LOG4CPLUS_FATAL(log_, errmsg);
                XCEPT_RETHROW(evf::Exception, errmsg, e);
        }

        return totalSize;
}

//______________________________________________________________________________
UInt_t SMProxy::sendErrorEvent(UInt_t fuResourceId, UInt_t runNumber,
                UInt_t evtNumber, UInt_t fuProcessId, UInt_t fuGuid, UChar_t *data,
                UInt_t dataSize) throw (evf::Exception) {
        UInt_t totalSize = 0;
        MemRef_t* bufRef = createFragmentChain(I2O_SM_ERROR, dataHeaderSize_, data,
                        dataSize, totalSize);

        I2O_SM_DATA_MESSAGE_FRAME *msg;
        MemRef_t* next = bufRef;
        do {
                msg = (I2O_SM_DATA_MESSAGE_FRAME*) next->getDataLocation();
                msg->rbBufferID = fuResourceId;
                msg->runID = runNumber;
                msg->eventID = evtNumber;
                msg->outModID = 0xffffffff;
                msg->fuProcID = fuProcessId;
                msg->fuGUID = fuGuid;
        } while ((next = next->getNextReference()));

        try {
                fuAppContext_->postFrame(bufRef, fuAppDesc_, smAppDesc_);
        } catch (xdaq::exception::Exception &e) {
                string errmsg = "Failed to post ERROR Message.";
                LOG4CPLUS_FATAL(log_, errmsg);
                XCEPT_RETHROW(evf::Exception, errmsg, e);
        }

        return totalSize;
}

//______________________________________________________________________________
UInt_t SMProxy::sendDqmEvent(UInt_t fuDqmId, UInt_t runNumber,
                UInt_t evtAtUpdate, UInt_t folderId, UInt_t fuProcessId, UInt_t fuGuid,
                UChar_t*data, UInt_t dataSize) throw (evf::Exception) {
        UInt_t totalSize = 0;
        MemRef_t* bufRef = createFragmentChain(I2O_SM_DQM, dqmHeaderSize_, data,
                        dataSize, totalSize);

        I2O_SM_DQM_MESSAGE_FRAME *msg;
        MemRef_t* next = bufRef;
        do {
                msg = (I2O_SM_DQM_MESSAGE_FRAME*) next->getDataLocation();
                msg->rbBufferID = fuDqmId;
                msg->runID = runNumber;
                msg->eventAtUpdateID = evtAtUpdate;
                msg->folderID = folderId;
                msg->fuProcID = fuProcessId;
                msg->fuGUID = fuGuid;
        } while ((next = next->getNextReference()));

        try {
                fuAppContext_->postFrame(bufRef, fuAppDesc_, smAppDesc_);
        } catch (xdaq::exception::Exception &e) {
                string errmsg = "Failed to post DQM Message.";
                LOG4CPLUS_FATAL(log_, errmsg);
                XCEPT_RETHROW(evf::Exception, errmsg, e);
        }

        return totalSize;
}

// implementation of private member functions

//______________________________________________________________________________
MemRef_t* SMProxy::createFragmentChain(UShort_t i2oFunctionCode,
                UInt_t headerSize, UChar_t *data, UInt_t dataSize, UInt_t &totalSize)
                throw (evf::Exception) {
        totalSize = 0;

        UInt_t fragmentDataSizeMax = I2O_MAX_SIZE - headerSize;
        UInt_t fragmentCount = (dataSize / fragmentDataSizeMax);
        if (dataSize % fragmentDataSizeMax)
                ++fragmentCount;

        UInt_t currentPosition = 0;
        UInt_t remainingDataSize = dataSize;

        MemRef_t *head(0);
        MemRef_t *tail(0);

        try {

                for (UInt_t iFragment = 0; iFragment < fragmentCount; iFragment++) {

                        UInt_t fragmentDataSize = fragmentDataSizeMax;
                        UInt_t fragmentSize = fragmentDataSize + headerSize;

                        if (remainingDataSize < fragmentDataSizeMax) {
                                fragmentDataSize = remainingDataSize;
                                fragmentSize = fragmentDataSize + headerSize;
                                if (fragmentSize & 0x7)
                                        fragmentSize = ((fragmentSize >> 3) + 1) << 3;
                        }

                        // allocate the fragment buffer from the pool
                        toolbox::mem::Reference *bufRef =
                                        toolbox::mem::getMemoryPoolFactory()->getFrame(i2oPool_,
                                                        fragmentSize);

                        // set up pointers to the allocated message buffer
                        I2O_MESSAGE_FRAME *stdMsg;
                        I2O_PRIVATE_MESSAGE_FRAME *pvtMsg;
                        I2O_SM_MULTIPART_MESSAGE_FRAME *msg;

                        stdMsg = (I2O_MESSAGE_FRAME*) bufRef->getDataLocation();
                        pvtMsg = (I2O_PRIVATE_MESSAGE_FRAME*) stdMsg;
                        msg = (I2O_SM_MULTIPART_MESSAGE_FRAME*) stdMsg;

                        stdMsg->VersionOffset = 0;
                        stdMsg->MsgFlags = 0; // normal message (not multicast)
                        stdMsg->MessageSize = fragmentSize >> 2;
                        stdMsg->Function = I2O_PRIVATE_MESSAGE;
                        stdMsg->InitiatorAddress = i2o::utils::getAddressMap()->getTid(
                                        fuAppDesc_);
                        stdMsg->TargetAddress = i2o::utils::getAddressMap()->getTid(
                                        smAppDesc_);

                        pvtMsg->XFunctionCode = i2oFunctionCode;
                        pvtMsg->OrganizationID = XDAQ_ORGANIZATION_ID;

                        msg->dataSize = fragmentDataSize;
                        msg->hltLocalId = fuAppDesc_->getLocalId();
                        msg->hltInstance = fuAppDesc_->getInstance();
                        msg->hltTid = i2o::utils::getAddressMap()->getTid(fuAppDesc_);
                        msg->numFrames = fragmentCount;
                        msg->frameCount = iFragment;
                        msg->originalSize = dataSize;

                        for (UInt_t i = 0; i < fuUrl_.size(); i++)
                                msg->hltURL[i] = fuUrl_[i];
                        msg->hltURL[fuUrl_.size()] = '\0';

                        for (UInt_t i = 0; i < fuClassName_.size(); i++)
                                msg->hltClassName[i] = fuClassName_[i];
                        msg->hltClassName[fuClassName_.size()] = '\0';

                        if (iFragment == 0) {
                                head = bufRef;
                                tail = bufRef;
                        } else {
                                tail->setNextReference(bufRef);
                                tail = bufRef;
                        }

                        if (fragmentDataSize != 0) {
                                UChar_t* targetAddr = (UChar_t*) msg + headerSize;
                                std::copy(data + currentPosition,
                                                data + currentPosition + fragmentDataSize, targetAddr);
                        }

                        bufRef->setDataSize(fragmentSize);
                        remainingDataSize -= fragmentDataSize;
                        currentPosition += fragmentDataSize;
                        totalSize += fragmentSize;

                } // for (iFragment ...)
        } catch (toolbox::mem::exception::Exception& e) {
                if (0 != head)
                        head->release();
                totalSize = 0;
                string errmsg = "Failed to allocate buffer reference.";
                LOG4CPLUS_FATAL(log_, errmsg);
                XCEPT_RETHROW(evf::Exception, errmsg, e);
        } catch (xdaq::exception::ApplicationDescriptorNotFound& e) {
                if (0 != head)
                        head->release();
                totalSize = 0;
                string errmsg = "Failed to get tid.";
                LOG4CPLUS_FATAL(log_, errmsg);
                XCEPT_RETHROW(evf::Exception, errmsg, e);
        }

        return head;
}