Historizing with HistoryManagerBase

[SoftDevel]

I need to extend my current OPC server with historizing. I try to do this according to the tutorial in the SDK documentation, but in contrast to there my history comes from a database. The access to the database is not thread safe and i don't know how that HistoryManagerBase works, do i need to synchronize the access to the database? Can there be e.g. multiple outstanding readRaw() calls? Can i just lock a mutex during the whole runtime of readRaw() or can this lead to a deadlock?

And how do i use this HistoryReadCPUserDataBase parameter? I've read this can specify the point where to continue to read history data or the server must even set this parameter if it cannot deliver all requested data, but the documentation of this class says it has a constructor and a destructor, and nothing more.

This btw. i have encountered many times now reading the documentation, the documentation shows only constructor/destructor and nothing more, but obvisously there is more, or it just says this wraps some opc data structure and then end-of-documentation. This makes it pretty much impossible to use these elements, currently i simply avoided all these elements as good as i can or luckily found some example and copied it blind in hope it works, but in the long term this is not a solution. We bought a SDK license and want to use this in a productive manner soon, is there a way to get a more complete documentation?

[Support Team]

Hi,

If you check the readme of the SDK and the documentation of the HistoryManager interface you can see that the Historical Access functionality is not released yet. At the last major SDK release, the Historical Access specification was not released and interface changes still possible. For this reason the implementation, examples and documentation of the toolkit layer helper class HistoryManagerBase is not complete.

Since the specification is now released, we know that the existing HistoryManager interfaces in the Server SDK does not require any change. Customers of the C++ SDK have already implemented OPC UA servers for full historian systems based on version 1.3.

The HistoryManager is an asynchronous interface. The implementation helper class HistoryManagerBase makes this interface synchronous where every incoming read of a node is executed in its own worker thread from a thread pool. If a client reads five variables, five worker threads will call HistoryManagerBase::readRaw() at the same time. You are responsible for locking access to your data source in the HistoryManagerBase if necessary.

History read operations may have more results than the server is allowed to return (maxValues) or he can return (number exceeds the max array size of serializer - default is 65535). In this case the server returns only the allowed or possible number and a continuation point. Your HistoryManager must create an instance of a class derived from HistoryReadCPUserDataBase. This class must contain the information necessary to continue a read operation (e.g. the last timestamp returned). The SDK is responsible for maintaining the returned continuation point (returned through ppContinuationPoint.

If ppContinuationPoint contains a valid pointer when called, it contains the continuation point provided by your HistoryManager in a previous call. Your HistoryManager is responsible for deleting the passed in continuation point object after extracting the necessary information to continue the read operation.

Best Regards,
Unified Automation Support Team

[Support Team]

Here is also the updated example from the server lesson 7 with continuation point handling. The update is from the next release version under development.

First the example specific continuation point class derived from HistoryReadCPUserDataBase. The time stamp of the next value to return is used as continuation point.

Code: Select all

// History manager specific continuation point class
class HistoryReadCPUserDataBA : public HistoryReadCPUserDataBase
{
public:
    HistoryReadCPUserDataBA(){}
    ~HistoryReadCPUserDataBA(){}

    // We use the next time stamp to return as continuation point
    UaDateTime m_newStartTime;
};

And the readRaw implementation with continuation point.

Code: Select all

UaStatus HistoryManagerBuildingAutomation::readRaw (
    const ServiceContext&       serviceContext,
    HistoryVariableHandle*      pVariableHandle,
    HistoryReadCPUserDataBase** ppContinuationPoint,
    OpcUa_TimestampsToReturn    timestampsToReturn,
    OpcUa_UInt32                maxValues,
    OpcUa_DateTime&             startTime,
    OpcUa_DateTime&             endTime,
    OpcUa_Boolean               returnBounds,
    UaDataValues&               dataValues)
{
    // The NodeManagerBase is creating history variable handles of the type HistoryVariableHandleUaNode
    if ( pVariableHandle )
    {
        return OpcUa_BadNodeIdUnknown;
    }
    HistoryVariableHandleUaNode* pUaNodeVariableHandle = (HistoryVariableHandleUaNode*)pVariableHandle;

    // Check if the NodeId provided in the HistoryVariableHandleUaNode is a valid node to read
    std::map<UaNodeId, HistorizedVariable*>::iterator it;
    it = m_mapVariables.find(pUaNodeVariableHandle->pUaNode()->nodeId());
    if ( (it == m_mapVariables.end()) || (it->second->m_isValid == OpcUa_False) )
    {
        return OpcUa_BadNodeIdUnknown;
    }

    UaStatus            ret;
    HistorizedVariable* pVariable = it->second;
    OpcUa_UInt32        i         = 0;
    UaDateTime          dtStart(startTime);
    UaDateTime          dtEnd(endTime);
    OpcUa_Int64         iStart    = dtStart;
    OpcUa_Int64         iEnd      = dtEnd;

    // Check if we have a continuation point
    HistoryReadCPUserDataBA* pContinuationPoint = (HistoryReadCPUserDataBA*)*ppContinuationPoint;
    if ( pContinuationPoint )
    {
        // Set time from continuation point as new start time
        dtStart = pContinuationPoint->m_newStartTime;
        iStart    = dtStart;
        // Delete continuation point
        delete pContinuationPoint;
        // Set ppContinuationPoint point in/out to NULL
        *ppContinuationPoint = NULL;
    }

    if ( maxValues == 0 )
    {
        maxValues = OpcUa_Int32_Max;
    }

    // Lock access to list of values
    UaMutexLocker lock(&pVariable->m_mutex);

    if ( iStart < iEnd )
    {
        // Read in forward direction
        std::list<UaDataValue>::iterator itValues;
        dataValues.create(pVariable->m_values.size());
        for ( itValues=pVariable->m_values.begin(); itValues!=pVariable->m_values.end(); itValues++ )
        {
            // Check if we reached max values
            if ( i == maxValues )
            {
                // Create a continuation point
                HistoryReadCPUserDataBA* pContinuationPoint = new HistoryReadCPUserDataBA;
                // Use next time stamp as starting point
                pContinuationPoint->m_newStartTime = itValues->serverTimestamp();
                // Return continuation point
                *ppContinuationPoint = pContinuationPoint;
                break;
            }

            UaDateTime  dtVal(itValues->serverTimestamp());
            OpcUa_Int64 iVal = dtVal;

            if ( iVal < iStart )
            {
                // We have not found the start time yet
                continue;
            }

            if ( iVal > iEnd )
            {
                // We are behind the end time
                break;
            }

            if ( (i == 0) && (returnBounds != OpcUa_False) && (itValues != pVariable->m_values.begin()) )
            {
                // Bounds handling
                itValues--;
                itValues->copyTo(&dataValues[i]);
                itValues++;
                i++;
            }

            itValues->copyTo(&dataValues[i]);

            i++;
        }

        // Make size smaller if necessary
        dataValues.resize(i);
    }
    else
    {
        // Read in inverse direction
        std::list<UaDataValue>::reverse_iterator ritValues;
        dataValues.create(pVariable->m_values.size());
        for ( ritValues=pVariable->m_values.rbegin(); ritValues!=pVariable->m_values.rend(); ritValues++ )
        {
            // Check if we reached max values
            if ( i == maxValues )
            {
                // Create a continuation point
                HistoryReadCPUserDataBA* pContinuationPoint = new HistoryReadCPUserDataBA;
                // Use next time stamp as starting point
                pContinuationPoint->m_newStartTime = ritValues->serverTimestamp();
                // Return continuation point
                *ppContinuationPoint = pContinuationPoint;
                break;
            }

            UaDateTime  dtVal(ritValues->serverTimestamp());
            OpcUa_Int64 iVal = dtVal;

            if ( iVal > iStart )
            {
                // We have not found the start time yet
                continue;
            }

            if ( iVal < iEnd )
            {
                // We are behind the end time
                break;
            }

            if ( (i == 0) && (returnBounds != OpcUa_False) && (ritValues != pVariable->m_values.rbegin()) )
            {
                // Bounds handling
                ritValues--;
                ritValues->copyTo(&dataValues[i]);
                ritValues++;
                i++;
            }

            ritValues->copyTo(&dataValues[i]);

            i++;
        }

        // Make size smaller if necessary
        dataValues.resize(i);
    }

    return ret;
}