| /* -*- 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> |
| */ |
| |
| #include "sync-logic.h" |
| #include "sync-diff-leaf.h" |
| #include "sync-full-leaf.h" |
| #include "sync-log.h" |
| |
| #include <boost/make_shared.hpp> |
| #include <boost/foreach.hpp> |
| #include <boost/lexical_cast.hpp> |
| #include <boost/date_time/posix_time/posix_time.hpp> |
| #include <vector> |
| |
| using namespace std; |
| using namespace boost; |
| |
| // INIT_LOGGER ("SyncLogic"); |
| |
| namespace Sync |
| { |
| |
| SyncLogic::SyncLogic (const std::string &syncPrefix, |
| LogicUpdateCallback onUpdate, |
| LogicRemoveCallback onRemove) |
| : m_syncPrefix (syncPrefix) |
| , m_onUpdate (onUpdate) |
| , m_onRemove (onRemove) |
| , m_ccnxHandle(new CcnxWrapper()) |
| , m_randomGenerator (static_cast<unsigned int> (std::time (0))) |
| , m_rangeUniformRandom (m_randomGenerator, uniform_int<> (20,80)) |
| { |
| #ifdef _DEBUG |
| #ifdef HAVE_LOG4CXX |
| // _LOG_FUNCTION (syncPrefix); |
| static int id = 0; |
| staticModuleLogger = log4cxx::Logger::getLogger ("SyncLogic." + lexical_cast<string> (id)); |
| id ++; |
| #endif |
| #endif |
| |
| m_ccnxHandle->setInterestFilter (m_syncPrefix, |
| bind (&SyncLogic::respondSyncInterest, this, _1)); |
| |
| m_scheduler.schedule (posix_time::seconds (0), |
| bind (&SyncLogic::sendSyncInterest, this), |
| REEXPRESSING_INTEREST); |
| } |
| |
| SyncLogic::~SyncLogic () |
| { |
| // _LOG_FUNCTION (this); |
| // cout << "SyncLogic::~SyncLogic ()" << endl; |
| |
| m_ccnxHandle.reset (); |
| } |
| |
| void |
| SyncLogic::respondSyncInterest (const string &interest) |
| { |
| _LOG_TRACE ("<< I " << interest); |
| //cout << "Respond Sync Interest" << endl; |
| string hash = interest.substr(interest.find_last_of("/") + 1); |
| // cout << "Received Sync Interest: " << hash << endl; |
| |
| DigestPtr digest = make_shared<Digest> (); |
| try |
| { |
| istringstream is (hash); |
| is >> *digest; |
| } |
| catch (Error::DigestCalculationError &e) |
| { |
| // log error. ignoring it for now, later we should log it |
| return; |
| } |
| |
| processSyncInterest (digest, interest, false); |
| } |
| |
| void |
| SyncLogic::processSyncInterest (DigestConstPtr digest, const std::string &interestName, bool timedProcessing/*=false*/) |
| { |
| recursive_mutex::scoped_lock lock (m_stateMutex); |
| |
| // Special case when state is not empty and we have received request with zero-root digest |
| if (digest->isZero () && !m_state.getDigest()->isZero ()) |
| { |
| _LOG_TRACE (">> D " << interestName << "/state" << " (zero)"); |
| |
| m_syncInterestTable.remove (interestName + "/state"); |
| m_ccnxHandle->publishData (interestName + "/state", |
| lexical_cast<string> (m_state), |
| m_syncResponseFreshness); |
| if (m_outstandingInterest == interestName) |
| { |
| m_scheduler.schedule (posix_time::seconds (0), |
| bind (&SyncLogic::sendSyncInterest, this), |
| REEXPRESSING_INTEREST); |
| } |
| return; |
| } |
| |
| if (*m_state.getDigest() == *digest) |
| { |
| // cout << interestName << "\n"; |
| if (digest->isZero ()) |
| { |
| _LOG_TRACE ("processSyncInterest (): Digest is zero, adding /state to PIT"); |
| m_syncInterestTable.insert (interestName + "/state"); |
| } |
| else |
| { |
| _LOG_TRACE ("processSyncInterest (): Same state. Adding to PIT"); |
| m_syncInterestTable.insert (interestName); |
| } |
| return; |
| } |
| |
| DiffStateContainer::iterator stateInDiffLog = m_log.find (digest); |
| |
| if (stateInDiffLog != m_log.end ()) |
| { |
| _LOG_TRACE (">> D " << interestName); |
| |
| m_syncInterestTable.remove (interestName); |
| m_ccnxHandle->publishData (interestName, |
| lexical_cast<string> (*(*stateInDiffLog)->diff ()), |
| m_syncResponseFreshness); |
| if (m_outstandingInterest == interestName) |
| { |
| m_scheduler.schedule (posix_time::seconds (0), |
| bind (&SyncLogic::sendSyncInterest, this), |
| REEXPRESSING_INTEREST); |
| } |
| return; |
| } |
| |
| if (!timedProcessing) |
| { |
| m_scheduler.schedule (posix_time::milliseconds (m_rangeUniformRandom ()) /*from 20 to 100ms*/, |
| bind (&SyncLogic::processSyncInterest, this, digest, interestName, true), |
| DELAYED_INTEREST_PROCESSING); |
| |
| } |
| else |
| { |
| _LOG_TRACE (">> D " << interestName << "/state" << " (timed processing)"); |
| |
| m_syncInterestTable.remove (interestName + "/state"); |
| m_ccnxHandle->publishData (interestName + "/state", |
| lexical_cast<string> (m_state), |
| m_syncResponseFreshness); |
| |
| if (m_outstandingInterest == interestName) |
| { |
| m_scheduler.schedule (posix_time::seconds (0), |
| bind (&SyncLogic::sendSyncInterest, this), |
| REEXPRESSING_INTEREST); |
| } |
| } |
| } |
| |
| void |
| SyncLogic::processSyncData (const string &name, const string &dataBuffer) |
| { |
| _LOG_TRACE ("<< D " << name); |
| DiffStatePtr diffLog = make_shared<DiffState> (); |
| |
| try |
| { |
| recursive_mutex::scoped_lock lock (m_stateMutex); |
| |
| string last = name.substr(name.find_last_of("/") + 1); |
| istringstream ss (dataBuffer); |
| |
| if (last == "state") |
| { |
| FullState full; |
| ss >> full; |
| BOOST_FOREACH (LeafConstPtr leaf, full.getLeaves()) // order doesn't matter |
| { |
| NameInfoConstPtr info = leaf->getInfo (); |
| SeqNo seq = leaf->getSeq (); |
| |
| bool inserted = false; |
| bool updated = false; |
| SeqNo oldSeq; |
| tie (inserted, updated, oldSeq) = m_state.update (info, seq); |
| |
| if (inserted || updated) |
| { |
| diffLog->update (info, seq); |
| m_onUpdate (info->toString (), seq.getSeq(), oldSeq); |
| } |
| } |
| } |
| else |
| { |
| DiffState diff; |
| ss >> diff; |
| BOOST_FOREACH (LeafConstPtr leaf, diff.getLeaves().get<ordered>()) |
| { |
| DiffLeafConstPtr diffLeaf = 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; |
| tie (inserted, updated, oldSeq) = m_state.update (info, seq); |
| |
| if (inserted || updated) |
| { |
| diffLog->update (info, seq); |
| m_onUpdate (info->toString (), seq.getSeq(), oldSeq); |
| } |
| } |
| else if (diffLeaf->getOperation() == REMOVE) |
| { |
| if (m_state.remove (info)) |
| { |
| diffLog->remove (info); |
| m_onRemove (info->toString ()); |
| } |
| } |
| else |
| { |
| BOOST_ASSERT (false); // just in case |
| } |
| } |
| } |
| |
| diffLog->setDigest(m_state.getDigest()); |
| if (m_log.size () > 0) |
| { |
| m_log.get<sequenced> ().front ()->setNext (diffLog); |
| } |
| m_log.erase (m_state.getDigest()); |
| /// @todo Optimization |
| m_log.insert (diffLog); |
| } |
| catch (Error::SyncXmlDecodingFailure &e) |
| { |
| diffLog.reset (); |
| // don't do anything |
| } |
| |
| // if state has changed, then it is safe to express a new interest |
| if (diffLog->getLeaves ().size () > 0) |
| { |
| m_scheduler.schedule (posix_time::seconds (0), |
| bind (&SyncLogic::sendSyncInterest, this), |
| REEXPRESSING_INTEREST); |
| } |
| else |
| { |
| // should not reexpress the same interest. Need at least wait for data lifetime |
| // Otherwise we will get immediate reply from the local daemon and there will be 100% utilization |
| m_scheduler.cancel (REEXPRESSING_INTEREST); |
| // m_scheduler.schedule (posix_time::seconds (0), |
| m_scheduler.schedule (posix_time::seconds (m_syncResponseFreshness) + posix_time::milliseconds (1), |
| bind (&SyncLogic::sendSyncInterest, this), |
| REEXPRESSING_INTEREST); |
| } |
| } |
| |
| void |
| SyncLogic::satisfyPendingSyncInterests (DiffStatePtr diffLog) |
| { |
| vector<string> pis = m_syncInterestTable.fetchAll (); |
| if (pis.size () > 0) |
| { |
| stringstream ss; |
| ss << *diffLog; |
| bool satisfiedOwnInterest = false; |
| |
| for (vector<string>::iterator ii = pis.begin(); ii != pis.end(); ++ii) |
| { |
| _LOG_TRACE (">> D " << *ii); |
| m_ccnxHandle->publishData (*ii, ss.str(), m_syncResponseFreshness); |
| |
| { |
| recursive_mutex::scoped_lock lock (m_stateMutex); |
| // _LOG_DEBUG (*ii << " == " << m_outstandingInterest << " = " << (*ii == m_outstandingInterest)); |
| satisfiedOwnInterest = satisfiedOwnInterest || (*ii == m_outstandingInterest) || (*ii == (m_outstandingInterest + "/state")); |
| } |
| } |
| |
| if (satisfiedOwnInterest) |
| { |
| _LOG_DEBUG ("Have satisfied our own interest. Scheduling interest reexpression"); |
| // we need to reexpress interest only if we satisfied our own interest |
| m_scheduler.schedule (posix_time::milliseconds (0), |
| bind (&SyncLogic::sendSyncInterest, this), |
| REEXPRESSING_INTEREST); |
| } |
| } |
| } |
| |
| void |
| SyncLogic::processPendingSyncInterests (DiffStatePtr diffLog) |
| { |
| //cout << "Process Pending Interests" <<endl; |
| 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); |
| // _LOG_DEBUG (*diffLog->getDigest () << " " << m_log.size ()); |
| } |
| |
| void |
| SyncLogic::addLocalNames (const string &prefix, uint32_t session, uint32_t seq) |
| { |
| DiffStatePtr diff; |
| { |
| //cout << "Add local names" <<endl; |
| recursive_mutex::scoped_lock lock (m_stateMutex); |
| NameInfoConstPtr info = StdNameInfo::FindOrCreate(prefix); |
| |
| SeqNo seqN (session, seq); |
| m_state.update(info, seqN); |
| |
| _LOG_DEBUG ("addLocalNames (): new state " << *m_state.getDigest ()); |
| |
| diff = make_shared<DiffState>(); |
| diff->update(info, seqN); |
| processPendingSyncInterests (diff); |
| } |
| |
| // _LOG_DEBUG ("PIT size: " << m_syncInterestTable.size ()); |
| satisfyPendingSyncInterests (diff); |
| } |
| |
| void |
| SyncLogic::remove(const string &prefix) |
| { |
| DiffStatePtr diff; |
| { |
| recursive_mutex::scoped_lock lock (m_stateMutex); |
| NameInfoConstPtr info = StdNameInfo::FindOrCreate(prefix); |
| m_state.remove(info); |
| |
| diff = make_shared<DiffState>(); |
| diff->remove(info); |
| |
| processPendingSyncInterests (diff); |
| } |
| |
| satisfyPendingSyncInterests (diff); |
| } |
| |
| void |
| SyncLogic::sendSyncInterest () |
| { |
| ostringstream os; |
| |
| { |
| // cout << "Sending Sync Interest" << endl; |
| recursive_mutex::scoped_lock lock (m_stateMutex); |
| |
| os << m_syncPrefix << "/" << *m_state.getDigest(); |
| |
| _LOG_TRACE (">> I " << os.str ()); |
| |
| m_outstandingInterest = os.str (); |
| } |
| |
| m_ccnxHandle->sendInterest (os.str (), |
| bind (&SyncLogic::processSyncData, this, _1, _2)); |
| |
| m_scheduler.cancel (REEXPRESSING_INTEREST); |
| m_scheduler.schedule (posix_time::seconds (4), |
| bind (&SyncLogic::sendSyncInterest, this), |
| REEXPRESSING_INTEREST); |
| } |
| |
| } |