resuminggraph.cpp File Reference

Basic Counter Graph with persistent storage. More...

Go to the source code of this file.

Detailed Description

Basic Counter Graph with persistent storage.

Introduction

This examples cover the most basic way to preserve state across Graph instances.

This covers how to resume state and a neat way to define initial state.

Graph Architecture

The example uses a minimal Graph with a single Detector and two topics. The Detector counts EventHappened and publishes the total in EventCount.

ResumingCounter

State Persistence; Snapshots & ResumeFromSnapshotTopicState

This example shows how GraphStateStore and StateSnapshot can be used in conjunction with ResumeFromSnapshotTopicState to provide an extensible, robust and transparent state persistence mechanism.

The diagram below shows the life cycle of StateSnapshots during boot.

State Lifetime FSM

This life cycle can be seen in the main of this example:

int main()
{
    DetectorGraph::StateSnapshot primeSnapshot = GetPrimeSnapshot();
    DetectorGraph::StateSnapshot resumeSnapshot = ReadSnapshot(primeSnapshot);

    ResumingGraph resumingGraph = ResumingGraph(resumeSnapshot);
    for (int i = 0; i < 7; ++i)
    {
        resumingGraph.ProcessData(EventHappened());
        cout << "EventCount = " << resumingGraph.GetLastTopicState<EventCount>()->count << endl;

        const auto latestSnapshot = resumingGraph.GetLastState();
        WriteSnapshot(*latestSnapshot);
    }
}

PrimeSnapshot is a StateSnapshot that contains the initial state/data that should always be used - regardless of file-based state persistence. In this example we synthesize PrimeSnapshot this way:

DetectorGraph::StateSnapshot GetPrimeSnapshot()
{
    std::list< ptr::shared_ptr<const DetectorGraph::TopicState> > topicStatesList;
    topicStatesList.push_back(ptr::shared_ptr<const DetectorGraph::TopicState>(new EventCount(1000)));
    // ... add any other TopicStates that need Prime states.
    return DetectorGraph::StateSnapshot(topicStatesList);
}

ResumeSnapshot is the aggregation of TopicStates contained in PrimeSnapshot and what is deserialized from storage in ReadSnapshot():

DetectorGraph::StateSnapshot ReadSnapshot(const DetectorGraph::StateSnapshot& primeSnapshot)
{
    std::list< ptr::shared_ptr<const DetectorGraph::TopicState> > topicStatesList;

    std::ifstream snapshotInStream(kSavedSnapshotFileName);
    if (snapshotInStream.is_open())
    {
        // Generally here you'd have a loop through the TopicStates found on
        // Storage and adding each one to `topicStatesList`
        {
            int count;
            snapshotInStream >> count;
            topicStatesList.push_back(ptr::shared_ptr<const DetectorGraph::TopicState>(new EventCount(count)));
        }
        // ... add deserialization of any other TopicStates of interest.
    }

    return DetectorGraph::StateSnapshot(primeSnapshot, topicStatesList);
}

ResumeSnapshot is then used to construct a ResumeFromSnapshotTopicState which is then posted to the graph to allow Detectors to resume/initialize their state.

    void Evaluate(const DetectorGraph::ResumeFromSnapshotTopicState& aResumeFrom)
    {
        const auto previousEventCount = aResumeFrom.snapshot.GetState<EventCount>();
        if (previousEventCount)
        {
            mEventCount = *previousEventCount;
        }
    }

From then on mStateStore in ResumingGraph is continually updated from within ResumingGraph::ProcessOutput():

        mStateStore.TakeNewSnapshot(mGraph.GetOutputList());

The up-to-date latestSnapshot can then be flushed to disk as necessary with WriteSnapshot():

void WriteSnapshot(const DetectorGraph::StateSnapshot& aSnapshot)
{
    std::ofstream snapshotOutStream(kSavedSnapshotFileName);

    // Generally here you'd have a loop through a list of TopicState
    // serializers giving each one a chance to flush that topic's contents to
    // disk. And you'd normally use something fancier for Serialization (e.g.
    // protobufs).
    // For the purposes of this demo we'll mock that in a single method:
    {
        ptr::shared_ptr<const EventCount> countPtr = aSnapshot.GetState<EventCount>();
        if (countPtr)
        {
            snapshotOutStream << countPtr->count << endl;
        }
        // ... add serialization of any other TopicStates of interest.
    }
}

Note

Different applications may chose how to do this step differently. Depending on your chosen Serialization technology it can be too expensive to call WriteSnapshot() for each new processed input. Some applications may choose to only call WriteSnapshot() during shutdown. A more sophisticated approach is to call WriteSnapshot() only when a TopicState in a set of critical ones changes. This can be done by iterating through the list in mGraph.GetOutputList() as it only contains the changed TopicStates.

Running the Program

Running the program produces:

$ ./resuminggraph.out
DetectorGraph: Graph Initialized
EventCount = 1001
EventCount = 1002
EventCount = 1003
EventCount = 1004
EventCount = 1005
EventCount = 1006
EventCount = 1007

$ ./resuminggraph.out
DetectorGraph: Graph Initialized
EventCount = 1008
EventCount = 1009
EventCount = 1010
EventCount = 1011
EventCount = 1012
EventCount = 1013
EventCount = 1014

$ cat resumingGraphSnapshot.txt
1014

Definition in file resuminggraph.cpp.