blob: aa08accfa31d6280ee84e75fd93e5c7885f91e8f [file] [log] [blame]
Yingdi Yuf7ede412014-08-30 20:37:52 -07001/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
2/*
Sonu Mishra0dadc572016-12-12 23:59:41 -08003 * Copyright (c) 2012-2017 University of California, Los Angeles
Yingdi Yuf7ede412014-08-30 20:37:52 -07004 *
5 * This file is part of ChronoSync, synchronization library for distributed realtime
6 * applications for NDN.
7 *
8 * ChronoSync is free software: you can redistribute it and/or modify it under the terms
9 * of the GNU General Public License as published by the Free Software Foundation, either
10 * version 3 of the License, or (at your option) any later version.
11 *
12 * ChronoSync is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
13 * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
14 * PURPOSE. See the GNU General Public License for more details.
15 *
16 * You should have received a copy of the GNU General Public License along with
17 * ChronoSync, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
18 *
19 * @author Zhenkai Zhu <http://irl.cs.ucla.edu/~zhenkai/>
20 * @author Chaoyi Bian <bcy@pku.edu.cn>
21 * @author Alexander Afanasyev <http://lasr.cs.ucla.edu/afanasyev/index.html>
22 * @author Yingdi Yu <yingdi@cs.ucla.edu>
Sonu Mishra0dadc572016-12-12 23:59:41 -080023 * @author Sonu Mishra <https://www.linkedin.com/in/mishrasonu>
Yingdi Yuf7ede412014-08-30 20:37:52 -070024 */
25
26#include "logic.hpp"
27#include "logger.hpp"
28
Alexander Afanasyev36eb3ed2017-01-11 12:35:58 -080029INIT_LOGGER(Logic);
Yingdi Yuf7ede412014-08-30 20:37:52 -070030
31#ifdef _DEBUG
32#define _LOG_DEBUG_ID(v) _LOG_DEBUG("Instance" << m_instanceId << ": " << v)
33#else
34#define _LOG_DEBUG_ID(v) _LOG_DEBUG(v)
35#endif
36
37namespace chronosync {
38
39using ndn::ConstBufferPtr;
40using ndn::EventId;
41
42const uint8_t EMPTY_DIGEST_VALUE[] = {
43 0xe3, 0xb0, 0xc4, 0x42, 0x98, 0xfc, 0x1c, 0x14,
44 0x9a, 0xfb, 0xf4, 0xc8, 0x99, 0x6f, 0xb9, 0x24,
45 0x27, 0xae, 0x41, 0xe4, 0x64, 0x9b, 0x93, 0x4c,
46 0xa4, 0x95, 0x99, 0x1b, 0x78, 0x52, 0xb8, 0x55
47};
48
49#ifdef _DEBUG
50int Logic::m_instanceCounter = 0;
51#endif
52
Yingdi Yucd339022014-11-05 17:51:19 -080053const ndn::Name Logic::DEFAULT_NAME;
Qiuhan Ding8c095fd2014-11-19 17:38:32 -080054const ndn::Name Logic::EMPTY_NAME;
Yingdi Yucd339022014-11-05 17:51:19 -080055const ndn::shared_ptr<ndn::Validator> Logic::DEFAULT_VALIDATOR;
Yingdi Yuf7ede412014-08-30 20:37:52 -070056const time::steady_clock::Duration Logic::DEFAULT_RESET_TIMER = time::seconds(0);
57const time::steady_clock::Duration Logic::DEFAULT_CANCEL_RESET_TIMER = time::milliseconds(500);
58const time::milliseconds Logic::DEFAULT_RESET_INTEREST_LIFETIME(1000);
59const time::milliseconds Logic::DEFAULT_SYNC_INTEREST_LIFETIME(1000);
60const time::milliseconds Logic::DEFAULT_SYNC_REPLY_FRESHNESS(1000);
Sonu Mishra4d3a2e02017-01-18 20:27:51 -080061const time::milliseconds Logic::DEFAULT_RECOVERY_INTEREST_LIFETIME(1000);
Yingdi Yuf7ede412014-08-30 20:37:52 -070062
63const ndn::ConstBufferPtr Logic::EMPTY_DIGEST(new ndn::Buffer(EMPTY_DIGEST_VALUE, 32));
64const ndn::name::Component Logic::RESET_COMPONENT("reset");
Sonu Mishra4d3a2e02017-01-18 20:27:51 -080065const ndn::name::Component Logic::RECOVERY_COMPONENT("recovery");
Yingdi Yuf7ede412014-08-30 20:37:52 -070066
67Logic::Logic(ndn::Face& face,
68 const Name& syncPrefix,
Qiuhan Ding8c095fd2014-11-19 17:38:32 -080069 const Name& defaultUserPrefix,
Yingdi Yuf7ede412014-08-30 20:37:52 -070070 const UpdateCallback& onUpdate,
Qiuhan Ding8c095fd2014-11-19 17:38:32 -080071 const Name& defaultSigningId,
Yingdi Yucd339022014-11-05 17:51:19 -080072 ndn::shared_ptr<ndn::Validator> validator,
Yingdi Yuf7ede412014-08-30 20:37:52 -070073 const time::steady_clock::Duration& resetTimer,
74 const time::steady_clock::Duration& cancelResetTimer,
75 const time::milliseconds& resetInterestLifetime,
76 const time::milliseconds& syncInterestLifetime,
Sonu Mishra4d3a2e02017-01-18 20:27:51 -080077 const time::milliseconds& syncReplyFreshness,
78 const time::milliseconds& recoveryInterestLifetime)
Yingdi Yuf7ede412014-08-30 20:37:52 -070079 : m_face(face)
80 , m_syncPrefix(syncPrefix)
Qiuhan Ding8c095fd2014-11-19 17:38:32 -080081 , m_defaultUserPrefix(defaultUserPrefix)
Yingdi Yuf7ede412014-08-30 20:37:52 -070082 , m_interestTable(m_face.getIoService())
83 , m_outstandingInterestId(0)
84 , m_isInReset(false)
85 , m_needPeriodReset(resetTimer > time::steady_clock::Duration::zero())
86 , m_onUpdate(onUpdate)
87 , m_scheduler(m_face.getIoService())
88 , m_randomGenerator(static_cast<unsigned int>(std::time(0)))
89 , m_rangeUniformRandom(m_randomGenerator, boost::uniform_int<>(100,500))
90 , m_reexpressionJitter(m_randomGenerator, boost::uniform_int<>(100,500))
91 , m_resetTimer(resetTimer)
92 , m_cancelResetTimer(cancelResetTimer)
93 , m_resetInterestLifetime(resetInterestLifetime)
94 , m_syncInterestLifetime(syncInterestLifetime)
95 , m_syncReplyFreshness(syncReplyFreshness)
Sonu Mishra4d3a2e02017-01-18 20:27:51 -080096 , m_recoveryInterestLifetime(recoveryInterestLifetime)
Qiuhan Ding8c095fd2014-11-19 17:38:32 -080097 , m_defaultSigningId(defaultSigningId)
Yingdi Yucd339022014-11-05 17:51:19 -080098 , m_validator(validator)
Yingdi Yuf7ede412014-08-30 20:37:52 -070099{
100#ifdef _DEBUG
101 m_instanceId = m_instanceCounter++;
102#endif
103
104 _LOG_DEBUG_ID(">> Logic::Logic");
105
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800106 addUserNode(m_defaultUserPrefix, m_defaultSigningId);
107
108
Yingdi Yuf7ede412014-08-30 20:37:52 -0700109 m_syncReset = m_syncPrefix;
110 m_syncReset.append("reset");
111
112 _LOG_DEBUG_ID("Listen to: " << m_syncPrefix);
113 m_syncRegisteredPrefixId =
114 m_face.setInterestFilter(m_syncPrefix,
115 bind(&Logic::onSyncInterest, this, _1, _2),
116 bind(&Logic::onSyncRegisterFailed, this, _1, _2));
117
Qiuhan Dinge246b622014-12-03 21:57:48 -0800118 sendSyncInterest();
Yingdi Yuf7ede412014-08-30 20:37:52 -0700119 _LOG_DEBUG_ID("<< Logic::Logic");
120}
121
122Logic::~Logic()
123{
Yingdi Yuf7ede412014-08-30 20:37:52 -0700124 m_scheduler.cancelAllEvents();
Yingdi Yu9d5679a2015-02-01 00:17:58 -0800125 m_interestTable.clear();
126 m_face.shutdown();
Yingdi Yuf7ede412014-08-30 20:37:52 -0700127}
128
129void
Qiuhan Dingfb8c9e02015-01-30 14:04:55 -0800130Logic::reset(bool isOnInterest)
Yingdi Yuf7ede412014-08-30 20:37:52 -0700131{
132 m_isInReset = true;
133
134 m_state.reset();
135 m_log.clear();
136
Qiuhan Dingfb8c9e02015-01-30 14:04:55 -0800137 if (!isOnInterest)
138 sendResetInterest();
Yingdi Yuf7ede412014-08-30 20:37:52 -0700139
140 // reset outstanding interest name, so that data for previous interest will be dropped.
141 if (m_outstandingInterestId != 0) {
142 m_face.removePendingInterest(m_outstandingInterestId);
143 m_outstandingInterestId = 0;
144 }
145
146 sendSyncInterest();
147
148 if (static_cast<bool>(m_delayedInterestProcessingId))
149 m_scheduler.cancelEvent(m_delayedInterestProcessingId);
150
151 m_delayedInterestProcessingId =
152 m_scheduler.scheduleEvent(m_cancelResetTimer,
153 bind(&Logic::cancelReset, this));
154}
155
156void
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800157Logic::setDefaultUserPrefix(const Name& defaultUserPrefix)
Yingdi Yuf7ede412014-08-30 20:37:52 -0700158{
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800159 if (defaultUserPrefix != EMPTY_NAME) {
160 if (m_nodeList.find(defaultUserPrefix) != m_nodeList.end()) {
161 m_defaultUserPrefix = defaultUserPrefix;
162 m_defaultSigningId = m_nodeList[defaultUserPrefix].signingId;
163 }
164 }
Yingdi Yuf7ede412014-08-30 20:37:52 -0700165}
166
167void
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800168Logic::addUserNode(const Name& userPrefix, const Name& signingId)
Yingdi Yuf7ede412014-08-30 20:37:52 -0700169{
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800170 if (userPrefix == EMPTY_NAME)
Yingdi Yuf7ede412014-08-30 20:37:52 -0700171 return;
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800172 if (m_defaultUserPrefix == EMPTY_NAME) {
173 m_defaultUserPrefix = userPrefix;
174 m_defaultSigningId = signingId;
175 }
176 if (m_nodeList.find(userPrefix) == m_nodeList.end()) {
177 m_nodeList[userPrefix].userPrefix = userPrefix;
178 m_nodeList[userPrefix].signingId = signingId;
179 Name sessionName = userPrefix;
180 sessionName.appendNumber(ndn::time::toUnixTimestamp(ndn::time::system_clock::now()).count());
181 m_nodeList[userPrefix].sessionName = sessionName;
182 m_nodeList[userPrefix].seqNo = 0;
Qiuhan Dingfb8c9e02015-01-30 14:04:55 -0800183 reset(false);
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800184 }
185}
Yingdi Yuf7ede412014-08-30 20:37:52 -0700186
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800187void
188Logic::removeUserNode(const Name& userPrefix)
189{
190 auto userNode = m_nodeList.find(userPrefix);
191 if (userNode != m_nodeList.end()) {
192 m_nodeList.erase(userNode);
193 if (m_defaultUserPrefix == userPrefix) {
194 if (!m_nodeList.empty()) {
195 m_defaultUserPrefix = m_nodeList.begin()->second.userPrefix;
196 m_defaultSigningId = m_nodeList.begin()->second.signingId;
197 }
198 else {
199 m_defaultUserPrefix = EMPTY_NAME;
200 m_defaultSigningId = DEFAULT_NAME;
201 }
Yingdi Yuf7ede412014-08-30 20:37:52 -0700202 }
Qiuhan Dingfb8c9e02015-01-30 14:04:55 -0800203 reset(false);
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800204 }
205}
Yingdi Yuf7ede412014-08-30 20:37:52 -0700206
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800207const Name&
208Logic::getSessionName(Name prefix)
209{
210 if (prefix == EMPTY_NAME)
211 prefix = m_defaultUserPrefix;
212 auto node = m_nodeList.find(prefix);
213 if (node != m_nodeList.end())
214 return node->second.sessionName;
215 else
216 throw Error("Refer to non-existent node:" + prefix.toUri());
217}
Yingdi Yuf7ede412014-08-30 20:37:52 -0700218
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800219const SeqNo&
220Logic::getSeqNo(Name prefix)
221{
222 if (prefix == EMPTY_NAME)
223 prefix = m_defaultUserPrefix;
224 auto node = m_nodeList.find(prefix);
225 if (node != m_nodeList.end())
226 return node->second.seqNo;
227 else
228 throw Logic::Error("Refer to non-existent node:" + prefix.toUri());
Yingdi Yuf7ede412014-08-30 20:37:52 -0700229
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800230}
231
232void
233Logic::updateSeqNo(const SeqNo& seqNo, const Name &updatePrefix)
234{
235 Name prefix;
236 if (updatePrefix == EMPTY_NAME) {
237 if (m_defaultUserPrefix == EMPTY_NAME)
238 return;
239 prefix = m_defaultUserPrefix;
240 }
241 else
242 prefix = updatePrefix;
243
244 auto it = m_nodeList.find(prefix);
245 if (it != m_nodeList.end()) {
246 NodeInfo& node = it->second;
247 _LOG_DEBUG_ID(">> Logic::updateSeqNo");
248 _LOG_DEBUG_ID("seqNo: " << seqNo << " m_seqNo: " << node.seqNo);
249 if (seqNo < node.seqNo || seqNo == 0)
250 return;
251
252 node.seqNo = seqNo;
253 _LOG_DEBUG_ID("updateSeqNo: m_seqNo " << node.seqNo);
254
255 if (!m_isInReset) {
256 _LOG_DEBUG_ID("updateSeqNo: not in Reset ");
257 ndn::ConstBufferPtr previousRoot = m_state.getRootDigest();
258 {
259 using namespace CryptoPP;
260
261 std::string hash;
262 StringSource(previousRoot->buf(), previousRoot->size(), true,
263 new HexEncoder(new StringSink(hash), false));
264 _LOG_DEBUG_ID("Hash: " << hash);
265 }
266
267 bool isInserted = false;
268 bool isUpdated = false;
269 SeqNo oldSeq;
270 boost::tie(isInserted, isUpdated, oldSeq) = m_state.update(node.sessionName,
271 node.seqNo);
272
273 _LOG_DEBUG_ID("Insert: " << std::boolalpha << isInserted);
274 _LOG_DEBUG_ID("Updated: " << std::boolalpha << isUpdated);
275 if (isInserted || isUpdated) {
276 DiffStatePtr commit = make_shared<DiffState>();
277 commit->update(node.sessionName, node.seqNo);
278 commit->setRootDigest(m_state.getRootDigest());
279 insertToDiffLog(commit, previousRoot);
280
281 satisfyPendingSyncInterests(prefix, commit);
282 }
Yingdi Yuf7ede412014-08-30 20:37:52 -0700283 }
284 }
285}
286
287ConstBufferPtr
288Logic::getRootDigest() const
289{
290 return m_state.getRootDigest();
291}
292
293void
294Logic::printState(std::ostream& os) const
295{
296 BOOST_FOREACH(ConstLeafPtr leaf, m_state.getLeaves())
297 {
298 os << *leaf << "\n";
299 }
300}
301
302std::set<Name>
303Logic::getSessionNames() const
304{
305 std::set<Name> sessionNames;
306
307 BOOST_FOREACH(ConstLeafPtr leaf, m_state.getLeaves())
308 {
309 sessionNames.insert(leaf->getSessionName());
310 }
311
312 return sessionNames;
313}
314
315void
316Logic::onSyncInterest(const Name& prefix, const Interest& interest)
317{
318 _LOG_DEBUG_ID(">> Logic::onSyncInterest");
319 Name name = interest.getName();
320
321 _LOG_DEBUG_ID("InterestName: " << name);
322
Sonu Mishra4d3a2e02017-01-18 20:27:51 -0800323 if (name.size() >= 1 && RESET_COMPONENT == name.get(-1)) {
324 processResetInterest(interest);
325 }
326 else if (name.size() >= 2 && RECOVERY_COMPONENT == name.get(-2)) {
327 processRecoveryInterest(interest);
328 }
329 else {
Yingdi Yuf7ede412014-08-30 20:37:52 -0700330 processSyncInterest(interest.shared_from_this());
331 }
Yingdi Yuf7ede412014-08-30 20:37:52 -0700332
333 _LOG_DEBUG_ID("<< Logic::onSyncInterest");
334}
335
336void
337Logic::onSyncRegisterFailed(const Name& prefix, const std::string& msg)
338{
339 //Sync prefix registration failed
340 _LOG_DEBUG_ID(">> Logic::onSyncRegisterFailed");
341}
342
343void
344Logic::onSyncData(const Interest& interest, Data& data)
345{
346 _LOG_DEBUG_ID(">> Logic::onSyncData");
Yingdi Yucd339022014-11-05 17:51:19 -0800347 if (static_cast<bool>(m_validator))
348 m_validator->validate(data,
349 bind(&Logic::onSyncDataValidated, this, _1),
350 bind(&Logic::onSyncDataValidationFailed, this, _1));
351 else
352 onSyncDataValidated(data.shared_from_this());
Yingdi Yuf7ede412014-08-30 20:37:52 -0700353 _LOG_DEBUG_ID("<< Logic::onSyncData");
354}
355
356void
357Logic::onResetData(const Interest& interest, Data& data)
358{
359 // This should not happened, drop the received data.
360}
361
362void
363Logic::onSyncTimeout(const Interest& interest)
364{
365 // It is OK. Others will handle the time out situation.
366 _LOG_DEBUG_ID(">> Logic::onSyncTimeout");
367 _LOG_DEBUG_ID("Interest: " << interest.getName());
368 _LOG_DEBUG_ID("<< Logic::onSyncTimeout");
369}
370
371void
372Logic::onSyncDataValidationFailed(const shared_ptr<const Data>& data)
373{
374 // SyncReply cannot be validated.
375}
376
377void
378Logic::onSyncDataValidated(const shared_ptr<const Data>& data)
379{
380 Name name = data->getName();
381 ConstBufferPtr digest = make_shared<ndn::Buffer>(name.get(-1).value(), name.get(-1).value_size());
382
383 processSyncData(name, digest, data->getContent().blockFromValue());
384}
385
386void
387Logic::processSyncInterest(const shared_ptr<const Interest>& interest,
388 bool isTimedProcessing/*=false*/)
389{
390 _LOG_DEBUG_ID(">> Logic::processSyncInterest");
391
392 const Name& name = interest->getName();
393 ConstBufferPtr digest =
394 make_shared<ndn::Buffer>(name.get(-1).value(), name.get(-1).value_size());
395
396 ConstBufferPtr rootDigest = m_state.getRootDigest();
397
398 // If the digest of the incoming interest is the same as root digest
399 // Put the interest into InterestTable
400 if (*rootDigest == *digest) {
401 _LOG_DEBUG_ID("Oh, we are in the same state");
402 m_interestTable.insert(interest, digest, false);
403
404 if (!m_isInReset)
405 return;
406
407 if (!isTimedProcessing) {
408 _LOG_DEBUG_ID("Non timed processing in reset");
409 // Still in reset, our own seq has not been put into state yet
410 // Do not hurry, some others may be also resetting and may send their reply
411 if (static_cast<bool>(m_delayedInterestProcessingId))
412 m_scheduler.cancelEvent(m_delayedInterestProcessingId);
413
414 time::milliseconds after(m_rangeUniformRandom());
415 _LOG_DEBUG_ID("After: " << after);
416 m_delayedInterestProcessingId =
417 m_scheduler.scheduleEvent(after,
418 bind(&Logic::processSyncInterest, this, interest, true));
419 }
420 else {
421 _LOG_DEBUG_ID("Timed processing in reset");
422 // Now we can get out of reset state by putting our own stuff into m_state.
423 cancelReset();
424 }
425
426 return;
427 }
428
429 // If the digest of incoming interest is an "empty" digest
Sonu Mishrae10acbc2017-01-18 14:14:05 -0800430 if (*digest == *EMPTY_DIGEST) {
Yingdi Yuf7ede412014-08-30 20:37:52 -0700431 _LOG_DEBUG_ID("Poor guy, he knows nothing");
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800432 sendSyncData(m_defaultUserPrefix, name, m_state);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700433 return;
434 }
435
436 DiffStateContainer::iterator stateIter = m_log.find(digest);
437 // If the digest of incoming interest can be found from the log
438 if (stateIter != m_log.end()) {
439 _LOG_DEBUG_ID("It is ok, you are so close");
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800440 sendSyncData(m_defaultUserPrefix, name, *(*stateIter)->diff());
Yingdi Yuf7ede412014-08-30 20:37:52 -0700441 return;
442 }
443
444 if (!isTimedProcessing) {
445 _LOG_DEBUG_ID("Let's wait, just wait for a while");
446 // Do not hurry, some incoming SyncReplies may help us to recognize the digest
Yingdi Yu53f5f042015-01-31 16:33:25 -0800447 bool doesExist = m_interestTable.has(digest);
448 m_interestTable.insert(interest, digest, true);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700449 if (doesExist)
450 // Original comment (not sure): somebody else replied, so restart random-game timer
451 // YY: Get the same SyncInterest again, refresh the timer
452 m_scheduler.cancelEvent(m_delayedInterestProcessingId);
453
454 m_delayedInterestProcessingId =
455 m_scheduler.scheduleEvent(time::milliseconds(m_rangeUniformRandom()),
456 bind(&Logic::processSyncInterest, this, interest, true));
457 }
458 else {
459 // OK, nobody is helping us, just tell the truth.
Sonu Mishra4d3a2e02017-01-18 20:27:51 -0800460 _LOG_DEBUG_ID("OK, nobody is helping us, let us try to recover");
Yingdi Yuf7ede412014-08-30 20:37:52 -0700461 m_interestTable.erase(digest);
Sonu Mishra4d3a2e02017-01-18 20:27:51 -0800462 sendRecoveryInterest(digest);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700463 }
464
465 _LOG_DEBUG_ID("<< Logic::processSyncInterest");
466}
467
468void
469Logic::processResetInterest(const Interest& interest)
470{
471 _LOG_DEBUG_ID(">> Logic::processResetInterest");
Qiuhan Dingfb8c9e02015-01-30 14:04:55 -0800472 reset(true);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700473}
474
475void
476Logic::processSyncData(const Name& name,
477 ndn::ConstBufferPtr digest,
478 const Block& syncReplyBlock)
479{
480 _LOG_DEBUG_ID(">> Logic::processSyncData");
Yingdi Yuf7ede412014-08-30 20:37:52 -0700481 DiffStatePtr commit = make_shared<DiffState>();
482 ndn::ConstBufferPtr previousRoot = m_state.getRootDigest();
483
484 try {
485 m_interestTable.erase(digest); // Remove satisfied interest from PIT
486
487 State reply;
488 reply.wireDecode(syncReplyBlock);
489
490 std::vector<MissingDataInfo> v;
491 BOOST_FOREACH(ConstLeafPtr leaf, reply.getLeaves().get<ordered>())
492 {
493 BOOST_ASSERT(leaf != 0);
494
495 const Name& info = leaf->getSessionName();
496 SeqNo seq = leaf->getSeq();
497
498 bool isInserted = false;
499 bool isUpdated = false;
500 SeqNo oldSeq;
501 boost::tie(isInserted, isUpdated, oldSeq) = m_state.update(info, seq);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700502 if (isInserted || isUpdated) {
503 commit->update(info, seq);
504
505 oldSeq++;
506 MissingDataInfo mdi = {info, oldSeq, seq};
507 v.push_back(mdi);
508 }
509 }
510
511 if (!v.empty()) {
512 m_onUpdate(v);
513
514 commit->setRootDigest(m_state.getRootDigest());
515 insertToDiffLog(commit, previousRoot);
516 }
517 else {
518 _LOG_DEBUG_ID("What? nothing new");
519 }
520 }
521 catch (State::Error&) {
522 _LOG_DEBUG_ID("Something really fishy happened during state decoding");
523 // Something really fishy happened during state decoding;
524 commit.reset();
525 return;
526 }
527
528 if (static_cast<bool>(commit) && !commit->getLeaves().empty()) {
529 // state changed and it is safe to express a new interest
530 time::steady_clock::Duration after = time::milliseconds(m_reexpressionJitter());
531 _LOG_DEBUG_ID("Reschedule sync interest after: " << after);
532 EventId eventId = m_scheduler.scheduleEvent(after,
533 bind(&Logic::sendSyncInterest, this));
534
535 m_scheduler.cancelEvent(m_reexpressingInterestId);
536 m_reexpressingInterestId = eventId;
537 }
538}
539
540void
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800541Logic::satisfyPendingSyncInterests(const Name& updatedPrefix, ConstDiffStatePtr commit)
Yingdi Yuf7ede412014-08-30 20:37:52 -0700542{
543 _LOG_DEBUG_ID(">> Logic::satisfyPendingSyncInterests");
544 try {
545 _LOG_DEBUG_ID("InterestTable size: " << m_interestTable.size());
546 for (InterestTable::const_iterator it = m_interestTable.begin();
547 it != m_interestTable.end(); it++) {
548 ConstUnsatisfiedInterestPtr request = *it;
Yingdi Yuf7ede412014-08-30 20:37:52 -0700549 if (request->isUnknown)
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800550 sendSyncData(updatedPrefix, request->interest->getName(), m_state);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700551 else
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800552 sendSyncData(updatedPrefix, request->interest->getName(), *commit);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700553 }
554 m_interestTable.clear();
555 }
556 catch (InterestTable::Error&) {
557 // ok. not really an error
558 }
559 _LOG_DEBUG_ID("<< Logic::satisfyPendingSyncInterests");
560}
561
562void
563Logic::insertToDiffLog(DiffStatePtr commit, ndn::ConstBufferPtr previousRoot)
564{
565 _LOG_DEBUG_ID(">> Logic::insertToDiffLog");
566 // Connect to the history
567 if (!m_log.empty())
568 (*m_log.find(previousRoot))->setNext(commit);
569
570 // Insert the commit
571 m_log.erase(commit->getRootDigest());
572 m_log.insert(commit);
573 _LOG_DEBUG_ID("<< Logic::insertToDiffLog");
574}
575
576void
577Logic::sendResetInterest()
578{
579 _LOG_DEBUG_ID(">> Logic::sendResetInterest");
580
581 if (m_needPeriodReset) {
582 _LOG_DEBUG_ID("Need Period Reset");
583 _LOG_DEBUG_ID("ResetTimer: " << m_resetTimer);
584
585 EventId eventId =
586 m_scheduler.scheduleEvent(m_resetTimer + ndn::time::milliseconds(m_reexpressionJitter()),
587 bind(&Logic::sendResetInterest, this));
588 m_scheduler.cancelEvent(m_resetInterestId);
589 m_resetInterestId = eventId;
590 }
591
592 Interest interest(m_syncReset);
593 interest.setMustBeFresh(true);
594 interest.setInterestLifetime(m_resetInterestLifetime);
595 m_face.expressInterest(interest,
596 bind(&Logic::onResetData, this, _1, _2),
597 bind(&Logic::onSyncTimeout, this, _1));
598
599 _LOG_DEBUG_ID("<< Logic::sendResetInterest");
600}
601
602void
603Logic::sendSyncInterest()
604{
605 _LOG_DEBUG_ID(">> Logic::sendSyncInterest");
606
607 Name interestName;
608 interestName.append(m_syncPrefix)
609 .append(ndn::name::Component(*m_state.getRootDigest()));
610
611 m_outstandingInterestName = interestName;
612
613#ifdef _DEBUG
614 printDigest(m_state.getRootDigest());
615#endif
616
617 EventId eventId =
Sonu Mishra0dadc572016-12-12 23:59:41 -0800618 m_scheduler.scheduleEvent(m_syncInterestLifetime / 2 +
Yingdi Yuf7ede412014-08-30 20:37:52 -0700619 ndn::time::milliseconds(m_reexpressionJitter()),
620 bind(&Logic::sendSyncInterest, this));
621 m_scheduler.cancelEvent(m_reexpressingInterestId);
622 m_reexpressingInterestId = eventId;
623
624 Interest interest(interestName);
625 interest.setMustBeFresh(true);
626 interest.setInterestLifetime(m_syncInterestLifetime);
627
628 m_outstandingInterestId = m_face.expressInterest(interest,
629 bind(&Logic::onSyncData, this, _1, _2),
630 bind(&Logic::onSyncTimeout, this, _1));
631
632 _LOG_DEBUG_ID("Send interest: " << interest.getName());
633 _LOG_DEBUG_ID("<< Logic::sendSyncInterest");
634}
635
636void
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800637Logic::sendSyncData(const Name& nodePrefix, const Name& name, const State& state)
Yingdi Yuf7ede412014-08-30 20:37:52 -0700638{
639 _LOG_DEBUG_ID(">> Logic::sendSyncData");
640 shared_ptr<Data> syncReply = make_shared<Data>(name);
641 syncReply->setContent(state.wireEncode());
642 syncReply->setFreshnessPeriod(m_syncReplyFreshness);
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800643 if (m_nodeList.find(nodePrefix) == m_nodeList.end())
644 return;
645 if (m_nodeList[nodePrefix].signingId.empty())
Yingdi Yucd339022014-11-05 17:51:19 -0800646 m_keyChain.sign(*syncReply);
647 else
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800648 m_keyChain.signByIdentity(*syncReply, m_nodeList[nodePrefix].signingId);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700649
650 m_face.put(*syncReply);
651
652 // checking if our own interest got satisfied
653 if (m_outstandingInterestName == name) {
654 // remove outstanding interest
655 if (m_outstandingInterestId != 0) {
656 m_face.removePendingInterest(m_outstandingInterestId);
657 m_outstandingInterestId = 0;
658 }
659
660 // re-schedule sending Sync interest
661 time::milliseconds after(m_reexpressionJitter());
662 _LOG_DEBUG_ID("Satisfy our own interest");
663 _LOG_DEBUG_ID("Reschedule sync interest after " << after);
664 EventId eventId = m_scheduler.scheduleEvent(after, bind(&Logic::sendSyncInterest, this));
665 m_scheduler.cancelEvent(m_reexpressingInterestId);
666 m_reexpressingInterestId = eventId;
667 }
668 _LOG_DEBUG_ID("<< Logic::sendSyncData");
669}
670
671void
672Logic::cancelReset()
673{
674 _LOG_DEBUG_ID(">> Logic::cancelReset");
675 if (!m_isInReset)
676 return;
677
678 m_isInReset = false;
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800679 for (const auto& node : m_nodeList) {
680 updateSeqNo(node.second.seqNo, node.first);
681 }
Yingdi Yuf7ede412014-08-30 20:37:52 -0700682 _LOG_DEBUG_ID("<< Logic::cancelReset");
683}
684
685void
686Logic::printDigest(ndn::ConstBufferPtr digest)
687{
688 using namespace CryptoPP;
689
690 std::string hash;
691 StringSource(digest->buf(), digest->size(), true,
692 new HexEncoder(new StringSink(hash), false));
693 _LOG_DEBUG_ID("Hash: " << hash);
694}
695
Sonu Mishra4d3a2e02017-01-18 20:27:51 -0800696void
697Logic::sendRecoveryInterest(ndn::ConstBufferPtr digest)
698{
699 _LOG_DEBUG_ID(">> Logic::sendRecoveryInterest");
700
701 Name interestName;
702 interestName.append(m_syncPrefix)
703 .append(RECOVERY_COMPONENT)
704 .append(ndn::name::Component(*digest));
705
706 Interest interest(interestName);
707 interest.setMustBeFresh(true);
708 interest.setInterestLifetime(m_recoveryInterestLifetime);
709
710 m_face.expressInterest(interest, bind(&Logic::onRecoveryData, this, _1, _2),
711 bind(&Logic::onRecoveryTimeout, this, _1));
712
713 _LOG_DEBUG_ID("interest: " << interest.getName());
714 _LOG_DEBUG_ID("<< Logic::sendRecoveryInterest");
715}
716
717void
718Logic::processRecoveryInterest(const Interest& interest)
719{
720 _LOG_DEBUG_ID(">> Logic::processRecoveryInterest");
721
722 const Name& name = interest.getName();
723 ConstBufferPtr digest = make_shared<ndn::Buffer>(name.get(-1).value(), name.get(-1).value_size());
724
725 ConstBufferPtr rootDigest = m_state.getRootDigest();
726
727 DiffStateContainer::iterator stateIter = m_log.find(digest);
728
729 if (stateIter != m_log.end() || *digest == *EMPTY_DIGEST || *rootDigest == *digest) {
730 _LOG_DEBUG_ID("I can help you recover");
731 sendSyncData(m_defaultUserPrefix, name, m_state);
732 return;
733 }
734 _LOG_DEBUG_ID("<< Logic::processRecoveryInterest");
735}
736
737void
738Logic::onRecoveryData(const Interest& interest, Data& data)
739{
740 _LOG_DEBUG_ID(">> Logic::onRecoveryData");
741 onSyncDataValidated(data.shared_from_this());
742 _LOG_DEBUG_ID("<< Logic::onRecoveryData");
743}
744
745void
746Logic::onRecoveryTimeout(const Interest& interest)
747{
748 _LOG_DEBUG_ID(">> Logic::onRecoveryTimeout");
749 _LOG_DEBUG_ID("Interest: " << interest.getName());
750 _LOG_DEBUG_ID("<< Logic::onRecoveryTimeout");
751}
752
753} // namespace chronosync