From 23657bc8a770734a74f73f6d07075130a366ef00 Mon Sep 17 00:00:00 2001 From: Mauro Sardara Date: Tue, 10 Mar 2020 15:50:18 +0100 Subject: [HICN-544] Do not block reading incoming messages in memif connector. Signed-off-by: Mauro Sardara Change-Id: I844dfa64a977c9c41bfc103bb110c274802b1839 Signed-off-by: Mauro Sardara --- libtransport/src/core/memif_connector.cc | 12 ++++++---- libtransport/src/core/memif_connector.h | 2 +- libtransport/src/core/portal.h | 16 ++++++------- .../src/protocols/byte_stream_reassembly.cc | 5 ++-- utils/src/hiperf.cc | 27 ++++++++++++++++------ 5 files changed, 40 insertions(+), 22 deletions(-) diff --git a/libtransport/src/core/memif_connector.cc b/libtransport/src/core/memif_connector.cc index 179db63e4..49f262ec8 100644 --- a/libtransport/src/core/memif_connector.cc +++ b/libtransport/src/core/memif_connector.cc @@ -305,11 +305,13 @@ void MemifConnector::sendCallback(const std::error_code &ec) { } } -void MemifConnector::processInputBuffer() { +void MemifConnector::processInputBuffer(std::uint16_t total_packets) { Packet::MemBufPtr ptr; - while (input_buffer_.pop(ptr)) { - receive_callback_(std::move(ptr)); + for (; total_packets > 0; total_packets--) { + if (input_buffer_.pop(ptr)) { + receive_callback_(std::move(ptr)); + } } } @@ -339,6 +341,7 @@ int MemifConnector::onInterrupt(memif_conn_handle_t conn, void *private_ctx, memif_connection_t *c = connector->memif_connection_.get(); int err = MEMIF_ERR_SUCCESS, ret_val; + uint16_t total_packets = 0; uint16_t rx; do { @@ -386,11 +389,12 @@ int MemifConnector::onInterrupt(memif_conn_handle_t conn, void *private_ctx, } c->rx_buf_num -= rx; + total_packets += rx; } while (ret_val == MEMIF_ERR_NOBUF); connector->io_service_.post( - std::bind(&MemifConnector::processInputBuffer, connector)); + std::bind(&MemifConnector::processInputBuffer, connector, total_packets)); return 0; diff --git a/libtransport/src/core/memif_connector.h b/libtransport/src/core/memif_connector.h index aafef1e56..693efd14c 100644 --- a/libtransport/src/core/memif_connector.h +++ b/libtransport/src/core/memif_connector.h @@ -96,7 +96,7 @@ class MemifConnector : public Connector { void sendCallback(const std::error_code &ec); - void processInputBuffer(); + void processInputBuffer(std::uint16_t total_packets); private: static utils::EpollEventReactor main_event_reactor_; diff --git a/libtransport/src/core/portal.h b/libtransport/src/core/portal.h index 34dfbd826..05715543a 100644 --- a/libtransport/src/core/portal.h +++ b/libtransport/src/core/portal.h @@ -24,6 +24,7 @@ #include #include #include +#include #include #include @@ -52,21 +53,20 @@ class HandlerMemory { static constexpr std::size_t memory_size = 1024 * 1024; public: - HandlerMemory() : index_(0) {} + HandlerMemory() {} HandlerMemory(const HandlerMemory &) = delete; HandlerMemory &operator=(const HandlerMemory &) = delete; TRANSPORT_ALWAYS_INLINE void *allocate(std::size_t size) { - return &storage_[index_++ % memory_size]; + return utils::FixedBlockAllocator<128, 4096>::getInstance() + ->allocateBlock(); } - TRANSPORT_ALWAYS_INLINE void deallocate(void *pointer) {} - - private: - // Storage space used for handler-based custom memory allocation. - typename std::aligned_storage<128>::type storage_[memory_size]; - uint32_t index_; + TRANSPORT_ALWAYS_INLINE void deallocate(void *pointer) { + utils::FixedBlockAllocator<128, 4096>::getInstance()->deallocateBlock( + pointer); + } #else public: HandlerMemory() {} diff --git a/libtransport/src/protocols/byte_stream_reassembly.cc b/libtransport/src/protocols/byte_stream_reassembly.cc index e15498bb1..12631637e 100644 --- a/libtransport/src/protocols/byte_stream_reassembly.cc +++ b/libtransport/src/protocols/byte_stream_reassembly.cc @@ -39,7 +39,7 @@ ByteStreamReassembly::ByteStreamReassembly( void ByteStreamReassembly::reassemble( std::unique_ptr &&manifest) { - if (TRANSPORT_EXPECT_TRUE(manifest != nullptr)) { + if (TRANSPORT_EXPECT_TRUE(manifest != nullptr) && read_buffer_->capacity()) { received_packets_.emplace( std::make_pair(manifest->getName().getSuffix(), nullptr)); assembleContent(); @@ -47,7 +47,8 @@ void ByteStreamReassembly::reassemble( } void ByteStreamReassembly::reassemble(ContentObject::Ptr &&content_object) { - if (TRANSPORT_EXPECT_TRUE(content_object != nullptr)) { + if (TRANSPORT_EXPECT_TRUE(content_object != nullptr) && + read_buffer_->capacity()) { received_packets_.emplace(std::make_pair( content_object->getName().getSuffix(), std::move(content_object))); assembleContent(); diff --git a/utils/src/hiperf.cc b/utils/src/hiperf.cc index 2b78a02b9..eabd12c86 100644 --- a/utils/src/hiperf.cc +++ b/utils/src/hiperf.cc @@ -72,6 +72,7 @@ struct ClientConfiguration { producer_certificate(""), passphrase(""), receive_buffer(nullptr), + receive_buffer_size_(128 * 1024), download_size(0), report_interval_milliseconds_(1000), transport_protocol_(CBR), @@ -92,6 +93,7 @@ struct ClientConfiguration { std::string producer_certificate; std::string passphrase; std::shared_ptr receive_buffer; + std::size_t receive_buffer_size_; std::size_t download_size; std::uint32_t report_interval_milliseconds_; TransportProtocolAlgorithms transport_protocol_; @@ -562,11 +564,10 @@ class HIperfClient { }; class Callback : public ConsumerSocket::ReadCallback { - static constexpr std::size_t read_size = 128 * 1024; - public: Callback(HIperfClient &hiperf_client) : client_(hiperf_client) { - client_.configuration_.receive_buffer = utils::MemBuf::create(read_size); + client_.configuration_.receive_buffer = + utils::MemBuf::create(client_.configuration_.receive_buffer_size_); } bool isBufferMovable() noexcept override { return false; } @@ -575,7 +576,7 @@ class HIperfClient { size_t *max_length) override { *application_buffer = client_.configuration_.receive_buffer->writableData(); - *max_length = read_size; + *max_length = client_.configuration_.receive_buffer_size_; } void readDataAvailable(std::size_t length) noexcept override {} @@ -583,7 +584,9 @@ class HIperfClient { void readBufferAvailable( std::unique_ptr &&buffer) noexcept override {} - size_t maxBufferSize() const override { return read_size; } + size_t maxBufferSize() const override { + return client_.configuration_.receive_buffer_size_; + } void readError(const std::error_code ec) noexcept override { std::cerr << "Error " << ec.message() << " while reading from socket" @@ -740,6 +743,7 @@ class HIperfServer { } void virtualProcessInterest(ProducerSocket &p, const Interest &interest) { + // std::cout << "Received interest " << interest.getName() << std::endl; content_objects_[content_objects_index_ & mask_]->setName( interest.getName()); producer_socket_->produce( @@ -1147,6 +1151,10 @@ void usage() { << std::endl; std::cerr << "-L\t\t\t" << "Set interest lifetime." << std::endl; + std::cerr << "-M\t\t\t" + << "Size of consumer input buffer. If 0, reassembly of packets " + "will be disabled." + << std::endl; std::cerr << "-W\t\t\t\t" << "Use a fixed congestion window " "for retrieving the data." @@ -1200,7 +1208,7 @@ int main(int argc, char *argv[]) { int opt; #ifndef _WIN32 while ((opt = getopt(argc, argv, - "DSCf:b:d:W:RMc:vA:s:rmlK:k:y:p:hi:xE:P:B:ItL:")) != + "DSCf:b:d:W:RM:c:vA:s:rmlK:k:y:p:hi:xE:P:B:ItL:")) != -1) { switch (opt) { // Common @@ -1214,7 +1222,7 @@ int main(int argc, char *argv[]) { } #else while ((opt = getopt(argc, argv, - "SCf:b:d:W:RMc:vA:s:rmlK:k:y:p:hi:xB:E:P:tL:")) != -1) { + "SCf:b:d:W:RM:c:vA:s:rmlK:k:y:p:hi:xB:E:P:tL:")) != -1) { switch (opt) { #endif case 'f': { @@ -1260,6 +1268,11 @@ int main(int argc, char *argv[]) { options = 1; break; } + case 'M': { + client_configuration.receive_buffer_size_ = std::stoull(optarg); + options = 1; + break; + } #ifdef SECURE_HICNTRANSPORT case 'P': { client_configuration.producer_prefix_ = Prefix(optarg); -- cgit 1.2.3-korg