aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--libtransport/src/hicn/transport/interfaces/socket_consumer.cc31
-rw-r--r--libtransport/src/hicn/transport/interfaces/socket_consumer.h29
-rw-r--r--libtransport/src/hicn/transport/interfaces/socket_producer.cc64
-rw-r--r--libtransport/src/hicn/transport/interfaces/socket_producer.h41
-rw-r--r--libtransport/src/hicn/transport/protocols/raaqm.cc2
5 files changed, 61 insertions, 106 deletions
diff --git a/libtransport/src/hicn/transport/interfaces/socket_consumer.cc b/libtransport/src/hicn/transport/interfaces/socket_consumer.cc
index 14cd27b6b..64b60101d 100644
--- a/libtransport/src/hicn/transport/interfaces/socket_consumer.cc
+++ b/libtransport/src/hicn/transport/interfaces/socket_consumer.cc
@@ -121,37 +121,6 @@ asio::io_service &ConsumerSocket::getIoService() {
return portal_->getIoService();
}
-// If the thread calling lambda_func is not the same of io_service, this
-// function reschedule the function on it
-template <typename Lambda, typename arg2>
-int ConsumerSocket::rescheduleOnIOService(int socket_option_key,
- arg2 socket_option_value,
- Lambda lambda_func) {
- // To enforce type check
- std::function<int(int, arg2)> func = lambda_func;
- int result = SOCKET_OPTION_SET;
- if (transport_protocol_->isRunning()) {
- std::mutex mtx;
- /* Condition variable for the wait */
- std::condition_variable cv;
- bool done = false;
- io_service_.dispatch([&socket_option_key, &socket_option_value, &mtx,
- &result, &done, &func]() {
- std::unique_lock<std::mutex> lck(mtx);
- done = true;
- result = func(socket_option_key, socket_option_value);
- });
- std::unique_lock<std::mutex> lck(mtx);
- if (!done) {
- cv.wait(lck);
- }
- } else {
- result = func(socket_option_key, socket_option_value);
- }
-
- return result;
-}
-
int ConsumerSocket::setSocketOption(int socket_option_key,
ReadCallback *socket_option_value) {
// Reschedule the function on the io_service to avoid race condition in case
diff --git a/libtransport/src/hicn/transport/interfaces/socket_consumer.h b/libtransport/src/hicn/transport/interfaces/socket_consumer.h
index e3620b269..eceee2d34 100644
--- a/libtransport/src/hicn/transport/interfaces/socket_consumer.h
+++ b/libtransport/src/hicn/transport/interfaces/socket_consumer.h
@@ -282,9 +282,36 @@ class ConsumerSocket : public BaseSocket {
ConsumerTimerCallback **socket_option_value);
protected:
+ // If the thread calling lambda_func is not the same of io_service, this
+ // function reschedule the function on it
template <typename Lambda, typename arg2>
int rescheduleOnIOService(int socket_option_key, arg2 socket_option_value,
- Lambda lambda_func);
+ Lambda lambda_func) {
+ // To enforce type check
+ std::function<int(int, arg2)> func = lambda_func;
+ int result = SOCKET_OPTION_SET;
+ if (transport_protocol_->isRunning()) {
+ std::mutex mtx;
+ /* Condition variable for the wait */
+ std::condition_variable cv;
+ bool done = false;
+ io_service_.dispatch([&socket_option_key, &socket_option_value, &mtx, &cv,
+ &result, &done, &func]() {
+ std::unique_lock<std::mutex> lck(mtx);
+ done = true;
+ result = func(socket_option_key, socket_option_value);
+ cv.notify_all();
+ });
+ std::unique_lock<std::mutex> lck(mtx);
+ if (!done) {
+ cv.wait(lck);
+ }
+ } else {
+ result = func(socket_option_key, socket_option_value);
+ }
+
+ return result;
+ }
private:
asio::io_service internal_io_service_;
diff --git a/libtransport/src/hicn/transport/interfaces/socket_producer.cc b/libtransport/src/hicn/transport/interfaces/socket_producer.cc
index f90197490..8f8fc1a79 100644
--- a/libtransport/src/hicn/transport/interfaces/socket_producer.cc
+++ b/libtransport/src/hicn/transport/interfaces/socket_producer.cc
@@ -391,70 +391,6 @@ void ProducerSocket::onInterest(Interest &interest) {
}
}
-// If the thread calling lambda_func is not the same of io_service, this
-// function reschedule the function on it
-template <typename Lambda, typename arg2>
-int ProducerSocket::rescheduleOnIOService(int socket_option_key,
- arg2 socket_option_value,
- Lambda lambda_func) {
- // To enforce type check
- std::function<int(int, arg2)> func = lambda_func;
- int result = SOCKET_OPTION_SET;
- if (listening_thread_.joinable() &&
- std::this_thread::get_id() != listening_thread_.get_id()) {
- std::mutex mtx;
- /* Condition variable for the wait */
- std::condition_variable cv;
- bool done = false;
- io_service_.dispatch([&socket_option_key, &socket_option_value, &mtx,
- &result, &done, &func]() {
- std::unique_lock<std::mutex> lck(mtx);
- done = true;
- result = func(socket_option_key, socket_option_value);
- });
- std::unique_lock<std::mutex> lck(mtx);
- if (!done) {
- cv.wait(lck);
- }
- } else {
- result = func(socket_option_key, socket_option_value);
- }
-
- return result;
-}
-
-// If the thread calling lambda_func is not the same of io_service, this
-// function reschedule the function on it
-template <typename Lambda, typename arg2>
-int ProducerSocket::rescheduleOnIOServiceWithReference(
- int socket_option_key, arg2 &socket_option_value, Lambda lambda_func) {
- // To enforce type check
- std::function<int(int, arg2 &)> func = lambda_func;
- int result = SOCKET_OPTION_SET;
- if (listening_thread_.joinable() &&
- std::this_thread::get_id() != this->listening_thread_.get_id()) {
- std::mutex mtx;
- /* Condition variable for the wait */
- std::condition_variable cv;
- std::unique_lock<std::mutex> lck(mtx);
- bool done = false;
- io_service_.dispatch([this, &socket_option_key, &socket_option_value, &mtx,
- &cv, &result, &done, &func]() {
- std::unique_lock<std::mutex> lck(mtx);
- done = true;
- result = func(socket_option_key, socket_option_value);
-
- if (!done) {
- cv.wait(lck);
- }
- });
- } else {
- result = func(socket_option_key, socket_option_value);
- }
-
- return result;
-}
-
int ProducerSocket::setSocketOption(int socket_option_key,
uint32_t socket_option_value) {
switch (socket_option_key) {
diff --git a/libtransport/src/hicn/transport/interfaces/socket_producer.h b/libtransport/src/hicn/transport/interfaces/socket_producer.h
index 5c617d761..709a2582b 100644
--- a/libtransport/src/hicn/transport/interfaces/socket_producer.h
+++ b/libtransport/src/hicn/transport/interfaces/socket_producer.h
@@ -160,15 +160,6 @@ class ProducerSocket : public Socket<BasePortal>,
virtual int getSocketOption(int socket_option_key,
std::string &socket_option_value);
- template <typename Lambda, typename arg2>
- int rescheduleOnIOService(int socket_option_key, arg2 socket_option_value,
- Lambda lambda_func);
-
- template <typename Lambda, typename arg2>
- int rescheduleOnIOServiceWithReference(int socket_option_key,
- arg2 &socket_option_value,
- Lambda lambda_func);
-
protected:
// Threads
std::thread listening_thread_;
@@ -215,6 +206,38 @@ class ProducerSocket : public Socket<BasePortal>,
ProducerContentCallback on_content_produced_;
+ // If the thread calling lambda_func is not the same of io_service, this
+ // function reschedule the function on it
+ template <typename Lambda, typename arg2>
+ int rescheduleOnIOService(int socket_option_key, arg2 socket_option_value,
+ Lambda lambda_func) {
+ // To enforce type check
+ std::function<int(int, arg2)> func = lambda_func;
+ int result = SOCKET_OPTION_SET;
+ if (listening_thread_.joinable() &&
+ std::this_thread::get_id() != listening_thread_.get_id()) {
+ std::mutex mtx;
+ /* Condition variable for the wait */
+ std::condition_variable cv;
+ bool done = false;
+ io_service_.dispatch([&socket_option_key, &socket_option_value,
+ &mtx, &cv, &result, &done, &func]() {
+ std::unique_lock<std::mutex> lck(mtx);
+ done = true;
+ result = func(socket_option_key, socket_option_value);
+ cv.notify_all();
+ });
+ std::unique_lock<std::mutex> lck(mtx);
+ if (!done) {
+ cv.wait(lck);
+ }
+ } else {
+ result = func(socket_option_key, socket_option_value);
+ }
+
+ return result;
+ }
+
private:
void listen();
diff --git a/libtransport/src/hicn/transport/protocols/raaqm.cc b/libtransport/src/hicn/transport/protocols/raaqm.cc
index c816158f9..779f9a9a1 100644
--- a/libtransport/src/hicn/transport/protocols/raaqm.cc
+++ b/libtransport/src/hicn/transport/protocols/raaqm.cc
@@ -519,7 +519,7 @@ void RaaqmTransportProtocol::onContentReassembled(std::error_code ec) {
interface::ConsumerSocket::ReadCallback *on_payload = VOID_HANDLER;
socket_->getSocketOption(READ_CALLBACK, &on_payload);
- if (on_payload) {
+ if (!on_payload) {
throw errors::RuntimeException(
"The read callback must be installed in the transport before "
"starting "