use std::chrono for time tracking

This commit is contained in:
Adrien Béraud committed 2014-12-24 01:13:43 -05:00
1 parent fde26fef5b
commit 33542e932a
8 files changed
+389 -331

No files matched your search

+67 -58
View File
@@ -34,6 +34,7 @@ THE SOFTWARE.
#include <vector>
#include <map>
#include <list>
#include <queue>
#include <functional>
#include <algorithm>
#include <memory>
@@ -65,7 +66,7 @@ public:
socklen_t sslen;
};
typedef int_fast8_t want_t;
using want_t = int_fast8_t;
Dht() {}
@@ -117,7 +118,7 @@ public:
int pingNode(const sockaddr*, socklen_t);
void periodic(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen, time_t *tosleep);
time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen);
/**
* Get a value by searching on all available protocols (IPv4, IPv6),
@@ -223,20 +224,26 @@ private:
static constexpr unsigned MAX_SEARCHES {1024};
/* Time for a request to timeout */
static constexpr time_t MAX_RESPONSE_TIME {6};
static constexpr std::chrono::seconds MAX_RESPONSE_TIME {6};
/* A search with no nodes will timeout after this time. */
static constexpr time_t SEARCH_TIMEOUT {6};
static constexpr std::chrono::seconds SEARCH_TIMEOUT {6};
/* The time after which we can send get requests for
a search in case of no answers. */
static constexpr time_t SEARCH_GET_STEP {6};
static constexpr std::chrono::seconds SEARCH_GET_STEP {6};
/* The time after which we consider a search to be expirable. */
static constexpr time_t SEARCH_EXPIRE_TIME {62 * 60};
static constexpr std::chrono::minutes SEARCH_EXPIRE_TIME {62};
/* The time after which we consider a node to be expirable. */
static constexpr time_t NODE_EXPIRE_TIME {15 * 60};
static constexpr std::chrono::minutes NODE_EXPIRE_TIME {15};
static constexpr std::chrono::minutes NODE_GOOD_TIME {120};
static constexpr std::chrono::seconds REANNOUNCE_MARGIN {3};
static constexpr std::chrono::seconds UDP_REPLY_TIME {15};
/* The maximum number of nodes that we snub. There is probably little
reason to increase this value. */
@@ -244,7 +251,7 @@ private:
static constexpr long unsigned MAX_REQUESTS_PER_SEC {400};
static constexpr time_t TOKEN_EXPIRE_TIME {10 * 60};
static constexpr std::chrono::seconds TOKEN_EXPIRE_TIME {10 * 60};
static constexpr unsigned TOKEN_SIZE {64};
@@ -252,29 +259,29 @@ private:
InfoHash id {};
sockaddr_storage ss;
socklen_t sslen {0};
time_t time {0}; /* time of last message received */
time_t reply_time {0}; /* time of last correct reply received */
time_t pinged_time {0}; /* time of last request */
time_point time {}; /* time of last message received */
time_point reply_time {}; /* time of last correct reply received */
time_point pinged_time {}; /* time of last request */
unsigned pinged {0}; /* how many requests we sent since last reply */
Node() {
std::fill_n((uint8_t*)&ss, sizeof(ss), 0);
}
Node(const InfoHash& id, const sockaddr* sa, socklen_t salen, time_t t, time_t reply_time)
Node(const InfoHash& id, const sockaddr* sa, socklen_t salen, time_point t, time_point reply_time)
: id(id), sslen(salen), time(t), reply_time(reply_time) {
std::copy_n((const uint8_t*)sa, salen, (uint8_t*)&ss);
}
bool isGood(time_t now) const;
bool isGood(time_point now) const;
NodeExport exportNode() const { return NodeExport {id, ss, sslen}; }
};
struct Bucket {
Bucket() : cached() {}
Bucket(sa_family_t af, const InfoHash& f = {}, time_t t = 0)
Bucket(sa_family_t af, const InfoHash& f = {}, time_point t = {})
: af(af), first(f), time(t), cached() {}
sa_family_t af {0};
InfoHash first {};
time_t time {0}; /* time of last reply in this bucket */
time_point time {}; /* time of last reply in this bucket */
std::list<Node> nodes {};
sockaddr_storage cached; /* the address of a likely candidate */
socklen_t cachedlen {0};
@@ -316,47 +323,52 @@ private:
SearchNode(const InfoHash& id) : id(id), ss() {}
struct RequestStatus {
time_t request_time; /* the time of the last unanswered announce request */
time_t reply_time; /* the time of the last announce confirmation */
time_point request_time {}; /* the time of the last unanswered announce request */
time_point reply_time {}; /* the time of the last announce confirmation */
RequestStatus() {};
RequestStatus(time_point q, time_point a = {}) : request_time(q), reply_time(a) {};
};
typedef std::map<Value::Id, RequestStatus> AnnounceStatusMap;
/**
* Can we use this node to listen/announce ?
*/
bool isSynced(time_t now) const {
bool isSynced(time_point now) const {
return /*pinged < 3 && replied &&*/ reply_time >= now - NODE_EXPIRE_TIME;
}
bool isAnnounced(Value::Id vid, const ValueType& type, time_t now) const {
bool isAnnounced(Value::Id vid, const ValueType& type, time_point now) const {
auto ack = acked.find(vid);
if (ack == acked.end()) {
return false;
}
return ack->second.reply_time + type.expiration > now;
}
time_t getAnnounceTime(AnnounceStatusMap::const_iterator ack, const ValueType& type) const {
time_point getAnnounceTime(AnnounceStatusMap::const_iterator ack, const ValueType& type) const {
if (ack == acked.end())
return request_time + 5;
return std::max<time_t>({ack->second.reply_time + type.expiration - 3, ack->second.request_time + 5, request_time + 5});
return request_time + MAX_RESPONSE_TIME;
return std::max<time_point>({
ack->second.reply_time + type.expiration - REANNOUNCE_MARGIN,
ack->second.request_time + MAX_RESPONSE_TIME,
request_time + MAX_RESPONSE_TIME
});
}
time_t getAnnounceTime(Value::Id vid, const ValueType& type) const {
time_point getAnnounceTime(Value::Id vid, const ValueType& type) const {
return getAnnounceTime(acked.find(vid), type);
}
time_t getListenTime() const {
time_t min_t = listenStatus.request_time + MAX_RESPONSE_TIME;
return listenStatus.reply_time ? std::max(listenStatus.reply_time + NODE_EXPIRE_TIME - 5, min_t) : min_t;
time_point getListenTime() const {
time_point min_t = listenStatus.request_time + MAX_RESPONSE_TIME;
return listenStatus.reply_time.time_since_epoch().count() ? std::max(listenStatus.reply_time + NODE_EXPIRE_TIME - REANNOUNCE_MARGIN, min_t) : min_t;
}
InfoHash id {};
sockaddr_storage ss;
socklen_t sslen {0};
time_t request_time {0}; /* the time of the last unanswered request */
time_t reply_time {0}; /* the time of the last reply with a token */
time_point request_time {}; /* the time of the last unanswered request */
time_point reply_time {}; /* the time of the last reply with a token */
unsigned pinged {0};
RequestStatus listenStatus {0, 0};
RequestStatus listenStatus {};
Blob token {};
@@ -368,7 +380,7 @@ private:
};
struct Get {
time_t start;
time_point start;
Value::Filter filter;
GetCallback get_cb;
DoneCallback done_cb;
@@ -398,7 +410,7 @@ private:
sa_family_t af;
uint16_t tid;
time_t step_time {0}; /* the time of the last search_step */
time_point step_time {}; /* the time of the last search_step */
bool done {false};
std::vector<SearchNode> nodes {SEARCH_NODES+1};
@@ -408,54 +420,54 @@ private:
std::map<size_t, LocalListener> listeners {};
size_t listener_token = 1;
bool insertNode(const InfoHash& id, const sockaddr*, socklen_t, time_t now, bool confirmed=false, const Blob& token={});
void insertBucket(const Bucket&, time_t now);
bool insertNode(const InfoHash& id, const sockaddr*, socklen_t, time_point now, bool confirmed=false, const Blob& token={});
void insertBucket(const Bucket&, time_point now);
/**
* Can we use this search to announce ?
*/
bool isSynced(time_t now) const;
bool isSynced(time_point now) const;
time_t getLastGetTime() const;
time_point getLastGetTime() const;
bool isUpdated(const SearchNode& sn) const;
/**
* Is this get operation done ?
*/
bool isDone(const Get& get, time_t now) const;
bool isDone(const Get& get, time_point now) const;
time_t getUpdateTime(time_t now) const;
time_point getUpdateTime(time_point now) const;
bool isAnnounced(Value::Id id, const ValueType& type, time_t now) const;
bool isAnnounced(Value::Id id, const ValueType& type, time_point now) const;
/**
* ret = 0 : no announce required.
* ret > 0 : (re-)announce required at time ret.
*/
time_t getAnnounceTime(const std::map<ValueType::Id, ValueType>& types) const;
time_point getAnnounceTime(const std::map<ValueType::Id, ValueType>& types) const;
/**
* ret = 0 : no listen required.
* ret > 0 : (re-)announce required at time ret.
*/
time_t getListenTime(time_t now) const;
time_point getListenTime(time_point now) const;
/**
* ret = 0 : no expiration
* ret > 0 : search will expire at time ret
*/
time_t getExpireTime() const;
time_point getExpireTime() const;
time_t getNextStepTime(const std::map<ValueType::Id, ValueType>& types, time_t now) const;
time_point getNextStepTime(const std::map<ValueType::Id, ValueType>& types, time_point now) const;
};
struct ValueStorage {
std::shared_ptr<Value> data {};
time_t time {0};
time_point time {};
ValueStorage() {}
ValueStorage(const std::shared_ptr<Value>& v, time_t t) : data(v), time(t) {}
ValueStorage(const std::shared_ptr<Value>& v, time_point t) : data(v), time(t) {}
};
/**
@@ -466,13 +478,13 @@ private:
sockaddr_storage ss;
socklen_t sslen {};
uint16_t tid {};
time_t time {};
time_point time {};
constexpr Listener() : ss() {}
Listener(const InfoHash& id, const sockaddr *from, socklen_t fromlen, uint16_t ttid, time_t t) : id(id), ss(), sslen(fromlen), tid(ttid), time(t) {
Listener(const InfoHash& id, const sockaddr *from, socklen_t fromlen, uint16_t ttid, time_point t) : id(id), ss(), sslen(fromlen), tid(ttid), time(t) {
memcpy(&ss, from, fromlen);
}
void refresh(const sockaddr *from, socklen_t fromlen, uint16_t ttid, time_t t) {
void refresh(const sockaddr *from, socklen_t fromlen, uint16_t ttid, time_point t) {
memcpy(&ss, from, fromlen);
sslen = fromlen;
tid = ttid;
@@ -573,16 +585,13 @@ private:
sockaddr_storage blacklist[BLACKLISTED_MAX] {};
unsigned next_blacklisted = 0;
struct timeval now {0, 0};
time_t mybucket_grow_time {0}, mybucket6_grow_time {0};
time_t expire_stuff_time {0};
time_t search_time {0};
time_t confirm_nodes_time {0};
time_t rotate_secrets_time {0};
time_t rate_limit_time {0};
// remaining requests for this second
long unsigned rate_limit_tokens {MAX_REQUESTS_PER_SEC};
time_point now;
time_point mybucket_grow_time {}, mybucket6_grow_time {};
time_point expire_stuff_time {};
time_point search_time {};
time_point confirm_nodes_time {};
time_point rotate_secrets_time {};
std::queue<time_point> rate_limit_time {};
// Networking & packet handling
int send(const char* buf, size_t len, int flags, const sockaddr*, socklen_t);
+20 -21
View File
@@ -85,48 +85,48 @@ public:
void dumpTables() const
{
std::unique_lock<std::mutex> lck(dht_mtx);
dht->dumpTables();
dht_->dumpTables();
}
InfoHash getId() const {
if (!dht)
if (!dht_)
return {};
return dht->getId();
return dht_->getId();
}
InfoHash getRoutingId() const {
if (!dht)
if (!dht_)
return {};
return dht->getRoutingId();
return dht_->getRoutingId();
}
std::vector<Dht::NodeExport> exportNodes() const {
std::unique_lock<std::mutex> lck(dht_mtx);
if (!dht)
if (!dht_)
return {};
return dht->exportNodes();
return dht_->exportNodes();
}
std::vector<Dht::ValuesExport> exportValues() const {
std::unique_lock<std::mutex> lck(dht_mtx);
if (!dht)
if (!dht_)
return {};
return dht->exportValues();
return dht_->exportValues();
}
void setLoggers(LogMethod&& error = NOLOG, LogMethod&& warn = NOLOG, LogMethod&& debug = NOLOG) {
std::unique_lock<std::mutex> lck(dht_mtx);
dht->setLoggers(std::forward<LogMethod>(error), std::forward<LogMethod>(warn), std::forward<LogMethod>(debug));
dht_->setLoggers(std::forward<LogMethod>(error), std::forward<LogMethod>(warn), std::forward<LogMethod>(debug));
}
void registerType(const ValueType& type) {
std::unique_lock<std::mutex> lck(dht_mtx);
dht->registerType(type);
dht_->registerType(type);
}
void importValues(const std::vector<Dht::ValuesExport>& values) {
std::unique_lock<std::mutex> lck(dht_mtx);
dht->importValues(values);
dht_->importValues(values);
}
bool isRunning() const {
@@ -136,23 +136,23 @@ public:
int getNodesStats(sa_family_t af, unsigned *good_return, unsigned *dubious_return, unsigned *cached_return, unsigned *incoming_return) const
{
std::unique_lock<std::mutex> lck(dht_mtx);
return dht->getNodesStats(af, good_return, dubious_return, cached_return, incoming_return);
return dht_->getNodesStats(af, good_return, dubious_return, cached_return, incoming_return);
}
std::string getStorageLog() const
{
std::unique_lock<std::mutex> lck(dht_mtx);
return dht->getStorageLog();
return dht_->getStorageLog();
}
std::string getRoutingTablesLog(sa_family_t af) const
{
std::unique_lock<std::mutex> lck(dht_mtx);
return dht->getRoutingTablesLog(af);
return dht_->getRoutingTablesLog(af);
}
std::string getSearchesLog(sa_family_t af) const
{
std::unique_lock<std::mutex> lck(dht_mtx);
return dht->getSearchesLog(af);
return dht_->getSearchesLog(af);
}
/**
@@ -160,9 +160,9 @@ public:
*/
void run(in_port_t port, const crypto::Identity identity, bool threaded = false, StatusCallback cb = nullptr);
void loop() {
time_point loop() {
std::unique_lock<std::mutex> lck(dht_mtx);
loop_();
return loop_();
}
void join();
@@ -170,9 +170,9 @@ public:
private:
void doRun(in_port_t port, const crypto::Identity identity);
void loop_();
time_point loop_();
std::unique_ptr<SecureDht> dht {};
std::unique_ptr<SecureDht> dht_ {};
mutable std::mutex dht_mtx {};
std::thread dht_thread {};
std::condition_variable cv {};
@@ -180,7 +180,6 @@ private:
std::thread rcv_thread {};
std::mutex sock_mtx {};
std::vector<std::pair<Blob, sockaddr_storage>> rcv {};
std::atomic<time_t> tosleep {0};
std::queue<std::function<void(SecureDht&)>> pending_ops {};
std::mutex storage_mtx {};
+1 -1
View File
@@ -136,7 +136,7 @@ private:
};
const ValueType CERTIFICATE_TYPE = {
8, "Certificate", 60 * 60 * 24 * 7,
8, "Certificate", std::chrono::hours(24 * 7),
// A certificate can only be stored at it's public key ID.
[](InfoHash id, std::shared_ptr<Value>& v, InfoHash, const sockaddr*, socklen_t) {
try {
+23
View File
@@ -32,6 +32,7 @@
#include <tuple>
#include <numeric>
#include <limits>
#include <chrono>
typedef std::vector<uint8_t> Blob;
@@ -70,6 +71,13 @@ namespace detail {
}
};
template <class T>
struct get_size_helper<std::chrono::time_point<T>> {
static constexpr size_t value(const std::chrono::time_point<T>&) {
return sizeof(typename std::chrono::time_point<T>::rep);
}
};
template <class tuple_type>
inline size_t get_tuple_size(const tuple_type& obj, int_<0>) {
constexpr size_t idx = std::tuple_size<tuple_type>::value-1;
@@ -159,6 +167,13 @@ namespace detail {
};
template <class T>
struct serialize_helper<std::chrono::time_point<T>> {
static void apply(const std::chrono::time_point<T>& obj, Blob::iterator& res) {
serializer(obj.time_since_epoch().count(), res);
}
};
template <class T>
struct serialize_helper {
static void apply(const T& obj, Blob::iterator& res) {
@@ -207,6 +222,14 @@ namespace detail {
}
};
template <class T>
struct deserialize_helper<std::chrono::time_point<T>> {
static std::chrono::time_point<T> apply(Blob::const_iterator& begin,
Blob::const_iterator end) {
return std::chrono::time_point<T>(typename T::duration(deserialize_helper<typename T::rep>::apply(begin,end)));
}
};
template <class T>
struct deserialize_helper<std::vector<T>> {
static std::vector<T> apply(Blob::const_iterator& begin,
+27 -3
View File
@@ -54,11 +54,35 @@ typedef uint16_t in_port_t;
#include <algorithm>
#include <functional>
#include <memory>
#include <chrono>
#include <cstdarg>
namespace dht {
using clock = std::chrono::steady_clock;
using time_point = clock::time_point;
using duration = clock::duration;
static constexpr time_point TIME_INVALID = {};
static constexpr time_point TIME_MAX {duration{std::numeric_limits<duration::rep>::max()}};
template <typename Duration = duration>
class uniform_duration_distribution : public std::uniform_int_distribution<typename Duration::rep> {
using Base = std::uniform_int_distribution<typename Duration::rep>;
using param_type = typename Base::param_type;
public:
uniform_duration_distribution(Duration min, Duration max) : Base(min.count(), max.count()) {}
template <class Generator>
Duration operator()(Generator && g) {
return Duration(Base::operator()(g));
}
template< class Generator >
Duration operator()( Generator && g, const param_type& params ) {
return Duration(Base::operator()(g, params));
}
};
/**
* Wrapper for logging methods
*/
@@ -100,10 +124,10 @@ struct ValueType {
typedef uint16_t Id;
ValueType () {}
ValueType (Id id, std::string name, time_t e = 60 * 60)
ValueType (Id id, std::string name, duration e = std::chrono::hours(1))
: id(id), name(name), expiration(e) {}
ValueType (Id id, std::string name, time_t e, StorePolicy&& sp, EditPolicy&& ep)
ValueType (Id id, std::string name, duration e, StorePolicy&& sp, EditPolicy&& ep)
: id(id), name(name), expiration(e), storePolicy(std::move(sp)), editPolicy(std::move(ep)) {}
virtual ~ValueType() {}
@@ -124,7 +148,7 @@ struct ValueType {
Id id {0};
std::string name {};
time_t expiration {60 * 60};
duration expiration {60 * 60};
StorePolicy storePolicy {DEFAULT_STORE_POLICY};
EditPolicy editPolicy {DEFAULT_EDIT_POLICY};
};
+220 -215
View File
File diff suppressed because it is too large. Load diff
+30 -32
View File
@@ -47,8 +47,8 @@ DhtRunner::run(in_port_t port, const crypto::Identity identity, bool threaded, S
dht_thread = std::thread([this]() {
while (running) {
std::unique_lock<std::mutex> lk(dht_mtx);
loop_();
cv.wait_for(lk, std::chrono::seconds( tosleep ), [this]() {
auto wakeup = loop_();
cv.wait_until(lk, wakeup, [this]() {
if (not running)
return true;
{
@@ -78,57 +78,60 @@ DhtRunner::join()
rcv_thread.join();
{
std::unique_lock<std::mutex> lck(dht_mtx);
dht.reset();
dht_.reset();
status4 = Dht::Status::Disconnected;
status6 = Dht::Status::Disconnected;
}
}
void
time_point
DhtRunner::loop_()
{
if (!dht) return;
time_t tosl = 5;
{
std::unique_lock<std::mutex> lck(sock_mtx);
if (!dht) return;
if (rcv.size()) {
for (const auto& pck : rcv) {
auto& buf = pck.first;
auto& from = pck.second;
dht->periodic(buf.data(), buf.size()-1, (sockaddr*)&from, from.ss_family == AF_INET ? sizeof(sockaddr_in) : sizeof(sockaddr_in6), &tosl);
}
rcv.clear();
} else {
dht->periodic(nullptr, 0, nullptr, 0, &tosl);
}
}
tosleep = tosl;
if (!dht_)
return {};
decltype(pending_ops) ops {};
{
std::unique_lock<std::mutex> lck(storage_mtx);
ops = std::move(pending_ops);
}
while (not ops.empty()) {
ops.front()(*dht);
ops.front()(*dht_);
ops.pop();
}
time_point wakeup {};
{
std::unique_lock<std::mutex> lck(sock_mtx);
if (not rcv.empty()) {
for (const auto& pck : rcv) {
auto& buf = pck.first;
auto& from = pck.second;
wakeup = dht_->periodic(buf.data(), buf.size()-1, (sockaddr*)&from, from.ss_family == AF_INET ? sizeof(sockaddr_in) : sizeof(sockaddr_in6));
}
rcv.clear();
} else {
wakeup = dht_->periodic(nullptr, 0, nullptr, 0);
}
}
if (statusCb) {
Dht::Status nstatus4 = dht->getStatus(AF_INET);
Dht::Status nstatus6 = dht->getStatus(AF_INET6);
Dht::Status nstatus4 = dht_->getStatus(AF_INET);
Dht::Status nstatus6 = dht_->getStatus(AF_INET6);
if (nstatus4 != status4 || nstatus6 != status6) {
status4 = nstatus4;
status6 = nstatus6;
statusCb(status4, status6);
}
}
return wakeup;
}
void
DhtRunner::doRun(in_port_t port, const crypto::Identity identity)
{
dht.reset();
dht_.reset();
int s = socket(PF_INET, SOCK_DGRAM, 0);
int s6 = socket(PF_INET6, SOCK_DGRAM, 0);
@@ -159,22 +162,17 @@ DhtRunner::doRun(in_port_t port, const crypto::Identity identity)
throw DhtException("Can't bind IPv6 socket");
}
dht = std::unique_ptr<SecureDht>(new SecureDht {s, s6, identity});
dht_ = std::unique_ptr<SecureDht>(new SecureDht {s, s6, identity});
rcv_thread = std::thread([this,s,s6]() {
std::mt19937 engine(std::random_device{}());
auto rand_delay = std::bind(std::uniform_int_distribution<uint32_t>(0, 1000000), engine);
try {
while (true) {
uint8_t buf[4096 * 64];
sockaddr_storage from;
socklen_t fromlen;
struct timeval tv;
struct timeval tv {.tv_sec = 3, .tv_usec = 0};
fd_set readfds;
tv.tv_sec = std::min<time_t>(tosleep.load(), 5);
tv.tv_usec = rand_delay();
//std::cout << "Dht::rcv_thread loop " << tv.tv_sec << "." << tv.tv_usec << std::endl;
FD_ZERO(&readfds);
if(s >= 0)
+1 -1
View File
@@ -79,7 +79,7 @@ ServiceAnnouncement::storePolicy(InfoHash, std::shared_ptr<Value>& v, InfoHash,
return true;
}
const ValueType ServiceAnnouncement::TYPE = {1, "Service Announcement", 15 * 60, ServiceAnnouncement::storePolicy, ValueType::DEFAULT_EDIT_POLICY};
const ValueType ServiceAnnouncement::TYPE = {1, "Service Announcement", std::chrono::minutes(15), ServiceAnnouncement::storePolicy, ValueType::DEFAULT_EDIT_POLICY};
void
Value::packToSign(Blob& res) const