diff options
Diffstat (limited to 'libtransport/src')
21 files changed, 683 insertions, 1760 deletions
diff --git a/libtransport/src/hicn/transport/http/CMakeLists.txt b/libtransport/src/hicn/transport/http/CMakeLists.txt index ddcf1fdc3..b24c80195 100644 --- a/libtransport/src/hicn/transport/http/CMakeLists.txt +++ b/libtransport/src/hicn/transport/http/CMakeLists.txt @@ -29,7 +29,6 @@ list(APPEND HEADER_FILES ${CMAKE_CURRENT_SOURCE_DIR}/facade.h ${CMAKE_CURRENT_SOURCE_DIR}/response.h ${CMAKE_CURRENT_SOURCE_DIR}/message.h - ${CMAKE_CURRENT_SOURCE_DIR}/callbacks.h ) set(SOURCE_FILES ${SOURCE_FILES} PARENT_SCOPE) diff --git a/libtransport/src/hicn/transport/http/callbacks.h b/libtransport/src/hicn/transport/http/callbacks.h deleted file mode 100644 index a0f3d5999..000000000 --- a/libtransport/src/hicn/transport/http/callbacks.h +++ /dev/null @@ -1,46 +0,0 @@ -/* - * Copyright (c) 2017-2019 Cisco and/or its affiliates. - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at: - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#pragma once - -#include <hicn/transport/http/server_publisher.h> - -#include <functional> -#include <memory> - -namespace transport { - -namespace http { - -enum class RC : uint8_t { - SUCCESS, - CONTENT_PUBLISHED, - ERR_UNDEFINED, -}; - -using OnHttpRequest = - std::function<void(std::shared_ptr<HTTPServerPublisher>&, const uint8_t*, - std::size_t, int request_id)>; -using DeadlineTimerCallback = std::function<void(const std::error_code& e)>; -using ReceiveCallback = std::function<void(const std::vector<uint8_t>&)>; -using OnPayloadCallback = - std::function<RC(const std::error_code& ec, const core::Name& name, - const ContentBuffer& payload)>; -using ContentSentCallback = - std::function<void(const std::error_code&, const core::Name&)>; - -} // namespace http - -} // namespace transport
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/http/client_connection.cc b/libtransport/src/hicn/transport/http/client_connection.cc index b31d89b6b..fb7dbdfac 100644 --- a/libtransport/src/hicn/transport/http/client_connection.cc +++ b/libtransport/src/hicn/transport/http/client_connection.cc @@ -27,6 +27,8 @@ using namespace transport; HTTPClientConnection::HTTPClientConnection() : consumer_(TransportProtocolAlgorithms::RAAQM, io_service_), + read_bytes_callback_(nullptr), + read_buffer_(nullptr), response_(std::make_shared<HTTPResponse>()), timer_(nullptr) { consumer_.setSocketOption( @@ -35,11 +37,7 @@ HTTPClientConnection::HTTPClientConnection() &HTTPClientConnection::verifyData, this, std::placeholders::_1, std::placeholders::_2)); - consumer_.setSocketOption( - ConsumerCallbacksOptions::CONTENT_RETRIEVED, - (ConsumerContentCallback)std::bind( - &HTTPClientConnection::processPayload, this, std::placeholders::_1, - std::placeholders::_2, std::placeholders::_3)); + consumer_.setSocketOption(ConsumerCallbacksOptions::READ_CALLBACK, this); consumer_.connect(); std::shared_ptr<typename ConsumerSocket::Portal> portal; @@ -47,36 +45,48 @@ HTTPClientConnection::HTTPClientConnection() timer_ = std::make_unique<asio::steady_timer>(portal->getIoService()); } -HTTPClientConnection &HTTPClientConnection::get( +HTTPClientConnection::RC HTTPClientConnection::get( const std::string &url, HTTPHeaders headers, HTTPPayload payload, - std::shared_ptr<HTTPResponse> response) { - return sendRequest(url, HTTPMethod::GET, headers, payload, response); + std::shared_ptr<HTTPResponse> response, ReadBytesCallback *callback) { + return sendRequest(url, HTTPMethod::GET, headers, payload, response, + callback); } -HTTPClientConnection &HTTPClientConnection::sendRequest( +HTTPClientConnection::RC HTTPClientConnection::sendRequest( const std::string &url, HTTPMethod method, HTTPHeaders headers, - HTTPPayload payload, std::shared_ptr<HTTPResponse> response) { + HTTPPayload payload, std::shared_ptr<HTTPResponse> response, + ReadBytesCallback *callback) { + current_url_ = url; + read_bytes_callback_ = callback; if (!response) { response = response_; } auto start = std::chrono::steady_clock::now(); HTTPRequest request(method, url, headers, payload); - std::string name = sendRequestGetReply(request, response); - auto end = std::chrono::steady_clock::now(); - - TRANSPORT_LOGI( - "%s %s [%s] duration: %llu [usec] %zu [bytes]\n", - method_map[method].c_str(), url.c_str(), name.c_str(), - (unsigned long long)std::chrono::duration_cast<std::chrono::microseconds>( - end - start) - .count(), - response->size()); + response->clear(); - return *this; + success_callback_ = [this, method = std::move(method), url = std::move(url), + start = std::move(start), + response = std::move(response)]( + std::size_t size) -> std::shared_ptr<HTTPResponse> { + auto end = std::chrono::steady_clock::now(); + TRANSPORT_LOGI( + "%s %s [%s] duration: %llu [usec] %zu [bytes]\n", + method_map[method].c_str(), url.c_str(), name_.str().c_str(), + (unsigned long long) + std::chrono::duration_cast<std::chrono::microseconds>(end - start) + .count(), + size); + + return response; + }; + + sendRequestGetReply(request, response); + return return_code_; } -std::string HTTPClientConnection::sendRequestGetReply( +void HTTPClientConnection::sendRequestGetReply( const HTTPRequest &request, std::shared_ptr<HTTPResponse> &response) { const std::string &request_string = request.getRequestString(); const std::string &locator = request.getLocator(); @@ -94,38 +104,28 @@ std::string HTTPClientConnection::sendRequestGetReply( &HTTPClientConnection::processLeavingInterest, this, std::placeholders::_1, std::placeholders::_2, request_string)); - // Send content to producer piggybacking it through first interest (to fix) - - response->clear(); - // Factor hicn name using hash + name_.str(""); - std::stringstream stream; - - stream << std::hex << http::default_values::ipv6_first_word << ":"; + name_ << std::hex << http::default_values::ipv6_first_word << ":"; for (uint16_t *word = (uint16_t *)&locator_hash; std::size_t(word) < (std::size_t(&locator_hash) + sizeof(locator_hash)); word++) { - stream << ":" << std::hex << *word; + name_ << ":" << std::hex << *word; } for (uint16_t *word = (uint16_t *)&request_hash; std::size_t(word) < (std::size_t(&request_hash) + sizeof(request_hash)); word++) { - stream << ":" << std::hex << *word; + name_ << ":" << std::hex << *word; } - stream << "|0"; - - ContentBuffer response_ptr = - std::static_pointer_cast<std::vector<uint8_t>>(response); + name_ << "|0"; - consumer_.consume(Name(stream.str()), response_ptr); + consumer_.consume(Name(name_.str())); consumer_.stop(); - - return stream.str(); } HTTPResponse HTTPClientConnection::response() { @@ -133,14 +133,6 @@ HTTPResponse HTTPClientConnection::response() { return std::move(*response_); } -void HTTPClientConnection::processPayload(ConsumerSocket &c, - std::size_t bytes_transferred, - const std::error_code &ec) { - if (ec) { - TRANSPORT_LOGE("Download failed!!"); - } -} - bool HTTPClientConnection::verifyData( ConsumerSocket &c, const core::ContentObject &contentObject) { if (contentObject.getPayloadType() == PayloadType::CONTENT_OBJECT) { @@ -192,6 +184,49 @@ HTTPClientConnection &HTTPClientConnection::setCertificate( return *this; } +// Read buffer management +void HTTPClientConnection::readBufferAvailable( + std::unique_ptr<utils::MemBuf> &&buffer) noexcept { + if (!read_bytes_callback_) { + if (!read_buffer_) { + read_buffer_ = std::move(buffer); + } else { + read_buffer_->prependChain(std::move(buffer)); + } + } else { + read_bytes_callback_->onBytesReceived(std::move(buffer)); + } +} + +// Read buffer management +void HTTPClientConnection::readError(const std::error_code ec) noexcept { + TRANSPORT_LOGE("Error %s during download of %s", ec.message().c_str(), + current_url_.c_str()); + if (read_bytes_callback_) { + read_bytes_callback_->onError(ec); + } + + return_code_ = HTTPClientConnection::RC::DOWNLOAD_FAILED; +} + +void HTTPClientConnection::readSuccess(std::size_t total_size) noexcept { + auto response = success_callback_(total_size); + if (read_bytes_callback_) { + read_bytes_callback_->onSuccess(total_size); + } else { + response->reserve(total_size); + const utils::MemBuf *head = read_buffer_.get(), *current = head; + do { + response->insert(response->end(), current->data(), current->tail()); + current = current->next(); + } while (current != head); + + read_buffer_.reset(); + } + + return_code_ = HTTPClientConnection::RC::DOWNLOAD_SUCCESS; +} + } // namespace http } // namespace transport diff --git a/libtransport/src/hicn/transport/http/client_connection.h b/libtransport/src/hicn/transport/http/client_connection.h index faad1864c..6c150f848 100644 --- a/libtransport/src/hicn/transport/http/client_connection.h +++ b/libtransport/src/hicn/transport/http/client_connection.h @@ -31,18 +31,30 @@ namespace http { using namespace interface; using namespace core; -class HTTPClientConnection { +class HTTPClientConnection : public ConsumerSocket::ReadCallback { + static constexpr uint32_t max_buffer_capacity = 64 * 1024; + public: + class ReadBytesCallback { + public: + virtual void onBytesReceived(std::unique_ptr<utils::MemBuf> &&buffer) = 0; + virtual void onSuccess(std::size_t bytes) = 0; + virtual void onError(const std::error_code ec) = 0; + }; + + enum class RC : uint32_t { DOWNLOAD_FAILED, DOWNLOAD_SUCCESS }; + HTTPClientConnection(); - HTTPClientConnection &get(const std::string &url, HTTPHeaders headers = {}, - HTTPPayload payload = {}, - std::shared_ptr<HTTPResponse> response = nullptr); + RC get(const std::string &url, HTTPHeaders headers = {}, + HTTPPayload payload = {}, + std::shared_ptr<HTTPResponse> response = nullptr, + ReadBytesCallback *callback = nullptr); - HTTPClientConnection &sendRequest( - const std::string &url, HTTPMethod method, HTTPHeaders headers = {}, - HTTPPayload payload = {}, - std::shared_ptr<HTTPResponse> response = nullptr); + RC sendRequest(const std::string &url, HTTPMethod method, + HTTPHeaders headers = {}, HTTPPayload payload = {}, + std::shared_ptr<HTTPResponse> response = nullptr, + ReadBytesCallback *callback = nullptr); HTTPResponse response(); @@ -55,11 +67,8 @@ class HTTPClientConnection { HTTPClientConnection &setCertificate(const std::string &cert_path); private: - void processPayload(interface::ConsumerSocket &c, - std::size_t bytes_transferred, const std::error_code &ec); - - std::string sendRequestGetReply(const HTTPRequest &request, - std::shared_ptr<HTTPResponse> &response); + void sendRequestGetReply(const HTTPRequest &request, + std::shared_ptr<HTTPResponse> &response); bool verifyData(interface::ConsumerSocket &c, const core::ContentObject &contentObject); @@ -68,12 +77,42 @@ class HTTPClientConnection { const core::Interest &interest, std::string &payload); + // Read callback + bool isBufferMovable() noexcept override { return true; } + void getReadBuffer(uint8_t **application_buffer, + size_t *max_length) override {} + void readDataAvailable(size_t length) noexcept override {} + size_t maxBufferSize() const override { return max_buffer_capacity; } + void readBufferAvailable( + std::unique_ptr<utils::MemBuf> &&buffer) noexcept override; + void readError(const std::error_code ec) noexcept override; + void readSuccess(std::size_t total_size) noexcept override; + asio::io_service io_service_; + // The consumer socket ConsumerSocket consumer_; + // The current url provided by the application + std::string current_url_; + // The current hICN name used for downloading + std::stringstream name_; + // Function to be called when the read is successful + std::function<std::shared_ptr<HTTPResponse>(std::size_t)> success_callback_; + // Return code for current download + RC return_code_; + + // Application provided callback for saving the received content during + // the download. If this callback is used, the HTTPClient will NOT save + // any byte internally. + ReadBytesCallback *read_bytes_callback_; + + // Internal read buffer and HTTP response, to be used if the application does + // not provide any read_bytes_callback + std::unique_ptr<utils::MemBuf> read_buffer_; std::shared_ptr<HTTPResponse> response_; + // Timer std::unique_ptr<asio::steady_timer> timer_; }; diff --git a/libtransport/src/hicn/transport/http/server_acceptor.h b/libtransport/src/hicn/transport/http/server_acceptor.h index 99480028a..4e7350b76 100644 --- a/libtransport/src/hicn/transport/http/server_acceptor.h +++ b/libtransport/src/hicn/transport/http/server_acceptor.h @@ -15,7 +15,6 @@ #pragma once -#include <hicn/transport/http/callbacks.h> #include <hicn/transport/http/default_values.h> #include <hicn/transport/http/request.h> #include <hicn/transport/http/server_publisher.h> @@ -31,6 +30,9 @@ namespace http { class HTTPServerAcceptor { friend class HTTPServerPublisher; + using OnHttpRequest = + std::function<void(std::shared_ptr<HTTPServerPublisher> &, + const uint8_t *, std::size_t, int request_id)>; public: HTTPServerAcceptor(std::string &&server_locator, OnHttpRequest callback); diff --git a/libtransport/src/hicn/transport/http/server_publisher.h b/libtransport/src/hicn/transport/http/server_publisher.h index f3c39c3fe..1f12fd8f9 100644 --- a/libtransport/src/hicn/transport/http/server_publisher.h +++ b/libtransport/src/hicn/transport/http/server_publisher.h @@ -63,8 +63,6 @@ class HTTPServerPublisher { std::unique_ptr<ProducerSocket> producer_; ProducerInterestCallback interest_enter_callback_; utils::UserCallback wait_callback_; - - ContentBuffer receive_buffer_; }; } // end namespace http diff --git a/libtransport/src/hicn/transport/interfaces/CMakeLists.txt b/libtransport/src/hicn/transport/interfaces/CMakeLists.txt index cbf371bac..2ff4fda56 100644 --- a/libtransport/src/hicn/transport/interfaces/CMakeLists.txt +++ b/libtransport/src/hicn/transport/interfaces/CMakeLists.txt @@ -19,15 +19,13 @@ list(APPEND HEADER_FILES ${CMAKE_CURRENT_SOURCE_DIR}/socket_producer.h ${CMAKE_CURRENT_SOURCE_DIR}/rtc_socket_consumer.h ${CMAKE_CURRENT_SOURCE_DIR}/rtc_socket_producer.h - ${CMAKE_CURRENT_SOURCE_DIR}/async_transport.h - ${CMAKE_CURRENT_SOURCE_DIR}/full_duplex_socket.h ${CMAKE_CURRENT_SOURCE_DIR}/publication_options.h ${CMAKE_CURRENT_SOURCE_DIR}/socket_options_default_values.h ${CMAKE_CURRENT_SOURCE_DIR}/socket_options_keys.h + ${CMAKE_CURRENT_SOURCE_DIR}/callbacks.h ) list(APPEND SOURCE_FILES - ${CMAKE_CURRENT_SOURCE_DIR}/full_duplex_socket.cc ${CMAKE_CURRENT_SOURCE_DIR}/rtc_socket_consumer.cc ${CMAKE_CURRENT_SOURCE_DIR}/rtc_socket_producer.cc ${CMAKE_CURRENT_SOURCE_DIR}/socket_producer.cc diff --git a/libtransport/src/hicn/transport/interfaces/async_transport.h b/libtransport/src/hicn/transport/interfaces/async_transport.h deleted file mode 100644 index 692dd318c..000000000 --- a/libtransport/src/hicn/transport/interfaces/async_transport.h +++ /dev/null @@ -1,641 +0,0 @@ - -/* - * Copyright (c) 2017-2019 Cisco and/or its affiliates. - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at: - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#pragma once - -#include <hicn/transport/interfaces/publication_options.h> -#include <hicn/transport/interfaces/socket.h> -#include <hicn/transport/portability/portability.h> - -#ifndef _WIN32 -#include <sys/uio.h> -#endif - -#include <memory> - -namespace transport { - -namespace interface { - -/* - * flags given by the application for write* calls - */ -enum class WriteFlags : uint32_t { - NONE = 0x00, - /* - * Whether to delay the output until a subsequent non-corked write. - * (Note: may not be supported in all subclasses or on all platforms.) - */ - CORK = 0x01, - /* - * for a socket that has ACK latency enabled, it will cause the kernel - * to fire a TCP ESTATS event when the last byte of the given write call - * will be acknowledged. - */ - EOR = 0x02, - /* - * this indicates that only the write side of socket should be shutdown - */ - WRITE_SHUTDOWN = 0x04, - /* - * use msg zerocopy if allowed - */ - WRITE_MSG_ZEROCOPY = 0x08, -}; - -/* - * union operator - */ -TRANSPORT_ALWAYS_INLINE WriteFlags operator|(WriteFlags a, WriteFlags b) { - return static_cast<WriteFlags>(static_cast<uint32_t>(a) | - static_cast<uint32_t>(b)); -} - -/* - * compound assignment union operator - */ -TRANSPORT_ALWAYS_INLINE WriteFlags &operator|=(WriteFlags &a, WriteFlags b) { - a = a | b; - return a; -} - -/* - * intersection operator - */ -TRANSPORT_ALWAYS_INLINE WriteFlags operator&(WriteFlags a, WriteFlags b) { - return static_cast<WriteFlags>(static_cast<uint32_t>(a) & - static_cast<uint32_t>(b)); -} - -/* - * compound assignment intersection operator - */ -TRANSPORT_ALWAYS_INLINE WriteFlags &operator&=(WriteFlags &a, WriteFlags b) { - a = a & b; - return a; -} - -/* - * exclusion parameter - */ -TRANSPORT_ALWAYS_INLINE WriteFlags operator~(WriteFlags a) { - return static_cast<WriteFlags>(~static_cast<uint32_t>(a)); -} - -/* - * unset operator - */ -TRANSPORT_ALWAYS_INLINE WriteFlags unSet(WriteFlags a, WriteFlags b) { - return a & ~b; -} - -/* - * inclusion operator - */ -TRANSPORT_ALWAYS_INLINE bool isSet(WriteFlags a, WriteFlags b) { - return (a & b) == b; -} - -class ConnectCallback { - public: - virtual ~ConnectCallback() = default; - - /** - * connectSuccess() will be invoked when the connection has been - * successfully established. - */ - virtual void connectSuccess() noexcept = 0; - - /** - * connectErr() will be invoked if the connection attempt fails. - * - * @param ex An exception describing the error that occurred. - */ - virtual void connectErr(const std::error_code ec) noexcept = 0; -}; - -/** - * AsyncSocket defines an asynchronous API for streaming I/O. - * - * This class provides an API to for asynchronously waiting for data - * on a streaming transport, and for asynchronously sending data. - * - * The APIs for reading and writing are intentionally asymmetric. Waiting for - * data to read is a persistent API: a callback is installed, and is notified - * whenever new data is available. It continues to be notified of new events - * until it is uninstalled. - * - * AsyncSocket does not provide read timeout functionality, because it - * typically cannot determine when the timeout should be active. Generally, a - * timeout should only be enabled when processing is blocked waiting on data - * from the remote endpoint. For server-side applications, the timeout should - * not be active if the server is currently processing one or more outstanding - * requests on this transport. For client-side applications, the timeout - * should not be active if there are no requests pending on the transport. - * Additionally, if a client has multiple pending requests, it will ususally - * want a separate timeout for each request, rather than a single read timeout. - * - * The write API is fairly intuitive: a user can request to send a block of - * data, and a callback will be informed once the entire block has been - * transferred to the kernel, or on error. AsyncSocket does provide a send - * timeout, since most callers want to give up if the remote end stops - * responding and no further progress can be made sending the data. - */ -class AsyncSocket { - public: - /** - * Close the transport. - * - * This gracefully closes the transport, waiting for all pending write - * requests to complete before actually closing the underlying transport. - * - * If a read callback is set, readEOF() will be called immediately. If there - * are outstanding write requests, the close will be delayed until all - * remaining writes have completed. No new writes may be started after - * close() has been called. - */ - virtual void close() = 0; - - /** - * Close the transport immediately. - * - * This closes the transport immediately, dropping any outstanding data - * waiting to be written. - * - * If a read callback is set, readEOF() will be called immediately. - * If there are outstanding write requests, these requests will be aborted - * and writeError() will be invoked immediately on all outstanding write - * callbacks. - */ - virtual void closeNow() = 0; - - /** - * Perform a half-shutdown of the write side of the transport. - * - * The caller should not make any more calls to write() or writev() after - * shutdownWrite() is called. Any future write attempts will fail - * immediately. - * - * Not all transport types support half-shutdown. If the underlying - * transport does not support half-shutdown, it will fully shutdown both the - * read and write sides of the transport. (Fully shutting down the socket is - * better than doing nothing at all, since the caller may rely on the - * shutdownWrite() call to notify the other end of the connection that no - * more data can be read.) - * - * If there is pending data still waiting to be written on the transport, - * the actual shutdown will be delayed until the pending data has been - * written. - * - * Note: There is no corresponding shutdownRead() equivalent. Simply - * uninstall the read callback if you wish to stop reading. (On TCP sockets - * at least, shutting down the read side of the socket is a no-op anyway.) - */ - virtual void shutdownWrite() = 0; - - /** - * Perform a half-shutdown of the write side of the transport. - * - * shutdownWriteNow() is identical to shutdownWrite(), except that it - * immediately performs the shutdown, rather than waiting for pending writes - * to complete. Any pending write requests will be immediately failed when - * shutdownWriteNow() is called. - */ - virtual void shutdownWriteNow() = 0; - - /** - * Determine if transport is open and ready to read or write. - * - * Note that this function returns false on EOF; you must also call error() - * to distinguish between an EOF and an error. - * - * @return true iff the transport is open and ready, false otherwise. - */ - virtual bool good() const = 0; - - /** - * Determine if the transport is readable or not. - * - * @return true iff the transport is readable, false otherwise. - */ - virtual bool readable() const = 0; - - /** - * Determine if the transport is writable or not. - * - * @return true iff the transport is writable, false otherwise. - */ - virtual bool writable() const { - // By default return good() - leave it to implementers to override. - return good(); - } - - /** - * Determine if the there is pending data on the transport. - * - * @return true iff the if the there is pending data, false otherwise. - */ - virtual bool isPending() const { return readable(); } - - /** - * Determine if transport is connected to the endpoint - * - * @return false iff the transport is connected, otherwise true - */ - virtual bool connected() const = 0; - - /** - * Determine if an error has occurred with this transport. - * - * @return true iff an error has occurred (not EOF). - */ - virtual bool error() const = 0; - - // /** - // * Attach the transport to a EventBase. - // * - // * This may only be called if the transport is not currently attached to a - // * EventBase (by an earlier call to detachEventBase()). - // * - // * This method must be invoked in the EventBase's thread. - // */ - // virtual void attachEventBase(EventBase* eventBase) = 0; - - // /** - // * Detach the transport from its EventBase. - // * - // * This may only be called when the transport is idle and has no reads or - // * writes pending. Once detached, the transport may not be used again - // until - // * it is re-attached to a EventBase by calling attachEventBase(). - // * - // * This method must be called from the current EventBase's thread. - // */ - // virtual void detachEventBase() = 0; - - // /** - // * Determine if the transport can be detached. - // * - // * This method must be called from the current EventBase's thread. - // */ - // virtual bool isDetachable() const = 0; - - /** - * Set the send timeout. - * - * If write requests do not make any progress for more than the specified - * number of milliseconds, fail all pending writes and close the transport. - * - * If write requests are currently pending when setSendTimeout() is called, - * the timeout interval is immediately restarted using the new value. - * - * @param milliseconds The timeout duration, in milliseconds. If 0, no - * timeout will be used. - */ - virtual void setSendTimeout(uint32_t milliseconds) = 0; - - /** - * Get the send timeout. - * - * @return Returns the current send timeout, in milliseconds. A return value - * of 0 indicates that no timeout is set. - */ - virtual uint32_t getSendTimeout() const = 0; - - virtual void connect(ConnectCallback *callback, - const core::Prefix &prefix_) = 0; - - // /** - // * Get the address of the local endpoint of this transport. - // * - // * This function may throw AsyncSocketException on error. - // * - // * @param address The local address will be stored in the specified - // * SocketAddress. - // */ - // virtual void getLocalAddress(* address) const = 0; - - virtual size_t getAppBytesWritten() const = 0; - virtual size_t getRawBytesWritten() const = 0; - virtual size_t getAppBytesReceived() const = 0; - virtual size_t getRawBytesReceived() const = 0; - - class BufferCallback { - public: - virtual ~BufferCallback() {} - virtual void onEgressBuffered() = 0; - virtual void onEgressBufferCleared() = 0; - }; - - ~AsyncSocket() = default; -}; - -class AsyncAcceptor { - public: - class AcceptCallback { - public: - virtual ~AcceptCallback() = default; - - /** - * connectionAccepted() is called whenever a new client connection is - * received. - * - * The AcceptCallback will remain installed after connectionAccepted() - * returns. - * - * @param fd The newly accepted client socket. The AcceptCallback - * assumes ownership of this socket, and is responsible - * for closing it when done. The newly accepted file - * descriptor will have already been put into - * non-blocking mode. - * @param clientAddr A reference to a SocketAddress struct containing the - * client's address. This struct is only guaranteed to - * remain valid until connectionAccepted() returns. - */ - virtual void connectionAccepted( - const core::Name &subscriber_name) noexcept = 0; - - /** - * acceptError() is called if an error occurs while accepting. - * - * The AcceptCallback will remain installed even after an accept error, - * as the errors are typically somewhat transient, such as being out of - * file descriptors. The server socket must be explicitly stopped if you - * wish to stop accepting after an error. - * - * @param ex An exception representing the error. - */ - virtual void acceptError(const std::exception &ex) noexcept = 0; - - /** - * acceptStarted() will be called in the callback's EventBase thread - * after this callback has been added to the AsyncServerSocket. - * - * acceptStarted() will be called before any calls to connectionAccepted() - * or acceptError() are made on this callback. - * - * acceptStarted() makes it easier for callbacks to perform initialization - * inside the callback thread. (The call to addAcceptCallback() must - * always be made from the AsyncServerSocket's primary EventBase thread. - * acceptStarted() provides a hook that will always be invoked in the - * callback's thread.) - * - * Note that the call to acceptStarted() is made once the callback is - * added, regardless of whether or not the AsyncServerSocket is actually - * accepting at the moment. acceptStarted() will be called even if the - * AsyncServerSocket is paused when the callback is added (including if - * the initial call to startAccepting() on the AsyncServerSocket has not - * been made yet). - */ - virtual void acceptStarted() noexcept {} - - /** - * acceptStopped() will be called when this AcceptCallback is removed from - * the AsyncServerSocket, or when the AsyncServerSocket is destroyed, - * whichever occurs first. - * - * No more calls to connectionAccepted() or acceptError() will be made - * after acceptStopped() is invoked. - */ - virtual void acceptStopped() noexcept {} - }; - - /** - * Wait for subscribers - * - */ - virtual void waitForSubscribers(AcceptCallback *cb) = 0; -}; - -class AsyncReader { - public: - class ReadCallback { - public: - virtual ~ReadCallback() = default; - - /** - * When data becomes available, getReadBuffer() will be invoked to get the - * buffer into which data should be read. - * - * This method allows the ReadCallback to delay buffer allocation until - * data becomes available. This allows applications to manage large - * numbers of idle connections, without having to maintain a separate read - * buffer for each idle connection. - * - * It is possible that in some cases, getReadBuffer() may be called - * multiple times before readDataAvailable() is invoked. In this case, the - * data will be written to the buffer returned from the most recent call to - * readDataAvailable(). If the previous calls to readDataAvailable() - * returned different buffers, the ReadCallback is responsible for ensuring - * that they are not leaked. - * - * If getReadBuffer() throws an exception, returns a nullptr buffer, or - * returns a 0 length, the ReadCallback will be uninstalled and its - * readError() method will be invoked. - * - * getReadBuffer() is not allowed to change the transport state before it - * returns. (For example, it should never uninstall the read callback, or - * set a different read callback.) - * - * @param bufReturn getReadBuffer() should update *bufReturn to contain the - * address of the read buffer. This parameter will never - * be nullptr. - * @param lenReturn getReadBuffer() should update *lenReturn to contain the - * maximum number of bytes that may be written to the read - * buffer. This parameter will never be nullptr. - * - * - * XXX TODO this does not seems to be completely true Checlk i/. - */ - virtual void getReadBuffer(void **bufReturn, size_t *lenReturn) = 0; - - /** - * readDataAvailable() will be invoked when data has been successfully read - * into the buffer returned by the last call to getReadBuffer(). - * - * The read callback remains installed after readDataAvailable() returns. - * It must be explicitly uninstalled to stop receiving read events. - * getReadBuffer() will be called at least once before each call to - * readDataAvailable(). getReadBuffer() will also be called before any - * call to readEOF(). - * - * @param len The number of bytes placed in the buffer. - */ - - virtual void readDataAvailable(size_t len) noexcept = 0; - - /** - * When data becomes available, isBufferMovable() will be invoked to figure - * out which API will be used, readBufferAvailable() or - * readDataAvailable(). If isBufferMovable() returns true, that means - * ReadCallback supports the IOBuf ownership transfer and - * readBufferAvailable() will be used. Otherwise, not. - - * By default, isBufferMovable() always return false. If - * readBufferAvailable() is implemented and to be invoked, You should - * overwrite isBufferMovable() and return true in the inherited class. - * - * This method allows the AsyncSocket/AsyncSSLSocket do buffer allocation by - * itself until data becomes available. Compared with the pre/post buffer - * allocation in getReadBuffer()/readDataAvailabe(), readBufferAvailable() - * has two advantages. First, this can avoid memcpy. E.g., in - * AsyncSSLSocket, the decrypted data was copied from the openssl internal - * buffer to the readbuf buffer. With the buffer ownership transfer, the - * internal buffer can be directly "moved" to ReadCallback. Second, the - * memory allocation can be more precise. The reason is - * AsyncSocket/AsyncSSLSocket can allocate the memory of precise size - * because they have more context about the available data than - * ReadCallback. Think about the getReadBuffer() pre-allocate 4072 bytes - * buffer, but the available data is always 16KB (max OpenSSL record size). - */ - - virtual bool isBufferMovable() noexcept { return false; } - - /** - * Suggested buffer size, allocated for read operations, - * if callback is movable and supports folly::IOBuf - */ - - virtual size_t maxBufferSize() const { - return 64 * 1024; // 64K - } - - /** - * readBufferAvailable() will be invoked when data has been successfully - * read. - * - * Note that only either readBufferAvailable() or readDataAvailable() will - * be invoked according to the return value of isBufferMovable(). The timing - * and aftereffect of readBufferAvailable() are the same as - * readDataAvailable() - * - * @param readBuf The unique pointer of read buffer. - */ - - // virtual void readBufferAvailable(uint8_t** buffer, std::size_t - // *buf_length) noexcept {} - - virtual void readBufferAvailable(ContentBuffer &&buffer) noexcept {} - - // virtual void readBufferAvailable(utils::SharableBuffer<uint8_t>&& buffer) - // noexcept {} - - /** - * readEOF() will be invoked when the transport is closed. - * - * The read callback will be automatically uninstalled immediately before - * readEOF() is invoked. - */ - virtual void readEOF() noexcept = 0; - - /** - * readError() will be invoked if an error occurs reading from the - * transport. - * - * The read callback will be automatically uninstalled immediately before - * readError() is invoked. - * - * @param ex An exception describing the error that occurred. - */ - virtual void readErr(const std::error_code ec) noexcept = 0; - }; - - // Read methods that aren't part of AsyncTransport. - virtual void setReadCB(ReadCallback *callback) = 0; - virtual ReadCallback *getReadCallback() const = 0; - - protected: - virtual ~AsyncReader() = default; -}; - -class AsyncWriter { - public: - class WriteCallback { - public: - virtual ~WriteCallback() = default; - - /** - * writeSuccess() will be invoked when all of the data has been - * successfully written. - * - * Note that this mainly signals that the buffer containing the data to - * write is no longer needed and may be freed or re-used. It does not - * guarantee that the data has been fully transmitted to the remote - * endpoint. For example, on socket-based transports, writeSuccess() only - * indicates that the data has been given to the kernel for eventual - * transmission. - */ - virtual void writeSuccess() noexcept = 0; - - /** - * writeError() will be invoked if an error occurs writing the data. - * - * @param bytesWritten The number of bytes that were successfull - * @param ex An exception describing the error that occurred. - */ - virtual void writeErr(size_t bytesWritten) noexcept = 0; - }; - - /** - * If you supply a non-null WriteCallback, exactly one of writeSuccess() - * or writeErr() will be invoked when the write completes. If you supply - * the same WriteCallback object for multiple write() calls, it will be - * invoked exactly once per call. The only way to cancel outstanding - * write requests is to close the socket (e.g., with closeNow() or - * shutdownWriteNow()). When closing the socket this way, writeErr() will - * still be invoked once for each outstanding write operation. - */ - virtual void write(WriteCallback *callback, const void *buf, size_t bytes, - const PublicationOptions &options, - WriteFlags flags = WriteFlags::NONE) = 0; - - /** - * If you supply a non-null WriteCallback, exactly one of writeSuccess() - * or writeErr() will be invoked when the write completes. If you supply - * the same WriteCallback object for multiple write() calls, it will be - * invoked exactly once per call. The only way to cancel outstanding - * write requests is to close the socket (e.g., with closeNow() or - * shutdownWriteNow()). When closing the socket this way, writeErr() will - * still be invoked once for each outstanding write operation. - */ - virtual void write(WriteCallback *callback, ContentBuffer &&output_buffer, - const PublicationOptions &options, - WriteFlags flags = WriteFlags::NONE) = 0; - - // /** - // * If you supply a non-null WriteCallback, exactly one of writeSuccess() - // * or writeErr() will be invoked when the write completes. If you supply - // * the same WriteCallback object for multiple write() calls, it will be - // * invoked exactly once per call. The only way to cancel outstanding - // * write requests is to close the socket (e.g., with closeNow() or - // * shutdownWriteNow()). When closing the socket this way, writeErr() will - // * still be invoked once for each outstanding write operation. - // */ - // virtual void writeChain( - // WriteCallback* callback, - // std::unique_ptr<IOBuf>&& buf, - // WriteFlags flags = WriteFlags::NONE) = 0; - - virtual void setWriteCB(WriteCallback *callback) = 0; - virtual WriteCallback *getWriteCallback() const = 0; - - protected: - virtual ~AsyncWriter() = default; -}; - -} // namespace interface - -} // namespace transport
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/interfaces/callbacks.h b/libtransport/src/hicn/transport/interfaces/callbacks.h new file mode 100644 index 000000000..24f47eb75 --- /dev/null +++ b/libtransport/src/hicn/transport/interfaces/callbacks.h @@ -0,0 +1,110 @@ +/* + * Copyright (c) 2017-2019 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include <functional> +#include <system_error> + +namespace utils { +class MemBuf; +} + +namespace transport { + +namespace protocol { + +class IcnObserver; +class TransportStatistics; + +} // namespace protocol + +namespace core { + +class ContentObject; +class Interest; +} // namespace core + +namespace interface { + +// Forward declarations +class ConsumerSocket; +class ProducerSocket; + +/** + * The ConsumerInterestCallback will be called in different parts of the + * consumer socket processing pipeline, with a ConsumerSocket and an Interest as + * parameters. + */ +using ConsumerInterestCallback = + std::function<void(ConsumerSocket &, const core::Interest &)>; + +/** + * The ConsumerTimerCallback is called periodically for exposing to applications + * a summary of the statistics of the transport protocol in use. + */ +using ConsumerTimerCallback = std::function<void( + ConsumerSocket &, const protocol::TransportStatistics &stats)>; + +/** + * The ProducerContentCallback will be called by the producer socket right after + * a content has been segmented and published. + */ +using ProducerContentCallback = std::function<void( + ProducerSocket &, const std::error_code &, uint64_t bytes_written)>; + +/** + * The ConsumerContentObjectCallback will be called in different parts of the + * consumer socket processing pipeline, with a ConsumerSocket and an + * ContentObject as parameters. + */ +using ConsumerContentObjectCallback = + std::function<void(ConsumerSocket &, const core::ContentObject &)>; + +/** + * The ConsumerContentObjectVerificationCallback will be called by the transport + * if an application is willing to verify each content object. Note that a + * better alternative is to instrument the transport to perform the verification + * autonomously, without requiring the intervention of the application. + */ +using ConsumerContentObjectVerificationCallback = + std::function<bool(ConsumerSocket &, const core::ContentObject &)>; + +/** + * The ConsumerManifestCallback will be called by the consumer socket when a + * manifest is received. + */ +using ConsumerManifestCallback = + std::function<void(ConsumerSocket &, const core::ContentObjectManifest &)>; + +/** + * The ProducerContentObjectCallback will be called in different parts of the + * consumer socket processing pipeline, with a ProducerSocket and an + * ContentObject as parameters. + */ +using ProducerContentObjectCallback = + std::function<void(ProducerSocket &, core::ContentObject &)>; + +/** + * The ProducerContentObjectCallback will be called in different parts of the + * consumer socket processing pipeline, with a ProducerSocket and an + * Interest as parameters. + */ +using ProducerInterestCallback = + std::function<void(ProducerSocket &, core::Interest &)>; + +} // namespace interface + +} // namespace transport
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/interfaces/full_duplex_socket.cc b/libtransport/src/hicn/transport/interfaces/full_duplex_socket.cc deleted file mode 100644 index fdd422dee..000000000 --- a/libtransport/src/hicn/transport/interfaces/full_duplex_socket.cc +++ /dev/null @@ -1,490 +0,0 @@ -/* - * Copyright (c) 2017-2019 Cisco and/or its affiliates. - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at: - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include <hicn/transport/interfaces/full_duplex_socket.h> -#include <hicn/transport/interfaces/socket_options_default_values.h> - -#include <memory> - -namespace transport { - -namespace interface { - -static const std::string producer_identity = "producer_socket"; - -AsyncFullDuplexSocket::AsyncFullDuplexSocket(const Prefix &locator) - : AsyncFullDuplexSocket(locator, internal_io_service_) {} - -AsyncFullDuplexSocket::AsyncFullDuplexSocket(const Prefix &locator, - asio::io_service &io_service) - : locator_(locator), - incremental_suffix_(0), - io_service_(io_service), - work_(io_service), - producer_(std::make_unique<ProducerSocket>(io_service_)), - consumer_(std::make_unique<ConsumerSocket>( - TransportProtocolAlgorithms::RAAQM /* , io_service_ */)), - read_callback_(nullptr), - write_callback_(nullptr), - connect_callback_(nullptr), - accept_callback_(nullptr), - internal_connect_callback_(new OnConnectCallback(*this)), - internal_signal_callback_(new OnSignalCallback(*this)), - send_timeout_milliseconds_(~0), - counters_({0}), - receive_buffer_(ContentBuffer()) { - using namespace transport; - using namespace std::placeholders; - producer_->registerPrefix(locator); - - producer_->setSocketOption( - ProducerCallbacksOptions::CACHE_MISS, - std::bind(&AsyncFullDuplexSocket::onControlInterest, this, _1, _2)); - - producer_->setSocketOption(GeneralTransportOptions::OUTPUT_BUFFER_SIZE, - uint32_t{150000}); - - ProducerContentCallback producer_callback = - std::bind(&AsyncFullDuplexSocket::onContentProduced, this, _1, _2, _3); - producer_->setSocketOption(ProducerCallbacksOptions::CONTENT_PRODUCED, - producer_callback); - - producer_->connect(); - - consumer_->setSocketOption(ConsumerCallbacksOptions::CONTENT_OBJECT_TO_VERIFY, - (ConsumerContentObjectVerificationCallback)[]( - ConsumerSocket & s, const ContentObject &c) - ->bool { return true; }); - - ConsumerContentCallback consumer_callback = - std::bind(&AsyncFullDuplexSocket::onContentRetrieved, this, _1, _2, _3); - consumer_->setSocketOption(ConsumerCallbacksOptions::CONTENT_RETRIEVED, - consumer_callback); - - consumer_->setSocketOption(GeneralTransportOptions::MAX_INTEREST_RETX, - uint32_t{4}); - - consumer_->connect(); -} - -void AsyncFullDuplexSocket::close() { - this->consumer_->stop(); - this->producer_->stop(); -} - -void AsyncFullDuplexSocket::closeNow() { close(); } - -void AsyncFullDuplexSocket::shutdownWrite() { producer_->stop(); } - -void AsyncFullDuplexSocket::shutdownWriteNow() { shutdownWrite(); } - -bool AsyncFullDuplexSocket::good() const { return true; } - -bool AsyncFullDuplexSocket::readable() const { - // TODO return status of consumer socket - return true; -} - -bool AsyncFullDuplexSocket::writable() const { - // TODO return status of producer socket - return true; -} - -bool AsyncFullDuplexSocket::isPending() const { - // TODO save if there are production operation in the ops queue - // in producer socket - return true; -} - -bool AsyncFullDuplexSocket::connected() const { - // No real connection here (ICN world). Return good - return good(); -} - -bool AsyncFullDuplexSocket::error() const { return !good(); } - -void AsyncFullDuplexSocket::setSendTimeout(uint32_t milliseconds) { - // TODO if production takes too much to complete - // let's abort the operation. - - // Normally with hicn this should be done for content - // pull, not for production. - - send_timeout_milliseconds_ = milliseconds; -} - -uint32_t AsyncFullDuplexSocket::getSendTimeout() const { - return send_timeout_milliseconds_; -} - -size_t AsyncFullDuplexSocket::getAppBytesWritten() const { - return counters_.app_bytes_written_; -} - -size_t AsyncFullDuplexSocket::getRawBytesWritten() const { return 0; } - -size_t AsyncFullDuplexSocket::getAppBytesReceived() const { - return counters_.app_bytes_read_; -} - -size_t AsyncFullDuplexSocket::getRawBytesReceived() const { return 0; } - -void AsyncFullDuplexSocket::connect(ConnectCallback *callback, - const core::Prefix &prefix) { - connect_callback_ = callback; - - // Create an interest for a subscription - auto interest = - core::Interest::Ptr(new core::Interest(prefix.makeRandomName())); - auto _payload = utils::MemBuf::create(sizeof(ActionMessage)); - _payload->append(sizeof(ActionMessage)); - auto payload = _payload->writableData(); - ActionMessage *subscription_message = - reinterpret_cast<ActionMessage *>(payload); - subscription_message->header.msg_type = MessageType::ACTION; - subscription_message->action = Action::SUBSCRIBE; - subscription_message->header.reserved[0] = 0; - subscription_message->header.reserved[1] = 0; - - // Set the name the other part should use for notifying a content production - sync_notification_ = std::move(locator_.makeRandomName()); - sync_notification_.copyToDestination( - reinterpret_cast<uint8_t *>(subscription_message->name)); - - TRANSPORT_LOGI( - "Trying to connect. Sending interest: %s, name for notifications: %s", - prefix.getName().toString().c_str(), - sync_notification_.toString().c_str()); - - interest->setLifetime(1000); - interest->appendPayload(std::move(_payload)); - consumer_->asyncSendInterest(std::move(interest), - internal_connect_callback_.get()); -} - -void AsyncFullDuplexSocket::write(WriteCallback *callback, const void *buf, - size_t bytes, - const PublicationOptions &options, - WriteFlags flags) { - using namespace transport; - - // 1 asynchronously write the content. I assume here the - // buffer contains the whole application frame. FIXME: check - // if this is true and fix it accordingly - std::cout << "Size of the PAYLOAD: " << bytes << std::endl; - - if (bytes > core::Packet::default_mtu - sizeof(PayloadMessage)) { - TRANSPORT_LOGI("Producing content with name %s", - options.getName().toString().c_str()); - producer_->asyncProduce(options.getName(), - reinterpret_cast<const uint8_t *>(buf), bytes); - signalProductionToSubscribers(options.getName()); - } else { - TRANSPORT_LOGI("Sending payload through interest"); - piggybackPayloadToSubscribers( - options.getName(), reinterpret_cast<const uint8_t *>(buf), bytes); - } -} - -void AsyncFullDuplexSocket::write(WriteCallback *callback, - ContentBuffer &&output_buffer, - const PublicationOptions &options, - WriteFlags flags) { - using namespace transport; - - // 1 asynchronously write the content. I assume here the - // buffer contains the whole application frame. FIXME: check - // if this is true and fix it accordingly - std::cout << "Size of the PAYLOAD: " << output_buffer->size() << std::endl; - - if (output_buffer->size() > - core::Packet::default_mtu - sizeof(PayloadMessage)) { - TRANSPORT_LOGI("Producing content with name %s", - options.getName().toString().c_str()); - producer_->asyncProduce(options.getName(), std::move(output_buffer)); - signalProductionToSubscribers(options.getName()); - } else { - TRANSPORT_LOGI("Sending payload through interest"); - piggybackPayloadToSubscribers(options.getName(), &(*output_buffer)[0], - output_buffer->size()); - } -} - -void AsyncFullDuplexSocket::piggybackPayloadToSubscribers( - const core::Name &name, const uint8_t *buffer, std::size_t bytes) { - for (auto &sub : subscribers_) { - auto interest = core::Interest::Ptr(new core::Interest(name)); - auto _payload = utils::MemBuf::create(bytes + sizeof(PayloadMessage)); - _payload->append(bytes + sizeof(PayloadMessage)); - auto payload = _payload->writableData(); - - PayloadMessage *interest_payload = - reinterpret_cast<PayloadMessage *>(payload); - interest_payload->header.msg_type = MessageType::PAYLOAD; - interest_payload->header.reserved[0] = 0; - interest_payload->header.reserved[1] = 0; - interest_payload->reserved[0] = 0; - std::memcpy(payload + sizeof(PayloadMessage), buffer, bytes); - interest->appendPayload(std::move(_payload)); - - // Set the timeout of 0.2 second - interest->setLifetime(1000); - interest->setName(sub); - interest->getWritableName().setSuffix(incremental_suffix_++); - // TRANSPORT_LOGI("Sending signalization to %s", - // interest->getName().toString().c_str()); - - consumer_->asyncSendInterest(std::move(interest), - internal_signal_callback_.get()); - } -} - -void AsyncFullDuplexSocket::signalProductionToSubscribers( - const core::Name &name) { - // Signal the other part we are producing a content - // Create an interest for a subscription - - for (auto &sub : subscribers_) { - auto interest = core::Interest::Ptr(new core::Interest(name)); - // Todo consider using preallocated pool of membufs - auto _payload = utils::MemBuf::create(sizeof(ActionMessage)); - _payload->append(sizeof(ActionMessage)); - auto payload = interest->getPayload()->writableData(); - - ActionMessage *produce_notification = - reinterpret_cast<ActionMessage *>(payload); - produce_notification->header.msg_type = MessageType::ACTION; - produce_notification->action = Action::SIGNAL_PRODUCTION; - produce_notification->header.reserved[0] = 0; - produce_notification->header.reserved[1] = 0; - name.copyToDestination( - reinterpret_cast<uint8_t *>(produce_notification->name)); - interest->appendPayload(std::move(_payload)); - - // Set the timeout of 0.2 second - interest->setLifetime(1000); - interest->setName(sub); - interest->getWritableName().setSuffix(incremental_suffix_++); - // TRANSPORT_LOGI("Sending signalization to %s", - // interest->getName().toString().c_str()); - - consumer_->asyncSendInterest(std::move(interest), - internal_signal_callback_.get()); - } -} - -void AsyncFullDuplexSocket::waitForSubscribers(AcceptCallback *cb) { - accept_callback_ = cb; -} - -std::shared_ptr<core::ContentObject> -AsyncFullDuplexSocket::decodeSynchronizationMessage( - const core::Interest &interest) { - auto mesg = interest.getPayload(); - const MessageHeader *header = - reinterpret_cast<const MessageHeader *>(mesg->data()); - - switch (header->msg_type) { - case MessageType::ACTION: { - // Check what is the action to perform - const ActionMessage *message = - reinterpret_cast<const ActionMessage *>(header); - - if (message->action == Action::SUBSCRIBE) { - // Add consumer to list on consumers to be notified - auto ret = - subscribers_.emplace(AF_INET6, (const uint8_t *)message->name, 0); - TRANSPORT_LOGI("Added subscriber %s :)", ret.first->toString().c_str()); - if (ret.second) { - accept_callback_->connectionAccepted(*ret.first); - } - - TRANSPORT_LOGI("Connection success!"); - - sync_notification_ = std::move(locator_.makeRandomName()); - return createSubscriptionResponse(sync_notification_); - - } else if (message->action == Action::CANCEL_SUBSCRIPTION) { - // XXX Modify name!!! Each allocated name allocates a 128 bit array. - subscribers_.erase( - core::Name(AF_INET6, (const uint8_t *)message->name, 0)); - return createAck(); - } else if (message->action == Action::SIGNAL_PRODUCTION) { - // trigger a reverse pull for the name contained in the message - core::Name n(AF_INET6, (const uint8_t *)message->name, 0); - std::cout << "PROD NOTIFICATION: Content to retrieve: " << n - << std::endl; - std::cout << "PROD NOTIFICATION: Interest name: " << interest.getName() - << std::endl; // << " compared to " << sync_notification_ << - // std::endl; - - if (sync_notification_.equals(interest.getName(), false)) { - std::cout << "Starting reverse pull for " << n << std::endl; - consumer_->asyncConsume(n, receive_buffer_); - return createAck(); - } - } else { - TRANSPORT_LOGE("Received unknown message. Dropping it."); - } - - break; - } - case MessageType::RESPONSE: { - throw errors::RuntimeException( - "The response should be a content object!!"); - } - case MessageType::PAYLOAD: { - // The interest contains the payload directly. - // We saved one round trip :) - - auto buffer = ContentBuffer(); - const uint8_t *data = mesg->data() + sizeof(PayloadMessage); - buffer->assign(data, data + mesg->length() - sizeof(PayloadMessage)); - read_callback_->readBufferAvailable(std::move(buffer)); - return createAck(); - } - default: { return std::shared_ptr<core::ContentObject>(nullptr); } - } - - return std::shared_ptr<core::ContentObject>(nullptr); -} - -void AsyncFullDuplexSocket::onControlInterest(ProducerSocket &s, - const core::Interest &i) { - auto payload = i.getPayload(); - if (payload->length()) { - // Try to decode payload and see if starting an async pull operation - auto response = decodeSynchronizationMessage(i); - if (response) { - response->setName(i.getName()); - s.produce(*response); - } - } -} - -void AsyncFullDuplexSocket::onContentProduced(ProducerSocket &producer, - const std::error_code &ec, - uint64_t bytes_written) { - if (write_callback_) { - if (!ec) { - write_callback_->writeSuccess(); - } else { - write_callback_->writeErr(bytes_written); - } - } -} - -void AsyncFullDuplexSocket::onContentRetrieved(ConsumerSocket &s, - std::size_t size, - const std::error_code &ec) { - // Sanity check - if (size != receive_buffer_->size()) { - TRANSPORT_LOGE( - "Received content size differs from size retrieved from the buffer."); - return; - } - - TRANSPORT_LOGI("Received content with size %zu", size); - if (!ec) { - read_callback_->readBufferAvailable(std::move(receive_buffer_)); - } else { - TRANSPORT_LOGE("Error retrieving content."); - } - // consumer_->stop(); -} - -void AsyncFullDuplexSocket::OnConnectCallback::onContentObject( - core::Interest::Ptr &&, core::ContentObject::Ptr &&content_object) { - // The ack message should contain the name to be used for notifying - // the production of the content to the other part - - if (content_object->getPayload()->length() == 0) { - TRANSPORT_LOGW("Connection response message empty...."); - return; - } - - SubscriptionResponseMessage *response = - reinterpret_cast<SubscriptionResponseMessage *>( - content_object->getPayload()->writableData()); - - if (response->response.header.msg_type == MessageType::RESPONSE) { - if (response->response.return_code == ReturnCode::OK) { - auto ret = - socket_.subscribers_.emplace(AF_INET6, (uint8_t *)response->name, 0); - TRANSPORT_LOGI("Successfully connected!!!! Subscriber added: %s", - ret.first->toString().c_str()); - socket_.connect_callback_->connectSuccess(); - } - } -} - -void AsyncFullDuplexSocket::OnSignalCallback::onContentObject( - core::Interest::Ptr &&, core::ContentObject::Ptr &&content_object) { - return; -} - -void AsyncFullDuplexSocket::OnSignalCallback::onTimeout( - core::Interest::Ptr &&interest) { - TRANSPORT_LOGE("Retransmitting signalization interest to %s!!", - interest->getName().toString().c_str()); - socket_.consumer_->asyncSendInterest(std::move(interest), - socket_.internal_signal_callback_.get()); -} - -void AsyncFullDuplexSocket::OnConnectCallback::onTimeout( - core::Interest::Ptr &&interest) { - socket_.connect_callback_->connectErr( - std::make_error_code(std::errc::not_connected)); -} - -std::shared_ptr<core::ContentObject> AsyncFullDuplexSocket::createAck() { - // Send the response back - core::Name name("b001::abcd"); - auto response = std::make_shared<core::ContentObject>(name); - auto _payload = utils::MemBuf::create(sizeof(ActionMessage)); - _payload->append(sizeof(ResponseMessage)); - auto payload = response->getPayload()->data(); - ResponseMessage *response_message = (ResponseMessage *)payload; - response_message->header.msg_type = MessageType::RESPONSE; - response_message->header.reserved[0] = 0; - response_message->header.reserved[1] = 0; - response_message->return_code = ReturnCode::OK; - response->appendPayload(std::move(_payload)); - response->setLifetime(0); - return response; -} - -std::shared_ptr<core::ContentObject> -AsyncFullDuplexSocket::createSubscriptionResponse(const core::Name &name) { - // Send the response back - core::Name tmp_name("b001::abcd"); - auto response = std::make_shared<core::ContentObject>(tmp_name); - auto _payload = utils::MemBuf::create(sizeof(SubscriptionResponseMessage)); - _payload->append(sizeof(SubscriptionResponseMessage)); - auto payload = _payload->data(); - SubscriptionResponseMessage *response_message = - (SubscriptionResponseMessage *)payload; - response_message->response.header.msg_type = MessageType::RESPONSE; - response_message->response.header.reserved[0] = 0; - response_message->response.header.reserved[1] = 0; - response_message->response.return_code = ReturnCode::OK; - name.copyToDestination(reinterpret_cast<uint8_t *>(response_message->name)); - response->appendPayload(std::move(_payload)); - response->setLifetime(0); - return response; -} - -} // namespace interface -} // namespace transport diff --git a/libtransport/src/hicn/transport/interfaces/full_duplex_socket.h b/libtransport/src/hicn/transport/interfaces/full_duplex_socket.h deleted file mode 100644 index 438325fdb..000000000 --- a/libtransport/src/hicn/transport/interfaces/full_duplex_socket.h +++ /dev/null @@ -1,243 +0,0 @@ -/* - * Copyright (c) 2017-2019 Cisco and/or its affiliates. - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at: - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -/* - * This class is created for sending/receiving data over an ICN network. - */ - -#pragma once - -#include <hicn/transport/core/prefix.h> -#include <hicn/transport/interfaces/async_transport.h> -#include <hicn/transport/interfaces/socket_consumer.h> -#include <hicn/transport/interfaces/socket_producer.h> -#include <hicn/transport/portability/portability.h> - -#include <unordered_set> -#include <vector> - -namespace transport { - -namespace interface { - -enum class MessageType : uint8_t { ACTION, RESPONSE, PAYLOAD }; - -enum class Action : uint8_t { - SUBSCRIBE, - CANCEL_SUBSCRIPTION, - SIGNAL_PRODUCTION, -}; - -enum class ReturnCode : uint8_t { - OK, - FAILED, -}; - -struct MessageHeader { - MessageType msg_type; - uint8_t reserved[2]; -}; - -struct ActionMessage { - MessageHeader header; - Action action; - uint64_t name[2]; -}; - -struct ResponseMessage { - MessageHeader header; - ReturnCode return_code; -}; - -struct SubscriptionResponseMessage { - ResponseMessage response; - uint64_t name[2]; -}; - -struct PayloadMessage { - MessageHeader header; - uint8_t reserved[1]; -}; - -// struct NotificationMessage { -// Action action; -// uint8_t reserved[3]; -// uint64_t -// } - -using core::Prefix; - -class AsyncFullDuplexSocket : public AsyncSocket, - public AsyncReader, - public AsyncWriter, - public AsyncAcceptor { - private: - struct Counters { - uint64_t app_bytes_written_; - uint64_t app_bytes_read_; - - TRANSPORT_ALWAYS_INLINE void updateBytesWritten(uint64_t bytes) { - app_bytes_written_ += bytes; - } - - TRANSPORT_ALWAYS_INLINE void updateBytesRead(uint64_t bytes) { - app_bytes_read_ += bytes; - } - }; - - public: - using UniquePtr = std::unique_ptr<AsyncFullDuplexSocket>; - using SharedPtr = std::unique_ptr<AsyncFullDuplexSocket>; - - AsyncFullDuplexSocket(const Prefix &locator, asio::io_service &io_service); - AsyncFullDuplexSocket(const core::Prefix &locator); - - ~AsyncFullDuplexSocket(){}; - - using ReadCallback = AsyncReader::ReadCallback; - using WriteCallback = AsyncWriter::WriteCallback; - - TRANSPORT_ALWAYS_INLINE void setReadCB(ReadCallback *callback) override { - read_callback_ = callback; - } - - TRANSPORT_ALWAYS_INLINE ReadCallback *getReadCallback() const override { - return read_callback_; - } - - TRANSPORT_ALWAYS_INLINE void setWriteCB(WriteCallback *callback) override { - write_callback_ = callback; - } - - TRANSPORT_ALWAYS_INLINE WriteCallback *getWriteCallback() const override { - return write_callback_; - } - - TRANSPORT_ALWAYS_INLINE const core::Prefix &getLocator() { return locator_; } - - void connect(ConnectCallback *callback, const core::Prefix &prefix) override; - - void write(WriteCallback *callback, const void *buf, size_t bytes, - const PublicationOptions &options, - WriteFlags flags = WriteFlags::NONE) override; - - virtual void write(WriteCallback *callback, ContentBuffer &&output_buffer, - const PublicationOptions &options, - WriteFlags flags = WriteFlags::NONE) override; - - void waitForSubscribers(AcceptCallback *cb) override; - - void close() override; - - void closeNow() override; - - void shutdownWrite() override; - - void shutdownWriteNow() override; - - bool good() const override; - - bool readable() const override; - - bool writable() const override; - - bool isPending() const override; - - bool connected() const override; - - bool error() const override; - - void setSendTimeout(uint32_t milliseconds) override; - - size_t getAppBytesWritten() const override; - size_t getRawBytesWritten() const override; - size_t getAppBytesReceived() const override; - size_t getRawBytesReceived() const override; - - uint32_t getSendTimeout() const override; - - private: - std::shared_ptr<core::ContentObject> decodeSynchronizationMessage( - const core::Interest &interest); - - class OnConnectCallback : public BasePortal::ConsumerCallback { - public: - OnConnectCallback(AsyncFullDuplexSocket &socket) : socket_(socket){}; - virtual ~OnConnectCallback() = default; - void onContentObject(core::Interest::Ptr &&, - core::ContentObject::Ptr &&content_object) override; - void onTimeout(core::Interest::Ptr &&interest) override; - - private: - AsyncFullDuplexSocket &socket_; - }; - - class OnSignalCallback : public BasePortal::ConsumerCallback { - public: - OnSignalCallback(AsyncFullDuplexSocket &socket) : socket_(socket){}; - virtual ~OnSignalCallback() = default; - void onContentObject(core::Interest::Ptr &&, - core::ContentObject::Ptr &&content_object); - void onTimeout(core::Interest::Ptr &&interest); - - private: - AsyncFullDuplexSocket &socket_; - }; - - void onControlInterest(ProducerSocket &s, const core::Interest &i); - void onContentProduced(ProducerSocket &producer, const std::error_code &ec, - uint64_t bytes_written); - void onContentRetrieved(ConsumerSocket &s, std::size_t size, - const std::error_code &ec); - - void signalProductionToSubscribers(const core::Name &name); - void piggybackPayloadToSubscribers(const core::Name &name, - const uint8_t *buffer, std::size_t bytes); - - std::shared_ptr<core::ContentObject> createAck(); - std::shared_ptr<core::ContentObject> createSubscriptionResponse( - const core::Name &name); - - core::Prefix locator_; - uint32_t incremental_suffix_; - core::Name sync_notification_; - // std::unique_ptr<BasePortal> portal_; - asio::io_service internal_io_service_; - asio::io_service &io_service_; - asio::io_service::work work_; - - // These names represent the "locator" of a certain - // peer that subscribed to this. - std::unordered_set<core::Name> subscribers_; - - // Useful for publishing / Retrieving data - std::unique_ptr<ProducerSocket> producer_; - std::unique_ptr<ConsumerSocket> consumer_; - - ReadCallback *read_callback_; - WriteCallback *write_callback_; - ConnectCallback *connect_callback_; - AcceptCallback *accept_callback_; - - std::unique_ptr<OnConnectCallback> internal_connect_callback_; - std::unique_ptr<OnSignalCallback> internal_signal_callback_; - - uint32_t send_timeout_milliseconds_; - struct Counters counters_; - ContentBuffer receive_buffer_; -}; - -} // namespace interface -} // namespace transport diff --git a/libtransport/src/hicn/transport/interfaces/socket.h b/libtransport/src/hicn/transport/interfaces/socket.h index 7d50d0fbd..90f6a3ef6 100644 --- a/libtransport/src/hicn/transport/interfaces/socket.h +++ b/libtransport/src/hicn/transport/interfaces/socket.h @@ -16,18 +16,10 @@ #pragma once #include <hicn/transport/config.h> -#include <hicn/transport/core/content_object.h> #include <hicn/transport/core/facade.h> -#include <hicn/transport/core/interest.h> -#include <hicn/transport/core/manifest_format_fixed.h> -#include <hicn/transport/core/manifest_inline.h> -#include <hicn/transport/core/name.h> +#include <hicn/transport/interfaces/callbacks.h> #include <hicn/transport/interfaces/socket_options_default_values.h> #include <hicn/transport/interfaces/socket_options_keys.h> -#include <hicn/transport/protocols/statistics.h> -#include <hicn/transport/utils/crypto_suite.h> -#include <hicn/transport/utils/identity.h> -#include <hicn/transport/utils/verifier.h> #define SOCKET_OPTION_GET 0 #define SOCKET_OPTION_NOT_GET 1 @@ -39,26 +31,14 @@ namespace transport { -namespace protocol { -class IcnObserver; -class TransportStatistics; -} // namespace protocol - namespace interface { +// Forward Declarations template <typename PortalType> class Socket; -class ConsumerSocket; -class ProducerSocket; - -// using Interest = core::Interest; -// using ContentObject = core::ContentObject; -// using Name = core::Name; -// using HashAlgorithm = core::HashAlgorithm; -// using CryptoSuite = utils::CryptoSuite; -// using Identity = utils::Identity; -// using Verifier = utils::Verifier; +// Define the portal and its connector, depending on the compilation options +// passed by the build tool. using HicnForwarderPortal = core::HicnForwarderPortal; #ifdef __linux__ @@ -76,40 +56,6 @@ using BaseSocket = Socket<HicnForwarderPortal>; using BasePortal = HicnForwarderPortal; #endif -using PayloadType = core::PayloadType; -using Prefix = core::Prefix; -using Array = utils::Array<uint8_t>; -using ContentBuffer = std::shared_ptr<std::vector<uint8_t>>; - -using ConsumerInterestCallback = - std::function<void(ConsumerSocket &, const core::Interest &)>; - -using ConsumerContentCallback = - std::function<void(ConsumerSocket &, std::size_t, const std::error_code &)>; - -using ConsumerTimerCallback = std::function<void( - ConsumerSocket &, const protocol::TransportStatistics &stats)>; - -using ProducerContentCallback = std::function<void( - ProducerSocket &, const std::error_code &, uint64_t bytes_written)>; - -using ConsumerContentObjectCallback = - std::function<void(ConsumerSocket &, const core::ContentObject &)>; - -using ConsumerContentObjectVerificationCallback = - std::function<bool(ConsumerSocket &, const core::ContentObject &)>; - -using ConsumerManifestCallback = - std::function<void(ConsumerSocket &, const core::ContentObjectManifest &)>; - -using ProducerContentObjectCallback = - std::function<void(ProducerSocket &, core::ContentObject &)>; - -using ProducerInterestCallback = - std::function<void(ProducerSocket &, core::Interest &)>; - -using namespace protocol; - template <typename PortalType> class Socket { static_assert(std::is_same<PortalType, HicnForwarderPortal>::value diff --git a/libtransport/src/hicn/transport/interfaces/socket_consumer.cc b/libtransport/src/hicn/transport/interfaces/socket_consumer.cc index ca9722849..37d545779 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_consumer.cc +++ b/libtransport/src/hicn/transport/interfaces/socket_consumer.cc @@ -48,7 +48,6 @@ ConsumerSocket::ConsumerSocket(int protocol, asio::io_service &io_service) is_async_(false), verifier_(std::make_shared<utils::Verifier>()), verify_signature_(false), - content_buffer_(nullptr), on_interest_output_(VOID_HANDLER), on_interest_timeout_(VOID_HANDLER), on_interest_satisfied_(VOID_HANDLER), @@ -56,7 +55,8 @@ ConsumerSocket::ConsumerSocket(int protocol, asio::io_service &io_service) on_content_object_verification_(VOID_HANDLER), on_content_object_(VOID_HANDLER), on_manifest_(VOID_HANDLER), - on_payload_retrieved_(VOID_HANDLER), + stats_summary_(VOID_HANDLER), + read_callback_(nullptr), virtual_download_(false), rtt_stats_(false), timer_interval_milliseconds_(0) { @@ -81,13 +81,11 @@ ConsumerSocket::~ConsumerSocket() { void ConsumerSocket::connect() { portal_->connect(); } -int ConsumerSocket::consume(const Name &name, ContentBuffer &receive_buffer) { +int ConsumerSocket::consume(const Name &name) { if (transport_protocol_->isRunning()) { return CONSUMER_BUSY; } - content_buffer_ = receive_buffer; - network_name_ = name; network_name_.setSuffix(0); is_async_ = false; @@ -97,14 +95,12 @@ int ConsumerSocket::consume(const Name &name, ContentBuffer &receive_buffer) { return CONSUMER_FINISHED; } -int ConsumerSocket::asyncConsume(const Name &name, - ContentBuffer &receive_buffer) { +int ConsumerSocket::asyncConsume(const Name &name) { if (!async_downloader_.stopped()) { - async_downloader_.add([this, receive_buffer, name]() { + async_downloader_.add([this, name]() { network_name_ = std::move(name); network_name_.setSuffix(0); is_async_ = true; - content_buffer_ = receive_buffer; transport_protocol_->start(); }); } diff --git a/libtransport/src/hicn/transport/interfaces/socket_consumer.h b/libtransport/src/hicn/transport/interfaces/socket_consumer.h index ceed53954..40344af5d 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_consumer.h +++ b/libtransport/src/hicn/transport/interfaces/socket_consumer.h @@ -19,6 +19,11 @@ #include <hicn/transport/interfaces/socket_options_default_values.h> #include <hicn/transport/protocols/protocol.h> #include <hicn/transport/utils/event_thread.h> +#include <hicn/transport/utils/verifier.h> + +extern "C" { +#include <parc/security/parc_KeyId.h> +} #define CONSUMER_FINISHED 0 #define CONSUMER_BUSY 1 @@ -28,28 +33,210 @@ namespace transport { namespace interface { +using namespace core; +using namespace protocol; + +/** + * @brief Main interface for consumer applications. + * + * The consumer socket is the main interface for a consumer application. + * It allows to retrieve an application data from one/many producers, by + * hiding all the complexity of the transport protocol used underneath. + */ class ConsumerSocket : public BaseSocket { public: + /** + * The ReadCallback is a class which can be used by the transport for both + * querying the application needs and notifying events. + * + * Beware that the methods of this class will be called synchronously while + * the transport is working, so the operations the application is performing + * on the data retrieved should be executed in another thread in an + * asynchronous manner. Blocking one of these callbacks means blocking the + * transport. + */ + class ReadCallback { + public: + virtual ~ReadCallback() = default; + + /** + * This API will specify to the transport whether the buffer should be + * allocated by the application (and then the retrieved content will be + * copied there) or the transport should allocate the buffer and "move" it + * to the application. In other words, if isBufferMovable return true, the + * transport will transfer the ownership of the read buffer to the + * application, without performing an additional copy, while if it returns + * false the transport will use the getReadBuffer API. + * + * By default this method returns true. + * + */ + virtual bool isBufferMovable() noexcept { return true; } + + /** + * This method will be called by the transport when the content is + * available. The application can then allocate its own buffer and provide + * the address to the transport, which will use it for writing the data. + * Note that if the application won't allocate enough memory this method + * will be called several times, until the internal read buffer will be + * emptied. For ensuring this method will be called once, applications + * should allocate at least maxBufferSize() bytes. + * + * @param application_buffer - Pointer to the application's buffer. + * @param max_length - The length of the application buffer. + */ + virtual void getReadBuffer(uint8_t **application_buffer, + size_t *max_length) = 0; + + /** + * This method will be called by the transport after calling getReadBuffer, + * in order to notify the application that length bytes are available in the + * buffer. The max_length size of the buffer could be larger than the actual + * amount of bytes written. + * + * @param length - The number of bytes placed in the buffer. + */ + virtual void readDataAvailable(size_t length) noexcept = 0; + + /** + * This method will be called by the transport for understanding how many + * bytes it should read (at most) before notifying the application. + * + * By default it reads 64 KB. + */ + virtual size_t maxBufferSize() const { return 64 * 1024; } + + /** + * This method will be called by the transport iff (isBufferMovable == + * true). The unique_ptr underlines the fact that the ownership of the + * buffer is being transferred to the application. + * + * @param buffer - The buffer + */ + virtual void readBufferAvailable( + std::unique_ptr<utils::MemBuf> &&buffer) noexcept {} + + /** + * readError() will be invoked if an error occurs reading from the + * transport. + * + * @param ec - An error code describing the error. + */ + virtual void readError(const std::error_code ec) noexcept = 0; + + /** + * This callback will be invoked when the whole content is retrieved. The + * transport itself knows when a content is retrieved (since it is not an + * opaque bytestream like TCP), and the transport itself is able to tell + * the application when the transfer is done. + */ + virtual void readSuccess(std::size_t total_size) noexcept = 0; + }; + + /** + * @brief Create a new consumer socket. + * + * @param protocol - The transport protocol to use. So far the following + * transport are supported: + * - CBR: Constant bitrate + * - Raaqm: Based on paper: Optimal multipath congestion control and request + * forwarding in information-centric networks: Protocol design and + * experimentation. G Carofiglio, M Gallo, L Muscariello. Computer Networks + * 110, 104-117 + * - RTC: Real time communication + */ explicit ConsumerSocket(int protocol); explicit ConsumerSocket(int protocol, asio::io_service &io_service); + /** + * @brief Destroy the consumer socket. + */ ~ConsumerSocket(); + /** + * @brief Connect the consumer socket to the underlying hICN forwarder. + */ void connect() override; - int consume(const Name &name, ContentBuffer &receive_buffer); - - int asyncConsume(const Name &name, ContentBuffer &receive_buffer); - + /** + * Retrieve a content using the protocol specified in the constructor. + * This function blocks until the whole content is downloaded. + * For monitoring the status of the download, the application MUST set the + * ConsumerRead callback. This callback will be called periodically (depending + * on the needs of the application), allowing the application to save the + * retrieved data. + * + * @param name - The name of the content to retrieve. + * + * @return CONSUMER_BUSY if a pending download exists + * @return CONSUMER_FINISHED when the download finishes + * + * Notice that the fact consume() returns CONSUMER_FINISHED does not imply the + * content retrieval succeeded. This information can be obtained from the + * error code in CONTENT_RETRIEVED callback. + */ + int consume(const Name &name); + int asyncConsume(const Name &name); + + /** + * Send an interest asynchronously in another thread, which is the same used + * for asyncConsume. + * + * @param interest - An Interest::Ptr to the interest. Notice that the + * application looses the ownership of the interest, which is transferred to + * the library itself. + * @param callback - A ConsumerCallback containing the events to be trigger in + * case of timeout or content reception. + * + */ void asyncSendInterest(Interest::Ptr &&interest, Portal::ConsumerCallback *callback); + /** + * Stops the consumer socket. If several downloads are queued (using + * asyncConsume), this call stops just the current one. + */ void stop(); + /** + * Resume the download from the same exact point it stopped. + */ void resume(); + /** + * Get the io_service which is running the transport protocol event loop. + * + * @return A reference to the internal io_service where the transport protocol + * is running. + */ asio::io_service &getIoService() override; + TRANSPORT_ALWAYS_INLINE int setSocketOption( + int socket_option_key, ReadCallback *socket_option_value) { + switch (socket_option_key) { + case ConsumerCallbacksOptions::READ_CALLBACK: + read_callback_ = socket_option_value; + break; + default: + return SOCKET_OPTION_NOT_SET; + } + + return SOCKET_OPTION_SET; + } + + TRANSPORT_ALWAYS_INLINE int getSocketOption( + int socket_option_key, ReadCallback **socket_option_value) { + switch (socket_option_key) { + case ConsumerCallbacksOptions::READ_CALLBACK: + *socket_option_value = read_callback_; + break; + default: + return SOCKET_OPTION_NOT_GET; + } + + return SOCKET_OPTION_GET; + } + TRANSPORT_ALWAYS_INLINE int setSocketOption(int socket_option_key, double socket_option_value) { switch (socket_option_key) { @@ -150,12 +337,6 @@ class ConsumerSocket : public BaseSocket { break; } - case ConsumerCallbacksOptions::CONTENT_RETRIEVED: - if (socket_option_value == VOID_HANDLER) { - on_payload_retrieved_ = VOID_HANDLER; - break; - } - case RateEstimationOptions::RATE_ESTIMATION_BATCH_PARAMETER: if (socket_option_value > 0) { rate_estimation_batching_parameter_ = socket_option_value; @@ -275,20 +456,6 @@ class ConsumerSocket : public BaseSocket { } TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, ConsumerContentCallback socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::CONTENT_RETRIEVED: - on_payload_retrieved_ = socket_option_value; - break; - - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption( int socket_option_key, ConsumerManifestCallback socket_option_value) { switch (socket_option_key) { case ConsumerCallbacksOptions::MANIFEST_INPUT: @@ -332,21 +499,6 @@ class ConsumerSocket : public BaseSocket { } TRANSPORT_ALWAYS_INLINE int setSocketOption( - int socket_option_key, - const std::shared_ptr<std::vector<uint8_t>> &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::APPLICATION_BUFFER: - content_buffer_ = socket_option_value; - break; - - default: - return SOCKET_OPTION_NOT_SET; - } - - return SOCKET_OPTION_SET; - } - - TRANSPORT_ALWAYS_INLINE int setSocketOption( int socket_option_key, const std::string &socket_option_value) { switch (socket_option_key) { case GeneralTransportOptions::CERTIFICATE: @@ -569,18 +721,6 @@ class ConsumerSocket : public BaseSocket { } TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, ConsumerContentCallback **socket_option_value) { - switch (socket_option_key) { - case ConsumerCallbacksOptions::CONTENT_RETRIEVED: - *socket_option_value = &on_payload_retrieved_; - return SOCKET_OPTION_GET; - - default: - return SOCKET_OPTION_NOT_GET; - } - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption( int socket_option_key, ConsumerManifestCallback **socket_option_value) { switch (socket_option_key) { case ConsumerCallbacksOptions::MANIFEST_INPUT: @@ -636,20 +776,6 @@ class ConsumerSocket : public BaseSocket { } TRANSPORT_ALWAYS_INLINE int getSocketOption( - int socket_option_key, - std::shared_ptr<std::vector<uint8_t>> &socket_option_value) { - switch (socket_option_key) { - case GeneralTransportOptions::APPLICATION_BUFFER: - socket_option_value = content_buffer_; - break; - default: - return SOCKET_OPTION_NOT_GET; - } - - return SOCKET_OPTION_GET; - } - - TRANSPORT_ALWAYS_INLINE int getSocketOption( int socket_option_key, std::string &socket_option_value) { switch (socket_option_key) { case DataLinkOptions::OUTPUT_INTERFACE: @@ -718,8 +844,6 @@ class ConsumerSocket : public BaseSocket { PARCKeyId *key_id_; bool verify_signature_; - ContentBuffer content_buffer_; - ConsumerInterestCallback on_interest_retransmission_; ConsumerInterestCallback on_interest_output_; ConsumerInterestCallback on_interest_timeout_; @@ -730,11 +854,10 @@ class ConsumerSocket : public BaseSocket { ConsumerContentObjectCallback on_content_object_; ConsumerManifestCallback on_manifest_; - - ConsumerContentCallback on_payload_retrieved_; - ConsumerTimerCallback stats_summary_; + ReadCallback *read_callback_; + // Virtual download for traffic generator bool virtual_download_; diff --git a/libtransport/src/hicn/transport/interfaces/socket_options_keys.h b/libtransport/src/hicn/transport/interfaces/socket_options_keys.h index bcf049310..c21108186 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_options_keys.h +++ b/libtransport/src/hicn/transport/interfaces/socket_options_keys.h @@ -78,7 +78,7 @@ typedef enum { CONTENT_OBJECT_INPUT = 411, MANIFEST_INPUT = 412, CONTENT_OBJECT_TO_VERIFY = 413, - CONTENT_RETRIEVED = 414, + READ_CALLBACK = 414, STATS_SUMMARY = 415 } ConsumerCallbacksOptions; diff --git a/libtransport/src/hicn/transport/interfaces/socket_producer.cc b/libtransport/src/hicn/transport/interfaces/socket_producer.cc index 3f9d4959d..c4cf95895 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_producer.cc +++ b/libtransport/src/hicn/transport/interfaces/socket_producer.cc @@ -14,6 +14,7 @@ */ #include <hicn/transport/interfaces/socket_producer.h> +#include <hicn/transport/utils/identity.h> #include <functional> @@ -336,16 +337,6 @@ void ProducerSocket::asyncProduce(const Name &suffix, const uint8_t *buf, } } -void ProducerSocket::asyncProduce(const Name &suffix, - ContentBuffer &&output_buffer) { - if (!async_thread_.stopped()) { - async_thread_.add( - [this, suff = suffix, buffer = std::move(output_buffer)]() { - produce(suff, &(*buffer)[0], buffer->size(), true); - }); - } -} - void ProducerSocket::onInterest(Interest &interest) { if (on_interest_input_ != VOID_HANDLER) { on_interest_input_(*this, interest); diff --git a/libtransport/src/hicn/transport/interfaces/socket_producer.h b/libtransport/src/hicn/transport/interfaces/socket_producer.h index 6ba5671cc..200c32a95 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_producer.h +++ b/libtransport/src/hicn/transport/interfaces/socket_producer.h @@ -30,6 +30,10 @@ #define REGISTRATION_FAILURE 2 #define REGISTRATION_IN_PROGRESS 3 +namespace utils { +class Identity; +} + namespace transport { namespace interface { @@ -59,7 +63,7 @@ class ProducerSocket : public Socket<BasePortal>, void asyncProduce(const Name &suffix, const uint8_t *buf, size_t buffer_size); - void asyncProduce(const Name &suffix, ContentBuffer &&output_buffer); + void asyncProduce(const Name &suffix); void asyncProduce(ContentObject &content_object); diff --git a/libtransport/src/hicn/transport/protocols/raaqm.cc b/libtransport/src/hicn/transport/protocols/raaqm.cc index b8a7c9610..7f0310e7c 100644 --- a/libtransport/src/hicn/transport/protocols/raaqm.cc +++ b/libtransport/src/hicn/transport/protocols/raaqm.cc @@ -38,14 +38,14 @@ RaaqmTransportProtocol::RaaqmTransportProtocol(ConsumerSocket *icnet_socket) } RaaqmTransportProtocol::~RaaqmTransportProtocol() { - if (this->rate_estimator_) { - delete this->rate_estimator_; + if (rate_estimator_) { + delete rate_estimator_; } } int RaaqmTransportProtocol::start() { - if (this->rate_estimator_) { - this->rate_estimator_->onStart(); + if (rate_estimator_) { + rate_estimator_->onStart(); } if (!cur_path_) { @@ -75,13 +75,13 @@ int RaaqmTransportProtocol::start() { choice_param); if (choice_param == 1) { - this->rate_estimator_ = new ALaTcpEstimator(); + rate_estimator_ = new ALaTcpEstimator(); } else { - this->rate_estimator_ = new SimpleEstimator(alpha, batching_param); + rate_estimator_ = new SimpleEstimator(alpha, batching_param); } socket_->getSocketOption(RateEstimationOptions::RATE_ESTIMATION_OBSERVER, - &this->rate_estimator_->observer_); + &rate_estimator_->observer_); // Current path auto cur_path = std::make_unique<RaaqmDataPath>( @@ -126,7 +126,7 @@ void RaaqmTransportProtocol::increaseWindow() { socket_->setSocketOption(GeneralTransportOptions::CURRENT_WINDOW_SIZE, current_window_size_); } - this->rate_estimator_->onWindowIncrease(current_window_size_); + rate_estimator_->onWindowIncrease(current_window_size_); } void RaaqmTransportProtocol::decreaseWindow() { @@ -145,7 +145,7 @@ void RaaqmTransportProtocol::decreaseWindow() { socket_->setSocketOption(GeneralTransportOptions::CURRENT_WINDOW_SIZE, current_window_size_); } - this->rate_estimator_->onWindowDecrease(current_window_size_); + rate_estimator_->onWindowDecrease(current_window_size_); } void RaaqmTransportProtocol::afterDataUnsatisfied(uint64_t segment) { @@ -158,8 +158,8 @@ void RaaqmTransportProtocol::afterContentReception( updatePathTable(content_object); increaseWindow(); updateRtt(interest.getName().getSuffix()); - this->rate_estimator_->onDataReceived((int)content_object.payloadSize() + - (int)content_object.headerSize()); + rate_estimator_->onDataReceived((int)content_object.payloadSize() + + (int)content_object.headerSize()); // Set drop probablility and window size accordingly RAAQM(); } @@ -368,7 +368,12 @@ void RaaqmTransportProtocol::onContentSegment( reassemble(std::move(content_object)); } else if (TRANSPORT_EXPECT_FALSE(incremental_suffix == index_manager_->getFinalSuffix())) { - onContentReassembled(std::make_error_code(std::errc(0))); + interface::ConsumerSocket::ReadCallback *on_payload = nullptr; + socket_->getSocketOption(READ_CALLBACK, &on_payload); + + if (on_payload != nullptr) { + on_payload->readSuccess(stats_.getBytesRecv()); + } } } else { // TODO Application policy check @@ -487,8 +492,8 @@ void RaaqmTransportProtocol::sendInterest(std::uint64_t next_suffix) { return; } - // This is set to ~0 so that the next interest_retransmissions_ + 1, performed - // by sendInterest, will result in 0 + // This is set to ~0 so that the next interest_retransmissions_ + 1, + // performed by sendInterest, will result in 0 interest_retransmissions_[next_suffix & mask] = ~0; interest_timepoints_[next_suffix & mask] = utils::SteadyClock::now(); sendInterest(std::move(interest)); @@ -502,16 +507,23 @@ void RaaqmTransportProtocol::sendInterest(Interest::Ptr &&interest) { } void RaaqmTransportProtocol::onContentReassembled(std::error_code ec) { - interface::ConsumerContentCallback *on_payload = nullptr; - socket_->getSocketOption(CONTENT_RETRIEVED, &on_payload); - if (*on_payload != VOID_HANDLER) { - std::shared_ptr<std::vector<uint8_t>> content_buffer; - socket_->getSocketOption( - interface::GeneralTransportOptions::APPLICATION_BUFFER, content_buffer); - (*on_payload)(*socket_, content_buffer->size(), ec); + interface::ConsumerSocket::ReadCallback *on_payload = nullptr; + socket_->getSocketOption(READ_CALLBACK, &on_payload); + + if (on_payload == nullptr) { + throw errors::RuntimeException( + "The read callback must be installed in the transport before " + "starting " + "the content retrieval."); + } + + if (!ec) { + on_payload->readSuccess(stats_.getBytesRecv()); + } else { + on_payload->readError(ec); } - this->rate_estimator_->onDownloadFinished(); + rate_estimator_->onDownloadFinished(); stop(); } @@ -526,8 +538,8 @@ void RaaqmTransportProtocol::updateRtt(uint64_t segment) { // Update stats updateStats((uint32_t)segment, rtt.count(), now); - if (this->rate_estimator_) { - this->rate_estimator_->onRttUpdate((double)rtt.count()); + if (rate_estimator_) { + rate_estimator_->onRttUpdate((double)rtt.count()); } cur_path_->insertNewRtt(rtt.count()); @@ -676,4 +688,4 @@ void RaaqmTransportProtocol::checkForStalePaths() { } // end namespace protocol -} // end namespace transport +} // namespace transport diff --git a/libtransport/src/hicn/transport/protocols/reassembly.cc b/libtransport/src/hicn/transport/protocols/reassembly.cc index 899f701c7..a2062df93 100644 --- a/libtransport/src/hicn/transport/protocols/reassembly.cc +++ b/libtransport/src/hicn/transport/protocols/reassembly.cc @@ -17,6 +17,7 @@ #include <hicn/transport/protocols/indexing_manager.h> #include <hicn/transport/protocols/reassembly.h> #include <hicn/transport/utils/array.h> +#include <hicn/transport/utils/membuf.h> namespace transport { @@ -30,7 +31,8 @@ BaseReassembly::BaseReassembly(interface::ConsumerSocket *icn_socket, manifest_index_manager_( std::make_unique<ManifestIndexManager>(icn_socket)), index_manager_(incremental_index_manager_.get()), - index_(0) { + index_(0), + read_buffer_(nullptr) { setContentCallback(content_callback); } @@ -54,30 +56,88 @@ void BaseReassembly::reassemble(ContentObject::Ptr &&content_object) { void BaseReassembly::copyContent(const ContentObject &content_object) { auto a = content_object.getPayload(); - - std::shared_ptr<std::vector<uint8_t>> content_buffer; - reassembly_consumer_socket_->getSocketOption( - interface::GeneralTransportOptions::APPLICATION_BUFFER, content_buffer); - - content_buffer->insert(content_buffer->end(), (uint8_t *)a->data(), - (uint8_t *)a->data() + a->length()); + auto payload_length = a->length(); + auto write_size = std::min(payload_length, read_buffer_->tailroom()); + auto additional_bytes = payload_length > read_buffer_->tailroom() + ? payload_length - read_buffer_->tailroom() + : 0; + + std::memcpy(read_buffer_->writableTail(), a->data(), write_size); + read_buffer_->append(write_size); + + if (!read_buffer_->tailroom()) { + notifyApplication(); + std::memcpy(read_buffer_->writableTail(), a->data() + write_size, + additional_bytes); + read_buffer_->append(additional_bytes); + } bool download_completed = index_manager_->getFinalSuffix() == content_object.getName().getSuffix(); if (TRANSPORT_EXPECT_FALSE(download_completed)) { + notifyApplication(); content_callback_->onContentReassembled(std::make_error_code(std::errc(0))); } } +void BaseReassembly::notifyApplication() { + interface::ConsumerSocket::ReadCallback *read_callback = nullptr; + reassembly_consumer_socket_->getSocketOption( + interface::ConsumerCallbacksOptions::READ_CALLBACK, &read_callback); + + if (TRANSPORT_EXPECT_FALSE(!read_callback)) { + TRANSPORT_LOGE("Read callback not installed!"); + return; + } + + if (read_callback->isBufferMovable()) { + // No need to perform an additional copy. The whole buffer will be + // tranferred to the application. + + read_callback->readBufferAvailable(std::move(read_buffer_)); + read_buffer_ = utils::MemBuf::create(read_callback->maxBufferSize()); + } else { + // The buffer will be copied into the application-provided buffer + uint8_t *buffer; + std::size_t length; + std::size_t total_length = read_buffer_->length(); + + while (read_buffer_->length()) { + buffer = nullptr; + length = 0; + read_callback->getReadBuffer(&buffer, &length); + + if (!buffer || !length) { + throw errors::RuntimeException( + "Invalid buffer provided by the application."); + } + + auto to_copy = std::min(read_buffer_->length(), length); + std::memcpy(buffer, read_buffer_->data(), to_copy); + read_buffer_->trimStart(to_copy); + } + + read_callback->readDataAvailable(total_length); + read_buffer_->clear(); + } +} + void BaseReassembly::reset() { manifest_index_manager_->reset(); incremental_index_manager_->reset(); index_ = index_manager_->getNextReassemblySegment(); received_packets_.clear(); + + // reset read buffer + interface::ConsumerSocket::ReadCallback *read_callback; + reassembly_consumer_socket_->getSocketOption( + interface::ConsumerCallbacksOptions::READ_CALLBACK, &read_callback); + + read_buffer_ = utils::MemBuf::create(read_callback->maxBufferSize()); } } // namespace protocol -} // end namespace transport +} // namespace transport diff --git a/libtransport/src/hicn/transport/protocols/reassembly.h b/libtransport/src/hicn/transport/protocols/reassembly.h index 9efddb773..79f0ea4d2 100644 --- a/libtransport/src/hicn/transport/protocols/reassembly.h +++ b/libtransport/src/hicn/transport/protocols/reassembly.h @@ -20,6 +20,10 @@ namespace transport { +namespace interface { +class ConsumerReadCallback; +} + namespace protocol { // Forward Declaration @@ -54,6 +58,9 @@ class BaseReassembly : public Reassembly { virtual void reset() override; + private: + void notifyApplication(); + protected: // The consumer socket interface::ConsumerSocket *reassembly_consumer_socket_; @@ -63,6 +70,7 @@ class BaseReassembly : public Reassembly { std::unordered_map<std::uint32_t, ContentObject::Ptr> received_packets_; uint64_t index_; + std::unique_ptr<utils::MemBuf> read_buffer_; }; } // namespace protocol diff --git a/libtransport/src/hicn/transport/protocols/rtc.cc b/libtransport/src/hicn/transport/protocols/rtc.cc index 9402d3b02..4205ade4e 100644 --- a/libtransport/src/hicn/transport/protocols/rtc.cc +++ b/libtransport/src/hicn/transport/protocols/rtc.cc @@ -161,12 +161,11 @@ void RTCTransportProtocol::updateDelayStats( uint32_t segmentNumber = content_object.getName().getSuffix(); uint32_t pkt = segmentNumber & modMask_; - if (inflightInterests_[pkt].state != sent_) - return; + if (inflightInterests_[pkt].state != sent_) return; - if(interestRetransmissions_.find(segmentNumber) != + if (interestRetransmissions_.find(segmentNumber) != interestRetransmissions_.end()) - //this packet was rtx at least once + // this packet was rtx at least once return; uint32_t pathLabel = content_object.getPathLabel(); @@ -329,8 +328,7 @@ void RTCTransportProtocol::increaseWindow() { } else { currentCWin_ = min( maxCWin_, - (uint32_t)ceil(currentCWin_ + - (1.0 / (double)currentCWin_))); // linear + (uint32_t)ceil(currentCWin_ + (1.0 / (double)currentCWin_))); // linear } } @@ -363,7 +361,6 @@ void RTCTransportProtocol::sendInterest(Name *interest_name, bool rtx) { if (!rtx) { inflightInterestsCount_++; } - } void RTCTransportProtocol::scheduleNextInterests() { @@ -373,9 +370,9 @@ void RTCTransportProtocol::scheduleNextInterests() { while (inflightInterestsCount_ < currentCWin_) { Name *interest_name = nullptr; socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, - &interest_name); + &interest_name); - //we send the packet only if it is not pending yet + // we send the packet only if it is not pending yet interest_name->setSuffix(actualSegment_); if (portal_->interestIsPending(*interest_name)) { actualSegment_++; @@ -383,11 +380,11 @@ void RTCTransportProtocol::scheduleNextInterests() { } uint32_t pkt = actualSegment_ & modMask_; - //if we already reacevied the content we don't ask it again - if(inflightInterests_[pkt].state == received_ && - inflightInterests_[pkt].sequence == actualSegment_) { - actualSegment_++; - continue; + // if we already reacevied the content we don't ask it again + if (inflightInterests_[pkt].state == received_ && + inflightInterests_[pkt].sequence == actualSegment_) { + actualSegment_++; + continue; } inflightInterests_[pkt].transmissionTime = @@ -419,93 +416,93 @@ void RTCTransportProtocol::scheduleAppNackRtx(std::vector<uint32_t> &nacks) { #endif } -void RTCTransportProtocol::addRetransmissions(uint32_t val){ - //add only val in the rtx list +void RTCTransportProtocol::addRetransmissions(uint32_t val) { + // add only val in the rtx list addRetransmissions(val, val + 1); } -void RTCTransportProtocol::addRetransmissions(uint32_t start, uint32_t stop){ - for(uint32_t i = start; i < stop; i++){ +void RTCTransportProtocol::addRetransmissions(uint32_t start, uint32_t stop) { + for (uint32_t i = start; i < stop; i++) { auto it = interestRetransmissions_.find(i); - if(it == interestRetransmissions_.end()){ + if (it == interestRetransmissions_.end()) { if (lastSegNacked_ <= i) { - //i must be larger than the last past nack received + // i must be larger than the last past nack received interestRetransmissions_[i] = 0; } - }//if the retransmission is already there the rtx timer will - //take care of it + } // if the retransmission is already there the rtx timer will + // take care of it } retransmit(true); } -void RTCTransportProtocol::retransmit(bool first_rtx){ +void RTCTransportProtocol::retransmit(bool first_rtx) { auto it = interestRetransmissions_.begin(); - //cut len to max HICN_MAX_RTX_SIZE - //since we use a map, the smaller (and so the older) sequence number are at - //the beginnin of the map - while(interestRetransmissions_.size() > HICN_MAX_RTX_SIZE){ + // cut len to max HICN_MAX_RTX_SIZE + // since we use a map, the smaller (and so the older) sequence number are at + // the beginnin of the map + while (interestRetransmissions_.size() > HICN_MAX_RTX_SIZE) { it = interestRetransmissions_.erase(it); } it = interestRetransmissions_.begin(); - while (it != interestRetransmissions_.end()){ + while (it != interestRetransmissions_.end()) { uint32_t pkt = it->first & modMask_; - if(inflightInterests_[pkt].sequence != it->first){ - //this packet is not anymore in the inflight buffer, erase it + if (inflightInterests_[pkt].sequence != it->first) { + // this packet is not anymore in the inflight buffer, erase it it = interestRetransmissions_.erase(it); continue; } - //we retransmitted the packet too many times - if(it->second >= HICN_MAX_RTX){ + // we retransmitted the packet too many times + if (it->second >= HICN_MAX_RTX) { it = interestRetransmissions_.erase(it); continue; } - //this packet is too old - if((lastReceived_ > it->first) && - (lastReceived_ - it->first) > HICN_MAX_RTX_MAX_AGE){ + // this packet is too old + if ((lastReceived_ > it->first) && + (lastReceived_ - it->first) > HICN_MAX_RTX_MAX_AGE) { it = interestRetransmissions_.erase(it); continue; } - if(first_rtx){ - //TODO (optimization) - //the rtx that we never sent (it->second == 0) are all at the - //end, so we can go directly there - if(it->second == 0){ + if (first_rtx) { + // TODO (optimization) + // the rtx that we never sent (it->second == 0) are all at the + // end, so we can go directly there + if (it->second == 0) { inflightInterests_[pkt].transmissionTime = std::chrono::duration_cast<std::chrono::milliseconds>( - std::chrono::steady_clock::now().time_since_epoch()) - .count(); + std::chrono::steady_clock::now().time_since_epoch()) + .count(); it->second++; Name *interest_name = nullptr; socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, - &interest_name); + &interest_name); interest_name->setSuffix(it->first); sendInterest(interest_name, true); } ++it; - }else{ - //base on time + } else { + // base on time uint64_t now = std::chrono::duration_cast<std::chrono::milliseconds>( - std::chrono::steady_clock::now().time_since_epoch()) - .count(); - if((now - inflightInterests_[pkt].transmissionTime) > 20){ - //XXX replace 20 with rtt + std::chrono::steady_clock::now().time_since_epoch()) + .count(); + if ((now - inflightInterests_[pkt].transmissionTime) > 20) { + // XXX replace 20 with rtt inflightInterests_[pkt].transmissionTime = std::chrono::duration_cast<std::chrono::milliseconds>( - std::chrono::steady_clock::now().time_since_epoch()) - .count(); + std::chrono::steady_clock::now().time_since_epoch()) + .count(); it->second++; Name *interest_name = nullptr; socket_->getSocketOption(GeneralTransportOptions::NETWORK_NAME, - &interest_name); + &interest_name); interest_name->setSuffix(it->first); sendInterest(interest_name, true); } @@ -514,13 +511,13 @@ void RTCTransportProtocol::retransmit(bool first_rtx){ } } -void RTCTransportProtocol::checkRtx(){ +void RTCTransportProtocol::checkRtx() { retransmit(false); rtx_timer_->expires_from_now(std::chrono::milliseconds(20)); rtx_timer_->async_wait([this](std::error_code ec) { - if (ec) return; - checkRtx(); - }); + if (ec) return; + checkRtx(); + }); } void RTCTransportProtocol::onTimeout(Interest::Ptr &&interest) { @@ -533,25 +530,25 @@ void RTCTransportProtocol::onTimeout(Interest::Ptr &&interest) { inflightInterestsCount_--; } - //check how many times we sent this packet - auto it = interestRetransmissions_.find(segmentNumber); - if(it != interestRetransmissions_.end() && it->second >= HICN_MAX_RTX){ - inflightInterests_[pkt].state = lost_; - } + // check how many times we sent this packet + auto it = interestRetransmissions_.find(segmentNumber); + if (it != interestRetransmissions_.end() && it->second >= HICN_MAX_RTX) { + inflightInterests_[pkt].state = lost_; + } - if(inflightInterests_[pkt].state == sent_) { - inflightInterests_[pkt].state = timeout1_; - } else if (inflightInterests_[pkt].state == timeout1_) { - inflightInterests_[pkt].state = timeout2_; - } else if (inflightInterests_[pkt].state == timeout2_) { - inflightInterests_[pkt].state = lost_; - } + if (inflightInterests_[pkt].state == sent_) { + inflightInterests_[pkt].state = timeout1_; + } else if (inflightInterests_[pkt].state == timeout1_) { + inflightInterests_[pkt].state = timeout2_; + } else if (inflightInterests_[pkt].state == timeout2_) { + inflightInterests_[pkt].state = lost_; + } - if(inflightInterests_[pkt].state == lost_) { - interestRetransmissions_.erase(segmentNumber); - }else{ - addRetransmissions(segmentNumber); - } + if (inflightInterests_[pkt].state == lost_) { + interestRetransmissions_.erase(segmentNumber); + } else { + addRetransmissions(segmentNumber); + } scheduleNextInterests(); } @@ -562,12 +559,11 @@ bool RTCTransportProtocol::checkIfProducerIsActive( uint32_t productionSeg = *payload; uint32_t productionRate = *(++payload); - if (productionRate == 0) { // the producer socket is not active // in this case we consider only the first nack if (nack_timer_used_) { - return false; + return false; } nack_timer_used_ = true; @@ -680,12 +676,12 @@ void RTCTransportProtocol::onContentObject( ((1 - HICN_ESTIMATED_PACKET_SIZE) * payload->length()); if (inflightInterests_[pkt].state == sent_) { - inflightInterestsCount_--; //packet sent without timeouts + inflightInterestsCount_--; // packet sent without timeouts } if (inflightInterests_[pkt].state == sent_ && interestRetransmissions_.find(segmentNumber) == - interestRetransmissions_.end()){ + interestRetransmissions_.end()) { // we count only non retransmitted data in order to take into accunt only // the transmition rate of the producer receivedBytes_ += (uint32_t)(content_object->headerSize() + @@ -693,7 +689,7 @@ void RTCTransportProtocol::onContentObject( updateDelayStats(*content_object); addRetransmissions(lastReceived_ + 1, segmentNumber); - //lastReceived_ is updated only for data packets received without RTX + // lastReceived_ is updated only for data packets received without RTX lastReceived_ = segmentNumber; } @@ -704,7 +700,7 @@ void RTCTransportProtocol::onContentObject( increaseWindow(); } - //in any case we remove the packet from the rtx list + // in any case we remove the packet from the rtx list interestRetransmissions_.erase(segmentNumber); if (schedule_next_interest) { @@ -715,24 +711,50 @@ void RTCTransportProtocol::onContentObject( void RTCTransportProtocol::returnContentToApplication( const ContentObject &content_object) { // return content to the user - auto a = content_object.getPayload(); + auto read_buffer = content_object.getPayload(); - a->trimStart(HICN_TIMESTAMP_SIZE); - uint8_t *start = a->writableData(); - unsigned size = (unsigned)a->length(); + read_buffer->trimStart(HICN_TIMESTAMP_SIZE); // set offset between hICN and RTP packets - uint16_t rtp_seq = ntohs(*(((uint16_t *)start) + 1)); + uint16_t rtp_seq = ntohs(*(((uint16_t *)read_buffer->writableData()) + 1)); RTPhICN_offset_ = content_object.getName().getSuffix() - rtp_seq; - std::shared_ptr<std::vector<uint8_t>> content_buffer; - socket_->getSocketOption(APPLICATION_BUFFER, content_buffer); - content_buffer->insert(content_buffer->end(), start, start + size); + interface::ConsumerSocket::ReadCallback *read_callback = nullptr; + socket_->getSocketOption(READ_CALLBACK, &read_callback); + + if (read_callback == nullptr) { + throw errors::RuntimeException( + "The read callback must be installed in the transport before starting " + "the content retrieval."); + } + + if (read_callback->isBufferMovable()) { + read_callback->readBufferAvailable( + utils::MemBuf::copyBuffer(read_buffer->data(), read_buffer->length())); + } else { + // The buffer will be copied into the application-provided buffer + uint8_t *buffer; + std::size_t length; + std::size_t total_length = read_buffer->length(); + + while (read_buffer->length()) { + buffer = nullptr; + length = 0; + read_callback->getReadBuffer(&buffer, &length); + + if (!buffer || !length) { + throw errors::RuntimeException( + "Invalid buffer provided by the application."); + } + + auto to_copy = std::min(read_buffer->length(), length); + + std::memcpy(buffer, read_buffer->data(), to_copy); + read_buffer->trimStart(to_copy); + } - ConsumerContentCallback *on_payload = nullptr; - socket_->getSocketOption(CONTENT_RETRIEVED, &on_payload); - if ((*on_payload) != VOID_HANDLER) { - (*on_payload)(*socket_, size, std::make_error_code(std::errc(0))); + read_callback->readDataAvailable(total_length); + read_buffer->clear(); } } |