IpcMsgQueue.cpp

Go to the documentation of this file.

// Copyright CERN 2012 - Developed in collaboration with GSI

#include <fesa-core/Core/IpcMsgQueue.h>

#include <fesa-core/Core/MessageTypes.h>
#include <fesa-core/Exception/FesaException.h>

#include <cmw-log/Logger.h>

#include <fstream> // To check the queue system limits
#include <sstream>
#include <cstring>
#include <cerrno>
#include <sys/resource.h> // get/set r_limit
#include <sys/stat.h>


namespace
{

CMW::Log::Logger& logger = CMW::Log::LoggerFactory::getLogger("FESA.FWK.fesa-core.Core.IpcMsgQueue");

} // namespace


namespace fesa
{

IpcMsgQueue::IpcMsgQueue(const std::string& name, uint32_t queuelength, bool blocking) :
    AbstractMsgQueue(name, IPC_MSG_QUEUE, queuelength)
{
    checkSystemMaxValues(queuelength);//check if the limits, provided by the system are sufficiant
    mq_attr attr;
    std::memset((void *) &attr, 0, sizeof(attr));
    attr.mq_msgsize = MSG_DATA_SIZE_MAX; // size of each message
    attr.mq_maxmsg = queuelength; // max number of messages
    attr.mq_flags = 0; // leave zero for mq_open

    int32_t mask = umask(0); // set the user mask to 0 in order to create the queue with the wanted rights
    int32_t oflags = O_CREAT | O_RDWR;
    if (blocking == false)
    {
        oflags |=  O_NONBLOCK; // Add the non blocking flag to the Q config
    }
    if ((msgQueue_ = mq_open(name.c_str(), oflags, 0666, &attr)) == (mqd_t) - 1)
    {
        std::string errMsg(std::strerror(errno));
        if (errno == EINVAL)
        {
            errMsg.append("\nRun the process as root!");
        }
        throw FesaException(__FILE__, __LINE__, FesaErrorCreatingMsgQueue.c_str(), name.c_str(), errMsg.c_str());
    }
    umask(mask); // restore the user mask
}

IpcMsgQueue::~IpcMsgQueue()
{
    mq_close( msgQueue_);
    mq_unlink(getQueueName().c_str());
}

void IpcMsgQueue::postMsg(AbstractMessage* msg)
{
    std::ostringstream stream(std::ios_base::out);
    msg->serialize(stream);//since we are a Inter Process Communication queue, we cannot send pointers
    if (mq_send(msgQueue_, stream.str().c_str(), MSG_DATA_SIZE_MAX, msg->prio_) == -1)
    {
        if (errno != EAGAIN)
        {
            throw FesaException(__FILE__, __LINE__, FesaErrorMsgQueueGeneric.c_str(), "Posting Msg on",
                                getQueueName().c_str(), std::strerror(errno));
        }
        throw FesaException(__FILE__, __LINE__, FesaErrorQueueLengthExceeded.c_str(), "Posting Msg on",
                            getQueueName().c_str());
    }
    if(logger.isLoggable(CMW::Log::Level::LL_TRACE))
    {
        std::ostringstream logMsg;
        logMsg << "Posted '" << stream.str() << "'";
        LOG_TRACE_IF(logger, logMsg.str());
    }
}

AbstractMessage* IpcMsgQueue::consumeMsg()
{
    ssize_t length;
    char data[MSG_DATA_SIZE_MAX];
    // on blocking mode, mq_receive waits, until a message arrives
    length = mq_receive(msgQueue_, data, MSG_DATA_SIZE_MAX, 0);
    if (length == -1)
    {
        throw FesaException(__FILE__, __LINE__, FesaErrorMsgQueueGeneric.c_str(), "Consuming Msg on",
                            getQueueName().c_str(), std::strerror(errno));
    }
    else
    {
        std::string message(data);
        std::istringstream stream(message);
        if(logger.isLoggable(CMW::Log::Level::LL_TRACE))
        {
            std::ostringstream logMsg;
            logMsg << "Received '" << stream.str() << "'";
            LOG_TRACE_IF(logger, logMsg.str());
        }
        uint32_t type;
        // get the type of the message from the stream
        stream >> type;
        switch (type)
        {
            case AutomaticNotificationMsg:
            {
                return new AutomaticNotificationMessage(stream);
                break;
            }
            case ManualNotificationMsg:
            {
                return new ManualNotificationMessage(stream);
                break;
            }
            case CommandMsg:
            {
                return new CommandMessage(stream);
                break;
            }
            case OnDemandMsg:
            {
                return new OnDemandMessage(stream);
                break;
            }
            default:
            {
                throw FesaException(__FILE__, __LINE__, FesaErrorUnknownMessageType.c_str(), getQueueName().c_str());
            }
        }
    }
    // Should never be executed but make gcc happy
    return (AbstractMessage*) NULL;
}

uint32_t IpcMsgQueue::getCurrentMsgCount()
{
    mq_attr attrs;

    if (mq_getattr(msgQueue_, &attrs) == -1)
    {
        throw FesaException(__FILE__, __LINE__, FesaErrorMsgQueueGeneric.c_str(), "Getting size on",
                            getQueueName().c_str(), std::strerror(errno));
    }

    return static_cast<uint32_t>(attrs.mq_curmsgs);
}

void IpcMsgQueue::purge()
{
    char msg[constants_.msg_size_max_]; // the max msg size
    while (getCurrentMsgCount() > 0)
    {
        if (mq_receive(msgQueue_, msg, constants_.msg_size_max_, NULL) == -1)
        {
            throw FesaException(__FILE__, __LINE__, FesaErrorMsgQueueGeneric.c_str(), "Purging queue on",
                                getQueueName().c_str(), std::strerror(errno));
        }
    }
}

void IpcMsgQueue::checkSystemMaxValues(uint32_t queuelength)
{
    bool changed = false;
    // ----------------------------------------
    // -- 1 -- First we check the system-limits
    // ----------------------------------------

    std::ifstream file_max_message_size;
    std::ifstream file_max_queue_length;
    uint32_t system_max_message_size;
    uint32_t system_max_queue_length;

    file_max_message_size.open("/proc/sys/fs/mqueue/msgsize_max", std::ifstream::in);
    file_max_message_size >> system_max_message_size;
    file_max_message_size.close();

    file_max_queue_length.open("/proc/sys/fs/mqueue/msg_max", std::ifstream::in);
    file_max_queue_length >> system_max_queue_length;
    file_max_queue_length.close();

    if (system_max_message_size < constants_.msg_size_max_)
    {
        std::stringstream out;//for conversions
        out << constants_.msg_size_max_;
        throw FesaException(__FILE__, __LINE__, FesaErrorSystemMessageSizeExceeded.c_str(), out.str().c_str(),
                            out.str().c_str(), out.str().c_str());
    }

    if (system_max_queue_length < queuelength)
    {
        std::stringstream out;//for conversions
        out << queuelength;
        throw FesaException(__FILE__, __LINE__, FesaErrorSystemQueueLengtExceeded.c_str(), out.str().c_str(),
                            out.str().c_str(), out.str().c_str());
    }

    // ----------------------------------------
    // -- 2 -- Now we check (and fix) the soft and hard-limit of our process
    // ----------------------------------------

    //get the current limits for this process
    rlimit currentLimit;
    getrlimit(RLIMIT_MSGQUEUE,&currentLimit);//check if limit is sufficient

    //calculate the needed limits
    uint32_t neededLimit = queuelength * static_cast<uint32_t>(sizeof(struct msg_msg *)) + queuelength * MSG_DATA_SIZE_MAX;

    if(currentLimit.rlim_max <= neededLimit)
    {
        currentLimit.rlim_max = neededLimit;
        changed = true;
    }
    if(currentLimit.rlim_cur <= neededLimit)
    {
        currentLimit.rlim_cur = neededLimit;
        changed = true;
    }
    if(changed)
    {
        if(setrlimit(RLIMIT_MSGQUEUE,&currentLimit) == -1)//check
        {
            throw FesaException(__FILE__, __LINE__, FesaErrorMsgQueueGeneric.c_str(), "Error while setting process",
                                "posix-message-queue-memory hard/soft-limit(rlim_max/rlim_cur). Operation 'setrlimit' could not been performed. Try to login as root, or to give this process the CAP_SYS_RESOURCE capability.",std::strerror(errno));
        }
    }
}

} // fesa