blob: ef96a866faa5d6ec6dc6c3db8b7fa1f5299f139a [file] [log] [blame]
Zhenkai Zhu8d935c82012-03-06 10:44:12 -08001/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
2/*
3 * Copyright (c) 2012 University of California, Los Angeles
4 *
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License version 2 as
7 * published by the Free Software Foundation;
8 *
9 * This program is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 * GNU General Public License for more details.
13 *
14 * You should have received a copy of the GNU General Public License
15 * along with this program; if not, write to the Free Software
16 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17 *
18 * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
Chaoyi Bian3e1eb162012-04-03 16:59:32 -070019 * Chaoyi Bian <bcy@pku.edu.cn>
Zhenkai Zhu8d935c82012-03-06 10:44:12 -080020 * Alexander Afanasyev <alexander.afanasyev@ucla.edu>
21 */
Chaoyi Bian44fff0c2012-03-07 21:07:22 -080022
Alexander Afanasyevb6cf56e2012-05-29 19:33:49 -070023#ifdef NS3_MODULE
Alexander Afanasyev0ac399e2012-04-27 13:28:28 -070024#include <ns3/ccnx-pit.h>
25#include <ns3/ccnx.h>
Alexander Afanasyevb6cf56e2012-05-29 19:33:49 -070026#endif
Alexander Afanasyev0ac399e2012-04-27 13:28:28 -070027
Chaoyi Bian11f294f2012-03-08 14:28:06 -080028#include "sync-logic.h"
Chaoyi Bian89ee2dc2012-03-09 14:06:01 -080029#include "sync-diff-leaf.h"
30#include "sync-full-leaf.h"
Alexander Afanasyevce001692013-07-14 11:34:41 -070031#include "sync-logging.h"
Zhenkai Zhu3cfdcb92012-06-06 15:20:10 -070032#include "sync-state.h"
Alexander Afanasyev4f9ea482012-03-15 11:57:29 -070033
Chaoyi Bian89ee2dc2012-03-09 14:06:01 -080034#include <boost/make_shared.hpp>
35#include <boost/foreach.hpp>
Alexander Afanasyev387ac952012-03-11 23:49:27 -070036#include <boost/lexical_cast.hpp>
Zhenkai Zhua5d06d72012-03-09 15:16:24 -080037#include <vector>
Chaoyi Bian44fff0c2012-03-07 21:07:22 -080038
39using namespace std;
40using namespace boost;
Yingdi Yu43e71612013-10-30 22:19:31 -070041using namespace ndn;
Chaoyi Bian44fff0c2012-03-07 21:07:22 -080042
Chaoyi Bianf007f922012-04-09 20:05:37 -070043INIT_LOGGER ("SyncLogic");
Alexander Afanasyev4f9ea482012-03-15 11:57:29 -070044
Alexander Afanasyev983b0e92012-04-26 12:44:18 -070045#ifdef NS3_MODULE
46#define GET_RANDOM(var) var.GetValue ()
47#else
48#define GET_RANDOM(var) var ()
49#endif
50
51#define TIME_SECONDS_WITH_JITTER(sec) \
52 (TIME_SECONDS (sec) + TIME_MILLISECONDS (GET_RANDOM (m_reexpressionJitter)))
53
Alexander Afanasyev42612262012-05-05 10:58:37 -070054#define TIME_MILLISECONDS_WITH_JITTER(ms) \
55 (TIME_MILLISECONDS (ms) + TIME_MILLISECONDS (GET_RANDOM (m_reexpressionJitter)))
Alexander Afanasyev983b0e92012-04-26 12:44:18 -070056
57
Chaoyi Bian44fff0c2012-03-07 21:07:22 -080058namespace Sync
59{
60
Yingdi Yu43e71612013-10-30 22:19:31 -070061 SyncLogic::SyncLogic (const Name& syncPrefix,
62 Ptr<SyncPolicyManager> syncPolicyManager,
63 LogicUpdateCallback onUpdate,
64 LogicRemoveCallback onRemove)
65 : m_state (new FullState)
66 , m_syncInterestTable (TIME_SECONDS (m_syncInterestReexpress))
67 , m_syncPrefix (syncPrefix)
68 , m_onUpdate (onUpdate)
69 , m_onRemove (onRemove)
70 , m_perBranch (false)
71 , m_policyManager(syncPolicyManager)
Alexander Afanasyev73f7f9a2012-04-09 15:45:47 -070072#ifndef NS3_MODULE
Yingdi Yu43e71612013-10-30 22:19:31 -070073 , m_randomGenerator (static_cast<unsigned int> (std::time (0)))
74 , m_rangeUniformRandom (m_randomGenerator, uniform_int<> (200,1000))
75 , m_reexpressionJitter (m_randomGenerator, uniform_int<> (100,500))
Alexander Afanasyev73f7f9a2012-04-09 15:45:47 -070076#else
Yingdi Yu43e71612013-10-30 22:19:31 -070077 , m_rangeUniformRandom (200,1000)
78 , m_reexpressionJitter (10,500)
Alexander Afanasyev860e6fe2012-03-15 17:30:31 -070079#endif
Yingdi Yu43e71612013-10-30 22:19:31 -070080 , m_recoveryRetransmissionInterval (m_defaultRecoveryRetransmitInterval)
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -070081{
Alexander Afanasyevb550d6a2012-04-09 15:03:16 -070082#ifndef NS3_MODULE
83 // In NS3 module these functions are moved to StartApplication method
Yingdi Yu43e71612013-10-30 22:19:31 -070084
85 Ptr<security::Keychain> keychain = Ptr<security::Keychain>(new security::Keychain(Ptr<security::IdentityManager>::Create(),
86 m_policyManager,
87 NULL));
88 m_handler = Ptr<Wrapper>(new Wrapper(keychain));
89 m_handler->setInterestFilter (m_syncPrefix,
90 bind (&SyncLogic::respondSyncInterest, this, _1));
Alexander Afanasyevbf2b4362012-03-12 23:55:09 -070091
Alexander Afanasyev983b0e92012-04-26 12:44:18 -070092 m_scheduler.schedule (TIME_SECONDS (0), // no need to add jitter
Alexander Afanasyev4f9ea482012-03-15 11:57:29 -070093 bind (&SyncLogic::sendSyncInterest, this),
94 REEXPRESSING_INTEREST);
Alexander Afanasyevb550d6a2012-04-09 15:03:16 -070095#endif
Chaoyi Bian44fff0c2012-03-07 21:07:22 -080096}
97
Yingdi Yu43e71612013-10-30 22:19:31 -070098SyncLogic::SyncLogic (const Name& syncPrefix,
99 Ptr<SyncPolicyManager> syncPolicyManager,
Zhenkai Zhub9f19592012-10-16 14:27:38 -0700100 LogicPerBranchCallback onUpdateBranch)
101 : m_state (new FullState)
102 , m_syncInterestTable (TIME_SECONDS (m_syncInterestReexpress))
103 , m_syncPrefix (syncPrefix)
104 , m_onUpdateBranch (onUpdateBranch)
105 , m_perBranch(true)
Yingdi Yu43e71612013-10-30 22:19:31 -0700106 , m_policyManager(syncPolicyManager)
Zhenkai Zhub9f19592012-10-16 14:27:38 -0700107#ifndef NS3_MODULE
108 , m_randomGenerator (static_cast<unsigned int> (std::time (0)))
109 , m_rangeUniformRandom (m_randomGenerator, uniform_int<> (200,1000))
110 , m_reexpressionJitter (m_randomGenerator, uniform_int<> (100,500))
111#else
112 , m_rangeUniformRandom (200,1000)
113 , m_reexpressionJitter (10,500)
114#endif
Alexander Afanasyevce001692013-07-14 11:34:41 -0700115 , m_recoveryRetransmissionInterval (m_defaultRecoveryRetransmitInterval)
Zhenkai Zhub9f19592012-10-16 14:27:38 -0700116{
117#ifndef NS3_MODULE
118 // In NS3 module these functions are moved to StartApplication method
Yingdi Yu43e71612013-10-30 22:19:31 -0700119
120 Ptr<security::Keychain> keychain = Ptr<security::Keychain>(new security::Keychain(Ptr<security::IdentityManager>::Create(),
121 m_policyManager,
122 NULL));
123 m_handler = Ptr<Wrapper>(new Wrapper(keychain));
Zhenkai Zhub9f19592012-10-16 14:27:38 -0700124
Yingdi Yu43e71612013-10-30 22:19:31 -0700125 m_handler->setInterestFilter (m_syncPrefix,
Zhenkai Zhub9f19592012-10-16 14:27:38 -0700126 bind (&SyncLogic::respondSyncInterest, this, _1));
127
128 m_scheduler.schedule (TIME_SECONDS (0), // no need to add jitter
129 bind (&SyncLogic::sendSyncInterest, this),
130 REEXPRESSING_INTEREST);
131#endif
132}
133
Alexander Afanasyevc1030192012-03-08 22:21:28 -0800134SyncLogic::~SyncLogic ()
Chaoyi Bian44fff0c2012-03-07 21:07:22 -0800135{
Yingdi Yu43e71612013-10-30 22:19:31 -0700136 m_handler->clearInterestFilter (Name(m_syncPrefix));
Alexander Afanasyev1b449c42012-03-13 20:24:07 -0700137}
Alexander Afanasyev387ac952012-03-11 23:49:27 -0700138
Alexander Afanasyevb550d6a2012-04-09 15:03:16 -0700139#ifdef NS3_MODULE
Alexander Afanasyev387ac952012-03-11 23:49:27 -0700140void
Alexander Afanasyevb550d6a2012-04-09 15:03:16 -0700141SyncLogic::StartApplication ()
Alexander Afanasyev387ac952012-03-11 23:49:27 -0700142{
Alexander Afanasyevb550d6a2012-04-09 15:03:16 -0700143 m_ccnxHandle->SetNode (GetNode ());
144 m_ccnxHandle->StartApplication ();
Alexander Afanasyev387ac952012-03-11 23:49:27 -0700145
Alexander Afanasyevb550d6a2012-04-09 15:03:16 -0700146 m_ccnxHandle->setInterestFilter (m_syncPrefix,
147 bind (&SyncLogic::respondSyncInterest, this, _1));
148
Alexander Afanasyev983b0e92012-04-26 12:44:18 -0700149 m_scheduler.schedule (TIME_SECONDS (0), // need to send first interests at exactly the same time
Alexander Afanasyevb550d6a2012-04-09 15:03:16 -0700150 bind (&SyncLogic::sendSyncInterest, this),
151 REEXPRESSING_INTEREST);
Alexander Afanasyev387ac952012-03-11 23:49:27 -0700152}
153
154void
Alexander Afanasyevb550d6a2012-04-09 15:03:16 -0700155SyncLogic::StopApplication ()
156{
Alexander Afanasyev434f1612012-04-21 21:19:15 -0700157 m_ccnxHandle->clearInterestFilter (m_syncPrefix);
Alexander Afanasyevb550d6a2012-04-09 15:03:16 -0700158 m_ccnxHandle->StopApplication ();
Alexander Afanasyev40942f42012-04-21 20:53:16 -0700159 m_scheduler.cancel (REEXPRESSING_INTEREST);
160 m_scheduler.cancel (DELAYED_INTEREST_PROCESSING);
Alexander Afanasyevb550d6a2012-04-09 15:03:16 -0700161}
162#endif
163
Zhenkai Zhueb1d8652012-12-23 15:25:39 -0800164void
165SyncLogic::stop()
166{
Yingdi Yu43e71612013-10-30 22:19:31 -0700167 m_handler->clearInterestFilter (Name(m_syncPrefix));
Zhenkai Zhueb1d8652012-12-23 15:25:39 -0800168 m_scheduler.cancel (REEXPRESSING_INTEREST);
169 m_scheduler.cancel (DELAYED_INTEREST_PROCESSING);
170}
171
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700172/**
173 * Two types of intersts
174 *
175 * Normal name: .../<hash>
176 * Recovery name: .../recovery/<hash>
177 */
178boost::tuple<DigestConstPtr, std::string>
179SyncLogic::convertNameToDigestAndType (const std::string &name)
Alexander Afanasyev387ac952012-03-11 23:49:27 -0700180{
Yingdi Yu43e71612013-10-30 22:19:31 -0700181 BOOST_ASSERT (name.find (m_syncPrefix.toUri()) == 0);
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700182
Yingdi Yu43e71612013-10-30 22:19:31 -0700183 string hash = name.substr (m_syncPrefix.toUri().size (), name.size ()-m_syncPrefix.toUri().size ());
Alexander Afanasyev282a10b2012-05-09 12:56:38 -0700184 if (hash[0] == '/')
185 hash = hash.substr (1, hash.size ()-1);
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700186 string interestType = "normal";
Alexander Afanasyev282a10b2012-05-09 12:56:38 -0700187
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700188 size_t pos = hash.find ('/');
189 if (pos != string::npos)
Alexander Afanasyev387ac952012-03-11 23:49:27 -0700190 {
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700191 interestType = hash.substr (0, pos);
192 hash = hash.substr (pos + 1);
Alexander Afanasyev387ac952012-03-11 23:49:27 -0700193 }
194
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700195 _LOG_TRACE (hash << ", " << interestType);
196
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700197 DigestPtr digest = make_shared<Digest> ();
198 istringstream is (hash);
199 is >> *digest;
200
201 return make_tuple (digest, interestType);
Alexander Afanasyev387ac952012-03-11 23:49:27 -0700202}
203
204void
Yingdi Yu43e71612013-10-30 22:19:31 -0700205SyncLogic::respondSyncInterest (ndn::Ptr<ndn::Interest> interest)
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700206{
Yingdi Yu43e71612013-10-30 22:19:31 -0700207 _LOG_DEBUG("respondSyncInterest: " << interest->getName().toUri());
208 string name = interest->getName().toUri();
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700209 try
210 {
211 _LOG_TRACE ("<< I " << name);
212
213 DigestConstPtr digest;
214 string type;
215 tie (digest, type) = convertNameToDigestAndType (name);
216
217 if (type == "normal") // kind of ineffective...
218 {
219 processSyncInterest (name, digest);
220 }
221 else if (type == "recovery")
222 {
223 processSyncRecoveryInterest (name, digest);
224 }
225 }
226 catch (Error::DigestCalculationError &e)
227 {
228 _LOG_TRACE ("Something fishy happened...");
229 // log error. ignoring it for now, later we should log it
230 return ;
231 }
232}
233
234void
Yingdi Yu43e71612013-10-30 22:19:31 -0700235SyncLogic::respondSyncData (Ptr<Data> data)
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700236{
Yingdi Yu43e71612013-10-30 22:19:31 -0700237 string name = data->getName().toUri();
238 const char *wireData = data->content().buf();
239 size_t len = data->content().size();
240
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700241 try
242 {
243 _LOG_TRACE ("<< D " << name);
244
245 DigestConstPtr digest;
246 string type;
247 tie (digest, type) = convertNameToDigestAndType (name);
248
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700249 if (type == "normal")
250 {
Zhenkai Zhu97e36bd2012-06-06 13:55:03 -0700251 processSyncData (name, digest, wireData, len);
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700252 }
253 else
254 {
Alexander Afanasyeva76010b2012-05-24 21:31:49 -0700255 // timer is always restarted when we schedule recovery
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700256 m_scheduler.cancel (REEXPRESSING_RECOVERY_INTEREST);
Zhenkai Zhu97e36bd2012-06-06 13:55:03 -0700257 processSyncData (name, digest, wireData, len);
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700258 }
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700259 }
260 catch (Error::DigestCalculationError &e)
261 {
262 _LOG_TRACE ("Something fishy happened...");
263 // log error. ignoring it for now, later we should log it
264 return;
265 }
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700266}
267
Yingdi Yu43e71612013-10-30 22:19:31 -0700268void
269SyncLogic::onSyncDataTimeout(Ptr<Closure> closure, Ptr<ndn::Interest> interest, int retry)
270{
271 if(retry > 0)
272 {
273 Ptr<Closure> newClosure = Ptr<Closure>(new Closure(closure->m_dataCallback,
274 boost::bind(&SyncLogic::onSyncDataTimeout,
275 this,
276 _1,
277 _2,
278 retry - 1),
279 closure->m_unverifiedCallback,
280 closure->m_stepCount)
281 );
282 m_handler->sendInterest(interest, newClosure);
283 }
284}
285
286void
287SyncLogic::onSyncDataUnverified()
288{}
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700289
290void
291SyncLogic::processSyncInterest (const std::string &name, DigestConstPtr digest, bool timedProcessing/*=false*/)
Alexander Afanasyev387ac952012-03-11 23:49:27 -0700292{
Yingdi Yu43e71612013-10-30 22:19:31 -0700293 _LOG_DEBUG("processSyncInterest");
Zhenkai Zhu2d3e2702012-10-15 14:18:05 -0700294 DigestConstPtr rootDigest;
295 {
296 recursive_mutex::scoped_lock lock (m_stateMutex);
297 rootDigest = m_state->getDigest();
298 }
Alexander Afanasyev387ac952012-03-11 23:49:27 -0700299
Alexander Afanasyevdd0eba72012-03-13 13:57:10 -0700300 // Special case when state is not empty and we have received request with zero-root digest
Zhenkai Zhu2d3e2702012-10-15 14:18:05 -0700301 if (digest->isZero () && !rootDigest->isZero ())
Alexander Afanasyevdd0eba72012-03-13 13:57:10 -0700302 {
Zhenkai Zhu2d3e2702012-10-15 14:18:05 -0700303
304 SyncStateMsg ssm;
305 {
306 recursive_mutex::scoped_lock lock (m_stateMutex);
307 ssm << (*m_state);
308 }
309 sendSyncData (name, digest, ssm);
Alexander Afanasyevdd0eba72012-03-13 13:57:10 -0700310 return;
311 }
Alexander Afanasyev763855a2012-03-13 14:17:37 -0700312
Zhenkai Zhu2d3e2702012-10-15 14:18:05 -0700313 if (*rootDigest == *digest)
Alexander Afanasyev4f9ea482012-03-15 11:57:29 -0700314 {
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700315 _LOG_TRACE ("processSyncInterest (): Same state. Adding to PIT");
Alexander Afanasyev2b6fbfb2012-05-24 10:57:14 -0700316 m_syncInterestTable.insert (digest, name, false);
Alexander Afanasyev4f9ea482012-03-15 11:57:29 -0700317 return;
318 }
Alexander Afanasyevdd0eba72012-03-13 13:57:10 -0700319
Alexander Afanasyev387ac952012-03-11 23:49:27 -0700320 DiffStateContainer::iterator stateInDiffLog = m_log.find (digest);
321
322 if (stateInDiffLog != m_log.end ())
323 {
Alexander Afanasyev235c6d72012-03-15 22:28:43 -0700324 DiffStateConstPtr stateDiff = (*stateInDiffLog)->diff ();
Alexander Afanasyev235c6d72012-03-15 22:28:43 -0700325
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700326 sendSyncData (name, digest, stateDiff);
Alexander Afanasyev387ac952012-03-11 23:49:27 -0700327 return;
328 }
329
330 if (!timedProcessing)
331 {
Alexander Afanasyev2b6fbfb2012-05-24 10:57:14 -0700332 bool exists = m_syncInterestTable.insert (digest, name, true);
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700333 if (exists) // somebody else replied, so restart random-game timer
Alexander Afanasyev3a229132012-04-25 15:07:26 -0700334 {
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700335 _LOG_DEBUG ("Unknown digest, but somebody may have already replied, so restart our timer");
336 m_scheduler.cancel (DELAYED_INTEREST_PROCESSING);
Alexander Afanasyev3a229132012-04-25 15:07:26 -0700337 }
Alexander Afanasyev03793b42012-05-01 13:03:07 -0700338
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700339 uint32_t waitDelay = GET_RANDOM (m_rangeUniformRandom);
340 _LOG_DEBUG ("Digest is not in the log. Schedule processing after small delay: " << waitDelay << "ms");
Alexander Afanasyev983b0e92012-04-26 12:44:18 -0700341
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700342 m_scheduler.schedule (TIME_MILLISECONDS (waitDelay),
343 bind (&SyncLogic::processSyncInterest, this, name, digest, true),
Alexander Afanasyevbf2b4362012-03-12 23:55:09 -0700344 DELAYED_INTEREST_PROCESSING);
Alexander Afanasyev387ac952012-03-11 23:49:27 -0700345 }
346 else
347 {
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700348 _LOG_TRACE (" (timed processing)");
Alexander Afanasyeva76010b2012-05-24 21:31:49 -0700349
350 m_recoveryRetransmissionInterval = m_defaultRecoveryRetransmitInterval;
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700351 sendSyncRecoveryInterests (digest);
Alexander Afanasyev387ac952012-03-11 23:49:27 -0700352 }
Chaoyi Bian44fff0c2012-03-07 21:07:22 -0800353}
354
Alexander Afanasyevc1030192012-03-08 22:21:28 -0800355void
Zhenkai Zhu97e36bd2012-06-06 13:55:03 -0700356SyncLogic::processSyncData (const std::string &name, DigestConstPtr digest, const char *wireData, size_t len)
Chaoyi Bian44fff0c2012-03-07 21:07:22 -0800357{
Alexander Afanasyev750d1872012-03-12 15:33:56 -0700358 DiffStatePtr diffLog = make_shared<DiffState> ();
Alexander Afanasyev2b6fbfb2012-05-24 10:57:14 -0700359 bool ownInterestSatisfied = false;
Alexander Afanasyeve4e2bf72012-03-12 12:44:54 -0700360
Alexander Afanasyev750d1872012-03-12 15:33:56 -0700361 try
362 {
Alexander Afanasyev750d1872012-03-12 15:33:56 -0700363
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700364 m_syncInterestTable.remove (name); // Remove satisfied interest from PIT
Alexander Afanasyev750d1872012-03-12 15:33:56 -0700365
Alexander Afanasyev2b6fbfb2012-05-24 10:57:14 -0700366 ownInterestSatisfied = (name == m_outstandingInterestName);
367
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700368 DiffState diff;
Zhenkai Zhu97e36bd2012-06-06 13:55:03 -0700369 SyncStateMsg msg;
Zhenkai Zhu3cfdcb92012-06-06 15:20:10 -0700370 if (!msg.ParseFromArray(wireData, len) || !msg.IsInitialized())
Zhenkai Zhu97e36bd2012-06-06 13:55:03 -0700371 {
372 //Throw
Zhenkai Zhu3cfdcb92012-06-06 15:20:10 -0700373 BOOST_THROW_EXCEPTION (Error::SyncStateMsgDecodingFailure () );
Zhenkai Zhu97e36bd2012-06-06 13:55:03 -0700374 }
375 msg >> diff;
376
Zhenkai Zhu1cb29292012-05-31 22:54:34 -0700377 vector<MissingDataInfo> v;
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700378 BOOST_FOREACH (LeafConstPtr leaf, diff.getLeaves().get<ordered>())
Alexander Afanasyev04fd8a82012-03-12 15:40:48 -0700379 {
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700380 DiffLeafConstPtr diffLeaf = dynamic_pointer_cast<const DiffLeaf> (leaf);
381 BOOST_ASSERT (diffLeaf != 0);
382
383 NameInfoConstPtr info = diffLeaf->getInfo();
Zhenkai Zhuba3c1d12012-10-03 23:26:35 -0700384 if (diffLeaf->getOperation() == UPDATE)
Alexander Afanasyev04fd8a82012-03-12 15:40:48 -0700385 {
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700386 SeqNo seq = diffLeaf->getSeq();
Alexander Afanasyev750d1872012-03-12 15:33:56 -0700387
388 bool inserted = false;
389 bool updated = false;
390 SeqNo oldSeq;
Zhenkai Zhu2d3e2702012-10-15 14:18:05 -0700391 {
392 recursive_mutex::scoped_lock lock (m_stateMutex);
393 tie (inserted, updated, oldSeq) = m_state->update (info, seq);
394 }
Chaoyi Bian89ee2dc2012-03-09 14:06:01 -0800395
Alexander Afanasyevd4adce52012-03-13 13:59:39 -0700396 if (inserted || updated)
Alexander Afanasyev750d1872012-03-12 15:33:56 -0700397 {
398 diffLog->update (info, seq);
Zhenkai Zhu68f04d52012-06-05 14:07:41 -0700399 if (!oldSeq.isValid())
400 {
401 oldSeq = SeqNo(seq.getSession(), 0);
402 }
403 else
404 {
405 ++oldSeq;
406 }
Zhenkai Zhuba3c1d12012-10-03 23:26:35 -0700407 // there is no need for application to process update on forwarder node
408 if (info->toString() != forwarderPrefix)
409 {
410 MissingDataInfo mdi = {info->toString(), oldSeq, seq};
Zhenkai Zhub9f19592012-10-16 14:27:38 -0700411 if (m_perBranch)
412 {
413 ostringstream interestName;
414 interestName << mdi.prefix << "/" << mdi.high.getSession() << "/" << mdi.high.getSeq();
415 m_onUpdateBranch(interestName.str());
416 }
417 else
418 {
419 v.push_back(mdi);
420 }
Zhenkai Zhuba3c1d12012-10-03 23:26:35 -0700421 }
Alexander Afanasyev04fd8a82012-03-12 15:40:48 -0700422 }
423 }
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700424 else if (diffLeaf->getOperation() == REMOVE)
Alexander Afanasyev04fd8a82012-03-12 15:40:48 -0700425 {
Zhenkai Zhu2d3e2702012-10-15 14:18:05 -0700426 recursive_mutex::scoped_lock lock (m_stateMutex);
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700427 if (m_state->remove (info))
Alexander Afanasyev750d1872012-03-12 15:33:56 -0700428 {
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700429 diffLog->remove (info);
Zhenkai Zhub9f19592012-10-16 14:27:38 -0700430 if (!m_perBranch)
431 {
432 m_onRemove (info->toString ());
433 }
Alexander Afanasyev04fd8a82012-03-12 15:40:48 -0700434 }
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700435 }
436 else
437 {
Alexander Afanasyev750d1872012-03-12 15:33:56 -0700438 }
439 }
440
Zhenkai Zhu1cb29292012-05-31 22:54:34 -0700441 if (!v.empty())
442 {
Zhenkai Zhub9f19592012-10-16 14:27:38 -0700443 if (!m_perBranch)
444 {
445 m_onUpdate(v);
446 }
Zhenkai Zhu1cb29292012-05-31 22:54:34 -0700447 }
448
Alexander Afanasyev235c6d72012-03-15 22:28:43 -0700449 insertToDiffLog (diffLog);
Alexander Afanasyev750d1872012-03-12 15:33:56 -0700450 }
Zhenkai Zhu97e36bd2012-06-06 13:55:03 -0700451 catch (Error::SyncStateMsgDecodingFailure &e)
Alexander Afanasyev750d1872012-03-12 15:33:56 -0700452 {
Alexander Afanasyev282a10b2012-05-09 12:56:38 -0700453 _LOG_TRACE ("Something really fishy happened during state decoding " <<
454 diagnostic_information (e));
Alexander Afanasyevd4cca472012-03-13 14:25:46 -0700455 diffLog.reset ();
456 // don't do anything
Alexander Afanasyev750d1872012-03-12 15:33:56 -0700457 }
458
Alexander Afanasyev2b6fbfb2012-05-24 10:57:14 -0700459 if ((diffLog != 0 && diffLog->getLeaves ().size () > 0) ||
460 ownInterestSatisfied)
Alexander Afanasyev750d1872012-03-12 15:33:56 -0700461 {
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700462 // Do it only if everything went fine and state changed
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700463
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700464 // this is kind of wrong
465 // satisfyPendingSyncInterests (diffLog); // if there are interests in PIT, there is a point to satisfy them using new state
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700466
Alexander Afanasyev282a10b2012-05-09 12:56:38 -0700467 // if state has changed, then it is safe to express a new interest
Alexander Afanasyevd4cca472012-03-13 14:25:46 -0700468 m_scheduler.cancel (REEXPRESSING_INTEREST);
Alexander Afanasyev282a10b2012-05-09 12:56:38 -0700469 m_scheduler.schedule (TIME_SECONDS_WITH_JITTER (0),
Alexander Afanasyevd4cca472012-03-13 14:25:46 -0700470 bind (&SyncLogic::sendSyncInterest, this),
471 REEXPRESSING_INTEREST);
Alexander Afanasyev04fd8a82012-03-12 15:40:48 -0700472 }
Chaoyi Bian44fff0c2012-03-07 21:07:22 -0800473}
474
Alexander Afanasyevc1030192012-03-08 22:21:28 -0800475void
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700476SyncLogic::processSyncRecoveryInterest (const std::string &name, DigestConstPtr digest)
Chaoyi Bian44fff0c2012-03-07 21:07:22 -0800477{
Yingdi Yu43e71612013-10-30 22:19:31 -0700478 _LOG_DEBUG("processSyncRecoveryInterest");
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700479 DiffStateContainer::iterator stateInDiffLog = m_log.find (digest);
480
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700481 if (stateInDiffLog == m_log.end ())
Alexander Afanasyevbf2b4362012-03-12 23:55:09 -0700482 {
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700483 _LOG_TRACE ("Could not find " << *digest << " in digest log");
484 return;
485 }
Alexander Afanasyev860e6fe2012-03-15 17:30:31 -0700486
Zhenkai Zhu2d3e2702012-10-15 14:18:05 -0700487 SyncStateMsg ssm;
488 {
489 recursive_mutex::scoped_lock lock (m_stateMutex);
490 ssm << (*m_state);
491 }
492 sendSyncData (name, digest, ssm);
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700493}
Alexander Afanasyev860e6fe2012-03-15 17:30:31 -0700494
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700495void
496SyncLogic::satisfyPendingSyncInterests (DiffStateConstPtr diffLog)
497{
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700498 DiffStatePtr fullStateLog = make_shared<DiffState> ();
499 {
500 recursive_mutex::scoped_lock lock (m_stateMutex);
501 BOOST_FOREACH (LeafConstPtr leaf, m_state->getLeaves ()/*.get<timed> ()*/)
502 {
503 fullStateLog->update (leaf->getInfo (), leaf->getSeq ());
504 /// @todo Impose limit on how many state info should be send out
505 }
506 }
507
508 try
Alexander Afanasyevc1030192012-03-08 22:21:28 -0800509 {
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700510 uint32_t counter = 0;
511 while (m_syncInterestTable.size () > 0)
Alexander Afanasyev860e6fe2012-03-15 17:30:31 -0700512 {
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700513 Interest interest = m_syncInterestTable.pop ();
514
515 if (!interest.m_unknown)
516 {
517 _LOG_TRACE (">> D " << interest.m_name);
518 sendSyncData (interest.m_name, interest.m_digest, diffLog);
519 }
520 else
521 {
522 _LOG_TRACE (">> D (unknown)" << interest.m_name);
523 sendSyncData (interest.m_name, interest.m_digest, fullStateLog);
524 }
525 counter ++;
Alexander Afanasyevf07ab352012-03-13 12:57:46 -0700526 }
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700527 _LOG_DEBUG ("Satisfied " << counter << " pending interests");
Alexander Afanasyevf07ab352012-03-13 12:57:46 -0700528 }
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700529 catch (Error::InterestTableIsEmpty &e)
530 {
531 // ok. not really an error
Chaoyi Bian44fff0c2012-03-07 21:07:22 -0800532 }
Chaoyi Bian44fff0c2012-03-07 21:07:22 -0800533}
534
Alexander Afanasyevc1030192012-03-08 22:21:28 -0800535void
Alexander Afanasyev235c6d72012-03-15 22:28:43 -0700536SyncLogic::insertToDiffLog (DiffStatePtr diffLog)
Alexander Afanasyev1b449c42012-03-13 20:24:07 -0700537{
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700538 diffLog->setDigest (m_state->getDigest());
Alexander Afanasyev1b449c42012-03-13 20:24:07 -0700539 if (m_log.size () > 0)
540 {
541 m_log.get<sequenced> ().front ()->setNext (diffLog);
542 }
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700543 m_log.erase (m_state->getDigest()); // remove diff state with the same digest. next pointers are still valid
Alexander Afanasyev1b449c42012-03-13 20:24:07 -0700544 /// @todo Optimization
Alexander Afanasyev7df73332012-03-15 12:31:49 -0700545 m_log.get<sequenced> ().push_front (diffLog);
Alexander Afanasyev1b449c42012-03-13 20:24:07 -0700546}
547
548void
Zhenkai Zhu0efa37b2012-03-12 13:54:12 -0700549SyncLogic::addLocalNames (const string &prefix, uint32_t session, uint32_t seq)
550{
Alexander Afanasyev1b449c42012-03-13 20:24:07 -0700551 DiffStatePtr diff;
552 {
553 //cout << "Add local names" <<endl;
554 recursive_mutex::scoped_lock lock (m_stateMutex);
555 NameInfoConstPtr info = StdNameInfo::FindOrCreate(prefix);
Alexander Afanasyev860e6fe2012-03-15 17:30:31 -0700556
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700557 _LOG_DEBUG ("addLocalNames (): old state " << *m_state->getDigest ());
Zhenkai Zhu0efa37b2012-03-12 13:54:12 -0700558
Zhenkai Zhu0efa37b2012-03-12 13:54:12 -0700559 SeqNo seqN (session, seq);
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700560 m_state->update(info, seqN);
Zhenkai Zhu0efa37b2012-03-12 13:54:12 -0700561
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700562 _LOG_DEBUG ("addLocalNames (): new state " << *m_state->getDigest ());
Alexander Afanasyev860e6fe2012-03-15 17:30:31 -0700563
Alexander Afanasyev1b449c42012-03-13 20:24:07 -0700564 diff = make_shared<DiffState>();
565 diff->update(info, seqN);
Alexander Afanasyev235c6d72012-03-15 22:28:43 -0700566 insertToDiffLog (diff);
Alexander Afanasyev1b449c42012-03-13 20:24:07 -0700567 }
568
Alexander Afanasyev860e6fe2012-03-15 17:30:31 -0700569 // _LOG_DEBUG ("PIT size: " << m_syncInterestTable.size ());
570 satisfyPendingSyncInterests (diff);
Zhenkai Zhu0efa37b2012-03-12 13:54:12 -0700571}
572
573void
Alexander Afanasyevd91497c2012-03-12 15:39:30 -0700574SyncLogic::remove(const string &prefix)
575{
Alexander Afanasyev1b449c42012-03-13 20:24:07 -0700576 DiffStatePtr diff;
577 {
578 recursive_mutex::scoped_lock lock (m_stateMutex);
579 NameInfoConstPtr info = StdNameInfo::FindOrCreate(prefix);
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700580 m_state->remove(info);
Zhenkai Zhu0efa37b2012-03-12 13:54:12 -0700581
Zhenkai Zhua2e0b082012-09-26 10:34:15 -0700582 // increment the sequence number for the forwarder node
583 NameInfoConstPtr forwarderInfo = StdNameInfo::FindOrCreate(forwarderPrefix);
Zhenkai Zhua2e0b082012-09-26 10:34:15 -0700584
Alexander Afanasyev7e1986f2012-10-04 10:21:10 -0700585 LeafContainer::iterator item = m_state->getLeaves ().find (forwarderInfo);
586 SeqNo seqNo (0);
587 if (item != m_state->getLeaves ().end ())
588 {
589 seqNo = (*item)->getSeq ();
590 ++seqNo;
591 }
592 m_state->update (forwarderInfo, seqNo);
Zhenkai Zhua2e0b082012-09-26 10:34:15 -0700593
Alexander Afanasyev1b449c42012-03-13 20:24:07 -0700594 diff = make_shared<DiffState>();
595 diff->remove(info);
Alexander Afanasyev7e1986f2012-10-04 10:21:10 -0700596 diff->update(forwarderInfo, seqNo);
Zhenkai Zhu0efa37b2012-03-12 13:54:12 -0700597
Alexander Afanasyev235c6d72012-03-15 22:28:43 -0700598 insertToDiffLog (diff);
Alexander Afanasyev1b449c42012-03-13 20:24:07 -0700599 }
600
Alexander Afanasyev860e6fe2012-03-15 17:30:31 -0700601 satisfyPendingSyncInterests (diff);
Zhenkai Zhu0efa37b2012-03-12 13:54:12 -0700602}
603
604void
Alexander Afanasyevc1030192012-03-08 22:21:28 -0800605SyncLogic::sendSyncInterest ()
Chaoyi Bian44fff0c2012-03-07 21:07:22 -0800606{
Yingdi Yu43e71612013-10-30 22:19:31 -0700607 _LOG_DEBUG("sendSyncInterest");
Alexander Afanasyev172d2b72012-03-08 23:43:39 -0800608 ostringstream os;
Chaoyi Bian89ee2dc2012-03-09 14:06:01 -0800609
Alexander Afanasyev860e6fe2012-03-15 17:30:31 -0700610 {
Alexander Afanasyev860e6fe2012-03-15 17:30:31 -0700611 recursive_mutex::scoped_lock lock (m_stateMutex);
Alexander Afanasyev4f9ea482012-03-15 11:57:29 -0700612
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700613 os << m_syncPrefix << "/" << *m_state->getDigest();
614 m_outstandingInterestName = os.str ();
Alexander Afanasyev860e6fe2012-03-15 17:30:31 -0700615 _LOG_TRACE (">> I " << os.str ());
Alexander Afanasyev860e6fe2012-03-15 17:30:31 -0700616 }
Alexander Afanasyevbf2b4362012-03-12 23:55:09 -0700617
Yingdi Yu43e71612013-10-30 22:19:31 -0700618 _LOG_DEBUG("sendSyncInterest: " << os.str());
619
Alexander Afanasyevbf2b4362012-03-12 23:55:09 -0700620 m_scheduler.cancel (REEXPRESSING_INTEREST);
Alexander Afanasyev983b0e92012-04-26 12:44:18 -0700621 m_scheduler.schedule (TIME_SECONDS_WITH_JITTER (m_syncInterestReexpress),
Alexander Afanasyevbf2b4362012-03-12 23:55:09 -0700622 bind (&SyncLogic::sendSyncInterest, this),
623 REEXPRESSING_INTEREST);
Yingdi Yu43e71612013-10-30 22:19:31 -0700624
625 Ptr<ndn::Interest> interestPtr = Ptr<ndn::Interest>(new ndn::Interest(os.str()));
626 Ptr<Closure> closure = Ptr<Closure> (new Closure(boost::bind(&SyncLogic::respondSyncData,
627 this,
628 _1),
629 boost::bind(&SyncLogic::onSyncDataTimeout,
630 this,
631 _1,
632 _2,
633 0),
634 boost::bind(&SyncLogic::onSyncDataUnverified,
635 this)));
636 m_handler->sendInterest(interestPtr, closure);
Chaoyi Bian44fff0c2012-03-07 21:07:22 -0800637}
638
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700639void
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700640SyncLogic::sendSyncRecoveryInterests (DigestConstPtr digest)
641{
642 ostringstream os;
643 os << m_syncPrefix << "/recovery/" << *digest;
644 _LOG_TRACE (">> I " << os.str ());
645
Alexander Afanasyeva76010b2012-05-24 21:31:49 -0700646 TimeDuration nextRetransmission = TIME_MILLISECONDS_WITH_JITTER (m_recoveryRetransmissionInterval);
647 m_recoveryRetransmissionInterval <<= 1;
648
649 m_scheduler.cancel (REEXPRESSING_RECOVERY_INTEREST);
650 if (m_recoveryRetransmissionInterval < 100*1000) // <100 seconds
651 {
652 m_scheduler.schedule (nextRetransmission,
653 bind (&SyncLogic::sendSyncRecoveryInterests, this, digest),
654 REEXPRESSING_RECOVERY_INTEREST);
655 }
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700656
Yingdi Yu43e71612013-10-30 22:19:31 -0700657 Ptr<ndn::Interest> interestPtr = Ptr<ndn::Interest>(new ndn::Interest(os.str()));
658 Ptr<Closure> closure = Ptr<Closure> (new Closure(boost::bind(&SyncLogic::respondSyncData,
659 this,
660 _1),
661 boost::bind(&SyncLogic::onSyncDataTimeout,
662 this,
663 _1,
664 _2,
665 0),
666 boost::bind(&SyncLogic::onSyncDataUnverified,
667 this)));
668 m_handler->sendInterest (interestPtr, closure);
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700669}
670
671
672void
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700673SyncLogic::sendSyncData (const std::string &name, DigestConstPtr digest, StateConstPtr state)
674{
Zhenkai Zhu2d3e2702012-10-15 14:18:05 -0700675 SyncStateMsg msg;
676 msg << (*state);
677 sendSyncData(name, digest, msg);
678}
679
680// pass in state msg instead of state, so that there is no need to lock the state until
681// this function returns
682void
683SyncLogic::sendSyncData (const std::string &name, DigestConstPtr digest, SyncStateMsg &ssm)
684{
Alexander Afanasyev46eb5262012-05-10 16:30:35 -0700685 _LOG_TRACE (">> D " << name);
Zhenkai Zhu3cfdcb92012-06-06 15:20:10 -0700686 int size = ssm.ByteSize();
687 char *wireData = new char[size];
688 ssm.SerializeToArray(wireData, size);
Yingdi Yu43e71612013-10-30 22:19:31 -0700689 Blob blob(wireData, size);
690 Name dataName(name);
691 Name signingIdentity = m_policyManager->inferSigningIdentity(dataName);
692 m_handler->publishDataByIdentity (dataName,
693 blob,
694 signingIdentity,
695 m_syncResponseFreshness); // in NS-3 it doesn't have any effect... yet
Chaoyi Bian6665d5a2012-06-07 14:16:44 -0700696 delete []wireData;
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700697
Alexander Afanasyev282a10b2012-05-09 12:56:38 -0700698 // checking if our own interest got satisfied
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700699 bool satisfiedOwnInterest = false;
700 {
701 recursive_mutex::scoped_lock lock (m_stateMutex);
702 satisfiedOwnInterest = (m_outstandingInterestName == name);
703 }
704
705 if (satisfiedOwnInterest)
706 {
707 _LOG_TRACE ("Satisfied our own Interest. Re-expressing (hopefully with a new digest)");
708
709 m_scheduler.cancel (REEXPRESSING_INTEREST);
710 m_scheduler.schedule (TIME_SECONDS_WITH_JITTER (0),
711 bind (&SyncLogic::sendSyncInterest, this),
712 REEXPRESSING_INTEREST);
713 }
714}
715
Zhenkai Zhue5660932012-06-04 15:25:20 -0700716string
717SyncLogic::getRootDigest()
718{
719 ostringstream os;
720 recursive_mutex::scoped_lock lock (m_stateMutex);
721 os << *m_state->getDigest();
722 return os.str();
723}
724
Alexander Afanasyev80237382012-10-04 10:20:47 -0700725size_t
726SyncLogic::getNumberOfBranches () const
727{
728 recursive_mutex::scoped_lock lock (m_stateMutex);
729 return m_state->getLeaves ().size ();
730}
731
Alexander Afanasyev5548c042012-10-04 19:10:09 -0700732void
733SyncLogic::printState () const
734{
735 recursive_mutex::scoped_lock lock (m_stateMutex);
736
737 BOOST_FOREACH (const boost::shared_ptr<Sync::Leaf> leaf, m_state->getLeaves ())
738 {
739 std::cout << *leaf << std::endl;
740 }
741}
742
Zhenkai Zhud5aec4b2012-10-09 12:01:19 -0700743std::map<std::string, bool>
744SyncLogic::getBranchPrefixes() const
745{
746 recursive_mutex::scoped_lock lock (m_stateMutex);
747
748 std::map<std::string, bool> m;
749
750 BOOST_FOREACH (const boost::shared_ptr<Sync::Leaf> leaf, m_state->getLeaves ())
751 {
752 std::string prefix = leaf->getInfo()->toString();
753 // do not return forwarder prefix
754 if (prefix != forwarderPrefix)
755 {
756 m.insert(pair<std::string, bool>(prefix, false));
757 }
758 }
759
760 return m;
761}
Alexander Afanasyevf3c03a92012-05-09 12:00:37 -0700762
Alexander Afanasyevc1030192012-03-08 22:21:28 -0800763}