Move non-public classes and functions to psync::detail namespace
CompressionScheme, CompressionError, and MissingDataInfo, which are public types, are instead moved to a new header PSync/common.hpp Change-Id: If1b8cb037cb321ff32c080c67df9dc3689223c00
This commit is contained in:
1 parent
aafef2bd91
commit
db7895644a
28 files changed
+304
-305
No files matched your search
@@ -0,0 +1,66 @@
|
||||
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
|
||||
/*
|
||||
* Copyright (c) 2014-2020, The University of Memphis
|
||||
*
|
||||
* This file is part of PSync.
|
||||
* See AUTHORS.md for complete list of PSync authors and contributors.
|
||||
*
|
||||
* PSync is free software: you can redistribute it and/or modify it under the terms
|
||||
* of the GNU Lesser General Public License as published by the Free Software Foundation,
|
||||
* either version 3 of the License, or (at your option) any later version.
|
||||
*
|
||||
* PSync is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
|
||||
* without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
|
||||
* PURPOSE. See the GNU Lesser General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Lesser General Public License along with
|
||||
* PSync, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
#ifndef PSYNC_COMMON_HPP
|
||||
#define PSYNC_COMMON_HPP
|
||||
|
||||
#include "PSync/detail/config.hpp"
|
||||
|
||||
#include <ndn-cxx/name.hpp>
|
||||
|
||||
#include <cstddef>
|
||||
#include <cstdint>
|
||||
#include <functional>
|
||||
#include <stdexcept>
|
||||
#include <vector>
|
||||
|
||||
namespace psync {
|
||||
|
||||
enum class CompressionScheme {
|
||||
NONE,
|
||||
ZLIB,
|
||||
GZIP,
|
||||
BZIP2,
|
||||
LZMA,
|
||||
ZSTD,
|
||||
#ifdef PSYNC_HAVE_ZLIB
|
||||
DEFAULT = ZLIB
|
||||
#else
|
||||
DEFAULT = NONE
|
||||
#endif
|
||||
};
|
||||
|
||||
class CompressionError : public std::runtime_error
|
||||
{
|
||||
public:
|
||||
using std::runtime_error::runtime_error;
|
||||
};
|
||||
|
||||
struct MissingDataInfo
|
||||
{
|
||||
ndn::Name prefix;
|
||||
uint64_t lowSeq;
|
||||
uint64_t highSeq;
|
||||
};
|
||||
|
||||
using UpdateCallback = std::function<void(const std::vector<MissingDataInfo>&)>;
|
||||
|
||||
} // namespace psync
|
||||
|
||||
#endif // PSYNC_COMMON_HPP
|
||||
+6
-10
@@ -20,10 +20,8 @@
|
||||
#include "PSync/consumer.hpp"
|
||||
#include "PSync/detail/state.hpp"
|
||||
|
||||
#include <ndn-cxx/util/logger.hpp>
|
||||
#include <ndn-cxx/security/validator-null.hpp>
|
||||
|
||||
#include <boost/algorithm/string.hpp>
|
||||
#include <ndn-cxx/util/logger.hpp>
|
||||
|
||||
namespace psync {
|
||||
|
||||
@@ -65,7 +63,7 @@ Consumer::addSubscription(const ndn::Name& prefix, uint64_t seqNo, bool callSync
|
||||
m_bloomFilter.insert(prefix.toUri());
|
||||
|
||||
if (callSyncDataCb && seqNo != 0) {
|
||||
m_onUpdate({MissingDataInfo(prefix, seqNo, seqNo)});
|
||||
m_onUpdate({{prefix, seqNo, seqNo}});
|
||||
}
|
||||
|
||||
return true;
|
||||
@@ -134,8 +132,7 @@ Consumer::onHelloData(const ndn::ConstBufferPtr& bufferPtr)
|
||||
|
||||
NDN_LOG_TRACE("m_iblt: " << std::hash<std::string>{}(m_iblt.toUri()));
|
||||
|
||||
State state{ndn::Block{bufferPtr}};
|
||||
|
||||
detail::State state{ndn::Block(bufferPtr)};
|
||||
std::vector<MissingDataInfo> updates;
|
||||
std::map<ndn::Name, uint64_t> availableSubscriptions;
|
||||
|
||||
@@ -148,7 +145,7 @@ Consumer::onHelloData(const ndn::ConstBufferPtr& bufferPtr)
|
||||
// m_prefixes (see addSubscription). So [] operator is safe to use.
|
||||
if (isSubscribed(prefix) && seq > m_prefixes[prefix]) {
|
||||
// In case we are behind on this prefix and consumer is subscribed to it
|
||||
updates.emplace_back(prefix, m_prefixes[prefix] + 1, seq);
|
||||
updates.push_back({prefix, m_prefixes[prefix] + 1, seq});
|
||||
m_prefixes[prefix] = seq;
|
||||
}
|
||||
availableSubscriptions.emplace(prefix, seq);
|
||||
@@ -233,8 +230,7 @@ Consumer::onSyncData(const ndn::ConstBufferPtr& bufferPtr)
|
||||
// Extract IBF from sync data name which is the last component
|
||||
m_iblt = m_syncDataName.getSubName(m_syncDataName.size() - 1, 1);
|
||||
|
||||
State state{ndn::Block{bufferPtr}};
|
||||
|
||||
detail::State state{ndn::Block(bufferPtr)};
|
||||
std::vector<MissingDataInfo> updates;
|
||||
|
||||
for (const auto& content : state) {
|
||||
@@ -244,7 +240,7 @@ Consumer::onSyncData(const ndn::ConstBufferPtr& bufferPtr)
|
||||
if (m_prefixes.find(prefix) == m_prefixes.end() || seq > m_prefixes[prefix]) {
|
||||
// If this is just the next seq number then we had already informed the consumer about
|
||||
// the previous sequence number and hence seq low and seq high should be equal to current seq
|
||||
updates.emplace_back(prefix, m_prefixes[prefix] + 1, seq);
|
||||
updates.push_back({prefix, m_prefixes[prefix] + 1, seq});
|
||||
m_prefixes[prefix] = seq;
|
||||
}
|
||||
// Else updates will be empty and consumer will not be notified.
|
||||
|
||||
+4
-6
@@ -20,9 +20,9 @@
|
||||
#ifndef PSYNC_CONSUMER_HPP
|
||||
#define PSYNC_CONSUMER_HPP
|
||||
|
||||
#include "PSync/common.hpp"
|
||||
#include "PSync/detail/access-specifiers.hpp"
|
||||
#include "PSync/detail/bloom-filter.hpp"
|
||||
#include "PSync/detail/util.hpp"
|
||||
|
||||
#include <ndn-cxx/face.hpp>
|
||||
#include <ndn-cxx/util/random.hpp>
|
||||
@@ -31,14 +31,12 @@
|
||||
#include <ndn-cxx/util/time.hpp>
|
||||
|
||||
#include <map>
|
||||
#include <vector>
|
||||
|
||||
namespace psync {
|
||||
|
||||
using namespace ndn::literals::time_literals;
|
||||
using namespace ndn::time_literals;
|
||||
|
||||
typedef std::function<void(const std::map<ndn::Name, uint64_t>&)> ReceiveHelloCallback;
|
||||
typedef std::function<void(const std::vector<MissingDataInfo>&)> UpdateCallback;
|
||||
using ReceiveHelloCallback = std::function<void(const std::map<ndn::Name, uint64_t>&)>;
|
||||
|
||||
const ndn::time::milliseconds HELLO_INTEREST_LIFETIME = 1_s;
|
||||
const ndn::time::milliseconds SYNC_INTEREST_LIFETIME = 1_s;
|
||||
@@ -190,7 +188,7 @@ PSYNC_PUBLIC_WITH_TESTS_ELSE_PRIVATE:
|
||||
UpdateCallback m_onUpdate;
|
||||
|
||||
// Bloom filter is used to store application/user's subscription list.
|
||||
BloomFilter m_bloomFilter;
|
||||
detail::BloomFilter m_bloomFilter;
|
||||
|
||||
ndn::time::milliseconds m_helloInterestLifetime;
|
||||
ndn::time::milliseconds m_syncInterestLifetime;
|
||||
|
||||
@@ -50,11 +50,11 @@
|
||||
|
||||
#include <algorithm>
|
||||
#include <cmath>
|
||||
#include <cstddef>
|
||||
#include <cstdlib>
|
||||
#include <limits>
|
||||
|
||||
namespace psync {
|
||||
namespace detail {
|
||||
|
||||
// https://github.com/ArashPartow/bloom
|
||||
|
||||
@@ -376,4 +376,5 @@ BloomFilter::generate_unique_salt()
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace detail
|
||||
} // namespace psync
|
||||
@@ -53,6 +53,7 @@
|
||||
#include <vector>
|
||||
|
||||
namespace psync {
|
||||
namespace detail {
|
||||
|
||||
class bloom_parameters;
|
||||
|
||||
@@ -141,6 +142,7 @@ private:
|
||||
double desired_false_positive_probability_ = 0.0;
|
||||
};
|
||||
|
||||
} // namespace detail
|
||||
} // namespace psync
|
||||
|
||||
#endif // PSYNC_DETAIL_BLOOM_FILTER_HPP
|
||||
+42
-42
@@ -44,10 +44,12 @@
|
||||
*/
|
||||
|
||||
#include "PSync/detail/iblt.hpp"
|
||||
#include "PSync/detail/util.hpp"
|
||||
|
||||
#include <ndn-cxx/util/exception.hpp>
|
||||
|
||||
namespace psync {
|
||||
namespace detail {
|
||||
|
||||
namespace be = boost::endian;
|
||||
|
||||
@@ -181,47 +183,6 @@ IBLT::operator-(const IBLT& other) const
|
||||
return result;
|
||||
}
|
||||
|
||||
bool
|
||||
operator==(const IBLT& iblt1, const IBLT& iblt2)
|
||||
{
|
||||
auto iblt1HashTable = iblt1.getHashTable();
|
||||
auto iblt2HashTable = iblt2.getHashTable();
|
||||
if (iblt1HashTable.size() != iblt2HashTable.size()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
size_t N = iblt1HashTable.size();
|
||||
|
||||
for (size_t i = 0; i < N; i++) {
|
||||
if (iblt1HashTable[i].count != iblt2HashTable[i].count ||
|
||||
iblt1HashTable[i].keySum != iblt2HashTable[i].keySum ||
|
||||
iblt1HashTable[i].keyCheck != iblt2HashTable[i].keyCheck)
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
bool
|
||||
operator!=(const IBLT& iblt1, const IBLT& iblt2)
|
||||
{
|
||||
return !(iblt1 == iblt2);
|
||||
}
|
||||
|
||||
std::ostream&
|
||||
operator<<(std::ostream& out, const IBLT& iblt)
|
||||
{
|
||||
out << "count keySum keyCheckMatch\n";
|
||||
for (const auto& entry : iblt.getHashTable()) {
|
||||
out << entry.count << " " << entry.keySum << " ";
|
||||
out << ((murmurHash3(N_HASHCHECK, entry.keySum) == entry.keyCheck) ||
|
||||
(entry.isEmpty())? "true" : "false");
|
||||
out << "\n";
|
||||
}
|
||||
|
||||
return out;
|
||||
}
|
||||
|
||||
void
|
||||
IBLT::appendToName(ndn::Name& name) const
|
||||
{
|
||||
@@ -256,7 +217,6 @@ IBLT::extractValueFromName(const ndn::name::Component& ibltName) const
|
||||
}
|
||||
|
||||
size_t n = decompressedBuf->size() / 4;
|
||||
|
||||
std::vector<uint32_t> values(n, 0);
|
||||
|
||||
for (size_t i = 0; i < n; i++) {
|
||||
@@ -268,4 +228,44 @@ IBLT::extractValueFromName(const ndn::name::Component& ibltName) const
|
||||
return values;
|
||||
}
|
||||
|
||||
bool
|
||||
operator==(const IBLT& iblt1, const IBLT& iblt2)
|
||||
{
|
||||
auto iblt1HashTable = iblt1.getHashTable();
|
||||
auto iblt2HashTable = iblt2.getHashTable();
|
||||
if (iblt1HashTable.size() != iblt2HashTable.size()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
size_t N = iblt1HashTable.size();
|
||||
|
||||
for (size_t i = 0; i < N; i++) {
|
||||
if (iblt1HashTable[i].count != iblt2HashTable[i].count ||
|
||||
iblt1HashTable[i].keySum != iblt2HashTable[i].keySum ||
|
||||
iblt1HashTable[i].keyCheck != iblt2HashTable[i].keyCheck)
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
bool
|
||||
operator!=(const IBLT& iblt1, const IBLT& iblt2)
|
||||
{
|
||||
return !(iblt1 == iblt2);
|
||||
}
|
||||
|
||||
std::ostream&
|
||||
operator<<(std::ostream& os, const IBLT& iblt)
|
||||
{
|
||||
os << "count keySum keyCheckMatch\n";
|
||||
for (const auto& entry : iblt.getHashTable()) {
|
||||
os << entry.count << " " << entry.keySum << " "
|
||||
<< ((entry.isEmpty() || murmurHash3(N_HASHCHECK, entry.keySum) == entry.keyCheck) ? "true" : "false")
|
||||
<< "\n";
|
||||
}
|
||||
return os;
|
||||
}
|
||||
|
||||
} // namespace detail
|
||||
} // namespace psync
|
||||
@@ -46,16 +46,15 @@
|
||||
#ifndef PSYNC_DETAIL_IBLT_HPP
|
||||
#define PSYNC_DETAIL_IBLT_HPP
|
||||
|
||||
#include "PSync/detail/util.hpp"
|
||||
#include "PSync/common.hpp"
|
||||
|
||||
#include <ndn-cxx/name.hpp>
|
||||
|
||||
#include <inttypes.h>
|
||||
#include <set>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
namespace psync {
|
||||
namespace detail {
|
||||
|
||||
class HashTableEntry
|
||||
{
|
||||
@@ -175,8 +174,9 @@ bool
|
||||
operator!=(const IBLT& iblt1, const IBLT& iblt2);
|
||||
|
||||
std::ostream&
|
||||
operator<<(std::ostream& out, const IBLT& iblt);
|
||||
operator<<(std::ostream& os, const IBLT& iblt);
|
||||
|
||||
} // namespace detail
|
||||
} // namespace psync
|
||||
|
||||
#endif // PSYNC_DETAIL_IBLT_HPP
|
||||
@@ -23,6 +23,7 @@
|
||||
#include <ndn-cxx/util/ostream-joiner.hpp>
|
||||
|
||||
namespace psync {
|
||||
namespace detail {
|
||||
|
||||
State::State(const ndn::Block& block)
|
||||
{
|
||||
@@ -107,4 +108,5 @@ operator<<(std::ostream& os, const State& state)
|
||||
return os;
|
||||
}
|
||||
|
||||
} // namespace detail
|
||||
} // namespace psync
|
||||
@@ -23,7 +23,6 @@
|
||||
#include <ndn-cxx/name.hpp>
|
||||
|
||||
namespace psync {
|
||||
|
||||
namespace tlv {
|
||||
|
||||
enum {
|
||||
@@ -32,13 +31,16 @@ enum {
|
||||
|
||||
} // namespace tlv
|
||||
|
||||
namespace detail {
|
||||
|
||||
class State
|
||||
{
|
||||
public:
|
||||
explicit State(const ndn::Block& block);
|
||||
|
||||
State() = default;
|
||||
|
||||
explicit
|
||||
State(const ndn::Block& block);
|
||||
|
||||
void
|
||||
addContent(const ndn::Name& prefix);
|
||||
|
||||
@@ -80,6 +82,7 @@ NDN_CXX_DECLARE_WIRE_ENCODE_INSTANTIATIONS(State);
|
||||
std::ostream&
|
||||
operator<<(std::ostream& os, const State& State);
|
||||
|
||||
} // namespace detail
|
||||
} // namespace psync
|
||||
|
||||
#endif // PSYNC_DETAIL_STATE_HPP
|
||||
+57
-65
@@ -18,8 +18,8 @@
|
||||
*
|
||||
* murmurHash3 was written by Austin Appleby, and is placed in the public
|
||||
* domain. The author hereby disclaims copyright to this source code.
|
||||
* https://github.com/aappleby/smhasher/blob/master/src/murmurHash3.cpp
|
||||
**/
|
||||
* https://github.com/aappleby/smhasher/blob/master/src/MurmurHash3.cpp
|
||||
*/
|
||||
|
||||
#include "PSync/detail/util.hpp"
|
||||
|
||||
@@ -27,6 +27,7 @@
|
||||
#include <ndn-cxx/util/backports.hpp>
|
||||
#include <ndn-cxx/util/exception.hpp>
|
||||
|
||||
#include <boost/iostreams/copy.hpp>
|
||||
#include <boost/iostreams/device/array.hpp>
|
||||
#include <boost/iostreams/filtering_stream.hpp>
|
||||
#ifdef PSYNC_HAVE_ZLIB
|
||||
@@ -44,32 +45,36 @@
|
||||
#ifdef PSYNC_HAVE_ZSTD
|
||||
#include <boost/iostreams/filter/zstd.hpp>
|
||||
#endif
|
||||
#include <boost/iostreams/copy.hpp>
|
||||
|
||||
namespace psync {
|
||||
namespace detail {
|
||||
|
||||
namespace bio = boost::iostreams;
|
||||
|
||||
static uint32_t
|
||||
static inline uint32_t
|
||||
ROTL32(uint32_t x, int8_t r)
|
||||
{
|
||||
return (x << r) | (x >> (32 - r));
|
||||
}
|
||||
|
||||
uint32_t
|
||||
murmurHash3(uint32_t nHashSeed, const std::vector<unsigned char>& vDataToHash)
|
||||
murmurHash3(const void* key, size_t len, uint32_t seed)
|
||||
{
|
||||
uint32_t h1 = nHashSeed;
|
||||
const uint8_t * data = (const uint8_t*)key;
|
||||
const int nblocks = len / 4;
|
||||
|
||||
uint32_t h1 = seed;
|
||||
|
||||
const uint32_t c1 = 0xcc9e2d51;
|
||||
const uint32_t c2 = 0x1b873593;
|
||||
|
||||
const size_t nblocks = vDataToHash.size() / 4;
|
||||
|
||||
//----------
|
||||
// body
|
||||
const uint32_t * blocks = (const uint32_t *)(&vDataToHash[0] + nblocks*4);
|
||||
|
||||
for (size_t i = -nblocks; i; i++) {
|
||||
const uint32_t * blocks = (const uint32_t *)(data + nblocks*4);
|
||||
|
||||
for (int i = -nblocks; i; i++)
|
||||
{
|
||||
uint32_t k1 = blocks[i];
|
||||
|
||||
k1 *= c1;
|
||||
@@ -83,27 +88,25 @@ murmurHash3(uint32_t nHashSeed, const std::vector<unsigned char>& vDataToHash)
|
||||
|
||||
//----------
|
||||
// tail
|
||||
const uint8_t * tail = (const uint8_t*)(&vDataToHash[0] + nblocks*4);
|
||||
|
||||
const uint8_t * tail = (const uint8_t*)(data + nblocks*4);
|
||||
|
||||
uint32_t k1 = 0;
|
||||
|
||||
switch (vDataToHash.size() & 3) {
|
||||
case 3:
|
||||
k1 ^= tail[2] << 16;
|
||||
switch (len & 3)
|
||||
{
|
||||
case 3: k1 ^= tail[2] << 16;
|
||||
NDN_CXX_FALLTHROUGH;
|
||||
|
||||
case 2:
|
||||
k1 ^= tail[1] << 8;
|
||||
case 2: k1 ^= tail[1] << 8;
|
||||
NDN_CXX_FALLTHROUGH;
|
||||
|
||||
case 1:
|
||||
k1 ^= tail[0];
|
||||
case 1: k1 ^= tail[0];
|
||||
k1 *= c1; k1 = ROTL32(k1,15); k1 *= c2; h1 ^= k1;
|
||||
}
|
||||
|
||||
//----------
|
||||
// finalization
|
||||
h1 ^= vDataToHash.size();
|
||||
|
||||
h1 ^= len;
|
||||
h1 ^= h1 >> 16;
|
||||
h1 *= 0x85ebca6b;
|
||||
h1 ^= h1 >> 13;
|
||||
@@ -113,65 +116,52 @@ murmurHash3(uint32_t nHashSeed, const std::vector<unsigned char>& vDataToHash)
|
||||
return h1;
|
||||
}
|
||||
|
||||
uint32_t
|
||||
murmurHash3(uint32_t nHashSeed, const std::string& str)
|
||||
{
|
||||
return murmurHash3(nHashSeed, std::vector<unsigned char>(str.begin(), str.end()));
|
||||
}
|
||||
|
||||
uint32_t
|
||||
murmurHash3(uint32_t nHashSeed, uint32_t value)
|
||||
{
|
||||
return murmurHash3(nHashSeed,
|
||||
std::vector<unsigned char>((unsigned char*)&value,
|
||||
(unsigned char*)&value + sizeof(uint32_t)));
|
||||
}
|
||||
|
||||
std::shared_ptr<ndn::Buffer>
|
||||
compress(CompressionScheme scheme, const uint8_t* buffer, size_t bufferSize)
|
||||
{
|
||||
ndn::OBufferStream out;
|
||||
bio::filtering_streambuf<bio::input> in;
|
||||
|
||||
switch (scheme) {
|
||||
case CompressionScheme::ZLIB:
|
||||
#ifdef PSYNC_HAVE_ZLIB
|
||||
in.push(bio::zlib_compressor(bio::zlib::best_compression));
|
||||
#else
|
||||
NDN_THROW(Error("ZLIB compression not supported!"));
|
||||
#endif
|
||||
break;
|
||||
#else
|
||||
NDN_THROW(CompressionError("ZLIB compression not supported!"));
|
||||
#endif
|
||||
|
||||
case CompressionScheme::GZIP:
|
||||
#ifdef PSYNC_HAVE_GZIP
|
||||
in.push(bio::gzip_compressor(bio::gzip::best_compression));
|
||||
#else
|
||||
NDN_THROW(Error("GZIP compression not supported!"));
|
||||
#endif
|
||||
break;
|
||||
#else
|
||||
NDN_THROW(CompressionError("GZIP compression not supported!"));
|
||||
#endif
|
||||
|
||||
case CompressionScheme::BZIP2:
|
||||
#ifdef PSYNC_HAVE_BZIP2
|
||||
in.push(bio::bzip2_compressor());
|
||||
#else
|
||||
NDN_THROW(Error("BZIP2 compression not supported!"));
|
||||
#endif
|
||||
break;
|
||||
#else
|
||||
NDN_THROW(CompressionError("BZIP2 compression not supported!"));
|
||||
#endif
|
||||
|
||||
case CompressionScheme::LZMA:
|
||||
#ifdef PSYNC_HAVE_LZMA
|
||||
in.push(bio::lzma_compressor(bio::lzma::best_compression));
|
||||
#else
|
||||
NDN_THROW(Error("LZMA compression not supported!"));
|
||||
#endif
|
||||
break;
|
||||
#else
|
||||
NDN_THROW(CompressionError("LZMA compression not supported!"));
|
||||
#endif
|
||||
|
||||
case CompressionScheme::ZSTD:
|
||||
#ifdef PSYNC_HAVE_ZSTD
|
||||
in.push(bio::zstd_compressor(bio::zstd::best_compression));
|
||||
#else
|
||||
NDN_THROW(Error("ZSTD compression not supported!"));
|
||||
#endif
|
||||
break;
|
||||
#else
|
||||
NDN_THROW(CompressionError("ZSTD compression not supported!"));
|
||||
#endif
|
||||
|
||||
case CompressionScheme::NONE:
|
||||
break;
|
||||
@@ -187,46 +177,47 @@ decompress(CompressionScheme scheme, const uint8_t* buffer, size_t bufferSize)
|
||||
{
|
||||
ndn::OBufferStream out;
|
||||
bio::filtering_streambuf<bio::input> in;
|
||||
|
||||
switch (scheme) {
|
||||
case CompressionScheme::ZLIB:
|
||||
#ifdef PSYNC_HAVE_ZLIB
|
||||
in.push(bio::zlib_decompressor());
|
||||
#else
|
||||
NDN_THROW(Error("ZLIB decompression not supported!"));
|
||||
#endif
|
||||
break;
|
||||
#else
|
||||
NDN_THROW(CompressionError("ZLIB decompression not supported!"));
|
||||
#endif
|
||||
|
||||
case CompressionScheme::GZIP:
|
||||
#ifdef PSYNC_HAVE_GZIP
|
||||
in.push(bio::gzip_decompressor());
|
||||
#else
|
||||
NDN_THROW(Error("GZIP compression not supported!"));
|
||||
#endif
|
||||
break;
|
||||
#else
|
||||
NDN_THROW(CompressionError("GZIP compression not supported!"));
|
||||
#endif
|
||||
|
||||
case CompressionScheme::BZIP2:
|
||||
#ifdef PSYNC_HAVE_BZIP2
|
||||
in.push(bio::bzip2_decompressor());
|
||||
#else
|
||||
NDN_THROW(Error("BZIP2 compression not supported!"));
|
||||
#endif
|
||||
break;
|
||||
#else
|
||||
NDN_THROW(CompressionError("BZIP2 compression not supported!"));
|
||||
#endif
|
||||
|
||||
case CompressionScheme::LZMA:
|
||||
#ifdef PSYNC_HAVE_LZMA
|
||||
in.push(bio::lzma_decompressor());
|
||||
#else
|
||||
NDN_THROW(Error("LZMA compression not supported!"));
|
||||
#endif
|
||||
break;
|
||||
#else
|
||||
NDN_THROW(CompressionError("LZMA compression not supported!"));
|
||||
#endif
|
||||
|
||||
case CompressionScheme::ZSTD:
|
||||
#ifdef PSYNC_HAVE_ZSTD
|
||||
in.push(bio::zstd_decompressor());
|
||||
#else
|
||||
NDN_THROW(Error("ZSTD compression not supported!"));
|
||||
#endif
|
||||
break;
|
||||
#else
|
||||
NDN_THROW(CompressionError("ZSTD compression not supported!"));
|
||||
#endif
|
||||
|
||||
case CompressionScheme::NONE:
|
||||
break;
|
||||
@@ -237,4 +228,5 @@ decompress(CompressionScheme scheme, const uint8_t* buffer, size_t bufferSize)
|
||||
return out.buf();
|
||||
}
|
||||
|
||||
} // namespace detail
|
||||
} // namespace psync
|
||||
+14
-45
@@ -15,60 +15,34 @@
|
||||
*
|
||||
* You should have received a copy of the GNU Lesser General Public License along with
|
||||
* PSync, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
|
||||
*
|
||||
* murmurHash3 was written by Austin Appleby, and is placed in the public
|
||||
* domain. The author hereby disclaims copyright to this source code.
|
||||
* https://github.com/aappleby/smhasher/blob/master/src/murmurHash3.cpp
|
||||
*/
|
||||
|
||||
#ifndef PSYNC_DETAIL_UTIL_HPP
|
||||
#define PSYNC_DETAIL_UTIL_HPP
|
||||
|
||||
#include "PSync/detail/config.hpp"
|
||||
#include "PSync/common.hpp"
|
||||
|
||||
#include <ndn-cxx/name.hpp>
|
||||
#include <ndn-cxx/encoding/buffer.hpp>
|
||||
|
||||
#include <inttypes.h>
|
||||
#include <vector>
|
||||
#include <string>
|
||||
|
||||
namespace psync {
|
||||
namespace detail {
|
||||
|
||||
uint32_t
|
||||
murmurHash3(uint32_t nHashSeed, const std::vector<unsigned char>& vDataToHash);
|
||||
murmurHash3(const void* key, size_t len, uint32_t seed);
|
||||
|
||||
uint32_t
|
||||
murmurHash3(uint32_t nHashSeed, const std::string& str);
|
||||
|
||||
uint32_t
|
||||
murmurHash3(uint32_t nHashSeed, uint32_t value);
|
||||
|
||||
struct MissingDataInfo
|
||||
inline uint32_t
|
||||
murmurHash3(uint32_t seed, const std::string& str)
|
||||
{
|
||||
MissingDataInfo(const ndn::Name& prefix, uint64_t lowSeq, uint64_t highSeq)
|
||||
: prefix(prefix)
|
||||
, lowSeq(lowSeq)
|
||||
, highSeq(highSeq)
|
||||
{}
|
||||
return murmurHash3(str.data(), str.size(), seed);
|
||||
}
|
||||
|
||||
ndn::Name prefix;
|
||||
uint64_t lowSeq;
|
||||
uint64_t highSeq;
|
||||
};
|
||||
|
||||
enum class CompressionScheme {
|
||||
NONE,
|
||||
ZLIB,
|
||||
GZIP,
|
||||
BZIP2,
|
||||
LZMA,
|
||||
ZSTD,
|
||||
#ifdef PSYNC_HAVE_ZLIB
|
||||
DEFAULT = ZLIB
|
||||
#else
|
||||
DEFAULT = NONE
|
||||
#endif
|
||||
};
|
||||
inline uint32_t
|
||||
murmurHash3(uint32_t seed, uint32_t value)
|
||||
{
|
||||
return murmurHash3(&value, sizeof(value), seed);
|
||||
}
|
||||
|
||||
std::shared_ptr<ndn::Buffer>
|
||||
compress(CompressionScheme scheme, const uint8_t* buffer, size_t bufferSize);
|
||||
@@ -76,12 +50,7 @@ compress(CompressionScheme scheme, const uint8_t* buffer, size_t bufferSize);
|
||||
std::shared_ptr<ndn::Buffer>
|
||||
decompress(CompressionScheme scheme, const uint8_t* buffer, size_t bufferSize);
|
||||
|
||||
class Error : public std::runtime_error
|
||||
{
|
||||
public:
|
||||
using std::runtime_error::runtime_error;
|
||||
};
|
||||
|
||||
} // namespace detail
|
||||
} // namespace psync
|
||||
|
||||
#endif // PSYNC_DETAIL_UTIL_HPP
|
||||
+25
-31
@@ -18,14 +18,14 @@
|
||||
*/
|
||||
|
||||
#include "PSync/full-producer.hpp"
|
||||
#include "PSync/detail/state.hpp"
|
||||
#include "PSync/detail/util.hpp"
|
||||
|
||||
#include <ndn-cxx/security/validator-null.hpp>
|
||||
#include <ndn-cxx/util/logger.hpp>
|
||||
#include <ndn-cxx/util/segment-fetcher.hpp>
|
||||
#include <ndn-cxx/security/validator-null.hpp>
|
||||
|
||||
#include <cstring>
|
||||
#include <limits>
|
||||
#include <functional>
|
||||
|
||||
namespace psync {
|
||||
|
||||
@@ -118,7 +118,7 @@ FullProducer::sendSyncInterest()
|
||||
});
|
||||
|
||||
m_fetcher->onError.connect([this] (uint32_t errorCode, const std::string& msg) {
|
||||
NDN_LOG_ERROR("Cannot fetch sync data, error: " << errorCode << " message: " << msg);
|
||||
NDN_LOG_ERROR("Cannot fetch sync data, error: " << errorCode << ", message: " << msg);
|
||||
// We would like to recover from errors like NoRoute NACK quicker than sync Interest timeout.
|
||||
// We don't react to Interest timeout here as we have scheduled the next sync Interest
|
||||
// to be sent in half the sync Interest lifetime + jitter above. So we would react to
|
||||
@@ -159,10 +159,10 @@ FullProducer::onSyncInterest(const ndn::Name& prefixName, const ndn::Interest& i
|
||||
|
||||
ndn::name::Component ibltName = interestName.get(interestName.size() - 1);
|
||||
|
||||
NDN_LOG_DEBUG("Full Sync Interest Received, nonce: " << interest.getNonce() <<
|
||||
NDN_LOG_DEBUG("Full sync Interest received, nonce: " << interest.getNonce() <<
|
||||
", hash: " << std::hash<ndn::Name>{}(interestName));
|
||||
|
||||
IBLT iblt(m_expectedNumEntries, m_ibltCompression);
|
||||
detail::IBLT iblt(m_expectedNumEntries, m_ibltCompression);
|
||||
try {
|
||||
iblt.initialize(ibltName);
|
||||
}
|
||||
@@ -171,7 +171,7 @@ FullProducer::onSyncInterest(const ndn::Name& prefixName, const ndn::Interest& i
|
||||
return;
|
||||
}
|
||||
|
||||
IBLT diff = m_iblt - iblt;
|
||||
auto diff = m_iblt - iblt;
|
||||
|
||||
std::set<uint32_t> positive;
|
||||
std::set<uint32_t> negative;
|
||||
@@ -185,7 +185,7 @@ FullProducer::onSyncInterest(const ndn::Name& prefixName, const ndn::Interest& i
|
||||
// Or send if we can't get neither positive nor negative differences
|
||||
if (positive.size() + negative.size() >= m_threshold ||
|
||||
(positive.size() == 0 && negative.size() == 0)) {
|
||||
State state;
|
||||
detail::State state;
|
||||
for (const auto& content : m_prefixes) {
|
||||
if (content.second != 0) {
|
||||
state.addContent(ndn::Name(content.first).appendNumber(content.second));
|
||||
@@ -200,7 +200,7 @@ FullProducer::onSyncInterest(const ndn::Name& prefixName, const ndn::Interest& i
|
||||
}
|
||||
}
|
||||
|
||||
State state;
|
||||
detail::State state;
|
||||
for (const auto& hash : positive) {
|
||||
auto nameIt = m_biMap.left.find(hash);
|
||||
if (nameIt != m_biMap.left.end()) {
|
||||
@@ -219,10 +219,10 @@ FullProducer::onSyncInterest(const ndn::Name& prefixName, const ndn::Interest& i
|
||||
return;
|
||||
}
|
||||
|
||||
auto& entry = m_pendingEntries.emplace(interestName, PendingEntryInfoFull{iblt, {}}).first->second;
|
||||
auto& entry = m_pendingEntries.emplace(interestName, PendingEntryInfo{iblt, {}}).first->second;
|
||||
entry.expirationEvent = m_scheduler.schedule(interest.getInterestLifetime(),
|
||||
[this, interest] {
|
||||
NDN_LOG_TRACE("Erase Pending Interest " << interest.getNonce());
|
||||
NDN_LOG_TRACE("Erase pending Interest " << interest.getNonce());
|
||||
m_pendingEntries.erase(interest.getName());
|
||||
});
|
||||
}
|
||||
@@ -238,7 +238,7 @@ FullProducer::sendSyncData(const ndn::Name& name, const ndn::Block& block)
|
||||
// TODO: Remove appending of hash - serves no purpose to the receiver
|
||||
ndn::Name dataName(ndn::Name(name).appendNumber(std::hash<ndn::Name>{}(nameWithIblt)));
|
||||
|
||||
auto content = compress(m_contentCompression, block.wire(), block.size());
|
||||
auto content = detail::compress(m_contentCompression, block.wire(), block.size());
|
||||
|
||||
// checking if our own interest got satisfied
|
||||
if (m_outstandingInterestName == name) {
|
||||
@@ -250,7 +250,7 @@ FullProducer::sendSyncData(const ndn::Name& name, const ndn::Block& block)
|
||||
m_outstandingInterestName = ndn::Name("");
|
||||
}
|
||||
|
||||
NDN_LOG_DEBUG("Sending Sync Data");
|
||||
NDN_LOG_DEBUG("Sending sync Data");
|
||||
|
||||
// Send data after removing pending sync interest on face
|
||||
m_segmentPublisher.publish(name, dataName, content, m_syncReplyFreshness);
|
||||
@@ -259,7 +259,7 @@ FullProducer::sendSyncData(const ndn::Name& name, const ndn::Block& block)
|
||||
sendSyncInterest();
|
||||
}
|
||||
else {
|
||||
NDN_LOG_DEBUG("Sending Sync Data");
|
||||
NDN_LOG_DEBUG("Sending sync Data");
|
||||
m_segmentPublisher.publish(name, dataName, content, m_syncReplyFreshness);
|
||||
}
|
||||
}
|
||||
@@ -269,26 +269,26 @@ FullProducer::onSyncData(const ndn::Interest& interest, const ndn::ConstBufferPt
|
||||
{
|
||||
deletePendingInterests(interest.getName());
|
||||
|
||||
State state;
|
||||
detail::State state;
|
||||
try {
|
||||
auto decompressed = decompress(m_contentCompression, bufferPtr->data(), bufferPtr->size());
|
||||
auto decompressed = detail::decompress(m_contentCompression, bufferPtr->data(), bufferPtr->size());
|
||||
state.wireDecode(ndn::Block(std::move(decompressed)));
|
||||
}
|
||||
catch (const std::exception& e) {
|
||||
NDN_LOG_ERROR("Cannot parse received Sync Data: " << e.what());
|
||||
NDN_LOG_ERROR("Cannot parse received sync Data: " << e.what());
|
||||
return;
|
||||
}
|
||||
|
||||
std::vector<MissingDataInfo> updates;
|
||||
NDN_LOG_DEBUG("Sync Data received: " << state);
|
||||
|
||||
NDN_LOG_DEBUG("Sync Data Received: " << state);
|
||||
std::vector<MissingDataInfo> updates;
|
||||
|
||||
for (const auto& content : state) {
|
||||
ndn::Name prefix = content.getPrefix(-1);
|
||||
uint64_t seq = content.get(content.size() - 1).toNumber();
|
||||
|
||||
if (m_prefixes.find(prefix) == m_prefixes.end() || m_prefixes[prefix] < seq) {
|
||||
updates.push_back(MissingDataInfo{prefix, m_prefixes[prefix] + 1, seq});
|
||||
updates.push_back({prefix, m_prefixes[prefix] + 1, seq});
|
||||
updateSeqNo(prefix, seq);
|
||||
// We should not call satisfyPendingSyncInterests here because we just
|
||||
// got data and deleted pending interest by calling deletePendingFullSyncInterests
|
||||
@@ -315,8 +315,8 @@ FullProducer::satisfyPendingInterests()
|
||||
NDN_LOG_DEBUG("Satisfying full sync interest: " << m_pendingEntries.size());
|
||||
|
||||
for (auto it = m_pendingEntries.begin(); it != m_pendingEntries.end();) {
|
||||
const PendingEntryInfoFull& entry = it->second;
|
||||
IBLT diff = m_iblt - entry.iblt;
|
||||
const auto& entry = it->second;
|
||||
auto diff = m_iblt - entry.iblt;
|
||||
std::set<uint32_t> positive;
|
||||
std::set<uint32_t> negative;
|
||||
|
||||
@@ -330,7 +330,7 @@ FullProducer::satisfyPendingInterests()
|
||||
}
|
||||
}
|
||||
|
||||
State state;
|
||||
detail::State state;
|
||||
for (const auto& hash : positive) {
|
||||
auto nameIt = m_biMap.left.find(hash);
|
||||
if (nameIt != m_biMap.left.end()) {
|
||||
@@ -354,15 +354,9 @@ FullProducer::satisfyPendingInterests()
|
||||
bool
|
||||
FullProducer::isFutureHash(const ndn::Name& prefix, const std::set<uint32_t>& negative)
|
||||
{
|
||||
uint32_t nextHash = murmurHash3(N_HASHCHECK,
|
||||
auto nextHash = detail::murmurHash3(detail::N_HASHCHECK,
|
||||
ndn::Name(prefix).appendNumber(m_prefixes[prefix] + 1).toUri());
|
||||
for (const auto& nHash : negative) {
|
||||
if (nHash == nextHash) {
|
||||
return true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
return negative.find(nextHash) != negative.end();
|
||||
}
|
||||
|
||||
void
|
||||
|
||||
+7
-18
@@ -21,31 +21,14 @@
|
||||
#define PSYNC_FULL_PRODUCER_HPP
|
||||
|
||||
#include "PSync/producer-base.hpp"
|
||||
#include "PSync/detail/state.hpp"
|
||||
|
||||
#include <map>
|
||||
#include <random>
|
||||
#include <set>
|
||||
|
||||
#include <ndn-cxx/face.hpp>
|
||||
#include <ndn-cxx/security/key-chain.hpp>
|
||||
#include <ndn-cxx/util/scheduler.hpp>
|
||||
#include <ndn-cxx/util/segment-fetcher.hpp>
|
||||
#include <ndn-cxx/util/time.hpp>
|
||||
|
||||
namespace psync {
|
||||
|
||||
// Name has to be different than PendingEntryInfo
|
||||
// used in partial-producer otherwise get strange segmentation-faults
|
||||
// when partial producer is destructed
|
||||
struct PendingEntryInfoFull
|
||||
{
|
||||
IBLT iblt;
|
||||
ndn::scheduler::ScopedEventId expirationEvent;
|
||||
};
|
||||
|
||||
typedef std::function<void(const std::vector<MissingDataInfo>&)> UpdateCallback;
|
||||
|
||||
const ndn::time::milliseconds SYNC_INTEREST_LIFTIME = 1_s;
|
||||
|
||||
/**
|
||||
@@ -187,7 +170,13 @@ private:
|
||||
isFutureHash(const ndn::Name& prefix, const std::set<uint32_t>& negative);
|
||||
|
||||
private:
|
||||
std::map<ndn::Name, PendingEntryInfoFull> m_pendingEntries;
|
||||
struct PendingEntryInfo
|
||||
{
|
||||
detail::IBLT iblt;
|
||||
ndn::scheduler::ScopedEventId expirationEvent;
|
||||
};
|
||||
|
||||
std::map<ndn::Name, PendingEntryInfo> m_pendingEntries;
|
||||
ndn::time::milliseconds m_syncInterestLifetime;
|
||||
UpdateCallback m_onUpdate;
|
||||
ndn::scheduler::ScopedEventId m_scheduledSyncInterestId;
|
||||
|
||||
+10
-12
@@ -82,10 +82,9 @@ PartialProducer::onHelloInterest(const ndn::Name& prefix, const ndn::Interest& i
|
||||
|
||||
NDN_LOG_DEBUG("Hello Interest Received, nonce: " << interest);
|
||||
|
||||
State state;
|
||||
|
||||
for (const auto& prefix : m_prefixes) {
|
||||
state.addContent(ndn::Name(prefix.first).appendNumber(prefix.second));
|
||||
detail::State state;
|
||||
for (const auto& p : m_prefixes) {
|
||||
state.addContent(ndn::Name(p.first).appendNumber(p.second));
|
||||
}
|
||||
NDN_LOG_DEBUG("sending content p: " << state);
|
||||
|
||||
@@ -137,11 +136,10 @@ PartialProducer::onSyncInterest(const ndn::Name& prefix, const ndn::Interest& in
|
||||
return;
|
||||
}
|
||||
|
||||
BloomFilter bf;
|
||||
IBLT iblt(m_expectedNumEntries, m_ibltCompression);
|
||||
|
||||
detail::BloomFilter bf;
|
||||
detail::IBLT iblt(m_expectedNumEntries, m_ibltCompression);
|
||||
try {
|
||||
bf = BloomFilter(projectedCount, falsePositiveProb, bfName);
|
||||
bf = detail::BloomFilter(projectedCount, falsePositiveProb, bfName);
|
||||
iblt.initialize(ibltName);
|
||||
}
|
||||
catch (const std::exception& e) {
|
||||
@@ -150,7 +148,7 @@ PartialProducer::onSyncInterest(const ndn::Name& prefix, const ndn::Interest& in
|
||||
}
|
||||
|
||||
// get the difference
|
||||
IBLT diff = m_iblt - iblt;
|
||||
auto diff = m_iblt - iblt;
|
||||
|
||||
// non-empty positive means we have some elements that the others don't
|
||||
std::set<uint32_t> positive;
|
||||
@@ -169,7 +167,7 @@ PartialProducer::onSyncInterest(const ndn::Name& prefix, const ndn::Interest& in
|
||||
}
|
||||
|
||||
// generate content for Sync reply
|
||||
State state;
|
||||
detail::State state;
|
||||
NDN_LOG_TRACE("Size of positive set " << positive.size());
|
||||
NDN_LOG_TRACE("Size of negative set " << negative.size());
|
||||
for (const auto& hash : positive) {
|
||||
@@ -211,7 +209,7 @@ PartialProducer::satisfyPendingSyncInterests(const ndn::Name& prefix) {
|
||||
for (auto it = m_pendingEntries.begin(); it != m_pendingEntries.end();) {
|
||||
const PendingEntryInfo& entry = it->second;
|
||||
|
||||
IBLT diff = m_iblt - entry.iblt;
|
||||
auto diff = m_iblt - entry.iblt;
|
||||
std::set<uint32_t> positive;
|
||||
std::set<uint32_t> negative;
|
||||
|
||||
@@ -228,7 +226,7 @@ PartialProducer::satisfyPendingSyncInterests(const ndn::Name& prefix) {
|
||||
continue;
|
||||
}
|
||||
|
||||
State state;
|
||||
detail::State state;
|
||||
if (entry.bf.contains(prefix.toUri()) || positive.size() + negative.size() >= m_threshold) {
|
||||
if (entry.bf.contains(prefix.toUri())) {
|
||||
state.addContent(ndn::Name(prefix).appendNumber(m_prefixes[prefix]));
|
||||
|
||||
@@ -32,13 +32,6 @@
|
||||
|
||||
namespace psync {
|
||||
|
||||
struct PendingEntryInfo
|
||||
{
|
||||
BloomFilter bf;
|
||||
IBLT iblt;
|
||||
ndn::scheduler::ScopedEventId expirationEvent;
|
||||
};
|
||||
|
||||
const ndn::time::milliseconds HELLO_REPLY_FRESHNESS = 1_s;
|
||||
|
||||
/**
|
||||
@@ -122,6 +115,13 @@ PSYNC_PUBLIC_WITH_TESTS_ELSE_PRIVATE:
|
||||
onSyncInterest(const ndn::Name& prefix, const ndn::Interest& interest);
|
||||
|
||||
PSYNC_PUBLIC_WITH_TESTS_ELSE_PRIVATE:
|
||||
struct PendingEntryInfo
|
||||
{
|
||||
detail::BloomFilter bf;
|
||||
detail::IBLT iblt;
|
||||
ndn::scheduler::ScopedEventId expirationEvent;
|
||||
};
|
||||
|
||||
std::map<ndn::Name, PendingEntryInfo> m_pendingEntries;
|
||||
ndn::ScopedRegisteredPrefixHandle m_registeredPrefix;
|
||||
ndn::time::milliseconds m_helloReplyFreshness;
|
||||
|
||||
@@ -18,14 +18,12 @@
|
||||
**/
|
||||
|
||||
#include "PSync/producer-base.hpp"
|
||||
#include "PSync/detail/util.hpp"
|
||||
|
||||
#include <ndn-cxx/util/exception.hpp>
|
||||
#include <ndn-cxx/util/logger.hpp>
|
||||
|
||||
#include <boost/algorithm/string.hpp>
|
||||
|
||||
#include <cstring>
|
||||
#include <functional>
|
||||
#include <limits>
|
||||
|
||||
namespace psync {
|
||||
@@ -118,7 +116,7 @@ ProducerBase::updateSeqNo(const ndn::Name& prefix, uint64_t seq)
|
||||
// Insert the new seq no in m_prefixes, m_biMap, and m_iblt
|
||||
it->second = seq;
|
||||
ndn::Name prefixWithSeq = ndn::Name(prefix).appendNumber(seq);
|
||||
uint32_t newHash = murmurHash3(N_HASHCHECK, prefixWithSeq.toUri());
|
||||
auto newHash = detail::murmurHash3(detail::N_HASHCHECK, prefixWithSeq.toUri());
|
||||
m_biMap.insert({newHash, prefixWithSeq});
|
||||
m_iblt.insert(newHash);
|
||||
}
|
||||
|
||||
@@ -20,20 +20,19 @@
|
||||
#ifndef PSYNC_PRODUCER_BASE_HPP
|
||||
#define PSYNC_PRODUCER_BASE_HPP
|
||||
|
||||
#include "PSync/common.hpp"
|
||||
#include "PSync/detail/access-specifiers.hpp"
|
||||
#include "PSync/detail/bloom-filter.hpp"
|
||||
#include "PSync/detail/iblt.hpp"
|
||||
#include "PSync/detail/util.hpp"
|
||||
#include "PSync/segment-publisher.hpp"
|
||||
|
||||
#include <ndn-cxx/face.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/scheduler.hpp>
|
||||
#include <ndn-cxx/util/time.hpp>
|
||||
#include <ndn-cxx/security/key-chain.hpp>
|
||||
#include <ndn-cxx/security/validator-config.hpp>
|
||||
|
||||
#include <boost/bimap.hpp>
|
||||
#include <boost/bimap/bimap.hpp>
|
||||
#include <boost/bimap/unordered_set_of.hpp>
|
||||
|
||||
#include <map>
|
||||
@@ -52,6 +51,7 @@ const ndn::time::milliseconds SYNC_REPLY_FRESHNESS = 1_s;
|
||||
*/
|
||||
class ProducerBase
|
||||
{
|
||||
public:
|
||||
class Error : public std::runtime_error
|
||||
{
|
||||
public:
|
||||
@@ -155,11 +155,8 @@ PSYNC_PUBLIC_WITH_TESTS_ELSE_PROTECTED:
|
||||
void
|
||||
onRegisterFailed(const ndn::Name& prefix, const std::string& msg) const;
|
||||
|
||||
using HashNameBiMap = bm::bimap<bm::unordered_set_of<uint32_t>,
|
||||
bm::unordered_set_of<ndn::Name, std::hash<ndn::Name>>>;
|
||||
|
||||
PSYNC_PUBLIC_WITH_TESTS_ELSE_PROTECTED:
|
||||
IBLT m_iblt;
|
||||
detail::IBLT m_iblt;
|
||||
uint32_t m_expectedNumEntries;
|
||||
// Threshold is used check if the differences are greater
|
||||
// than it and whether we need to update the other side.
|
||||
@@ -167,6 +164,9 @@ PSYNC_PUBLIC_WITH_TESTS_ELSE_PROTECTED:
|
||||
|
||||
// prefix and sequence number
|
||||
std::map<ndn::Name, uint64_t> m_prefixes;
|
||||
|
||||
using HashNameBiMap = bm::bimap<bm::unordered_set_of<uint32_t>,
|
||||
bm::unordered_set_of<ndn::Name, std::hash<ndn::Name>>>;
|
||||
HashNameBiMap m_biMap;
|
||||
|
||||
ndn::Face& m_face;
|
||||
|
||||
@@ -42,7 +42,7 @@ SegmentPublisher::publish(const ndn::Name& interestName, const ndn::Name& dataNa
|
||||
|
||||
void
|
||||
SegmentPublisher::publish(const ndn::Name& interestName, const ndn::Name& dataName,
|
||||
const std::shared_ptr<const ndn::Buffer>& buffer,
|
||||
const ndn::ConstBufferPtr& buffer,
|
||||
ndn::time::milliseconds freshness,
|
||||
const ndn::security::SigningInfo& signingInfo)
|
||||
{
|
||||
|
||||
@@ -31,17 +31,14 @@
|
||||
|
||||
namespace psync {
|
||||
|
||||
const int MAX_SEGMENTS_STORED = 100;
|
||||
|
||||
/**
|
||||
* @brief Segment Publisher to publish segmented data
|
||||
*
|
||||
*/
|
||||
class SegmentPublisher
|
||||
{
|
||||
public:
|
||||
SegmentPublisher(ndn::Face& face, ndn::KeyChain& keyChain,
|
||||
size_t imsLimit = MAX_SEGMENTS_STORED);
|
||||
size_t imsLimit = 100);
|
||||
|
||||
/**
|
||||
* @brief Put all the segments in memory.
|
||||
@@ -68,7 +65,7 @@ public:
|
||||
*/
|
||||
void
|
||||
publish(const ndn::Name& interestName, const ndn::Name& dataName,
|
||||
const std::shared_ptr<const ndn::Buffer>& buffer, ndn::time::milliseconds freshness,
|
||||
const ndn::ConstBufferPtr& buffer, ndn::time::milliseconds freshness,
|
||||
const ndn::security::SigningInfo& signingInfo = ndn::security::SigningInfo());
|
||||
|
||||
/**
|
||||
|
||||
@@ -24,8 +24,7 @@
|
||||
#include <ndn-cxx/name.hpp>
|
||||
|
||||
namespace psync {
|
||||
|
||||
using namespace ndn;
|
||||
namespace detail {
|
||||
|
||||
BOOST_AUTO_TEST_SUITE(TestBloomFilter)
|
||||
|
||||
@@ -40,7 +39,7 @@ BOOST_AUTO_TEST_CASE(Basic)
|
||||
|
||||
BOOST_AUTO_TEST_CASE(NameAppendAndExtract)
|
||||
{
|
||||
Name bfName("/test");
|
||||
ndn::Name bfName("/test");
|
||||
BloomFilter bf(100, 0.001);
|
||||
bf.insert("/memphis");
|
||||
bf.appendToName(bfName);
|
||||
@@ -56,4 +55,5 @@ BOOST_AUTO_TEST_CASE(NameAppendAndExtract)
|
||||
|
||||
BOOST_AUTO_TEST_SUITE_END()
|
||||
|
||||
} // namespace detail
|
||||
} // namespace psync
|
||||
@@ -18,11 +18,11 @@
|
||||
**/
|
||||
|
||||
#include "PSync/full-producer.hpp"
|
||||
#include "PSync/detail/util.hpp"
|
||||
|
||||
#include "tests/boost-test.hpp"
|
||||
#include "tests/io-fixture.hpp"
|
||||
|
||||
#include <ndn-cxx/name.hpp>
|
||||
#include <ndn-cxx/mgmt/nfd/control-parameters.hpp>
|
||||
#include <ndn-cxx/util/dummy-client-face.hpp>
|
||||
|
||||
@@ -80,7 +80,7 @@ BOOST_AUTO_TEST_CASE(OnSyncDataDecodeFailure)
|
||||
BOOST_CHECK_NO_THROW(node.onSyncData(syncInterest, badCompress));
|
||||
|
||||
const uint8_t test[] = {'t', 'e', 's', 't'};
|
||||
auto goodCompressBadBlock = compress(node.m_contentCompression, &test[0], sizeof(test));
|
||||
auto goodCompressBadBlock = detail::compress(node.m_contentCompression, test, sizeof(test));
|
||||
BOOST_CHECK_NO_THROW(node.onSyncData(syncInterest, goodCompressBadBlock));
|
||||
}
|
||||
|
||||
|
||||
@@ -20,11 +20,11 @@
|
||||
#include "PSync/full-producer.hpp"
|
||||
#include "PSync/consumer.hpp"
|
||||
#include "PSync/detail/state.hpp"
|
||||
#include "PSync/detail/util.hpp"
|
||||
|
||||
#include "tests/boost-test.hpp"
|
||||
#include "tests/io-fixture.hpp"
|
||||
|
||||
#include <ndn-cxx/name.hpp>
|
||||
#include <ndn-cxx/util/dummy-client-face.hpp>
|
||||
|
||||
namespace psync {
|
||||
@@ -390,7 +390,7 @@ BOOST_AUTO_TEST_CASE(DelayedSecondSegment)
|
||||
addNode(0);
|
||||
|
||||
int i = 0;
|
||||
State state;
|
||||
detail::State state;
|
||||
std::shared_ptr<Buffer> compressed;
|
||||
do {
|
||||
Name prefixToPublish("userNode0-" + to_string(i++));
|
||||
@@ -400,13 +400,13 @@ BOOST_AUTO_TEST_CASE(DelayedSecondSegment)
|
||||
state.addContent(Name(prefixToPublish).appendNumber(nodes[0]->m_prefixes[prefixToPublish]));
|
||||
|
||||
auto block = state.wireEncode();
|
||||
compressed = compress(nodes[0]->m_contentCompression, block.wire(), block.size());
|
||||
compressed = detail::compress(nodes[0]->m_contentCompression, block.wire(), block.size());
|
||||
} while (compressed->size() < (MAX_NDN_PACKET_SIZE >> 1));
|
||||
|
||||
advanceClocks(ndn::time::milliseconds(10), 100);
|
||||
|
||||
Name syncInterestName(syncPrefix);
|
||||
IBLT iblt(40, nodes[0]->m_ibltCompression);
|
||||
detail::IBLT iblt(40, nodes[0]->m_ibltCompression);
|
||||
iblt.appendToName(syncInterestName);
|
||||
|
||||
nodes[0]->onSyncInterest(syncPrefix, Interest(syncInterestName));
|
||||
|
||||
+8
-8
@@ -22,9 +22,8 @@
|
||||
|
||||
#include "tests/boost-test.hpp"
|
||||
|
||||
#include <ndn-cxx/interest.hpp>
|
||||
|
||||
namespace psync {
|
||||
namespace detail {
|
||||
|
||||
using namespace ndn;
|
||||
|
||||
@@ -32,7 +31,7 @@ BOOST_AUTO_TEST_SUITE(TestIBLT)
|
||||
|
||||
BOOST_AUTO_TEST_CASE(Equal)
|
||||
{
|
||||
int size = 10;
|
||||
const size_t size = 10;
|
||||
|
||||
IBLT iblt1(size, CompressionScheme::DEFAULT);
|
||||
IBLT iblt2(size, CompressionScheme::DEFAULT);
|
||||
@@ -107,7 +106,7 @@ BOOST_AUTO_TEST_CASE(NameAppendAndExtract)
|
||||
|
||||
BOOST_AUTO_TEST_CASE(CopyInsertErase)
|
||||
{
|
||||
int size = 10;
|
||||
const size_t size = 10;
|
||||
|
||||
IBLT iblt1(size, CompressionScheme::DEFAULT);
|
||||
|
||||
@@ -136,7 +135,8 @@ BOOST_AUTO_TEST_CASE(HigherSeqTest)
|
||||
{
|
||||
// The case where we can't recognize if the rcvd IBF has higher sequence number
|
||||
// Relevant to full sync case
|
||||
int size = 10;
|
||||
|
||||
const size_t size = 10;
|
||||
|
||||
IBLT ownIBF(size, CompressionScheme::DEFAULT);
|
||||
IBLT rcvdIBF(size, CompressionScheme::DEFAULT);
|
||||
@@ -160,10 +160,9 @@ BOOST_AUTO_TEST_CASE(HigherSeqTest)
|
||||
|
||||
BOOST_AUTO_TEST_CASE(Difference)
|
||||
{
|
||||
int size = 10;
|
||||
const size_t size = 10;
|
||||
|
||||
IBLT ownIBF(size, CompressionScheme::DEFAULT);
|
||||
|
||||
IBLT rcvdIBF = ownIBF;
|
||||
|
||||
IBLT diff = ownIBF - rcvdIBF;
|
||||
@@ -200,7 +199,7 @@ BOOST_AUTO_TEST_CASE(DifferenceBwOversizedIBFs)
|
||||
// Check that we can still list the difference
|
||||
// even though we can't list the IBFs itself
|
||||
|
||||
int size = 10;
|
||||
const size_t size = 10;
|
||||
|
||||
IBLT ownIBF(size, CompressionScheme::DEFAULT);
|
||||
|
||||
@@ -231,4 +230,5 @@ BOOST_AUTO_TEST_CASE(DifferenceBwOversizedIBFs)
|
||||
|
||||
BOOST_AUTO_TEST_SUITE_END()
|
||||
|
||||
} // namespace detail
|
||||
} // namespace psync
|
||||
@@ -82,7 +82,7 @@ BOOST_AUTO_TEST_CASE(SameSyncInterest)
|
||||
syncInterestName.append("sync");
|
||||
Name syncInterestPrefix = syncInterestName;
|
||||
|
||||
BloomFilter bf(20, 0.001);
|
||||
detail::BloomFilter bf(20, 0.001);
|
||||
bf.appendToName(syncInterestName);
|
||||
|
||||
producer.m_iblt.appendToName(syncInterestName);
|
||||
@@ -133,7 +133,7 @@ BOOST_AUTO_TEST_CASE(OnSyncInterest)
|
||||
// Sync interest with malicious IBF
|
||||
syncInterestName = syncPrefix;
|
||||
syncInterestName.append("sync");
|
||||
BloomFilter bf(20, 0.001);
|
||||
detail::BloomFilter bf(20, 0.001);
|
||||
bf.appendToName(syncInterestName);
|
||||
syncInterestName.append("fake-name");
|
||||
BOOST_CHECK_NO_THROW(producer.onSyncInterest(syncInterestName, Interest(syncInterestName)));
|
||||
|
||||
@@ -18,7 +18,6 @@
|
||||
**/
|
||||
|
||||
#include "PSync/producer-base.hpp"
|
||||
#include "PSync/detail/util.hpp"
|
||||
|
||||
#include "tests/boost-test.hpp"
|
||||
|
||||
@@ -32,7 +31,7 @@ using namespace ndn;
|
||||
|
||||
BOOST_AUTO_TEST_SUITE(TestProducerBase)
|
||||
|
||||
BOOST_AUTO_TEST_CASE(Ctor)
|
||||
BOOST_AUTO_TEST_CASE(Constructor)
|
||||
{
|
||||
util::DummyClientFace face;
|
||||
BOOST_REQUIRE_NO_THROW(ProducerBase(40, face, Name("/psync"), Name("/testUser")));
|
||||
|
||||
@@ -25,9 +25,9 @@
|
||||
|
||||
#include <ndn-cxx/data.hpp>
|
||||
#include <ndn-cxx/interest.hpp>
|
||||
#include <ndn-cxx/security/validator-null.hpp>
|
||||
#include <ndn-cxx/util/dummy-client-face.hpp>
|
||||
#include <ndn-cxx/util/segment-fetcher.hpp>
|
||||
#include <ndn-cxx/security/validator-null.hpp>
|
||||
|
||||
namespace psync {
|
||||
|
||||
@@ -88,7 +88,7 @@ public:
|
||||
shared_ptr<util::SegmentFetcher> fetcher;
|
||||
Name dataName;
|
||||
time::milliseconds freshness = 1_s;
|
||||
State state;
|
||||
detail::State state;
|
||||
|
||||
int numComplete = 0;
|
||||
int numRepliesFromStore = 0;
|
||||
|
||||
+6
-11
@@ -24,8 +24,7 @@
|
||||
#include <ndn-cxx/data.hpp>
|
||||
|
||||
namespace psync {
|
||||
|
||||
using namespace ndn;
|
||||
namespace detail {
|
||||
|
||||
BOOST_AUTO_TEST_SUITE(TestState)
|
||||
|
||||
@@ -36,17 +35,14 @@ BOOST_AUTO_TEST_CASE(EncodeDecode)
|
||||
state.addContent(ndn::Name("test2"));
|
||||
|
||||
// Simulate getting buffer content from segment fetcher
|
||||
Data data;
|
||||
ndn::Data data;
|
||||
data.setContent(state.wireEncode());
|
||||
ndn::Buffer buffer(data.getContent().value_size());
|
||||
std::copy(data.getContent().value_begin(),
|
||||
data.getContent().value_end(),
|
||||
buffer.begin());
|
||||
|
||||
ndn::ConstBufferPtr bufferPtr = make_shared<ndn::Buffer>(buffer);
|
||||
|
||||
ndn::Block block(std::move(bufferPtr));
|
||||
|
||||
ndn::Block block(std::make_shared<ndn::Buffer>(buffer));
|
||||
State rcvdState;
|
||||
rcvdState.wireDecode(block);
|
||||
|
||||
@@ -58,16 +54,14 @@ BOOST_AUTO_TEST_CASE(EmptyContent)
|
||||
State state;
|
||||
|
||||
// Simulate getting buffer content from segment fetcher
|
||||
Data data;
|
||||
ndn::Data data;
|
||||
data.setContent(state.wireEncode());
|
||||
ndn::Buffer buffer(data.getContent().value_size());
|
||||
std::copy(data.getContent().value_begin(),
|
||||
data.getContent().value_end(),
|
||||
buffer.begin());
|
||||
ndn::ConstBufferPtr bufferPtr = make_shared<ndn::Buffer>(buffer);
|
||||
|
||||
ndn::Block block(std::move(bufferPtr));
|
||||
|
||||
ndn::Block block(std::make_shared<ndn::Buffer>(buffer));
|
||||
BOOST_CHECK_NO_THROW(State state2(block));
|
||||
|
||||
State state2(block);
|
||||
@@ -90,4 +84,5 @@ BOOST_AUTO_TEST_CASE(ReEncode)
|
||||
|
||||
BOOST_AUTO_TEST_SUITE_END()
|
||||
|
||||
} // namespace detail
|
||||
} // namespace psync
|
||||
+8
-8
@@ -22,10 +22,11 @@
|
||||
#include "tests/boost-test.hpp"
|
||||
|
||||
namespace psync {
|
||||
namespace detail {
|
||||
|
||||
BOOST_AUTO_TEST_SUITE(TestUtil)
|
||||
|
||||
BOOST_AUTO_TEST_CASE(Basic)
|
||||
BOOST_AUTO_TEST_CASE(Compression)
|
||||
{
|
||||
std::vector<CompressionScheme> available = {CompressionScheme::ZLIB, CompressionScheme::GZIP,
|
||||
CompressionScheme::BZIP2, CompressionScheme::LZMA,
|
||||
@@ -49,6 +50,9 @@ BOOST_AUTO_TEST_CASE(Basic)
|
||||
supported.push_back(CompressionScheme::ZSTD);
|
||||
#endif
|
||||
|
||||
std::set_difference(available.begin(), available.end(), supported.begin(), supported.end(),
|
||||
std::inserter(notSupported, notSupported.begin()));
|
||||
|
||||
const uint8_t uncompressed[] = {'t', 'e', 's', 't'};
|
||||
|
||||
for (const auto& s : supported) {
|
||||
@@ -57,17 +61,13 @@ BOOST_AUTO_TEST_CASE(Basic)
|
||||
BOOST_CHECK_NO_THROW(decompress(s, compressed->data(), compressed->size()));
|
||||
}
|
||||
|
||||
std::set_difference(available.begin(), available.end(), supported.begin(), supported.end(),
|
||||
std::inserter(notSupported, notSupported.begin()));
|
||||
|
||||
for (const auto& s : notSupported) {
|
||||
BOOST_CHECK_THROW(compress(s, uncompressed, sizeof(uncompressed)),
|
||||
std::runtime_error);
|
||||
BOOST_CHECK_THROW(decompress(s, uncompressed, sizeof(uncompressed)),
|
||||
std::runtime_error);
|
||||
BOOST_CHECK_THROW(compress(s, uncompressed, sizeof(uncompressed)), CompressionError);
|
||||
BOOST_CHECK_THROW(decompress(s, uncompressed, sizeof(uncompressed)), CompressionError);
|
||||
}
|
||||
}
|
||||
|
||||
BOOST_AUTO_TEST_SUITE_END()
|
||||
|
||||
} // namespace detail
|
||||
} // namespace psync
|
||||
Reference in new issue
Block a user