23 #include <ndn-cxx/util/logger.hpp> 34 const ndn::Name& syncPrefix,
35 const ndn::Name& userPrefix,
36 ndn::time::milliseconds helloReplyFreshness,
37 ndn::time::milliseconds syncReplyFreshness,
40 userPrefix, syncReplyFreshness, ibltCompression)
41 , m_helloReplyFreshness(helloReplyFreshness)
44 [
this] (
const ndn::Name& syncPrefix) {
46 std::bind(&PartialProducer::onHelloInterest,
this, _1, _2));
48 std::bind(&PartialProducer::onSyncInterest,
this, _1, _2));
60 uint64_t newSeq = seq.value_or(
m_prefixes[prefix] + 1);
62 NDN_LOG_INFO(
"Publish: " << prefix <<
"/" << newSeq);
66 satisfyPendingSyncInterests(prefix);
70 PartialProducer::onHelloInterest(
const ndn::Name& prefix,
const ndn::Interest& interest)
78 if (interest.getName().get(interest.getName().size()-1).toUri() !=
"hello" &&
79 interest.getName().get(interest.getName().size()-4).toUri() !=
"hello") {
83 NDN_LOG_DEBUG(
"Hello Interest Received, nonce: " << interest);
88 state.
addContent(ndn::Name(prefix.first).appendNumber(prefix.second));
90 NDN_LOG_DEBUG(
"sending content p: " << state);
92 ndn::Name helloDataName = prefix;
100 PartialProducer::onSyncInterest(
const ndn::Name& prefix,
const ndn::Interest& interest)
106 NDN_LOG_DEBUG(
"Sync Interest Received, nonce: " << interest.getNonce() <<
107 " hash: " << std::hash<std::string>{}(interest.getName().toUri()));
109 ndn::Name nameWithoutSyncPrefix = interest.getName().getSubName(prefix.size());
110 ndn::Name interestName;
112 if (nameWithoutSyncPrefix.size() == 4) {
114 interestName = interest.getName();
116 else if (nameWithoutSyncPrefix.size() == 6) {
118 interestName = interest.getName().getPrefix(-2);
124 ndn::name::Component bfName, ibltName;
125 unsigned int projectedCount;
126 double falsePositiveProb;
128 projectedCount = interestName.get(interestName.size()-4).toNumber();
129 falsePositiveProb = interestName.get(interestName.size()-3).toNumber()/1000.;
130 bfName = interestName.get(interestName.size()-2);
132 ibltName = interestName.get(interestName.size()-1);
134 catch (
const std::exception& e) {
135 NDN_LOG_ERROR(
"Cannot extract bloom filter and IBF from sync interest: " << e.what());
136 NDN_LOG_ERROR(
"Format: /<syncPrefix>/sync/<BF-count>/<BF-false-positive-probability>/<BF>/<IBF>");
144 bf =
BloomFilter(projectedCount, falsePositiveProb, bfName);
147 catch (
const std::exception& e) {
148 NDN_LOG_WARN(e.what());
156 std::set<uint32_t> positive;
157 std::set<uint32_t> negative;
159 NDN_LOG_TRACE(
"Number elements in IBF: " <<
m_prefixes.size());
163 NDN_LOG_TRACE(
"Result of listEntries on the difference: " << peel);
166 NDN_LOG_DEBUG(
"Can't decode the difference, sending application Nack");
173 NDN_LOG_TRACE(
"Size of positive set " << positive.size());
174 NDN_LOG_TRACE(
"Size of negative set " << negative.size());
175 for (
const auto& hash : positive) {
176 auto nameIt =
m_biMap.left.find(hash);
177 if (nameIt !=
m_biMap.left.end()) {
178 if (bf.
contains(nameIt->second.getPrefix(-1).toUri())) {
181 NDN_LOG_DEBUG(
"Content: " << nameIt->second <<
" " << nameIt->first);
186 NDN_LOG_TRACE(
"m_threshold: " <<
m_threshold <<
" Total: " << positive.size() + negative.size());
191 ndn::Name syncDataName = interestName;
199 auto& entry = m_pendingEntries.emplace(interestName,
PendingEntryInfo{bf, iblt, {}}).first->second;
200 entry.expirationEvent =
m_scheduler.schedule(interest.getInterestLifetime(),
202 NDN_LOG_TRACE(
"Erase Pending Interest " << interest.getNonce());
203 m_pendingEntries.erase(interest.getName());
208 PartialProducer::satisfyPendingSyncInterests(
const ndn::Name& prefix) {
209 NDN_LOG_TRACE(
"size of pending interest: " << m_pendingEntries.size());
211 for (
auto it = m_pendingEntries.begin(); it != m_pendingEntries.end();) {
215 std::set<uint32_t> positive;
216 std::set<uint32_t> negative;
220 NDN_LOG_TRACE(
"Result of listEntries on the difference: " << peel);
222 NDN_LOG_TRACE(
"Number elements in IBF: " <<
m_prefixes.size());
223 NDN_LOG_TRACE(
"m_threshold: " <<
m_threshold <<
" Total: " << positive.size() + negative.size());
226 NDN_LOG_TRACE(
"Decoding of differences with stored IBF unsuccessful, deleting pending interest");
227 m_pendingEntries.erase(it++);
235 NDN_LOG_DEBUG(
"sending sync content " << prefix <<
" " << std::to_string(
m_prefixes[prefix]));
238 NDN_LOG_DEBUG(
"Sending with empty content to send latest IBF to consumer");
242 ndn::Name syncDataName = it->first;
248 m_pendingEntries.erase(it++);
void sendApplicationNack(const ndn::Name &name)
Sends a data packet with content type nack.
void updateSeqNo(const ndn::Name &prefix, uint64_t seq)
Update m_prefixes and IBF with the given prefix and seq.
void initialize(const ndn::name::Component &ibltName)
Populate the hash table using the vector representation of IBLT.
void addContent(const ndn::Name &prefix)
const ndn::Block & wireEncode() const
NDN_LOG_INIT(psync.Consumer)
Partial sync logic to publish data names.
Invertible Bloom Lookup Table (Invertible Bloom Filter)
void publish(const ndn::Name &interestName, const ndn::Name &dataName, const ndn::Block &block, ndn::time::milliseconds freshness, const ndn::security::SigningInfo &signingInfo=ndn::security::v2::KeyChain::getDefaultSigningInfo())
Put all the segments in memory.
CompressionScheme m_ibltCompression
bool contains(const std::string &key) const
void publishName(const ndn::Name &prefix, ndn::optional< uint64_t > seq=ndn::nullopt)
Publish name to let subscribed consumers know.
const std::vector< ndn::Name > & getContent() const
uint32_t m_expectedNumEntries
bool replyFromStore(const ndn::Name &interestName)
Try to reply from memory, return false if we cannot find the segment.
bool listEntries(std::set< uint32_t > &positive, std::set< uint32_t > &negative) const
List all the entries in the IBLT.
void appendToName(ndn::Name &name) const
Appends self to name.
void onRegisterFailed(const ndn::Name &prefix, const std::string &msg) const
Logs a message if setting an interest filter fails.
PartialProducer(size_t expectedNumEntries, ndn::Face &face, const ndn::Name &syncPrefix, const ndn::Name &userPrefix, ndn::time::milliseconds helloReplyFreshness=HELLO_REPLY_FRESHNESS, ndn::time::milliseconds syncReplyFreshness=SYNC_REPLY_FRESHNESS, CompressionScheme ibltCompression=CompressionScheme::NONE)
constructor
ndn::Scheduler m_scheduler
ndn::time::milliseconds m_syncReplyFreshness
SegmentPublisher m_segmentPublisher
Base class for PartialProducer and FullProducer.
std::map< ndn::Name, uint64_t > m_prefixes