mirror of
https://github.com/Xahau/xahaud.git
synced 2025-12-06 17:27:52 +00:00
Add TimeKeeper:
This class tracks Ripple network time and closing time. Conflicts: src/ripple/ledger/ReadView.h
This commit is contained in:
Vinnie Falco
466
src/ripple/core/impl/SNTPClock.cpp
Normal file
466
src/ripple/core/impl/SNTPClock.cpp
Normal file
@@ -0,0 +1,466 @@
|
||||
//------------------------------------------------------------------------------
|
||||
/*
|
||||
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 <BeastConfig.h>
|
||||
#include <ripple/basics/Log.h>
|
||||
#include <ripple/basics/ThreadName.h>
|
||||
#include <ripple/crypto/RandomNumbers.h>
|
||||
#include <ripple/core/impl/SNTPClock.h>
|
||||
#include <beast/asio/placeholders.h>
|
||||
#include <beast/threads/Thread.h>
|
||||
#include <boost/asio.hpp>
|
||||
#include <boost/optional.hpp>
|
||||
#include <cmath>
|
||||
#include <deque>
|
||||
#include <map>
|
||||
#include <beast/cxx14/memory.h> // <memory>
|
||||
#include <mutex>
|
||||
#include <thread>
|
||||
|
||||
namespace ripple {
|
||||
|
||||
// #define SNTP_DEBUG
|
||||
|
||||
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 };
|
||||
|
||||
// NTP query frequency - 4 minutes
|
||||
#define NTP_QUERY_FREQUENCY (4 * 60)
|
||||
|
||||
// NTP minimum interval to query same servers - 3 minutes
|
||||
#define NTP_MIN_QUERY (3 * 60)
|
||||
|
||||
// NTP sample window (should be odd)
|
||||
#define NTP_SAMPLE_WINDOW 9
|
||||
|
||||
// NTP timestamp constant
|
||||
#define NTP_UNIX_OFFSET 0x83AA7E80
|
||||
|
||||
// NTP timestamp validity
|
||||
#define 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:
|
||||
struct Query
|
||||
{
|
||||
bool replied;
|
||||
time_t sent; // VFALCO time_t, really?
|
||||
std::uint32_t nonce;
|
||||
|
||||
Query (time_t j = time_t(-1))
|
||||
: replied (false)
|
||||
, sent (j)
|
||||
{
|
||||
}
|
||||
};
|
||||
|
||||
beast::Journal j_;
|
||||
std::mutex mutable mutex_;
|
||||
std::thread thread_;
|
||||
boost::asio::io_service io_service_;
|
||||
boost::optional<
|
||||
boost::asio::io_service::work> work_;
|
||||
|
||||
std::map <boost::asio::ip::udp::endpoint, Query> queries_;
|
||||
boost::asio::ip::udp::socket socket_;
|
||||
boost::asio::deadline_timer timer_;
|
||||
boost::asio::ip::udp::resolver resolver_;
|
||||
std::vector<std::pair<std::string, time_t>> servers_;
|
||||
int offset_;
|
||||
time_t lastUpdate_;
|
||||
std::deque<int> 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_ (time_t(-1))
|
||||
, buf_ (256)
|
||||
{
|
||||
}
|
||||
|
||||
~SNTPClientImp ()
|
||||
{
|
||||
if (thread_.joinable())
|
||||
{
|
||||
error_code ec;
|
||||
timer_.cancel(ec);
|
||||
socket_.cancel(ec);
|
||||
work_ = boost::none;
|
||||
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<std::mutex> lock (mutex_);
|
||||
for (auto const& item : servers)
|
||||
servers_.emplace_back(
|
||||
item, time_t(-1));
|
||||
}
|
||||
queryAll();
|
||||
|
||||
using namespace boost::asio;
|
||||
socket_.open (ip::udp::v4 ());
|
||||
socket_.async_receive_from (buffer (buf_, 256),
|
||||
ep_, std::bind(
|
||||
&SNTPClientImp::onRead, this,
|
||||
beast::asio::placeholders::error,
|
||||
beast::asio::placeholders::bytes_transferred));
|
||||
timer_.expires_from_now(
|
||||
boost::posix_time::seconds(NTP_QUERY_FREQUENCY));
|
||||
timer_.async_wait(std::bind(
|
||||
&SNTPClientImp::onTimer, this,
|
||||
beast::asio::placeholders::error));
|
||||
|
||||
// VFALCO Is it correct to launch the thread
|
||||
// here after queuing I/O?
|
||||
//
|
||||
thread_ = std::thread(&SNTPClientImp::doRun, this);
|
||||
}
|
||||
|
||||
time_point
|
||||
now() const override
|
||||
{
|
||||
std::lock_guard<std::mutex> lock (mutex_);
|
||||
auto const when = clock_type::now();
|
||||
if ((lastUpdate_ == (time_t)-1) ||
|
||||
((lastUpdate_ + NTP_TIMESTAMP_VALID) < time(nullptr)))
|
||||
return when;
|
||||
return when + std::chrono::seconds(offset_);
|
||||
}
|
||||
|
||||
duration
|
||||
offset() const override
|
||||
{
|
||||
std::lock_guard<std::mutex> lock (mutex_);
|
||||
return std::chrono::seconds(offset_);
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------
|
||||
|
||||
void doRun ()
|
||||
{
|
||||
setCallingThreadName("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(
|
||||
boost::posix_time::seconds (NTP_QUERY_FREQUENCY));
|
||||
timer_.async_wait(std::bind(
|
||||
&SNTPClientImp::onTimer, this,
|
||||
beast::asio::placeholders::error));
|
||||
}
|
||||
|
||||
void
|
||||
onRead (error_code const& ec, std::size_t bytes_xferd)
|
||||
{
|
||||
using namespace boost::asio;
|
||||
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<std::mutex> 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 (nullptr) > (query->second.sent + 1))
|
||||
{
|
||||
JLOG(j_.warning) <<
|
||||
"SNTP: Late response from " << ep_;
|
||||
}
|
||||
else if (bytes_xferd < 48)
|
||||
{
|
||||
JLOG(j_.warning) <<
|
||||
"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_.warning) <<
|
||||
"SNTP: Reply from " << ep_ << "had wrong nonce";
|
||||
}
|
||||
else
|
||||
{
|
||||
processReply ();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
socket_.async_receive_from(buffer(buf_, 256),
|
||||
ep_, std::bind(&SNTPClientImp::onRead, this,
|
||||
beast::asio::placeholders::error,
|
||||
beast::asio::placeholders::bytes_transferred));
|
||||
}
|
||||
|
||||
//--------------------------------------------------------------------------
|
||||
|
||||
void addServer (std::string const& server)
|
||||
{
|
||||
std::lock_guard<std::mutex> lock (mutex_);
|
||||
servers_.push_back (std::make_pair (server, time_t(-1)));
|
||||
}
|
||||
|
||||
void queryAll ()
|
||||
{
|
||||
while (doQuery ())
|
||||
{
|
||||
}
|
||||
}
|
||||
|
||||
bool doQuery ()
|
||||
{
|
||||
std::lock_guard<std::mutex> lock (mutex_);
|
||||
std::vector< std::pair<std::string, time_t> >::iterator best = servers_.end ();
|
||||
|
||||
for (auto iter = servers_.begin (), end = best;
|
||||
iter != end; ++iter)
|
||||
if ((best == end) || (iter->second == time_t(-1)) || (iter->second < best->second))
|
||||
best = iter;
|
||||
|
||||
if (best == servers_.end ())
|
||||
{
|
||||
JLOG(j_.trace) <<
|
||||
"SNTP: No server to query";
|
||||
return false;
|
||||
}
|
||||
|
||||
time_t now = time (nullptr);
|
||||
|
||||
if ((best->second != time_t(-1)) && ((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,
|
||||
beast::asio::placeholders::error,
|
||||
beast::asio::placeholders::iterator));
|
||||
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;
|
||||
}
|
||||
|
||||
ip::udp::resolver::iterator sel = it;
|
||||
int i = 1;
|
||||
|
||||
while (++it != ip::udp::resolver::iterator())
|
||||
if ((rand () % ++i) == 0)
|
||||
sel = it;
|
||||
|
||||
if (sel != ip::udp::resolver::iterator ())
|
||||
{
|
||||
std::lock_guard<std::mutex> lock (mutex_);
|
||||
Query& query = queries_[*sel];
|
||||
time_t now = time (nullptr);
|
||||
|
||||
if ((query.sent == now) || ((query.sent + 1) == 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;
|
||||
random_fill (&query.nonce);
|
||||
reinterpret_cast<std::uint32_t*> (SNTPQueryData)[NTP_OFF_XMITTS_INT] = static_cast<std::uint32_t> (time (nullptr)) + NTP_UNIX_OFFSET;
|
||||
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,
|
||||
beast::asio::placeholders::error,
|
||||
beast::asio::placeholders::bytes_transferred));
|
||||
}
|
||||
}
|
||||
|
||||
void onSend (error_code const& ec, std::size_t)
|
||||
{
|
||||
if (ec == boost::asio::error::operation_aborted)
|
||||
return;
|
||||
|
||||
if (ec)
|
||||
{
|
||||
JLOG(j_.warning) <<
|
||||
"SNTPClock::onSend: " << ec.message();
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
void processReply ()
|
||||
{
|
||||
assert (buf_.size () >= 48);
|
||||
std::uint32_t* recvBuffer = reinterpret_cast<std::uint32_t*> (&buf_.front ());
|
||||
|
||||
unsigned info = ntohl (recvBuffer[NTP_OFF_INFO]);
|
||||
int64_t timev = 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;
|
||||
}
|
||||
|
||||
std::int64_t now = static_cast<int> (time (nullptr));
|
||||
timev -= now;
|
||||
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_ == -1) || (offset_ == 1))
|
||||
offset_ = 0;
|
||||
|
||||
if (timev || offset_)
|
||||
{
|
||||
JLOG(j_.trace) << "SNTP: Offset is " << timev <<
|
||||
", new system offset is " << offset_;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
//------------------------------------------------------------------------------
|
||||
|
||||
std::unique_ptr<SNTPClock>
|
||||
make_SNTPClock (beast::Journal j)
|
||||
{
|
||||
return std::make_unique<SNTPClientImp>(j);
|
||||
}
|
||||
|
||||
} // ripple
|
||||
Reference in New Issue
Block a user