1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
//
// Copyright Aliaksei Levin (levlam@telegram.org), Arseny Smirnov (arseny30@gmail.com) 2014-2026
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//
#pragma once
#include "td/actor/SchedulerLocalStorage.h"
#include "td/utils/common.h"
#include "td/utils/format.h"
#include "td/utils/StringBuilder.h"
#include "td/utils/Time.h"
#include <atomic>
#include <memory>
namespace td {
class NetStatsCallback {
public:
virtual void on_read(uint64 bytes) = 0;
virtual void on_write(uint64 bytes) = 0;
NetStatsCallback() = default;
NetStatsCallback(const NetStatsCallback &) = delete;
NetStatsCallback &operator=(const NetStatsCallback &) = delete;
virtual ~NetStatsCallback() = default;
};
struct NetStatsData {
uint64 read_size = 0;
uint64 write_size = 0;
uint64 count = 0;
double duration = 0;
};
inline NetStatsData operator+(const NetStatsData &a, const NetStatsData &b) {
NetStatsData res;
res.read_size = a.read_size + b.read_size;
res.write_size = a.write_size + b.write_size;
res.count = a.count + b.count;
res.duration = a.duration + b.duration;
return res;
}
inline NetStatsData operator-(const NetStatsData &a, const NetStatsData &b) {
NetStatsData res;
CHECK(a.read_size >= b.read_size);
res.read_size = a.read_size - b.read_size;
CHECK(a.write_size >= b.write_size);
res.write_size = a.write_size - b.write_size;
CHECK(a.count >= b.count);
res.count = a.count - b.count;
CHECK(a.duration >= b.duration);
res.duration = a.duration - b.duration;
return res;
}
inline StringBuilder &operator<<(StringBuilder &sb, const NetStatsData &data) {
return sb << tag("Rx size", format::as_size(data.read_size)) << tag("Tx size", format::as_size(data.write_size))
<< tag("count", data.count) << tag("duration", format::as_time(data.duration));
}
class NetStats {
public:
class Callback {
public:
virtual void on_stats_updated() = 0;
Callback() = default;
Callback(const Callback &) = delete;
Callback &operator=(const Callback &) = delete;
virtual ~Callback() = default;
};
std::shared_ptr<NetStatsCallback> get_callback() const {
return impl_;
}
NetStatsData get_stats() const {
return impl_->get_stats();
}
// do it before get_callback
void set_callback(unique_ptr<Callback> callback) {
impl_->set_callback(std::move(callback));
}
private:
class Impl final : public NetStatsCallback {
public:
NetStatsData get_stats() const {
NetStatsData res;
local_net_stats_.for_each([&](auto &stats) {
res.read_size += stats.read_size.load(std::memory_order_relaxed);
res.write_size += stats.write_size.load(std::memory_order_relaxed);
});
return res;
}
void set_callback(unique_ptr<Callback> callback) {
callback_ = std::move(callback);
}
private:
struct LocalNetStats {
double last_update = 0;
uint64 unsync_size = 0;
std::atomic<uint64> read_size{0};
std::atomic<uint64> write_size{0};
};
SchedulerLocalStorage<LocalNetStats> local_net_stats_;
unique_ptr<Callback> callback_;
void on_read(uint64 size) final {
auto &stats = local_net_stats_.get();
stats.read_size.fetch_add(size, std::memory_order_relaxed);
on_change(stats, size);
}
void on_write(uint64 size) final {
auto &stats = local_net_stats_.get();
stats.write_size.fetch_add(size, std::memory_order_relaxed);
on_change(stats, size);
}
void on_change(LocalNetStats &stats, uint64 size) {
stats.unsync_size += size;
auto now = Time::now();
if (stats.unsync_size > 10000 || now - stats.last_update > 300) {
stats.unsync_size = 0;
stats.last_update = now;
callback_->on_stats_updated();
}
}
};
std::shared_ptr<Impl> impl_{std::make_shared<Impl>()};
};
} // namespace td
|
