mirror of
https://codeberg.org/wownero/wownero-lws
synced 2026-01-09 15:15:15 -08:00
Restarting on reorg is broken; fix asio::io_context::restart calls (#177)
This commit is contained in:
Lee *!* Clagett
committed by
Lee *!* Clagett
Lee *!* Clagett
parent
8d854e9bc6
commit
44278d0d11
153
src/scanner.cpp
153
src/scanner.cpp
@@ -960,95 +960,98 @@ namespace lws
|
|||||||
std::vector<std::shared_ptr<rpc::scanner::queue>> queues;
|
std::vector<std::shared_ptr<rpc::scanner::queue>> queues;
|
||||||
queues.resize(thread_count);
|
queues.resize(thread_count);
|
||||||
|
|
||||||
struct join_
|
|
||||||
{
|
{
|
||||||
scanner_sync& self;
|
struct join_
|
||||||
rpc::context& ctx;
|
|
||||||
std::vector<std::shared_ptr<rpc::scanner::queue>>& queues;
|
|
||||||
std::vector<boost::thread>& threads;
|
|
||||||
|
|
||||||
~join_() noexcept
|
|
||||||
{
|
{
|
||||||
self.stop();
|
scanner_sync& self;
|
||||||
if (self.has_shutdown())
|
rpc::context& ctx;
|
||||||
ctx.raise_abort_process();
|
std::vector<std::shared_ptr<rpc::scanner::queue>>& queues;
|
||||||
else
|
std::vector<boost::thread>& threads;
|
||||||
ctx.raise_abort_scan();
|
|
||||||
|
|
||||||
for (const auto& queue : queues)
|
~join_() noexcept
|
||||||
{
|
{
|
||||||
if (queue)
|
self.stop();
|
||||||
queue->stop();
|
if (self.has_shutdown())
|
||||||
|
ctx.raise_abort_process();
|
||||||
|
else
|
||||||
|
ctx.raise_abort_scan();
|
||||||
|
|
||||||
|
for (const auto& queue : queues)
|
||||||
|
{
|
||||||
|
if (queue)
|
||||||
|
queue->stop();
|
||||||
|
}
|
||||||
|
for (auto& thread : threads)
|
||||||
|
thread.join();
|
||||||
}
|
}
|
||||||
for (auto& thread : threads)
|
} join{self, ctx, queues, threads};
|
||||||
thread.join();
|
|
||||||
|
/*
|
||||||
|
The algorithm here is extremely basic. Users are divided evenly amongst
|
||||||
|
the configurable thread count, and grouped by scan height. If an old
|
||||||
|
account appears, some accounts (grouped on that thread) will be delayed
|
||||||
|
in processing waiting for that account to catch up. Its not the greatest,
|
||||||
|
but this "will have to do" - but we're getting closer to fixing that
|
||||||
|
too.
|
||||||
|
|
||||||
|
Another "issue" is that each thread works independently instead of more
|
||||||
|
cooperatively for scanning. This requires a bit more synchronization, so
|
||||||
|
was left for later. Its likely worth doing to reduce the number of
|
||||||
|
transfers from the daemon, and the bottleneck on the writes into LMDB.
|
||||||
|
*/
|
||||||
|
|
||||||
|
self.stop_ = false;
|
||||||
|
|
||||||
|
boost::thread::attributes attrs;
|
||||||
|
attrs.set_stack_size(THREAD_STACK_SIZE);
|
||||||
|
|
||||||
|
std::sort(users.begin(), users.end(), by_height{});
|
||||||
|
|
||||||
|
MINFO("Starting scan loops on " << thread_count << " thread(s) with " << users.size() << " account(s)");
|
||||||
|
|
||||||
|
for (std::size_t i = 0; i < queues.size(); ++i)
|
||||||
|
{
|
||||||
|
queues[i] = std::make_shared<rpc::scanner::queue>();
|
||||||
|
|
||||||
|
// this can create threads with no active accounts, they just wait
|
||||||
|
const std::size_t count = users.size() / (queues.size() - i);
|
||||||
|
std::vector<lws::account> thread_users{
|
||||||
|
std::make_move_iterator(users.end() - count), std::make_move_iterator(users.end())
|
||||||
|
};
|
||||||
|
users.erase(users.end() - count, users.end());
|
||||||
|
|
||||||
|
auto data = std::make_shared<thread_data>(
|
||||||
|
MONERO_UNWRAP(ctx.connect()), disk.clone(), std::move(thread_users), queues[i], opts
|
||||||
|
);
|
||||||
|
threads.emplace_back(attrs, std::bind(&do_scan_loop, std::ref(self), std::move(data), i == 0));
|
||||||
}
|
}
|
||||||
} join{self, ctx, queues, threads};
|
|
||||||
|
|
||||||
/*
|
users.clear();
|
||||||
The algorithm here is extremely basic. Users are divided evenly amongst
|
users.shrink_to_fit();
|
||||||
the configurable thread count, and grouped by scan height. If an old
|
|
||||||
account appears, some accounts (grouped on that thread) will be delayed
|
|
||||||
in processing waiting for that account to catch up. Its not the greatest,
|
|
||||||
but this "will have to do" - but we're getting closer to fixing that
|
|
||||||
too.
|
|
||||||
|
|
||||||
Another "issue" is that each thread works independently instead of more
|
auto server = std::make_shared<rpc::scanner::server>(
|
||||||
cooperatively for scanning. This requires a bit more synchronization, so
|
self.io_,
|
||||||
was left for later. Its likely worth doing to reduce the number of
|
disk.clone(),
|
||||||
transfers from the daemon, and the bottleneck on the writes into LMDB.
|
MONERO_UNWRAP(ctx.connect()),
|
||||||
*/
|
queues,
|
||||||
|
std::move(active),
|
||||||
self.stop_ = false;
|
self.webhooks_.ssl_context()
|
||||||
|
|
||||||
boost::thread::attributes attrs;
|
|
||||||
attrs.set_stack_size(THREAD_STACK_SIZE);
|
|
||||||
|
|
||||||
std::sort(users.begin(), users.end(), by_height{});
|
|
||||||
|
|
||||||
MINFO("Starting scan loops on " << thread_count << " thread(s) with " << users.size() << " account(s)");
|
|
||||||
|
|
||||||
for (std::size_t i = 0; i < queues.size(); ++i)
|
|
||||||
{
|
|
||||||
queues[i] = std::make_shared<rpc::scanner::queue>();
|
|
||||||
|
|
||||||
// this can create threads with no active accounts, they just wait
|
|
||||||
const std::size_t count = users.size() / (queues.size() - i);
|
|
||||||
std::vector<lws::account> thread_users{
|
|
||||||
std::make_move_iterator(users.end() - count), std::make_move_iterator(users.end())
|
|
||||||
};
|
|
||||||
users.erase(users.end() - count, users.end());
|
|
||||||
|
|
||||||
auto data = std::make_shared<thread_data>(
|
|
||||||
MONERO_UNWRAP(ctx.connect()), disk.clone(), std::move(thread_users), queues[i], opts
|
|
||||||
);
|
);
|
||||||
threads.emplace_back(attrs, std::bind(&do_scan_loop, std::ref(self), std::move(data), i == 0));
|
|
||||||
}
|
|
||||||
|
|
||||||
users.clear();
|
rpc::scanner::server::start_user_checking(server);
|
||||||
users.shrink_to_fit();
|
if (!lws_server_addr.empty())
|
||||||
|
rpc::scanner::server::start_acceptor(server, lws_server_addr, std::move(lws_server_pass));
|
||||||
|
|
||||||
auto server = std::make_shared<rpc::scanner::server>(
|
// Blocks until sigint, local scanner issue, storage issue, or exception
|
||||||
self.io_,
|
self.io_.restart();
|
||||||
disk.clone(),
|
self.io_.run();
|
||||||
MONERO_UNWRAP(ctx.connect()),
|
|
||||||
queues,
|
|
||||||
std::move(active),
|
|
||||||
self.webhooks_.ssl_context()
|
|
||||||
);
|
|
||||||
|
|
||||||
rpc::scanner::server::start_user_checking(server);
|
rpc::scanner::server::stop(server);
|
||||||
if (!lws_server_addr.empty())
|
} // block until all threads join
|
||||||
rpc::scanner::server::start_acceptor(std::move(server), lws_server_addr, std::move(lws_server_pass));
|
|
||||||
|
|
||||||
// Blocks until sigint, local scanner issue, storage issue, or exception
|
|
||||||
self.io_.run();
|
|
||||||
self.io_.restart();
|
|
||||||
|
|
||||||
// Make sure server stops because we could re-start after blockchain sync
|
// Make sure server stops because we could re-start after blockchain sync
|
||||||
rpc::scanner::server::stop(server);
|
|
||||||
self.io_.poll();
|
|
||||||
self.io_.restart();
|
self.io_.restart();
|
||||||
|
self.io_.poll();
|
||||||
}
|
}
|
||||||
|
|
||||||
template<typename R, typename Q>
|
template<typename R, typename Q>
|
||||||
@@ -1389,8 +1392,8 @@ namespace lws
|
|||||||
the correct timer was run. */
|
the correct timer was run. */
|
||||||
while (!has_shutdown() && ready.wait_for(std::chrono::seconds{0}) == std::future_status::timeout)
|
while (!has_shutdown() && ready.wait_for(std::chrono::seconds{0}) == std::future_status::timeout)
|
||||||
{
|
{
|
||||||
sync_.io_.run_one();
|
|
||||||
sync_.io_.restart();
|
sync_.io_.restart();
|
||||||
|
sync_.io_.run_one();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
|
|||||||
Reference in New Issue
Block a user