diff --git a/.gitignore b/.gitignore index dc40a00..712ddb9 100644 --- a/.gitignore +++ b/.gitignore @@ -32,3 +32,4 @@ __pycache__/ # Other /VERSION.info +/results/ diff --git a/examples/chat-pubsub-regex.cpp b/examples/chat-pubsub-regex.cpp new file mode 100644 index 0000000..6013eda --- /dev/null +++ b/examples/chat-pubsub-regex.cpp @@ -0,0 +1,167 @@ +/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */ +/* + * Copyright (c) 2012-2023 University of California, Los Angeles + * + * This file is part of ndn-svs, synchronization library for distributed realtime + * applications for NDN. + * + * ndn-svs library 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, in version 2.1 of the License. + * + * ndn-svs library 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. + */ + +#include +#include +#include +#include +#include +#include + +#include + +using namespace ndn::svs; + +struct Options +{ + std::string prefix; + std::string m_id; +}; + +class Program +{ +public: + Program(const Options& options) + : m_options(options) + { + // Use HMAC signing for Sync Interests + // Note: this is not generally recommended, but is used here for simplicity + SecurityOptions secOpts(m_keyChain); + secOpts.interestSigner->signingInfo.setSigningHmacKey("dGhpcyBpcyBhIHNlY3JldCBtZXNzYWdl"); + + // Sign data packets using SHA256 (for simplicity) + secOpts.dataSigner->signingInfo.setSha256Signing(); + + // Do not fetch publications older than 10 seconds + SVSPubSubOptions opts; + opts.useTimestamp = true; + opts.maxPubAge = ndn::time::seconds(10); + + // Create the Pub/Sub instance + m_svsps = std::make_shared( + ndn::Name(m_options.prefix), + ndn::Name(m_options.m_id), + face, + std::bind(&Program::onMissingData, this, _1), + opts, + secOpts); + + std::cout << "SVS client starting: " << m_options.m_id << std::endl; + + // Subscribe to all data packets with prefix /chat (the "topic") + m_svsps->subscribeWithRegex(ndn::Regex("^"), [] (const auto& subData) + { + std::string content(reinterpret_cast(subData.data.data()), subData.data.size()); + std::cout << subData.producerPrefix << " [" << subData.seqNo << "] : " << + subData.name << " : "; + if (content.length() > 200) { + std::cout << "[LONG] " << content.length() << " bytes" + << " [" << std::hash{}(content) << "]"; + } else { + std::cout << content; + } + std::cout << std::endl; + }); + } + + void + run() + { + // Begin processing face events in a separate thread. + std::thread svsThread([this] { face.processEvents(); }); + + // Announce our presence. + // Note that the SVS-PS instance is thread-safe. + publishMsg("User " + m_options.m_id + " has joined the groupchat"); + + // Read from stdin and publish messages. + std::string userInput; + while (true) { + std::getline(std::cin, userInput); + publishMsg(userInput); + } + + // Wait for the SVS-PS thread to finish. + svsThread.join(); + } + +protected: + /** + * Callback on receving a new State Vector from another node. + * This will be called regardless of whether the missing data contains any topics + * or producers that we are subscribed to. + */ + void + onMissingData(const std::vector&) + { + // Ignore any other missing data for this example + } + + /** + * Publish a string message to the group + */ + void + publishMsg(const std::string& msg) + { + // Message to send + std::string content = msg; + + // If the message starts with "SEND " generate a new message + // with random content with length after send + if (msg.length() > 5 && msg.substr(0, 5) == "SEND ") { + auto len = std::stoi(msg.substr(5)); + + content = std::string(len, 'a'); + std::srand(std::time(nullptr)); + for (auto& c : content) + c = std::rand() % 26 + 'a'; + + std::cout << "> Sending random message with hash [" << std::hash{}(content) << "]" << std::endl; + } + + // Note that unlike SVSync, names can be arbitrary, + // and need not be prefixed with the producer prefix. + ndn::Name name("chat"); // topic of publication + name.append(m_options.m_id); // who sent this + name.appendTimestamp(); // and when + + m_svsps->publish(name, ndn::make_span(reinterpret_cast(content.data()), content.size())); + } + +private: + const Options m_options; + ndn::Face face; + std::shared_ptr m_svsps; + ndn::KeyChain m_keyChain; +}; + +int +main(int argc, char** argv) +{ + if (argc != 2) { + std::cerr << "Usage: " << argv[0] << " " << std::endl; + return 1; + } + + Options opt; + opt.prefix = "/ndn/svs"; + opt.m_id = argv[1]; + + Program program(opt); + program.run(); + + return 0; +} diff --git a/examples/chat.cpp b/examples/chat.cpp index 36d4eca..d0aefe6 100644 --- a/examples/chat.cpp +++ b/examples/chat.cpp @@ -84,7 +84,7 @@ class Program for (ndn::svs::SeqNo s = v[i].low; s <= v[i].high; ++s) { // Request a single data packet using the SVSync API ndn::svs::NodeID nid = v[i].nodeId; - m_svs->fetchData(nid, s, [nid](const auto& data) { + m_svs->fetchData(nid, v[i].bootstrapTime, s, [nid](const auto& data) { std::string content(reinterpret_cast(data.getContent().value()), data.getContent().value_size()); std::cout << data.getName() << " : " << content << std::endl; diff --git a/ndn-svs/common.hpp b/ndn-svs/common.hpp index 477207d..ea055da 100644 --- a/ndn-svs/common.hpp +++ b/ndn-svs/common.hpp @@ -40,6 +40,7 @@ namespace ndn::svs { // Type and constant declarations for State Vector Sync (SVS) using NodeID = ndn::Name; using SeqNo = uint64_t; +using BootstrapTime = uint64_t; using ndn::security::ValidationError; diff --git a/ndn-svs/core.cpp b/ndn-svs/core.cpp index a22fa4d..2bd627d 100644 --- a/ndn-svs/core.cpp +++ b/ndn-svs/core.cpp @@ -21,8 +21,12 @@ #include #include #include +#include + +#include #include +#include #ifdef NDN_SVS_COMPRESSION #include @@ -33,6 +37,205 @@ namespace ndn::svs { +NDN_LOG_INIT(ndn_svs.SyncTimeline); + +namespace { + +using SteadyClock = std::chrono::steady_clock; + +static uint64_t +elapsedMs(const SteadyClock::time_point& start, const SteadyClock::time_point& end) +{ + return std::chrono::duration_cast(end - start).count(); +} + +static uint64_t +elapsedUs(const SteadyClock::time_point& start, const SteadyClock::time_point& end) +{ + return std::chrono::duration_cast(end - start).count(); +} + +static BootstrapTime +getCurrentBootstrapTime() +{ + return static_cast( + std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()).count()); +} + +static std::string +syncTraceKey(const Interest& interest) +{ + return interest.getName().toUri(); +} + +static bool +decodeSyncParameters(const Interest& interest, ndn::Block& params, VersionVector& vv) +{ + if (!interest.hasApplicationParameters()) { + return false; + } + + params = interest.getApplicationParameters(); + params.parse(); + +#ifdef NDN_SVS_COMPRESSION + // The spec requires that if an LZMA block is present, then no other blocks + // are present; the whole ApplicationParameters payload is compressed. + if (params.find(tlv::LzmaBlock) != params.elements_end()) { + auto lzmaBlock = params.get(tlv::LzmaBlock); + + boost::iostreams::filtering_istreambuf in; + in.push(boost::iostreams::lzma_decompressor()); + in.push(boost::iostreams::array_source(reinterpret_cast(lzmaBlock.value()), + lzmaBlock.value_size())); + ndn::OBufferStream decompressed; + boost::iostreams::copy(in, decompressed); + + auto parsed = ndn::Block::fromBuffer(decompressed.buf()); + if (!std::get<0>(parsed)) { + return false; + } + + params = std::get<1>(parsed); + params.parse(); + } +#endif + + vv = VersionVector(params.get(tlv::StateVector)); + return true; +} + +} // namespace + +struct SVSyncCore::SyncProcessingJob +{ + Interest interest; + VersionVector localVector; + uint64_t stateGeneration = 0; + uint64_t incomingFace = 0; + SteadyClock::time_point receivedAt; + SteadyClock::time_point validatedAt; + std::string traceKey; +}; + +struct SVSyncCore::SyncProcessingResult +{ + SyncProcessingJob job; + bool ok = false; + bool decodeFailed = false; + VersionVector remoteVector; + MergeComputationResult merge; + bool myVectorNew = false; + std::vector missingData; + uint64_t parseUs = 0; + uint64_t decodeUs = 0; + uint64_t compareUs = 0; + uint64_t missingUs = 0; + uint64_t workerUs = 0; +}; + +struct SVSyncCore::SyncProductionJob +{ + VersionVector localVector; + ndn::Block extraBlock; + bool hasExtraBlock = false; + uint64_t stateGeneration = 0; + SteadyClock::time_point submittedAt; + std::string traceKey; +}; + +struct SVSyncCore::SyncProductionResult +{ + SyncProductionJob job; + bool ok = false; + Interest interest; + bool signedInWorker = false; + bool extraBlockBuiltInWorker = false; + uint64_t encodeUs = 0; + uint64_t signUs = 0; + uint64_t workerUs = 0; +}; + +class SVSyncCore::SyncWorkerPool +{ +public: + explicit + SyncWorkerPool(size_t workerThreads, size_t maxQueueSize) + : m_maxQueueSize(std::max(1, maxQueueSize)) + { + workerThreads = std::max(1, workerThreads); + m_workers.reserve(workerThreads); + for (size_t i = 0; i < workerThreads; ++i) { + m_workers.emplace_back([this] { run(); }); + } + } + + ~SyncWorkerPool() + { + shutdown(); + } + + bool + post(std::function job, size_t& queueDepth) + { + std::lock_guard lock(m_mutex); + if (m_stopped || m_queue.size() >= m_maxQueueSize) { + queueDepth = m_queue.size(); + return false; + } + m_queue.push(std::move(job)); + queueDepth = m_queue.size(); + m_cv.notify_one(); + return true; + } + + void + shutdown() + { + { + std::lock_guard lock(m_mutex); + if (m_stopped) { + return; + } + m_stopped = true; + } + m_cv.notify_all(); + for (auto& worker : m_workers) { + if (worker.joinable()) { + worker.join(); + } + } + } + +private: + void + run() + { + while (true) { + std::function job; + { + std::unique_lock lock(m_mutex); + m_cv.wait(lock, [this] { return m_stopped || !m_queue.empty(); }); + if (m_stopped && m_queue.empty()) { + return; + } + job = std::move(m_queue.front()); + m_queue.pop(); + } + job(); + } + } + +private: + size_t m_maxQueueSize; + std::vector m_workers; + std::queue> m_queue; + std::mutex m_mutex; + std::condition_variable m_cv; + bool m_stopped = false; +}; + SVSyncCore::SVSyncCore(ndn::Face& face, const Name& syncPrefix, const UpdateCallback& onUpdate, @@ -42,6 +245,7 @@ SVSyncCore::SVSyncCore(ndn::Face& face, , m_syncPrefix(syncPrefix) , m_securityOptions(securityOptions) , m_id(nid) + , m_bootstrapTime(getCurrentBootstrapTime()) , m_onUpdate(onUpdate) , m_maxSuppressionTime(500_ms) , m_periodicSyncTime(30_s) @@ -61,6 +265,143 @@ SVSyncCore::SVSyncCore(ndn::Face& face, [](auto&&...) { NDN_THROW(Error("Failed to register sync prefix")); }); } +SVSyncCore::~SVSyncCore() +{ + setParallelSyncProduction(false); + setParallelSyncProcessing(false); +} + +void +SVSyncCore::setParallelSyncProcessing(bool enabled, size_t workerThreads, + size_t maxQueueSize) +{ + if (!enabled) { + m_parallelSyncProcessing = false; + if (m_syncProcessingAlive) { + m_syncProcessingAlive->store(false, std::memory_order_relaxed); + } + if (m_syncWorkerPool) { + m_syncWorkerPool->shutdown(); + m_syncWorkerPool.reset(); + } + return; + } + + m_syncProcessingAlive = std::make_shared>(true); + if (!m_syncWorkerPool) { + m_syncWorkerPool = std::make_unique(workerThreads, maxQueueSize); + } + m_parallelSyncProcessing = true; +} + +void +SVSyncCore::setParallelSyncProduction(bool enabled, size_t workerThreads, + size_t maxQueueSize, + bool signInWorker, + bool buildExtraBlockInWorker) +{ + if (!enabled) { + m_parallelSyncProduction = false; + m_parallelSyncProductionSigning = false; + m_parallelSyncProductionExtraBlock = false; + if (m_syncProductionAlive) { + m_syncProductionAlive->store(false, std::memory_order_relaxed); + } + if (m_syncProductionWorkerPool) { + m_syncProductionWorkerPool->shutdown(); + m_syncProductionWorkerPool.reset(); + } + return; + } + + m_syncProductionAlive = std::make_shared>(true); + if (!m_syncProductionWorkerPool) { + m_syncProductionWorkerPool = std::make_unique(workerThreads, maxQueueSize); + } + m_parallelSyncProductionSigning = signInWorker; + m_parallelSyncProductionExtraBlock = buildExtraBlockInWorker; + m_parallelSyncProduction = true; +} + +void +SVSyncCore::setSyncInterestBatching(bool enabled, time::milliseconds window) +{ + std::lock_guard lock(m_schedulerMutex); + m_syncInterestBatching.store(enabled, std::memory_order_relaxed); + m_syncInterestBatchWindow = window; + if (!enabled) { + m_publicationSyncPending = false; + m_publicationSyncEvent.cancel(); + } +} + +void +SVSyncCore::setMaxSuppressionTime(time::milliseconds delay) +{ + if (delay < 0_ms) { + delay = 0_ms; + } + + std::lock_guard lock(m_schedulerMutex); + m_maxSuppressionTime = delay; + m_intrReplyDist = std::uniform_int_distribution<>(0, m_maxSuppressionTime.count()); +} + +void +SVSyncCore::setPeriodicSyncTime(time::milliseconds interval, double jitter) +{ + if (interval < 1_ms) { + interval = 1_ms; + } + if (jitter < 0.0) { + jitter = 0.0; + } + if (jitter > 1.0) { + jitter = 1.0; + } + + std::lock_guard lock(m_schedulerMutex); + m_periodicSyncTime = interval; + m_periodicSyncJitter = jitter; + const auto lower = static_cast( + std::max(1, static_cast( + std::llround(m_periodicSyncTime.count() * (1.0 - m_periodicSyncJitter))))); + const auto upper = static_cast( + std::max(lower, static_cast( + std::llround(m_periodicSyncTime.count() * (1.0 + m_periodicSyncJitter))))); + m_retxDist = std::uniform_int_distribution<>(lower, upper); +} + +SVSyncCore::SyncProcessingStats +SVSyncCore::getSyncProcessingStats() const +{ + SyncProcessingStats stats; + stats.syncJobsSubmitted = m_syncJobsSubmitted.load(); + stats.syncJobsCompleted = m_syncJobsCompleted.load(); + stats.syncJobsDropped = m_syncJobsDropped.load(); + stats.syncJobsStale = m_syncJobsStale.load(); + stats.syncWorkerQueueDepth = m_syncWorkerQueueDepth.load(); + stats.syncWorkerProcessingMs = m_syncWorkerProcessingMs.load(); + stats.syncMainThreadPublishMs = m_syncMainThreadPublishMs.load(); + stats.syncInterestSerialHandlerMs = m_syncInterestSerialHandlerMs.load(); + stats.syncInterestParallelTotalMs = m_syncInterestParallelTotalMs.load(); + stats.syncInterestMainThreadBlockingMs = m_syncInterestMainThreadBlockingMs.load(); + stats.syncProductionJobsSubmitted = m_syncProductionJobsSubmitted.load(); + stats.syncProductionJobsCompleted = m_syncProductionJobsCompleted.load(); + stats.syncProductionJobsDropped = m_syncProductionJobsDropped.load(); + stats.syncProductionJobsStale = m_syncProductionJobsStale.load(); + stats.syncProductionWorkerQueueDepth = m_syncProductionWorkerQueueDepth.load(); + stats.syncProductionWorkerProcessingMs = m_syncProductionWorkerProcessingMs.load(); + stats.syncProductionParallelTotalMs = m_syncProductionParallelTotalMs.load(); + return stats; +} + +void +SVSyncCore::incrementStat(std::atomic& counter, uint64_t value) const +{ + counter.fetch_add(value, std::memory_order_relaxed); +} + static inline int suppressionCurve(int constFactor, int value) { @@ -91,8 +432,15 @@ SVSyncCore::sendInitialInterest() void SVSyncCore::onSyncInterest(const Interest& interest) { + auto receivedAt = SteadyClock::now(); + auto traceKey = syncTraceKey(interest); + NDN_LOG_TRACE("event=sync_interest_received key=" << traceKey); + NDN_LOG_TRACE("event=validation_start key=" << traceKey); + switch (m_securityOptions.interestSigner->signingInfo.getSignerType()) { case security::SigningInfo::SIGNER_TYPE_NULL: + NDN_LOG_TRACE("event=validation_done key=" << traceKey << + " validation=none elapsed_us=" << elapsedUs(receivedAt, SteadyClock::now())); onSyncInterestValidated(interest); return; @@ -100,16 +448,33 @@ SVSyncCore::onSyncInterest(const Interest& interest) if (security::verifySignature(interest, m_keyChainMem.getTpm(), m_securityOptions.interestSigner->signingInfo.getSignerName(), - DigestAlgorithm::SHA256)) + DigestAlgorithm::SHA256)) { + NDN_LOG_TRACE("event=validation_done key=" << traceKey << + " validation=hmac elapsed_us=" << elapsedUs(receivedAt, SteadyClock::now())); onSyncInterestValidated(interest); + } return; default: - if (m_securityOptions.validator) - m_securityOptions.validator->validate( - interest, std::bind(&SVSyncCore::onSyncInterestValidated, this, _1), nullptr); - else + if (m_securityOptions.validator) { + m_securityOptions.validator->validate(interest, + [this, traceKey, receivedAt] (const Interest& validInterest) { + NDN_LOG_TRACE("event=validation_done key=" << traceKey << + " validation=validator elapsed_us=" << + elapsedUs(receivedAt, SteadyClock::now())); + onSyncInterestValidated(validInterest); + }, + [traceKey, receivedAt] (const Interest&, const auto&) { + NDN_LOG_DEBUG("event=validation_done key=" << traceKey << + " validation=failed elapsed_us=" << + elapsedUs(receivedAt, SteadyClock::now())); + }); + } + else { + NDN_LOG_TRACE("event=validation_done key=" << traceKey << + " validation=none elapsed_us=" << elapsedUs(receivedAt, SteadyClock::now())); onSyncInterestValidated(interest); + } return; } } @@ -117,69 +482,168 @@ SVSyncCore::onSyncInterest(const Interest& interest) void SVSyncCore::onSyncInterestValidated(const Interest& interest) { - // Get incoming face (this is needed by NLSR) - uint64_t incomingFace = 0; + if (!m_parallelSyncProcessing || !m_syncWorkerPool) { + onSyncInterestValidatedSerial(interest); + return; + } + + const auto mainStart = SteadyClock::now(); + const auto traceKey = syncTraceKey(interest); + SyncProcessingJob job; + job.interest = interest; + job.receivedAt = mainStart; + job.validatedAt = mainStart; + job.traceKey = traceKey; + + NDN_LOG_TRACE("event=sync_interest_parse_start mode=parallel-main key=" << traceKey); { auto tag = interest.getTag(); if (tag) { - incomingFace = tag->get(); + job.incomingFace = tag->get(); } } + { + std::lock_guard lock(m_vvMutex); + job.localVector = m_vv; + job.stateGeneration = m_stateGeneration; + } + NDN_LOG_TRACE("event=sync_interest_parse_done mode=parallel-main key=" << traceKey << + " elapsed_us=" << elapsedUs(mainStart, SteadyClock::now())); - // Check for invalid Interest - if (!interest.hasApplicationParameters()) { + size_t queueDepth = 0; + auto* pool = m_syncWorkerPool.get(); + auto alive = m_syncProcessingAlive; + bool queued = pool->post([this, alive, job = std::move(job)] { + SyncProcessingResult result; + result.job = job; + const auto workerStart = SteadyClock::now(); + + try { + const auto parseStart = SteadyClock::now(); + ndn::Block params; + VersionVector remoteVector; + if (!decodeSyncParameters(result.job.interest, params, remoteVector)) { + NDN_THROW(ndn::tlv::Error("Failed to decode sync parameters")); + } + const auto parseDone = SteadyClock::now(); + result.parseUs = elapsedUs(parseStart, parseDone); + NDN_LOG_TRACE("event=sync_interest_parse_done mode=worker key=" << result.job.traceKey << + " elapsed_us=" << result.parseUs); + + const auto decodeStart = SteadyClock::now(); + NDN_LOG_TRACE("event=state_vector_decode_start mode=worker key=" << result.job.traceKey); + result.remoteVector = remoteVector; + const auto decodeDone = SteadyClock::now(); + result.decodeUs = elapsedUs(decodeStart, decodeDone); + NDN_LOG_TRACE("event=state_vector_decode_done mode=worker key=" << result.job.traceKey << + " elapsed_us=" << result.decodeUs); + + const auto compareStart = SteadyClock::now(); + NDN_LOG_TRACE("event=state_compare_start mode=worker key=" << result.job.traceKey); + result.merge = computeMergeStateVector(result.job.localVector, result.remoteVector); + result.myVectorNew = result.merge.myVectorNew; + const auto compareDone = SteadyClock::now(); + result.compareUs = elapsedUs(compareStart, compareDone); + NDN_LOG_TRACE("event=state_compare_done mode=worker key=" << result.job.traceKey << + " elapsed_us=" << result.compareUs); + + const auto missingStart = SteadyClock::now(); + NDN_LOG_TRACE("event=missing_data_compute_start mode=worker key=" << result.job.traceKey); + result.missingData = result.merge.missingData; + const auto missingDone = SteadyClock::now(); + result.missingUs = elapsedUs(missingStart, missingDone); + NDN_LOG_TRACE("event=missing_data_compute_done mode=worker key=" << result.job.traceKey << + " elapsed_us=" << result.missingUs); + result.ok = true; + } + catch (const ndn::tlv::Error&) { + result.decodeFailed = true; + } + + const auto workerDone = SteadyClock::now(); + result.workerUs = elapsedUs(workerStart, workerDone); + incrementStat(m_syncWorkerProcessingMs, result.workerUs / 1000); + NDN_LOG_TRACE("event=sync_worker_processing_ms key=" << result.job.traceKey << + " elapsed_ms=" << (result.workerUs / 1000)); + + m_face.getIoContext().post([this, alive, result = std::move(result)] () mutable { + if (!alive || !alive->load(std::memory_order_relaxed)) { + return; + } + processSyncInterestResult(std::move(result)); + }); + }, queueDepth); + + m_syncWorkerQueueDepth.store(queueDepth, std::memory_order_relaxed); + + if (!queued) { + incrementStat(m_syncJobsDropped); + NDN_LOG_DEBUG("event=sync_job_queue_full key=" << traceKey << + " queue_depth=" << queueDepth << " action=fallback_serial"); + onSyncInterestValidatedSerial(interest); return; } - // Decode state parameters - ndn::Block params = interest.getApplicationParameters(); - params.parse(); - -#ifdef NDN_SVS_COMPRESSION - // Decompress if necessary. The spec requires that if an LZMA block is - // present, then no other blocks are present (everything is compressed - // together) - if (params.find(tlv::LzmaBlock) != params.elements_end()) { - auto lzmaBlock = params.get(tlv::LzmaBlock); + incrementStat(m_syncJobsSubmitted); + incrementStat(m_syncInterestMainThreadBlockingMs, elapsedMs(mainStart, SteadyClock::now())); +} - boost::iostreams::filtering_istreambuf in; - in.push(boost::iostreams::lzma_decompressor()); - in.push(boost::iostreams::array_source(reinterpret_cast(lzmaBlock.value()), - lzmaBlock.value_size())); - ndn::OBufferStream decompressed; - boost::iostreams::copy(in, decompressed); +void +SVSyncCore::onSyncInterestValidatedSerial(const Interest& interest, bool countSerialStats) +{ + const auto handlerStart = SteadyClock::now(); + const auto traceKey = syncTraceKey(interest); - auto parsed = ndn::Block::fromBuffer(decompressed.buf()); - if (!std::get<0>(parsed)) { - // TODO: log error parsing inner block - return; + // Get incoming face (this is needed by NLSR) + uint64_t incomingFace = 0; + const auto parseStart = SteadyClock::now(); + NDN_LOG_TRACE("event=sync_interest_parse_start mode=serial key=" << traceKey); + { + auto tag = interest.getTag(); + if (tag) { + incomingFace = tag->get(); } - - params = std::get<1>(parsed); - params.parse(); } -#endif + NDN_LOG_TRACE("event=sync_interest_parse_done mode=serial key=" << traceKey << + " elapsed_us=" << elapsedUs(parseStart, SteadyClock::now())); - // Get state vector - std::shared_ptr vvOther; + ndn::Block params; + VersionVector vvOther; + const auto decodeStart = SteadyClock::now(); + NDN_LOG_TRACE("event=state_vector_decode_start mode=serial key=" << traceKey); try { - vvOther = std::make_shared(params.get(tlv::StateVector)); - } catch (ndn::tlv::Error&) { - // TODO: log error + if (!decodeSyncParameters(interest, params, vvOther)) { + NDN_LOG_DEBUG("event=state_vector_decode_failed mode=serial key=" << traceKey); + return; + } + } + catch (const ndn::tlv::Error&) { + NDN_LOG_DEBUG("event=state_vector_decode_failed mode=serial key=" << traceKey); return; } + NDN_LOG_TRACE("event=state_vector_decode_done mode=serial key=" << traceKey << + " elapsed_us=" << elapsedUs(decodeStart, SteadyClock::now())); // Read extra mapping blocks if (m_recvExtraBlock) { try { - m_recvExtraBlock(params.get(tlv::MappingData), *vvOther); + m_recvExtraBlock(params.get(tlv::MappingData), vvOther); } catch (std::exception&) { // TODO: log error but continue } } // Merge state vector - auto result = mergeStateVector(*vvOther); + const auto compareStart = SteadyClock::now(); + NDN_LOG_TRACE("event=state_compare_start mode=serial key=" << traceKey); + auto result = mergeStateVector(vvOther); + NDN_LOG_TRACE("event=state_compare_done mode=serial key=" << traceKey << + " elapsed_us=" << elapsedUs(compareStart, SteadyClock::now())); + + const auto missingStart = SteadyClock::now(); + NDN_LOG_TRACE("event=missing_data_compute_start mode=serial key=" << traceKey); + NDN_LOG_TRACE("event=missing_data_compute_done mode=serial key=" << traceKey << + " elapsed_us=" << elapsedUs(missingStart, SteadyClock::now())); // Callback if missing data found if (!result.missingInfo.empty()) { @@ -189,7 +653,7 @@ SVSyncCore::onSyncInterestValidated(const Interest& interest) } // Try to record; the call will check if in suppression state - if (recordVector(*vvOther)) + if (recordVector(vvOther)) return; // If incoming state identical/newer to local vector, reset timer @@ -197,7 +661,7 @@ SVSyncCore::onSyncInterestValidated(const Interest& interest) if (!result.myVectorNew) { retxSyncInterest(false, 0); } else { - enterSuppressionState(*vvOther); + enterSuppressionState(vvOther); // Check how much time is left on the timer, // reset to ~m_intrReplyDist if more than that. int delay = m_intrReplyDist(m_rng); @@ -210,6 +674,106 @@ SVSyncCore::onSyncInterestValidated(const Interest& interest) retxSyncInterest(false, delay); } } + + const auto totalMs = elapsedMs(handlerStart, SteadyClock::now()); + NDN_LOG_TRACE("event=total_sync_interest_handler_ms mode=serial key=" << traceKey << + " elapsed_ms=" << totalMs); + NDN_LOG_TRACE("event=main_loop_blocked_ms mode=serial key=" << traceKey << + " elapsed_ms=" << totalMs); + if (countSerialStats) { + incrementStat(m_syncInterestSerialHandlerMs, totalMs); + incrementStat(m_syncInterestMainThreadBlockingMs, totalMs); + } +} + +void +SVSyncCore::processSyncInterestResult(SyncProcessingResult result) +{ + const auto mainStart = SteadyClock::now(); + const auto& traceKey = result.job.traceKey; + + if (!result.ok || result.decodeFailed) { + NDN_LOG_DEBUG("event=sync_parallel_result_decode_failed key=" << traceKey); + return; + } + + bool stale = false; + uint64_t currentGeneration = 0; + { + std::lock_guard lock(m_vvMutex); + if (m_stateGeneration != result.job.stateGeneration) { + stale = true; + currentGeneration = m_stateGeneration; + } + else { + m_vv = result.merge.mergedVector; + if (result.merge.otherVectorNew) { + ++m_stateGeneration; + } + } + } + + if (stale) { + incrementStat(m_syncJobsStale); + NDN_LOG_DEBUG("event=sync_job_stale key=" << traceKey << + " captured_generation=" << result.job.stateGeneration << + " current_generation=" << currentGeneration << + " action=fallback_serial"); + // Keep protocol behavior conservative: recompute against current state on + // the Face/io_context thread instead of applying an old worker snapshot. + onSyncInterestValidatedSerial(result.job.interest, false); + return; + } + + if (m_recvExtraBlock && result.job.interest.hasApplicationParameters()) + { + try { + ndn::Block params; + VersionVector remoteVector; + if (decodeSyncParameters(result.job.interest, params, remoteVector)) { + m_recvExtraBlock(params.get(tlv::MappingData), result.remoteVector); + } + } + catch (std::exception&) {} + } + + if (!result.missingData.empty()) + { + for (auto& e : result.missingData) + e.incomingFace = result.job.incomingFace; + m_onUpdate(result.missingData); + } + + if (recordVector(result.remoteVector)) { + incrementStat(m_syncJobsCompleted); + return; + } + + if (!result.myVectorNew) + { + retxSyncInterest(false, 0); + } + else + { + enterSuppressionState(result.remoteVector); + int delay = m_intrReplyDist(m_rng); + delay = suppressionCurve(m_maxSuppressionTime.count(), delay); + + if (getCurrentTime() + delay * 1000 < m_nextSyncInterest) + { + retxSyncInterest(false, delay); + } + } + + const auto mainMs = elapsedMs(mainStart, SteadyClock::now()); + const auto totalMs = elapsedMs(result.job.receivedAt, SteadyClock::now()); + incrementStat(m_syncJobsCompleted); + incrementStat(m_syncInterestMainThreadBlockingMs, mainMs); + incrementStat(m_syncInterestParallelTotalMs, totalMs); + NDN_LOG_TRACE("event=sync_interest_parallel_total_ms key=" << traceKey << + " elapsed_ms=" << totalMs); + NDN_LOG_TRACE("event=main_loop_blocked_ms mode=parallel key=" << traceKey << + " elapsed_ms=" << mainMs); } void @@ -238,12 +802,172 @@ SVSyncCore::retxSyncInterest(bool send, unsigned int delay) } } +void +SVSyncCore::schedulePublicationSync() +{ + std::lock_guard lock(m_schedulerMutex); + if (m_publicationSyncPending) { + return; + } + + m_publicationSyncPending = true; + m_publicationSyncEvent = m_scheduler.schedule(m_syncInterestBatchWindow, [this] { + { + std::lock_guard lock(m_schedulerMutex); + m_publicationSyncPending = false; + } + sendLocalPublicationSyncInterest(); + }); +} + +void +SVSyncCore::sendLocalPublicationSyncInterest() +{ + { + std::lock_guard lock(m_recordedVvMutex); + m_recordedVv = nullptr; + } + + sendSyncInterest(); + + unsigned int delay = m_retxDist(m_rng); + std::lock_guard lock(m_schedulerMutex); + m_nextSyncInterest = getCurrentTime() + 1000 * delay; + m_retxEvent = m_scheduler.schedule(time::milliseconds(delay), + [this] { retxSyncInterest(true, 0); }); +} + void SVSyncCore::sendSyncInterest() { if (!m_initialized) return; + if (!m_parallelSyncProduction || !m_syncProductionWorkerPool) { + sendSyncInterestSerial(); + return; + } + + const auto mainStart = SteadyClock::now(); + SyncProductionJob job; + job.submittedAt = mainStart; + job.traceKey = m_syncPrefix.toUri(); + + NDN_LOG_TRACE("event=response_encode_start mode=parallel-main key=" << job.traceKey); + { + std::lock_guard lock(m_vvMutex); + job.localVector = m_vv; + job.stateGeneration = m_stateGeneration; + } + + if (m_getExtraBlock && !m_parallelSyncProductionExtraBlock) { + job.extraBlock = m_getExtraBlock(job.localVector); + job.hasExtraBlock = true; + } + NDN_LOG_TRACE("event=response_encode_snapshot_done mode=parallel-main key=" << job.traceKey << + " elapsed_us=" << elapsedUs(mainStart, SteadyClock::now())); + + size_t queueDepth = 0; + auto* pool = m_syncProductionWorkerPool.get(); + auto alive = m_syncProductionAlive; + bool queued = pool->post([this, alive, job = std::move(job)] { + SyncProductionResult result; + result.job = job; + const auto workerStart = SteadyClock::now(); + + try { + const auto encodeStart = SteadyClock::now(); + ndn::Block extraBlock; + bool hasExtraBlock = result.job.hasExtraBlock; + if (m_getExtraBlock && m_parallelSyncProductionExtraBlock) { + extraBlock = m_getExtraBlock(result.job.localVector); + hasExtraBlock = true; + result.extraBlockBuiltInWorker = true; + } + else if (result.job.hasExtraBlock) { + extraBlock = result.job.extraBlock; + } + + ndn::encoding::EncodingBuffer enc; + size_t length = 0; + if (hasExtraBlock) { + length += ndn::encoding::prependBlock(enc, extraBlock); + } + length += ndn::encoding::prependBlock(enc, result.job.localVector.encode()); + enc.prependVarNumber(length); + enc.prependVarNumber(ndn::tlv::ApplicationParameters); + + ndn::Block wire = enc.block(); + wire.encode(); +#ifdef NDN_SVS_COMPRESSION + boost::iostreams::filtering_istreambuf in; + in.push(boost::iostreams::lzma_compressor()); + in.push(boost::iostreams::array_source(reinterpret_cast(wire.data()), wire.size())); + ndn::OBufferStream compressed; + boost::iostreams::copy(in, compressed); + wire = ndn::Block(tlv::LzmaBlock, compressed.buf()); + wire.encode(); +#endif + + result.interest.setName(Name(m_syncPrefix).appendVersion(2)); + result.interest.setApplicationParameters(wire); + result.interest.setInterestLifetime(1_ms); + result.encodeUs = elapsedUs(encodeStart, SteadyClock::now()); + + if (m_parallelSyncProductionSigning) { + const auto signStart = SteadyClock::now(); + NDN_LOG_TRACE("event=response_sign_start mode=parallel-worker key=" + << result.interest.getName()); + { + std::lock_guard signingLock(m_syncProductionSigningMutex); + signSyncInterest(result.interest); + } + result.signUs = elapsedUs(signStart, SteadyClock::now()); + result.signedInWorker = true; + NDN_LOG_TRACE("event=response_sign_done mode=parallel-worker key=" + << result.interest.getName() << " elapsed_us=" << result.signUs); + } + + result.ok = true; + } + catch (const std::exception&) { + result.ok = false; + } + + result.workerUs = elapsedUs(workerStart, SteadyClock::now()); + incrementStat(m_syncProductionWorkerProcessingMs, result.workerUs / 1000); + + m_face.getIoContext().post([this, alive, result = std::move(result)] () mutable { + if (!alive || !alive->load(std::memory_order_relaxed)) { + return; + } + processSyncProductionResult(std::move(result)); + }); + }, queueDepth); + + m_syncProductionWorkerQueueDepth.store(queueDepth, std::memory_order_relaxed); + + if (!queued) { + incrementStat(m_syncProductionJobsDropped); + NDN_LOG_DEBUG("event=sync_production_queue_full key=" << m_syncPrefix << + " queue_depth=" << queueDepth << " action=fallback_serial"); + sendSyncInterestSerial(); + return; + } + + incrementStat(m_syncProductionJobsSubmitted); + incrementStat(m_syncInterestMainThreadBlockingMs, elapsedMs(mainStart, SteadyClock::now())); +} + +void +SVSyncCore::sendSyncInterestSerial() +{ + if (!m_initialized) + return; + + const auto publishStart = SteadyClock::now(); + NDN_LOG_TRACE("event=response_encode_start key=" << m_syncPrefix); + // Build app parameters ndn::encoding::EncodingBuffer enc; { @@ -279,7 +1003,11 @@ SVSyncCore::sendSyncInterest() Interest interest(Name(m_syncPrefix).appendVersion(2)); interest.setApplicationParameters(wire); interest.setInterestLifetime(1_ms); + NDN_LOG_TRACE("event=response_encode_done key=" << interest.getName() << + " elapsed_us=" << elapsedUs(publishStart, SteadyClock::now())); + const auto signStart = SteadyClock::now(); + NDN_LOG_TRACE("event=response_sign_start key=" << interest.getName()); switch (m_securityOptions.interestSigner->signingInfo.getSignerType()) { case security::SigningInfo::SIGNER_TYPE_NULL: break; @@ -292,44 +1020,143 @@ SVSyncCore::sendSyncInterest() m_securityOptions.interestSigner->sign(interest); break; } + NDN_LOG_TRACE("event=response_sign_done key=" << interest.getName() << + " elapsed_us=" << elapsedUs(signStart, SteadyClock::now())); + const auto faceStart = SteadyClock::now(); + NDN_LOG_TRACE("event=face_put_start key=" << interest.getName() << + " operation=expressInterest"); m_face.expressInterest(interest, nullptr, nullptr, nullptr); + NDN_LOG_TRACE("event=face_put_done key=" << interest.getName() << + " operation=expressInterest elapsed_us=" << elapsedUs(faceStart, SteadyClock::now())); + incrementStat(m_syncMainThreadPublishMs, elapsedMs(publishStart, SteadyClock::now())); +} + +void +SVSyncCore::processSyncProductionResult(SyncProductionResult result) +{ + const auto mainStart = SteadyClock::now(); + const auto& traceKey = result.job.traceKey; + + if (!result.ok) { + incrementStat(m_syncProductionJobsDropped); + NDN_LOG_DEBUG("event=sync_production_failed key=" << traceKey); + return; + } + + uint64_t currentGeneration = 0; + { + std::lock_guard lock(m_vvMutex); + currentGeneration = m_stateGeneration; + } + + if (currentGeneration != result.job.stateGeneration && !result.extraBlockBuiltInWorker) { + incrementStat(m_syncProductionJobsStale); + NDN_LOG_DEBUG("event=sync_production_stale key=" << traceKey << + " captured_generation=" << result.job.stateGeneration << + " current_generation=" << currentGeneration << + " action=drop"); + return; + } + + if (currentGeneration != result.job.stateGeneration) { + incrementStat(m_syncProductionJobsStale); + NDN_LOG_DEBUG("event=sync_production_stale key=" << traceKey << + " captured_generation=" << result.job.stateGeneration << + " current_generation=" << currentGeneration << + " action=send_extra_block"); + } + + NDN_LOG_TRACE("event=response_encode_done mode=parallel key=" << result.interest.getName() << + " elapsed_us=" << result.encodeUs); + + if (!result.signedInWorker) { + const auto signStart = SteadyClock::now(); + NDN_LOG_TRACE("event=response_sign_start mode=parallel key=" << result.interest.getName()); + signSyncInterest(result.interest); + NDN_LOG_TRACE("event=response_sign_done mode=parallel key=" << result.interest.getName() << + " elapsed_us=" << elapsedUs(signStart, SteadyClock::now())); + } + else { + NDN_LOG_TRACE("event=response_sign_done mode=parallel-worker-result key=" + << result.interest.getName() << " elapsed_us=" << result.signUs); + } + + const auto faceStart = SteadyClock::now(); + NDN_LOG_TRACE("event=face_put_start mode=parallel key=" << result.interest.getName() << + " operation=expressInterest"); + m_face.expressInterest(result.interest, nullptr, nullptr, nullptr); + NDN_LOG_TRACE("event=face_put_done mode=parallel key=" << result.interest.getName() << + " operation=expressInterest elapsed_us=" << elapsedUs(faceStart, SteadyClock::now())); + + const auto mainMs = elapsedMs(mainStart, SteadyClock::now()); + const auto totalMs = elapsedMs(result.job.submittedAt, SteadyClock::now()); + incrementStat(m_syncProductionJobsCompleted); + incrementStat(m_syncProductionParallelTotalMs, totalMs); + incrementStat(m_syncMainThreadPublishMs, mainMs); + incrementStat(m_syncInterestMainThreadBlockingMs, mainMs); +} + +void +SVSyncCore::signSyncInterest(Interest& interest) +{ + switch (m_securityOptions.interestSigner->signingInfo.getSignerType()) { + case security::SigningInfo::SIGNER_TYPE_NULL: + break; + + case security::SigningInfo::SIGNER_TYPE_HMAC: + m_keyChainMem.sign(interest, m_securityOptions.interestSigner->signingInfo); + break; + + default: + m_securityOptions.interestSigner->sign(interest); + break; + } } SVSyncCore::MergeResult SVSyncCore::mergeStateVector(const VersionVector& vvOther) { std::lock_guard lock(m_vvMutex); - SVSyncCore::MergeResult result; - - // Check if other vector has newer state - for (const auto& entry : vvOther) { - NodeID nidOther = entry.first; - SeqNo seqOther = entry.second; - SeqNo seqCurrent = m_vv.get(nidOther); - if (seqCurrent < seqOther) { - result.otherVectorNew = true; + auto result = computeMergeStateVector(m_vv, vvOther); + m_vv = result.mergedVector; + if (result.otherVectorNew) { + ++m_stateGeneration; + } - SeqNo startSeq = m_vv.get(nidOther) + 1; - result.missingInfo.push_back({ nidOther, startSeq, seqOther, 0 }); + return {result.myVectorNew, result.otherVectorNew, result.missingData}; +} - m_vv.set(nidOther, seqOther); +SVSyncCore::MergeComputationResult +SVSyncCore::computeMergeStateVector(const VersionVector& localVector, + const VersionVector& remoteVector) +{ + MergeComputationResult result; + result.mergedVector = localVector; + + for (const auto& [nidOther, seqEntries] : remoteVector.getAllEntries()) { + for (const auto& [bootstrapTime, seqOther] : seqEntries) { + SeqNo seqCurrent = result.mergedVector.get(nidOther, bootstrapTime); + + if (seqCurrent < seqOther) { + result.otherVectorNew = true; + result.missingData.push_back({nidOther, seqCurrent + 1, seqOther, 0, bootstrapTime}); + result.mergedVector.set(nidOther, bootstrapTime, seqOther); + } } } - // Check if I have newer state - for (const auto& entry : m_vv) { - NodeID nid = entry.first; - SeqNo seq = entry.second; - SeqNo seqOther = vvOther.get(nid); - - // Ignore this node if it was last updated within network RTT - if (time::system_clock::now() - m_vv.getLastUpdate(nid) < m_maxSuppressionTime) - continue; + for (const auto& [nid, seqEntries] : result.mergedVector.getAllEntries()) { + for (const auto& [bootstrapTime, seq] : seqEntries) { + SeqNo seqOther = remoteVector.get(nid, bootstrapTime); - if (seqOther < seq) { - result.myVectorNew = true; + if (seqOther < seq) { + result.myVectorNew = true; + break; + } + } + if (result.myVectorNew) { break; } } @@ -347,23 +1174,43 @@ SVSyncCore::getSeqNo(const NodeID& nid) const { std::lock_guard lock(m_vvMutex); NodeID t_nid = (nid == EMPTY_NODE_ID) ? m_id : nid; + if (t_nid == m_id) { + return m_vv.get(t_nid, m_bootstrapTime); + } return m_vv.get(t_nid); } void SVSyncCore::updateSeqNo(const SeqNo& seq, const NodeID& nid) +{ + NodeID t_nid = (nid == EMPTY_NODE_ID) ? m_id : nid; + updateSeqNo(seq, m_bootstrapTime, t_nid); +} + +void +SVSyncCore::updateSeqNo(const SeqNo& seq, BootstrapTime bootstrapTime, const NodeID& nid) { NodeID t_nid = (nid == EMPTY_NODE_ID) ? m_id : nid; SeqNo prev; { std::lock_guard lock(m_vvMutex); - prev = m_vv.get(t_nid); - m_vv.set(t_nid, seq); + prev = m_vv.get(t_nid, bootstrapTime); + m_vv.set(t_nid, bootstrapTime, seq); + if (seq > prev) { + ++m_stateGeneration; + } } if (seq > prev) - retxSyncInterest(false, 1); + { + if (m_syncInterestBatching.load(std::memory_order_relaxed)) { + schedulePublicationSync(); + } + else { + sendLocalPublicationSyncInterest(); + } + } } std::set @@ -395,13 +1242,13 @@ SVSyncCore::recordVector(const VersionVector& vvOther) std::lock_guard lock1(m_vvMutex); - for (const auto& entry : vvOther) { - NodeID nidOther = entry.first; - SeqNo seqOther = entry.second; - SeqNo seqCurrent = m_recordedVv->get(nidOther); + for (const auto& [nidOther, seqEntries] : vvOther.getAllEntries()) { + for (const auto& [bootstrapTime, seqOther] : seqEntries) { + SeqNo seqCurrent = m_recordedVv->get(nidOther, bootstrapTime); - if (seqCurrent < seqOther) { - m_recordedVv->set(nidOther, seqOther); + if (seqCurrent < seqOther) { + m_recordedVv->set(nidOther, bootstrapTime, seqOther); + } } } diff --git a/ndn-svs/core.hpp b/ndn-svs/core.hpp index 977a762..68d7438 100644 --- a/ndn-svs/core.hpp +++ b/ndn-svs/core.hpp @@ -25,7 +25,10 @@ #include #include +#include #include +#include +#include namespace ndn::svs { @@ -40,6 +43,8 @@ class MissingDataInfo SeqNo high; /// @brief ndn::lp::IncomingFaceIdTag uint64_t incomingFace; + /// @brief bootstrap time identifying this node's current SVS session + BootstrapTime bootstrapTime = 0; }; /** @@ -78,6 +83,76 @@ class SVSyncCore : noncopyable const SecurityOptions& securityOptions = SecurityOptions::DEFAULT, const NodeID& nid = EMPTY_NODE_ID); + ~SVSyncCore(); + + struct SyncProcessingStats + { + uint64_t syncJobsSubmitted = 0; + uint64_t syncJobsCompleted = 0; + uint64_t syncJobsDropped = 0; + uint64_t syncJobsStale = 0; + uint64_t syncWorkerQueueDepth = 0; + uint64_t syncWorkerProcessingMs = 0; + uint64_t syncMainThreadPublishMs = 0; + uint64_t syncInterestSerialHandlerMs = 0; + uint64_t syncInterestParallelTotalMs = 0; + uint64_t syncInterestMainThreadBlockingMs = 0; + uint64_t syncProductionJobsSubmitted = 0; + uint64_t syncProductionJobsCompleted = 0; + uint64_t syncProductionJobsDropped = 0; + uint64_t syncProductionJobsStale = 0; + uint64_t syncProductionWorkerQueueDepth = 0; + uint64_t syncProductionWorkerProcessingMs = 0; + uint64_t syncProductionParallelTotalMs = 0; + }; + + void + setParallelSyncProcessing(bool enabled, size_t workerThreads = 1, + size_t maxQueueSize = 1024); + + void + setParallelSyncProduction(bool enabled, size_t workerThreads = 1, + size_t maxQueueSize = 1024, + bool signInWorker = false, + bool buildExtraBlockInWorker = false); + + /** + * @brief Experimentally coalesce locally triggered sync interests. + * + * This only delays sync interests triggered by local publication + * updateSeqNo() calls. Sync interests needed for reply/suppression and the + * wire format are unchanged. + */ + void + setSyncInterestBatching(bool enabled, + time::milliseconds window = 5_ms); + + /** + * @brief Set the maximum randomized suppression delay for replying to Sync + * Interests when this node has newer state. + * + * This is an implementation timer only; it does not affect the SVS wire + * format or protocol semantics. Smaller values are useful for small, + * latency-sensitive groups, while larger values reduce duplicate Sync + * Interests in large groups. + */ + void + setMaxSuppressionTime(time::milliseconds delay); + + /** + * @brief Set the periodic Sync Interest retransmission interval. + * + * This is a diagnostic/testing override. Production and benchmark runs should + * normally keep the library default because this timer affects piggyback + * opportunities. Values below 1 ms are clamped to 1 ms. + */ + void + setPeriodicSyncTime(time::milliseconds interval, + double jitter = 0.1); + + SyncProcessingStats + getSyncProcessingStats() const; + /** * @brief Reset the sync tree (and restart synchronization again) * @@ -106,6 +181,11 @@ class SVSyncCore : noncopyable */ SeqNo getSeqNo(const NodeID& nid = EMPTY_NODE_ID) const; + BootstrapTime getBootstrapTime() const + { + return m_bootstrapTime; + } + /** * @brief Update the seqNo of the local session * @@ -116,6 +196,8 @@ class SVSyncCore : noncopyable */ void updateSeqNo(const SeqNo& seq, const NodeID& nid = EMPTY_NODE_ID); + void updateSeqNo(const SeqNo& seq, BootstrapTime bootstrapTime, const NodeID& nid); + /// @brief Get all the nodeIDs std::set getNodeIds() const; @@ -158,6 +240,10 @@ class SVSyncCore : noncopyable void onSyncInterestValidated(const Interest& interest); + void + onSyncInterestValidatedSerial(const Interest& interest, + bool countSerialStats = true); + /** * @brief Mark the instance as initialized and send the first interest */ @@ -190,6 +276,19 @@ class SVSyncCore : noncopyable std::vector missingInfo; }; + void + sendSyncInterestSerial(); + + /** + * @brief Send a Sync Interest for a local publication. + * + * Local publications must advertise the newly increased sequence number even + * if a recently received remote state vector would otherwise suppress a + * normal retransmission timer. + */ + void + sendLocalPublicationSyncInterest(); + /** * @brief Merge state vector into the current * @param vvOther state vector to merge in @@ -197,6 +296,18 @@ class SVSyncCore : noncopyable */ MergeResult mergeStateVector(const VersionVector& vvOther); + struct MergeComputationResult + { + bool myVectorNew = false; + bool otherVectorNew = false; + std::vector missingData; + VersionVector mergedVector; + }; + + static MergeComputationResult + computeMergeStateVector(const VersionVector& localVector, + const VersionVector& remoteVector); + /** * @brief Record vector by merging it into m_recordedVv * @param vvOther state vector to merge in @@ -226,11 +337,33 @@ class SVSyncCore : noncopyable static inline const NodeID EMPTY_NODE_ID; private: + struct SyncProcessingJob; + struct SyncProcessingResult; + struct SyncProductionJob; + struct SyncProductionResult; + class SyncWorkerPool; + + void + processSyncInterestResult(SyncProcessingResult result); + + void + processSyncProductionResult(SyncProductionResult result); + + void + signSyncInterest(Interest& interest); + + void + schedulePublicationSync(); + + void + incrementStat(std::atomic& counter, uint64_t value = 1) const; + // Communication ndn::Face& m_face; const Name m_syncPrefix; const SecurityOptions m_securityOptions; const NodeID m_id; + const BootstrapTime m_bootstrapTime; ndn::ScopedRegisteredPrefixHandle m_syncRegisteredPrefix; const UpdateCallback m_onUpdate; @@ -238,6 +371,7 @@ class SVSyncCore : noncopyable // State VersionVector m_vv; mutable std::mutex m_vvMutex; + uint64_t m_stateGeneration = 0; // Aggregates incoming vectors while in suppression state std::unique_ptr m_recordedVv = nullptr; mutable std::mutex m_recordedVvMutex; @@ -270,12 +404,44 @@ class SVSyncCore : noncopyable mutable std::mutex m_schedulerMutex; scheduler::ScopedEventId m_retxEvent; scheduler::ScopedEventId m_packetEvent; + scheduler::ScopedEventId m_publicationSyncEvent; + bool m_publicationSyncPending = false; // Time at which the next sync interest will be sent std::atomic_long m_nextSyncInterest; // Prevent sending interests before initialization bool m_initialized = false; + std::atomic m_syncInterestBatching{false}; + time::milliseconds m_syncInterestBatchWindow = 5_ms; + + std::atomic m_parallelSyncProcessing{false}; + std::shared_ptr> m_syncProcessingAlive; + std::unique_ptr m_syncWorkerPool; + std::atomic m_syncJobsSubmitted{0}; + std::atomic m_syncJobsCompleted{0}; + std::atomic m_syncJobsDropped{0}; + std::atomic m_syncJobsStale{0}; + std::atomic m_syncWorkerQueueDepth{0}; + std::atomic m_syncWorkerProcessingMs{0}; + std::atomic m_syncMainThreadPublishMs{0}; + std::atomic m_syncInterestSerialHandlerMs{0}; + std::atomic m_syncInterestParallelTotalMs{0}; + std::atomic m_syncInterestMainThreadBlockingMs{0}; + + std::atomic m_parallelSyncProduction{false}; + std::atomic m_parallelSyncProductionSigning{false}; + std::atomic m_parallelSyncProductionExtraBlock{false}; + std::shared_ptr> m_syncProductionAlive; + std::unique_ptr m_syncProductionWorkerPool; + std::mutex m_syncProductionSigningMutex; + std::atomic m_syncProductionJobsSubmitted{0}; + std::atomic m_syncProductionJobsCompleted{0}; + std::atomic m_syncProductionJobsDropped{0}; + std::atomic m_syncProductionJobsStale{0}; + std::atomic m_syncProductionWorkerQueueDepth{0}; + std::atomic m_syncProductionWorkerProcessingMs{0}; + std::atomic m_syncProductionParallelTotalMs{0}; }; } // namespace ndn::svs diff --git a/ndn-svs/mapping-provider.cpp b/ndn-svs/mapping-provider.cpp index 7d47150..f056aca 100644 --- a/ndn-svs/mapping-provider.cpp +++ b/ndn-svs/mapping-provider.cpp @@ -17,8 +17,42 @@ #include "mapping-provider.hpp" #include "tlv.hpp" +#include + namespace ndn::svs { +namespace { + +Name::Component +makeBootstrapComponent(BootstrapTime bootstrapTime) +{ + return Name::Component("t=" + std::to_string(bootstrapTime)); +} + +Name::Component +makeSeqComponent(SeqNo seqNo) +{ + return Name::Component("seq=" + std::to_string(seqNo)); +} + +uint64_t +parseNamedNumber(const Name::Component& component, std::string_view prefix) +{ + const auto value = component.toUri(); + if (value.rfind(prefix, 0) != 0) { + NDN_THROW(std::invalid_argument("unexpected named number component")); + } + return std::stoull(value.substr(prefix.size())); +} + +Name +makeMappingKey(const NodeID& nodeId, BootstrapTime bootstrapTime, SeqNo seqNo) +{ + return Name(nodeId).append(makeBootstrapComponent(bootstrapTime)).append(makeSeqComponent(seqNo)); +} + +} // namespace + MappingList::MappingList() = default; MappingList::MappingList(const NodeID& nid) @@ -39,8 +73,17 @@ MappingList::MappingList(const Block& block) if (it->type() == tlv::MappingEntry) { it->parse(); - // SeqNo and ApplicationName - SeqNo seqNo = ndn::encoding::readNonNegativeInteger(it->elements().at(0)); + auto seqNoEntry = it->elements().at(0); + seqNoEntry.parse(); + if (seqNoEntry.type() != tlv::SeqNoEntry || + seqNoEntry.elements().size() < 2 || + seqNoEntry.elements().at(0).type() != tlv::BootstrapTime || + seqNoEntry.elements().at(1).type() != tlv::SeqNo) { + NDN_THROW(ndn::tlv::Error("SeqNoEntry", seqNoEntry.type())); + } + BootstrapTime bootstrapTime = + ndn::encoding::readNonNegativeInteger(seqNoEntry.elements().at(0)); + SeqNo seqNo = ndn::encoding::readNonNegativeInteger(seqNoEntry.elements().at(1)); Name name(it->elements().at(1)); // Additional blocks @@ -48,7 +91,7 @@ MappingList::MappingList(const Block& block) for (auto it2 = it->elements().begin() + 2; it2 != it->elements().end(); it2++) blocks.push_back(*it2); - pairs.push_back({ seqNo, std::make_pair(name, blocks) }); + pairs.push_back({ bootstrapTime, seqNo, std::make_pair(name, blocks) }); continue; } } @@ -60,18 +103,25 @@ MappingList::encode() const ndn::encoding::EncodingBuffer enc; size_t totalLength = 0; - for (const auto& [seq, mapping] : pairs) { + for (const auto& entry : pairs) { size_t entryLength = 0; // Additional blocks - for (const auto& block : mapping.second) + for (const auto& block : entry.mapping.second) entryLength += ndn::encoding::prependBlock(enc, block); // Name - entryLength += ndn::encoding::prependBlock(enc, mapping.first.wireEncode()); - - // SeqNo - entryLength += ndn::encoding::prependNonNegativeIntegerBlock(enc, tlv::SeqNo, seq); + entryLength += ndn::encoding::prependBlock(enc, entry.mapping.first.wireEncode()); + + // SeqNoEntry + size_t seqEntryLength = 0; + seqEntryLength += ndn::encoding::prependNonNegativeIntegerBlock(enc, tlv::SeqNo, + entry.seqNo); + seqEntryLength += ndn::encoding::prependNonNegativeIntegerBlock(enc, tlv::BootstrapTime, + entry.bootstrapTime); + entryLength += enc.prependVarNumber(seqEntryLength); + entryLength += enc.prependVarNumber(tlv::SeqNoEntry); + entryLength += seqEntryLength; totalLength += enc.prependVarNumber(entryLength); totalLength += enc.prependVarNumber(tlv::MappingEntry); @@ -101,27 +151,38 @@ MappingProvider::MappingProvider(const Name& syncPrefix, } void -MappingProvider::insertMapping(const NodeID& nodeId, const SeqNo& seqNo, const MappingEntryPair& entry) +MappingProvider::insertMapping(const NodeID& nodeId, BootstrapTime bootstrapTime, + const SeqNo& seqNo, const MappingEntryPair& entry) { - m_map[Name(nodeId).appendNumber(seqNo)] = entry; + std::lock_guard lock(m_mapMutex); + m_map[makeMappingKey(nodeId, bootstrapTime, seqNo)] = entry; } MappingEntryPair -MappingProvider::getMapping(const NodeID& nodeId, const SeqNo& seqNo) +MappingProvider::getMapping(const NodeID& nodeId, BootstrapTime bootstrapTime, + const SeqNo& seqNo) { - return m_map.at(Name(nodeId).appendNumber(seqNo)); + std::lock_guard lock(m_mapMutex); + return m_map.at(makeMappingKey(nodeId, bootstrapTime, seqNo)); } void MappingProvider::onMappingQuery(const Interest& interest) { - MissingDataInfo query = parseMappingQueryDataName(interest.getName()); + MissingDataInfo query; + try { + query = parseMappingQueryDataName(interest.getName()); + } + catch (const std::exception&) { + return; + } + MappingList queryResponse(query.nodeId); for (SeqNo i = query.low; i <= std::max(query.high, query.low); i++) { try { - auto mapping = getMapping(query.nodeId, i); - queryResponse.pairs.emplace_back(i, mapping); + auto mapping = getMapping(query.nodeId, query.bootstrapTime, i); + queryResponse.pairs.push_back({query.bootstrapTime, i, mapping}); } catch (const std::exception&) { // TODO: don't give up if not everything is found // Instead return whatever we have and let the client request @@ -167,11 +228,11 @@ MappingProvider::fetchNameMapping(const MissingDataInfo& info, MappingList list(block); // Add all mappings to self - for (const auto& [seq, mapping] : list.pairs) { + for (const auto& entry : list.pairs) { try { - getMapping(info.nodeId, seq); + getMapping(info.nodeId, entry.bootstrapTime, entry.seqNo); } catch (const std::exception&) { - insertMapping(info.nodeId, seq, mapping); + insertMapping(info.nodeId, entry.bootstrapTime, entry.seqNo, entry.mapping); } } @@ -192,17 +253,19 @@ MappingProvider::getMappingQueryDataName(const MissingDataInfo& info) return Name(info.nodeId) .append(m_syncPrefix) .append("MAPPING") - .appendNumber(info.low) - .appendNumber(info.high); + .append(makeBootstrapComponent(info.bootstrapTime)) + .append(makeSeqComponent(info.low)) + .append(makeSeqComponent(info.high)); } MissingDataInfo MappingProvider::parseMappingQueryDataName(const Name& name) { MissingDataInfo info; - info.low = name.get(-2).toNumber(); - info.high = name.get(-1).toNumber(); - info.nodeId = name.getPrefix(-3 - m_syncPrefix.size()); + info.bootstrapTime = parseNamedNumber(name.get(-3), "t="); + info.low = parseNamedNumber(name.get(-2), "seq="); + info.high = parseNamedNumber(name.get(-1), "seq="); + info.nodeId = name.getPrefix(-4 - m_syncPrefix.size()); return info; } diff --git a/ndn-svs/mapping-provider.hpp b/ndn-svs/mapping-provider.hpp index 7c99812..d0726e7 100644 --- a/ndn-svs/mapping-provider.hpp +++ b/ndn-svs/mapping-provider.hpp @@ -21,11 +21,19 @@ #include "fetcher.hpp" #include +#include namespace ndn::svs { using MappingEntryPair = std::pair>; +struct MappingEntry +{ + BootstrapTime bootstrapTime = 0; + SeqNo seqNo = 0; + MappingEntryPair mapping; +}; + /** * @brief TLV type for mapping list */ @@ -44,7 +52,7 @@ class MappingList public: NodeID nodeId; - std::vector> pairs; + std::vector pairs; }; /** @@ -65,14 +73,16 @@ class MappingProvider : noncopyable /** * @brief Insert a mapping entry into the store */ - void insertMapping(const NodeID& nodeId, const SeqNo& seqNo, const MappingEntryPair& entry); + void insertMapping(const NodeID& nodeId, BootstrapTime bootstrapTime, + const SeqNo& seqNo, const MappingEntryPair& entry); /** * @brief Get a mapping and throw if not found * * @returns Corresponding application name */ - MappingEntryPair getMapping(const NodeID& nodeId, const SeqNo& seqNo); + MappingEntryPair getMapping(const NodeID& nodeId, BootstrapTime bootstrapTime, + const SeqNo& seqNo); /** * @brief Retrieve the data mappings for encapsulated data packets @@ -119,6 +129,7 @@ class MappingProvider : noncopyable ndn::ScopedRegisteredPrefixHandle m_registeredPrefix; std::map m_map; + std::mutex m_mapMutex; }; } // namespace ndn::svs diff --git a/ndn-svs/security-options.cpp b/ndn-svs/security-options.cpp index 2927675..d67bc3d 100644 --- a/ndn-svs/security-options.cpp +++ b/ndn-svs/security-options.cpp @@ -25,7 +25,12 @@ BaseSigner::~BaseSigner() = default; void KeyChainSigner::sign(Interest& interest) const { - m_keyChain.sign(interest, signingInfo); + auto params = signingInfo; + auto sigInfo = params.getSignatureInfo(); + sigInfo.setTime(getFreshInterestTimestamp()); + params.setSignatureInfo(sigInfo); + params.setSignedInterestFormat(security::SignedInterestFormat::V03); + m_keyChain.sign(interest, params); } void @@ -34,6 +39,24 @@ KeyChainSigner::sign(Data& data) const m_keyChain.sign(data, signingInfo); } +time::system_clock::time_point +KeyChainSigner::getFreshInterestTimestamp() const +{ + std::lock_guard lock(m_interestTimestampMutex); + + auto timestamp = time::system_clock::now(); + if (time::duration_cast(timestamp - m_lastInterestTimestamp) > + time::milliseconds(0)) { + m_lastInterestTimestamp = timestamp; + } + else { + m_lastInterestTimestamp += time::milliseconds(1); + timestamp = m_lastInterestTimestamp; + } + + return timestamp; +} + SecurityOptions::SecurityOptions(KeyChain& keyChain) : interestSigner(std::make_shared(keyChain)) , dataSigner(std::make_shared(keyChain)) diff --git a/ndn-svs/security-options.hpp b/ndn-svs/security-options.hpp index ac34621..364c2a9 100644 --- a/ndn-svs/security-options.hpp +++ b/ndn-svs/security-options.hpp @@ -19,6 +19,8 @@ #include "common.hpp" +#include + namespace ndn::svs { /** @@ -85,8 +87,13 @@ class KeyChainSigner : public BaseSigner void sign(Data& data) const override; +private: + time::system_clock::time_point getFreshInterestTimestamp() const; + private: KeyChain& m_keyChain; + mutable std::mutex m_interestTimestampMutex; + mutable time::system_clock::time_point m_lastInterestTimestamp; }; /** diff --git a/ndn-svs/svspubsub.cpp b/ndn-svs/svspubsub.cpp index edcd003..6c2ce6f 100644 --- a/ndn-svs/svspubsub.cpp +++ b/ndn-svs/svspubsub.cpp @@ -15,13 +15,41 @@ */ #include "svspubsub.hpp" +#include "tlv.hpp" #include +#include +#include +#include + +#include #include +#include namespace ndn::svs { +NDN_LOG_INIT(ndn_svs.SVSPubSub); + +namespace { + +size_t +envSizeOrDefault(const char* name, size_t defaultValue) +{ + const char* value = std::getenv(name); + if (value == nullptr || *value == '\0') { + return defaultValue; + } + try { + return std::max(1, static_cast(std::stoull(value))); + } + catch (...) { + return defaultValue; + } +} + +} // namespace + SVSPubSub::SVSPubSub(const Name& syncPrefix, const Name& nodePrefix, ndn::Face& face, @@ -41,9 +69,29 @@ SVSPubSub::SVSPubSub(const Name& syncPrefix, securityOptions, options.dataStore) , m_mappingProvider(syncPrefix, nodePrefix, face, securityOptions) + , m_asyncPublishWorkers(4) + , m_asyncPublishAlive(std::make_shared(true)) { + MAX_SIZE_OF_APPLICATION_PARAMETERS = + envSizeOrDefault("NDNSF_SVS_MAX_APP_PARAMS_BYTES", + MAX_SIZE_OF_APPLICATION_PARAMETERS); + MAX_SIZE_OF_PIGGYDATA = + envSizeOrDefault("NDNSF_SVS_MAX_PIGGYDATA_BYTES", + MAX_SIZE_OF_PIGGYDATA); + m_piggyDataCacheLimit = + envSizeOrDefault("NDNSF_SVS_PIGGYDATA_CACHE_LIMIT", + m_piggyDataCacheLimit); m_svsync.getCore().setGetExtraBlockCallback(std::bind(&SVSPubSub::onGetExtraData, this, _1)); - m_svsync.getCore().setRecvExtraBlockCallback(std::bind(&SVSPubSub::onRecvExtraData, this, _1)); + m_svsync.getCore().setRecvExtraBlockCallback(std::bind(&SVSPubSub::onRecvExtraData, this, _1, _2)); +} + +SVSPubSub::~SVSPubSub() +{ + if (m_asyncPublishAlive) { + m_asyncPublishAlive->store(false, std::memory_order_relaxed); + } + m_asyncPublishWorkers.stop(); + m_asyncPublishWorkers.join(); } SeqNo @@ -59,7 +107,8 @@ SVSPubSub::publish(const Name& name, auto finalBlock = name::Component::fromSegment(nSegments - 1); NodeID nid = nodePrefix == EMPTY_NAME ? m_dataPrefix : nodePrefix; - SeqNo seqNo = m_svsync.getCore().getSeqNo(nid) + 1; + SeqNo seqNo = reserveSeqNo(nid); + BootstrapTime bootstrapTime = m_svsync.getCore().getBootstrapTime(); for (size_t i = 0; i < nSegments; i++) { // Create encapsulated segment @@ -80,7 +129,7 @@ SVSPubSub::publish(const Name& name, } // Insert mapping and manually update the sequence number - insertMapping(nid, seqNo, name, mappingBlocks); + insertMapping(nid, bootstrapTime, seqNo, name, mappingBlocks); m_svsync.getCore().updateSeqNo(seqNo, nid); return seqNo; } else { @@ -88,21 +137,376 @@ SVSPubSub::publish(const Name& name, data.setContent(value); data.setFreshnessPeriod(freshnessPeriod); m_securityOptions.dataSigner->sign(data); + // if the data size is smaller than MAX_SIZE_OF_PIGGYDATA, add it to the piggyback queue + const auto wireSize = data.wireEncode().size(); + if (wireSize <= MAX_SIZE_OF_PIGGYDATA) { + std::lock_guard lock(m_extraDataMutex); + m_piggyDataQueue.push_back({data, 0}); + } + else { + NDN_LOG_TRACE("event=piggyback_skip reason=data_too_large bytes=" << wireSize + << " limit=" << MAX_SIZE_OF_PIGGYDATA + << " data=" << data.getName()); + } return publishPacket(data, nodePrefix); } } SeqNo -SVSPubSub::publishPacket(const Data& data, const Name& nodePrefix, std::vector mappingBlocks) +SVSPubSub::publishAsync(const Name& name, span value, + const Name& nodePrefix, time::milliseconds freshnessPeriod, + std::vector mappingBlocks) +{ + NodeID nid = nodePrefix == EMPTY_NAME ? m_dataPrefix : nodePrefix; + SeqNo seqNo = reserveSeqNo(nid); + + AsyncPublication publication; + publication.kind = AsyncPublication::Kind::Bytes; + publication.bootstrapTime = m_svsync.getCore().getBootstrapTime(); + publication.seqNo = seqNo; + publication.name = name; + publication.nodePrefix = nid; + publication.freshnessPeriod = freshnessPeriod; + publication.value.assign(value.begin(), value.end()); + publication.mappingBlocks = std::move(mappingBlocks); + enqueueAsyncPublication(std::move(publication)); + + return seqNo; +} + + +SeqNo +SVSPubSub::publish(const Name& name, + const Name& nodePrefix, time::milliseconds freshnessPeriod, + std::vector mappingBlocks) { + // Segment the data if larger than MAX_DATA_SIZE NodeID nid = nodePrefix == EMPTY_NAME ? m_dataPrefix : nodePrefix; - SeqNo seqNo = m_svsync.publishData(data.wireEncode(), data.getFreshnessPeriod(), nid, ndn::tlv::Data); - insertMapping(nid, seqNo, data.getName(), mappingBlocks); + SeqNo seqNo = reserveSeqNo(nid); + BootstrapTime bootstrapTime = m_svsync.getCore().getBootstrapTime(); + + // Insert mapping and manually update the sequence number + insertMapping(nid, bootstrapTime, seqNo, name, mappingBlocks); + m_svsync.getCore().updateSeqNo(seqNo, nid); + + return seqNo; +} + +SeqNo +SVSPubSub::publishAsync(const Name& name, + const Name& nodePrefix, time::milliseconds freshnessPeriod, + std::vector mappingBlocks) +{ + NodeID nid = nodePrefix == EMPTY_NAME ? m_dataPrefix : nodePrefix; + SeqNo seqNo = reserveSeqNo(nid); + + AsyncPublication publication; + publication.kind = AsyncPublication::Kind::NameOnly; + publication.bootstrapTime = m_svsync.getCore().getBootstrapTime(); + publication.seqNo = seqNo; + publication.name = name; + publication.nodePrefix = nid; + publication.freshnessPeriod = freshnessPeriod; + publication.mappingBlocks = std::move(mappingBlocks); + enqueueAsyncPublication(std::move(publication)); + + return seqNo; +} + +SeqNo +SVSPubSub::publishPacket(const Data& data, const Name& nodePrefix, + std::vector mappingBlocks) +{ + NodeID nid = nodePrefix == EMPTY_NAME ? m_dataPrefix : nodePrefix; + SeqNo seqNo = reserveSeqNo(nid); + BootstrapTime bootstrapTime = m_svsync.getCore().getBootstrapTime(); + m_svsync.insertDataAtSeq(data.wireEncode(), data.getFreshnessPeriod(), + nid, seqNo, ndn::tlv::Data); + insertMapping(nid, bootstrapTime, seqNo, data.getName(), mappingBlocks); + m_svsync.getCore().updateSeqNo(seqNo, nid); + return seqNo; +} + +SeqNo +SVSPubSub::publishPacketAsync(const Data& data, const Name& nodePrefix, + std::vector mappingBlocks) +{ + NodeID nid = nodePrefix == EMPTY_NAME ? m_dataPrefix : nodePrefix; + SeqNo seqNo = reserveSeqNo(nid); + + AsyncPublication publication; + publication.kind = AsyncPublication::Kind::Packet; + publication.bootstrapTime = m_svsync.getCore().getBootstrapTime(); + publication.seqNo = seqNo; + publication.name = data.getName(); + publication.nodePrefix = nid; + publication.freshnessPeriod = data.getFreshnessPeriod(); + publication.packet = data; + publication.mappingBlocks = std::move(mappingBlocks); + enqueueAsyncPublication(std::move(publication)); + + return seqNo; +} + +SeqNo +SVSPubSub::reserveSeqNo(const NodeID& nid) +{ + std::lock_guard lock(m_asyncPublishMutex); + SeqNo& reserved = m_reservedSeqNo[nid]; + reserved = std::max(reserved, m_svsync.getCore().getSeqNo(nid)); + SeqNo seqNo = ++reserved; + if (m_nextAsyncCommitSeq.find(nid) == m_nextAsyncCommitSeq.end()) { + m_nextAsyncCommitSeq[nid] = seqNo; + } return seqNo; } void -SVSPubSub::insertMapping(const NodeID& nid, SeqNo seqNo, const Name& name, std::vector additional) +SVSPubSub::enqueueAsyncPublication(AsyncPublication publication) +{ + auto alive = m_asyncPublishAlive; + boost::asio::post(m_asyncPublishWorkers, + [this, alive, publication = std::move(publication)] () mutable { + if (!alive || !alive->load(std::memory_order_relaxed)) { + return; + } + prepareAsyncPublication(std::move(publication)); + }); +} + +void +SVSPubSub::prepareAsyncPublication(AsyncPublication publication) +{ + PreparedPublication prepared; + try { + switch (publication.kind) { + case AsyncPublication::Kind::Bytes: + prepared = prepareReservedBytes(publication); + break; + case AsyncPublication::Kind::NameOnly: + prepared = prepareReservedNameOnly(publication); + break; + case AsyncPublication::Kind::Packet: + prepared = prepareReservedPacket(publication); + break; + } + } + catch (const std::exception& e) { + NDN_LOG_DEBUG("event=async_publish_prepare_failed node=" << publication.nodePrefix + << " seq=" << publication.seqNo << " error=" << e.what()); + prepared.seqNo = publication.seqNo; + prepared.nodePrefix = publication.nodePrefix; + prepared.ok = false; + } + + auto alive = m_asyncPublishAlive; + m_face.getIoContext().post([this, alive, prepared = std::move(prepared)] () mutable { + if (!alive || !alive->load(std::memory_order_relaxed)) { + return; + } + onPreparedPublication(std::move(prepared)); + }); +} + +SVSPubSub::PreparedPublication +SVSPubSub::prepareReservedBytes(const AsyncPublication& publication) +{ + PreparedPublication prepared; + prepared.seqNo = publication.seqNo; + prepared.bootstrapTime = publication.bootstrapTime; + prepared.nodePrefix = publication.nodePrefix; + prepared.mappingName = publication.name; + prepared.freshnessPeriod = publication.freshnessPeriod; + prepared.mappingBlocks = publication.mappingBlocks; + + if (publication.value.size() > MAX_DATA_SIZE) { + size_t nSegments = (publication.value.size() / MAX_DATA_SIZE) + 1; + auto finalBlock = name::Component::fromSegment(nSegments - 1); + + for (size_t i = 0; i < nSegments; i++) + { + auto segmentName = Name(publication.name).appendVersion(0).appendSegment(i); + auto segment = Data(segmentName); + segment.setFreshnessPeriod(publication.freshnessPeriod); + + const uint8_t* segVal = publication.value.data() + i * MAX_DATA_SIZE; + const size_t segValSize = std::min(publication.value.size() - i * MAX_DATA_SIZE, + MAX_DATA_SIZE); + segment.setContent(ndn::make_span(segVal, segValSize)); + + segment.setFinalBlock(finalBlock); + { + std::lock_guard lock(m_asyncPublishSigningMutex); + m_securityOptions.dataSigner->sign(segment); + } + + Data outer(m_svsync.getDataName(publication.nodePrefix, publication.bootstrapTime, + publication.seqNo) + .appendVersion(0).appendSegment(i)); + outer.setContent(segment.wireEncode()); + outer.setFreshnessPeriod(publication.freshnessPeriod); + outer.setContentType(ndn::tlv::Data); + outer.setFinalBlock(finalBlock); + { + std::lock_guard lock(m_asyncPublishSigningMutex); + m_securityOptions.dataSigner->sign(outer); + } + outer.wireEncode(); + m_svsync.insertPreparedData(outer, false); + prepared.outerPackets.push_back(std::move(outer)); + } + + insertMapping(publication.nodePrefix, publication.bootstrapTime, publication.seqNo, + publication.name, publication.mappingBlocks); + return prepared; + } + + ndn::Data data(publication.name); + data.setContent(make_span(publication.value.data(), publication.value.size())); + data.setFreshnessPeriod(publication.freshnessPeriod); + { + std::lock_guard lock(m_asyncPublishSigningMutex); + m_securityOptions.dataSigner->sign(data); + } + const auto wireSize = data.wireEncode().size(); + if (wireSize <= MAX_SIZE_OF_PIGGYDATA) { + prepared.piggyPacket = data; + } + else { + NDN_LOG_TRACE("event=piggyback_skip reason=data_too_large bytes=" << wireSize + << " limit=" << MAX_SIZE_OF_PIGGYDATA + << " data=" << data.getName()); + } + + AsyncPublication packetPublication; + packetPublication.kind = AsyncPublication::Kind::Packet; + packetPublication.bootstrapTime = publication.bootstrapTime; + packetPublication.seqNo = publication.seqNo; + packetPublication.name = data.getName(); + packetPublication.nodePrefix = publication.nodePrefix; + packetPublication.freshnessPeriod = data.getFreshnessPeriod(); + packetPublication.packet = data; + packetPublication.mappingBlocks = publication.mappingBlocks; + auto packetPrepared = prepareReservedPacket(packetPublication); + packetPrepared.piggyPacket = prepared.piggyPacket; + return packetPrepared; +} + +SVSPubSub::PreparedPublication +SVSPubSub::prepareReservedNameOnly(const AsyncPublication& publication) +{ + PreparedPublication prepared; + prepared.seqNo = publication.seqNo; + prepared.bootstrapTime = publication.bootstrapTime; + prepared.nodePrefix = publication.nodePrefix; + prepared.mappingName = publication.name; + prepared.freshnessPeriod = publication.freshnessPeriod; + prepared.mappingBlocks = publication.mappingBlocks; + insertMapping(publication.nodePrefix, publication.bootstrapTime, publication.seqNo, + publication.name, publication.mappingBlocks); + return prepared; +} + +SVSPubSub::PreparedPublication +SVSPubSub::prepareReservedPacket(const AsyncPublication& publication) +{ + PreparedPublication prepared; + prepared.seqNo = publication.seqNo; + prepared.bootstrapTime = publication.bootstrapTime; + prepared.nodePrefix = publication.nodePrefix; + prepared.mappingName = publication.packet.getName(); + prepared.freshnessPeriod = publication.packet.getFreshnessPeriod(); + prepared.mappingBlocks = publication.mappingBlocks; + prepared.putFirstPacketToFace = true; + + Data outer(m_svsync.getDataName(publication.nodePrefix, publication.bootstrapTime, + publication.seqNo)); + outer.setContent(publication.packet.wireEncode()); + outer.setFreshnessPeriod(publication.packet.getFreshnessPeriod()); + outer.setContentType(ndn::tlv::Data); + { + std::lock_guard lock(m_asyncPublishSigningMutex); + m_securityOptions.dataSigner->sign(outer); + } + outer.wireEncode(); + m_svsync.insertPreparedData(outer, false); + prepared.outerPackets.push_back(std::move(outer)); + insertMapping(publication.nodePrefix, publication.bootstrapTime, publication.seqNo, + publication.packet.getName(), publication.mappingBlocks); + return prepared; +} + +void +SVSPubSub::onPreparedPublication(PreparedPublication publication) +{ + const auto nodePrefix = publication.nodePrefix; + { + std::lock_guard lock(m_asyncPublishMutex); + m_preparedPublications[nodePrefix][publication.seqNo] = std::move(publication); + } + commitReadyPreparedPublications(nodePrefix); +} + +void +SVSPubSub::commitReadyPreparedPublications(const NodeID& nid) +{ + std::vector ready; + SeqNo highestSeq = 0; + while (true) { + { + std::lock_guard lock(m_asyncPublishMutex); + auto nextIt = m_nextAsyncCommitSeq.find(nid); + if (nextIt == m_nextAsyncCommitSeq.end()) { + break; + } + auto nodeIt = m_preparedPublications.find(nid); + if (nodeIt == m_preparedPublications.end()) { + break; + } + auto pubIt = nodeIt->second.find(nextIt->second); + if (pubIt == nodeIt->second.end()) { + break; + } + highestSeq = pubIt->second.seqNo; + ready.push_back(std::move(pubIt->second)); + nodeIt->second.erase(pubIt); + if (nodeIt->second.empty()) { + m_preparedPublications.erase(nodeIt); + } + ++nextIt->second; + } + } + + if (ready.empty()) { + return; + } + + for (const auto& publication : ready) { + commitPreparedPublication(publication); + } + m_svsync.getCore().updateSeqNo(highestSeq, nid); +} + +void +SVSPubSub::commitPreparedPublication(const PreparedPublication& publication) +{ + if (!publication.ok) { + return; + } + + if (publication.putFirstPacketToFace && !publication.outerPackets.empty()) { + m_svsync.putPreparedData(publication.outerPackets.front()); + } + + if (publication.piggyPacket) { + std::lock_guard lock(m_extraDataMutex); + m_piggyDataQueue.push_back({*publication.piggyPacket, 0}); + } +} + +void +SVSPubSub::insertMapping(const NodeID& nid, BootstrapTime bootstrapTime, SeqNo seqNo, + const Name& name, std::vector additional) { // additional is a copy deliberately // this way we can add well-known mappings to the list @@ -112,7 +516,7 @@ SVSPubSub::insertMapping(const NodeID& nid, SeqNo seqNo, const Name& name, std:: unsigned long now = std::chrono::duration_cast( std::chrono::system_clock::now().time_since_epoch()) .count(); - auto timestamp = Name::Component::fromNumber(now, tlv::TimestampNameComponent); + auto timestamp = Name::Component::fromNumber(now, ndn::tlv::TimestampNameComponent); additional.push_back(timestamp); } @@ -120,13 +524,16 @@ SVSPubSub::insertMapping(const NodeID& nid, SeqNo seqNo, const Name& name, std:: MappingEntryPair entry = { name, additional }; // notify subscribers in next sync interest - if (m_notificationMappingList.nodeId == EMPTY_NAME || m_notificationMappingList.nodeId == nid) { - m_notificationMappingList.nodeId = nid; - m_notificationMappingList.pairs.push_back({ seqNo, entry }); + { + std::lock_guard lock(m_extraDataMutex); + if (m_notificationMappingList.nodeId == EMPTY_NAME || m_notificationMappingList.nodeId == nid) { + m_notificationMappingList.nodeId = nid; + m_notificationMappingList.pairs.push_back({ bootstrapTime, seqNo, entry }); + } } // send mapping to provider - m_mappingProvider.insertMapping(nid, seqNo, entry); + m_mappingProvider.insertMapping(nid, bootstrapTime, seqNo, entry); } uint32_t @@ -139,10 +546,17 @@ SVSPubSub::subscribe(const Name& prefix, const SubscriptionCallback& callback, b } uint32_t -SVSPubSub::subscribeToProducer(const Name& nodePrefix, - const SubscriptionCallback& callback, - bool prefetch, - bool packets) +SVSPubSub::subscribeWithRegex(const Regex ®ex, const SubscriptionCallback &callback,bool autofetch, bool packets) +{ + uint32_t handle = ++m_subscriptionCount; + Subscription sub = { handle, ndn::Name(), callback, packets, false, autofetch, std::make_shared(regex)}; + m_regexSubscriptions.push_back(sub); + return handle; +} + +uint32_t +SVSPubSub::subscribeToProducer(const Name& nodePrefix, const SubscriptionCallback& callback, + bool prefetch, bool packets) { uint32_t handle = ++m_subscriptionCount; Subscription sub = { handle, nodePrefix, callback, packets, prefetch }; @@ -164,6 +578,7 @@ SVSPubSub::unsubscribe(uint32_t handle) unsub(m_producerSubscriptions); unsub(m_prefixSubscriptions); + unsub(m_regexSubscriptions); } void @@ -177,16 +592,17 @@ SVSPubSub::updateCallbackInternal(const std::vector& info) if (sub.prefix.isPrefixOf(streamName)) { // Add to fetching queue for (SeqNo i = stream.low; i <= stream.high; i++) - m_fetchMap[std::pair(stream.nodeId, i)].push_back(sub); + m_fetchMap[PublicationKey(stream.nodeId, stream.bootstrapTime, i)].push_back(sub); // Prefetch next available data if (sub.prefetch) - m_svsync.fetchData(stream.nodeId, stream.high + 1, [](auto&&...) {}); // do nothing with prefetch + m_svsync.fetchData(stream.nodeId, stream.bootstrapTime, + stream.high + 1, [](auto&&...) {}); // do nothing with prefetch } } - // Fetch all mappings if we have prefix subscription(s) - if (!m_prefixSubscriptions.empty()) { + // Fetch all mappings if we have prefix subscription(s) or regex subscription(s) + if (!m_prefixSubscriptions.empty() || !m_regexSubscriptions.empty()) { MissingDataInfo remainingInfo = stream; // Attemt to find what we already know about mapping @@ -195,7 +611,7 @@ SVSPubSub::updateCallbackInternal(const std::vector& info) for (SeqNo i = remainingInfo.low; i <= remainingInfo.high; i++) { try { // throws if mapping not found - this->processMapping(stream.nodeId, i); + this->processMapping(stream.nodeId, stream.bootstrapTime, i); remainingInfo.low++; } catch (const std::exception&) { break; @@ -216,8 +632,8 @@ SVSPubSub::updateCallbackInternal(const std::vector& info) truncatedRemainingInfo, [this, remainingInfo, streamName](const MappingList& list) { bool queued = false; - for (const auto& [seq, mapping] : list.pairs) - queued |= this->processMapping(streamName, seq); + for (const auto& entry : list.pairs) + queued |= this->processMapping(streamName, entry.bootstrapTime, entry.seqNo); if (queued) this->fetchAll(); @@ -234,17 +650,22 @@ SVSPubSub::updateCallbackInternal(const std::vector& info) } bool -SVSPubSub::processMapping(const NodeID& nodeId, SeqNo seqNo) +SVSPubSub::processMapping(const NodeID& nodeId, BootstrapTime bootstrapTime, SeqNo seqNo) { + const PublicationKey publication(nodeId, bootstrapTime, seqNo); + if (hasPiggyDeliveredPublication(publication)) { + return false; + } + // this will throw if mapping not found - auto mapping = m_mappingProvider.getMapping(nodeId, seqNo); + auto mapping = m_mappingProvider.getMapping(nodeId, bootstrapTime, seqNo); // check if timestamp is too old if (m_opts.maxPubAge > 0_ms) { // look for the additional timestamp block // if no timestamp block is present, we just skip this step for (const auto& block : mapping.second) { - if (block.type() != tlv::TimestampNameComponent) + if (block.type() != ndn::tlv::TimestampNameComponent) continue; unsigned long now = std::chrono::duration_cast( @@ -259,15 +680,68 @@ SVSPubSub::processMapping(const NodeID& nodeId, SeqNo seqNo) } } - // check if known mapping matches subscription bool queued = false; + bool deliveredFromPiggy = false; + auto queueOrDeliver = [this, &queued, &deliveredFromPiggy, &mapping, &publication, + &nodeId, seqNo](const Subscription& sub) { + if (sub.autofetch) { + std::optional packet; + { + std::lock_guard lock(m_extraDataMutex); + auto data = m_piggyDataCache.lower_bound(mapping.first); + auto exactData = m_piggyDataCache.find(mapping.first); + if (exactData != m_piggyDataCache.end()) { + packet = exactData->second; + } + else if (data != m_piggyDataCache.end() && + mapping.first.isPrefixOf(data->first)) { + packet = data->second; + } + } + if (packet) { + SubscriptionData subData = { + mapping.first, + packet->getContent().value_bytes(), + nodeId, + seqNo, + packet + }; + sub.callback(subData); + deliveredFromPiggy = true; + } + else { + m_fetchMap[publication].push_back(sub); + queued = true; + } + } + else { + SubscriptionData subData = { + mapping.first, + ndn::span{}, + nodeId, + seqNo, + std::nullopt + }; + sub.callback(subData); + } + }; + + // check if known mapping matches subscription for (const auto& sub : m_prefixSubscriptions) { if (sub.prefix.isPrefixOf(mapping.first)) { - m_fetchMap[std::pair(nodeId, seqNo)].push_back(sub); - queued = true; + queueOrDeliver(sub); + } + } + for (const auto& sub : m_regexSubscriptions) { + if (sub.regex->match(mapping.first)) { + queueOrDeliver(sub); } } + if (deliveredFromPiggy) { + rememberPiggyDeliveredPublication(publication); + } + return queued; } @@ -282,14 +756,54 @@ SVSPubSub::fetchAll() m_fetchingMap[key] = true; // Fetch first data packet - const auto& [nodeId, seqNo] = key; - m_svsync.fetchData(nodeId, seqNo, std::bind(&SVSPubSub::onSyncData, this, _1, key), 12); + const auto& [nodeId, bootstrapTime, seqNo] = key; + try { + const auto mapping = m_mappingProvider.getMapping(nodeId, bootstrapTime, seqNo); + std::optional packet; + { + std::lock_guard lock(m_extraDataMutex); + auto data = m_piggyDataCache.lower_bound(mapping.first); + auto exactData = m_piggyDataCache.find(mapping.first); + if (exactData != m_piggyDataCache.end()) { + packet = exactData->second; + } + else if (data != m_piggyDataCache.end() && + mapping.first.isPrefixOf(data->first)) { + packet = data->second; + } + } + if (packet) { + SubscriptionData subData = { + mapping.first, + packet->getContent().value_bytes(), + nodeId, + seqNo, + packet, + }; + for (const auto& sub : m_fetchMap[key]) { + sub.callback(subData); + } + rememberPiggyDeliveredPublication(key); + NDN_LOG_TRACE("event=piggyback_cache_satisfy data=" << packet->getName() + << " node=" << nodeId + << " seq=" << seqNo); + cleanUpFetch(key); + continue; + } + } + catch (const std::exception&) { + } + m_svsync.fetchData(nodeId, bootstrapTime, seqNo, + std::bind(&SVSPubSub::onSyncData, this, _1, key), 12); } } void -SVSPubSub::onSyncData(const Data& firstData, const std::pair& publication) +SVSPubSub::onSyncData(const Data& firstData, const PublicationKey& publication) { + const auto& nodeId = std::get<0>(publication); + const auto seqNo = std::get<2>(publication); + // Make sure the data is encapsulated if (firstData.getContentType() != ndn::tlv::Data) { m_fetchingMap[publication] = false; @@ -302,7 +816,7 @@ SVSPubSub::onSyncData(const Data& firstData, const std::pair& publi // Return data to packet subscriptions SubscriptionData subData = { - innerData.getName(), innerContent.value_bytes(), publication.first, publication.second, innerData, + innerData.getName(), innerContent.value_bytes(), nodeId, seqNo, innerData, }; // Function to return data to subscriptions @@ -362,7 +876,8 @@ SVSPubSub::onSyncData(const Data& firstData, const std::pair& publi // Return data to packet subscriptions SubscriptionData subData = { - innerName, *finalBuffer, publication.first, publication.second, std::nullopt, + innerName, *finalBuffer, std::get<0>(publication), std::get<2>(publication), + std::nullopt, }; for (const auto& sub : this->m_fetchMap[publication]) @@ -418,30 +933,227 @@ SVSPubSub::onSyncData(const Data& firstData, const std::pair& publi } void -SVSPubSub::cleanUpFetch(const std::pair& publication) +SVSPubSub::cleanUpFetch(const PublicationKey& publication) { m_fetchMap.erase(publication); m_fetchingMap.erase(publication); } +bool +SVSPubSub::hasPiggyDeliveredPublication(const PublicationKey& publication) +{ + std::lock_guard lock(m_extraDataMutex); + return m_piggyDeliveredPublications.find(publication) != m_piggyDeliveredPublications.end(); +} + +void +SVSPubSub::rememberPiggyDeliveredPublication(const PublicationKey& publication) +{ + std::lock_guard lock(m_extraDataMutex); + if (m_piggyDeliveredPublications.find(publication) == m_piggyDeliveredPublications.end()) { + m_piggyDeliveredPublicationOrder.push_back(publication); + } + m_piggyDeliveredPublications[publication] = true; + while (m_piggyDeliveredPublications.size() > m_piggyDeliveredPublicationLimit && + !m_piggyDeliveredPublicationOrder.empty()) { + m_piggyDeliveredPublications.erase(m_piggyDeliveredPublicationOrder.front()); + m_piggyDeliveredPublicationOrder.pop_front(); + } +} + +bool +SVSPubSub::satisfyPendingFetchFromPiggyData(const Data& data) +{ + std::vector>> ready; + + for (const auto& entry : m_fetchMap) + { + const auto& publication = entry.first; + try { + auto mapping = m_mappingProvider.getMapping(std::get<0>(publication), + std::get<1>(publication), + std::get<2>(publication)); + if (mapping.first == data.getName()) { + ready.push_back(entry); + } + } + catch (const std::exception&) { + } + } + + for (const auto& [publication, subscriptions] : ready) + { + std::optional packet(data); + SubscriptionData subData = { + data.getName(), + data.getContent().value_bytes(), + std::get<0>(publication), + std::get<2>(publication), + packet, + }; + + for (const auto& sub : subscriptions) { + sub.callback(subData); + } + rememberPiggyDeliveredPublication(publication); + cleanUpFetch(publication); + } + + NDN_LOG_TRACE("event=piggyback_satisfy data=" << data.getName() + << " matches=" << ready.size()); + + return !ready.empty(); +} + Block SVSPubSub::onGetExtraData(const VersionVector&) { - MappingList copy = m_notificationMappingList; + std::lock_guard lock(m_extraDataMutex); + + MappingList includedMappings(m_notificationMappingList.nodeId); + size_t mappingCount = 0; + size_t piggyCount = 0; + if (m_notificationMappingList.nodeId != EMPTY_NAME) { + for (const auto& entry : m_notificationMappingList.pairs) { + MappingList candidate = includedMappings; + candidate.pairs.push_back(entry); + auto candidateBlock = candidate.encode(); + if (candidateBlock.size() <= MAX_SIZE_OF_APPLICATION_PARAMETERS) { + includedMappings = std::move(candidate); + ++mappingCount; + } + // Overflow mappings are intentionally omitted from this latency-oriented + // piggyback path; peers can still fetch them from MappingProvider. + } + } + + ndn::Block block = includedMappings.encode(); + block.parse(); + size_t size = block.size(); + size_t retainedCount = 0; + size_t expiredCount = 0; + std::deque retained; + for (auto it = m_piggyDataQueue.rbegin(); it != m_piggyDataQueue.rend(); ++it) { + auto dataBlock = it->data.wireEncode(); + if (size + dataBlock.size() > MAX_SIZE_OF_APPLICATION_PARAMETERS) { + ++it->missed; + if (it->missed < 3) { + retained.push_front(*it); + ++retainedCount; + } + else { + ++expiredCount; + NDN_LOG_TRACE("event=piggyback_drop reason=missed_too_many bytes=" + << dataBlock.size() + << " data=" << it->data.getName()); + } + continue; + } + + block.push_back(dataBlock); + size += dataBlock.size(); + ++piggyCount; + } + m_piggyDataQueue = std::move(retained); + block.encode(); + + NDN_LOG_TRACE("event=piggyback_build mappings=" << mappingCount + << " data=" << piggyCount + << " retained=" << retainedCount + << " expired=" << expiredCount + << " bytes=" << block.size() + << " appParamLimit=" << MAX_SIZE_OF_APPLICATION_PARAMETERS + << " piggyDataLimit=" << MAX_SIZE_OF_PIGGYDATA); + m_notificationMappingList = MappingList(); - return copy.encode(); + + return block; } void -SVSPubSub::onRecvExtraData(const Block& block) +SVSPubSub::onRecvExtraData(const Block& block, const VersionVector&) { + std::vector receivedMappings; try { + NDN_LOG_TRACE("event=piggyback_recv_block type=" << block.type() + << " bytes=" << block.size()); MappingList list(block); - for (const auto& p : list.pairs) { - m_mappingProvider.insertMapping(list.nodeId, p.first, p.second); + for (const auto& entry : list.pairs) { + m_mappingProvider.insertMapping(list.nodeId, entry.bootstrapTime, entry.seqNo, entry.mapping); + receivedMappings.emplace_back(list.nodeId, entry.bootstrapTime, entry.seqNo); } + NDN_LOG_TRACE("event=piggyback_recv_mapping_list node=" << list.nodeId + << " mappings=" << list.pairs.size()); + } catch (const std::exception& e) { + NDN_LOG_TRACE("event=piggyback_recv_mapping_list_error error=" << e.what()); + } + + try { + block.parse(); + size_t mappingCount = 0; + size_t dataCount = 0; + for (const auto& childBlock : block.elements()) { + NDN_LOG_TRACE("event=piggyback_recv_child type=" << childBlock.type() + << " bytes=" << childBlock.size()); + if (childBlock.type() == ndn::svs::tlv::MappingData) { + MappingList childList(childBlock); + for (const auto& entry : childList.pairs) { + m_mappingProvider.insertMapping(childList.nodeId, entry.bootstrapTime, + entry.seqNo, entry.mapping); + receivedMappings.emplace_back(childList.nodeId, entry.bootstrapTime, entry.seqNo); + } + mappingCount += childList.pairs.size(); + } + + if (childBlock.type() == ndn::tlv::Data) { + ndn::Data data(childBlock); + satisfyPendingFetchFromPiggyData(data); + try { + std::lock_guard lock(m_extraDataMutex); + const auto dataName = data.getName(); + if (m_piggyDataCache.find(dataName) == m_piggyDataCache.end()) { + m_piggyDataCacheOrder.push_back(dataName); + } + m_piggyDataCache[dataName] = data; + const auto fullName = data.getFullName(); + if (m_piggyDataCache.find(fullName) == m_piggyDataCache.end()) { + m_piggyDataCacheOrder.push_back(fullName); + } + m_piggyDataCache[fullName] = data; + while (m_piggyDataCache.size() > m_piggyDataCacheLimit && + !m_piggyDataCacheOrder.empty()) { + m_piggyDataCache.erase(m_piggyDataCacheOrder.front()); + m_piggyDataCacheOrder.pop_front(); + } + } + catch (const std::exception&) { + } + ++dataCount; + } + } + NDN_LOG_TRACE("event=piggyback_recv_children mappings=" << mappingCount + << " data=" << dataCount); } catch (const std::exception&) { } + + bool queued = false; + size_t processed = 0; + for (const auto& [nodeId, bootstrapTime, seqNo] : receivedMappings) { + try { + queued |= processMapping(nodeId, bootstrapTime, seqNo); + ++processed; + } + catch (const std::exception&) { + } + } + if (queued) { + fetchAll(); + } + if (!receivedMappings.empty()) { + NDN_LOG_TRACE("event=piggyback_process_mappings processed=" << processed + << " received=" << receivedMappings.size() + << " queued=" << queued); + } } } // namespace ndn::svs diff --git a/ndn-svs/svspubsub.hpp b/ndn-svs/svspubsub.hpp index c34686f..685f36b 100644 --- a/ndn-svs/svspubsub.hpp +++ b/ndn-svs/svspubsub.hpp @@ -24,6 +24,17 @@ #include "svsync.hpp" #include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include namespace ndn::svs { @@ -77,7 +88,7 @@ class SVSPubSub : noncopyable const SVSPubSubOptions& options = {}, const SecurityOptions& securityOptions = SecurityOptions::DEFAULT); - virtual ~SVSPubSub() = default; + virtual ~SVSPubSub(); struct SubscriptionData { @@ -116,6 +127,41 @@ class SVSPubSub : noncopyable time::milliseconds freshnessPeriod = FRESH_FOREVER, std::vector mappingBlocks = {}); + /** + * @brief Queue a publication and return immediately with a reserved seqNo. + * + * The existing publish() API remains synchronous. This API copies the input + * bytes, reserves a sequence number, and prepares signing, encoding, data-store, + * and mapping work on the async publication worker pool. The Face/io_context + * thread only commits prepared publications in sequence order and performs the + * final Face/update operations. + */ + SeqNo + publishAsync(const Name& name, span value, + const Name& nodePrefix = EMPTY_NAME, + time::milliseconds freshnessPeriod = FRESH_FOREVER, + std::vector mappingBlocks = {}); + + /** + * @brief Publish data names only on the pub/sub group. + * + * @param name name for the publication + * @param nodePrefix Name to publish the data under + * @param freshnessPeriod freshness period for the data + * @param mappingBlocks Additional blocks to be published with the mapping (use sparingly) + */ + SeqNo + publish(const Name& name, + const Name& nodePrefix = EMPTY_NAME, + time::milliseconds freshnessPeriod = FRESH_FOREVER, + std::vector mappingBlocks = {}); + + SeqNo + publishAsync(const Name& name, + const Name& nodePrefix = EMPTY_NAME, + time::milliseconds freshnessPeriod = FRESH_FOREVER, + std::vector mappingBlocks = {}); + /** * @brief Subscribe to a application name prefix. * @@ -127,6 +173,18 @@ class SVSPubSub : noncopyable */ uint32_t subscribe(const Name& prefix, const SubscriptionCallback& callback, bool packets = false); + /** + * @brief Subscribe with a regex to name. + * + * @param regex regex of the application data + * @param callback Callback when new data is received + * @param packets Subscribe to the raw Data packets instead of BLOBs + * + * @returns Handle to the subscription + */ + uint32_t + subscribeWithRegex(const Regex& regex, const SubscriptionCallback& callback, bool autofetch = true, bool packets = false); + /** * @brief Subscribe to a data producer * @@ -165,13 +223,18 @@ class SVSPubSub : noncopyable const Name& nodePrefix = EMPTY_NAME, std::vector mappingBlocks = {}); + SeqNo + publishPacketAsync(const Data& data, + const Name& nodePrefix = EMPTY_NAME, + std::vector mappingBlocks = {}); + /** @brief Get the underlying sync */ SVSync& getSVSync() { return m_svsync; } -private: +NDN_SVS_PUBLIC_WITH_TESTS_ELSE_PRIVATE: struct Subscription { uint32_t id; @@ -179,29 +242,110 @@ class SVSPubSub : noncopyable SubscriptionCallback callback; bool isPacketSubscription; bool prefetch; + bool autofetch = true; + std::shared_ptr regex = make_shared("^<>+$"); }; - void onSyncData(const Data& syncData, const std::pair& publication); + using PublicationKey = std::tuple; + + void onSyncData(const Data& syncData, const PublicationKey& publication); void updateCallbackInternal(const std::vector& info); Block onGetExtraData(const VersionVector& vv); - void onRecvExtraData(const Block& block); + void onRecvExtraData(const Block& block, const VersionVector& vv); + + bool + satisfyPendingFetchFromPiggyData(const Data& data); + + bool + hasPiggyDeliveredPublication(const PublicationKey& publication); + + void + rememberPiggyDeliveredPublication(const PublicationKey& publication); /// @brief Insert a mapping entry into the store - void insertMapping(const NodeID& nid, SeqNo seqNo, const Name& name, std::vector additional); + void insertMapping(const NodeID& nid, BootstrapTime bootstrapTime, SeqNo seqNo, + const Name& name, std::vector additional); /** * @brief Get and process mapping from store. * @returns true if new publications were queued for fetch * @throws std::exception error if mapping is not found */ - bool processMapping(const NodeID& nodeId, SeqNo seqNo); + bool processMapping(const NodeID& nodeId, BootstrapTime bootstrapTime, SeqNo seqNo); void fetchAll(); - void cleanUpFetch(const std::pair& publication); + void cleanUpFetch(const PublicationKey& publication); + + SeqNo + reserveSeqNo(const NodeID& nid); + + struct AsyncPublication + { + enum class Kind + { + Bytes, + NameOnly, + Packet + }; + + Kind kind = Kind::Bytes; + BootstrapTime bootstrapTime = 0; + SeqNo seqNo = 0; + Name name; + Name nodePrefix; + time::milliseconds freshnessPeriod = FRESH_FOREVER; + std::vector value; + Data packet; + std::vector mappingBlocks; + }; + + struct PreparedPublication + { + SeqNo seqNo = 0; + BootstrapTime bootstrapTime = 0; + Name nodePrefix; + Name mappingName; + time::milliseconds freshnessPeriod = FRESH_FOREVER; + std::vector outerPackets; + std::optional piggyPacket; + std::vector mappingBlocks; + bool putFirstPacketToFace = false; + bool ok = true; + }; + + struct PiggyDataEntry + { + Data data; + size_t missed = 0; + }; + + void + enqueueAsyncPublication(AsyncPublication publication); + + void + prepareAsyncPublication(AsyncPublication publication); + + void + onPreparedPublication(PreparedPublication publication); + + void + commitReadyPreparedPublications(const NodeID& nid); + + void + commitPreparedPublication(const PreparedPublication& publication); + + PreparedPublication + prepareReservedBytes(const AsyncPublication& publication); + + PreparedPublication + prepareReservedNameOnly(const AsyncPublication& publication); + + PreparedPublication + prepareReservedPacket(const AsyncPublication& publication); public: static inline const Name EMPTY_NAME; @@ -230,10 +374,37 @@ class SVSPubSub : noncopyable uint32_t m_subscriptionCount; std::vector m_producerSubscriptions; std::vector m_prefixSubscriptions; + std::vector m_regexSubscriptions; // Queue of publications to fetch - std::map, std::vector> m_fetchMap; - std::map, bool> m_fetchingMap; + std::map> m_fetchMap; + std::map m_fetchingMap; + + size_t MAX_SIZE_OF_APPLICATION_PARAMETERS = 4096; + size_t MAX_SIZE_OF_PIGGYDATA = 800; + // Pending piggyback Data. Newer entries are tried first; entries that miss + // several Sync Interest opportunities are dropped and peers fetch normally. + std::deque m_piggyDataQueue; + // A bounded cache for received piggyback Data. This intentionally avoids + // ndn-cxx InMemoryStorage because SVS can touch this path from sync callback + // and subscription delivery paths close together when parallel sync is enabled. + std::map m_piggyDataCache; + std::deque m_piggyDataCacheOrder; + size_t m_piggyDataCacheLimit = 1024; + std::map m_piggyDeliveredPublications; + std::deque m_piggyDeliveredPublicationOrder; + size_t m_piggyDeliveredPublicationLimit = 4096; + std::mutex m_extraDataMutex; + + std::mutex m_asyncPublishMutex; + std::map m_reservedSeqNo; + std::deque m_asyncPublishQueue; + std::map m_nextAsyncCommitSeq; + std::map> m_preparedPublications; + + boost::asio::thread_pool m_asyncPublishWorkers; + std::shared_ptr m_asyncPublishAlive; + std::mutex m_asyncPublishSigningMutex; }; } // namespace ndn::svs diff --git a/ndn-svs/svsync-base.cpp b/ndn-svs/svsync-base.cpp index 0f2f4b8..3cc269f 100644 --- a/ndn-svs/svsync-base.cpp +++ b/ndn-svs/svsync-base.cpp @@ -65,7 +65,25 @@ SVSyncBase::publishData(const Block& content, NodeID pubId = id != EMPTY_NODE_ID ? id : m_id; SeqNo newSeq = m_core.getSeqNo(pubId) + 1; - Name dataName = getDataName(pubId, newSeq); + publishDataAtSeq(content, freshness, pubId, newSeq, contentType); + return newSeq; +} + +void +SVSyncBase::publishDataAtSeq(const Block& content, const ndn::time::milliseconds& freshness, + const NodeID& nid, const SeqNo seq, uint32_t contentType) +{ + NodeID pubId = nid != EMPTY_NODE_ID ? nid : m_id; + insertDataAtSeq(content, freshness, pubId, seq, contentType); + m_core.updateSeqNo(seq, pubId); +} + +void +SVSyncBase::insertDataAtSeq(const Block& content, const ndn::time::milliseconds& freshness, + const NodeID& nid, const SeqNo seq, uint32_t contentType) +{ + NodeID pubId = nid != EMPTY_NODE_ID ? nid : m_id; + Name dataName = getDataName(pubId, m_core.getBootstrapTime(), seq); auto data = std::make_shared(dataName); data->setContent(content); data->setFreshnessPeriod(freshness); @@ -73,11 +91,11 @@ SVSyncBase::publishData(const Block& content, m_securityOptions.dataSigner->sign(*data); - m_dataStore->insert(*data); - m_core.updateSeqNo(newSeq, pubId); + { + std::lock_guard lock(m_dataStoreMutex); + m_dataStore->insert(*data); + } m_face.put(*data); - - return newSeq; } void @@ -89,20 +107,46 @@ SVSyncBase::insertDataSegment(const Block& content, const Name::Component& finalBlock, uint32_t contentType) { - Name dataName = getDataName(nid, seq).appendVersion(0).appendSegment(segNo); + Name dataName = getDataName(nid, m_core.getBootstrapTime(), seq).appendVersion(0).appendSegment(segNo); auto data = std::make_shared(dataName); data->setContent(content); data->setFreshnessPeriod(freshness); data->setContentType(contentType); data->setFinalBlock(finalBlock); m_securityOptions.dataSigner->sign(*data); - m_dataStore->insert(*data); + { + std::lock_guard lock(m_dataStoreMutex); + m_dataStore->insert(*data); + } +} + +void +SVSyncBase::insertPreparedData(const Data& data, bool putToFace) +{ + auto preparedData = std::make_shared(data); + { + std::lock_guard lock(m_dataStoreMutex); + m_dataStore->insert(*preparedData); + } + if (putToFace) { + m_face.put(*preparedData); + } +} + +void +SVSyncBase::putPreparedData(const Data& data) +{ + m_face.put(data); } void SVSyncBase::onDataInterest(const Interest& interest) { - auto data = m_dataStore->find(interest); + std::shared_ptr data; + { + std::lock_guard lock(m_dataStoreMutex); + data = m_dataStore->find(interest); + } if (data != nullptr) m_face.put(*data); } @@ -116,18 +160,43 @@ SVSyncBase::fetchData(const NodeID& nid, DataValidationErrorCallback onValidationFailed = std::bind(&SVSyncBase::onDataValidationFailed, this, _1, _2); TimeoutCallback onTimeout = [](auto&&...) {}; - fetchData(nid, seqNo, onValidated, onValidationFailed, onTimeout, nRetries); + fetchData(nid, m_core.getBootstrapTime(), seqNo, onValidated, onValidationFailed, onTimeout, nRetries); +} + +void +SVSyncBase::fetchData(const NodeID& nid, + const BootstrapTime& bootstrapTime, + const SeqNo& seqNo, + const DataValidatedCallback& onValidated, + int nRetries) +{ + DataValidationErrorCallback onValidationFailed = + std::bind(&SVSyncBase::onDataValidationFailed, this, _1, _2); + TimeoutCallback onTimeout = [](auto&&...) {}; + fetchData(nid, bootstrapTime, seqNo, onValidated, onValidationFailed, onTimeout, nRetries); +} + +void +SVSyncBase::fetchData(const NodeID& nid, + const SeqNo& seqNo, + const DataValidatedCallback& onValidated, + const DataValidationErrorCallback& onValidationFailed, + const TimeoutCallback& onTimeout, + int nRetries) +{ + fetchData(nid, m_core.getBootstrapTime(), seqNo, onValidated, onValidationFailed, onTimeout, nRetries); } void SVSyncBase::fetchData(const NodeID& nid, + const BootstrapTime& bootstrapTime, const SeqNo& seqNo, const DataValidatedCallback& onValidated, const DataValidationErrorCallback& onValidationFailed, const TimeoutCallback& onTimeout, int nRetries) { - Name interestName = getDataName(nid, seqNo); + Name interestName = getDataName(nid, bootstrapTime, seqNo); Interest interest(interestName); interest.setCanBePrefix(true); interest.setInterestLifetime(2_s); @@ -143,8 +212,10 @@ SVSyncBase::fetchData(const NodeID& nid, void SVSyncBase::onDataValidated(const Data& data, const DataValidatedCallback& dataCallback) { - if (shouldCache(data)) + if (shouldCache(data)) { + std::lock_guard lock(m_dataStoreMutex); m_dataStore->insert(data); + } dataCallback(data); } diff --git a/ndn-svs/svsync-base.hpp b/ndn-svs/svsync-base.hpp index 8daa8fd..d4427a9 100644 --- a/ndn-svs/svsync-base.hpp +++ b/ndn-svs/svsync-base.hpp @@ -22,6 +22,8 @@ #include "security-options.hpp" #include "store.hpp" +#include + namespace ndn::svs { /** @@ -94,6 +96,29 @@ class SVSyncBase : noncopyable const NodeID& nid = EMPTY_NODE_ID, uint32_t contentType = ndn::tlv::Content); + /** + * @brief Publish a data packet with an already reserved sequence number. + * + * This is intended for non-blocking publishers that reserve seqNo before + * deferring actual publication to the Face/io_context thread. + */ + void + publishDataAtSeq(const Block& content, const ndn::time::milliseconds& freshness, + const NodeID& nid, const SeqNo seq, + uint32_t contentType = ndn::tlv::Content); + + /** + * @brief Insert a data packet with an already reserved sequence number + * without announcing the sequence number. + * + * Pub/Sub uses this to make mapping and piggyback metadata visible before + * the local state vector is advanced and the Sync Interest is sent. + */ + void + insertDataAtSeq(const Block& content, const ndn::time::milliseconds& freshness, + const NodeID& nid, const SeqNo seq, + uint32_t contentType = ndn::tlv::Content); + /** * Insert segment into the store without changing the sequence number. * Intended for inserting segments of a large publication. @@ -113,6 +138,24 @@ class SVSyncBase : noncopyable const Name::Component& finalBlock, uint32_t contentType = ndn::tlv::Content); + /** + * @brief Insert an already-built and signed outer SVS Data packet. + * + * This is used by parallel publishers that construct and sign Data packets on + * worker threads. The caller remains responsible for invoking updateSeqNo() + * on the Face/io_context thread after all related mapping state is ready. + */ + void + insertPreparedData(const Data& data, bool putToFace = false); + + /** + * @brief Put an already-prepared Data packet on the Face without touching the + * store. Used after worker-thread insertion to keep Face operations on the + * Face/io_context thread. + */ + void + putPreparedData(const Data& data); + /** * @brief Retrive a data packet with a particular seqNo from a session * @@ -127,6 +170,12 @@ class SVSyncBase : noncopyable const DataValidatedCallback& onValidated, int nRetries = 0); + void fetchData(const NodeID& nid, + const BootstrapTime& bootstrapTime, + const SeqNo& seq, + const DataValidatedCallback& onValidated, + int nRetries = 0); + /** * @brief Retrive a data packet with a particular seqNo from a session * @@ -146,6 +195,14 @@ class SVSyncBase : noncopyable const TimeoutCallback& onTimeout, int nRetries = 0); + void fetchData(const NodeID& nid, + const BootstrapTime& bootstrapTime, + const SeqNo& seq, + const DataValidatedCallback& onValidated, + const DataValidationErrorCallback& onValidationFailed, + const TimeoutCallback& onTimeout, + int nRetries = 0); + /** @brief Get the underlying data store */ DataStore& getDataStore() { @@ -158,6 +215,16 @@ class SVSyncBase : noncopyable return m_core; } + Name getDataName(const NodeID& nid, const BootstrapTime& bootstrapTime, const SeqNo& seqNo) + { + return makeDataName(nid, bootstrapTime, seqNo); + } + + Name getDataName(const NodeID& nid, const SeqNo& seqNo) + { + return makeDataName(nid, m_core.getBootstrapTime(), seqNo); + } + protected: /** * @brief Return data name for a given packet @@ -167,7 +234,8 @@ class SVSyncBase : noncopyable * data prefix for proper functionality, or the application must * independently produce data under the prefix. */ - virtual Name getDataName(const NodeID& nid, const SeqNo& seqNo) = 0; + virtual Name makeDataName(const NodeID& nid, const BootstrapTime& bootstrapTime, + const SeqNo& seqNo) = 0; public: static inline const NodeID EMPTY_NODE_ID; @@ -204,6 +272,7 @@ class SVSyncBase : noncopyable const UpdateCallback m_onUpdate; std::shared_ptr m_dataStore; + std::mutex m_dataStoreMutex; SVSyncCore m_core; }; diff --git a/ndn-svs/svsync-shared.hpp b/ndn-svs/svsync-shared.hpp index 4a38331..79e76ea 100644 --- a/ndn-svs/svsync-shared.hpp +++ b/ndn-svs/svsync-shared.hpp @@ -19,6 +19,8 @@ #include "svsync-base.hpp" +#include + namespace ndn::svs { /** @@ -48,11 +50,6 @@ class SVSyncShared : public SVSyncBase { } - Name getDataName(const NodeID& nid, const SeqNo& seqNo) override - { - return Name(m_dataPrefix).append(nid).appendNumber(seqNo); - } - /** @brief Set whether data of other nodes is also cached and served */ void setCacheAll(bool val) { @@ -60,6 +57,14 @@ class SVSyncShared : public SVSyncBase } private: + Name makeDataName(const NodeID& nid, const BootstrapTime& bootstrapTime, + const SeqNo& seqNo) override + { + return Name(m_dataPrefix).append(nid) + .append("t=" + std::to_string(bootstrapTime)) + .append("seq=" + std::to_string(seqNo)); + } + bool shouldCache(const Data&) const override { return m_cacheAll; diff --git a/ndn-svs/svsync.hpp b/ndn-svs/svsync.hpp index c48b77a..37d719b 100644 --- a/ndn-svs/svsync.hpp +++ b/ndn-svs/svsync.hpp @@ -19,6 +19,8 @@ #include "svsync-base.hpp" +#include + namespace ndn::svs { /** @@ -47,9 +49,13 @@ class SVSync : public SVSyncBase { } - Name getDataName(const NodeID& nid, const SeqNo& seqNo) override +private: + Name makeDataName(const NodeID& nid, const BootstrapTime& bootstrapTime, + const SeqNo& seqNo) override { - return Name(nid).append(m_syncPrefix).appendNumber(seqNo); + return Name(nid).append(m_syncPrefix) + .append("t=" + std::to_string(bootstrapTime)) + .append("seq=" + std::to_string(seqNo)); } }; diff --git a/ndn-svs/tlv.hpp b/ndn-svs/tlv.hpp index 2794b59..7be080e 100644 --- a/ndn-svs/tlv.hpp +++ b/ndn-svs/tlv.hpp @@ -25,10 +25,12 @@ enum : uint32_t { StateVector = 201, StateVectorEntry = 202, - SeqNo = 204, MappingData = 205, MappingEntry = 206, + SeqNoEntry = 210, LzmaBlock = 211, + BootstrapTime = 212, + SeqNo = 214, }; } // namespace ndn::svs::tlv diff --git a/ndn-svs/version-vector.cpp b/ndn-svs/version-vector.cpp index 141527e..9487bee 100644 --- a/ndn-svs/version-vector.cpp +++ b/ndn-svs/version-vector.cpp @@ -17,8 +17,24 @@ #include "version-vector.hpp" #include "tlv.hpp" +#include +#include + namespace ndn::svs { +namespace { + +bool +isBootstrapTimeTooFarInFuture(BootstrapTime bootstrapTime) +{ + const auto now = static_cast( + std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()).count()); + return bootstrapTime > now + 86400; +} + +} // namespace + VersionVector::VersionVector(const ndn::Block& block) { if (block.type() != tlv::StateVector) @@ -31,10 +47,32 @@ VersionVector::VersionVector(const ndn::Block& block) NDN_THROW(ndn::tlv::Error("StateVectorEntry", it->type())); it->parse(); - NodeID nodeId(it->elements().at(0)); - SeqNo seqNo = ndn::encoding::readNonNegativeInteger(it->elements().at(1)); + const auto& elements = it->elements(); + if (elements.empty() || elements.front().type() != ndn::tlv::Name) { + NDN_THROW(ndn::tlv::Error("Name", elements.empty() ? 0 : elements.front().type())); + } + + NodeID nodeId(elements.front()); + for (auto seqIt = elements.begin() + 1; seqIt != elements.end(); ++seqIt) { + if (seqIt->type() == tlv::SeqNoEntry) { + seqIt->parse(); + if (seqIt->elements().size() < 2 || + seqIt->elements().at(0).type() != tlv::BootstrapTime || + seqIt->elements().at(1).type() != tlv::SeqNo) { + NDN_THROW(ndn::tlv::Error("SeqNoEntry", seqIt->type())); + } + BootstrapTime bootstrapTime = + ndn::encoding::readNonNegativeInteger(seqIt->elements().at(0)); + if (isBootstrapTimeTooFarInFuture(bootstrapTime)) { + NDN_THROW(Error("State vector bootstrap time is too far in the future")); + } + SeqNo seqNo = ndn::encoding::readNonNegativeInteger(seqIt->elements().at(1)); + set(nodeId, bootstrapTime, seqNo); + continue; + } - m_map[nodeId] = seqNo; + NDN_THROW(ndn::tlv::Error("SeqNoEntry", seqIt->type())); + } } } @@ -44,9 +82,18 @@ VersionVector::encode() const ndn::encoding::EncodingBuffer enc; size_t totalLength = 0; - for (auto it = m_map.rbegin(); it != m_map.rend(); it++) { - // SeqNo - size_t entryLength = ndn::encoding::prependNonNegativeIntegerBlock(enc, tlv::SeqNo, it->second); + for (auto it = m_entries.rbegin(); it != m_entries.rend(); it++) { + size_t entryLength = 0; + + for (auto seqIt = it->second.rbegin(); seqIt != it->second.rend(); ++seqIt) { + size_t seqEntryLength = 0; + seqEntryLength += ndn::encoding::prependNonNegativeIntegerBlock(enc, tlv::SeqNo, seqIt->second); + seqEntryLength += ndn::encoding::prependNonNegativeIntegerBlock(enc, tlv::BootstrapTime, + seqIt->first); + entryLength += enc.prependVarNumber(seqEntryLength); + entryLength += enc.prependVarNumber(tlv::SeqNoEntry); + entryLength += seqEntryLength; + } // NodeID (Name) entryLength += ndn::encoding::prependBlock(enc, it->first.wireEncode()); @@ -65,10 +112,30 @@ std::string VersionVector::toStr() const { std::ostringstream stream; - for (const auto& elem : m_map) { - stream << elem.first << ":" << elem.second << " "; + for (const auto& elem : m_entries) { + stream << elem.first << ":"; + for (const auto& seqEntry : elem.second) { + stream << "[" << seqEntry.first << "," << seqEntry.second << "]"; + } + stream << " "; } return stream.str(); } +void +VersionVector::refreshLatest(const NodeID& nid) +{ + auto node = m_entries.find(nid); + if (node == m_entries.end() || node->second.empty()) { + m_latestMap.erase(nid); + return; + } + + SeqNo latestSeqNo = 0; + for (const auto& [bootstrapTime, seqNo] : node->second) { + latestSeqNo = std::max(latestSeqNo, seqNo); + } + m_latestMap[nid] = latestSeqNo; +} + } // namespace ndn::svs diff --git a/ndn-svs/version-vector.hpp b/ndn-svs/version-vector.hpp index c31761e..d440921 100644 --- a/ndn-svs/version-vector.hpp +++ b/ndn-svs/version-vector.hpp @@ -19,7 +19,9 @@ #include "common.hpp" +#include #include +#include namespace ndn::svs { @@ -33,6 +35,7 @@ class VersionVector }; using const_iterator = std::map::const_iterator; + using BootstrapSeqMap = std::map; VersionVector() = default; @@ -46,16 +49,51 @@ class VersionVector std::string toStr() const; SeqNo set(const NodeID& nid, SeqNo seqNo) + = delete; + + SeqNo set(const NodeID& nid, BootstrapTime bootstrapTime, SeqNo seqNo) { - m_map[nid] = seqNo; + m_entries[nid][bootstrapTime] = seqNo; + refreshLatest(nid); m_lastUpdate[nid] = time::system_clock::now(); return seqNo; } SeqNo get(const NodeID& nid) const { - auto elem = m_map.find(nid); - return elem == m_map.end() ? 0 : elem->second; + auto elem = m_latestMap.find(nid); + return elem == m_latestMap.end() ? 0 : elem->second; + } + + SeqNo get(const NodeID& nid, BootstrapTime bootstrapTime) const + { + auto node = m_entries.find(nid); + if (node == m_entries.end()) { + return 0; + } + auto entry = node->second.find(bootstrapTime); + return entry == node->second.end() ? 0 : entry->second; + } + + const BootstrapSeqMap& getEntries(const NodeID& nid) const + { + static const BootstrapSeqMap EMPTY_ENTRIES; + auto elem = m_entries.find(nid); + return elem == m_entries.end() ? EMPTY_ENTRIES : elem->second; + } + + BootstrapTime getBootstrapTime(const NodeID& nid) const + { + auto node = m_entries.find(nid); + if (node == m_entries.end() || node->second.empty()) { + return 0; + } + return node->second.rbegin()->first; + } + + const std::map& getAllEntries() const + { + return m_entries; } time::system_clock::time_point getLastUpdate(const NodeID& nid) const @@ -66,21 +104,25 @@ class VersionVector const_iterator begin() const noexcept { - return m_map.begin(); + return m_latestMap.begin(); } const_iterator end() const noexcept { - return m_map.end(); + return m_latestMap.end(); } bool has(const NodeID& nid) const { - return m_map.find(nid) != end(); + return m_entries.find(nid) != m_entries.end(); } private: - std::map m_map; + void refreshLatest(const NodeID& nid); + +private: + std::map m_entries; + std::map m_latestMap; std::map m_lastUpdate; }; diff --git a/tests/unit-tests/core.t.cpp b/tests/unit-tests/core.t.cpp index 75f2de1..0557aed 100644 --- a/tests/unit-tests/core.t.cpp +++ b/tests/unit-tests/core.t.cpp @@ -15,12 +15,53 @@ */ #include "core.hpp" +#include "tlv.hpp" #include "tests/boost-test.hpp" +#include + +#include + +#include + namespace ndn::tests { using namespace ndn::svs; +using namespace std::chrono_literals; + +static Interest +makeSyncInterest(const Name& syncPrefix, const VersionVector& vv) +{ + ndn::encoding::EncodingBuffer enc; + size_t length = 0; + length += ndn::encoding::prependBlock(enc, vv.encode()); + enc.prependVarNumber(length); + enc.prependVarNumber(ndn::tlv::ApplicationParameters); + + Interest interest(Name(syncPrefix).appendVersion(2)); + interest.setApplicationParameters(enc.block()); + return interest; +} + +static VersionVector +getSyncInterestVector(const Interest& interest) +{ + auto params = interest.getApplicationParameters(); + params.parse(); + return VersionVector(params.get(ndn::svs::tlv::StateVector)); +} + +static void +runIoUntil(Face& face, const std::function& done) +{ + auto deadline = std::chrono::steady_clock::now() + 2s; + while (!done() && std::chrono::steady_clock::now() < deadline) { + face.getIoContext().restart(); + face.getIoContext().run_for(10ms); + std::this_thread::sleep_for(1ms); + } +} class CoreFixture { @@ -32,7 +73,7 @@ class CoreFixture } protected: - Face m_face; + DummyClientFace m_face; Name m_syncPrefix; SVSyncCore m_core; }; @@ -50,8 +91,8 @@ BOOST_AUTO_TEST_CASE(MergeStateVector) BOOST_CHECK_EQUAL(missingInfo.size(), 0); VersionVector v1; - v1.set("one", 1); - v1.set("two", 2); + v1.set("one", 100, 1); + v1.set("two", 200, 2); missingInfo = m_core.mergeStateVector(v1).missingInfo; v = m_core.getState(); @@ -61,9 +102,9 @@ BOOST_AUTO_TEST_CASE(MergeStateVector) BOOST_CHECK_EQUAL(missingInfo.size(), 2); VersionVector v2; - v2.set("one", 1); - v2.set("two", 1); - v2.set("three", 3); + v2.set("one", 100, 1); + v2.set("two", 200, 1); + v2.set("three", 300, 3); missingInfo = m_core.mergeStateVector(v2).missingInfo; v = m_core.getState(); @@ -73,10 +114,181 @@ BOOST_AUTO_TEST_CASE(MergeStateVector) BOOST_REQUIRE_EQUAL(missingInfo.size(), 1); BOOST_CHECK_EQUAL(missingInfo[0].nodeId, "three"); + BOOST_CHECK_EQUAL(missingInfo[0].bootstrapTime, 300); BOOST_CHECK_EQUAL(missingInfo[0].low, 1); BOOST_CHECK_EQUAL(missingInfo[0].high, 3); } +BOOST_AUTO_TEST_CASE(ComputeMergeStateVector) +{ + VersionVector local; + local.set("one", 100, 2); + local.set("local-only", 700, 7); + + VersionVector remote; + remote.set("one", 100, 5); + remote.set("two", 200, 1); + + auto result = SVSyncCore::computeMergeStateVector(local, remote); + + BOOST_CHECK(result.myVectorNew); + BOOST_CHECK(result.otherVectorNew); + BOOST_CHECK_EQUAL(result.mergedVector.get("one"), 5); + BOOST_CHECK_EQUAL(result.mergedVector.get("two"), 1); + BOOST_CHECK_EQUAL(result.mergedVector.get("local-only"), 7); + + BOOST_REQUIRE_EQUAL(result.missingData.size(), 2); + BOOST_CHECK_EQUAL(result.missingData[0].nodeId, "one"); + BOOST_CHECK_EQUAL(result.missingData[0].bootstrapTime, 100); + BOOST_CHECK_EQUAL(result.missingData[0].low, 3); + BOOST_CHECK_EQUAL(result.missingData[0].high, 5); + BOOST_CHECK_EQUAL(result.missingData[1].nodeId, "two"); + BOOST_CHECK_EQUAL(result.missingData[1].bootstrapTime, 200); + BOOST_CHECK_EQUAL(result.missingData[1].low, 1); + BOOST_CHECK_EQUAL(result.missingData[1].high, 1); +} + +BOOST_AUTO_TEST_CASE(ComputeMergeStateVectorKeepsRestartEpochsSeparate) +{ + VersionVector local; + local.set("node", 100, 10); + + VersionVector remote; + remote.set("node", 200, 1); + + auto result = SVSyncCore::computeMergeStateVector(local, remote); + BOOST_CHECK(result.myVectorNew); + BOOST_CHECK(result.otherVectorNew); + BOOST_CHECK_EQUAL(result.mergedVector.get("node", 100), 10); + BOOST_CHECK_EQUAL(result.mergedVector.get("node", 200), 1); + BOOST_REQUIRE_EQUAL(result.missingData.size(), 1); + BOOST_CHECK_EQUAL(result.missingData[0].nodeId, "node"); + BOOST_CHECK_EQUAL(result.missingData[0].bootstrapTime, 200); + BOOST_CHECK_EQUAL(result.missingData[0].low, 1); + BOOST_CHECK_EQUAL(result.missingData[0].high, 1); +} + +BOOST_AUTO_TEST_CASE(ParallelSyncInterestHandling) +{ + m_core.setParallelSyncProcessing(true, 2, 16); + + VersionVector remote; + remote.set("peer", 300, 3); + m_core.onSyncInterestValidated(makeSyncInterest(m_syncPrefix, remote)); + + runIoUntil(m_face, [this] { + auto stats = m_core.getSyncProcessingStats(); + return stats.syncJobsCompleted >= 1 || stats.syncJobsStale >= 1; + }); + + auto stats = m_core.getSyncProcessingStats(); + BOOST_CHECK_EQUAL(stats.syncJobsSubmitted, 1); + BOOST_CHECK_EQUAL(stats.syncJobsDropped, 0); + BOOST_CHECK_EQUAL(m_core.getSeqNo("peer"), 3); +} + +BOOST_AUTO_TEST_CASE(ParallelSyncInterestStressAndStaleFallback) +{ + m_core.setParallelSyncProcessing(true, 2, 32); + + VersionVector staleRemote; + staleRemote.set("stale-peer", 100, 1); + m_core.onSyncInterestValidated(makeSyncInterest(m_syncPrefix, staleRemote)); + m_core.updateSeqNo(1, "local-node"); + + for (uint64_t seq = 1; seq <= 20; ++seq) { + VersionVector remote; + remote.set("peer", 300, seq); + m_core.onSyncInterestValidated(makeSyncInterest(m_syncPrefix, remote)); + } + + runIoUntil(m_face, [this] { + auto stats = m_core.getSyncProcessingStats(); + return stats.syncJobsCompleted + stats.syncJobsStale >= stats.syncJobsSubmitted; + }); + + auto stats = m_core.getSyncProcessingStats(); + BOOST_CHECK_EQUAL(stats.syncJobsDropped, 0); + BOOST_CHECK_GE(stats.syncJobsSubmitted, 21); + BOOST_CHECK_GE(stats.syncJobsStale, 1); + BOOST_CHECK_EQUAL(m_core.getSeqNo("peer"), 20); + BOOST_CHECK_EQUAL(m_core.getSeqNo("local-node"), 1); +} + +BOOST_AUTO_TEST_CASE(PublicationSyncBatchingCoalescesLocalUpdates) +{ + m_core.setSyncInterestBatching(true, 20_ms); + m_core.sendInitialInterest(); + + runIoUntil(m_face, [this] { + return !m_face.sentInterests.empty(); + }); + m_face.sentInterests.clear(); + + for (uint64_t seq = 1; seq <= 10; ++seq) { + m_core.updateSeqNo(seq, "local-node"); + } + + m_face.getIoContext().restart(); + m_face.getIoContext().run_for(std::chrono::milliseconds(5)); + BOOST_CHECK_EQUAL(m_face.sentInterests.size(), 0); + + runIoUntil(m_face, [this] { + return !m_face.sentInterests.empty(); + }); + + BOOST_REQUIRE_EQUAL(m_face.sentInterests.size(), 1); + VersionVector sentVector = getSyncInterestVector(m_face.sentInterests.front()); + BOOST_CHECK_EQUAL(sentVector.get("local-node"), 10); +} + +BOOST_AUTO_TEST_CASE(ParallelSyncProductionHandling) +{ + m_core.setParallelSyncProduction(true, 2, 16); + m_core.sendInitialInterest(); + + runIoUntil(m_face, [this] { + return !m_face.sentInterests.empty(); + }); + m_face.sentInterests.clear(); + + m_core.updateSeqNo(1, "local-node"); + + runIoUntil(m_face, [this] { + auto stats = m_core.getSyncProcessingStats(); + return !m_face.sentInterests.empty() || + stats.syncProductionJobsDropped > 0 || + stats.syncProductionJobsStale > 0; + }); + + auto stats = m_core.getSyncProcessingStats(); + BOOST_CHECK_EQUAL(stats.syncProductionJobsDropped, 0); + BOOST_CHECK_GE(stats.syncProductionJobsSubmitted, 1); + BOOST_CHECK_GE(stats.syncProductionJobsCompleted, 1); + BOOST_REQUIRE_EQUAL(m_face.sentInterests.size(), 1); + + VersionVector sentVector = getSyncInterestVector(m_face.sentInterests.front()); + BOOST_CHECK_EQUAL(sentVector.get("local-node"), 1); +} + +BOOST_AUTO_TEST_CASE(ParallelSyncProductionDropsStaleResult) +{ + m_core.setParallelSyncProduction(true, 1, 16); + m_core.sendInitialInterest(); + m_core.updateSeqNo(1, "local-node"); + + runIoUntil(m_face, [this] { + auto stats = m_core.getSyncProcessingStats(); + return stats.syncProductionJobsSubmitted >= 1 && + stats.syncProductionJobsCompleted + stats.syncProductionJobsStale + + stats.syncProductionJobsDropped >= 1; + }); + + auto stats = m_core.getSyncProcessingStats(); + BOOST_CHECK_EQUAL(stats.syncProductionJobsDropped, 0); + BOOST_CHECK_GE(stats.syncProductionJobsSubmitted, 1); +} + BOOST_AUTO_TEST_SUITE_END() } // namespace ndn::tests diff --git a/tests/unit-tests/security-options.t.cpp b/tests/unit-tests/security-options.t.cpp new file mode 100644 index 0000000..9b2374f --- /dev/null +++ b/tests/unit-tests/security-options.t.cpp @@ -0,0 +1,50 @@ +/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */ +/* + * Copyright (c) 2012-2025 University of California, Los Angeles + * + * This file is part of ndn-svs, synchronization library for distributed realtime + * applications for NDN. + */ + +#include "security-options.hpp" + +#include "tests/boost-test.hpp" + +#include + +namespace ndn::tests { + +using namespace ndn::svs; + +BOOST_AUTO_TEST_SUITE(TestSecurityOptions) + +BOOST_AUTO_TEST_CASE(SignedInterestTimestampIsMonotonic) +{ + KeyChain keyChain("pib-memory:", "tpm-memory:"); + auto identity = keyChain.createIdentity("/ndn/svs/test/signer"); + + KeyChainSigner signer(keyChain); + signer.signingInfo = security::signingByIdentity(identity); + + std::optional previous; + for (size_t i = 0; i < 32; ++i) { + Interest interest(Name("/ndn/svs/test/sync").appendNumber(i)); + interest.setApplicationParameters(makeStringBlock(tlv::ApplicationParameters, "state")); + + signer.sign(interest); + + BOOST_REQUIRE(interest.getSignatureInfo().has_value()); + auto timestamp = interest.getSignatureInfo()->getTime(); + BOOST_REQUIRE(timestamp.has_value()); + + auto timestampMs = time::toUnixTimestamp(*timestamp); + if (previous) { + BOOST_CHECK_GT(timestampMs.count(), previous->count()); + } + previous = timestampMs; + } +} + +BOOST_AUTO_TEST_SUITE_END() + +} // namespace ndn::tests diff --git a/tests/unit-tests/svspubsub.t.cpp b/tests/unit-tests/svspubsub.t.cpp new file mode 100644 index 0000000..2ed4d0d --- /dev/null +++ b/tests/unit-tests/svspubsub.t.cpp @@ -0,0 +1,121 @@ +/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */ +/* + * Copyright (c) 2012-2023 University of California, Los Angeles + * + * This file is part of ndn-svs, synchronization library for distributed realtime + * applications for NDN. + */ + +#include "svspubsub.hpp" + +#include "tests/boost-test.hpp" + +#include + +#include + +#include + +namespace ndn::tests { + +using namespace ndn::svs; +using namespace std::chrono_literals; + +static void +runIoUntil(Face& face, const std::function& done) +{ + auto deadline = std::chrono::steady_clock::now() + 2s; + while (!done() && std::chrono::steady_clock::now() < deadline) { + face.getIoContext().restart(); + face.getIoContext().run_for(10ms); + std::this_thread::sleep_for(1ms); + } +} + +BOOST_AUTO_TEST_SUITE(TestSVSPubSub) + +BOOST_AUTO_TEST_CASE(LatePiggyDataSatisfiesPendingFetch) +{ + DummyClientFace face; + KeyChain keyChain("pib-memory:svspubsub-test", "tpm-memory:svspubsub-test"); + keyChain.createIdentity("/svspubsub-test"); + SVSPubSubOptions opts; + opts.useTimestamp = false; + + SVSPubSub pubsub("/sync", "/local", face, + [] (const std::vector&) {}, + opts); + + size_t callbackCount = 0; + std::string received; + Name producer; + SeqNo receivedSeq = 0; + + pubsub.subscribe("/app", [&] (const SVSPubSub::SubscriptionData& data) { + ++callbackCount; + received.assign(reinterpret_cast(data.data.data()), data.data.size()); + producer = data.producerPrefix; + receivedSeq = data.seqNo; + }); + + BootstrapTime bootstrapTime = 100; + pubsub.insertMapping("/peer", bootstrapTime, 7, "/app/item", {}); + BOOST_CHECK(pubsub.processMapping("/peer", bootstrapTime, 7)); + BOOST_CHECK_EQUAL(callbackCount, 0); + + Data piggyData("/app/item"); + const std::string payload = "piggy"; + piggyData.setContent(make_span(reinterpret_cast(payload.data()), + payload.size())); + keyChain.sign(piggyData); + + MappingList extra("/peer"); + extra.pairs.push_back({bootstrapTime, 7, {Name("/app/item"), {}}}); + Block params = extra.encode(); + params.push_back(piggyData.wireEncode()); + params.encode(); + + VersionVector vv; + pubsub.onRecvExtraData(params, vv); + + BOOST_CHECK_EQUAL(callbackCount, 1); + BOOST_CHECK_EQUAL(received, payload); + BOOST_CHECK_EQUAL(producer, Name("/peer")); + BOOST_CHECK_EQUAL(receivedSeq, 7); + + pubsub.onRecvExtraData(params, vv); + BOOST_CHECK_EQUAL(callbackCount, 1); +} + +BOOST_AUTO_TEST_CASE(AsyncPublishQueuesWithoutImmediateFacePut) +{ + DummyClientFace face; + SVSPubSubOptions opts; + opts.useTimestamp = false; + + SVSPubSub pubsub("/sync", "/local", face, + [] (const std::vector&) {}, + opts); + + const std::string payload1 = "one"; + const std::string payload2 = "two"; + auto seq1 = pubsub.publishAsync("/app/one", + make_span(reinterpret_cast(payload1.data()), + payload1.size())); + auto seq2 = pubsub.publishAsync("/app/two", + make_span(reinterpret_cast(payload2.data()), + payload2.size())); + + BOOST_CHECK_EQUAL(seq1, 1); + BOOST_CHECK_EQUAL(seq2, 2); + BOOST_CHECK_EQUAL(face.sentData.size(), 0); + + runIoUntil(face, [&] { return face.sentData.size() >= 2; }); + + BOOST_CHECK_EQUAL(pubsub.getSVSync().getCore().getSeqNo("/local"), 2); + BOOST_REQUIRE_GE(face.sentData.size(), 2); +} + +BOOST_AUTO_TEST_SUITE_END() + +} // namespace ndn::tests diff --git a/tests/unit-tests/version-vector.t.cpp b/tests/unit-tests/version-vector.t.cpp index 2b8d967..13b2234 100644 --- a/tests/unit-tests/version-vector.t.cpp +++ b/tests/unit-tests/version-vector.t.cpp @@ -28,8 +28,8 @@ class VersionVectorFixture protected: VersionVectorFixture() { - v.set("one", 1); - v.set("two", 2); + v.set("one", 100, 1); + v.set("two", 200, 2); } protected: @@ -47,8 +47,9 @@ BOOST_AUTO_TEST_CASE(Get) BOOST_AUTO_TEST_CASE(Set) { - BOOST_CHECK_EQUAL(v.set("four", 44), 44); + BOOST_CHECK_EQUAL(v.set("four", 400, 44), 44); BOOST_CHECK_EQUAL(v.get("four"), 44); + BOOST_CHECK_EQUAL(v.get("four", 400), 44); } BOOST_AUTO_TEST_CASE(Iterate) @@ -66,31 +67,44 @@ BOOST_AUTO_TEST_CASE(Iterate) BOOST_AUTO_TEST_CASE(EncodeDecode) { ndn::Block block = v.encode(); - BOOST_CHECK_EQUAL(block.value_size(), 24); + BOOST_CHECK_GT(block.value_size(), 24); VersionVector dv(block); BOOST_CHECK_EQUAL(dv.get("one"), 1); BOOST_CHECK_EQUAL(dv.get("two"), 2); + BOOST_CHECK_EQUAL(dv.get("one", 100), 1); + BOOST_CHECK_EQUAL(dv.get("two", 200), 2); } -BOOST_AUTO_TEST_CASE(DecodeStatic) +BOOST_AUTO_TEST_CASE(RejectLegacyDirectSeqNo) { - // Hex: CA0A070508036F6E65CC0101CA0A0705080374776FCC0102 + // Legacy pre-v3 format used StateVectorEntry(Name, SeqNo) without SeqNoEntry. constexpr std::string_view encoded{ "\xCA\x0A\x07\x05\x08\x03\x6F\x6E\x65\xCC\x01\x01" "\xCA\x0A\x07\x05\x08\x03\x74\x77\x6F\xCC\x01\x02" }; - VersionVector dv(ndn::encoding::makeStringBlock(svs::tlv::StateVector, encoded)); - BOOST_CHECK_EQUAL(dv.get("one"), 1); - BOOST_CHECK_EQUAL(dv.get("two"), 2); + BOOST_CHECK_THROW(VersionVector(ndn::encoding::makeStringBlock(svs::tlv::StateVector, encoded)), + ndn::tlv::Error); +} + +BOOST_AUTO_TEST_CASE(MultipleBootstraps) +{ + VersionVector vector; + vector.set("node", 100, 10); + vector.set("node", 200, 1); + + VersionVector decoded(vector.encode()); + BOOST_CHECK_EQUAL(decoded.get("node", 100), 10); + BOOST_CHECK_EQUAL(decoded.get("node", 200), 1); + BOOST_CHECK_EQUAL(decoded.getEntries("node").size(), 2); } BOOST_AUTO_TEST_CASE(Ordering) { VersionVector v1; - v1.set("one", 1); - v1.set("two", 2); + v1.set("one", 100, 1); + v1.set("two", 200, 2); VersionVector v2; - v2.set("two", 2); - v2.set("one", 1); + v2.set("two", 200, 2); + v2.set("one", 100, 1); Block v1e = v1.encode(); Block v2e = v2.encode(); diff --git a/wscript b/wscript index db8a643..0eb5ad0 100644 --- a/wscript +++ b/wscript @@ -76,8 +76,8 @@ def configure(conf): boost_libs.append('iostreams') conf.check_boost(lib=boost_libs, mt=True) - if conf.env.BOOST_VERSION_NUMBER < 107400: - conf.fatal('The minimum supported version of Boost is 1.74.0.\n' + if conf.env.BOOST_VERSION_NUMBER < 107100: + conf.fatal('The minimum supported version of Boost is 1.71.0.\n' 'Please upgrade your distribution or manually install a newer version of Boost.\n' 'For more information, see https://redmine.named-data.net/projects/nfd/wiki/Boost')