blob: df0fcd14a03347fdc17be635e011d97271921ca2 [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
Ashlesh Gawande687cf922017-05-30 15:04:16 -050029#include <ndn-cxx/util/string-helper.hpp>
30
Alexander Afanasyev36eb3ed2017-01-11 12:35:58 -080031INIT_LOGGER(Logic);
Yingdi Yuf7ede412014-08-30 20:37:52 -070032
33#ifdef _DEBUG
34#define _LOG_DEBUG_ID(v) _LOG_DEBUG("Instance" << m_instanceId << ": " << v)
35#else
36#define _LOG_DEBUG_ID(v) _LOG_DEBUG(v)
37#endif
38
39namespace chronosync {
40
41using ndn::ConstBufferPtr;
42using ndn::EventId;
43
44const uint8_t EMPTY_DIGEST_VALUE[] = {
45 0xe3, 0xb0, 0xc4, 0x42, 0x98, 0xfc, 0x1c, 0x14,
46 0x9a, 0xfb, 0xf4, 0xc8, 0x99, 0x6f, 0xb9, 0x24,
47 0x27, 0xae, 0x41, 0xe4, 0x64, 0x9b, 0x93, 0x4c,
48 0xa4, 0x95, 0x99, 0x1b, 0x78, 0x52, 0xb8, 0x55
49};
50
51#ifdef _DEBUG
52int Logic::m_instanceCounter = 0;
53#endif
54
Yingdi Yucd339022014-11-05 17:51:19 -080055const ndn::Name Logic::DEFAULT_NAME;
Qiuhan Ding8c095fd2014-11-19 17:38:32 -080056const ndn::Name Logic::EMPTY_NAME;
Alexander Afanasyeve9eda8a2017-03-09 14:40:03 -080057const std::shared_ptr<ndn::Validator> Logic::DEFAULT_VALIDATOR;
Yingdi Yuf7ede412014-08-30 20:37:52 -070058const time::steady_clock::Duration Logic::DEFAULT_RESET_TIMER = time::seconds(0);
59const time::steady_clock::Duration Logic::DEFAULT_CANCEL_RESET_TIMER = time::milliseconds(500);
60const time::milliseconds Logic::DEFAULT_RESET_INTEREST_LIFETIME(1000);
61const time::milliseconds Logic::DEFAULT_SYNC_INTEREST_LIFETIME(1000);
62const time::milliseconds Logic::DEFAULT_SYNC_REPLY_FRESHNESS(1000);
Sonu Mishra4d3a2e02017-01-18 20:27:51 -080063const time::milliseconds Logic::DEFAULT_RECOVERY_INTEREST_LIFETIME(1000);
Yingdi Yuf7ede412014-08-30 20:37:52 -070064
65const ndn::ConstBufferPtr Logic::EMPTY_DIGEST(new ndn::Buffer(EMPTY_DIGEST_VALUE, 32));
66const ndn::name::Component Logic::RESET_COMPONENT("reset");
Sonu Mishra4d3a2e02017-01-18 20:27:51 -080067const ndn::name::Component Logic::RECOVERY_COMPONENT("recovery");
Yingdi Yuf7ede412014-08-30 20:37:52 -070068
69Logic::Logic(ndn::Face& face,
70 const Name& syncPrefix,
Qiuhan Ding8c095fd2014-11-19 17:38:32 -080071 const Name& defaultUserPrefix,
Yingdi Yuf7ede412014-08-30 20:37:52 -070072 const UpdateCallback& onUpdate,
Qiuhan Ding8c095fd2014-11-19 17:38:32 -080073 const Name& defaultSigningId,
Alexander Afanasyeve9eda8a2017-03-09 14:40:03 -080074 std::shared_ptr<ndn::Validator> validator,
Yingdi Yuf7ede412014-08-30 20:37:52 -070075 const time::steady_clock::Duration& resetTimer,
76 const time::steady_clock::Duration& cancelResetTimer,
77 const time::milliseconds& resetInterestLifetime,
78 const time::milliseconds& syncInterestLifetime,
Sonu Mishra4d3a2e02017-01-18 20:27:51 -080079 const time::milliseconds& syncReplyFreshness,
80 const time::milliseconds& recoveryInterestLifetime)
Yingdi Yuf7ede412014-08-30 20:37:52 -070081 : m_face(face)
82 , m_syncPrefix(syncPrefix)
Qiuhan Ding8c095fd2014-11-19 17:38:32 -080083 , m_defaultUserPrefix(defaultUserPrefix)
Yingdi Yuf7ede412014-08-30 20:37:52 -070084 , m_interestTable(m_face.getIoService())
85 , m_outstandingInterestId(0)
86 , m_isInReset(false)
87 , m_needPeriodReset(resetTimer > time::steady_clock::Duration::zero())
88 , m_onUpdate(onUpdate)
89 , m_scheduler(m_face.getIoService())
90 , m_randomGenerator(static_cast<unsigned int>(std::time(0)))
91 , m_rangeUniformRandom(m_randomGenerator, boost::uniform_int<>(100,500))
92 , m_reexpressionJitter(m_randomGenerator, boost::uniform_int<>(100,500))
93 , m_resetTimer(resetTimer)
94 , m_cancelResetTimer(cancelResetTimer)
95 , m_resetInterestLifetime(resetInterestLifetime)
96 , m_syncInterestLifetime(syncInterestLifetime)
97 , m_syncReplyFreshness(syncReplyFreshness)
Sonu Mishra4d3a2e02017-01-18 20:27:51 -080098 , m_recoveryInterestLifetime(recoveryInterestLifetime)
Yingdi Yucd339022014-11-05 17:51:19 -080099 , m_validator(validator)
Yingdi Yuf7ede412014-08-30 20:37:52 -0700100{
101#ifdef _DEBUG
102 m_instanceId = m_instanceCounter++;
103#endif
104
105 _LOG_DEBUG_ID(">> Logic::Logic");
106
Ashlesh Gawande687cf922017-05-30 15:04:16 -0500107 addUserNode(m_defaultUserPrefix, defaultSigningId);
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800108
109
Yingdi Yuf7ede412014-08-30 20:37:52 -0700110 m_syncReset = m_syncPrefix;
111 m_syncReset.append("reset");
112
113 _LOG_DEBUG_ID("Listen to: " << m_syncPrefix);
114 m_syncRegisteredPrefixId =
115 m_face.setInterestFilter(m_syncPrefix,
116 bind(&Logic::onSyncInterest, this, _1, _2),
117 bind(&Logic::onSyncRegisterFailed, this, _1, _2));
118
Qiuhan Dinge246b622014-12-03 21:57:48 -0800119 sendSyncInterest();
Yingdi Yuf7ede412014-08-30 20:37:52 -0700120 _LOG_DEBUG_ID("<< Logic::Logic");
121}
122
123Logic::~Logic()
124{
Yingdi Yuf7ede412014-08-30 20:37:52 -0700125 m_scheduler.cancelAllEvents();
Yingdi Yu9d5679a2015-02-01 00:17:58 -0800126 m_interestTable.clear();
127 m_face.shutdown();
Yingdi Yuf7ede412014-08-30 20:37:52 -0700128}
129
130void
Qiuhan Dingfb8c9e02015-01-30 14:04:55 -0800131Logic::reset(bool isOnInterest)
Yingdi Yuf7ede412014-08-30 20:37:52 -0700132{
133 m_isInReset = true;
134
135 m_state.reset();
136 m_log.clear();
137
Qiuhan Dingfb8c9e02015-01-30 14:04:55 -0800138 if (!isOnInterest)
139 sendResetInterest();
Yingdi Yuf7ede412014-08-30 20:37:52 -0700140
141 // reset outstanding interest name, so that data for previous interest will be dropped.
142 if (m_outstandingInterestId != 0) {
143 m_face.removePendingInterest(m_outstandingInterestId);
144 m_outstandingInterestId = 0;
145 }
146
147 sendSyncInterest();
148
149 if (static_cast<bool>(m_delayedInterestProcessingId))
150 m_scheduler.cancelEvent(m_delayedInterestProcessingId);
151
152 m_delayedInterestProcessingId =
153 m_scheduler.scheduleEvent(m_cancelResetTimer,
154 bind(&Logic::cancelReset, this));
155}
156
157void
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800158Logic::setDefaultUserPrefix(const Name& defaultUserPrefix)
Yingdi Yuf7ede412014-08-30 20:37:52 -0700159{
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800160 if (defaultUserPrefix != EMPTY_NAME) {
161 if (m_nodeList.find(defaultUserPrefix) != m_nodeList.end()) {
162 m_defaultUserPrefix = defaultUserPrefix;
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800163 }
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;
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800174 }
175 if (m_nodeList.find(userPrefix) == m_nodeList.end()) {
176 m_nodeList[userPrefix].userPrefix = userPrefix;
177 m_nodeList[userPrefix].signingId = signingId;
178 Name sessionName = userPrefix;
179 sessionName.appendNumber(ndn::time::toUnixTimestamp(ndn::time::system_clock::now()).count());
180 m_nodeList[userPrefix].sessionName = sessionName;
181 m_nodeList[userPrefix].seqNo = 0;
Qiuhan Dingfb8c9e02015-01-30 14:04:55 -0800182 reset(false);
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800183 }
184}
Yingdi Yuf7ede412014-08-30 20:37:52 -0700185
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800186void
187Logic::removeUserNode(const Name& userPrefix)
188{
189 auto userNode = m_nodeList.find(userPrefix);
190 if (userNode != m_nodeList.end()) {
191 m_nodeList.erase(userNode);
192 if (m_defaultUserPrefix == userPrefix) {
193 if (!m_nodeList.empty()) {
194 m_defaultUserPrefix = m_nodeList.begin()->second.userPrefix;
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800195 }
196 else {
197 m_defaultUserPrefix = EMPTY_NAME;
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800198 }
Yingdi Yuf7ede412014-08-30 20:37:52 -0700199 }
Qiuhan Dingfb8c9e02015-01-30 14:04:55 -0800200 reset(false);
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800201 }
202}
Yingdi Yuf7ede412014-08-30 20:37:52 -0700203
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800204const Name&
205Logic::getSessionName(Name prefix)
206{
207 if (prefix == EMPTY_NAME)
208 prefix = m_defaultUserPrefix;
209 auto node = m_nodeList.find(prefix);
210 if (node != m_nodeList.end())
211 return node->second.sessionName;
212 else
Alexander Afanasyeve9eda8a2017-03-09 14:40:03 -0800213 BOOST_THROW_EXCEPTION(Error("Refer to non-existent node:" + prefix.toUri()));
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800214}
Yingdi Yuf7ede412014-08-30 20:37:52 -0700215
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800216const SeqNo&
217Logic::getSeqNo(Name prefix)
218{
219 if (prefix == EMPTY_NAME)
220 prefix = m_defaultUserPrefix;
221 auto node = m_nodeList.find(prefix);
222 if (node != m_nodeList.end())
223 return node->second.seqNo;
224 else
Alexander Afanasyeve9eda8a2017-03-09 14:40:03 -0800225 BOOST_THROW_EXCEPTION(Logic::Error("Refer to non-existent node:" + prefix.toUri()));
Yingdi Yuf7ede412014-08-30 20:37:52 -0700226
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800227}
228
229void
Alexander Afanasyeve9eda8a2017-03-09 14:40:03 -0800230Logic::updateSeqNo(const SeqNo& seqNo, const Name& updatePrefix)
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800231{
232 Name prefix;
233 if (updatePrefix == EMPTY_NAME) {
234 if (m_defaultUserPrefix == EMPTY_NAME)
235 return;
236 prefix = m_defaultUserPrefix;
237 }
238 else
239 prefix = updatePrefix;
240
241 auto it = m_nodeList.find(prefix);
242 if (it != m_nodeList.end()) {
243 NodeInfo& node = it->second;
244 _LOG_DEBUG_ID(">> Logic::updateSeqNo");
245 _LOG_DEBUG_ID("seqNo: " << seqNo << " m_seqNo: " << node.seqNo);
246 if (seqNo < node.seqNo || seqNo == 0)
247 return;
248
249 node.seqNo = seqNo;
250 _LOG_DEBUG_ID("updateSeqNo: m_seqNo " << node.seqNo);
251
252 if (!m_isInReset) {
253 _LOG_DEBUG_ID("updateSeqNo: not in Reset ");
254 ndn::ConstBufferPtr previousRoot = m_state.getRootDigest();
255 {
Ashlesh Gawande687cf922017-05-30 15:04:16 -0500256 std::string hash = ndn::toHex(previousRoot->buf(), previousRoot->size(), false);
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800257 _LOG_DEBUG_ID("Hash: " << hash);
258 }
259
260 bool isInserted = false;
261 bool isUpdated = false;
262 SeqNo oldSeq;
263 boost::tie(isInserted, isUpdated, oldSeq) = m_state.update(node.sessionName,
264 node.seqNo);
265
266 _LOG_DEBUG_ID("Insert: " << std::boolalpha << isInserted);
267 _LOG_DEBUG_ID("Updated: " << std::boolalpha << isUpdated);
268 if (isInserted || isUpdated) {
269 DiffStatePtr commit = make_shared<DiffState>();
270 commit->update(node.sessionName, node.seqNo);
271 commit->setRootDigest(m_state.getRootDigest());
272 insertToDiffLog(commit, previousRoot);
273
274 satisfyPendingSyncInterests(prefix, commit);
Sonu Mishraf42aa2c2017-01-22 18:47:33 -0800275 formAndSendExcludeInterest(prefix, *commit, previousRoot);
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800276 }
Yingdi Yuf7ede412014-08-30 20:37:52 -0700277 }
278 }
279}
280
281ConstBufferPtr
282Logic::getRootDigest() const
283{
284 return m_state.getRootDigest();
285}
286
287void
288Logic::printState(std::ostream& os) const
289{
290 BOOST_FOREACH(ConstLeafPtr leaf, m_state.getLeaves())
291 {
292 os << *leaf << "\n";
293 }
294}
295
296std::set<Name>
297Logic::getSessionNames() const
298{
299 std::set<Name> sessionNames;
300
301 BOOST_FOREACH(ConstLeafPtr leaf, m_state.getLeaves())
302 {
303 sessionNames.insert(leaf->getSessionName());
304 }
305
306 return sessionNames;
307}
308
309void
310Logic::onSyncInterest(const Name& prefix, const Interest& interest)
311{
312 _LOG_DEBUG_ID(">> Logic::onSyncInterest");
313 Name name = interest.getName();
314
315 _LOG_DEBUG_ID("InterestName: " << name);
316
Sonu Mishra4d3a2e02017-01-18 20:27:51 -0800317 if (name.size() >= 1 && RESET_COMPONENT == name.get(-1)) {
318 processResetInterest(interest);
319 }
320 else if (name.size() >= 2 && RECOVERY_COMPONENT == name.get(-2)) {
321 processRecoveryInterest(interest);
322 }
323 else {
Yingdi Yuf7ede412014-08-30 20:37:52 -0700324 processSyncInterest(interest.shared_from_this());
325 }
Yingdi Yuf7ede412014-08-30 20:37:52 -0700326
327 _LOG_DEBUG_ID("<< Logic::onSyncInterest");
328}
329
330void
331Logic::onSyncRegisterFailed(const Name& prefix, const std::string& msg)
332{
333 //Sync prefix registration failed
334 _LOG_DEBUG_ID(">> Logic::onSyncRegisterFailed");
335}
336
337void
Alexander Afanasyeve9eda8a2017-03-09 14:40:03 -0800338Logic::onSyncData(const Interest& interest, const Data& data)
Yingdi Yuf7ede412014-08-30 20:37:52 -0700339{
340 _LOG_DEBUG_ID(">> Logic::onSyncData");
Sonu Mishraf42aa2c2017-01-22 18:47:33 -0800341 // if (static_cast<bool>(m_validator))
342 // m_validator->validate(data,
343 // bind(&Logic::onSyncDataValidated, this, _1),
344 // bind(&Logic::onSyncDataValidationFailed, this, _1));
345 // else
346 // onSyncDataValidated(data.shared_from_this());
347
348 if (interest.getExclude().empty()) {
349 _LOG_DEBUG_ID("First data");
Yingdi Yucd339022014-11-05 17:51:19 -0800350 onSyncDataValidated(data.shared_from_this());
Sonu Mishraf42aa2c2017-01-22 18:47:33 -0800351 }
352 else {
353 _LOG_DEBUG_ID("Data obtained using exclude filter");
354 onSyncDataValidated(data.shared_from_this(), false);
355 }
356 sendExcludeInterest(interest, data);
357
Yingdi Yuf7ede412014-08-30 20:37:52 -0700358 _LOG_DEBUG_ID("<< Logic::onSyncData");
359}
360
361void
Alexander Afanasyeve9eda8a2017-03-09 14:40:03 -0800362Logic::onResetData(const Interest& interest, const Data& data)
Yingdi Yuf7ede412014-08-30 20:37:52 -0700363{
364 // This should not happened, drop the received data.
365}
366
367void
368Logic::onSyncTimeout(const Interest& interest)
369{
370 // It is OK. Others will handle the time out situation.
371 _LOG_DEBUG_ID(">> Logic::onSyncTimeout");
372 _LOG_DEBUG_ID("Interest: " << interest.getName());
373 _LOG_DEBUG_ID("<< Logic::onSyncTimeout");
374}
375
376void
377Logic::onSyncDataValidationFailed(const shared_ptr<const Data>& data)
378{
379 // SyncReply cannot be validated.
380}
381
382void
Sonu Mishraf42aa2c2017-01-22 18:47:33 -0800383Logic::onSyncDataValidated(const shared_ptr<const Data>& data, bool firstData)
Yingdi Yuf7ede412014-08-30 20:37:52 -0700384{
385 Name name = data->getName();
386 ConstBufferPtr digest = make_shared<ndn::Buffer>(name.get(-1).value(), name.get(-1).value_size());
387
Sonu Mishraf42aa2c2017-01-22 18:47:33 -0800388 processSyncData(name, digest, data->getContent().blockFromValue(), firstData);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700389}
390
391void
392Logic::processSyncInterest(const shared_ptr<const Interest>& interest,
393 bool isTimedProcessing/*=false*/)
394{
395 _LOG_DEBUG_ID(">> Logic::processSyncInterest");
396
397 const Name& name = interest->getName();
398 ConstBufferPtr digest =
399 make_shared<ndn::Buffer>(name.get(-1).value(), name.get(-1).value_size());
400
401 ConstBufferPtr rootDigest = m_state.getRootDigest();
402
403 // If the digest of the incoming interest is the same as root digest
404 // Put the interest into InterestTable
405 if (*rootDigest == *digest) {
406 _LOG_DEBUG_ID("Oh, we are in the same state");
407 m_interestTable.insert(interest, digest, false);
408
409 if (!m_isInReset)
410 return;
411
412 if (!isTimedProcessing) {
413 _LOG_DEBUG_ID("Non timed processing in reset");
414 // Still in reset, our own seq has not been put into state yet
415 // Do not hurry, some others may be also resetting and may send their reply
416 if (static_cast<bool>(m_delayedInterestProcessingId))
417 m_scheduler.cancelEvent(m_delayedInterestProcessingId);
418
419 time::milliseconds after(m_rangeUniformRandom());
420 _LOG_DEBUG_ID("After: " << after);
421 m_delayedInterestProcessingId =
422 m_scheduler.scheduleEvent(after,
423 bind(&Logic::processSyncInterest, this, interest, true));
424 }
425 else {
426 _LOG_DEBUG_ID("Timed processing in reset");
427 // Now we can get out of reset state by putting our own stuff into m_state.
428 cancelReset();
429 }
430
431 return;
432 }
433
434 // If the digest of incoming interest is an "empty" digest
Sonu Mishrae10acbc2017-01-18 14:14:05 -0800435 if (*digest == *EMPTY_DIGEST) {
Yingdi Yuf7ede412014-08-30 20:37:52 -0700436 _LOG_DEBUG_ID("Poor guy, he knows nothing");
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800437 sendSyncData(m_defaultUserPrefix, name, m_state);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700438 return;
439 }
440
441 DiffStateContainer::iterator stateIter = m_log.find(digest);
442 // If the digest of incoming interest can be found from the log
443 if (stateIter != m_log.end()) {
444 _LOG_DEBUG_ID("It is ok, you are so close");
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800445 sendSyncData(m_defaultUserPrefix, name, *(*stateIter)->diff());
Yingdi Yuf7ede412014-08-30 20:37:52 -0700446 return;
447 }
448
449 if (!isTimedProcessing) {
450 _LOG_DEBUG_ID("Let's wait, just wait for a while");
451 // Do not hurry, some incoming SyncReplies may help us to recognize the digest
Yingdi Yu53f5f042015-01-31 16:33:25 -0800452 bool doesExist = m_interestTable.has(digest);
453 m_interestTable.insert(interest, digest, true);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700454 if (doesExist)
455 // Original comment (not sure): somebody else replied, so restart random-game timer
456 // YY: Get the same SyncInterest again, refresh the timer
457 m_scheduler.cancelEvent(m_delayedInterestProcessingId);
458
459 m_delayedInterestProcessingId =
460 m_scheduler.scheduleEvent(time::milliseconds(m_rangeUniformRandom()),
461 bind(&Logic::processSyncInterest, this, interest, true));
462 }
463 else {
464 // OK, nobody is helping us, just tell the truth.
Sonu Mishra4d3a2e02017-01-18 20:27:51 -0800465 _LOG_DEBUG_ID("OK, nobody is helping us, let us try to recover");
Yingdi Yuf7ede412014-08-30 20:37:52 -0700466 m_interestTable.erase(digest);
Sonu Mishra4d3a2e02017-01-18 20:27:51 -0800467 sendRecoveryInterest(digest);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700468 }
469
470 _LOG_DEBUG_ID("<< Logic::processSyncInterest");
471}
472
473void
474Logic::processResetInterest(const Interest& interest)
475{
476 _LOG_DEBUG_ID(">> Logic::processResetInterest");
Qiuhan Dingfb8c9e02015-01-30 14:04:55 -0800477 reset(true);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700478}
479
480void
481Logic::processSyncData(const Name& name,
482 ndn::ConstBufferPtr digest,
Sonu Mishraf42aa2c2017-01-22 18:47:33 -0800483 const Block& syncReplyBlock,
484 bool firstData)
Yingdi Yuf7ede412014-08-30 20:37:52 -0700485{
486 _LOG_DEBUG_ID(">> Logic::processSyncData");
Yingdi Yuf7ede412014-08-30 20:37:52 -0700487 DiffStatePtr commit = make_shared<DiffState>();
488 ndn::ConstBufferPtr previousRoot = m_state.getRootDigest();
489
490 try {
491 m_interestTable.erase(digest); // Remove satisfied interest from PIT
492
493 State reply;
494 reply.wireDecode(syncReplyBlock);
495
496 std::vector<MissingDataInfo> v;
497 BOOST_FOREACH(ConstLeafPtr leaf, reply.getLeaves().get<ordered>())
498 {
499 BOOST_ASSERT(leaf != 0);
500
501 const Name& info = leaf->getSessionName();
502 SeqNo seq = leaf->getSeq();
503
504 bool isInserted = false;
505 bool isUpdated = false;
506 SeqNo oldSeq;
507 boost::tie(isInserted, isUpdated, oldSeq) = m_state.update(info, seq);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700508 if (isInserted || isUpdated) {
509 commit->update(info, seq);
510
511 oldSeq++;
512 MissingDataInfo mdi = {info, oldSeq, seq};
513 v.push_back(mdi);
514 }
515 }
516
517 if (!v.empty()) {
518 m_onUpdate(v);
519
520 commit->setRootDigest(m_state.getRootDigest());
521 insertToDiffLog(commit, previousRoot);
522 }
523 else {
524 _LOG_DEBUG_ID("What? nothing new");
525 }
526 }
Alexander Afanasyeve9eda8a2017-03-09 14:40:03 -0800527 catch (const State::Error&) {
Yingdi Yuf7ede412014-08-30 20:37:52 -0700528 _LOG_DEBUG_ID("Something really fishy happened during state decoding");
529 // Something really fishy happened during state decoding;
530 commit.reset();
531 return;
532 }
533
Sonu Mishraf42aa2c2017-01-22 18:47:33 -0800534 if (static_cast<bool>(commit) && !commit->getLeaves().empty() && firstData) {
Yingdi Yuf7ede412014-08-30 20:37:52 -0700535 // state changed and it is safe to express a new interest
536 time::steady_clock::Duration after = time::milliseconds(m_reexpressionJitter());
537 _LOG_DEBUG_ID("Reschedule sync interest after: " << after);
538 EventId eventId = m_scheduler.scheduleEvent(after,
539 bind(&Logic::sendSyncInterest, this));
540
541 m_scheduler.cancelEvent(m_reexpressingInterestId);
542 m_reexpressingInterestId = eventId;
543 }
544}
545
546void
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800547Logic::satisfyPendingSyncInterests(const Name& updatedPrefix, ConstDiffStatePtr commit)
Yingdi Yuf7ede412014-08-30 20:37:52 -0700548{
549 _LOG_DEBUG_ID(">> Logic::satisfyPendingSyncInterests");
550 try {
551 _LOG_DEBUG_ID("InterestTable size: " << m_interestTable.size());
552 for (InterestTable::const_iterator it = m_interestTable.begin();
553 it != m_interestTable.end(); it++) {
554 ConstUnsatisfiedInterestPtr request = *it;
Yingdi Yuf7ede412014-08-30 20:37:52 -0700555 if (request->isUnknown)
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800556 sendSyncData(updatedPrefix, request->interest->getName(), m_state);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700557 else
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800558 sendSyncData(updatedPrefix, request->interest->getName(), *commit);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700559 }
560 m_interestTable.clear();
561 }
Alexander Afanasyeve9eda8a2017-03-09 14:40:03 -0800562 catch (const InterestTable::Error&) {
Yingdi Yuf7ede412014-08-30 20:37:52 -0700563 // ok. not really an error
564 }
565 _LOG_DEBUG_ID("<< Logic::satisfyPendingSyncInterests");
566}
567
568void
569Logic::insertToDiffLog(DiffStatePtr commit, ndn::ConstBufferPtr previousRoot)
570{
571 _LOG_DEBUG_ID(">> Logic::insertToDiffLog");
572 // Connect to the history
573 if (!m_log.empty())
574 (*m_log.find(previousRoot))->setNext(commit);
575
576 // Insert the commit
577 m_log.erase(commit->getRootDigest());
578 m_log.insert(commit);
579 _LOG_DEBUG_ID("<< Logic::insertToDiffLog");
580}
581
582void
583Logic::sendResetInterest()
584{
585 _LOG_DEBUG_ID(">> Logic::sendResetInterest");
586
587 if (m_needPeriodReset) {
588 _LOG_DEBUG_ID("Need Period Reset");
589 _LOG_DEBUG_ID("ResetTimer: " << m_resetTimer);
590
591 EventId eventId =
592 m_scheduler.scheduleEvent(m_resetTimer + ndn::time::milliseconds(m_reexpressionJitter()),
593 bind(&Logic::sendResetInterest, this));
594 m_scheduler.cancelEvent(m_resetInterestId);
595 m_resetInterestId = eventId;
596 }
597
598 Interest interest(m_syncReset);
599 interest.setMustBeFresh(true);
600 interest.setInterestLifetime(m_resetInterestLifetime);
601 m_face.expressInterest(interest,
602 bind(&Logic::onResetData, this, _1, _2),
Alexander Afanasyeve9eda8a2017-03-09 14:40:03 -0800603 bind(&Logic::onSyncTimeout, this, _1), // Nack
Yingdi Yuf7ede412014-08-30 20:37:52 -0700604 bind(&Logic::onSyncTimeout, this, _1));
605
606 _LOG_DEBUG_ID("<< Logic::sendResetInterest");
607}
608
609void
610Logic::sendSyncInterest()
611{
612 _LOG_DEBUG_ID(">> Logic::sendSyncInterest");
613
614 Name interestName;
615 interestName.append(m_syncPrefix)
616 .append(ndn::name::Component(*m_state.getRootDigest()));
617
618 m_outstandingInterestName = interestName;
619
620#ifdef _DEBUG
621 printDigest(m_state.getRootDigest());
622#endif
623
624 EventId eventId =
Sonu Mishra0dadc572016-12-12 23:59:41 -0800625 m_scheduler.scheduleEvent(m_syncInterestLifetime / 2 +
Yingdi Yuf7ede412014-08-30 20:37:52 -0700626 ndn::time::milliseconds(m_reexpressionJitter()),
627 bind(&Logic::sendSyncInterest, this));
628 m_scheduler.cancelEvent(m_reexpressingInterestId);
629 m_reexpressingInterestId = eventId;
630
631 Interest interest(interestName);
632 interest.setMustBeFresh(true);
633 interest.setInterestLifetime(m_syncInterestLifetime);
634
635 m_outstandingInterestId = m_face.expressInterest(interest,
636 bind(&Logic::onSyncData, this, _1, _2),
Alexander Afanasyeve9eda8a2017-03-09 14:40:03 -0800637 bind(&Logic::onSyncTimeout, this, _1), // Nack
Yingdi Yuf7ede412014-08-30 20:37:52 -0700638 bind(&Logic::onSyncTimeout, this, _1));
639
640 _LOG_DEBUG_ID("Send interest: " << interest.getName());
641 _LOG_DEBUG_ID("<< Logic::sendSyncInterest");
642}
643
644void
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800645Logic::sendSyncData(const Name& nodePrefix, const Name& name, const State& state)
Yingdi Yuf7ede412014-08-30 20:37:52 -0700646{
647 _LOG_DEBUG_ID(">> Logic::sendSyncData");
648 shared_ptr<Data> syncReply = make_shared<Data>(name);
649 syncReply->setContent(state.wireEncode());
650 syncReply->setFreshnessPeriod(m_syncReplyFreshness);
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800651 if (m_nodeList.find(nodePrefix) == m_nodeList.end())
652 return;
653 if (m_nodeList[nodePrefix].signingId.empty())
Yingdi Yucd339022014-11-05 17:51:19 -0800654 m_keyChain.sign(*syncReply);
655 else
Ashlesh Gawande687cf922017-05-30 15:04:16 -0500656 m_keyChain.sign(*syncReply, security::signingByIdentity(m_nodeList[nodePrefix].signingId));
Yingdi Yuf7ede412014-08-30 20:37:52 -0700657
658 m_face.put(*syncReply);
659
660 // checking if our own interest got satisfied
661 if (m_outstandingInterestName == name) {
662 // remove outstanding interest
663 if (m_outstandingInterestId != 0) {
664 m_face.removePendingInterest(m_outstandingInterestId);
665 m_outstandingInterestId = 0;
666 }
667
668 // re-schedule sending Sync interest
669 time::milliseconds after(m_reexpressionJitter());
670 _LOG_DEBUG_ID("Satisfy our own interest");
671 _LOG_DEBUG_ID("Reschedule sync interest after " << after);
672 EventId eventId = m_scheduler.scheduleEvent(after, bind(&Logic::sendSyncInterest, this));
673 m_scheduler.cancelEvent(m_reexpressingInterestId);
674 m_reexpressingInterestId = eventId;
675 }
676 _LOG_DEBUG_ID("<< Logic::sendSyncData");
677}
678
679void
680Logic::cancelReset()
681{
682 _LOG_DEBUG_ID(">> Logic::cancelReset");
683 if (!m_isInReset)
684 return;
685
686 m_isInReset = false;
Qiuhan Ding8c095fd2014-11-19 17:38:32 -0800687 for (const auto& node : m_nodeList) {
688 updateSeqNo(node.second.seqNo, node.first);
689 }
Yingdi Yuf7ede412014-08-30 20:37:52 -0700690 _LOG_DEBUG_ID("<< Logic::cancelReset");
691}
692
693void
694Logic::printDigest(ndn::ConstBufferPtr digest)
695{
Ashlesh Gawande687cf922017-05-30 15:04:16 -0500696 std::string hash = ndn::toHex(digest->buf(), digest->size(), false);
Yingdi Yuf7ede412014-08-30 20:37:52 -0700697 _LOG_DEBUG_ID("Hash: " << hash);
698}
699
Sonu Mishra4d3a2e02017-01-18 20:27:51 -0800700void
701Logic::sendRecoveryInterest(ndn::ConstBufferPtr digest)
702{
703 _LOG_DEBUG_ID(">> Logic::sendRecoveryInterest");
704
705 Name interestName;
706 interestName.append(m_syncPrefix)
707 .append(RECOVERY_COMPONENT)
708 .append(ndn::name::Component(*digest));
709
710 Interest interest(interestName);
711 interest.setMustBeFresh(true);
712 interest.setInterestLifetime(m_recoveryInterestLifetime);
713
Alexander Afanasyeve9eda8a2017-03-09 14:40:03 -0800714 m_face.expressInterest(interest,
715 bind(&Logic::onRecoveryData, this, _1, _2),
716 bind(&Logic::onRecoveryTimeout, this, _1), // Nack
717 bind(&Logic::onRecoveryTimeout, this, _1));
Sonu Mishra4d3a2e02017-01-18 20:27:51 -0800718
719 _LOG_DEBUG_ID("interest: " << interest.getName());
720 _LOG_DEBUG_ID("<< Logic::sendRecoveryInterest");
721}
722
723void
724Logic::processRecoveryInterest(const Interest& interest)
725{
726 _LOG_DEBUG_ID(">> Logic::processRecoveryInterest");
727
728 const Name& name = interest.getName();
729 ConstBufferPtr digest = make_shared<ndn::Buffer>(name.get(-1).value(), name.get(-1).value_size());
730
731 ConstBufferPtr rootDigest = m_state.getRootDigest();
732
733 DiffStateContainer::iterator stateIter = m_log.find(digest);
734
735 if (stateIter != m_log.end() || *digest == *EMPTY_DIGEST || *rootDigest == *digest) {
736 _LOG_DEBUG_ID("I can help you recover");
737 sendSyncData(m_defaultUserPrefix, name, m_state);
738 return;
739 }
740 _LOG_DEBUG_ID("<< Logic::processRecoveryInterest");
741}
742
743void
Alexander Afanasyeve9eda8a2017-03-09 14:40:03 -0800744Logic::onRecoveryData(const Interest& interest, const Data& data)
Sonu Mishra4d3a2e02017-01-18 20:27:51 -0800745{
746 _LOG_DEBUG_ID(">> Logic::onRecoveryData");
747 onSyncDataValidated(data.shared_from_this());
748 _LOG_DEBUG_ID("<< Logic::onRecoveryData");
749}
750
751void
752Logic::onRecoveryTimeout(const Interest& interest)
753{
754 _LOG_DEBUG_ID(">> Logic::onRecoveryTimeout");
755 _LOG_DEBUG_ID("Interest: " << interest.getName());
756 _LOG_DEBUG_ID("<< Logic::onRecoveryTimeout");
757}
758
Sonu Mishraf42aa2c2017-01-22 18:47:33 -0800759void
760Logic::sendExcludeInterest(const Interest& interest, const Data& data)
761{
762 _LOG_DEBUG_ID(">> Logic::sendExcludeInterest");
763
764 Name interestName = interest.getName();
765 Interest excludeInterest(interestName);
766
767 Exclude exclude = interest.getExclude();
768 exclude.excludeOne(data.getFullName().get(-1));
769 excludeInterest.setExclude(exclude);
770
771 excludeInterest.setInterestLifetime(m_syncInterestLifetime);
772
Ashlesh Gawanded31d6b12017-03-31 11:43:22 -0500773 if (excludeInterest.wireEncode().size() > ndn::MAX_NDN_PACKET_SIZE) {
774 return;
775 }
776
Alexander Afanasyeve9eda8a2017-03-09 14:40:03 -0800777 m_face.expressInterest(excludeInterest,
778 bind(&Logic::onSyncData, this, _1, _2),
779 bind(&Logic::onSyncTimeout, this, _1), // Nack
780 bind(&Logic::onSyncTimeout, this, _1));
Sonu Mishraf42aa2c2017-01-22 18:47:33 -0800781
782 _LOG_DEBUG_ID("Send interest: " << excludeInterest.getName());
783 _LOG_DEBUG_ID("<< Logic::sendExcludeInterest");
784}
785
786void
787Logic::formAndSendExcludeInterest(const Name& nodePrefix, const State& commit, ndn::ConstBufferPtr previousRoot)
788{
789 _LOG_DEBUG_ID(">> Logic::formAndSendExcludeInterest");
790 Name interestName;
791 interestName.append(m_syncPrefix)
792 .append(ndn::name::Component(*previousRoot));
793 Interest interest(interestName);
794
795 shared_ptr<Data> data = make_shared<Data>(interestName);
796 data->setContent(commit.wireEncode());
797 data->setFreshnessPeriod(m_syncReplyFreshness);
798 if (m_nodeList.find(nodePrefix) == m_nodeList.end())
799 return;
800 if (m_nodeList[nodePrefix].signingId.empty())
801 m_keyChain.sign(*data);
802 else
Ashlesh Gawande687cf922017-05-30 15:04:16 -0500803 m_keyChain.sign(*data, security::signingByIdentity(m_nodeList[nodePrefix].signingId));
Sonu Mishraf42aa2c2017-01-22 18:47:33 -0800804
805 sendExcludeInterest(interest, *data);
806
807 _LOG_DEBUG_ID("<< Logic::formAndSendExcludeInterest");
808}
809
Alexander Afanasyeve9eda8a2017-03-09 14:40:03 -0800810} // namespace chronosync