ARCHIVE/rippled
1
0
Fork 0
mirror of https://github.com/XRPLF/rippled.git synced 2025-12-06 17:27:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

Format first-party source according to .clang-format

Browse Source
This commit is contained in:
Pretty Printer
2020-04-17 09:56:34 -05:00
committed by manojsdoshi
parent 65dfc5d19e
commit 50760c6935
1076 changed files with 86161 additions and 77449 deletions
Download Patch File Download Diff File Expand all files Collapse all files

167
src/ripple/rpc/impl/ShardArchiveHandler.cpp
View File

@@ -19,12 +19,12 @@
#include <ripple/app/misc/NetworkOPs.h>
#include <ripple/basics/Archive.h>
#include <ripple/basics/BasicConfig.h>
#include <ripple/core/ConfigSections.h>
#include <ripple/nodestore/DatabaseShard.h>
#include <ripple/rpc/ShardArchiveHandler.h>
#include <ripple/basics/BasicConfig.h>
#include <ripple/rpc/impl/Handler.h>
#include <ripple/protocol/ErrorCodes.h>
#include <ripple/rpc/ShardArchiveHandler.h>
#include <ripple/rpc/impl/Handler.h>
#include <memory>
@@ -40,10 +40,12 @@ ShardArchiveHandler::pointer ShardArchiveHandler::instance_ = nullptr;
boost::filesystem::path
ShardArchiveHandler::getDownloadDirectory(Config const& config)
{
return get(config.section(
ConfigSection::shardDatabase()), "download_path",
get(config.section(ConfigSection::shardDatabase()),
"path", "")) / "download";
return get(config.section(ConfigSection::shardDatabase()),
"download_path",
get(config.section(ConfigSection::shardDatabase()),
"path",
"")) /
"download";
}
auto
@@ -55,8 +57,7 @@ ShardArchiveHandler::getInstance() -> pointer
}
auto
ShardArchiveHandler::getInstance(Application& app,
Stoppable& parent) -> pointer
ShardArchiveHandler::getInstance(Application& app, Stoppable& parent) -> pointer
{
std::lock_guard lock(instance_mutex_);
assert(!instance_);
@@ -67,7 +68,8 @@ ShardArchiveHandler::getInstance(Application& app,
}
auto
ShardArchiveHandler::recoverInstance(Application& app, Stoppable& parent) -> pointer
ShardArchiveHandler::recoverInstance(Application& app, Stoppable& parent)
-> pointer
{
std::lock_guard lock(instance_mutex_);
assert(!instance_);
@@ -98,9 +100,9 @@ ShardArchiveHandler::ShardArchiveHandler(
{
assert(app_.getShardStore());
if(recovery)
downloader_.reset(new DatabaseDownloader (
app_.getIOService(), j_, app_.config()));
if (recovery)
downloader_.reset(
new DatabaseDownloader(app_.getIOService(), j_, app_.config()));
}
bool
@@ -111,15 +113,15 @@ ShardArchiveHandler::init()
create_directories(downloadDir_);
sqliteDB_ = std::make_unique<DatabaseCon>(
downloadDir_ ,
downloadDir_,
stateDBName,
DownloaderDBPragma,
ShardArchiveHandlerDBInit);
}
catch(std::exception const& e)
catch (std::exception const& e)
{
JLOG(j_.error()) << "exception: " << e.what()
<< " in function: " << __func__;
<< " in function: " << __func__;
return false;
}
@@ -134,7 +136,8 @@ ShardArchiveHandler::initFromDB()
{
using namespace boost::filesystem;
assert(exists(downloadDir_ / stateDBName) &&
assert(
exists(downloadDir_ / stateDBName) &&
is_regular_file(downloadDir_ / stateDBName));
sqliteDB_ = std::make_unique<DatabaseCon>(
@@ -145,8 +148,8 @@ ShardArchiveHandler::initFromDB()
auto& session{sqliteDB_->getSession()};
soci::rowset<soci::row> rs = (session.prepare
<< "SELECT * FROM State;");
soci::rowset<soci::row> rs =
(session.prepare << "SELECT * FROM State;");
std::lock_guard<std::mutex> lock(m_);
@@ -156,8 +159,8 @@ ShardArchiveHandler::initFromDB()
if (!parseUrl(url, it->get<std::string>(1)))
{
JLOG(j_.error()) << "Failed to parse url: "
<< it->get<std::string>(1);
JLOG(j_.error())
<< "Failed to parse url: " << it->get<std::string>(1);
continue;
}
@@ -167,16 +170,16 @@ ShardArchiveHandler::initFromDB()
// Failed to load anything
// from the state database.
if(archives_.empty())
if (archives_.empty())
{
release();
return false;
}
}
catch(std::exception const& e)
catch (std::exception const& e)
{
JLOG(j_.error()) << "exception: " << e.what()
<< " in function: " << __func__;
<< " in function: " << __func__;
return false;
}
@@ -199,7 +202,8 @@ ShardArchiveHandler::onStop()
}
bool
ShardArchiveHandler::add(std::uint32_t shardIndex,
ShardArchiveHandler::add(
std::uint32_t shardIndex,
std::pair<parsedURL, std::string>&& url)
{
std::lock_guard<std::mutex> lock(m_);
@@ -210,24 +214,24 @@ ShardArchiveHandler::add(std::uint32_t shardIndex,
auto& session{sqliteDB_->getSession()};
session << "INSERT INTO State VALUES (:index, :url);",
soci::use(shardIndex),
soci::use(url.second);
soci::use(shardIndex), soci::use(url.second);
return true;
}
bool
ShardArchiveHandler::add(std::uint32_t shardIndex, parsedURL&& url,
ShardArchiveHandler::add(
std::uint32_t shardIndex,
parsedURL&& url,
std::lock_guard<std::mutex> const&)
{
if (process_)
{
JLOG(j_.error()) <<
"Download and import already in progress";
JLOG(j_.error()) << "Download and import already in progress";
return false;
}
auto const it {archives_.find(shardIndex)};
auto const it{archives_.find(shardIndex)};
if (it != archives_.end())
return url == it->second;
@@ -245,20 +249,17 @@ ShardArchiveHandler::start()
std::lock_guard lock(m_);
if (!app_.getShardStore())
{
JLOG(j_.error()) <<
"No shard store available";
JLOG(j_.error()) << "No shard store available";
return false;
}
if (process_)
{
JLOG(j_.warn()) <<
"Archives already being processed";
JLOG(j_.warn()) << "Archives already being processed";
return false;
}
if (archives_.empty())
{
JLOG(j_.warn()) <<
"No archives to process";
JLOG(j_.warn()) << "No archives to process";
return false;
}
@@ -276,8 +277,7 @@ ShardArchiveHandler::start()
}
catch (std::exception const& e)
{
JLOG(j_.error()) <<
"exception: " << e.what();
JLOG(j_.error()) << "exception: " << e.what();
return false;
}
@@ -301,16 +301,15 @@ ShardArchiveHandler::next(std::lock_guard<std::mutex>& l)
}
// Create a temp archive directory at the root
auto const shardIndex {archives_.begin()->first};
auto const dstDir {downloadDir_ / std::to_string(shardIndex)};
auto const shardIndex{archives_.begin()->first};
auto const dstDir{downloadDir_ / std::to_string(shardIndex)};
try
{
create_directory(dstDir);
}
catch (std::exception const& e)
{
JLOG(j_.error()) <<
"exception: " << e.what();
JLOG(j_.error()) << "exception: " << e.what();
remove(l);
return next(l);
}
@@ -318,19 +317,21 @@ ShardArchiveHandler::next(std::lock_guard<std::mutex>& l)
// Download the archive. Process in another thread
// to prevent holding up the lock if the downloader
// sleeps.
auto const& url {archives_.begin()->second};
auto const& url{archives_.begin()->second};
app_.getJobQueue().addJob(
jtCLIENT, "ShardArchiveHandler",
[this, ptr = shared_from_this(), url, dstDir](Job&)
{
jtCLIENT,
"ShardArchiveHandler",
[this, ptr = shared_from_this(), url, dstDir](Job&) {
if (!downloader_->download(
url.domain,
std::to_string(url.port.get_value_or(443)),
url.path,
11,
dstDir / "archive.tar.lz4",
std::bind(&ShardArchiveHandler::complete,
ptr, std::placeholders::_1)))
url.domain,
std::to_string(url.port.get_value_or(443)),
url.path,
11,
dstDir / "archive.tar.lz4",
std::bind(
&ShardArchiveHandler::complete,
ptr,
std::placeholders::_1)))
{
std::lock_guard<std::mutex> l(m_);
remove(l);
@@ -351,10 +352,10 @@ ShardArchiveHandler::complete(path dstPath)
{
if (!is_regular_file(dstPath))
{
auto ar {archives_.begin()};
JLOG(j_.error()) <<
"Downloading shard id " << ar->first <<
" form URL " << ar->second.domain << ar->second.path;
auto ar{archives_.begin()};
JLOG(j_.error())
<< "Downloading shard id " << ar->first << " form URL "
<< ar->second.domain << ar->second.path;
remove(lock);
next(lock);
return;
@@ -362,8 +363,7 @@ ShardArchiveHandler::complete(path dstPath)
}
catch (std::exception const& e)
{
JLOG(j_.error()) <<
"exception: " << e.what();
JLOG(j_.error()) << "exception: " << e.what();
remove(lock);
next(lock);
return;
@@ -372,21 +372,21 @@ ShardArchiveHandler::complete(path dstPath)
// Process in another thread to not hold up the IO service
app_.getJobQueue().addJob(
jtCLIENT, "ShardArchiveHandler",
[=, dstPath = std::move(dstPath), ptr = shared_from_this()](Job&)
{
jtCLIENT,
"ShardArchiveHandler",
[=, dstPath = std::move(dstPath), ptr = shared_from_this()](Job&) {
// If not synced then defer and retry
auto const mode {ptr->app_.getOPs().getOperatingMode()};
auto const mode{ptr->app_.getOPs().getOperatingMode()};
if (mode != OperatingMode::FULL)
{
std::lock_guard lock(m_);
timer_.expires_from_now(static_cast<std::chrono::seconds>(
(static_cast<std::size_t>(OperatingMode::FULL)
- static_cast<std::size_t>(mode)) * 10));
(static_cast<std::size_t>(OperatingMode::FULL) -
static_cast<std::size_t>(mode)) *
10));
timer_.async_wait(
[=, dstPath = std::move(dstPath), ptr = std::move(ptr)]
(boost::system::error_code const& ec)
{
[=, dstPath = std::move(dstPath), ptr = std::move(ptr)](
boost::system::error_code const& ec) {
if (ec != boost::asio::error::operation_aborted)
ptr->complete(std::move(dstPath));
});
@@ -410,7 +410,7 @@ ShardArchiveHandler::process(path const& dstPath)
shardIndex = archives_.begin()->first;
}
auto const shardDir {dstPath.parent_path() / std::to_string(shardIndex)};
auto const shardDir{dstPath.parent_path() / std::to_string(shardIndex)};
try
{
// Extract the downloaded archive
@@ -419,35 +419,31 @@ ShardArchiveHandler::process(path const& dstPath)
// The extracted root directory name must match the shard index
if (!is_directory(shardDir))
{
JLOG(j_.error()) <<
"Shard " << shardIndex <<
" mismatches archive shard directory";
JLOG(j_.error()) << "Shard " << shardIndex
<< " mismatches archive shard directory";
return;
}
}
catch (std::exception const& e)
{
JLOG(j_.error()) <<
"exception: " << e.what();
JLOG(j_.error()) << "exception: " << e.what();
return;
}
// Import the shard into the shard store
if (!app_.getShardStore()->importShard(shardIndex, shardDir))
{
JLOG(j_.error()) <<
"Importing shard " << shardIndex;
JLOG(j_.error()) << "Importing shard " << shardIndex;
return;
}
JLOG(j_.debug()) <<
"Shard " << shardIndex << " downloaded and imported";
JLOG(j_.debug()) << "Shard " << shardIndex << " downloaded and imported";
}
void
ShardArchiveHandler::remove(std::lock_guard<std::mutex>&)
{
auto const shardIndex {archives_.begin()->first};
auto const shardIndex{archives_.begin()->first};
app_.getShardStore()->removePreShard(shardIndex);
archives_.erase(shardIndex);
@@ -456,15 +452,14 @@ ShardArchiveHandler::remove(std::lock_guard<std::mutex>&)
session << "DELETE FROM State WHERE ShardIndex = :index;",
soci::use(shardIndex);
auto const dstDir {downloadDir_ / std::to_string(shardIndex)};
auto const dstDir{downloadDir_ / std::to_string(shardIndex)};
try
{
remove_all(dstDir);
}
catch (std::exception const& e)
{
JLOG(j_.error()) <<
"exception: " << e.what();
JLOG(j_.error()) << "exception: " << e.what();
}
}
@@ -494,11 +489,11 @@ ShardArchiveHandler::doRelease(std::lock_guard<std::mutex> const&)
catch (std::exception const& e)
{
JLOG(j_.error()) << "exception: " << e.what()
<< " in function: " << __func__;
<< " in function: " << __func__;
}
downloader_.reset();
}
} // RPC
} // ripple
} // namespace RPC
} // namespace ripple