blob: 293d90cd30e5976c4905643a60fa64a97dde6e92 [file] [log] [blame]
/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
/*
* Copyright (c) 2012 University of California, Los Angeles
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License version 2 as
* published by the Free Software Foundation;
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*
* Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
* Chaoyi Bian <bcy@pku.edu.cn>
* Alexander Afanasyev <alexander.afanasyev@ucla.edu>
* Yingdi Yu <yingdi@cs.ucla.edu>
*/
#ifdef NS3_MODULE
#include <ns3/ccnx-pit.h>
#include <ns3/ccnx.h>
#endif
#include "sync-logic.h"
#include "sync-diff-leaf.h"
#include "sync-full-leaf.h"
#include "sync-logging.h"
#include "sync-state.h"
#include <boost/make_shared.hpp>
#include <boost/foreach.hpp>
#include <boost/lexical_cast.hpp>
#include <vector>
using namespace std;
using namespace ndn;
INIT_LOGGER ("SyncLogic");
#ifdef NS3_MODULE
#define GET_RANDOM(var) var.GetValue ()
#else
#define GET_RANDOM(var) var ()
#endif
#define TIME_SECONDS_WITH_JITTER(sec) \
(TIME_SECONDS (sec) + TIME_MILLISECONDS (GET_RANDOM (m_reexpressionJitter)))
#define TIME_MILLISECONDS_WITH_JITTER(ms) \
(TIME_MILLISECONDS (ms) + TIME_MILLISECONDS (GET_RANDOM (m_reexpressionJitter)))
namespace Sync
{
SyncLogic::SyncLogic (const Name& syncPrefix,
shared_ptr<Validator> validator,
shared_ptr<Face> face,
LogicUpdateCallback onUpdate,
LogicRemoveCallback onRemove)
: m_state (new FullState)
, m_syncInterestTable (TIME_SECONDS (m_syncInterestReexpress))
, m_syncPrefix (syncPrefix)
, m_onUpdate (onUpdate)
, m_onRemove (onRemove)
, m_perBranch (false)
, m_validator(validator)
, m_keyChain(new KeyChain())
, m_face(face)
, m_scheduler(*face->ioService())
#ifndef NS3_MODULE
, m_randomGenerator (static_cast<unsigned int> (std::time (0)))
, m_rangeUniformRandom (m_randomGenerator, boost::uniform_int<> (200,1000))
, m_reexpressionJitter (m_randomGenerator, boost::uniform_int<> (100,500))
#else
, m_rangeUniformRandom (200,1000)
, m_reexpressionJitter (10,500)
#endif
, m_recoveryRetransmissionInterval (m_defaultRecoveryRetransmitInterval)
{
#ifndef NS3_MODULE
// In NS3 module these functions are moved to StartApplication method
m_syncRegisteredPrefixId = m_face->setInterestFilter (m_syncPrefix,
bind(&SyncLogic::onSyncInterest, this, _1, _2),
bind(&SyncLogic::onSyncRegisterFailed, this, _1));
m_reexpressingInterestId = m_scheduler.scheduleEvent (time::seconds (0), // no need to add jitter
bind (&SyncLogic::sendSyncInterest, this));
#endif
}
SyncLogic::SyncLogic (const Name& syncPrefix,
shared_ptr<Validator> validator,
shared_ptr<Face> face,
LogicPerBranchCallback onUpdateBranch)
: m_state (new FullState)
, m_syncInterestTable (TIME_SECONDS (m_syncInterestReexpress))
, m_syncPrefix (syncPrefix)
, m_onUpdateBranch (onUpdateBranch)
, m_perBranch(true)
, m_validator(validator)
, m_keyChain(new KeyChain())
, m_face(face)
, m_scheduler(*face->ioService())
#ifndef NS3_MODULE
, m_randomGenerator (static_cast<unsigned int> (std::time (0)))
, m_rangeUniformRandom (m_randomGenerator, boost::uniform_int<> (200,1000))
, m_reexpressionJitter (m_randomGenerator, boost::uniform_int<> (100,500))
#else
, m_rangeUniformRandom (200,1000)
, m_reexpressionJitter (10,500)
#endif
, m_recoveryRetransmissionInterval (m_defaultRecoveryRetransmitInterval)
{
#ifndef NS3_MODULE
// In NS3 module these functions are moved to StartApplication method
m_syncRegisteredPrefixId = m_face->setInterestFilter (m_syncPrefix,
bind(&SyncLogic::onSyncInterest, this, _1, _2),
bind(&SyncLogic::onSyncRegisterFailed, this, _1));
m_reexpressingInterestId = m_scheduler.scheduleEvent (time::seconds (0), // no need to add jitter
bind (&SyncLogic::sendSyncInterest, this));
#endif
}
SyncLogic::~SyncLogic ()
{
// m_face->unsetInterestFilter(m_syncRegisteredPrefixId);
}
#ifdef NS3_MODULE
void
SyncLogic::StartApplication ()
{
m_ccnxHandle->SetNode (GetNode ());
m_ccnxHandle->StartApplication ();
m_ccnxHandle->setInterestFilter (m_syncPrefix,
func_lib::bind (&SyncLogic::respondSyncInterest, this, _1));
m_scheduler.schedule (TIME_SECONDS (0), // need to send first interests at exactly the same time
func_lib::bind (&SyncLogic::sendSyncInterest, this),
REEXPRESSING_INTEREST);
}
void
SyncLogic::StopApplication ()
{
m_ccnxHandle->clearInterestFilter (m_syncPrefix);
m_ccnxHandle->StopApplication ();
m_scheduler.cancel (REEXPRESSING_INTEREST);
m_scheduler.cancel (DELAYED_INTEREST_PROCESSING);
}
#endif
void
SyncLogic::stop()
{
m_face->unsetInterestFilter(m_syncRegisteredPrefixId);
m_scheduler.cancelEvent (m_reexpressingInterestId);
m_scheduler.cancelEvent (m_delayedInterestProcessingId);
}
/**
* Two types of intersts
*
* Normal name: .../<hash>
* Recovery name: .../recovery/<hash>
*/
boost::tuple<DigestConstPtr, std::string>
SyncLogic::convertNameToDigestAndType (const Name &name)
{
BOOST_ASSERT (m_syncPrefix.isPrefixOf(name));
int nameLengthDiff = name.size() - m_syncPrefix.size();
BOOST_ASSERT (nameLengthDiff > 0);
BOOST_ASSERT (nameLengthDiff < 3);
string hash = name.get(-1).toEscapedString();
string interestType;
if(nameLengthDiff == 1)
interestType = "normal";
else
interestType = name.get(-2).toEscapedString();
_LOG_TRACE (hash << ", " << interestType);
DigestPtr digest = boost::make_shared<Digest> ();
istringstream is (hash);
is >> *digest;
return make_tuple (digest, interestType);
}
void
SyncLogic::onSyncInterest (const shared_ptr<const Name>& prefix,
const shared_ptr<const ndn::Interest>& interest)
{
Name name = interest->getName();
_LOG_DEBUG("respondSyncInterest: " << name);
try
{
_LOG_TRACE ("<< I " << name);
DigestConstPtr digest;
string type;
tie (digest, type) = convertNameToDigestAndType (name);
if (type == "normal") // kind of ineffective...
{
processSyncInterest (name, digest);
}
else if (type == "recovery")
{
processSyncRecoveryInterest (name, digest);
}
}
catch (Error::DigestCalculationError &e)
{
_LOG_TRACE ("Something fishy happened...");
// log error. ignoring it for now, later we should log it
return ;
}
}
void
SyncLogic::onSyncRegisterFailed(const shared_ptr<const Name>& prefix)
{
_LOG_DEBUG("Sync prefix registration failed!");
}
void
SyncLogic::onSyncData(const shared_ptr<const ndn::Interest>& interest,
const shared_ptr<Data>& data,
const OnDataValidated& onValidated,
const OnDataValidationFailed& onValidationFailed)
{ m_validator->validate(data, onValidated, onValidationFailed); }
void
SyncLogic::onSyncTimeout(const shared_ptr<const ndn::Interest>& interest)
{
// It is OK. Others will handle the time out situation.
}
void
SyncLogic::onSyncDataValidationFailed(const shared_ptr<const Data>& data)
{
_LOG_DEBUG("Sync data cannot be verified!");
}
void
SyncLogic::onSyncDataValidated(const shared_ptr<const Data>& data)
{
Name name = data->getName();
const char* wireData = (const char*)data->getContent().value();
size_t len = data->getContent().value_size();
try
{
_LOG_TRACE ("<< D " << name);
DigestConstPtr digest;
string type;
tie (digest, type) = convertNameToDigestAndType (name);
if (type == "normal")
{
processSyncData (name, digest, wireData, len);
}
else
{
// timer is always restarted when we schedule recovery
m_scheduler.cancelEvent (m_reexpressingRecoveryInterestId);
processSyncData (name, digest, wireData, len);
}
}
catch (Error::DigestCalculationError &e)
{
_LOG_TRACE ("Something fishy happened...");
// log error. ignoring it for now, later we should log it
return;
}
}
void
SyncLogic::processSyncInterest (const Name &name, DigestConstPtr digest, bool timedProcessing/*=false*/)
{
_LOG_DEBUG("processSyncInterest");
DigestConstPtr rootDigest;
{
boost::recursive_mutex::scoped_lock lock (m_stateMutex);
rootDigest = m_state->getDigest();
}
// Special case when state is not empty and we have received request with zero-root digest
if (digest->isZero () && !rootDigest->isZero ())
{
SyncStateMsg ssm;
{
boost::recursive_mutex::scoped_lock lock (m_stateMutex);
ssm << (*m_state);
}
sendSyncData (name, digest, ssm);
return;
}
if (*rootDigest == *digest)
{
_LOG_TRACE ("processSyncInterest (): Same state. Adding to PIT");
m_syncInterestTable.insert (digest, name.toUri(), false);
return;
}
DiffStateContainer::iterator stateInDiffLog = m_log.find (digest);
if (stateInDiffLog != m_log.end ())
{
DiffStateConstPtr stateDiff = (*stateInDiffLog)->diff ();
sendSyncData (name, digest, stateDiff);
return;
}
if (!timedProcessing)
{
bool exists = m_syncInterestTable.insert (digest, name.toUri(), true);
if (exists) // somebody else replied, so restart random-game timer
{
_LOG_DEBUG ("Unknown digest, but somebody may have already replied, so restart our timer");
m_scheduler.cancelEvent (m_delayedInterestProcessingId);
}
uint32_t waitDelay = GET_RANDOM (m_rangeUniformRandom);
_LOG_DEBUG ("Digest is not in the log. Schedule processing after small delay: " << waitDelay << "ms");
m_delayedInterestProcessingId = m_scheduler.scheduleEvent (time::milliseconds (waitDelay),
bind (&SyncLogic::processSyncInterest, this, name, digest, true));
}
else
{
_LOG_TRACE (" (timed processing)");
m_recoveryRetransmissionInterval = m_defaultRecoveryRetransmitInterval;
sendSyncRecoveryInterests (digest);
}
}
void
SyncLogic::processSyncData (const Name &name, DigestConstPtr digest, const char *wireData, size_t len)
{
DiffStatePtr diffLog = boost::make_shared<DiffState> ();
bool ownInterestSatisfied = false;
try
{
m_syncInterestTable.remove (name.toUri()); // Remove satisfied interest from PIT
ownInterestSatisfied = (name == m_outstandingInterestName);
DiffState diff;
SyncStateMsg msg;
if (!msg.ParseFromArray(wireData, len) || !msg.IsInitialized())
{
//Throw
BOOST_THROW_EXCEPTION (Error::SyncStateMsgDecodingFailure () );
}
msg >> diff;
vector<MissingDataInfo> v;
BOOST_FOREACH (LeafConstPtr leaf, diff.getLeaves().get<ordered>())
{
DiffLeafConstPtr diffLeaf = boost::dynamic_pointer_cast<const DiffLeaf> (leaf);
BOOST_ASSERT (diffLeaf != 0);
NameInfoConstPtr info = diffLeaf->getInfo();
if (diffLeaf->getOperation() == UPDATE)
{
SeqNo seq = diffLeaf->getSeq();
bool inserted = false;
bool updated = false;
SeqNo oldSeq;
{
boost::recursive_mutex::scoped_lock lock (m_stateMutex);
boost::tie (inserted, updated, oldSeq) = m_state->update (info, seq);
}
if (inserted || updated)
{
diffLog->update (info, seq);
if (!oldSeq.isValid())
{
oldSeq = SeqNo(seq.getSession(), 0);
}
else
{
++oldSeq;
}
// there is no need for application to process update on forwarder node
if (info->toString() != forwarderPrefix)
{
MissingDataInfo mdi = {info->toString(), oldSeq, seq};
if (m_perBranch)
{
ostringstream interestName;
interestName << mdi.prefix << "/" << mdi.high.getSession() << "/" << mdi.high.getSeq();
m_onUpdateBranch(interestName.str());
}
else
{
v.push_back(mdi);
}
}
}
}
else if (diffLeaf->getOperation() == REMOVE)
{
boost::recursive_mutex::scoped_lock lock (m_stateMutex);
if (m_state->remove (info))
{
diffLog->remove (info);
if (!m_perBranch)
{
m_onRemove (info->toString ());
}
}
}
else
{
}
}
if (!v.empty())
{
if (!m_perBranch)
{
m_onUpdate(v);
}
}
insertToDiffLog (diffLog);
}
catch (Error::SyncStateMsgDecodingFailure &e)
{
_LOG_TRACE ("Something really fishy happened during state decoding " <<
diagnostic_information (e));
diffLog.reset ();
// don't do anything
}
if ((diffLog != 0 && diffLog->getLeaves ().size () > 0) ||
ownInterestSatisfied)
{
// Do it only if everything went fine and state changed
// this is kind of wrong
// satisfyPendingSyncInterests (diffLog); // if there are interests in PIT, there is a point to satisfy them using new state
// if state has changed, then it is safe to express a new interest
m_scheduler.cancelEvent (m_reexpressingInterestId);
m_reexpressingInterestId = m_scheduler.scheduleEvent (time::milliseconds(GET_RANDOM (m_reexpressionJitter)),
bind (&SyncLogic::sendSyncInterest, this));
}
}
void
SyncLogic::processSyncRecoveryInterest (const Name &name, DigestConstPtr digest)
{
_LOG_DEBUG("processSyncRecoveryInterest");
DiffStateContainer::iterator stateInDiffLog = m_log.find (digest);
if (stateInDiffLog == m_log.end ())
{
_LOG_TRACE ("Could not find " << *digest << " in digest log");
return;
}
SyncStateMsg ssm;
{
boost::recursive_mutex::scoped_lock lock (m_stateMutex);
ssm << (*m_state);
}
sendSyncData (name, digest, ssm);
}
void
SyncLogic::satisfyPendingSyncInterests (DiffStateConstPtr diffLog)
{
DiffStatePtr fullStateLog = boost::make_shared<DiffState> ();
{
boost::recursive_mutex::scoped_lock lock (m_stateMutex);
BOOST_FOREACH (LeafConstPtr leaf, m_state->getLeaves ()/*.get<timed> ()*/)
{
fullStateLog->update (leaf->getInfo (), leaf->getSeq ());
/// @todo Impose limit on how many state info should be send out
}
}
try
{
uint32_t counter = 0;
while (m_syncInterestTable.size () > 0)
{
Sync::Interest interest = m_syncInterestTable.pop ();
if (!interest.m_unknown)
{
_LOG_TRACE (">> D " << interest.m_name);
sendSyncData (interest.m_name, interest.m_digest, diffLog);
}
else
{
_LOG_TRACE (">> D (unknown)" << interest.m_name);
sendSyncData (interest.m_name, interest.m_digest, fullStateLog);
}
counter ++;
}
_LOG_DEBUG ("Satisfied " << counter << " pending interests");
}
catch (Error::InterestTableIsEmpty &e)
{
// ok. not really an error
}
}
void
SyncLogic::insertToDiffLog (DiffStatePtr diffLog)
{
diffLog->setDigest (m_state->getDigest());
if (m_log.size () > 0)
{
m_log.get<sequenced> ().front ()->setNext (diffLog);
}
m_log.erase (m_state->getDigest()); // remove diff state with the same digest. next pointers are still valid
/// @todo Optimization
m_log.get<sequenced> ().push_front (diffLog);
}
void
SyncLogic::addLocalNames (const Name &prefix, uint64_t session, uint64_t seq)
{
DiffStatePtr diff;
{
//cout << "Add local names" <<endl;
boost::recursive_mutex::scoped_lock lock (m_stateMutex);
NameInfoConstPtr info = StdNameInfo::FindOrCreate(prefix.toUri());
_LOG_DEBUG ("addLocalNames (): old state " << *m_state->getDigest ());
SeqNo seqN (session, seq);
m_state->update(info, seqN);
_LOG_DEBUG ("addLocalNames (): new state " << *m_state->getDigest ());
diff = boost::make_shared<DiffState>();
diff->update(info, seqN);
insertToDiffLog (diff);
}
// _LOG_DEBUG ("PIT size: " << m_syncInterestTable.size ());
satisfyPendingSyncInterests (diff);
}
void
SyncLogic::remove(const Name &prefix)
{
DiffStatePtr diff;
{
boost::recursive_mutex::scoped_lock lock (m_stateMutex);
NameInfoConstPtr info = StdNameInfo::FindOrCreate(prefix.toUri());
m_state->remove(info);
// increment the sequence number for the forwarder node
NameInfoConstPtr forwarderInfo = StdNameInfo::FindOrCreate(forwarderPrefix);
LeafContainer::iterator item = m_state->getLeaves ().find (forwarderInfo);
SeqNo seqNo (0);
if (item != m_state->getLeaves ().end ())
{
seqNo = (*item)->getSeq ();
++seqNo;
}
m_state->update (forwarderInfo, seqNo);
diff = boost::make_shared<DiffState>();
diff->remove(info);
diff->update(forwarderInfo, seqNo);
insertToDiffLog (diff);
}
satisfyPendingSyncInterests (diff);
}
void
SyncLogic::sendSyncInterest ()
{
_LOG_DEBUG("sendSyncInterest");
{
boost::recursive_mutex::scoped_lock lock (m_stateMutex);
m_outstandingInterestName = m_syncPrefix;
ostringstream os;
os << *m_state->getDigest();
m_outstandingInterestName.append(os.str());
_LOG_TRACE (">> I " << m_outstandingInterestName);
}
_LOG_DEBUG("sendSyncInterest: " << m_outstandingInterestName);
m_scheduler.cancelEvent (m_reexpressingInterestId);
m_reexpressingInterestId = m_scheduler.scheduleEvent (time::seconds(m_syncInterestReexpress)+time::milliseconds(GET_RANDOM (m_reexpressionJitter)),
bind (&SyncLogic::sendSyncInterest, this));
ndn::Interest interest(m_outstandingInterestName);
interest.setMustBeFresh(true);
OnDataValidated onValidated = bind(&SyncLogic::onSyncDataValidated, this, _1);
OnDataValidationFailed onValidationFailed = bind(&SyncLogic::onSyncDataValidationFailed, this, _1);
m_face->expressInterest(interest,
bind(&SyncLogic::onSyncData, this, _1, _2, onValidated, onValidationFailed),
bind(&SyncLogic::onSyncTimeout, this, _1));
}
void
SyncLogic::sendSyncRecoveryInterests (DigestConstPtr digest)
{
ostringstream os;
os << *digest;
Name interestName = m_syncPrefix;
interestName.append("recovery").append(os.str());
time::Duration nextRetransmission = time::milliseconds (m_recoveryRetransmissionInterval + GET_RANDOM (m_reexpressionJitter));
m_recoveryRetransmissionInterval <<= 1;
m_scheduler.cancelEvent (m_reexpressingRecoveryInterestId);
if (m_recoveryRetransmissionInterval < 100*1000) // <100 seconds
m_reexpressingRecoveryInterestId = m_scheduler.scheduleEvent (nextRetransmission,
bind (&SyncLogic::sendSyncRecoveryInterests, this, digest));
ndn::Interest interest(interestName);
interest.setMustBeFresh(true);
OnDataValidated onValidated = bind(&SyncLogic::onSyncDataValidated, this, _1);
OnDataValidationFailed onValidationFailed = bind(&SyncLogic::onSyncDataValidationFailed, this, _1);
m_face->expressInterest(interest,
bind(&SyncLogic::onSyncData, this, _1, _2, onValidated, onValidationFailed),
bind(&SyncLogic::onSyncTimeout, this, _1));
}
void
SyncLogic::sendSyncData (const Name &name, DigestConstPtr digest, StateConstPtr state)
{
SyncStateMsg msg;
msg << (*state);
sendSyncData(name, digest, msg);
}
// pass in state msg instead of state, so that there is no need to lock the state until
// this function returns
void
SyncLogic::sendSyncData (const Name &name, DigestConstPtr digest, SyncStateMsg &ssm)
{
_LOG_TRACE (">> D " << name);
int size = ssm.ByteSize();
char *wireData = new char[size];
ssm.SerializeToArray(wireData, size);
Data syncData(name);
syncData.setContent(reinterpret_cast<const uint8_t*>(wireData), size);
m_keyChain->sign(syncData);
m_face->put(syncData);
delete []wireData;
// checking if our own interest got satisfied
bool satisfiedOwnInterest = false;
{
boost::recursive_mutex::scoped_lock lock (m_stateMutex);
satisfiedOwnInterest = (m_outstandingInterestName == name);
}
if (satisfiedOwnInterest)
{
_LOG_TRACE ("Satisfied our own Interest. Re-expressing (hopefully with a new digest)");
m_scheduler.cancelEvent (m_reexpressingInterestId);
m_reexpressingInterestId = m_scheduler.scheduleEvent (time::milliseconds(GET_RANDOM (m_reexpressionJitter)),
bind (&SyncLogic::sendSyncInterest, this));
}
}
string
SyncLogic::getRootDigest()
{
ostringstream os;
boost::recursive_mutex::scoped_lock lock (m_stateMutex);
os << *m_state->getDigest();
return os.str();
}
size_t
SyncLogic::getNumberOfBranches () const
{
boost::recursive_mutex::scoped_lock lock (m_stateMutex);
return m_state->getLeaves ().size ();
}
void
SyncLogic::printState () const
{
boost::recursive_mutex::scoped_lock lock (m_stateMutex);
BOOST_FOREACH (const boost::shared_ptr<Sync::Leaf> leaf, m_state->getLeaves ())
{
std::cout << *leaf << std::endl;
}
}
std::map<std::string, bool>
SyncLogic::getBranchPrefixes() const
{
boost::recursive_mutex::scoped_lock lock (m_stateMutex);
std::map<std::string, bool> m;
BOOST_FOREACH (const boost::shared_ptr<Sync::Leaf> leaf, m_state->getLeaves ())
{
std::string prefix = leaf->getInfo()->toString();
// do not return forwarder prefix
if (prefix != forwarderPrefix)
{
m.insert(pair<std::string, bool>(prefix, false));
}
}
return m;
}
}