Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

RXI-1122 fix history mode after reconnect #110

Merged
merged 1 commit into from
Jun 5, 2024
Merged

RXI-1122 fix history mode after reconnect #110

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 11 additions & 6 deletions src/xbwd/client/ChainListener.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,12 +131,17 @@ ChainListener::onConnect()

// Clear on re-connect
inRequest_ = false;
ledgerReqMax_ = 0;
ledgerProcessedDoor_ = 0;
ledgerProcessedSubmit_ = 0;
prevLedgerIndex_ = 0;
txnHistoryIndex_ = 0;
hp_.clear();

// Resume only if history finished
if (hp_.state_ != HistoryProcessor::FINISHED)
{
ledgerReqMax_ = 0;
ledgerProcessedDoor_ = 0;
ledgerProcessedSubmit_ = 0;
prevLedgerIndex_ = 0;
txnHistoryIndex_ = 0;
hp_.clear();
}

if (signAccount_)
{
Expand Down
114 changes: 72 additions & 42 deletions src/xbwd/client/WebsocketClient.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ WebsocketClient::cleanup()
{
ios_.post(strand_.wrap([this] {
timer_.cancel();
if (!peerClosed_)
if (state_ == ST_CONNECTED)
{
{
std::lock_guard l{m_};
Expand All @@ -66,15 +66,15 @@ WebsocketClient::cleanup()

std::lock_guard l(shutdownM_);
timer_.cancel();
isShutdown_ = true;
state_ = ST_SHUTDOWN;
shutdownCv_.notify_one();
}));
}
}
else
{
std::lock_guard<std::mutex> l(shutdownM_);
isShutdown_ = true;
state_ = ST_SHUTDOWN;
shutdownCv_.notify_one();
}
}));
Expand All @@ -83,12 +83,12 @@ WebsocketClient::cleanup()
void
WebsocketClient::shutdown()
{
if (isShutdown_)
if (state_ == ST_SHUTDOWN)
return;
cleanup();
std::unique_lock l{shutdownM_};
if (!isShutdown_)
shutdownCv_.wait(l, [this] { return isShutdown_.load(); });
if (state_ != ST_SHUTDOWN)
shutdownCv_.wait(l, [this] { return state_ == ST_SHUTDOWN; });
}

WebsocketClient::WebsocketClient(
Expand Down Expand Up @@ -122,34 +122,40 @@ WebsocketClient::~WebsocketClient()
void
WebsocketClient::connect()
{
std::lock_guard<std::mutex> l(shutdownM_);
if (isShutdown_)
return;

try
{
rb_.clear();
// TODO: Change all the beast::IP:Endpoints to boost endpoints
stream_.connect(ep_);
peerClosed_ = false;
ws_.set_option(boost::beast::websocket::stream_base::decorator(
[&](boost::beast::websocket::request_type& req) {
for (auto const& h : headers_)
req.set(h.first, h.second);
}));
ws_.handshake(
ep_.address().to_string() + ":" + std::to_string(ep_.port()), "/");
{
std::lock_guard<std::mutex> ls(shutdownM_);
if (state_ == ST_SHUTDOWN)
return;

std::lock_guard lw{m_};

rb_.clear();
// TODO: Change all the beast::IP:Endpoints to boost endpoints
stream_.connect(ep_);
ws_.set_option(boost::beast::websocket::stream_base::decorator(
[&](boost::beast::websocket::request_type& req) {
for (auto const& h : headers_)
req.set(h.first, h.second);
}));
ws_.handshake(
ep_.address().to_string() + ":" + std::to_string(ep_.port()),
"/");
state_ = ST_CONNECTED;

JLOGV(
j_.info(),
"WebsocketClient connected to",
jv("ip", ep_.address()),
jv("port", ep_.port()));
JLOGV(
j_.info(),
"WebsocketClient connected to",
jv("ip", ep_.address()),
jv("port", ep_.port()));

ws_.async_read(
rb_,
std::bind(
&WebsocketClient::onReadMsg, this, std::placeholders::_1));
}

ws_.async_read(
rb_,
std::bind(
&WebsocketClient::onReadMsg, this, std::placeholders::_1));
onConnectCallback_();
}
catch (std::exception& e)
Expand All @@ -170,6 +176,21 @@ WebsocketClient::send(
Json::Value params,
std::string const& chain)
{
{
std::lock_guard<std::mutex> l(shutdownM_);
if (state_ != ST_CONNECTED)
{
// Attestations will be re-send after TTL, everything else - after
// reconnect
JLOGV(
j_.trace(),
"WebsocketClient::send",
jv("chainType", chain),
jv("error", "not connected"));
return 0;
}
}

params[ripple::jss::method] = cmd;
params[ripple::jss::jsonrpc] = "2.0";
params[ripple::jss::ripplerpc] = "2.0";
Expand All @@ -189,7 +210,6 @@ WebsocketClient::send(
}
catch (...)
{
std::lock_guard<std::mutex> l(shutdownM_);
reconnect("exception at sending data");
}
return id;
Expand All @@ -200,18 +220,19 @@ WebsocketClient::onReadMsg(error_code const& ec)
{
if (ec)
{
auto const& reason = ws_.reason();
boost::beast::websocket::close_reason reason;
{
std::lock_guard l{m_};
reason = ws_.reason();
}

JLOGV(
j_.error(),
"WebsocketClient::onReadMsg error",
jv("ec", ec),
jv("code", reason.code),
jv("msg", reason.reason));
if (ec == boost::beast::websocket::error::closed)
peerClosed_ = true;

std::lock_guard<std::mutex> l(shutdownM_);
reconnect("error reading data");
return;
}
Expand All @@ -222,17 +243,26 @@ WebsocketClient::onReadMsg(error_code const& ec)
messageCv_.notify_one();
}

rb_.clear();
ws_.async_read(
rb_,
std::bind(&WebsocketClient::onReadMsg, this, std::placeholders::_1));
{
std::lock_guard l{m_};
if (state_ != ST_CONNECTED)
return;
rb_.clear();
ws_.async_read(
rb_,
std::bind(
&WebsocketClient::onReadMsg, this, std::placeholders::_1));
}
}

void
WebsocketClient::reconnect(std::string_view reason)
{
if (isShutdown_)
std::lock_guard<std::mutex> l(shutdownM_);

if (state_ == ST_SHUTDOWN)
return;
state_ = ST_INIT;

JLOGV(j_.info(), "WebsocketClient::reconnect()", jv("reason", reason));

Expand All @@ -257,7 +287,7 @@ WebsocketClient::runCallbacks()
{
std::uint64_t maxSize = 0;

for (; !isShutdown_;)
for (; state_ != ST_SHUTDOWN;)
{
processingQueue_.clear();
{
Expand All @@ -280,7 +310,7 @@ WebsocketClient::runCallbacks()

for (auto const& rb : processingQueue_)
{
if (isShutdown_)
if (state_ == ST_SHUTDOWN)
break;

auto const s = buffer_string(rb.data());
Expand Down
7 changes: 3 additions & 4 deletions src/xbwd/client/WebsocketClient.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,8 @@ class WebsocketClient

// mutex for shutdown
std::mutex shutdownM_;
std::atomic_bool isShutdown_ = false;
enum estate : std::int32_t { ST_INIT, ST_CONNECTED, ST_SHUTDOWN };
std::atomic_int32_t state_ = ST_INIT;
std::condition_variable shutdownCv_;

boost::asio::io_service& ios_;
Expand All @@ -65,8 +66,6 @@ class WebsocketClient
m_) ws_;
boost::beast::multi_buffer rb_;

std::atomic_bool peerClosed_{true};

std::function<void(Json::Value const&)> onMessageCallback_;
std::atomic_uint32_t nextId_{0};

Expand Down Expand Up @@ -111,7 +110,7 @@ class WebsocketClient
shutdown() EXCLUDES(shutdownM_);

void
reconnect(std::string_view reason) REQUIRES(shutdownM_);
reconnect(std::string_view reason) EXCLUDES(shutdownM_);

private:
void
Expand Down