Minor code cleanups

Change-Id: I2b67216754ba30563a65a2697e6801e363986da1
This commit is contained in:
Davide Pesavento committed 2022-09-25 14:06:26 -04:00
1 parent c45a4eaef4
commit 03426efa2c
14 files changed
+106 -131

No files matched your search

+4 -4
View File
@@ -35,10 +35,10 @@ namespace psync {
using namespace ndn::time_literals; using namespace ndn::time_literals;
const ndn::time::milliseconds HELLO_INTEREST_LIFETIME = 1_s; inline constexpr ndn::time::milliseconds HELLO_INTEREST_LIFETIME = 1_s;
const ndn::time::milliseconds HELLO_REPLY_FRESHNESS = 1_s; inline constexpr ndn::time::milliseconds HELLO_REPLY_FRESHNESS = 1_s;
const ndn::time::milliseconds SYNC_INTEREST_LIFETIME = 1_s; inline constexpr ndn::time::milliseconds SYNC_INTEREST_LIFETIME = 1_s;
const ndn::time::milliseconds SYNC_REPLY_FRESHNESS = 1_s; inline constexpr ndn::time::milliseconds SYNC_REPLY_FRESHNESS = 1_s;
enum class CompressionScheme { enum class CompressionScheme {
NONE, NONE,
+16 -12
View File
@@ -48,8 +48,6 @@
namespace psync::detail { namespace psync::detail {
namespace bio = boost::iostreams;
static inline uint32_t static inline uint32_t
ROTL32(uint32_t x, int8_t r) ROTL32(uint32_t x, int8_t r)
{ {
@@ -125,10 +123,14 @@ murmurHash3(uint32_t seed, const ndn::Name& name)
std::shared_ptr<ndn::Buffer> std::shared_ptr<ndn::Buffer>
compress(CompressionScheme scheme, ndn::span<const uint8_t> buffer) compress(CompressionScheme scheme, ndn::span<const uint8_t> buffer)
{ {
ndn::OBufferStream out; namespace bio = boost::iostreams;
bio::filtering_streambuf<bio::input> in;
bio::filtering_istreambuf in;
switch (scheme) { switch (scheme) {
case CompressionScheme::NONE:
break;
case CompressionScheme::ZLIB: case CompressionScheme::ZLIB:
#ifdef PSYNC_HAVE_ZLIB #ifdef PSYNC_HAVE_ZLIB
in.push(bio::zlib_compressor(bio::zlib::best_compression)); in.push(bio::zlib_compressor(bio::zlib::best_compression));
@@ -168,11 +170,10 @@ compress(CompressionScheme scheme, ndn::span<const uint8_t> buffer)
#else #else
NDN_THROW(CompressionError("ZSTD compression not supported!")); NDN_THROW(CompressionError("ZSTD compression not supported!"));
#endif #endif
case CompressionScheme::NONE:
break;
} }
in.push(bio::array_source(reinterpret_cast<const char*>(buffer.data()), buffer.size())); in.push(bio::array_source(reinterpret_cast<const char*>(buffer.data()), buffer.size()));
ndn::OBufferStream out;
bio::copy(in, out); bio::copy(in, out);
return out.buf(); return out.buf();
@@ -181,10 +182,14 @@ compress(CompressionScheme scheme, ndn::span<const uint8_t> buffer)
std::shared_ptr<ndn::Buffer> std::shared_ptr<ndn::Buffer>
decompress(CompressionScheme scheme, ndn::span<const uint8_t> buffer) decompress(CompressionScheme scheme, ndn::span<const uint8_t> buffer)
{ {
ndn::OBufferStream out; namespace bio = boost::iostreams;
bio::filtering_streambuf<bio::input> in;
bio::filtering_istreambuf in;
switch (scheme) { switch (scheme) {
case CompressionScheme::NONE:
break;
case CompressionScheme::ZLIB: case CompressionScheme::ZLIB:
#ifdef PSYNC_HAVE_ZLIB #ifdef PSYNC_HAVE_ZLIB
in.push(bio::zlib_decompressor()); in.push(bio::zlib_decompressor());
@@ -224,11 +229,10 @@ decompress(CompressionScheme scheme, ndn::span<const uint8_t> buffer)
#else #else
NDN_THROW(CompressionError("ZSTD compression not supported!")); NDN_THROW(CompressionError("ZSTD compression not supported!"));
#endif #endif
case CompressionScheme::NONE:
break;
} }
in.push(bio::array_source(reinterpret_cast<const char*>(buffer.data()), buffer.size())); in.push(bio::array_source(reinterpret_cast<const char*>(buffer.data()), buffer.size()));
ndn::OBufferStream out;
bio::copy(in, out); bio::copy(in, out);
return out.buf(); return out.buf();
+6 -9
View File
@@ -21,10 +21,9 @@
#include "PSync/detail/state.hpp" #include "PSync/detail/state.hpp"
#include "PSync/detail/util.hpp" #include "PSync/detail/util.hpp"
#include <ndn-cxx/lp/tags.hpp>
#include <ndn-cxx/security/validator-null.hpp> #include <ndn-cxx/security/validator-null.hpp>
#include <ndn-cxx/util/logger.hpp> #include <ndn-cxx/util/logger.hpp>
#include <ndn-cxx/util/segment-fetcher.hpp>
#include <ndn-cxx/lp/tags.hpp>
#include <cstring> #include <cstring>
@@ -37,7 +36,7 @@ FullProducer::FullProducer(ndn::Face& face,
size_t expectedNumEntries, size_t expectedNumEntries,
const ndn::Name& syncPrefix, const ndn::Name& syncPrefix,
const ndn::Name& userPrefix, const ndn::Name& userPrefix,
const UpdateCallback& onUpdateCallBack, UpdateCallback onUpdateCb,
ndn::time::milliseconds syncInterestLifetime, ndn::time::milliseconds syncInterestLifetime,
ndn::time::milliseconds syncReplyFreshness, ndn::time::milliseconds syncReplyFreshness,
CompressionScheme ibltCompression, CompressionScheme ibltCompression,
@@ -45,13 +44,11 @@ FullProducer::FullProducer(ndn::Face& face,
: ProducerBase(face, keyChain, expectedNumEntries, syncPrefix, userPrefix, : ProducerBase(face, keyChain, expectedNumEntries, syncPrefix, userPrefix,
syncReplyFreshness, ibltCompression, contentCompression) syncReplyFreshness, ibltCompression, contentCompression)
, m_syncInterestLifetime(syncInterestLifetime) , m_syncInterestLifetime(syncInterestLifetime)
, m_onUpdate(onUpdateCallBack) , m_onUpdate(std::move(onUpdateCb))
, m_jitter(100, 500)
{ {
m_registeredPrefix = m_face.setInterestFilter( m_registeredPrefix = m_face.setInterestFilter(ndn::InterestFilter(m_syncPrefix).allowLoopback(false),
ndn::InterestFilter(m_syncPrefix).allowLoopback(false), [this] (auto&&... args) { onSyncInterest(std::forward<decltype(args)>(args)...); },
std::bind(&FullProducer::onSyncInterest, this, _1, _2), [] (auto&&... args) { onRegisterFailed(std::forward<decltype(args)>(args)...); });
std::bind(&FullProducer::onRegisterFailed, this, _1, _2));
// Should we do this after setInterestFilter success call back // Should we do this after setInterestFilter success call back
// (Currently following ChronoSync's way) // (Currently following ChronoSync's way)
+5 -5
View File
@@ -51,9 +51,9 @@ public:
* @param expectedNumEntries Expected number of entries in IBF * @param expectedNumEntries Expected number of entries in IBF
* @param syncPrefix The prefix of the sync group * @param syncPrefix The prefix of the sync group
* @param userPrefix The prefix of the first user in the group * @param userPrefix The prefix of the first user in the group
* @param onUpdateCallBack The call back to be called when there is new data * @param onUpdateCallBack The callback to be invoked when there is new data
* @param syncInterestLifetime lifetime of the sync interest * @param syncInterestLifetime Lifetime of the sync interest
* @param syncReplyFreshness freshness of sync data * @param syncReplyFreshness FreshnessPeriod of sync data
* @param ibltCompression Compression scheme to use for IBF * @param ibltCompression Compression scheme to use for IBF
* @param contentCompression Compression scheme to use for Data content * @param contentCompression Compression scheme to use for Data content
*/ */
@@ -62,7 +62,7 @@ public:
size_t expectedNumEntries, size_t expectedNumEntries,
const ndn::Name& syncPrefix, const ndn::Name& syncPrefix,
const ndn::Name& userPrefix, const ndn::Name& userPrefix,
const UpdateCallback& onUpdateCallBack, UpdateCallback onUpdateCallBack,
ndn::time::milliseconds syncInterestLifetime = SYNC_INTEREST_LIFETIME, ndn::time::milliseconds syncInterestLifetime = SYNC_INTEREST_LIFETIME,
ndn::time::milliseconds syncReplyFreshness = SYNC_REPLY_FRESHNESS, ndn::time::milliseconds syncReplyFreshness = SYNC_REPLY_FRESHNESS,
CompressionScheme ibltCompression = CompressionScheme::DEFAULT, CompressionScheme ibltCompression = CompressionScheme::DEFAULT,
@@ -186,7 +186,7 @@ private:
ndn::time::milliseconds m_syncInterestLifetime; ndn::time::milliseconds m_syncInterestLifetime;
UpdateCallback m_onUpdate; UpdateCallback m_onUpdate;
ndn::scheduler::ScopedEventId m_scheduledSyncInterestId; ndn::scheduler::ScopedEventId m_scheduledSyncInterestId;
std::uniform_int_distribution<> m_jitter; std::uniform_int_distribution<> m_jitter{100, 500};
ndn::Name m_outstandingInterestName; ndn::Name m_outstandingInterestName;
ndn::ScopedRegisteredPrefixHandle m_registeredPrefix; ndn::ScopedRegisteredPrefixHandle m_registeredPrefix;
std::shared_ptr<ndn::util::SegmentFetcher> m_fetcher; std::shared_ptr<ndn::util::SegmentFetcher> m_fetcher;
+2 -3
View File
@@ -23,7 +23,6 @@
#include <ndn-cxx/util/logger.hpp> #include <ndn-cxx/util/logger.hpp>
#include <cstring> #include <cstring>
#include <limits>
namespace psync { namespace psync {
@@ -45,13 +44,13 @@ PartialProducer::PartialProducer(ndn::Face& face,
, m_helloReplyFreshness(helloReplyFreshness) , m_helloReplyFreshness(helloReplyFreshness)
{ {
m_registeredPrefix = m_face.registerPrefix(m_syncPrefix, m_registeredPrefix = m_face.registerPrefix(m_syncPrefix,
[this] (const ndn::Name& syncPrefix) { [this] (const auto&) {
m_face.setInterestFilter(ndn::Name(m_syncPrefix).append(HELLO), m_face.setInterestFilter(ndn::Name(m_syncPrefix).append(HELLO),
std::bind(&PartialProducer::onHelloInterest, this, _1, _2)); std::bind(&PartialProducer::onHelloInterest, this, _1, _2));
m_face.setInterestFilter(ndn::Name(m_syncPrefix).append(SYNC), m_face.setInterestFilter(ndn::Name(m_syncPrefix).append(SYNC),
std::bind(&PartialProducer::onSyncInterest, this, _1, _2)); std::bind(&PartialProducer::onSyncInterest, this, _1, _2));
}, },
std::bind(&PartialProducer::onRegisterFailed, this, _1, _2)); [] (auto&&... args) { onRegisterFailed(std::forward<decltype(args)>(args)...); });
} }
void void
+3 -9
View File
@@ -20,14 +20,8 @@
#ifndef PSYNC_PARTIAL_PRODUCER_HPP #ifndef PSYNC_PARTIAL_PRODUCER_HPP
#define PSYNC_PARTIAL_PRODUCER_HPP #define PSYNC_PARTIAL_PRODUCER_HPP
#include "PSync/detail/bloom-filter.hpp"
#include "PSync/producer-base.hpp" #include "PSync/producer-base.hpp"
#include "PSync/detail/bloom-filter.hpp"
#include <map>
#include <ndn-cxx/face.hpp>
#include <ndn-cxx/security/key-chain.hpp>
#include <ndn-cxx/util/scheduler.hpp>
namespace psync { namespace psync {
@@ -53,8 +47,8 @@ public:
* @param expectedNumEntries Expected number of entries in IBF * @param expectedNumEntries Expected number of entries in IBF
* @param syncPrefix The prefix of the sync group * @param syncPrefix The prefix of the sync group
* @param userPrefix The prefix of the first user in the group * @param userPrefix The prefix of the first user in the group
* @param syncReplyFreshness freshness of sync data * @param helloReplyFreshness FreshnessPeriod of hello data
* @param helloReplyFreshness freshness of hello data * @param syncReplyFreshness FreshnessPeriod of sync data
* @param ibltCompression Compression scheme to use for IBF * @param ibltCompression Compression scheme to use for IBF
*/ */
PartialProducer(ndn::Face& face, PartialProducer(ndn::Face& face,
+7 -9
View File
@@ -23,9 +23,6 @@
#include <ndn-cxx/util/exception.hpp> #include <ndn-cxx/util/exception.hpp>
#include <ndn-cxx/util/logger.hpp> #include <ndn-cxx/util/logger.hpp>
#include <cstring>
#include <limits>
namespace psync { namespace psync {
NDN_LOG_INIT(psync.ProducerBase); NDN_LOG_INIT(psync.ProducerBase);
@@ -127,22 +124,23 @@ void
ProducerBase::sendApplicationNack(const ndn::Name& name) ProducerBase::sendApplicationNack(const ndn::Name& name)
{ {
NDN_LOG_DEBUG("Sending application nack"); NDN_LOG_DEBUG("Sending application nack");
ndn::Name dataName(name); ndn::Name dataName(name);
m_iblt.appendToName(dataName); m_iblt.appendToName(dataName);
dataName.appendSegment(0); dataName.appendSegment(0);
ndn::Data data(dataName); ndn::Data data(dataName);
data.setFreshnessPeriod(m_syncReplyFreshness); data.setContentType(ndn::tlv::ContentType_Nack)
data.setContentType(ndn::tlv::ContentType_Nack); .setFreshnessPeriod(m_syncReplyFreshness)
data.setFinalBlock(dataName[-1]); .setFinalBlock(dataName[-1]);
m_keyChain.sign(data); m_keyChain.sign(data);
m_face.put(data); m_face.put(data);
} }
void void
ProducerBase::onRegisterFailed(const ndn::Name& prefix, const std::string& msg) const ProducerBase::onRegisterFailed(const ndn::Name& prefix, const std::string& msg)
{ {
NDN_LOG_ERROR("ProducerBase::onRegisterFailed(" << prefix << "): " << msg); NDN_LOG_ERROR("onRegisterFailed(" << prefix << "): " << msg);
NDN_THROW(Error(msg)); NDN_THROW(Error(msg));
} }
+4 -5
View File
@@ -27,7 +27,6 @@
#include <ndn-cxx/face.hpp> #include <ndn-cxx/face.hpp>
#include <ndn-cxx/security/key-chain.hpp> #include <ndn-cxx/security/key-chain.hpp>
#include <ndn-cxx/security/validator-config.hpp>
#include <ndn-cxx/util/random.hpp> #include <ndn-cxx/util/random.hpp>
#include <ndn-cxx/util/scheduler.hpp> #include <ndn-cxx/util/scheduler.hpp>
@@ -43,7 +42,7 @@ namespace bm = boost::bimaps;
/** /**
* @brief Base class for PartialProducer and FullProducer * @brief Base class for PartialProducer and FullProducer
* *
* Contains code common to both * Contains code common to both.
*/ */
class ProducerBase class ProducerBase
{ {
@@ -149,10 +148,10 @@ PSYNC_PUBLIC_WITH_TESTS_ELSE_PROTECTED:
sendApplicationNack(const ndn::Name& name); sendApplicationNack(const ndn::Name& name);
/** /**
* @brief Logs a message if setting an interest filter fails * @brief Logs a message and throws if setting an interest filter fails
*/ */
void [[noreturn]] static void
onRegisterFailed(const ndn::Name& prefix, const std::string& msg) const; onRegisterFailed(const ndn::Name& prefix, const std::string& msg);
PSYNC_PUBLIC_WITH_TESTS_ELSE_PROTECTED: PSYNC_PUBLIC_WITH_TESTS_ELSE_PROTECTED:
ndn::Face& m_face; ndn::Face& m_face;
+6 -7
View File
@@ -1,6 +1,6 @@
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */ /* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/* /*
* Copyright (c) 2014-2020, The University of Memphis * Copyright (c) 2014-2022, The University of Memphis
* *
* This file is part of PSync. * This file is part of PSync.
* See AUTHORS.md for complete list of PSync authors and contributors. * See AUTHORS.md for complete list of PSync authors and contributors.
@@ -19,7 +19,7 @@
#include <PSync/consumer.hpp> #include <PSync/consumer.hpp>
#include <ndn-cxx/name.hpp> #include <ndn-cxx/face.hpp>
#include <ndn-cxx/util/logger.hpp> #include <ndn-cxx/util/logger.hpp>
#include <ndn-cxx/util/random.hpp> #include <ndn-cxx/util/random.hpp>
@@ -44,7 +44,6 @@ public:
std::bind(&PSyncConsumer::afterReceiveHelloData, this, _1), std::bind(&PSyncConsumer::afterReceiveHelloData, this, _1),
std::bind(&PSyncConsumer::processSyncUpdate, this, _1), std::bind(&PSyncConsumer::processSyncUpdate, this, _1),
m_nSub, 0.001) m_nSub, 0.001)
, m_rng(ndn::random::getRandomNumberEngine())
{ {
// This starts the consumer side by sending a hello interest to the producer // This starts the consumer side by sending a hello interest to the producer
// When the producer responds with hello data, afterReceiveHelloData is called // When the producer responds with hello data, afterReceiveHelloData is called
@@ -96,18 +95,18 @@ private:
private: private:
ndn::Face m_face; ndn::Face m_face;
int m_nSub;
int m_nSub;
psync::Consumer m_consumer; psync::Consumer m_consumer;
ndn::random::RandomNumberEngine& m_rng;
ndn::random::RandomNumberEngine& m_rng{ndn::random::getRandomNumberEngine()};
}; };
int int
main(int argc, char* argv[]) main(int argc, char* argv[])
{ {
if (argc != 3) { if (argc != 3) {
std::cout << "usage: " << argv[0] << " " std::cerr << "Usage: " << argv[0] << " <sync-prefix> <number-of-subscriptions>\n";
<< "<sync-prefix> <number-of-subscriptions>" << std::endl;
return 1; return 1;
} }
+13 -16
View File
@@ -43,18 +43,16 @@ public:
*/ */
Producer(const ndn::Name& syncPrefix, const std::string& userPrefix, Producer(const ndn::Name& syncPrefix, const std::string& userPrefix,
int numDataStreams, int maxNumPublish) int numDataStreams, int maxNumPublish)
: m_fullProducer(m_face, m_keyChain, 80, syncPrefix, userPrefix, : m_producer(m_face, m_keyChain, 80, syncPrefix, userPrefix,
std::bind(&Producer::processSyncUpdate, this, _1), std::bind(&Producer::processSyncUpdate, this, _1),
1600_ms, 1600_ms) 1600_ms, 1600_ms)
, m_maxNumPublish(maxNumPublish) , m_maxNumPublish(maxNumPublish)
, m_rng(ndn::random::getRandomNumberEngine())
, m_rangeUniformRandom(0, 60000)
{ {
// Add user prefixes and schedule updates for them in specified interval // Add user prefixes and schedule updates for them in specified interval
for (int i = 0; i < numDataStreams; i++) { for (int i = 0; i < numDataStreams; i++) {
ndn::Name prefix(userPrefix + "-" + std::to_string(i)); ndn::Name prefix(userPrefix + "-" + std::to_string(i));
m_fullProducer.addUserNode(prefix); m_producer.addUserNode(prefix);
m_scheduler.schedule(ndn::time::milliseconds(m_rangeUniformRandom(m_rng)), m_scheduler.schedule(ndn::time::milliseconds(m_uniformRand(m_rng)),
[this, prefix] { doUpdate(prefix); }); [this, prefix] { doUpdate(prefix); });
} }
} }
@@ -69,13 +67,13 @@ private:
void void
doUpdate(const ndn::Name& prefix) doUpdate(const ndn::Name& prefix)
{ {
m_fullProducer.publishName(prefix); m_producer.publishName(prefix);
uint64_t seqNo = m_fullProducer.getSeqNo(prefix).value(); uint64_t seqNo = m_producer.getSeqNo(prefix).value();
NDN_LOG_INFO("Publish: " << prefix << "/" << seqNo); NDN_LOG_INFO("Publish: " << prefix << "/" << seqNo);
if (seqNo < m_maxNumPublish) { if (seqNo < m_maxNumPublish) {
m_scheduler.schedule(ndn::time::milliseconds(m_rangeUniformRandom(m_rng)), m_scheduler.schedule(ndn::time::milliseconds(m_uniformRand(m_rng)),
[this, prefix] { doUpdate(prefix); }); [this, prefix] { doUpdate(prefix); });
} }
} }
@@ -95,20 +93,19 @@ private:
ndn::KeyChain m_keyChain; ndn::KeyChain m_keyChain;
ndn::Scheduler m_scheduler{m_face.getIoService()}; ndn::Scheduler m_scheduler{m_face.getIoService()};
psync::FullProducer m_fullProducer; psync::FullProducer m_producer;
uint64_t m_maxNumPublish; uint64_t m_maxNumPublish;
ndn::random::RandomNumberEngine& m_rng; ndn::random::RandomNumberEngine& m_rng{ndn::random::getRandomNumberEngine()};
std::uniform_int_distribution<> m_rangeUniformRandom; std::uniform_int_distribution<> m_uniformRand{0, 60000};
}; };
int int
main(int argc, char* argv[]) main(int argc, char* argv[])
{ {
if (argc != 5) { if (argc != 5) {
std::cout << "usage: " << argv[0] << " <syncPrefix> <user-prefix> " std::cerr << "Usage: " << argv[0] << " <sync-prefix> <user-prefix> "
<< "<number-of-user-prefixes> <max-number-of-updates-per-user-prefix>" << "<number-of-user-prefixes> <max-number-of-updates-per-user-prefix>\n";
<< std::endl;
return 1; return 1;
} }
+8 -11
View File
@@ -41,8 +41,6 @@ public:
int numDataStreams, int maxNumPublish) int numDataStreams, int maxNumPublish)
: m_producer(m_face, m_keyChain, 40, syncPrefix, userPrefix + "-0") : m_producer(m_face, m_keyChain, 40, syncPrefix, userPrefix + "-0")
, m_maxNumPublish(maxNumPublish) , m_maxNumPublish(maxNumPublish)
, m_rng(ndn::random::getRandomNumberEngine())
, m_rangeUniformRandom(0, 60000)
{ {
// Add user prefixes and schedule updates for them // Add user prefixes and schedule updates for them
for (int i = 0; i < numDataStreams; i++) { for (int i = 0; i < numDataStreams; i++) {
@@ -52,8 +50,8 @@ public:
// Note that this does not add the already added userPrefix-0 in the constructor // Note that this does not add the already added userPrefix-0 in the constructor
m_producer.addUserNode(updateName); m_producer.addUserNode(updateName);
// Each user prefix is updated at random interval between 0 and 60 second // Each user prefix is updated at a random interval between 0 and 60 seconds
m_scheduler.schedule(ndn::time::milliseconds(m_rangeUniformRandom(m_rng)), m_scheduler.schedule(ndn::time::milliseconds(m_uniformRand(m_rng)),
[this, updateName] { doUpdate(updateName); }); [this, updateName] { doUpdate(updateName); });
} }
} }
@@ -75,8 +73,8 @@ private:
NDN_LOG_INFO("Publish: " << updateName << "/" << seqNo); NDN_LOG_INFO("Publish: " << updateName << "/" << seqNo);
if (seqNo < m_maxNumPublish) { if (seqNo < m_maxNumPublish) {
// Schedule the next update for this user prefix b/w 0 and 60 seconds // Schedule the next update for this user prefix between 0 and 60 seconds
m_scheduler.schedule(ndn::time::milliseconds(m_rangeUniformRandom(m_rng)), m_scheduler.schedule(ndn::time::milliseconds(m_uniformRand(m_rng)),
[this, updateName] { doUpdate(updateName); }); [this, updateName] { doUpdate(updateName); });
} }
} }
@@ -89,17 +87,16 @@ private:
psync::PartialProducer m_producer; psync::PartialProducer m_producer;
uint64_t m_maxNumPublish; uint64_t m_maxNumPublish;
ndn::random::RandomNumberEngine& m_rng; ndn::random::RandomNumberEngine& m_rng{ndn::random::getRandomNumberEngine()};
std::uniform_int_distribution<int> m_rangeUniformRandom; std::uniform_int_distribution<> m_uniformRand{0, 60000};
}; };
int int
main(int argc, char* argv[]) main(int argc, char* argv[])
{ {
if (argc != 5) { if (argc != 5) {
std::cout << "usage: " << argv[0] << " <sync-prefix> <user-prefix> " std::cerr << "Usage: " << argv[0] << " <sync-prefix> <user-prefix> "
<< "<number-of-user-prefixes> <max-number-of-updates-per-user-prefix>" << "<number-of-user-prefixes> <max-number-of-updates-per-user-prefix>\n";
<< std::endl;
return 1; return 1;
} }
+3 -13
View File
@@ -1,6 +1,6 @@
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */ /* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/* /*
* Copyright (c) 2014-2020, The University of Memphis * Copyright (c) 2014-2022, The University of Memphis
* *
* This file is part of PSync. * This file is part of PSync.
* See AUTHORS.md for complete list of PSync authors and contributors. * See AUTHORS.md for complete list of PSync authors and contributors.
@@ -22,7 +22,6 @@
#include "tests/boost-test.hpp" #include "tests/boost-test.hpp"
#include "tests/io-fixture.hpp" #include "tests/io-fixture.hpp"
#include <ndn-cxx/name.hpp>
#include <ndn-cxx/util/dummy-client-face.hpp> #include <ndn-cxx/util/dummy-client-face.hpp>
namespace psync { namespace psync {
@@ -31,18 +30,9 @@ using namespace ndn;
BOOST_AUTO_TEST_SUITE(TestConsumer) BOOST_AUTO_TEST_SUITE(TestConsumer)
BOOST_AUTO_TEST_CASE(Constructor)
{
util::DummyClientFace face({true, true});
BOOST_REQUIRE_NO_THROW(Consumer(Name("/psync"), face,
[] (const auto&) {},
[] (const auto&) {},
40, 0.001));
}
BOOST_AUTO_TEST_CASE(AddSubscription) BOOST_AUTO_TEST_CASE(AddSubscription)
{ {
util::DummyClientFace face({true, true}); util::DummyClientFace face;
Consumer consumer(Name("/psync"), face, Consumer consumer(Name("/psync"), face,
[] (const auto&) {}, [] (const auto&) {},
[] (const auto&) {}, [] (const auto&) {},
@@ -57,7 +47,7 @@ BOOST_AUTO_TEST_CASE(AddSubscription)
BOOST_FIXTURE_TEST_CASE(ConstantTimeoutForFirstSegment, tests::IoFixture) BOOST_FIXTURE_TEST_CASE(ConstantTimeoutForFirstSegment, tests::IoFixture)
{ {
util::DummyClientFace face(m_io, {true, true}); util::DummyClientFace face(m_io);
Consumer consumer(Name("/psync"), face, Consumer consumer(Name("/psync"), face,
[] (const auto&) {}, [] (const auto&) {},
[] (const auto&) {}, [] (const auto&) {},
+18 -17
View File
@@ -41,18 +41,18 @@ protected:
{ {
BOOST_ASSERT(id >= 0 && id < MAX_NODES); BOOST_ASSERT(id >= 0 && id < MAX_NODES);
userPrefixes[id] = "/userPrefix" + std::to_string(id); userPrefixes[id] = "/userPrefix" + std::to_string(id);
faces[id] = std::make_shared<util::DummyClientFace>(m_io, m_keyChain, faces[id] = std::make_unique<util::DummyClientFace>(m_io, m_keyChain,
util::DummyClientFace::Options{true, true}); util::DummyClientFace::Options{true, true});
nodes[id] = std::make_shared<FullProducer>(*faces[id], m_keyChain, 40, syncPrefix, userPrefixes[id], nodes[id] = std::make_unique<FullProducer>(*faces[id], m_keyChain, 40, syncPrefix, userPrefixes[id],
[] (const auto&) {}); [] (const auto&) {});
} }
void void
clearNodes() clearNodes()
{ {
faces.fill(nullptr); nodes = {};
userPrefixes.fill(Name()); faces = {};
nodes.fill(nullptr); userPrefixes = {};
} }
/** /**
@@ -60,8 +60,8 @@ protected:
* @param id update originator node index. * @param id update originator node index.
* @param i user prefix index. * @param i user prefix index.
*/ */
Name static Name
makeSubPrefix(int id, int i) const makeSubPrefix(int id, int i)
{ {
return "/userNode" + std::to_string(id) + "-" + std::to_string(i); return "/userNode" + std::to_string(id) + "-" + std::to_string(i);
} }
@@ -141,15 +141,16 @@ protected:
* @param maxTotalUpdates maximum totalUpdates parameter. * @param maxTotalUpdates maximum totalUpdates parameter.
* @param f test function. * @param f test function.
* *
* This method searches for totalUpdates∈[minTotalUpdates,maxTotalUpdates] until there is * This method searches for totalUpdates ∈ [minTotalUpdates,maxTotalUpdates] until
* at least one execution that caused an IBF decode failure above threshold. * there is at least one execution that caused an IBF decode failure above threshold.
* If such an execution is never achieved within the range, fail the test case. * If such an execution never occurs within the range, the test case fails.
* *
* Current FullSync logic cannot reliably recover from an IBF decode failure below threshold. * Current FullSync logic cannot reliably recover from an IBF decode failure below threshold.
* Hence, that condition is not tested. * Hence, that condition is not tested.
*/ */
void void
searchIbfDecodeFailures(int minTotalUpdates, int maxTotalUpdates, std::function<void(int totalUpdates)> f) searchIbfDecodeFailures(int minTotalUpdates, int maxTotalUpdates,
const std::function<void(int totalUpdates)>& f)
{ {
bool hasAboveThreshold = false; bool hasAboveThreshold = false;
for (int totalUpdates = minTotalUpdates; totalUpdates <= maxTotalUpdates; ++totalUpdates) { for (int totalUpdates = minTotalUpdates; totalUpdates <= maxTotalUpdates; ++totalUpdates) {
@@ -173,8 +174,8 @@ protected:
const Name syncPrefix = "/psync"; const Name syncPrefix = "/psync";
static constexpr int MAX_NODES = 4; static constexpr int MAX_NODES = 4;
std::array<Name, MAX_NODES> userPrefixes; std::array<Name, MAX_NODES> userPrefixes;
std::array<std::shared_ptr<util::DummyClientFace>, MAX_NODES> faces; std::array<std::unique_ptr<util::DummyClientFace>, MAX_NODES> faces;
std::array<std::shared_ptr<FullProducer>, MAX_NODES> nodes; std::array<std::unique_ptr<FullProducer>, MAX_NODES> nodes;
static constexpr uint64_t NOT_EXIST = std::numeric_limits<uint64_t>::max(); static constexpr uint64_t NOT_EXIST = std::numeric_limits<uint64_t>::max();
}; };
@@ -466,7 +467,7 @@ BOOST_AUTO_TEST_CASE(DelayedSecondSegment)
int i = 0; int i = 0;
detail::State state; detail::State state;
std::shared_ptr<ndn::Buffer> compressed; std::shared_ptr<Buffer> compressed;
do { do {
auto prefixToPublish = makeSubPrefix(0, i++); auto prefixToPublish = makeSubPrefix(0, i++);
nodes[0]->addUserNode(prefixToPublish); nodes[0]->addUserNode(prefixToPublish);
@@ -476,7 +477,7 @@ BOOST_AUTO_TEST_CASE(DelayedSecondSegment)
auto block = state.wireEncode(); auto block = state.wireEncode();
compressed = detail::compress(nodes[0]->m_contentCompression, block); compressed = detail::compress(nodes[0]->m_contentCompression, block);
} while (compressed->size() < (ndn::MAX_NDN_PACKET_SIZE >> 1)); } while (compressed->size() < (MAX_NDN_PACKET_SIZE >> 1));
advanceClocks(10_ms, 100); advanceClocks(10_ms, 100);
@@ -484,7 +485,7 @@ BOOST_AUTO_TEST_CASE(DelayedSecondSegment)
detail::IBLT iblt(40, nodes[0]->m_ibltCompression); detail::IBLT iblt(40, nodes[0]->m_ibltCompression);
iblt.appendToName(syncInterestName); iblt.appendToName(syncInterestName);
nodes[0]->onSyncInterest(syncPrefix, ndn::Interest(syncInterestName)); nodes[0]->onSyncInterest(syncPrefix, Interest(syncInterestName));
advanceClocks(10_ms); advanceClocks(10_ms);
@@ -500,7 +501,7 @@ BOOST_AUTO_TEST_CASE(DelayedSecondSegment)
interestName.appendSegment(1); interestName.appendSegment(1);
faces[0]->sentData.clear(); faces[0]->sentData.clear();
nodes[0]->onSyncInterest(syncPrefix, ndn::Interest(interestName)); nodes[0]->onSyncInterest(syncPrefix, Interest(interestName));
advanceClocks(10_ms); advanceClocks(10_ms);
// Should have repopulated SegmentPublisher // Should have repopulated SegmentPublisher
+11 -11
View File
@@ -24,7 +24,7 @@
#include "tests/io-fixture.hpp" #include "tests/io-fixture.hpp"
#include "tests/key-chain-fixture.hpp" #include "tests/key-chain-fixture.hpp"
#include <ndn-cxx/name.hpp> #include <array>
#include <ndn-cxx/util/dummy-client-face.hpp> #include <ndn-cxx/util/dummy-client-face.hpp>
namespace psync { namespace psync {
@@ -33,10 +33,10 @@ using namespace ndn;
class PartialSyncFixture : public tests::IoFixture, public tests::KeyChainFixture class PartialSyncFixture : public tests::IoFixture, public tests::KeyChainFixture
{ {
public: protected:
PartialSyncFixture() PartialSyncFixture()
{ {
producer = make_shared<PartialProducer>(face, m_keyChain, 40, syncPrefix, userPrefix); producer = std::make_unique<PartialProducer>(face, m_keyChain, 40, syncPrefix, userPrefix);
addUserNodes("testUser", 10); addUserNodes("testUser", 10);
} }
@@ -52,13 +52,13 @@ public:
void void
addConsumer(int id, const std::vector<std::string>& subscribeTo, bool linkToProducer = true) addConsumer(int id, const std::vector<std::string>& subscribeTo, bool linkToProducer = true)
{ {
consumerFaces[id] = std::make_shared<util::DummyClientFace>(m_io, m_keyChain, consumerFaces[id] = std::make_unique<util::DummyClientFace>(m_io, m_keyChain,
util::DummyClientFace::Options{true, true}); util::DummyClientFace::Options{true, true});
if (linkToProducer) { if (linkToProducer) {
face.linkTo(*consumerFaces[id]); face.linkTo(*consumerFaces[id]);
} }
consumers[id] = std::make_shared<Consumer>(syncPrefix, *consumerFaces[id], consumers[id] = std::make_unique<Consumer>(syncPrefix, *consumerFaces[id],
[&, id] (const auto& availableSubs) { [&, id] (const auto& availableSubs) {
numHelloDataRcvd++; numHelloDataRcvd++;
BOOST_CHECK(checkSubList(availableSubs)); BOOST_CHECK(checkSubList(availableSubs));
@@ -133,14 +133,14 @@ public:
protected: protected:
util::DummyClientFace face{m_io, m_keyChain, {true, true}}; util::DummyClientFace face{m_io, m_keyChain, {true, true}};
Name syncPrefix{"psync"}; const Name syncPrefix{"psync"};
Name userPrefix{"testUser-0"}; const Name userPrefix{"testUser-0"};
shared_ptr<PartialProducer> producer; std::unique_ptr<PartialProducer> producer;
std::map<Name, uint64_t> oldSeqMap; std::map<Name, uint64_t> oldSeqMap;
shared_ptr<Consumer> consumers[3]; std::array<std::unique_ptr<Consumer>, 3> consumers;
shared_ptr<util::DummyClientFace> consumerFaces[3]; std::array<std::unique_ptr<util::DummyClientFace>, 3> consumerFaces;
int numHelloDataRcvd = 0; int numHelloDataRcvd = 0;
int numSyncDataRcvd = 0; int numSyncDataRcvd = 0;
}; };
@@ -292,7 +292,7 @@ BOOST_AUTO_TEST_CASE(ReplicatedProducer)
util::DummyClientFace face2(m_io, m_keyChain, {true, true}); util::DummyClientFace face2(m_io, m_keyChain, {true, true});
PartialProducer replicatedProducer(face2, m_keyChain, 40, syncPrefix, userPrefix); PartialProducer replicatedProducer(face2, m_keyChain, 40, syncPrefix, userPrefix);
for (int i = 1; i < 10; i++) { for (int i = 1; i < 10; i++) {
replicatedProducer.addUserNode("testUser-" + std::to_string(i)); replicatedProducer.addUserNode("testUser-" + std::to_string(i));
} }
advanceClocks(ndn::time::milliseconds(10)); advanceClocks(ndn::time::milliseconds(10));
replicatedProducer.publishName("testUser-2"); replicatedProducer.publishName("testUser-2");