State Collection Interface Documentation (Version before 2.0) - FREEDM-DGI/FREEDM GitHub Wiki

The DGI State Collection module is implemented based on the Chandy-Lamport algorithm [1], which is used to collect consistent states of all participants in a distributed system. A consistent global state is one corresponding to a consistent cut. A consistent cut is left closed under the causal precedence relation. In another words, if one event is belongs to a cut, and all events happened before this event also belong to the cut, then the cut is considered to be consistent. The algorithm works as follows:

• The initiator starts state collection by recording its own states and broadcasting a marker out to other peers. At the same time, the initiator starts recording messages from other peers until it receives the marker back from other peers.
• Upon receiving the marker for the first time, the peer records its own state, forwards the marker to others (include the initiator) and starts recording messages from other peers until it receives the marker back from other peers.

The following diagram illustrates the Chandy-Lamport algorithm working on three nodes. The initiator is the leader node chosen by Group Management module in DGI.

Marker: std::pair<std::string, int>. Marker is tracked by the local node’s UUID in addition to a locally incremented integer.

Collected states: multimap <Marker, ptree>. Collected states use the ptree structure and are recorded in multimap with marker’s version. A ptree is a data structure in Boost that stores an arbitrarily deeply nested tree of values, indexed at each level by some key [2]. Each node of the tree stores its own value, plus an ordered list of its subnodes and their keys. The tree allows easy access to any of its nodes by means of a path, which is a concatenation of multiple keys. Based on different types of states (gateway, channel message and so on), different states will be prepared as a ptree in the following way.

//channel message

m_state.put("sc.type", "Message");
m_state.put("sc.transit.value", VALUE);
m_state.put("sc.transit.source", SOURCE);
m_state.put("sc.transit.destin", DESTINATION);

//TakeSnapshot()collects specific state for specific device by calling function in CPhysicalDeviceManager.cpp. The valueType could be gateway for SST, generation for DRER, storage for DESD, drain for LOAD and state for FID

m_state.put("sc.type", valueType);
m_state.put("sc.gateway", VALUE);
m_state.put("sc.source", SOURCE);

• Initiate(): The initiator use this function to start a state collection by recording its own states and broadcasting the marker out to all other peers. Currently, the initiator is the leader node chosen by the Group Management module.
• TakeSnapshot (deviceType, valueType): This function is used to save a local state. Given the type of the device (deviceType) and the type of the value (valueType), request state will be recorded.

Currently, the function is only implemented to recorded value for a specific device.

TakeSnapshot(std::string deviceType, std::string valueType)
{
    device::SettingValue PowerValue;
    //calling GetValue function in CPhysicalDeviceManger.cpp to obtain state value of specific device
	PowerValue = m_phyDevManager->GetValue(deviceType, valueType, &device::SumValues);
    //save local value into ptree m_curstate
    m_curstate.put("sc.type", valueType);
    m_curstate.put("sc.value", PowerValue);
    m_curstate.put("sc.source", GetUUID());
}

Examples on different devices, those parameters are received from messages from other module requesting for specific device:

TakeSnapshot("SST", "gateway");
TakeSnapshot("DRER", "generation");
TakeSnapshot("DESD", "storage");
TakeSnapshot("LOAD", "drain");
TakeSnapshot("FID", "state");

• SendStateBack(): This function indicates any node except the initiator has completed its local state collection and is called to send its collected states to the initiator.
• HandleRead(): This function is called upon every incoming message.
• StateResponse(): This function is called after the initiator has received states from all other peers and has completed its local state collection. All collected states will be separated into different kinds of strings: for example, string for gateway value, string for channel message.

State Collection is a passive process that won’t be started until it receives a message from other modules. It communicates with other modules through message passing. The message should be created with contents of “request”, the requested device type, the requested value type, UUID of the node and module sending the request. Currently, the state collection can collect states of one specific device. The following example illustrates the Load Balance (LB) calling State Collection (SC) for the gateway value and messages passing back and forth:

#include "StateCollection.hpp"
#include "SCPeerNode.hpp"
#include "device/DeviceMath.hpp"
#include "CMessage.hpp"

#include <algorithm>
#include <cassert>
#include <exception>
#include <sys/types.h>
#include <unistd.h>
#include <iomanip>
#include <fstream>
#include <string>
#include <stdlib.h>
#include <iostream>
#include <set>

#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/function.hpp>
#include <boost/range/adaptor/map.hpp>
#include <boost/foreach.hpp>

#define foreach     BOOST_FOREACH

#include <vector>
#include <boost/assign/list_of.hpp>
#include <boost/serialization/string.hpp>
#include <boost/serialization/vector.hpp>
#include <boost/serialization/shared_ptr.hpp>

#include "CConnection.hpp"
#include "CBroker.hpp"

#include <boost/property_tree/ptree.hpp>
using boost::property_tree::ptree;
#include "CLogger.hpp"
#include <map>

deviceType and valueType will be used as parameters sending to TakeSnapshot function :

CMessage m_sc;
m_sc.m_submessages.put("sc", "request");
m_sc.m_submessages.put("sc.deviceType", "SST");
m_sc.m_submessages.put("sc.valueType", "gateway");
m_sc.m_submessages.put("sc.source", GetUUID());
m_sc.m_submessages.put("sc.module", "lb");`

if (GetPeer(GetUUID()) != NULL)
{
    try
    {
        GetPeer(GetUUID())->AsyncSend(m_sc);
    }
	catch (boost::system::system_error& e)
    {
        Logger.Info << "Couldn't Send Message To Peer" << std::endl;
    }
}
else
{
    Logger.Info << "Peer doesn't exist" << std::endl;
}

Each node will call Takesnapshot() to save its local gateway. Each peer node except the Initiator will call SendStateBack() to send collected state back to the initiator. After receiving all peers’ “Done” message, the initiator will call StateResponse() to categorize collected states into a value string and in-transit the value string back to the requested module - LB.

m_.m_submessages.put("lb",, "CollectedState");
if ("sc.type"==m_valueType)
    m_.m_submessages.add("CollectedState.state.value", stateValueString);
else if ("sc.type"=="transit message")
    m_.m_submessages.add("CollectedState.intransit.value", transitValuesString);

LB can print out the states based on the type:

if(pt.get<std::string>("lb") == "CollectedState")    
{	
    foreach(ptree::value_type &v, pt.get_child("CollectedState.state"))
    {
        Logger.Notice << "SC module returned state values: " << v.second.data() << std::endl;
    }
    foreach(ptree::value_type &v, pt.get_child("CollectedState.intransit"))
    {
        Logger.Notice << "SC module returned intransit message: " << v.second.data() << std::endl;
    }
}

eg.
SC module returned state values: 20
SC module returned state values: -5
SC module returned state values: -20
SC module returned intransit message: Demand from UUID#1 to UUID#2
SC module returned intransit message: Accept from UUID#2 to UUID#3

[1] Distributed Snapshots: Determining Global States of Distributed Systems, ACM Transactions on Computer Systems, Vol. 3, No. 1, 1985, pp. 63-75.

[2] http://www.boost.org/doc/libs/1_41_0/doc/html/property_tree.html. Accessed on July 16th, 2012.