SubscriptionFilter.cpp

Go to the documentation of this file.

// Copyright CERN 2012 - Developed in collaboration with GSI

#include <fesa-core/RDADeviceServer/SubscriptionFilter.h>

#include <fesa-core/RDADeviceServer/NotificationThread.h>
// #include <fesa-core/RDADeviceServer/FesaDeviceServer.h> // TODO this dependency is just to retrieve the update policy .. to be reviewed
#include <fesa-core/Server/Property.h>
#include <fesa-core/DataStore/AbstractDevice.h>
#include <fesa-core/Synchronization/SynchronizationLabObjectFactory.h>
#include <fesa-core/Synchronization/CycleDescriptor.h>
#include <fesa-core/Utilities/Lock.h>
#include <fesa-core/Utilities/Mutex.h>

#include <cmw-rda/ValueChangeListener.h>
#include <cmw-rda/Data.h>
#include <cmw-log/Logger.h>

#include <boost/lexical_cast.hpp>

#include <iostream>

namespace
{
CMW::Log::Logger& logger = CMW::Log::LoggerFactory::getLogger("FESA.FWK.fesa-core.RDADeviceServer.SubscriptionFilter");
}


namespace fesa
{

AbstractSubscriptionFilter::AbstractSubscriptionFilter(AbstractDevice& device, Property& property, const rdaData& newFilter, bool useFastUpdate) :
    device_(device),
    property_(property),
    filter_(newFilter),
    useFastUpdate_(useFastUpdate)
{
}

AbstractSubscriptionFilter::~AbstractSubscriptionFilter()
{
}

void AbstractSubscriptionFilter::addSubscriber(const std::string& cycleSelector, rdaValueChangeListener* subscriber)
{
    // Create a new Listener
    boost::shared_ptr<FesaListener> l(new FesaListener(subscriber));

    LstnrMapItr itr = listeners_.find(cycleSelector);
    if (itr != listeners_.end())
    {
        // we found a listenerVector for this cycleSelector. Add the new listener
        (*itr).second->push_back(l);
    }
    else
    {
        boost::shared_ptr<ListenerVector> lv(new ListenerVector());
        lv->push_back(l);
        listeners_[cycleSelector] = lv;
    }
}

bool AbstractSubscriptionFilter::removeSubscriber(const std::string& cycleSelector, rdaValueChangeListener* subscriber, bool& isEmpty, bool& toBeDeleted)
{
    LstnrMapItr itr = listeners_.find(cycleSelector);
    if (itr != listeners_.end())
    {
        LstnrItr itrLstnr = (*itr).second->begin();
        LstnrItr itrEndLstnr = (*itr).second->end();
        for (; itrLstnr != itrEndLstnr; ++itrLstnr)
        {
            if (((*itrLstnr)->subscriber_) == subscriber) // check if the two references referred to the same rdaValueChangeListener instance
            {
                (*itr).second->erase(itrLstnr);
                if ((*itr).second->size() == 0)
                {
                    listeners_.erase(itr);
                }
                // leave the loop
                isEmpty = listeners_.empty();
                // only useful for background subsriptionFilter
                toBeDeleted = isEmpty;
                return true;
            }
        }
    }
    isEmpty = listeners_.empty();
    toBeDeleted = isEmpty;
    // subscriber was not found.
    return false;
}

bool AbstractSubscriptionFilter::matchFilter(const rdaData& newfilter)
{
    if (newfilter.size() == filter_.size()) //could match ..lets try
    {
        rdaDataIterator newFilterIter(newfilter);
        rdaDataIterator oldFilterIter(filter_);
        rdaDataEntry* newFilterEntry = newFilterIter.next();
        rdaDataEntry* oldFilterEntry = oldFilterIter.next();
        if (newFilterEntry == NULL && oldFilterEntry == NULL)
            return true; //we found an empty filter matching the new empty filter
        while (*newFilterEntry == *oldFilterEntry)//as long, as filter's entry do not differ
        {
            //compare next elements of the two filters
            newFilterEntry = newFilterIter.next();
            oldFilterEntry = oldFilterIter.next();
            if (newFilterEntry == NULL && oldFilterEntry == NULL)
                return true;//we reached the end of the filter, without getting kicked out --> we have a match
        }//end while -- comparison of two filters
    }//end if filter size is equal
    return false;//we reach the end of the loop?  --> nothing found
}

void AbstractSubscriptionFilter::cycleFirstUpdate(const std::string& cycleSelector, const rdaData& newFilter, rdaValueChangeListener* l)
{
    const rdaData emptydata;
    rdaData value;
    try
    {
        property_.get(device_, cycleSelector, newFilter, value, SUBSCRIBE);
    }
    catch (FesaException& ex)
    {
        try
        {
            l->ioFailed(rdaIOError("FESA", ex.getErrorCodeAsLong(), ex.getMessage().c_str()));
        }
        catch (rdaInternalError& ex)
        {
        }
        return;
    }
    try
    {
        //thats the "direct","slow" notification
        l->valueUpdated(emptydata, value, true);
    }
    catch (rdaInternalError& ex)
    {
    };

}

void AbstractSubscriptionFilter::allCyclesFirstUpdate(const rdaData& filter, rdaValueChangeListener* l)
{
    // get the correspondent cycle descriptor
    SynchronizationLabObjectFactory* syncFactory = SynchronizationLabObjectFactory::getInstance();
    CycleDescriptor* timingDescriptor = syncFactory->getCycleDescriptor(device_.timingDomain.get(),device_.mainMuxCriterion.get());

    const std::vector<int32_t>& idCol = timingDescriptor->getCycleIdsCol();

    //for all cycles in this Descriptor
    for (std::vector<int32_t>::const_iterator itr = idCol.begin(); itr != idCol.end(); itr++)
    {

        // the cycleName is returned by copy, therefore it's safe even if in the meanwhile
        // the timingDescriptor selectorNames has been refreshed
        std::string cycleName = timingDescriptor->getCycleSelectorName((*itr));
        cycleFirstUpdate(cycleName, filter, l);
    }
}


void AbstractSubscriptionFilter::reportInternalError(const std::string& category, int32_t errorCode, const std::string& message, bool changed)
{
    char* c = const_cast<char*> (category.c_str());
    char* m = const_cast<char*> (message.c_str());
    rdaData emptydata;

    LstnrMapItr itr = listeners_.begin();
    LstnrMapItr itrEnd = listeners_.end();
    for(; itr!=itrEnd; ++itr)
    {
        LstnrItr itrLstnr = (*itr).second->begin();
        LstnrItr itrEndLstnr = (*itr).second->end();
        for (; itrLstnr != itrEndLstnr; ++itrLstnr)
        {
            try {
                bool newError = changed;
                if (!(*itrLstnr)->reportErrorOnce_)
                {
                    newError = true;
                    (*itrLstnr)->reportErrorOnce_ = true;
                }
                (*itrLstnr)->subscriber_->ioFailed(c,errorCode,m,newError);
            }
            catch (rdaInternalError& ex)
            {
            }
        }
    }
    rdaValueChangeListener::flush(emptydata);
}

void AbstractSubscriptionFilter::printDebugState(std::stringstream& str)
{
    std::stringstream temp;
    temp << filter_;
    std::string s(temp.str());
    std::string::size_type pos;
    while ((pos=s.find('\n')) != std::string::npos)
    {
        s.replace(pos,1," ");
    }
    str << "\t\t\tFilter: " << s << "\n";
    LstnrMapItr itr = listeners_.begin();
    LstnrMapItr itrEnd = listeners_.end();
    for(; itr!=itrEnd; ++itr)
    {
        str << "\t\t\t\tCycleSelector: " << (*itr).first << "\n";
        LstnrItr itrLstnr = (*itr).second->begin();
        LstnrItr itrEndLstnr = (*itr).second->end();
        str << "\t\t\t\tListeners:\n";
        for (; itrLstnr != itrEndLstnr; ++itrLstnr)
        {
            str << "\t\t\t\t\t" << (*itrLstnr)->subscriber_ << "\n";
        }
    }
}



// Specialized class: SubscriptionFilter provide basically a specific implementation for the notify
SubscriptionFilter::SubscriptionFilter(AbstractDevice& device, Property& property, const rdaData& newFilter, bool useFastUpdate, NotificationThread* pNotificationThread):
    AbstractSubscriptionFilter(device, property, newFilter, useFastUpdate), pNotificationThread_(pNotificationThread), cancelled_(false)
{
}

SubscriptionFilter::~SubscriptionFilter()
{

}

void SubscriptionFilter::addSubscriber(const std::string& cycleSelector, rdaValueChangeListener* subscriber)
{
    // Create a new Listener
    boost::shared_ptr<FesaListener> l(new FesaListener(subscriber));

    // CriticalSection requiring to lock the Listeners using the mutex kept by the backgroundThread running notify()
    Lock lockListener(pNotificationThread_->getMutex());
    LstnrMapItr itr = listeners_.find(cycleSelector);
    if (itr != listeners_.end())
    {
        // we found a listenerVector for this cycleSelector. Add the new listener
        (*itr).second->push_back(l);
    }
    else
    {
        boost::shared_ptr<ListenerVector> lv(new ListenerVector());
        lv->push_back(l);
        listeners_[cycleSelector] = lv;
    }
}



bool SubscriptionFilter::removeSubscriber(const std::string& cycleSelector, rdaValueChangeListener* subscriber, bool& isEmpty, bool& toBeDeleted)
{
    LOG_DEBUG_IF(logger, "removeSubscriber");
    // CriticalSection requiring to lock the Listeners using the mutex kept by the backgroundThread running notify()
    Lock lockListener(pNotificationThread_->getMutex());
    LstnrMapItr itr = listeners_.find(cycleSelector);
    if (itr != listeners_.end())
    {
        LstnrItr itrLstnr = (*itr).second->begin();
        LstnrItr itrEndLstnr = (*itr).second->end();
        for (; itrLstnr != itrEndLstnr; ++itrLstnr)
        {
            if ((*itrLstnr)->subscriber_ == subscriber) // check if the two references referred to the same rdaValueChangeListener instance
            {
                (*itr).second->erase(itrLstnr);
                if ((*itr).second->size() == 0)
                {
                    listeners_.erase(itr);
                }
                // leave the loop
                isEmpty = listeners_.empty();
                toBeDeleted = isEmpty;
                if (isEmpty && pendingNotification_)
                {
                    // The subscriptionFilter is empty therefore it will be deleted, but because
                    // there is pending notification, we have to remove it from teh background thread queue
                    bool isInUse = pNotificationThread_->removeNotificationEvent(this);
                    pendingNotification_--;
                    if (isInUse)
                    {
                        toBeDeleted = false;
                        cancelled_ = true;
                    }
                }
                return true;
            }
        }
    }
    isEmpty = listeners_.empty();
    // subscriber was not found.
    return false;
}

// The main goal is to minimize the critical code to avoid as more as possible priority inversion, namely,
// a background:notify should not prevent
void SubscriptionFilter::notify(MultiplexingContext& muxContext,
                                RequestType& reqType, std::vector<std::string>& validCS)
{
    //Iterate trough the validCS
    std::vector<std::string>::iterator itrValidCS = validCS.begin();
    std::vector<std::string>::iterator itrEndValidCS = validCS.end();

    // CriticalSection requiring to lock the Listeners
    Lock lockListener(pNotificationThread_->getMutex());
    for (; itrValidCS != itrEndValidCS; ++itrValidCS)
    {
        // find an entry using the key
        LstnrMapItr itrLstnrMap = listeners_.find((*itrValidCS));
        if (itrLstnrMap != listeners_.end())
        {
            pNotificationThread_->addNotificationEvent(muxContext, reqType, validCS, this);
            ++pendingNotification_;
        }
    }
}

// The main goal is to minimize the critical code to avoid as more as possible priority inversion, namely,
// a background:notify should not prevent
void SubscriptionFilter::doUpdate(MultiplexingContext& muxContext, RequestType& reqType, std::vector<std::string>& validCS)
{
    //Iterate trough the validCS
    std::vector<std::string>::iterator itrValidCS = validCS.begin();
    std::vector<std::string>::iterator itrEndValidCS = validCS.end();
    const rdaData emptydata;
    rdaData pushdata;
    const rdaData* dp = &pushdata;
    bool gotData = false;
    bool gotError = false;
    std::string currentError("");
    int32_t currentErrorCode = 0;
    bool needData = false;
    int64_t dataStamp;

    Mutex& notificationThreadMutex = pNotificationThread_->getMutex();
    // CriticalSection requiring to lock the Listeners
    {
        Lock lockListener(notificationThreadMutex);
        for (; itrValidCS != itrEndValidCS; ++itrValidCS)
        {
            // find an entry using the key
            LstnrMapItr itrLstnrMap = listeners_.find((*itrValidCS));
            if (itrLstnrMap != listeners_.end())
            {
                needData = true;
                break;
            }
        }
        pendingNotification_--;
    }
    // end of the critical section: concurrent thread may run and change SubscriptionTree

    // Background property is accessed out of the critical section
    // to minimize the notification inversion priority
    if(needData)
    {
        try
        {
            dataStamp = property_.getDataTimestamp(device_, muxContext, filter_);
            property_.get(device_, muxContext, filter_, pushdata, reqType);
            gotData = true;
        }
        catch (FesaException& ex)
        {
            // cache the errorCode and errorMessage for the following listeners
            currentErrorCode = static_cast<int32_t>(ex.getErrorCodeAsLong());
            currentError = ex.getMessage();
            gotError = true;
        }
    }

    // CriticalSection requiring to lock the Listeners
    {
        Lock lockListener(notificationThreadMutex);
        if (cancelled_)
        {
            // This SubscriptionFilter has been mark to be deleted in meanwhile, just return;
            delete this;
            return;
        }
        for (; itrValidCS != itrEndValidCS; ++itrValidCS)
        {
            // find an entry using the key
            LstnrMapItr itrLstnrMap = listeners_.find((*itrValidCS));
            if (itrLstnrMap != listeners_.end())
            {
                LstnrItr itrLstnr = (*itrLstnrMap).second->begin();
                LstnrItr itrEndLstnr = (*itrLstnrMap).second->end();
                for (; itrLstnr != itrEndLstnr; ++itrLstnr)
                {
                    if (gotData)
                    {
                        // FIXME there is probably a bug with the subscriber object when one unsubscribe see LOG_ERROR_IF(logger, boost::lexical_cast<std::string>((*itrLstnr)->subscriber_->isOnChange()));
                        bool isOnChange = (*itrLstnr)->subscriber_->isOnChange();
                        if (gotData && (!isOnChange || (isOnChange && dataStamp) > (*itrLstnr)->updateStamp_))
                        {
                            try
                            {
                                if (useFastUpdate_)
                                    (*itrLstnr)->subscriber_->valueUpdated(*dp, true);
                                else
                                    (*itrLstnr)->subscriber_->valueUpdated(emptydata, *dp, true);
                                (*itrLstnr)->updateStamp_ = dataStamp;
                            }
                            catch (rdaInternalError& ie)
                            {
                            };
                        }
                    }
                    if (gotError)
                    {
                        try
                        {
                            (*itrLstnr)->subscriber_->ioFailed(rdaIOError("FESA", currentErrorCode,
                                                                          currentError.c_str()));
                            continue;
                        }
                        catch (rdaInternalError& ie)
                        {
                            // TODO : continue or break check if it's make sense to try to set all the subscriber
                            continue;
                        }
                    }
                }
            }
        }
    }
    // end of the critical section: concurrent thread may run and change SubscriptionTree
}

} // fesa