external/srs
1
0
Fork 0
mirror of https://github.com/ossrs/srs.git synced 2025-03-09 15:49:59 +00:00
Code Issues Releases Wiki Activity

Refactor: Extract SrsNetworkKbps from SrsKbps. v5.0.53

Browse source
This commit is contained in:
winlin 2022-08-30 13:41:27 +08:00
parent 937605b18c
commit 2c259bd95b
20 changed files with 949 additions and 662 deletions
Download patch file Download diff file Expand all files Collapse all files

18
trunk/src/protocol/srs_protocol_http_client.cpp
View file

@ -253,8 +253,7 @@ SrsHttpClient::SrsHttpClient()
{
transport = NULL;
ssl_transport = NULL;
clk = new SrsWallClock();
kbps = new SrsKbps(clk);
kbps = new SrsNetworkKbps();
parser = NULL;
recv_timeout = timeout = SRS_UTIME_NO_TIMEOUT;
port = 0;
@ -263,9 +262,8 @@ SrsHttpClient::SrsHttpClient()
SrsHttpClient::~SrsHttpClient()
{
disconnect();
srs_freep(kbps);
srs_freep(clk);
srs_freep(parser);
}
@ -410,18 +408,18 @@ void SrsHttpClient::set_recv_timeout(srs_utime_t tm)
recv_timeout = tm;
}
void SrsHttpClient::kbps_sample(const char* label, int64_t age)
void SrsHttpClient::kbps_sample(const char* label, srs_utime_t age)
{
kbps->sample();
int sr = kbps->get_send_kbps();
int sr30s = kbps->get_send_kbps_30s();
int sr5m = kbps->get_send_kbps_5m();
int rr = kbps->get_recv_kbps();
int rr30s = kbps->get_recv_kbps_30s();
int rr5m = kbps->get_recv_kbps_5m();
srs_trace("<- %s time=%" PRId64 ", okbps=%d,%d,%d, ikbps=%d,%d,%d", label, age, sr, sr30s, sr5m, rr, rr30s, rr5m);
srs_trace("<- %s time=%" PRId64 ", okbps=%d,%d,%d, ikbps=%d,%d,%d", label, srsu2ms(age), sr, sr30s, sr5m, rr, rr30s, rr5m);
}
void SrsHttpClient::disconnect()
@ -450,9 +448,9 @@ srs_error_t SrsHttpClient::connect()
// Set the recv/send timeout in srs_utime_t.
transport->set_recv_timeout(recv_timeout);
transport->set_send_timeout(timeout);
kbps->set_io(transport, transport);
kbps->set_io(transport, transport);
if (schema_ != "https") {
return err;
}

7
trunk/src/protocol/srs_protocol_http_client.hpp
View file

@ -21,7 +21,7 @@ class SrsHttpUri;
class SrsHttpParser;
class ISrsHttpMessage;
class SrsStSocket;
class SrsKbps;
class SrsNetworkKbps;
class SrsWallClock;
class SrsTcpClient;
@ -63,8 +63,7 @@ private:
SrsTcpClient* transport;
SrsHttpParser* parser;
std::map<std::string, std::string> headers;
SrsKbps* kbps;
SrsWallClock* clk;
SrsNetworkKbps* kbps;
private:
// The timeout in srs_utime_t.
srs_utime_t timeout;
@ -103,7 +102,7 @@ public:
public:
virtual void set_recv_timeout(srs_utime_t tm);
public:
virtual void kbps_sample(const char* label, int64_t age);
virtual void kbps_sample(const char* label, srs_utime_t age);
private:
virtual void disconnect();
virtual srs_error_t connect();

223
trunk/src/protocol/srs_protocol_kbps.cpp
View file

@ -11,52 +11,46 @@
SrsKbpsSlice::SrsKbpsSlice(SrsWallClock* c)
{
clk = c;
io = NULL;
last_bytes = io_bytes_base = starttime = bytes = delta_bytes = 0;
starttime = 0;
bytes = 0;
}
SrsKbpsSlice::~SrsKbpsSlice()
{
}
int64_t SrsKbpsSlice::get_total_bytes()
{
return bytes + last_bytes - io_bytes_base;
}
void SrsKbpsSlice::sample()
{
srs_utime_t now = clk->now();
int64_t total_bytes = get_total_bytes();
if (sample_30s.time < 0) {
sample_30s.update(total_bytes, now, 0);
sample_30s.update(bytes, now, 0);
}
if (sample_1m.time < 0) {
sample_1m.update(total_bytes, now, 0);
sample_1m.update(bytes, now, 0);
}
if (sample_5m.time < 0) {
sample_5m.update(total_bytes, now, 0);
sample_5m.update(bytes, now, 0);
}
if (sample_60m.time < 0) {
sample_60m.update(total_bytes, now, 0);
sample_60m.update(bytes, now, 0);
}
if (now - sample_30s.time >= 30 * SRS_UTIME_SECONDS) {
int kbps = (int)((total_bytes - sample_30s.total) * 8 / srsu2ms(now - sample_30s.time));
sample_30s.update(total_bytes, now, kbps);
int kbps = (int)((bytes - sample_30s.total) * 8 / srsu2ms(now - sample_30s.time));
sample_30s.update(bytes, now, kbps);
}
if (now - sample_1m.time >= 60 * SRS_UTIME_SECONDS) {
int kbps = (int)((total_bytes - sample_1m.total) * 8 / srsu2ms(now - sample_1m.time));
sample_1m.update(total_bytes, now, kbps);
int kbps = (int)((bytes - sample_1m.total) * 8 / srsu2ms(now - sample_1m.time));
sample_1m.update(bytes, now, kbps);
}
if (now - sample_5m.time >= 300 * SRS_UTIME_SECONDS) {
int kbps = (int)((total_bytes - sample_5m.total) * 8 / srsu2ms(now - sample_5m.time));
sample_5m.update(total_bytes, now, kbps);
int kbps = (int)((bytes - sample_5m.total) * 8 / srsu2ms(now - sample_5m.time));
sample_5m.update(bytes, now, kbps);
}
if (now - sample_60m.time >= 3600 * SRS_UTIME_SECONDS) {
int kbps = (int)((total_bytes - sample_60m.total) * 8 / srsu2ms(now - sample_60m.time));
sample_60m.update(total_bytes, now, kbps);
int kbps = (int)((bytes - sample_60m.total) * 8 / srsu2ms(now - sample_60m.time));
sample_60m.update(bytes, now, kbps);
}
}
@ -138,57 +132,22 @@ void SrsNetworkDelta::remark(int64_t* in, int64_t* out)
in_delta_ = out_delta_ = 0;
}
SrsKbps::SrsKbps(SrsWallClock* c) : is(c), os(c)
SrsKbps::SrsKbps(SrsWallClock* c)
{
clk = c;
clk = c ? c : _srs_clock;
is = new SrsKbpsSlice(clk);
os = new SrsKbpsSlice(clk);
}
SrsKbps::~SrsKbps()
{
}
void SrsKbps::set_io(ISrsProtocolStatistic* in, ISrsProtocolStatistic* out)
{
// set input stream
// now, set start time.
if (is.starttime == 0) {
is.starttime = clk->now();
}
// save the old in bytes.
if (is.io) {
is.bytes += is.io->get_recv_bytes() - is.io_bytes_base;
}
// use new io.
is.io = in;
is.last_bytes = is.io_bytes_base = 0;
if (in) {
is.last_bytes = is.io_bytes_base = in->get_recv_bytes();
}
// resample
is.sample();
// set output stream
// now, set start time.
if (os.starttime == 0) {
os.starttime = clk->now();
}
// save the old in bytes.
if (os.io) {
os.bytes += os.io->get_send_bytes() - os.io_bytes_base;
}
// use new io.
os.io = out;
os.last_bytes = os.io_bytes_base = 0;
if (out) {
os.last_bytes = os.io_bytes_base = out->get_send_bytes();
}
// resample
os.sample();
srs_freep(is);
srs_freep(os);
}
int SrsKbps::get_send_kbps()
{
int duration = srsu2ms(clk->now() - is.starttime);
int duration = srsu2ms(clk->now() - is->starttime);
if (duration <= 0) {
return 0;
}
@ -199,7 +158,7 @@ int SrsKbps::get_send_kbps()
int SrsKbps::get_recv_kbps()
{
int duration = srsu2ms(clk->now() - os.starttime);
int duration = srsu2ms(clk->now() - os->starttime);
if (duration <= 0) {
return 0;
}
@ -210,90 +169,118 @@ int SrsKbps::get_recv_kbps()
int SrsKbps::get_send_kbps_30s()
{
return os.sample_30s.rate;
return os->sample_30s.rate;
}
int SrsKbps::get_recv_kbps_30s()
{
return is.sample_30s.rate;
return is->sample_30s.rate;
}
int SrsKbps::get_send_kbps_5m()
{
return os.sample_5m.rate;
return os->sample_5m.rate;
}
int SrsKbps::get_recv_kbps_5m()
{
return is.sample_5m.rate;
return is->sample_5m.rate;
}
void SrsKbps::add_delta(ISrsKbpsDelta* delta)
{
if (!delta) return;
int64_t in, out;
delta->remark(&in, &out);
add_delta(in, out);
}
void SrsKbps::add_delta(int64_t in, int64_t out)
{
// update the total bytes
is.last_bytes += in;
os.last_bytes += out;
is->bytes += in;
os->bytes += out;
// we donot sample, please use sample() to do resample.
}
void SrsKbps::sample()
{
// update the total bytes
if (os.io) {
os.last_bytes = os.io->get_send_bytes();
}
if (is.io) {
is.last_bytes = is.io->get_recv_bytes();
}
// resample
is.sample();
os.sample();
is->sample();
os->sample();
}
int64_t SrsKbps::get_send_bytes()
{
// we must calc the send bytes dynamically,
// to not depends on the sample(which used to calc the kbps).
// @read https://github.com/ossrs/srs/issues/588
// session start bytes.
int64_t bytes = os.bytes;
// When exists active session, use it to get the last bytes.
if (os.io) {
bytes += os.io->get_send_bytes() - os.io_bytes_base;
return bytes;
}
// When no active session, the last_bytes record the last valid bytes.
// TODO: Maybe the bellow bytes is zero, because the ios.io.out is NULL.
bytes += os.last_bytes - os.io_bytes_base;
return bytes;
return os->bytes;
}
int64_t SrsKbps::get_recv_bytes()
{
// we must calc the send bytes dynamically,
// to not depends on the sample(which used to calc the kbps).
// @read https://github.com/ossrs/srs/issues/588
// session start bytes.
int64_t bytes = is.bytes;
// When exists active session, use it to get the last bytes.
if (is.io) {
bytes += is.io->get_recv_bytes() - is.io_bytes_base;
return bytes;
}
// When no active session, the last_bytes record the last valid bytes.
// TODO: Maybe the bellow bytes is zero, because the ios.io.out is NULL.
bytes += is.last_bytes - is.io_bytes_base;
return bytes;
return is->bytes;
}
SrsNetworkKbps::SrsNetworkKbps(SrsWallClock* clock)
{
delta_ = new SrsNetworkDelta();
kbps_ = new SrsKbps(clock);
}
SrsNetworkKbps::~SrsNetworkKbps()
{
srs_freep(kbps_);
srs_freep(delta_);
}
void SrsNetworkKbps::set_io(ISrsProtocolStatistic* in, ISrsProtocolStatistic* out)
{
delta_->set_io(in, out);
}
void SrsNetworkKbps::sample()
{
kbps_->add_delta(delta_);
kbps_->sample();
}
int SrsNetworkKbps::get_send_kbps()
{
return kbps_->get_send_kbps();
}
int SrsNetworkKbps::get_recv_kbps()
{
return kbps_->get_recv_kbps();
}
int SrsNetworkKbps::get_send_kbps_30s()
{
return kbps_->get_send_kbps_30s();
}
int SrsNetworkKbps::get_recv_kbps_30s()
{
return kbps_->get_recv_kbps_30s();
}
int SrsNetworkKbps::get_send_kbps_5m()
{
return kbps_->get_send_kbps_5m();
}
int SrsNetworkKbps::get_recv_kbps_5m()
{
return kbps_->get_recv_kbps_5m();
}
int64_t SrsNetworkKbps::get_send_bytes()
{
return kbps_->get_send_bytes();
}
int64_t SrsNetworkKbps::get_recv_bytes()
{
return kbps_->get_recv_bytes();
}

144
trunk/src/protocol/srs_protocol_kbps.hpp
View file

@ -13,64 +13,42 @@
#include <srs_kernel_kbps.hpp>
/**
* a slice of kbps statistic, for input or output.
* a slice contains a set of sessions, which has a base offset of bytes,
* where a slice is:
* starttime(oldest session startup time)
* bytes(total bytes of previous sessions)
* io_bytes_base(bytes offset of current session)
* last_bytes(bytes of current session)
* so, the total send bytes now is:
* send_bytes = bytes + last_bytes - io_bytes_base
* so, the bytes sent duration current session is:
* send_bytes = last_bytes - io_bytes_base
* @remark use set_io to start new session.
* @remakr the slice is a data collection object driven by SrsKbps.
* The slice of kbps statistic, for input or output.
*/
class SrsKbpsSlice
{
private:
SrsWallClock* clk;
public:
// the slice io used for SrsKbps to invoke,
// the SrsKbpsSlice itself never use it.
ISrsProtocolStatistic* io;
// session startup bytes
// @remark, use total_bytes() to get the total bytes of slice.
int64_t bytes;
// slice starttime, the first time to record bytes.
srs_utime_t starttime;
// session startup bytes number for io when set it,
// the base offset of bytes for io.
int64_t io_bytes_base;
// last updated bytes number,
// cache for io maybe freed.
int64_t last_bytes;
// samples
SrsRateSample sample_30s;
SrsRateSample sample_1m;
SrsRateSample sample_5m;
SrsRateSample sample_60m;
public:
// for the delta bytes.
int64_t delta_bytes;
public:
SrsKbpsSlice(SrsWallClock* clk);
virtual ~SrsKbpsSlice();
public:
// Get current total bytes, it doesn't depend on sample().
virtual int64_t get_total_bytes();
// Resample the slice to calculate the kbps.
virtual void sample();
};
/**
* the interface which provices delta of bytes.
* for a delta, for example, a live stream connection, we can got the delta by:
* The interface which provices delta of bytes. For example, we got a delta from a TCP client:
* ISrsKbpsDelta* delta = ...;
* Now, we can add delta simple to a kbps:
* kbps->add_delta(delta);
* Or by multiple kbps:
* int64_t in, out;
* delta->remark(&in, &out);
* kbps->add_delta(in, out);
* kbps1->add_delta(in, out);
* kbpsN->add_delta(in, out);
* Then you're able to use the kbps object.
*/
class ISrsKbpsDelta
{
@ -78,9 +56,8 @@ public:
ISrsKbpsDelta();
virtual ~ISrsKbpsDelta();
public:
/**
* resample to generate the value of delta bytes.
*/
// Resample to get the value of delta bytes.
// @remark If no delta bytes, both in and out will be set to 0.
virtual void remark(int64_t* in, int64_t* out) = 0;
};
@ -115,6 +92,7 @@ public:
SrsNetworkDelta();
virtual ~SrsNetworkDelta();
public:
// Switch the under-layer network io, we use the bytes as a fresh delta.
virtual void set_io(ISrsProtocolStatistic* in, ISrsProtocolStatistic* out);
// Interface ISrsKbpsDelta.
public:
@ -122,86 +100,64 @@ public:
};
/**
* to statistic the kbps of io.
* itself can be a statistic source, for example, used for SRS bytes stat.
* there are some usage scenarios:
* 1. connections to calc kbps by sample():
* To statistic the kbps. For example, we got a set of connections and add the total delta:
* SrsKbps* kbps = ...;
* kbps->set_io(in, out)
* for conn in connections:
* kbps->add_delta(conn->delta()) // Which return an ISrsKbpsDelta object.
* Then we sample and got the total kbps:
* kbps->sample()
* kbps->get_xxx_kbps().
* the connections know how many bytes already send/recv.
* 2. server to calc kbps by add_delta():
* SrsKbps* kbps = ...;
* kbps->set_io(NULL, NULL)
* for each connection in connections:
* ISrsKbpsDelta* delta = connection; // where connection implements ISrsKbpsDelta
* int64_t in, out;
* delta->remark(&in, &out)
* kbps->add_delta(in, out)
* kbps->sample()
* kbps->get_xxx_kbps().
* 3. kbps used as ISrsProtocolStatistic, to provides raw bytes:
* SrsKbps* kbps = ...;
* kbps->set_io(in, out);
* // both kbps->get_recv_bytes() and kbps->get_send_bytes() are available.
* // we can use the kbps as the data source of another kbps:
* SrsKbps* user = ...;
* user->set_io(kbps, kbps);
* the server never know how many bytes already send/recv, for the connection maybe closed.
*/
class SrsKbps : public ISrsProtocolStatistic
class SrsKbps
{
private:
SrsKbpsSlice is;
SrsKbpsSlice os;
SrsKbpsSlice* is;
SrsKbpsSlice* os;
SrsWallClock* clk;
public:
// We won't free the clock c.
SrsKbps(SrsWallClock* c);
// Note that we won't free the clock c.
SrsKbps(SrsWallClock* c = NULL);
virtual ~SrsKbps();
public:
/**
* set io to start new session.
* set the underlayer reader/writer,
* if the io destroied, for instance, the forwarder reconnect,
* user must set the io of SrsKbps to NULL to continue to use the kbps object.
* @param in the input stream statistic. can be NULL.
* @param out the output stream statistic. can be NULL.
* @remark if in/out is NULL, use the cached data for kbps.
* @remark User must set_io(NULL, NULL) then free the in and out.
*/
virtual void set_io(ISrsProtocolStatistic* in, ISrsProtocolStatistic* out);
public:
/**
* get total kbps, duration is from the startup of io.
* @remark, use sample() to update data.
*/
// Get total average kbps.
virtual int get_send_kbps();
virtual int get_recv_kbps();
// 30s
// Get the average kbps in 30s.
virtual int get_send_kbps_30s();
virtual int get_recv_kbps_30s();
// 5m
// Get the average kbps in 5m or 300s.
virtual int get_send_kbps_5m();
virtual int get_recv_kbps_5m();
public:
/**
* add delta to kbps clac mechenism.
* we donot know the total bytes, but know the delta, for instance,
* for rtmp server to calc total bytes and kbps.
* @remark user must invoke sample() to calc result after invoke this method.
* @param delta, assert should never be NULL.
*/
// Add delta to kbps. Please call sample() after all deltas are added to kbps.
virtual void add_delta(int64_t in, int64_t out);
/**
* resample all samples, ignore if in/out is NULL.
* used for user to calc the kbps, to sample new kbps value.
* @remark if user, for instance, the rtmp server to calc the total bytes,
* use the add_delta() is better solutions.
*/
virtual void add_delta(ISrsKbpsDelta* delta);
// Sample the kbps to get the kbps in N seconds.
virtual void sample();
// Interface ISrsProtocolStatistic
public:
virtual int64_t get_send_bytes();
virtual int64_t get_recv_bytes();
};
// A sugar to use SrsNetworkDelta and SrsKbps.
class SrsNetworkKbps
{
private:
SrsNetworkDelta* delta_;
SrsKbps* kbps_;
public:
SrsNetworkKbps(SrsWallClock* c = NULL);
virtual ~SrsNetworkKbps();
public:
virtual void set_io(ISrsProtocolStatistic* in, ISrsProtocolStatistic* out);
virtual void sample();
public:
virtual int get_send_kbps();
virtual int get_recv_kbps();
virtual int get_send_kbps_30s();
virtual int get_recv_kbps_30s();
virtual int get_send_kbps_5m();
virtual int get_recv_kbps_5m();
public:
virtual int64_t get_send_bytes();
virtual int64_t get_recv_bytes();

24
trunk/src/protocol/srs_protocol_rtmp_conn.cpp
View file

@ -18,9 +18,8 @@ using namespace std;
SrsBasicRtmpClient::SrsBasicRtmpClient(string r, srs_utime_t ctm, srs_utime_t stm)
{
clk = new SrsWallClock();
kbps = new SrsKbps(clk);
kbps = new SrsNetworkKbps();
url = r;
connect_timeout = ctm;
stream_timeout = stm;
@ -39,7 +38,6 @@ SrsBasicRtmpClient::~SrsBasicRtmpClient()
{
close();
srs_freep(kbps);
srs_freep(clk);
srs_freep(req);
}
@ -52,7 +50,7 @@ srs_error_t SrsBasicRtmpClient::connect()
transport = new SrsTcpClient(req->host, req->port, srs_utime_t(connect_timeout));
client = new SrsRtmpClient(transport);
kbps->set_io(transport, transport);
if ((err = transport->connect()) != srs_success) {
close();
return srs_error_wrap(err, "connect");
@ -168,32 +166,32 @@ srs_error_t SrsBasicRtmpClient::play(int chunk_size, bool with_vhost, std::strin
return err;
}
void SrsBasicRtmpClient::kbps_sample(const char* label, int64_t age)
void SrsBasicRtmpClient::kbps_sample(const char* label, srs_utime_t age)
{
kbps->sample();
int sr = kbps->get_send_kbps();
int sr30s = kbps->get_send_kbps_30s();
int sr5m = kbps->get_send_kbps_5m();
int rr = kbps->get_recv_kbps();
int rr30s = kbps->get_recv_kbps_30s();
int rr5m = kbps->get_recv_kbps_5m();
srs_trace("<- %s time=%" PRId64 ", okbps=%d,%d,%d, ikbps=%d,%d,%d", label, age, sr, sr30s, sr5m, rr, rr30s, rr5m);
srs_trace("<- %s time=%" PRId64 ", okbps=%d,%d,%d, ikbps=%d,%d,%d", label, srsu2ms(age), sr, sr30s, sr5m, rr, rr30s, rr5m);
}
void SrsBasicRtmpClient::kbps_sample(const char* label, int64_t age, int msgs)
void SrsBasicRtmpClient::kbps_sample(const char* label, srs_utime_t age, int msgs)
{
kbps->sample();
int sr = kbps->get_send_kbps();
int sr30s = kbps->get_send_kbps_30s();
int sr5m = kbps->get_send_kbps_5m();
int rr = kbps->get_recv_kbps();
int rr30s = kbps->get_recv_kbps_30s();
int rr5m = kbps->get_recv_kbps_5m();
srs_trace("<- %s time=%" PRId64 ", msgs=%d, okbps=%d,%d,%d, ikbps=%d,%d,%d", label, age, msgs, sr, sr30s, sr5m, rr, rr30s, rr5m);
srs_trace("<- %s time=%" PRId64 ", msgs=%d, okbps=%d,%d,%d, ikbps=%d,%d,%d", label, srsu2ms(age), msgs, sr, sr30s, sr5m, rr, rr30s, rr5m);
}
int SrsBasicRtmpClient::sid()

9
trunk/src/protocol/srs_protocol_rtmp_conn.hpp
View file

@ -17,7 +17,7 @@ class SrsRtmpClient;
class SrsCommonMessage;
class SrsSharedPtrMessage;
class SrsPacket;
class SrsKbps;
class SrsNetworkKbps;
class SrsWallClock;
// The simple RTMP client, provides friendly APIs.
@ -38,8 +38,7 @@ protected:
private:
SrsTcpClient* transport;
SrsRtmpClient* client;
SrsKbps* kbps;
SrsWallClock* clk;
SrsNetworkKbps* kbps;
int stream_id;
public:
// Constructor.
@ -59,8 +58,8 @@ protected:
public:
virtual srs_error_t publish(int chunk_size, bool with_vhost = true, std::string* pstream = NULL);
virtual srs_error_t play(int chunk_size, bool with_vhost = true, std::string* pstream = NULL);
virtual void kbps_sample(const char* label, int64_t age);
virtual void kbps_sample(const char* label, int64_t age, int msgs);
virtual void kbps_sample(const char* label, srs_utime_t age);
virtual void kbps_sample(const char* label, srs_utime_t age, int msgs);
virtual int sid();
public:
virtual srs_error_t recv_message(SrsCommonMessage** pmsg);