ARCHIVE/xahaud
3
0
Fork 0
mirror of https://github.com/Xahau/xahaud.git synced 2025-11-19 18:15:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Eliminate the built-in SNTP support (fixes #4207): (#4628)

Browse Source
This commit is contained in:
Nik Bougalis
2023-09-26 17:35:31 -07:00
committed by tequ
parent 6cf6416f15
commit 9838bdf214
17 changed files with 130 additions and 930 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
Builds/CMake/RippledCore.cmake
View File

@@ -588,9 +588,7 @@ target_sources (rippled PRIVATE
src/ripple/core/impl/JobQueue.cpp
src/ripple/core/impl/LoadEvent.cpp
src/ripple/core/impl/LoadMonitor.cpp
src/ripple/core/impl/SNTPClock.cpp
src/ripple/core/impl/SociDB.cpp
src/ripple/core/impl/TimeKeeper.cpp
src/ripple/core/impl/Workers.cpp
src/ripple/core/Pg.cpp
#[===============================[
@@ -968,7 +966,6 @@ if (tests)
src/test/jtx/impl/AMMTest.cpp
src/test/jtx/impl/Env.cpp
src/test/jtx/impl/JSONRPCClient.cpp
src/test/jtx/impl/ManualTimeKeeper.cpp
src/test/jtx/impl/TestHelpers.cpp
src/test/jtx/impl/WSClient.cpp
src/test/jtx/impl/hook.cpp

1
Builds/levelization/results/ordering.txt
View File

@@ -168,7 +168,6 @@ test.nodestore > test.unit_test
test.overlay > ripple.app
test.overlay > ripple.basics
test.overlay > ripple.beast
test.overlay > ripple.core
test.overlay > ripple.overlay
test.overlay > ripple.peerfinder
test.overlay > ripple.protocol

19
cfg/rippled-example.cfg
View File

@@ -463,19 +463,6 @@
#
#
#
# [sntp_servers]
#
# IP address or domain of NTP servers to use for time synchronization.
#
# These NTP servers are suitable for rippled servers located in the United
# States:
# time.windows.com
# time.apple.com
# time.nist.gov
# pool.ntp.org
#
#
#
# [max_transactions]
#
# Configure the maximum number of transactions to have in the job queue
@@ -1716,12 +1703,6 @@ advisory_delete=0
[debug_logfile]
/var/log/rippled/debug.log
[sntp_servers]
time.windows.com
time.apple.com
time.nist.gov
pool.ntp.org
# To use the XRP test network
# (see https://xrpl.org/connect-your-rippled-to-the-xrp-test-net.html),
# use the following [ips] section:

19
cfg/rippled-reporting.cfg
View File

@@ -450,19 +450,6 @@
#
#
#
# [sntp_servers]
#
# IP address or domain of NTP servers to use for time synchronization.
#
# These NTP servers are suitable for rippled servers located in the United
# States:
# time.windows.com
# time.apple.com
# time.nist.gov
# pool.ntp.org
#
#
#
# [max_transactions]
#
# Configure the maximum number of transactions to have in the job queue
@@ -1662,12 +1649,6 @@ advisory_delete=0
[debug_logfile]
/var/log/rippled-reporting/debug.log
[sntp_servers]
time.windows.com
time.apple.com
time.nist.gov
pool.ntp.org
# To use the XRP test network
# (see https://xrpl.org/connect-your-rippled-to-the-xrp-test-net.html),
# use the following [ips] section:

3
src/ripple/app/main/Application.cpp
View File

@@ -1185,9 +1185,6 @@ ApplicationImp::setup(boost::program_options::variables_map const& cmdline)
// Optionally turn off logging to console.
logs_->silent(config_->silent());
if (!config_->standalone())
timeKeeper_->run(config_->SNTP_SERVERS);
if (!initRelationalDatabase() || !initNodeStore())
return false;

4
src/ripple/app/main/Main.cpp
View File

@@ -802,10 +802,8 @@ run(int argc, char** argv)
if (vm.count("debug"))
setDebugLogSink(logs->makeSink("Debug", beast::severities::kTrace));
auto timeKeeper = make_TimeKeeper(logs->journal("TimeKeeper"));
auto app = make_Application(
std::move(config), std::move(logs), std::move(timeKeeper));
std::move(config), std::move(logs), std::make_unique<TimeKeeper>());
if (!app->setup(vm))
return -1;

18
src/ripple/app/misc/NetworkOPs.cpp
View File

@@ -688,11 +688,10 @@ private:
sPeerStatus, // Peer status changes.
sConsensusPhase, // Consensus phase
sBookChanges, // Per-ledger order book changes
sLastEntry = sBookChanges // as this name implies, any new entry
// must be ADDED ABOVE this one
sLastEntry // Any new entry must be ADDED ABOVE this one
};
std::array<SubMapType, SubTypes::sLastEntry + 1> mStreamMaps;
std::array<SubMapType, SubTypes::sLastEntry> mStreamMaps;
ServerFeeSummary mLastFeeSummary;
@@ -2651,13 +2650,10 @@ NetworkOPsImp::getServerInfo(bool human, bool admin, bool counters)
l[jss::reserve_inc_native] =
lpClosed->fees().increment.decimalXRP();
auto const nowOffset = app_.timeKeeper().nowOffset();
if (std::abs(nowOffset.count()) >= 60)
l[jss::system_time_offset] = nowOffset.count();
auto const closeOffset = app_.timeKeeper().closeOffset();
if (std::abs(closeOffset.count()) >= 60)
l[jss::close_time_offset] = closeOffset.count();
if (auto const closeOffset = app_.timeKeeper().closeOffset();
std::abs(closeOffset.count()) >= 60)
l[jss::close_time_offset] =
static_cast<std::uint32_t>(closeOffset.count());
#if RIPPLED_REPORTING
std::int64_t const dbAge =

2
src/ripple/beast/clock/abstract_clock.h
View File

@@ -70,7 +70,7 @@ public:
abstract_clock(abstract_clock const&) = default;
/** Returns the current time. */
virtual time_point
[[nodiscard]] virtual time_point
now() const = 0;
};

8
src/ripple/core/Config.h
View File

@@ -151,9 +151,11 @@ public:
bool nodeToShard = false;
bool ELB_SUPPORT = false;
std::vector<std::string> IPS; // Peer IPs from rippled.cfg.
std::vector<std::string> IPS_FIXED; // Fixed Peer IPs from rippled.cfg.
std::vector<std::string> SNTP_SERVERS; // SNTP servers from rippled.cfg.
// Entries from [ips] config stanza
std::vector<std::string> IPS;
// Entries from [ips_fixed] config stanza
std::vector<std::string> IPS_FIXED;
std::map<std::string, PublicKey>
IMPORT_VL_KEYS; // hex string -> class PublicKey (for caching purposes)

122
src/ripple/core/TimeKeeper.h
View File

@@ -22,73 +22,99 @@
#include <ripple/basics/chrono.h>
#include <ripple/beast/clock/abstract_clock.h>
#include <ripple/beast/utility/Journal.h>
#include <string>
#include <vector>
#include <atomic>
namespace ripple {
/** Manages various times used by the server. */
class TimeKeeper : public beast::abstract_clock<NetClock>
{
private:
std::atomic<std::chrono::seconds> closeOffset_{};
// Adjust system_clock::time_point for NetClock epoch
static constexpr time_point
adjust(std::chrono::system_clock::time_point when)
{
return time_point(std::chrono::duration_cast<duration>(
when.time_since_epoch() - days(10957)));
}
public:
virtual ~TimeKeeper() = default;
/** Launch the internal thread.
/** Returns the current time, using the server's clock.
The internal thread synchronizes local network time
using the provided list of SNTP servers.
It's possible for servers to have a different value for network
time, especially if they do not use some external mechanism for
time synchronization (e.g. NTP or SNTP). This is fine.
This estimate is not directly visible to other servers over the
protocol, but it is possible for them to make an educated guess
if this server publishes proposals or validations.
@note The network time is adjusted for the "Ripple epoch" which
was arbitrarily defined as 2000-01-01T00:00:00Z by Arthur
Britto and David Schwartz during early development of the
code. No rationale has been provided for this curious and
annoying, but otherwise unimportant, choice.
*/
virtual void
run(std::vector<std::string> const& servers) = 0;
[[nodiscard]] time_point
now() const override
{
return adjust(std::chrono::system_clock::now());
}
/** Returns the estimate of wall time, in network time.
/** Returns the predicted close time, in network time.
The network time is wall time adjusted for the Ripple
epoch, the beginning of January 1st, 2000 UTC. Each server
can compute a different value for network time. Other
servers value for network time is not directly observable,
but good guesses can be made by looking at validators'
positions on close times.
Servers compute network time by adjusting a local wall
clock using SNTP and then adjusting for the epoch.
The predicted close time represents the notional "center" of the
network. Each server assumes that its clock is correct and tries
to pull the close time towards its measure of network time.
*/
virtual time_point
now() const override = 0;
/** Returns the close time, in network time.
Close time is the time the network would agree that
a ledger closed, if a ledger closed right now.
The close time represents the notional "center"
of the network. Each server assumes its clock
is correct, and tries to pull the close time towards
its measure of network time.
*/
virtual time_point
closeTime() const = 0;
/** Adjust the close time.
This is called in response to received validations.
*/
virtual void
adjustCloseTime(std::chrono::duration<std::int32_t> amount) = 0;
[[nodiscard]] time_point
closeTime() const
{
return now() + closeOffset_.load();
}
// This may return a negative value
virtual std::chrono::duration<std::int32_t>
nowOffset() const = 0;
[[nodiscard]] std::chrono::seconds
closeOffset() const
{
return closeOffset_.load();
}
// This may return a negative value
virtual std::chrono::duration<std::int32_t>
closeOffset() const = 0;
/** Adjust the close time, based on the network's view of time. */
std::chrono::seconds
adjustCloseTime(std::chrono::seconds by)
{
using namespace std::chrono_literals;
auto offset = closeOffset_.load();
if (by == 0s && offset == 0s)
return offset;
// The close time adjustment is serialized externally to this
// code. The compare/exchange only serves as a weak check and
// should not fail. Even if it does, it's safe to simply just
// skip the adjustment.
closeOffset_.compare_exchange_strong(offset, [by, offset]() {
// Ignore small offsets and push the close time
// towards our wall time.
if (by > 1s)
return offset + ((by + 3s) / 4);
if (by < -1s)
return offset + ((by - 3s) / 4);
return (offset * 3) / 4;
}());
return closeOffset_.load();
}
};
extern std::unique_ptr<TimeKeeper>
make_TimeKeeper(beast::Journal j);
} // namespace ripple
#endif

36
src/ripple/core/impl/Config.cpp
View File

@@ -206,14 +206,10 @@ parseIniFile(std::string const& strInput, const bool bTrim)
IniFileSections::mapped_type*
getIniFileSection(IniFileSections& secSource, std::string const& strSection)
{
IniFileSections::iterator it;
IniFileSections::mapped_type* smtResult;
it = secSource.find(strSection);
if (it == secSource.end())
smtResult = nullptr;
else
smtResult = &(it->second);
return smtResult;
if (auto it = secSource.find(strSection); it != secSource.end())
return &(it->second);
return nullptr;
}
bool
@@ -223,22 +219,21 @@ getSingleSection(
std::string& strValue,
beast::Journal j)
{
IniFileSections::mapped_type* pmtEntries =
getIniFileSection(secSource, strSection);
bool bSingle = pmtEntries && 1 == pmtEntries->size();
auto const pmtEntries = getIniFileSection(secSource, strSection);
if (bSingle)
if (pmtEntries && pmtEntries->size() == 1)
{
strValue = (*pmtEntries)[0];
}
else if (pmtEntries)
{
JLOG(j.warn()) << boost::str(
boost::format("Section [%s]: requires 1 line not %d lines.") %
strSection % pmtEntries->size());
return true;
}
return bSingle;
if (pmtEntries)
{
JLOG(j.warn()) << "Section '" << strSection << "': requires 1 line not "
<< pmtEntries->size() << " lines.";
}
return false;
}
//------------------------------------------------------------------------------
@@ -464,9 +459,6 @@ Config::loadFromString(std::string const& fileContents)
if (auto s = getIniFileSection(secConfig, SECTION_IPS_FIXED))
IPS_FIXED = *s;
if (auto s = getIniFileSection(secConfig, SECTION_SNTP))
SNTP_SERVERS = *s;
// if the user has specified ip:port then replace : with a space.
auto replaceColons = [](std::vector<std::string>& strVec) {
const static std::regex e(":([0-9]+)$");

491
src/ripple/core/impl/SNTPClock.cpp
View File

@@ -1,491 +0,0 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2012, 2013 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#include <ripple/basics/Log.h>
#include <ripple/basics/random.h>
#include <ripple/beast/core/CurrentThreadName.h>
#include <ripple/core/impl/SNTPClock.h>
#include <boost/asio.hpp>
#include <cmath>
#include <deque>
#include <map>
#include <memory>
#include <mutex>
#include <optional>
#include <thread>
namespace ripple {
static uint8_t SNTPQueryData[48] = {
0x1B, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0};
using namespace std::chrono_literals;
// NTP query frequency - 4 minutes
auto constexpr NTP_QUERY_FREQUENCY = 4min;
// NTP minimum interval to query same servers - 3 minutes
auto constexpr NTP_MIN_QUERY = 3min;
// NTP sample window (should be odd)
#define NTP_SAMPLE_WINDOW 9
// NTP timestamp constant
auto constexpr NTP_UNIX_OFFSET = 0x83AA7E80s;
// NTP timestamp validity
auto constexpr NTP_TIMESTAMP_VALID = (NTP_QUERY_FREQUENCY + NTP_MIN_QUERY) * 2;
// SNTP packet offsets
#define NTP_OFF_INFO 0
#define NTP_OFF_ROOTDELAY 1
#define NTP_OFF_ROOTDISP 2
#define NTP_OFF_REFERENCEID 3
#define NTP_OFF_REFTS_INT 4
#define NTP_OFF_REFTS_FRAC 5
#define NTP_OFF_ORGTS_INT 6
#define NTP_OFF_ORGTS_FRAC 7
#define NTP_OFF_RECVTS_INT 8
#define NTP_OFF_RECVTS_FRAC 9
#define NTP_OFF_XMITTS_INT 10
#define NTP_OFF_XMITTS_FRAC 11
class SNTPClientImp : public SNTPClock
{
private:
template <class Duration>
using sys_time = std::chrono::time_point<clock_type, Duration>;
using sys_seconds = sys_time<std::chrono::seconds>;
struct Query
{
bool replied;
sys_seconds sent;
std::uint32_t nonce;
explicit Query(sys_seconds j = sys_seconds::max())
: replied(false), sent(j)
{
}
};
beast::Journal const j_;
std::mutex mutable mutex_;
std::thread thread_;
boost::asio::io_service io_service_;
std::optional<boost::asio::io_service::work> work_;
std::map<boost::asio::ip::udp::endpoint, Query> queries_;
boost::asio::ip::udp::socket socket_;
boost::asio::basic_waitable_timer<std::chrono::system_clock> timer_;
boost::asio::ip::udp::resolver resolver_;
std::vector<std::pair<std::string, sys_seconds>> servers_;
std::chrono::seconds offset_;
sys_seconds lastUpdate_;
std::deque<std::chrono::seconds> offsets_;
std::vector<uint8_t> buf_;
boost::asio::ip::udp::endpoint ep_;
public:
using error_code = boost::system::error_code;
explicit SNTPClientImp(beast::Journal j)
: j_(j)
, work_(io_service_)
, socket_(io_service_)
, timer_(io_service_)
, resolver_(io_service_)
, offset_(0)
, lastUpdate_(sys_seconds::max())
, buf_(256)
{
}
~SNTPClientImp() override
{
if (thread_.joinable())
{
error_code ec;
timer_.cancel(ec);
socket_.cancel(ec);
work_ = std::nullopt;
thread_.join();
}
}
//--------------------------------------------------------------------------
void
run(const std::vector<std::string>& servers) override
{
std::vector<std::string>::const_iterator it = servers.begin();
if (it == servers.end())
{
JLOG(j_.info()) << "SNTP: no server specified";
return;
}
{
std::lock_guard lock(mutex_);
for (auto const& item : servers)
servers_.emplace_back(item, sys_seconds::max());
}
queryAll();
using namespace boost::asio;
socket_.open(ip::udp::v4());
socket_.bind(ep_);
socket_.async_receive_from(
buffer(buf_, 256),
ep_,
std::bind(
&SNTPClientImp::onRead,
this,
std::placeholders::_1,
std::placeholders::_2));
timer_.expires_from_now(NTP_QUERY_FREQUENCY);
timer_.async_wait(
std::bind(&SNTPClientImp::onTimer, this, std::placeholders::_1));
thread_ = std::thread(&SNTPClientImp::doRun, this);
}
time_point
now() const override
{
std::lock_guard lock(mutex_);
using namespace std::chrono;
auto const when = time_point_cast<seconds>(clock_type::now());
if ((lastUpdate_ == sys_seconds::max()) ||
((lastUpdate_ + NTP_TIMESTAMP_VALID) <
time_point_cast<seconds>(clock_type::now())))
return when;
return when + offset_;
}
duration
offset() const override
{
std::lock_guard lock(mutex_);
return offset_;
}
//--------------------------------------------------------------------------
void
doRun()
{
beast::setCurrentThreadName("rippled: SNTPClock");
io_service_.run();
}
void
onTimer(error_code const& ec)
{
using namespace boost::asio;
if (ec == error::operation_aborted)
return;
if (ec)
{
JLOG(j_.error()) << "SNTPClock::onTimer: " << ec.message();
return;
}
doQuery();
timer_.expires_from_now(NTP_QUERY_FREQUENCY);
timer_.async_wait(
std::bind(&SNTPClientImp::onTimer, this, std::placeholders::_1));
}
void
onRead(error_code const& ec, std::size_t bytes_xferd)
{
using namespace boost::asio;
using namespace std::chrono;
if (ec == error::operation_aborted)
return;
// VFALCO Should we return on any error?
/*
if (ec)
return;
*/
if (!ec)
{
JLOG(j_.trace()) << "SNTP: Packet from " << ep_;
std::lock_guard lock(mutex_);
auto const query = queries_.find(ep_);
if (query == queries_.end())
{
JLOG(j_.debug()) << "SNTP: Reply from " << ep_
<< " found without matching query";
}
else if (query->second.replied)
{
JLOG(j_.debug()) << "SNTP: Duplicate response from " << ep_;
}
else
{
query->second.replied = true;
if (time_point_cast<seconds>(clock_type::now()) >
(query->second.sent + 1s))
{
JLOG(j_.warn()) << "SNTP: Late response from " << ep_;
}
else if (bytes_xferd < 48)
{
JLOG(j_.warn()) << "SNTP: Short reply from " << ep_ << " ("
<< bytes_xferd << ") " << buf_.size();
}
else if (
reinterpret_cast<std::uint32_t*>(
&buf_[0])[NTP_OFF_ORGTS_FRAC] != query->second.nonce)
{
JLOG(j_.warn())
<< "SNTP: Reply from " << ep_ << "had wrong nonce";
}
else
{
processReply();
}
}
}
socket_.async_receive_from(
buffer(buf_, 256),
ep_,
std::bind(
&SNTPClientImp::onRead,
this,
std::placeholders::_1,
std::placeholders::_2));
}
//--------------------------------------------------------------------------
void
addServer(std::string const& server)
{
std::lock_guard lock(mutex_);
servers_.push_back(std::make_pair(server, sys_seconds::max()));
}
void
queryAll()
{
while (doQuery())
{
}
}
bool
doQuery()
{
std::lock_guard lock(mutex_);
auto best = servers_.end();
for (auto iter = servers_.begin(), end = best; iter != end; ++iter)
if ((best == end) || (iter->second == sys_seconds::max()) ||
(iter->second < best->second))
best = iter;
if (best == servers_.end())
{
JLOG(j_.trace()) << "SNTP: No server to query";
return false;
}
using namespace std::chrono;
auto now = time_point_cast<seconds>(clock_type::now());
if ((best->second != sys_seconds::max()) &&
((best->second + NTP_MIN_QUERY) >= now))
{
JLOG(j_.trace()) << "SNTP: All servers recently queried";
return false;
}
best->second = now;
boost::asio::ip::udp::resolver::query query(
boost::asio::ip::udp::v4(), best->first, "ntp");
resolver_.async_resolve(
query,
std::bind(
&SNTPClientImp::resolveComplete,
this,
std::placeholders::_1,
std::placeholders::_2));
JLOG(j_.trace()) << "SNTPClock: Resolve pending for " << best->first;
return true;
}
void
resolveComplete(
error_code const& ec,
boost::asio::ip::udp::resolver::iterator it)
{
using namespace boost::asio;
if (ec == error::operation_aborted)
return;
if (ec)
{
JLOG(j_.trace()) << "SNTPClock::resolveComplete: " << ec.message();
return;
}
assert(it != ip::udp::resolver::iterator());
auto sel = it;
int i = 1;
while (++it != ip::udp::resolver::iterator())
{
if (rand_int(i++) == 0)
sel = it;
}
if (sel != ip::udp::resolver::iterator())
{
std::lock_guard lock(mutex_);
Query& query = queries_[*sel];
using namespace std::chrono;
auto now = time_point_cast<seconds>(clock_type::now());
if ((query.sent == now) || ((query.sent + 1s) == now))
{
// This can happen if the same IP address is reached through
// multiple names
JLOG(j_.trace()) << "SNTP: Redundant query suppressed";
return;
}
query.replied = false;
query.sent = now;
query.nonce = rand_int<std::uint32_t>();
// The following line of code will overflow at 2036-02-07 06:28:16
// UTC
// due to the 32 bit cast.
reinterpret_cast<std::uint32_t*>(
SNTPQueryData)[NTP_OFF_XMITTS_INT] =
static_cast<std::uint32_t>(
(time_point_cast<seconds>(clock_type::now()) +
NTP_UNIX_OFFSET)
.time_since_epoch()
.count());
reinterpret_cast<std::uint32_t*>(
SNTPQueryData)[NTP_OFF_XMITTS_FRAC] = query.nonce;
socket_.async_send_to(
buffer(SNTPQueryData, 48),
*sel,
std::bind(
&SNTPClientImp::onSend,
this,
std::placeholders::_1,
std::placeholders::_2));
}
}
void
onSend(error_code const& ec, std::size_t)
{
if (ec == boost::asio::error::operation_aborted)
return;
if (ec)
{
JLOG(j_.warn()) << "SNTPClock::onSend: " << ec.message();
return;
}
}
void
processReply()
{
using namespace std::chrono;
assert(buf_.size() >= 48);
std::uint32_t* recvBuffer =
reinterpret_cast<std::uint32_t*>(&buf_.front());
unsigned info = ntohl(recvBuffer[NTP_OFF_INFO]);
auto timev = seconds{ntohl(recvBuffer[NTP_OFF_RECVTS_INT])};
unsigned stratum = (info >> 16) & 0xff;
if ((info >> 30) == 3)
{
JLOG(j_.info()) << "SNTP: Alarm condition " << ep_;
return;
}
if ((stratum == 0) || (stratum > 14))
{
JLOG(j_.info()) << "SNTP: Unreasonable stratum (" << stratum
<< ") from " << ep_;
return;
}
using namespace std::chrono;
auto now = time_point_cast<seconds>(clock_type::now());
timev -= now.time_since_epoch();
timev -= NTP_UNIX_OFFSET;
// add offset to list, replacing oldest one if appropriate
offsets_.push_back(timev);
if (offsets_.size() >= NTP_SAMPLE_WINDOW)
offsets_.pop_front();
lastUpdate_ = now;
// select median time
auto offsetList = offsets_;
std::sort(offsetList.begin(), offsetList.end());
auto j = offsetList.size();
auto it = std::next(offsetList.begin(), j / 2);
offset_ = *it;
if ((j % 2) == 0)
offset_ = (offset_ + (*--it)) / 2;
// debounce: small corrections likely
// do more harm than good
if ((offset_ == -1s) || (offset_ == 1s))
offset_ = 0s;
if (timev != 0s || offset_ != 0s)
{
JLOG(j_.trace()) << "SNTP: Offset is " << timev.count()
<< ", new system offset is " << offset_.count();
}
}
};
//------------------------------------------------------------------------------
std::unique_ptr<SNTPClock>
make_SNTPClock(beast::Journal j)
{
return std::make_unique<SNTPClientImp>(j);
}
} // namespace ripple

47
src/ripple/core/impl/SNTPClock.h
View File

@@ -1,47 +0,0 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2012, 2013 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#ifndef RIPPLE_NET_SNTPCLOCK_H_INCLUDED
#define RIPPLE_NET_SNTPCLOCK_H_INCLUDED
#include <ripple/beast/clock/abstract_clock.h>
#include <ripple/beast/utility/Journal.h>
#include <chrono>
#include <memory>
#include <string>
#include <vector>
namespace ripple {
/** A clock based on system_clock and adjusted for SNTP. */
class SNTPClock : public beast::abstract_clock<std::chrono::system_clock>
{
public:
virtual void
run(std::vector<std::string> const& servers) = 0;
virtual duration
offset() const = 0;
};
extern std::unique_ptr<SNTPClock> make_SNTPClock(beast::Journal);
} // namespace ripple
#endif

124
src/ripple/core/impl/TimeKeeper.cpp
View File

@@ -1,124 +0,0 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2012, 2013 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#include <ripple/basics/Log.h>
#include <ripple/core/TimeKeeper.h>
#include <ripple/core/impl/SNTPClock.h>
#include <memory>
#include <mutex>
namespace ripple {
class TimeKeeperImpl : public TimeKeeper
{
private:
beast::Journal const j_;
std::mutex mutable mutex_;
std::chrono::duration<std::int32_t> closeOffset_;
std::unique_ptr<SNTPClock> clock_;
// Adjust system_clock::time_point for NetClock epoch
static time_point
adjust(std::chrono::system_clock::time_point when)
{
return time_point(std::chrono::duration_cast<duration>(
when.time_since_epoch() - days(10957)));
}
public:
explicit TimeKeeperImpl(beast::Journal j)
: j_(j), closeOffset_{}, clock_(make_SNTPClock(j))
{
}
void
run(std::vector<std::string> const& servers) override
{
clock_->run(servers);
}
time_point
now() const override
{
std::lock_guard lock(mutex_);
return adjust(clock_->now());
}
time_point
closeTime() const override
{
std::lock_guard lock(mutex_);
return adjust(clock_->now()) + closeOffset_;
}
void
adjustCloseTime(std::chrono::duration<std::int32_t> amount) override
{
using namespace std::chrono;
auto const s = amount.count();
std::lock_guard lock(mutex_);
// Take large offsets, ignore small offsets,
// push the close time towards our wall time.
if (s > 1)
closeOffset_ += seconds((s + 3) / 4);
else if (s < -1)
closeOffset_ += seconds((s - 3) / 4);
else
closeOffset_ = (closeOffset_ * 3) / 4;
if (closeOffset_.count() != 0)
{
if (std::abs(closeOffset_.count()) < 60)
{
JLOG(j_.info()) << "TimeKeeper: Close time offset now "
<< closeOffset_.count();
}
else
{
JLOG(j_.warn()) << "TimeKeeper: Large close time offset = "
<< closeOffset_.count();
}
}
}
std::chrono::duration<std::int32_t>
nowOffset() const override
{
using namespace std::chrono;
using namespace std;
lock_guard lock(mutex_);
return duration_cast<chrono::duration<int32_t>>(clock_->offset());
}
std::chrono::duration<std::int32_t>
closeOffset() const override
{
std::lock_guard lock(mutex_);
return closeOffset_;
}
};
//------------------------------------------------------------------------------
std::unique_ptr<TimeKeeper>
make_TimeKeeper(beast::Journal j)
{
return std::make_unique<TimeKeeperImpl>(j);
}
} // namespace ripple

47
src/test/jtx/ManualTimeKeeper.h
View File

@@ -21,45 +21,30 @@
#define RIPPLE_TEST_MANUALTIMEKEEPER_H_INCLUDED
#include <ripple/core/TimeKeeper.h>
#include <mutex>
#include <atomic>
namespace ripple {
namespace test {
class ManualTimeKeeper : public TimeKeeper
{
public:
ManualTimeKeeper();
void
run(std::vector<std::string> const& servers) override;
time_point
now() const override;
time_point
closeTime() const override;
void
adjustCloseTime(std::chrono::duration<std::int32_t> amount) override;
std::chrono::duration<std::int32_t>
nowOffset() const override;
std::chrono::duration<std::int32_t>
closeOffset() const override;
void
set(time_point now);
private:
// Adjust system_clock::time_point for NetClock epoch
static time_point
adjust(std::chrono::system_clock::time_point when);
std::atomic<time_point> now_{};
std::mutex mutable mutex_;
std::chrono::duration<std::int32_t> closeOffset_;
time_point now_;
public:
ManualTimeKeeper() = default;
time_point
now() const override
{
return now_.load();
}
void
set(time_point now)
{
now_.store(now);
}
};
} // namespace test

95
src/test/jtx/impl/ManualTimeKeeper.cpp
View File

@@ -1,95 +0,0 @@
//------------------------------------------------------------------------------
/*
This file is part of rippled: https://github.com/ripple/rippled
Copyright (c) 2012-2015 Ripple Labs Inc.
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#include <test/jtx/ManualTimeKeeper.h>
namespace ripple {
namespace test {
using namespace std::chrono_literals;
ManualTimeKeeper::ManualTimeKeeper() : closeOffset_{}, now_(0s)
{
}
void
ManualTimeKeeper::run(std::vector<std::string> const& servers)
{
}
auto
ManualTimeKeeper::now() const -> time_point
{
std::lock_guard lock(mutex_);
return now_;
}
auto
ManualTimeKeeper::closeTime() const -> time_point
{
std::lock_guard lock(mutex_);
return now_ + closeOffset_;
}
void
ManualTimeKeeper::adjustCloseTime(std::chrono::duration<std::int32_t> amount)
{
// Copied from TimeKeeper::adjustCloseTime
using namespace std::chrono;
auto const s = amount.count();
std::lock_guard lock(mutex_);
// Take large offsets, ignore small offsets,
// push the close time towards our wall time.
if (s > 1)
closeOffset_ += seconds((s + 3) / 4);
else if (s < -1)
closeOffset_ += seconds((s - 3) / 4);
else
closeOffset_ = (closeOffset_ * 3) / 4;
}
std::chrono::duration<std::int32_t>
ManualTimeKeeper::nowOffset() const
{
return {};
}
std::chrono::duration<std::int32_t>
ManualTimeKeeper::closeOffset() const
{
std::lock_guard lock(mutex_);
return closeOffset_;
}
void
ManualTimeKeeper::set(time_point now)
{
std::lock_guard lock(mutex_);
now_ = now;
}
auto
ManualTimeKeeper::adjust(std::chrono::system_clock::time_point when)
-> time_point
{
return time_point(std::chrono::duration_cast<duration>(
when.time_since_epoch() - days(10957)));
}
} // namespace test
} // namespace ripple

21
src/test/overlay/compression_test.cpp
View File

@@ -20,9 +20,9 @@
#include <ripple/app/ledger/Ledger.h>
#include <ripple/app/ledger/LedgerMaster.h>
#include <ripple/app/misc/Manifest.h>
#include <ripple/basics/random.h>
#include <ripple/beast/unit_test.h>
#include <ripple/beast/utility/Journal.h>
#include <ripple/core/TimeKeeper.h>
#include <ripple/overlay/Compression.h>
#include <ripple/overlay/Message.h>
#include <ripple/overlay/impl/Handshake.h>
@@ -227,8 +227,7 @@ public:
auto transaction = std::make_shared<protocol::TMTransaction>();
transaction->set_rawtransaction(usdTxBlob);
transaction->set_status(protocol::tsNEW);
auto tk = make_TimeKeeper(logs.journal("TimeKeeper"));
transaction->set_receivetimestamp(tk->now().time_since_epoch().count());
transaction->set_receivetimestamp(rand_int<std::uint64_t>());
transaction->set_deferred(true);
return transaction;
@@ -263,19 +262,23 @@ public:
ledgerData->set_error(protocol::TMReplyError::reNO_LEDGER);
ledgerData->mutable_nodes()->Reserve(n);
uint256 parentHash(0);
NetClock::duration const resolution{10};
NetClock::time_point ct{resolution};
for (int i = 0; i < n; i++)
{
LedgerInfo info;
auto tk = make_TimeKeeper(logs.journal("TimeKeeper"));
info.seq = i;
info.parentCloseTime = tk->now();
info.parentCloseTime = ct;
info.hash = ripple::sha512Half(i);
info.txHash = ripple::sha512Half(i + 1);
info.accountHash = ripple::sha512Half(i + 2);
info.parentHash = parentHash;
info.drops = XRPAmount(10);
info.closeTimeResolution = tk->now().time_since_epoch();
info.closeTime = tk->now();
info.closeTimeResolution = resolution;
info.closeTime = ct;
ct += resolution;
parentHash = ledgerHash(info);
Serializer nData;
ripple::addRaw(info, nData);
@@ -341,7 +344,7 @@ public:
Serializer s1;
st.add(s1);
list->set_signature(s1.data(), s1.size());
list->set_blob(strHex(s.getString()));
list->set_blob(strHex(s.slice()));
return list;
}
@@ -375,7 +378,7 @@ public:
st.add(s1);
auto& blob = *list->add_blobs();
blob.set_signature(s1.data(), s1.size());
blob.set_blob(strHex(s.getString()));
blob.set_blob(strHex(s.slice()));
return list;
}