SunnyQjm/PSync
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Use ndn-cxx's Segmenter in SegmentPublisher

Browse Source 
This commit also makes lld the preferred linker on Linux

Change-Id: I3a66f2e79a5378d005046c61a8ec2d49345508d6
This commit is contained in:
Davide Pesavento
2022-11-17 19:07:04 -05:00
parent 03426efa2c
commit b68f28497e
9 changed files with 62 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci.yml
+2
View File
@@ -52,6 +52,8 @@ jobs:
include:
- os: macos-12
xcode: '13.4'
- os: macos-12
xcode: '14.1'
steps:
- name: Set up Xcode
uses: maxim-lobanov/setup-xcode@v1
.github/workflows/docs.yml
+3 -1
View File
@@ -15,7 +15,7 @@ jobs:
strategy:
fail-fast: false
matrix:
os: [macos-12, ubuntu-20.04]
os: [macos-latest, ubuntu-latest]
env:
JOB_NAME: Docs
steps:
@@ -27,5 +27,7 @@ jobs:
./.jenkins
- name: Build documentation
run: |
pybindir=$(python3 -c 'import sysconfig; print(sysconfig.get_path("scripts", "posix_user"))')
export PATH="${pybindir}${PATH:+:}${PATH}"
./waf --color=yes configure
./waf --color=yes docs
.jenkins
+1
View File
@@ -16,6 +16,7 @@ case $(uname) in
# Emulate a subset of os-release(5)
export ID=macos
export VERSION_ID=$(sw_vers -productVersion)
export PATH="/usr/local/bin${PATH:+:}${PATH}"
if [[ -x /opt/homebrew/bin/brew ]]; then
eval "$(/opt/homebrew/bin/brew shellenv)"
elif [[ -x /usr/local/bin/brew ]]; then
.jenkins.d/00-deps.sh
+5 -11
View File
@@ -25,20 +25,14 @@ if [[ $ID == macos ]]; then
fi
brew update
brew install --formula "${FORMULAE[@]}"
if (( ${#PIP_PKGS[@]} )); then
pip3 install --upgrade --upgrade-strategy=eager "${PIP_PKGS[@]}"
fi
elif [[ $ID_LIKE == *debian* ]]; then
sudo apt-get -qq update
sudo apt-get -qy install "${APT_PKGS[@]}"
if (( ${#PIP_PKGS[@]} )); then
pip3 install --user --upgrade --upgrade-strategy=eager "${PIP_PKGS[@]}"
fi
elif [[ $ID_LIKE == *fedora* ]]; then
sudo dnf -y install gcc-c++ libasan pkgconf-pkg-config python3 \
sudo dnf -y install gcc-c++ libasan lld pkgconf-pkg-config python3 \
boost-devel openssl-devel sqlite-devel
fi
if (( ${#PIP_PKGS[@]} )); then
pip3 install --user --upgrade --upgrade-strategy=eager "${PIP_PKGS[@]}"
fi
.waf-tools/default-compiler-flags.py
+1 -3
View File
@@ -137,9 +137,7 @@ class GccBasicFlags(CompilerFlags):
def getGeneralFlags(self, conf):
flags = super(GccBasicFlags, self).getGeneralFlags(conf)
flags['CXXFLAGS'] += ['-std=c++17']
if Utils.unversioned_sys_platform() == 'linux':
flags['LINKFLAGS'] += ['-fuse-ld=gold']
elif Utils.unversioned_sys_platform() == 'freebsd':
if Utils.unversioned_sys_platform() != 'darwin':
flags['LINKFLAGS'] += ['-fuse-ld=lld']
return flags
PSync/full-producer.cpp
+2 -2
View File
@@ -266,14 +266,14 @@ FullProducer::sendSyncData(const ndn::Name& name, const ndn::Block& block)
NDN_LOG_DEBUG("Sending sync Data");
// Send data after removing pending sync interest on face
m_segmentPublisher.publish(name, dataName, content, m_syncReplyFreshness);
m_segmentPublisher.publish(name, dataName, *content, m_syncReplyFreshness);
NDN_LOG_TRACE("Renewing sync interest");
sendSyncInterest();
}
else {
NDN_LOG_DEBUG("Sending sync Data");
m_segmentPublisher.publish(name, dataName, content, m_syncReplyFreshness);
m_segmentPublisher.publish(name, dataName, *content, m_syncReplyFreshness);
}
}
PSync/segment-publisher.cpp
+15 -57
View File
@@ -19,79 +19,37 @@
#include "PSync/segment-publisher.hpp"
#include <ndn-cxx/name-component.hpp>
namespace psync {
SegmentPublisher::SegmentPublisher(ndn::Face& face, ndn::KeyChain& keyChain, size_t imsLimit)
SegmentPublisher::SegmentPublisher(ndn::Face& face, ndn::KeyChain& keyChain,
const ndn::security::SigningInfo& signingInfo, size_t imsLimit)
: m_face(face)
, m_scheduler(m_face.getIoService())
, m_keyChain(keyChain)
, m_segmenter(keyChain, signingInfo)
, m_ims(imsLimit)
{
}
void
SegmentPublisher::publish(const ndn::Name& interestName, const ndn::Name& dataName,
const ndn::Block& block, ndn::time::milliseconds freshness,
const ndn::security::SigningInfo& signingInfo)
ndn::span<const uint8_t> buffer, ndn::time::milliseconds freshness)
{
auto buf = std::make_shared<const ndn::Buffer>(block.begin(), block.end());
publish(interestName, dataName, buf, freshness, signingInfo);
}
auto segments = m_segmenter.segment(buffer, ndn::Name(dataName).appendVersion(),
ndn::MAX_NDN_PACKET_SIZE >> 1, freshness);
for (const auto& data : segments) {
m_ims.insert(*data, freshness);
m_scheduler.schedule(freshness, [this, name = data->getName()] { m_ims.erase(name); });
}
void
SegmentPublisher::publish(const ndn::Name& interestName, const ndn::Name& dataName,
const ndn::ConstBufferPtr& buffer,
ndn::time::milliseconds freshness,
const ndn::security::SigningInfo& signingInfo)
{
// Put on face only the segment which has a pending interest,
// otherwise the segment is unsolicited
uint64_t interestSegment = 0;
if (interestName[-1].isSegment()) {
interestSegment = interestName[-1].toSegment();
}
const uint8_t* rawBuffer = buffer->data();
const uint8_t* segmentBegin = rawBuffer;
const uint8_t* end = rawBuffer + buffer->size();
const size_t maxPacketSize = ndn::MAX_NDN_PACKET_SIZE >> 1;
uint64_t totalSegments = buffer->size() / maxPacketSize;
ndn::Name segmentPrefix(dataName);
segmentPrefix.appendVersion();
uint64_t segmentNo = 0;
do {
const uint8_t* segmentEnd = segmentBegin + maxPacketSize;
if (segmentEnd > end) {
segmentEnd = end;
}
ndn::Name segmentName(segmentPrefix);
segmentName.appendSegment(segmentNo);
// We get a std::exception: bad_weak_ptr from m_ims if we don't use shared_ptr for data
auto data = std::make_shared<ndn::Data>(segmentName);
data->setContent(ndn::span<const uint8_t>(segmentBegin, segmentEnd));
data->setFreshnessPeriod(freshness);
data->setFinalBlock(ndn::name::Component::fromSegment(totalSegments));
m_keyChain.sign(*data, signingInfo);
segmentBegin = segmentEnd;
// Put on face only the segment which has a pending interest
// otherwise the segment is unsolicited
if (interestSegment == segmentNo) {
m_face.put(*data);
}
m_ims.insert(*data, freshness);
m_scheduler.schedule(freshness, [this, segmentName] { m_ims.erase(segmentName); });
++segmentNo;
} while (segmentBegin < end);
if (interestSegment < segments.size()) {
m_face.put(*segments[interestSegment]);
}
}
bool
PSync/segment-publisher.hpp
+8 -25
View File
@@ -1,6 +1,6 @@
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/*
* Copyright (c) 2014-2020, The University of Memphis
* Copyright (c) 2014-2022, The University of Memphis
*
* This file is part of PSync.
* See AUTHORS.md for complete list of PSync authors and contributors.
@@ -23,55 +23,38 @@
#include "PSync/detail/access-specifiers.hpp"
#include <ndn-cxx/face.hpp>
#include <ndn-cxx/name.hpp>
#include <ndn-cxx/ims/in-memory-storage-fifo.hpp>
#include <ndn-cxx/security/key-chain.hpp>
#include <ndn-cxx/util/scheduler.hpp>
#include <ndn-cxx/util/time.hpp>
#include <ndn-cxx/util/segmenter.hpp>
namespace psync {
/**
* @brief Segment Publisher to publish segmented data
* @brief Helper class to publish segmented data.
*/
class SegmentPublisher
{
public:
SegmentPublisher(ndn::Face& face, ndn::KeyChain& keyChain,
const ndn::security::SigningInfo& signingInfo = ndn::security::SigningInfo(),
size_t imsLimit = 100);
/**
* @brief Put all the segments in memory.
*
* @param interestName the interest name, to determine the sequence to be answered immediately
* @param dataName the data name, has components after interest name
* @param block the content of the data
* @param freshness freshness of the segments
* @param signingInfo signing info to sign the data with
*/
void
publish(const ndn::Name& interestName, const ndn::Name& dataName,
const ndn::Block& block, ndn::time::milliseconds freshness,
const ndn::security::SigningInfo& signingInfo = ndn::security::SigningInfo());
/**
* @brief Put all the segments in memory.
*
* @param interestName the interest name, to determine the sequence to be answered immediately
* @param dataName the data name, has components after interest name
* @param buffer the content of the data
* @param freshness freshness of the segments
* @param signingInfo signing info to sign the data with
* @param freshness freshness period of the segments
*/
void
publish(const ndn::Name& interestName, const ndn::Name& dataName,
const ndn::ConstBufferPtr& buffer, ndn::time::milliseconds freshness,
const ndn::security::SigningInfo& signingInfo = ndn::security::SigningInfo());
ndn::span<const uint8_t> buffer, ndn::time::milliseconds freshness);
/**
* @brief Try to reply from memory, return false if we cannot find the segment.
*
* The caller is then expected to use publish if this returns false.
* The caller is then expected to use publish() if this returns false.
*/
bool
replyFromStore(const ndn::Name& interestName);
@@ -79,7 +62,7 @@ public:
private:
ndn::Face& m_face;
ndn::Scheduler m_scheduler;
ndn::KeyChain& m_keyChain;
ndn::util::Segmenter m_segmenter;
PSYNC_PUBLIC_WITH_TESTS_ELSE_PRIVATE:
ndn::InMemoryStorageFifo m_ims;
tests/test-segment-publisher.cpp
+25 -8
View File
@@ -38,7 +38,7 @@ protected:
SegmentPublisherFixture()
{
m_face.setInterestFilter(InterestFilter("/hello/world"),
bind(&SegmentPublisherFixture::onInterest, this, _1, _2),
bind(&SegmentPublisherFixture::onInterest, this, _2),
[] (auto&&...) { BOOST_CHECK(false); });
advanceClocks(10_ms);
@@ -63,7 +63,7 @@ protected:
}
void
onInterest(const Name& prefix, const Interest& interest)
onInterest(const Interest& interest)
{
if (publisher.replyFromStore(interest.getName())) {
numRepliesFromStore++;
@@ -82,7 +82,7 @@ protected:
protected:
util::DummyClientFace m_face{m_io, m_keyChain, {true, true}};
SegmentPublisher publisher{m_face, m_keyChain};
shared_ptr<util::SegmentFetcher> fetcher;
std::shared_ptr<util::SegmentFetcher> fetcher;
Name dataName;
detail::State state;
@@ -97,6 +97,7 @@ BOOST_FIXTURE_TEST_SUITE(TestSegmentPublisher, SegmentPublisherFixture)
BOOST_AUTO_TEST_CASE(Basic)
{
BOOST_CHECK_EQUAL(publisher.m_ims.size(), 0);
expressInterest(Interest("/hello/world"));
BOOST_CHECK_EQUAL(numComplete, 1);
// First segment is answered directly in publish,
@@ -104,12 +105,20 @@ BOOST_AUTO_TEST_CASE(Basic)
BOOST_CHECK_EQUAL(numRepliesFromStore, 2);
BOOST_CHECK_EQUAL(publisher.m_ims.size(), 3);
for (const auto& data : publisher.m_ims) {
BOOST_TEST_CONTEXT(data.getName()) {
BOOST_REQUIRE_EQUAL(data.getName().size(), 4);
BOOST_CHECK(data.getName()[-1].isSegment());
BOOST_CHECK(data.getName()[-2].isVersion());
}
}
numRepliesFromStore = 0;
expressInterest(Interest("/hello/world"));
BOOST_CHECK_EQUAL(numComplete, 2);
BOOST_CHECK_EQUAL(numRepliesFromStore, 3);
advanceClocks(time::milliseconds(freshness));
advanceClocks(freshness);
BOOST_CHECK_EQUAL(publisher.m_ims.size(), 0);
numRepliesFromStore = 0;
@@ -118,7 +127,7 @@ BOOST_AUTO_TEST_CASE(Basic)
BOOST_CHECK_EQUAL(numRepliesFromStore, 2);
numRepliesFromStore = 0;
m_face.expressInterest(Interest("/hello/world/").setCanBePrefix(true),
m_face.expressInterest(Interest("/hello/world").setCanBePrefix(true),
[this] (auto&&...) { this->numComplete++; },
[] (auto&&...) { BOOST_CHECK(false); },
[] (auto&&...) { BOOST_CHECK(false); });
@@ -127,10 +136,10 @@ BOOST_AUTO_TEST_CASE(Basic)
BOOST_CHECK_EQUAL(numRepliesFromStore, 1);
}
BOOST_AUTO_TEST_CASE(LargerDataName)
BOOST_AUTO_TEST_CASE(LongerDataName)
{
BOOST_CHECK_EQUAL(publisher.m_ims.size(), 0);
dataName = Name("/hello/world/IBF");
BOOST_CHECK_EQUAL(publisher.m_ims.size(), 0);
expressInterest(Interest("/hello/world"));
BOOST_CHECK_EQUAL(numComplete, 1);
@@ -139,7 +148,15 @@ BOOST_AUTO_TEST_CASE(LargerDataName)
BOOST_CHECK_EQUAL(numRepliesFromStore, 2);
BOOST_CHECK_EQUAL(publisher.m_ims.size(), 3);
advanceClocks(time::milliseconds(freshness));
for (const auto& data : publisher.m_ims) {
BOOST_TEST_CONTEXT(data.getName()) {
BOOST_REQUIRE_EQUAL(data.getName().size(), 5);
BOOST_CHECK(data.getName()[-1].isSegment());
BOOST_CHECK(data.getName()[-2].isVersion());
}
}
advanceClocks(freshness);
BOOST_CHECK_EQUAL(publisher.m_ims.size(), 0);
}