diff options
Diffstat (limited to 'libtransport')
12 files changed, 128 insertions, 149 deletions
diff --git a/libtransport/src/hicn/transport/core/hicn_binary_api.c b/libtransport/src/hicn/transport/core/hicn_binary_api.c index d909578a2..3868c0a14 100644 --- a/libtransport/src/hicn/transport/core/hicn_binary_api.c +++ b/libtransport/src/hicn/transport/core/hicn_binary_api.c @@ -134,8 +134,6 @@ int hicn_binary_api_register_cons_app( CONTEXT_SAVE(context_store, api, mp) - TRANSPORT_LOGI("Message created"); - return vpp_binary_api_send_request_wait_reply(api->vpp_api, mp); } diff --git a/libtransport/src/hicn/transport/core/portal.h b/libtransport/src/hicn/transport/core/portal.h index cadf37e8c..d9051c23c 100644 --- a/libtransport/src/hicn/transport/core/portal.h +++ b/libtransport/src/hicn/transport/core/portal.h @@ -49,18 +49,19 @@ static constexpr uint32_t pool_size = 2048; class HandlerMemory { #ifdef __vpp__ - static constexpr std::size_t memory_size = 1024 * 512; + static constexpr std::size_t memory_size = 1024 * 1024; + public: - HandlerMemory() : index_(0) { } + HandlerMemory() : index_(0) {} - HandlerMemory(const HandlerMemory&) = delete; - HandlerMemory& operator=(const HandlerMemory&) = delete; + HandlerMemory(const HandlerMemory &) = delete; + HandlerMemory &operator=(const HandlerMemory &) = delete; - TRANSPORT_ALWAYS_INLINE void* allocate(std::size_t size) { + TRANSPORT_ALWAYS_INLINE void *allocate(std::size_t size) { return &storage_[index_++ % memory_size]; } - TRANSPORT_ALWAYS_INLINE void deallocate(void* pointer) { } + TRANSPORT_ALWAYS_INLINE void deallocate(void *pointer) {} private: // Storage space used for handler-based custom memory allocation. @@ -68,16 +69,16 @@ class HandlerMemory { uint32_t index_; #else public: - HandlerMemory() { } + HandlerMemory() {} - HandlerMemory(const HandlerMemory&) = delete; - HandlerMemory& operator=(const HandlerMemory&) = delete; + HandlerMemory(const HandlerMemory &) = delete; + HandlerMemory &operator=(const HandlerMemory &) = delete; - TRANSPORT_ALWAYS_INLINE void* allocate(std::size_t size) { + TRANSPORT_ALWAYS_INLINE void *allocate(std::size_t size) { return ::operator new(size); } - TRANSPORT_ALWAYS_INLINE void deallocate(void* pointer) { + TRANSPORT_ALWAYS_INLINE void deallocate(void *pointer) { ::operator delete(pointer); } #endif @@ -90,34 +91,36 @@ class HandlerAllocator { public: using value_type = T; - explicit HandlerAllocator(HandlerMemory& mem) - : memory_(mem) {} + explicit HandlerAllocator(HandlerMemory &mem) : memory_(mem) {} template <typename U> - HandlerAllocator(const HandlerAllocator<U>& other) noexcept - : memory_(other.memory_) { } + HandlerAllocator(const HandlerAllocator<U> &other) noexcept + : memory_(other.memory_) {} - TRANSPORT_ALWAYS_INLINE bool operator==(const HandlerAllocator& other) const noexcept { + TRANSPORT_ALWAYS_INLINE bool operator==(const HandlerAllocator &other) const + noexcept { return &memory_ == &other.memory_; } - TRANSPORT_ALWAYS_INLINE bool operator!=(const HandlerAllocator& other) const noexcept { + TRANSPORT_ALWAYS_INLINE bool operator!=(const HandlerAllocator &other) const + noexcept { return &memory_ != &other.memory_; } - TRANSPORT_ALWAYS_INLINE T* allocate(std::size_t n) const { - return static_cast<T*>(memory_.allocate(sizeof(T) * n)); + TRANSPORT_ALWAYS_INLINE T *allocate(std::size_t n) const { + return static_cast<T *>(memory_.allocate(sizeof(T) * n)); } - TRANSPORT_ALWAYS_INLINE void deallocate(T* p, std::size_t /*n*/) const { + TRANSPORT_ALWAYS_INLINE void deallocate(T *p, std::size_t /*n*/) const { return memory_.deallocate(p); } private: - template <typename> friend class HandlerAllocator; + template <typename> + friend class HandlerAllocator; // The underlying memory. - HandlerMemory& memory_; + HandlerMemory &memory_; }; // Wrapper class template for handler objects to allow handler memory @@ -129,35 +132,33 @@ class CustomAllocatorHandler { public: using allocator_type = HandlerAllocator<Handler>; - CustomAllocatorHandler(HandlerMemory& m, Handler h) - : memory_(m), - handler_(h) { } + CustomAllocatorHandler(HandlerMemory &m, Handler h) + : memory_(m), handler_(h) {} allocator_type get_allocator() const noexcept { return allocator_type(memory_); } - template <typename ...Args> - void operator()(Args&&... args) { + template <typename... Args> + void operator()(Args &&... args) { handler_(std::forward<Args>(args)...); } private: - HandlerMemory& memory_; + HandlerMemory &memory_; Handler handler_; }; // Helper function to wrap a handler object to add custom allocation. template <typename Handler> inline CustomAllocatorHandler<Handler> makeCustomAllocatorHandler( - HandlerMemory& m, Handler h) { + HandlerMemory &m, Handler h) { return CustomAllocatorHandler<Handler>(m, h); } class Pool { public: - Pool(asio::io_service &io_service) - : io_service_(io_service) { + Pool(asio::io_service &io_service) : io_service_(io_service) { increasePendingInterestPool(); increaseInterestPool(); increaseContentObjectPool(); @@ -223,7 +224,7 @@ class Pool { asio::io_service &io_service_; }; -} +} // namespace portal_details using PendingInterestHashTable = std::unordered_map<uint32_t, PendingInterest::Ptr>; @@ -266,6 +267,7 @@ class Portal { typename ForwarderInt::ConnectorType>, ForwarderInt>::value, "ForwarderInt must inherit from ForwarderInterface!"); + public: class ConsumerCallback { public: @@ -290,7 +292,7 @@ class Portal { std::bind(&Portal::setLocalRoutes, this), io_service_, app_name_), forwarder_interface_(connector_), - packet_pool_(io_service) { } + packet_pool_(io_service) {} void setConsumerCallback(ConsumerCallback *consumer_callback) { consumer_callback_ = consumer_callback; @@ -310,7 +312,7 @@ class Portal { forwarder_interface_.connect(is_consumer); } - ~Portal() { stopEventsLoop(true); } + ~Portal() { killConnection(); } TRANSPORT_ALWAYS_INLINE bool interestIsPending(const Name &name) { auto it = @@ -338,11 +340,9 @@ class Portal { std::move(on_content_object_callback)); pending_interest->setOnTimeoutCallback( std::move(on_interest_timeout_callback)); - pending_interest->startCountdown( - portal_details::makeCustomAllocatorHandler( - async_callback_memory_, - std::bind(&Portal<ForwarderInt>::timerHandler, this, - std::placeholders::_1, hash))); + pending_interest->startCountdown(portal_details::makeCustomAllocatorHandler( + async_callback_memory_, std::bind(&Portal<ForwarderInt>::timerHandler, + this, std::placeholders::_1, hash))); pending_interest_hash_table_.emplace( std::make_pair(hash, std::move(pending_interest))); } @@ -398,15 +398,13 @@ class Portal { forwarder_interface_.send(content_object); } - TRANSPORT_ALWAYS_INLINE void stopEventsLoop(bool kill_connection = false) { - if (kill_connection) { - forwarder_interface_.closeConnection(); + TRANSPORT_ALWAYS_INLINE void stopEventsLoop() { + if (!io_service_.stopped()) { + io_service_.dispatch([this]() { + clear(); + io_service_.stop(); + }); } - - io_service_.post([this]() { - clear(); - io_service_.stop(); - }); } TRANSPORT_ALWAYS_INLINE void killConnection() { connector_.close(); } @@ -489,12 +487,10 @@ class Portal { if (interest_ptr->getOnDataCallback() != UNSET_CALLBACK) { interest_ptr->on_content_object_callback_( - std::move(interest_ptr->getInterest()), - std::move(content_object)); + std::move(interest_ptr->getInterest()), std::move(content_object)); } else if (consumer_callback_) { consumer_callback_->onContentObject( - std::move(interest_ptr->getInterest()), - std::move(content_object)); + std::move(interest_ptr->getInterest()), std::move(content_object)); } } else { TRANSPORT_LOGW("No pending interests for current content (%s)", diff --git a/libtransport/src/hicn/transport/core/vpp_binary_api.c b/libtransport/src/hicn/transport/core/vpp_binary_api.c index 09ffb2ec6..d54ef257e 100644 --- a/libtransport/src/hicn/transport/core/vpp_binary_api.c +++ b/libtransport/src/hicn/transport/core/vpp_binary_api.c @@ -173,7 +173,7 @@ void vpp_binary_api_send_receive_ping(vpp_binary_api_t *api) { CONTEXT_SAVE(context_store, api, mp_ping); - TRANSPORT_LOGI("Sending ping id %u", mp_ping->_vl_msg_id); + TRANSPORT_LOGD("Sending ping id %u", mp_ping->_vl_msg_id); vpp_binary_api_send_request_wait_reply(api, mp_ping); } @@ -202,7 +202,7 @@ int vpp_binary_api_set_int_state(vpp_binary_api_t *api, uint32_t sw_index, CONTEXT_SAVE(context_store, api, mp); - TRANSPORT_LOGI("Sending set int flags id %u", mp->_vl_msg_id); + TRANSPORT_LOGD("Sending set int flags id %u", mp->_vl_msg_id); return vpp_binary_api_send_request_wait_reply(api, mp); } @@ -211,7 +211,7 @@ void vpp_binary_api_send_request(vpp_binary_api_t *api, void *request) { vl_generic_request_t *req = NULL; req = (vl_generic_request_t *)request; - TRANSPORT_LOGI("Sending a request to VPP (id=%d).\n", ntohs(req->_vl_msg_id)); + TRANSPORT_LOGD("Sending a request to VPP (id=%d).\n", ntohs(req->_vl_msg_id)); S(api, req); } diff --git a/libtransport/src/hicn/transport/core/vpp_forwarder_interface.cc b/libtransport/src/hicn/transport/core/vpp_forwarder_interface.cc index 0d622f9a3..303b753be 100644 --- a/libtransport/src/hicn/transport/core/vpp_forwarder_interface.cc +++ b/libtransport/src/hicn/transport/core/vpp_forwarder_interface.cc @@ -86,12 +86,10 @@ uint32_t VPPForwarderInterface::getMemifConfiguration() { void VPPForwarderInterface::consumerConnection() { hicn_consumer_input_params input = {0}; - hicn_consumer_output_params output; + hicn_consumer_output_params output = {0}; ip_address_t ip4_address; ip_address_t ip6_address; - std::memset(&output, 0, sizeof(hicn_consumer_output_params)); - output.src4 = &ip4_address; output.src6 = &ip6_address; @@ -120,7 +118,7 @@ void VPPForwarderInterface::producerConnection() { void VPPForwarderInterface::connect(bool is_consumer) { std::lock_guard<std::mutex> connection_lock(global_lock_); - srand(time(NULL)); + srand(time(nullptr)); int secret = rand() % (1 << 10); std::stringstream app_name; app_name << "Libtransport_" << secret; @@ -209,18 +207,17 @@ void VPPForwarderInterface::registerRoute(Prefix &prefix) { void VPPForwarderInterface::closeConnection() { if (VPPForwarderInterface::api_) { + connector_.close(); + if (sw_if_index_ != uint32_t(~0)) { int ret = memif_binary_api_delete_memif(VPPForwarderInterface::memif_api_, sw_if_index_); - if (ret < 0) { TRANSPORT_LOGE("Error deleting memif with sw idx %u.", sw_if_index_); } } vpp_binary_api_destroy(VPPForwarderInterface::api_); - connector_.close(); - VPPForwarderInterface::api_ = nullptr; } } diff --git a/libtransport/src/hicn/transport/interfaces/full_duplex_socket.h b/libtransport/src/hicn/transport/interfaces/full_duplex_socket.h index b47432460..1d7ad3cb1 100644 --- a/libtransport/src/hicn/transport/interfaces/full_duplex_socket.h +++ b/libtransport/src/hicn/transport/interfaces/full_duplex_socket.h @@ -104,9 +104,7 @@ class AsyncFullDuplexSocket : public AsyncSocket, AsyncFullDuplexSocket(const Prefix &locator, asio::io_service &io_service); AsyncFullDuplexSocket(const core::Prefix &locator); - ~AsyncFullDuplexSocket() { - TRANSPORT_LOGI("Adios AsyncFullDuplexSocket!!!"); - }; + ~AsyncFullDuplexSocket(){}; using ReadCallback = AsyncReader::ReadCallback; using WriteCallback = AsyncWriter::WriteCallback; diff --git a/libtransport/src/hicn/transport/interfaces/socket_consumer.cc b/libtransport/src/hicn/transport/interfaces/socket_consumer.cc index febe66853..ca9722849 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_consumer.cc +++ b/libtransport/src/hicn/transport/interfaces/socket_consumer.cc @@ -62,25 +62,21 @@ ConsumerSocket::ConsumerSocket(int protocol, asio::io_service &io_service) timer_interval_milliseconds_(0) { switch (protocol) { case TransportProtocolAlgorithms::CBR: - transport_protocol_ = std::make_shared<CbrTransportProtocol>(this); + transport_protocol_ = std::make_unique<CbrTransportProtocol>(this); break; case TransportProtocolAlgorithms::RTC: - transport_protocol_ = std::make_shared<RTCTransportProtocol>(this); + transport_protocol_ = std::make_unique<RTCTransportProtocol>(this); break; case TransportProtocolAlgorithms::RAAQM: default: - transport_protocol_ = std::make_shared<RaaqmTransportProtocol>(this); + transport_protocol_ = std::make_unique<RaaqmTransportProtocol>(this); break; } } ConsumerSocket::~ConsumerSocket() { stop(); - async_downloader_.stop(); - - transport_protocol_.reset(); - portal_.reset(); } void ConsumerSocket::connect() { portal_->connect(); } @@ -132,10 +128,9 @@ void ConsumerSocket::asyncSendInterest(Interest::Ptr &&interest, void ConsumerSocket::stop() { if (transport_protocol_->isRunning()) { + std::cout << "Stopping transport protocol " << std::endl; transport_protocol_->stop(); } - - // is_running_ = false; } void ConsumerSocket::resume() { diff --git a/libtransport/src/hicn/transport/interfaces/socket_consumer.h b/libtransport/src/hicn/transport/interfaces/socket_consumer.h index 0f1ad537c..a50aeb583 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_consumer.h +++ b/libtransport/src/hicn/transport/interfaces/socket_consumer.h @@ -676,7 +676,7 @@ class ConsumerSocket : public BaseSocket { } protected: - std::shared_ptr<TransportProtocol> transport_protocol_; + std::unique_ptr<TransportProtocol> transport_protocol_; private: // context inner state variables diff --git a/libtransport/src/hicn/transport/interfaces/socket_producer.cc b/libtransport/src/hicn/transport/interfaces/socket_producer.cc index 497c40c99..d89fc9367 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_producer.cc +++ b/libtransport/src/hicn/transport/interfaces/socket_producer.cc @@ -15,6 +15,8 @@ #include <hicn/transport/interfaces/socket_producer.h> +#include <functional> + namespace transport { namespace interface { @@ -30,15 +32,12 @@ ProducerSocket::ProducerSocket(asio::io_service &io_service) data_packet_size_(default_values::content_object_packet_size), content_object_expiry_time_(default_values::content_object_expiry_time), output_buffer_(default_values::producer_socket_output_buffer_size), - async_thread_(), registration_status_(REGISTRATION_NOT_ATTEMPTED), making_manifest_(false), signature_type_(SHA_256), hash_algorithm_(HashAlgorithm::SHA_256), input_buffer_capacity_(default_values::producer_socket_input_buffer_size), input_buffer_size_(0), - processing_thread_stop_(false), - listening_thread_stop_(false), on_interest_input_(VOID_HANDLER), on_interest_dropped_input_buffer_(VOID_HANDLER), on_interest_inserted_input_buffer_(VOID_HANDLER), @@ -49,18 +48,10 @@ ProducerSocket::ProducerSocket(asio::io_service &io_service) on_content_object_in_output_buffer_(VOID_HANDLER), on_content_object_output_(VOID_HANDLER), on_content_object_evicted_from_output_buffer_(VOID_HANDLER), - on_content_produced_(VOID_HANDLER) { - listening_thread_stop_ = false; -} + on_content_produced_(VOID_HANDLER) {} ProducerSocket::~ProducerSocket() { - processing_thread_stop_ = true; - portal_->stopEventsLoop(true); - - if (processing_thread_.joinable()) { - processing_thread_.join(); - } - + stop(); if (listening_thread_.joinable()) { listening_thread_.join(); } @@ -308,8 +299,8 @@ uint32_t ProducerSocket::produce(Name content_name, const uint8_t *buf, if (is_last_manifest) { manifest->setFinalManifest(is_last_manifest); } + manifest->encode(); - // Time t0 = std::chrono::steady_clock::now(); identity_->getSigner().sign(*manifest); passContentObjectToCallbacks(manifest); } @@ -324,32 +315,20 @@ uint32_t ProducerSocket::produce(Name content_name, const uint8_t *buf, void ProducerSocket::asyncProduce(ContentObject &content_object) { if (!async_thread_.stopped()) { - // async_thread_.add(std::bind(&ProducerSocket::produce, this, - // content_object)); + auto co_ptr = std::static_pointer_cast<ContentObject>( + content_object.shared_from_this()); + async_thread_.add([this, content_object = std::move(co_ptr)]() { + produce(*content_object); + }); } } -// void ProducerSocket::asyncProduce(const Name &suffix, -// const uint8_t *buf, -// size_t buffer_size, -// AsyncProduceCallback && handler) { -// if (!async_thread_.stopped()) { -// async_thread_.add([this, buffer = buf, size = buffer_size, cb = -// std::move(handler)] () { -// uint64_t bytes_written = produce(suff, buffer, size, 0, false); -// auto ec = std::make_errc(0); -// cb(*this, ec, bytes_written); -// }); -// } -// } - void ProducerSocket::asyncProduce(const Name &suffix, const uint8_t *buf, size_t buffer_size) { if (!async_thread_.stopped()) { - async_thread_.add( - [this, suff = suffix, buffer = buf, size = buffer_size]() { - produce(suff, buffer, size, true); - }); + async_thread_.add([this, suffix, buffer = buf, size = buffer_size]() { + produce(suffix, buffer, size, 0, false); + }); } } diff --git a/libtransport/src/hicn/transport/interfaces/socket_producer.h b/libtransport/src/hicn/transport/interfaces/socket_producer.h index 4f38fb30e..1fdbabe2e 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_producer.h +++ b/libtransport/src/hicn/transport/interfaces/socket_producer.h @@ -525,6 +525,10 @@ class ProducerSocket : public Socket<BasePortal>, return SOCKET_OPTION_GET; } + private: + // Threads + std::thread listening_thread_; + protected: asio::io_service internal_io_service_; asio::io_service &io_service_; @@ -538,7 +542,6 @@ class ProducerSocket : public Socket<BasePortal>, private: utils::EventThread async_thread_; - int registration_status_; bool making_manifest_; @@ -560,12 +563,6 @@ class ProducerSocket : public Socket<BasePortal>, std::atomic_size_t input_buffer_capacity_; std::atomic_size_t input_buffer_size_; - // threads - std::thread listening_thread_; - std::thread processing_thread_; - volatile bool processing_thread_stop_; - volatile bool listening_thread_stop_; - // callbacks protected: ProducerInterestCallback on_interest_input_; diff --git a/libtransport/src/hicn/transport/protocols/protocol.h b/libtransport/src/hicn/transport/protocols/protocol.h index 6a3b23753..6911eada5 100644 --- a/libtransport/src/hicn/transport/protocols/protocol.h +++ b/libtransport/src/hicn/transport/protocols/protocol.h @@ -39,7 +39,7 @@ class TransportProtocol : public interface::BasePortal::ConsumerCallback, public: TransportProtocol(interface::ConsumerSocket *icn_socket); - virtual ~TransportProtocol() { stop(); }; + virtual ~TransportProtocol() = default; TRANSPORT_ALWAYS_INLINE bool isRunning() { return is_running_; } diff --git a/libtransport/src/hicn/transport/utils/epoll_event_reactor.cc b/libtransport/src/hicn/transport/utils/epoll_event_reactor.cc index 6df9e5656..a97e89500 100644 --- a/libtransport/src/hicn/transport/utils/epoll_event_reactor.cc +++ b/libtransport/src/hicn/transport/utils/epoll_event_reactor.cc @@ -48,30 +48,6 @@ int EpollEventReactor::addFileDescriptor(int fd, uint32_t events) { return 0; } -int EpollEventReactor::addFileDescriptor(int fd, uint32_t events, - EventCallback &callback) { - auto it = event_callback_map_.find(fd); - - if (it == event_callback_map_.end()) { - event_callback_map_[fd] = callback; - return addFileDescriptor(fd, events); - } - - return 0; -} - -int EpollEventReactor::addFileDescriptor(int fd, uint32_t events, - EventCallback &&callback) { - auto it = event_callback_map_.find(fd); - - if (it == event_callback_map_.end()) { - event_callback_map_[fd] = callback; - return addFileDescriptor(fd, events); - } - - return 0; -} - int EpollEventReactor::modFileDescriptor(int fd, uint32_t events) { if (TRANSPORT_EXPECT_FALSE(fd < 0)) { TRANSPORT_LOGE("invalid fd %d", fd); @@ -109,6 +85,7 @@ int EpollEventReactor::delFileDescriptor(int fd) { return -1; } + utils::SpinLock::Acquire locked(event_callback_map_lock_); event_callback_map_.erase(fd); return 0; @@ -117,6 +94,8 @@ int EpollEventReactor::delFileDescriptor(int fd) { void EpollEventReactor::runEventLoop(int timeout) { Event evt[128]; int en = 0; + EventCallbackMap::iterator it; + EventCallback callback; // evt.events = EPOLLIN | EPOLLOUT; sigset_t sigset; @@ -134,11 +113,26 @@ void EpollEventReactor::runEventLoop(int timeout) { for (int i = 0; i < en; i++) { if (evt[i].data.fd > 0) { - auto it = event_callback_map_.find(evt[i].data.fd); + { + utils::SpinLock::Acquire locked(event_callback_map_lock_); + it = event_callback_map_.find(evt[i].data.fd); + } + if (TRANSPORT_EXPECT_FALSE(it == event_callback_map_.end())) { TRANSPORT_LOGE("unexpected event. fd %d", evt[i].data.fd); } else { - event_callback_map_[evt[i].data.fd](evt[i]); + { + utils::SpinLock::Acquire locked(event_callback_map_lock_); + callback = event_callback_map_[evt[i].data.fd]; + } + + callback(evt[i]); + + // In the callback the epoll event reactor could have been stopped, + // then we need to check whether the event loop is still running. + if (TRANSPORT_EXPECT_FALSE(!run_event_loop_)) { + return; + } } } else { TRANSPORT_LOGE("unexpected event. fd %d", evt[i].data.fd); @@ -150,6 +144,8 @@ void EpollEventReactor::runEventLoop(int timeout) { void EpollEventReactor::runOneEvent() { Event evt; int en = 0; + EventCallbackMap::iterator it; + EventCallback callback; // evt.events = EPOLLIN | EPOLLOUT; sigset_t sigset; @@ -165,11 +161,20 @@ void EpollEventReactor::runOneEvent() { } if (TRANSPORT_EXPECT_TRUE(evt.data.fd > 0)) { - auto it = event_callback_map_.find(evt.data.fd); + { + utils::SpinLock::Acquire locked(event_callback_map_lock_); + it = event_callback_map_.find(evt.data.fd); + } + if (TRANSPORT_EXPECT_FALSE(it == event_callback_map_.end())) { TRANSPORT_LOGE("unexpected event. fd %d", evt.data.fd); } else { - event_callback_map_[evt.data.fd](evt); + { + utils::SpinLock::Acquire locked(event_callback_map_lock_); + callback = event_callback_map_[evt.data.fd]; + } + + callback(evt); } } else { TRANSPORT_LOGE("unexpected event. fd %d", evt.data.fd); diff --git a/libtransport/src/hicn/transport/utils/epoll_event_reactor.h b/libtransport/src/hicn/transport/utils/epoll_event_reactor.h index dbb87c6c5..04c10fc7e 100644 --- a/libtransport/src/hicn/transport/utils/epoll_event_reactor.h +++ b/libtransport/src/hicn/transport/utils/epoll_event_reactor.h @@ -16,6 +16,7 @@ #pragma once #include <hicn/transport/utils/event_reactor.h> +#include <hicn/transport/utils/spinlock.h> #include <sys/epoll.h> #include <atomic> @@ -38,9 +39,22 @@ class EpollEventReactor : public EventReactor { ~EpollEventReactor(); - int addFileDescriptor(int fd, uint32_t events, EventCallback &callback); + template <typename EventHandler> + int addFileDescriptor(int fd, uint32_t events, EventHandler &&callback) { + auto it = event_callback_map_.find(fd); + int ret = 0; - int addFileDescriptor(int fd, uint32_t events, EventCallback &&callback); + if (it == event_callback_map_.end()) { + { + utils::SpinLock::Acquire locked(event_callback_map_lock_); + event_callback_map_[fd] = std::forward<EventHandler &&>(callback); + } + + ret = addFileDescriptor(fd, events); + } + + return ret; + } int delFileDescriptor(int fd); @@ -60,7 +74,7 @@ class EpollEventReactor : public EventReactor { int epoll_fd_; std::atomic_bool run_event_loop_; EventCallbackMap event_callback_map_; - std::mutex event_callback_map_mutex_; + utils::SpinLock event_callback_map_lock_; }; } // namespace utils |