logic: handle sending data packets larger than MAX_NDN_PACKET_SIZE

refs: #4140

Change-Id: I722604a55765ffc1f27639f4d21e69f118cec77d

common.hpp

@@ -1,6 +1,6 @@
 /* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
 /*
- * Copyright (c) 2012-2017 University of California, Los Angeles
+ * Copyright (c) 2012-2018 University of California, Los Angeles
  *
  * This file is part of ChronoSync, synchronization library for distributed realtime
  * applications for NDN.
@@ -25,7 +25,7 @@
 
 #include "config.hpp"
 
-#ifdef _TESTS
+#ifdef CHRONOSYNC_HAVE_TESTS
 #define CHRONOSYNC_VIRTUAL_WITH_TESTS virtual
 #define CHRONOSYNC_PUBLIC_WITH_TESTS_ELSE_PROTECTED public
 #define CHRONOSYNC_PUBLIC_WITH_TESTS_ELSE_PRIVATE public

src/logic.cpp

@@ -86,9 +86,9 @@ Logic::Logic(ndn::Face& face,
   , m_needPeriodReset(resetTimer > time::steady_clock::Duration::zero())
   , m_onUpdate(onUpdate)
   , m_scheduler(m_face.getIoService())
-  , m_randomGenerator(static_cast<unsigned int>(std::time(0)))
-  , m_rangeUniformRandom(m_randomGenerator, boost::uniform_int<>(100,500))
-  , m_reexpressionJitter(m_randomGenerator, boost::uniform_int<>(100,500))
+  , m_rng(std::random_device{}())
+  , m_rangeUniformRandom(100, 500)
+  , m_reexpressionJitter(100, 500)
   , m_resetTimer(resetTimer)
   , m_cancelResetTimer(cancelResetTimer)
   , m_resetInterestLifetime(resetInterestLifetime)
@@ -413,7 +413,7 @@ Logic::processSyncInterest(const Interest& interest, bool isTimedProcessing/*=fa
       if (static_cast<bool>(m_delayedInterestProcessingId))
         m_scheduler.cancelEvent(m_delayedInterestProcessingId);
 
-      time::milliseconds after(m_rangeUniformRandom());
+      time::milliseconds after(m_rangeUniformRandom(m_rng));
       _LOG_DEBUG_ID("After: " << after);
       m_delayedInterestProcessingId =
         m_scheduler.scheduleEvent(after,
@@ -454,7 +454,7 @@ Logic::processSyncInterest(const Interest& interest, bool isTimedProcessing/*=fa
       m_scheduler.cancelEvent(m_delayedInterestProcessingId);
 
     m_delayedInterestProcessingId =
-      m_scheduler.scheduleEvent(time::milliseconds(m_rangeUniformRandom()),
+      m_scheduler.scheduleEvent(time::milliseconds(m_rangeUniformRandom(m_rng)),
                                 bind(&Logic::processSyncInterest, this, interest, true));
   }
   else {
@@ -530,7 +530,7 @@ Logic::processSyncData(const Name& name,
 
   if (static_cast<bool>(commit) && !commit->getLeaves().empty() && firstData) {
     // state changed and it is safe to express a new interest
-    time::steady_clock::Duration after = time::milliseconds(m_reexpressionJitter());
+    time::steady_clock::Duration after = time::milliseconds(m_reexpressionJitter(m_rng));
     _LOG_DEBUG_ID("Reschedule sync interest after: " << after);
     EventId eventId = m_scheduler.scheduleEvent(after,
                                                 bind(&Logic::sendSyncInterest, this));
@@ -587,7 +587,7 @@ Logic::sendResetInterest()
     _LOG_DEBUG_ID("ResetTimer: " << m_resetTimer);
 
     EventId eventId =
-      m_scheduler.scheduleEvent(m_resetTimer + ndn::time::milliseconds(m_reexpressionJitter()),
+      m_scheduler.scheduleEvent(m_resetTimer + ndn::time::milliseconds(m_reexpressionJitter(m_rng)),
                                 bind(&Logic::sendResetInterest, this));
     m_scheduler.cancelEvent(m_resetInterestId);
     m_resetInterestId = eventId;
@@ -621,7 +621,7 @@ Logic::sendSyncInterest()
 
   EventId eventId =
     m_scheduler.scheduleEvent(m_syncInterestLifetime / 2 +
-                              ndn::time::milliseconds(m_reexpressionJitter()),
+                              ndn::time::milliseconds(m_reexpressionJitter(m_rng)),
                               bind(&Logic::sendSyncInterest, this));
   m_scheduler.cancelEvent(m_reexpressingInterestId);
   m_reexpressingInterestId = eventId;
@@ -639,20 +639,56 @@ Logic::sendSyncInterest()
   _LOG_DEBUG_ID("<< Logic::sendSyncInterest");
 }
 
+void
+Logic::trimState(State& partialState, const State& state, size_t maxSize)
+{
+  partialState.reset();
+  State tmp;
+  std::vector<ConstLeafPtr> leaves;
+  for (const ConstLeafPtr& leaf : state.getLeaves()) {
+    leaves.push_back(leaf);
+  }
+
+  std::shuffle(leaves.begin(), leaves.end(), m_rng);
+
+  for (const auto& constLeafPtr : leaves) {
+    tmp.update(constLeafPtr->getSessionName(), constLeafPtr->getSeq());
+    if (tmp.wireEncode().size() >= maxSize) {
+      break;
+    }
+    partialState.update(constLeafPtr->getSessionName(), constLeafPtr->getSeq());
+  }
+}
+
 void
 Logic::sendSyncData(const Name& nodePrefix, const Name& name, const State& state)
 {
   _LOG_DEBUG_ID(">> Logic::sendSyncData");
+  if (m_nodeList.find(nodePrefix) == m_nodeList.end())
+    return;
+
   Data syncReply(name);
   syncReply.setContent(state.wireEncode());
   syncReply.setFreshnessPeriod(m_syncReplyFreshness);
-  if (m_nodeList.find(nodePrefix) == m_nodeList.end())
-    return;
+
   if (m_nodeList[nodePrefix].signingId.empty())
     m_keyChain.sign(syncReply);
   else
     m_keyChain.sign(syncReply, security::signingByIdentity(m_nodeList[nodePrefix].signingId));
 
+  if (syncReply.wireEncode().size() > ndn::MAX_NDN_PACKET_SIZE) {
+    _LOG_DEBUG("Sync reply size exceeded MAX_NDN_PACKET_SIZE");
+    auto maxContentSize = ndn::MAX_NDN_PACKET_SIZE - (syncReply.wireEncode().size() - state.wireEncode().size());
+    State partialState;
+    trimState(partialState, state, maxContentSize);
+    syncReply.setContent(partialState.wireEncode());
+
+    if (m_nodeList[nodePrefix].signingId.empty())
+      m_keyChain.sign(syncReply);
+    else
+      m_keyChain.sign(syncReply, security::signingByIdentity(m_nodeList[nodePrefix].signingId));
+  }
+
   m_face.put(syncReply);
 
   // checking if our own interest got satisfied
@@ -664,7 +700,7 @@ Logic::sendSyncData(const Name& nodePrefix, const Name& name, const State& state
     }
 
     // re-schedule sending Sync interest
-    time::milliseconds after(m_reexpressionJitter());
+    time::milliseconds after(m_reexpressionJitter(m_rng));
     _LOG_DEBUG_ID("Satisfy our own interest");
     _LOG_DEBUG_ID("Reschedule sync interest after " << after);
     EventId eventId = m_scheduler.scheduleEvent(after, bind(&Logic::sendSyncInterest, this));

src/logic.hpp

@@ -1,6 +1,6 @@
 /* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
 /*
- * Copyright (c) 2012-2017 University of California, Los Angeles
+ * Copyright (c) 2012-2018 University of California, Los Angeles
  *
  * This file is part of ChronoSync, synchronization library for distributed realtime
  * applications for NDN.
@@ -33,10 +33,10 @@
 #include <boost/archive/iterators/transform_width.hpp>
 #include <boost/assert.hpp>
 #include <boost/iterator/transform_iterator.hpp>
-#include <boost/random.hpp>
 #include <boost/throw_exception.hpp>
 
 #include <memory>
+#include <random>
 #include <unordered_map>
 
 namespace chronosync {
@@ -230,6 +230,9 @@ CHRONOSYNC_PUBLIC_WITH_TESTS_ELSE_PRIVATE:
     return m_state;
   }
 
+  /// @brief Trim @p state to a subset @p partialState whose encoding does not exceed @p maxSize
+  void
+  trimState(State& partialState, const State& state, size_t maxSize);
 
 private:
   /**
@@ -498,9 +501,9 @@ private:
   ndn::EventId m_resetInterestId;
 
   // Timer
-  boost::mt19937 m_randomGenerator;
-  boost::variate_generator<boost::mt19937&, boost::uniform_int<> > m_rangeUniformRandom;
-  boost::variate_generator<boost::mt19937&, boost::uniform_int<> > m_reexpressionJitter;
+  std::mt19937 m_rng;
+  std::uniform_int_distribution<> m_rangeUniformRandom;
+  std::uniform_int_distribution<> m_reexpressionJitter;
   /// @brief Timer to send next reset 0 for no reset
   time::steady_clock::Duration m_resetTimer;
   /// @brief Timer to cancel reset state
@@ -518,7 +521,6 @@ private:
   ndn::KeyChain m_keyChain;
   std::shared_ptr<Validator> m_validator;
 
-
 #ifdef _DEBUG
   int m_instanceId;
   static int s_instanceCounter;

tests/unit-tests/test-logic.cpp

@@ -1,6 +1,6 @@
 /* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
 /*
- * Copyright (c) 2012-2017 University of California, Los Angeles
+ * Copyright (c) 2012-2018 University of California, Los Angeles
  *
  * This file is part of ChronoSync, synchronization library for distributed realtime
  * applications for NDN.
@@ -319,6 +319,35 @@ BOOST_AUTO_TEST_CASE(MultipleUserUnderOneLogic)
   BOOST_CHECK_EQUAL(handler[1]->logic.getSessionNames().size(), 2);
 }
 
+BOOST_FIXTURE_TEST_CASE(ExplodeData, ndn::tests::IdentityManagementTimeFixture)
+{
+  Name syncPrefix("/ndn/broadcast/sync");
+  Name userPrefix("/user");
+  ndn::util::DummyClientFace face;
+  Logic logic(face, syncPrefix, userPrefix, bind(onUpdate, _1));
+
+  State state;
+  int i = 0;
+  while (state.wireEncode().size() < ndn::MAX_NDN_PACKET_SIZE) {
+    Name name("/test1");
+    name.append(std::to_string(i));
+    state.update(name, i++);
+  }
+
+  Data syncReply(syncPrefix);
+  syncReply.setContent(state.wireEncode());
+  m_keyChain.sign(syncReply);
+
+  BOOST_REQUIRE(syncReply.wireEncode().size() > ndn::MAX_NDN_PACKET_SIZE);
+
+  State partialState;
+  auto maxSize = ndn::MAX_NDN_PACKET_SIZE - (syncReply.wireEncode().size() - state.wireEncode().size());
+  logic.trimState(partialState, state, maxSize);
+
+  syncReply.setContent(partialState.wireEncode());
+  BOOST_REQUIRE(syncReply.wireEncode().size() < ndn::MAX_NDN_PACKET_SIZE);
+}
+
 BOOST_AUTO_TEST_SUITE_END()
 
 } // namespace test