Plugging out Ethernet cable crash library; if using TLS #353
Comments
|
Thank you for reporting the issue. I couldn't reproduce the issue on my environment. But my environment something different from yours because it is difficult to prepare plugging out environment. I use three terminals.
Concept socat outputbroker outputclient_cli outputWhen I kill socat, then I got the following result. brokerclient_cliWhich version are you using? If you using async_mqtt from github clone, commit hash is helpful. Anyway, I will post my code analysis comment later. |
|
I added log outputting code to the following part: async_mqtt/include/async_mqtt/util/impl/stream_close.hpp Lines 29 to 100 in 0b12246 This is the log added code: template <typename Self>
void operator()(
Self& self,
error_code ec = error_code{}
) {
auto& a_strm{*strm};
if (state == dispatch) {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "async operation start. state: dispatch -> close";
state = close;
as::dispatch(
a_strm.get_executor(),
as::append(
force_move(self),
error_code{},
std::ref(a_strm.nl_)
)
);
}
else {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "async operation finish. state: complete";
BOOST_ASSERT(state == complete);
a_strm.storing_cbs_.clear();
a_strm.sending_cbs_.clear();
self.complete(ec);
}
}
template <typename Self, typename Layer>
void operator()(
Self& self,
error_code /* ec */,
std::reference_wrapper<Layer> stream
) {
auto& a_strm{*strm};
BOOST_ASSERT(state == close);
if constexpr(has_async_close<Layer>::value) {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "has_async_close";
if constexpr (has_next_layer<Layer>::value) {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "call async_close";
layer_customize<Layer>::async_close(
stream.get(),
as::append(
force_move(self),
std::ref(stream.get().next_layer())
)
);
}
else {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "NOT has_next_layer (lowest layer (TCP)). call async_close. "
<< "state: close -> complete";
state = complete;
layer_customize<Layer>::async_close(
stream.get(),
force_move(self)
);
}
}
else {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "NOT has_async_close";
if constexpr (has_next_layer<Layer>::value) {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "skip this layer";
as::dispatch(
a_strm.get_executor(),
as::append(
force_move(self),
error_code{},
std::ref(a_strm.nl_)
)
);
}
else {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "NOT has_next_layer (lowest layer (TCP)) state: close -> complete";
state = complete;
as::dispatch(
a_strm.get_executor(),
force_move(self)
);
}
}
}Then I got the following result by the same operation (socat kill) This is the expected log outputs. If you replaced this part of the code with my log added version and cheking the log, it is very helpful to analyze the issue. |
|
We checkout the 9.0.0 tag. |
|
And this is from main(0b12246): |
|
tag 9.0.0 os https://github.com/redboltz/async_mqtt/blob/9.0.0/include/async_mqtt/util/impl/stream_close.hpp It seems that the line number is something weird. Here is 9.0.0 with log code from the line 1. // Copyright Takatoshi Kondo 2022
//
// Distributed under the Boost Software License, Version 1.0.
// (See accompanying file LICENSE_1_0.txt or copy at
// http://www.boost.org/LICENSE_1_0.txt)
#if !defined(ASYNC_MQTT_UTIL_IMPL_STREAM_CLOSE_HPP)
#define ASYNC_MQTT_UTIL_IMPL_STREAM_CLOSE_HPP
#include <async_mqtt/util/stream.hpp>
#include <async_mqtt/protocol_version.hpp>
namespace async_mqtt {
namespace detail {
template <typename NextLayer>
struct stream_impl<NextLayer>::stream_close_op {
using stream_type = this_type;
using stream_type_sp = std::shared_ptr<stream_type>;
std::shared_ptr<stream_type> strm;
enum {
dispatch,
close,
complete
} state = dispatch;
template <typename Self>
void operator()(
Self& self,
error_code ec = error_code{}
) {
auto& a_strm{*strm};
if (state == dispatch) {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "async operation start. state: dispatch -> close";
state = close;
as::dispatch(
a_strm.get_executor(),
as::append(
force_move(self),
error_code{},
std::ref(a_strm.nl_)
)
);
}
else {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "async operation finish. state: complete";
BOOST_ASSERT(state == complete);
a_strm.storing_cbs_.clear();
a_strm.sending_cbs_.clear();
self.complete(ec);
}
}
template <typename Self, typename Layer>
void operator()(
Self& self,
error_code /* ec */,
std::reference_wrapper<Layer> stream
) {
auto& a_strm{*strm};
BOOST_ASSERT(state == close);
if constexpr(has_async_close<Layer>::value) {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "has_async_close";
if constexpr (has_next_layer<Layer>::value) {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "call async_close";
layer_customize<Layer>::async_close(
stream.get(),
as::append(
force_move(self),
std::ref(stream.get().next_layer())
)
);
}
else {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "NOT has_next_layer (lowest layer (TCP)). call async_close. "
<< "state: close -> complete";
state = complete;
layer_customize<Layer>::async_close(
stream.get(),
force_move(self)
);
}
}
else {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "NOT has_async_close";
if constexpr (has_next_layer<Layer>::value) {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "skip this layer";
as::dispatch(
a_strm.get_executor(),
as::append(
force_move(self),
error_code{},
std::ref(a_strm.nl_)
)
);
}
else {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "NOT has_next_layer (lowest layer (TCP)) state: close -> complete";
state = complete;
as::dispatch(
a_strm.get_executor(),
force_move(self)
);
}
}
}
}; |
is a blank line. No logs added before the line 28. It is something weird too. |
|
Sorry, the problem is that I have automatic code formatting. // Copyright Takatoshi Kondo 2022
//
// Distributed under the Boost Software License, Version 1.0.
// (See accompanying file LICENSE_1_0.txt or copy at
// http://www.boost.org/LICENSE_1_0.txt)
#if !defined(ASYNC_MQTT_UTIL_IMPL_STREAM_CLOSE_HPP)
#define ASYNC_MQTT_UTIL_IMPL_STREAM_CLOSE_HPP
#include <async_mqtt/protocol_version.hpp>
#include <async_mqtt/util/stream.hpp>
namespace async_mqtt {
namespace detail {
template <typename NextLayer> struct stream_impl<NextLayer>::stream_close_op {
using stream_type = this_type;
using stream_type_sp = std::shared_ptr<stream_type>;
std::shared_ptr<stream_type> strm;
enum { dispatch, close, complete } state = dispatch;
template <typename Self>
void operator()(Self &self, error_code ec = error_code{}) {
auto &a_strm{*strm};
if (state == dispatch) {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "async operation start. state: dispatch -> close";
state = close;
as::dispatch(
a_strm.get_executor(),
as::append(force_move(self), error_code{}, std::ref(a_strm.nl_)));
} else {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "async operation finish. state: complete";
BOOST_ASSERT(state == complete);
a_strm.storing_cbs_.clear();
a_strm.sending_cbs_.clear();
self.complete(ec);
}
}
template <typename Self, typename Layer>
void operator()(Self &self, error_code /* ec */,
std::reference_wrapper<Layer> stream) {
auto &a_strm{*strm};
BOOST_ASSERT(state == close);
if constexpr (has_async_close<Layer>::value) {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get()) << "has_async_close";
if constexpr (has_next_layer<Layer>::value) {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get()) << "call async_close";
layer_customize<Layer>::async_close(
stream.get(),
as::append(force_move(self), std::ref(stream.get().next_layer())));
} else {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "NOT has_next_layer (lowest layer (TCP)). call async_close. "
<< "state: close -> complete";
state = complete;
layer_customize<Layer>::async_close(stream.get(), force_move(self));
}
} else {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get()) << "NOT has_async_close";
if constexpr (has_next_layer<Layer>::value) {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get()) << "skip this layer";
as::dispatch(
a_strm.get_executor(),
as::append(force_move(self), error_code{}, std::ref(a_strm.nl_)));
} else {
ASYNC_MQTT_LOG("mqtt_impl", fatal)
<< ASYNC_MQTT_ADD_VALUE(address, strm.get())
<< "NOT has_next_layer (lowest layer (TCP)) state: close -> "
"complete";
state = complete;
as::dispatch(a_strm.get_executor(), force_move(self));
}
}
}
};
} // namespace detail
template <typename NextLayer>
template <typename CompletionToken>
BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(CompletionToken, void())
stream<NextLayer>::async_close(CompletionToken &&token) {
BOOST_ASSERT(impl_);
return as::async_compose<CompletionToken, void(error_code)>(
typename impl_type::stream_close_op{impl_}, token, get_executor());
}
} // namespace async_mqtt
#endif // ASYNC_MQTT_UTIL_IMPL_STREAM_CLOSE_HPP` |
|
Please use ```cpp notation. |
|
Please keep the original code. Auto code fomatting confused me. |
|
I edited your comment only for applying ```. It seems that the async close sequence itself works expectedly. |
|
Okay, I changed the file using vim instead. Posting the output that should match your numbering. |
|
By the way, I prepared trial MQTT broker that support TCP, TLS, ws, and wss. When you unplugged your client machine, what is happened? On my environment, no disconnection is happened (I wait 1 minutes). I guess that It depends on TCP setting. |
|
You mentioned that you used degger, and at the line 67, assertion failed line, you checked the variable I think that it helps to know where the unexpected call is come from. Thank you very much for your cooperation. |
|
By the way which version of Boost are you using ? |
|
I'm using 1.84. I will drive to my company now and see if I can try to connect to your server using an ethernet cable. |
|
#354 has been merged. Now, client_cli log works fine. |
Please re-run using updated |
|
I create #355,. I am not 100% sure but It would fix this issue. |
|
Here is the output. I had to set keep-alive in order to crash it using your server: |
|
Thank you for reporitng. I think that the following line is the reason of the problematic behavior. If my hypothetis is right, the issue should be solved by #355. Please try it. |
|
To your server, it seems to work! I will test my original setup as well, but here is the output connecting to your server: |
|
I tested my original setup and it works as well! Thank you for the quick response! |
|
Finally I reproduced the issue. Here is the sequence using one phisical machine.
Important point is kill broker not socat. Making the situation that I've tested as follows:
I will release the fix as 9.0.1 soon. |
|
#355 has been merged, and version 9.0.1 has been released. Thank you for all your hard work; it was truly appreciated and valuable. |
|
No problem! Thank you for resolving the issue so quickly! |
Merged
|
#357 is merged. It is a better fix for this issue. And fix other TLS close problem. |
|
It works with my setup as well: |
|
Thank you for checking. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Hello,
if using TLS; there is an assert that crash the library if the Ethernet cable is plugged out between the client and broker.
This is the output from client_cli:
Debug build:
Release build:
, using a debugger shows that the state is "complete" and therefor the program crash.
Doing the same test using TCP works without problem.
For now we have created a patch that replace the assert with an if statement. Then the library doesn't crash in either Debug or Release builds. This fix may of course have other impacts that we haven't seen yet.
Regards,
Christian
The text was updated successfully, but these errors were encountered: