From 1630918b0fa511f2caf0d03241004f4cd47a6aba Mon Sep 17 00:00:00 2001 From: winlin Date: Tue, 30 Aug 2022 07:46:12 +0800 Subject: [PATCH] Refactor: Use compositor for ISrsKbpsDelta. v5.0.51 --- trunk/doc/CHANGELOG.md | 1 + trunk/src/app/srs_app_caster_flv.cpp | 15 ++--- trunk/src/app/srs_app_caster_flv.hpp | 8 +-- trunk/src/app/srs_app_conn.cpp | 8 --- trunk/src/app/srs_app_conn.hpp | 8 --- trunk/src/app/srs_app_http_conn.cpp | 18 +++--- trunk/src/app/srs_app_http_conn.hpp | 20 ++---- trunk/src/app/srs_app_http_static.cpp | 10 +-- trunk/src/app/srs_app_rtc_conn.cpp | 4 +- trunk/src/app/srs_app_rtc_conn.hpp | 5 +- trunk/src/app/srs_app_rtc_server.cpp | 5 +- trunk/src/app/srs_app_rtmp_conn.cpp | 30 +++------ trunk/src/app/srs_app_rtmp_conn.hpp | 14 ++-- trunk/src/app/srs_app_server.cpp | 31 ++++++--- trunk/src/app/srs_app_server.hpp | 2 +- trunk/src/app/srs_app_srt_conn.cpp | 30 +++------ trunk/src/app/srs_app_srt_conn.hpp | 9 ++- trunk/src/app/srs_app_srt_server.cpp | 17 +++-- trunk/src/app/srs_app_srt_server.hpp | 2 +- trunk/src/app/srs_app_statistic.cpp | 11 ++-- trunk/src/core/srs_core_version5.hpp | 2 +- trunk/src/protocol/srs_protocol_kbps.cpp | 70 +++++++++++++------- trunk/src/protocol/srs_protocol_kbps.hpp | 37 ++++++----- trunk/src/utest/srs_utest_protocol.cpp | 81 ++++++++++++++++-------- 24 files changed, 225 insertions(+), 213 deletions(-) diff --git a/trunk/doc/CHANGELOG.md b/trunk/doc/CHANGELOG.md index 8c036866b..584f0ca47 100644 --- a/trunk/doc/CHANGELOG.md +++ b/trunk/doc/CHANGELOG.md @@ -7,6 +7,7 @@ The changelog for SRS. ## SRS 5.0 Changelog +* v5.0, 2022-08-30, Refactor: Use compositor for ISrsKbpsDelta. v5.0.51 * v5.0, 2022-08-29, RTC: Stat the WebRTC clients bandwidth. v5.0.50 * v5.0, 2022-08-29, HLS: Stat the HLS streaming clients bandwidth. v5.0.49 * v5.0, 2022-08-28, URL: Use SrsHttpUri to parse URL and query. v5.0.48 diff --git a/trunk/src/app/srs_app_caster_flv.cpp b/trunk/src/app/srs_app_caster_flv.cpp index 483babb32..7c5db1d3b 100644 --- a/trunk/src/app/srs_app_caster_flv.cpp +++ b/trunk/src/app/srs_app_caster_flv.cpp @@ -68,9 +68,9 @@ srs_error_t SrsAppCasterFlv::on_tcp_client(srs_netfd_t stfd) srs_warn("empty ip for fd=%d", srs_netfd_fileno(stfd)); } - ISrsStartableConneciton* conn = new SrsDynamicHttpConn(this, stfd, http_mux, ip, port); + SrsDynamicHttpConn* conn = new SrsDynamicHttpConn(this, stfd, http_mux, ip, port); conns.push_back(conn); - + if ((err = conn->start()) != srs_success) { return srs_error_wrap(err, "start tcp listener"); } @@ -80,14 +80,14 @@ srs_error_t SrsAppCasterFlv::on_tcp_client(srs_netfd_t stfd) void SrsAppCasterFlv::remove(ISrsResource* c) { - ISrsStartableConneciton* conn = dynamic_cast(c); + ISrsConnection* conn = dynamic_cast(c); - std::vector::iterator it; + std::vector::iterator it; if ((it = std::find(conns.begin(), conns.end(), conn)) != conns.end()) { conns.erase(it); } - // fixbug: ISrsStartableConneciton for CasterFlv is not freed, which could cause memory leak + // fixbug: ISrsConnection for CasterFlv is not freed, which could cause memory leak // so, free conn which is not managed by SrsServer->conns; // @see: https://github.com/ossrs/srs/issues/826 manager->remove(c); @@ -297,11 +297,6 @@ srs_error_t SrsDynamicHttpConn::start() return conn->start(); } -void SrsDynamicHttpConn::remark(int64_t* in, int64_t* out) -{ - conn->remark(in, out); -} - SrsHttpFileReader::SrsHttpFileReader(ISrsHttpResponseReader* h) { http = h; diff --git a/trunk/src/app/srs_app_caster_flv.hpp b/trunk/src/app/srs_app_caster_flv.hpp index 44c713012..54d61cf08 100644 --- a/trunk/src/app/srs_app_caster_flv.hpp +++ b/trunk/src/app/srs_app_caster_flv.hpp @@ -35,7 +35,7 @@ class SrsAppCasterFlv : public ISrsTcpHandler, public ISrsResourceManager, publi private: std::string output; SrsHttpServeMux* http_mux; - std::vector conns; + std::vector conns; SrsResourceManager* manager; public: SrsAppCasterFlv(SrsConfDirective* c); @@ -54,7 +54,8 @@ public: }; // The dynamic http connection, never drop the body. -class SrsDynamicHttpConn : public ISrsStartableConneciton, public ISrsHttpConnOwner, public ISrsReloadHandler +class SrsDynamicHttpConn : public ISrsConnection, public ISrsStartable, public ISrsHttpConnOwner + , public ISrsReloadHandler { private: // The manager object to manage the connection. @@ -92,9 +93,6 @@ public: // Interface ISrsStartable public: virtual srs_error_t start(); -// Interface ISrsKbpsDelta -public: - virtual void remark(int64_t* in, int64_t* out); }; // The http wrapper for file reader, to read http post stream like a file. diff --git a/trunk/src/app/srs_app_conn.cpp b/trunk/src/app/srs_app_conn.cpp index 929ebb0d6..394baa05f 100644 --- a/trunk/src/app/srs_app_conn.cpp +++ b/trunk/src/app/srs_app_conn.cpp @@ -414,14 +414,6 @@ ISrsExpire::~ISrsExpire() { } -ISrsStartableConneciton::ISrsStartableConneciton() -{ -} - -ISrsStartableConneciton::~ISrsStartableConneciton() -{ -} - SrsTcpConnection::SrsTcpConnection(srs_netfd_t c) { stfd = c; diff --git a/trunk/src/app/srs_app_conn.hpp b/trunk/src/app/srs_app_conn.hpp index 9158dab1d..6cba50549 100644 --- a/trunk/src/app/srs_app_conn.hpp +++ b/trunk/src/app/srs_app_conn.hpp @@ -135,14 +135,6 @@ public: virtual void expire() = 0; }; -// Interface for connection that is startable. -class ISrsStartableConneciton : public ISrsConnection, public ISrsStartable, public ISrsKbpsDelta -{ -public: - ISrsStartableConneciton(); - virtual ~ISrsStartableConneciton(); -}; - // The basic connection of SRS, for TCP based protocols, // all connections accept from listener must extends from this base class, // server will add the connection to manager, and delete it when remove. diff --git a/trunk/src/app/srs_app_http_conn.cpp b/trunk/src/app/srs_app_http_conn.cpp index c2408cf14..7e22233bb 100644 --- a/trunk/src/app/srs_app_http_conn.cpp +++ b/trunk/src/app/srs_app_http_conn.cpp @@ -62,9 +62,8 @@ SrsHttpConn::SrsHttpConn(ISrsHttpConnOwner* handler, ISrsProtocolReadWriter* fd, ip = cip; port = cport; create_time = srsu2ms(srs_get_system_time()); - clk = new SrsWallClock(); - kbps = new SrsKbps(clk); - kbps->set_io(skt, skt); + delta_ = new SrsNetworkDelta(); + delta_->set_io(skt, skt); trd = new SrsSTCoroutine("http", this, _srs_context->get_id()); } @@ -76,8 +75,7 @@ SrsHttpConn::~SrsHttpConn() srs_freep(parser); srs_freep(cors); - srs_freep(kbps); - srs_freep(clk); + srs_freep(delta_); } std::string SrsHttpConn::desc() @@ -85,9 +83,9 @@ std::string SrsHttpConn::desc() return "HttpConn"; } -void SrsHttpConn::remark(int64_t* in, int64_t* out) +ISrsKbpsDelta* SrsHttpConn::delta() { - kbps->remark(in, out); + return delta_; } srs_error_t SrsHttpConn::start() @@ -404,7 +402,7 @@ srs_error_t SrsHttpxConn::on_conn_done(srs_error_t r0) bool exists = false; SrsStatistic::instance()->on_disconnect(get_id().c_str(), &exists); if (exists) { - SrsStatistic::instance()->kbps_add_delta(get_id().c_str(), this); + SrsStatistic::instance()->kbps_add_delta(get_id().c_str(), conn->delta()); } // Because we use manager to manage this object, @@ -465,9 +463,9 @@ srs_error_t SrsHttpxConn::start() return conn->start(); } -void SrsHttpxConn::remark(int64_t* in, int64_t* out) +ISrsKbpsDelta* SrsHttpxConn::delta() { - conn->remark(in, out); + return conn->delta(); } SrsHttpServer::SrsHttpServer(SrsServer* svr) diff --git a/trunk/src/app/srs_app_http_conn.hpp b/trunk/src/app/srs_app_http_conn.hpp index 28469c131..3cc143ed3 100644 --- a/trunk/src/app/srs_app_http_conn.hpp +++ b/trunk/src/app/srs_app_http_conn.hpp @@ -36,6 +36,7 @@ class SrsHttpUri; class SrsHttpMessage; class SrsHttpStreamServer; class SrsHttpStaticServer; +class SrsNetworkDelta; // The owner of HTTP connection. class ISrsHttpConnOwner @@ -59,7 +60,7 @@ public: // TODO: FIXME: Should rename to roundtrip or responder, not connection. // The http connection which request the static or stream content. -class SrsHttpConn : public ISrsStartableConneciton, public ISrsCoroutineHandler +class SrsHttpConn : public ISrsConnection, public ISrsStartable, public ISrsCoroutineHandler , public ISrsExpire { protected: @@ -76,12 +77,8 @@ protected: std::string ip; int port; private: - // The connection total kbps. - // not only the rtmp or http connection, all type of connection are - // need to statistic the kbps of io. - // The SrsStatistic will use it indirectly to statistic the bytes delta of current connection. - SrsKbps* kbps; - SrsWallClock* clk; + // The delta for statistic. + SrsNetworkDelta* delta_; // The create time in milliseconds. // for current connection to log self create time and calculate the living time. int64_t create_time; @@ -91,9 +88,8 @@ public: // Interface ISrsResource. public: virtual std::string desc(); -// Interface ISrsKbpsDelta public: - virtual void remark(int64_t* in, int64_t* out); + ISrsKbpsDelta* delta(); // Interface ISrsStartable public: virtual srs_error_t start(); @@ -127,8 +123,7 @@ public: }; // Drop body of request, only process the response. -class SrsHttpxConn : public ISrsStartableConneciton, public ISrsHttpConnOwner - , public ISrsReloadHandler +class SrsHttpxConn : public ISrsConnection, public ISrsStartable, public ISrsHttpConnOwner, public ISrsReloadHandler { private: // The manager object to manage the connection. @@ -168,9 +163,8 @@ public: // Interface ISrsStartable public: virtual srs_error_t start(); -// Interface ISrsKbpsDelta public: - virtual void remark(int64_t* in, int64_t* out); + ISrsKbpsDelta* delta(); }; // The http server, use http stream or static server to serve requests. diff --git a/trunk/src/app/srs_app_http_static.cpp b/trunk/src/app/srs_app_http_static.cpp index 05997cebe..a8a137551 100644 --- a/trunk/src/app/srs_app_http_static.cpp +++ b/trunk/src/app/srs_app_http_static.cpp @@ -70,8 +70,6 @@ SrsHlsStream::~SrsHlsStream() srs_error_t SrsHlsStream::serve_m3u8_ctx(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, ISrsFileReaderFactory* factory, string fullpath, SrsRequest* req) { - srs_error_t err = srs_success; - string ctx = r->query_get(SRS_CONTEXT_IN_HLS); // Always make the ctx alive now. @@ -99,13 +97,13 @@ void SrsHlsStream::on_serve_ts_ctx(ISrsHttpResponseWriter* w, ISrsHttpMessage* r SrsHttpConn* hc = dynamic_cast(hr->connection()); srs_assert(hc); - ISrsKbpsDelta* conn = dynamic_cast(hc); - srs_assert(conn); + ISrsKbpsDelta* delta = hc->delta(); + srs_assert(delta); // Only update the delta, because SrsServer will sample it. Note that SrsServer also does the stat for all clients // including this one, but it should be ignored because the id is not matched, and instead we use the hls_ctx as // session id to match the client. - SrsStatistic::instance()->kbps_add_delta(ctx, conn); + SrsStatistic::instance()->kbps_add_delta(ctx, delta); } srs_error_t SrsHlsStream::serve_new_session(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, SrsRequest* req) @@ -467,8 +465,6 @@ srs_error_t SrsVodStream::serve_mp4_stream(ISrsHttpResponseWriter* w, ISrsHttpMe srs_error_t SrsVodStream::serve_m3u8_ctx(ISrsHttpResponseWriter * w, ISrsHttpMessage * r, std::string fullpath) { - srs_error_t err = srs_success; - SrsHttpMessage* hr = dynamic_cast(r); srs_assert(hr); diff --git a/trunk/src/app/srs_app_rtc_conn.cpp b/trunk/src/app/srs_app_rtc_conn.cpp index fc23b3b4c..712fa09b0 100644 --- a/trunk/src/app/srs_app_rtc_conn.cpp +++ b/trunk/src/app/srs_app_rtc_conn.cpp @@ -1945,9 +1945,9 @@ vector SrsRtcConnection::peer_addresses() return addresses; } -void SrsRtcConnection::remark(int64_t* in, int64_t* out) +ISrsKbpsDelta* SrsRtcConnection::delta() { - delta_->remark(in, out); + return delta_; } const SrsContextId& SrsRtcConnection::get_id() diff --git a/trunk/src/app/srs_app_rtc_conn.hpp b/trunk/src/app/srs_app_rtc_conn.hpp index baf3b1c7e..425016d70 100644 --- a/trunk/src/app/srs_app_rtc_conn.hpp +++ b/trunk/src/app/srs_app_rtc_conn.hpp @@ -434,7 +434,7 @@ private: // // For performance, we use non-public from resource, // see https://stackoverflow.com/questions/3747066/c-cannot-convert-from-base-a-to-derived-type-b-via-virtual-base-a -class SrsRtcConnection : public ISrsResource, public ISrsDisposingHandler, public ISrsExpire, public ISrsKbpsDelta +class SrsRtcConnection : public ISrsResource, public ISrsDisposingHandler, public ISrsExpire { friend class SrsSecurityTransport; friend class SrsRtcPlayStream; @@ -513,9 +513,8 @@ public: std::string username(); // Get all addresses client used. std::vector peer_addresses(); -// Interface ISrsKbpsDelta. public: - virtual void remark(int64_t* in, int64_t* out); + virtual ISrsKbpsDelta* delta(); // Interface ISrsResource. public: virtual const SrsContextId& get_id(); diff --git a/trunk/src/app/srs_app_rtc_server.cpp b/trunk/src/app/srs_app_rtc_server.cpp index 221c831b6..baf34069a 100644 --- a/trunk/src/app/srs_app_rtc_server.cpp +++ b/trunk/src/app/srs_app_rtc_server.cpp @@ -643,10 +643,7 @@ srs_error_t SrsRtcServer::on_timer(srs_utime_t interval) // Update stat if session is alive. if (session->is_alive()) { nn_rtc_conns++; - - ISrsKbpsDelta* conn = dynamic_cast(session); - SrsStatistic::instance()->kbps_add_delta(session->get_id().c_str(), conn); - + SrsStatistic::instance()->kbps_add_delta(session->get_id().c_str(), session->delta()); continue; } diff --git a/trunk/src/app/srs_app_rtmp_conn.cpp b/trunk/src/app/srs_app_rtmp_conn.cpp index eb7dd1c3a..65a15505e 100644 --- a/trunk/src/app/srs_app_rtmp_conn.cpp +++ b/trunk/src/app/srs_app_rtmp_conn.cpp @@ -101,9 +101,8 @@ SrsRtmpConn::SrsRtmpConn(SrsServer* svr, srs_netfd_t c, string cip, int cport) ip = cip; port = cport; create_time = srsu2ms(srs_get_system_time()); - clk = new SrsWallClock(); - kbps = new SrsKbps(clk); - kbps->set_io(skt, skt); + delta_ = new SrsNetworkDelta(); + delta_->set_io(skt, skt); trd = new SrsSTCoroutine("rtmp", this, _srs_context->get_id()); rtmp = new SrsRtmpServer(skt); @@ -137,8 +136,7 @@ SrsRtmpConn::~SrsRtmpConn() } srs_freep(trd); - srs_freep(kbps); - srs_freep(clk); + srs_freep(delta_); srs_freep(skt); srs_freep(info); @@ -342,9 +340,9 @@ srs_error_t SrsRtmpConn::on_reload_vhost_publish(string vhost) return err; } -void SrsRtmpConn::remark(int64_t* in, int64_t* out) +ISrsKbpsDelta* SrsRtmpConn::delta() { - kbps->remark(in, out); + return delta_; } srs_error_t SrsRtmpConn::service_cycle() @@ -765,10 +763,7 @@ srs_error_t SrsRtmpConn::do_playing(SrsLiveSource* source, SrsLiveConsumer* cons // reportable if (pprint->can_print()) { - kbps->sample(); - srs_trace("-> " SRS_CONSTS_LOG_PLAY " time=%d, msgs=%d, okbps=%d,%d,%d, ikbps=%d,%d,%d, mw=%d/%d", - (int)pprint->age(), count, kbps->get_send_kbps(), kbps->get_send_kbps_30s(), kbps->get_send_kbps_5m(), - kbps->get_recv_kbps(), kbps->get_recv_kbps_30s(), kbps->get_recv_kbps_5m(), srsu2msi(mw_sleep), mw_msgs); + srs_trace("-> " SRS_CONSTS_LOG_PLAY " time=%d, msgs=%d, mw=%d/%d", (int)pprint->age(), count, srsu2msi(mw_sleep), mw_msgs); } if (count <= 0) { @@ -891,8 +886,7 @@ srs_error_t SrsRtmpConn::do_publishing(SrsLiveSource* source, SrsPublishRecvThre if (true) { bool mr = _srs_config->get_mr_enabled(req->vhost); srs_utime_t mr_sleep = _srs_config->get_mr_sleep(req->vhost); - srs_trace("start publish mr=%d/%d, p1stpt=%d, pnt=%d, tcp_nodelay=%d", - mr, srsu2msi(mr_sleep), srsu2msi(publish_1stpkt_timeout), srsu2msi(publish_normal_timeout), tcp_nodelay); + srs_trace("start publish mr=%d/%d, p1stpt=%d, pnt=%d, tcp_nodelay=%d", mr, srsu2msi(mr_sleep), srsu2msi(publish_1stpkt_timeout), srsu2msi(publish_normal_timeout), tcp_nodelay); } int64_t nb_msgs = 0; @@ -935,13 +929,9 @@ srs_error_t SrsRtmpConn::do_publishing(SrsLiveSource* source, SrsPublishRecvThre // reportable if (pprint->can_print()) { - kbps->sample(); bool mr = _srs_config->get_mr_enabled(req->vhost); srs_utime_t mr_sleep = _srs_config->get_mr_sleep(req->vhost); - srs_trace("<- " SRS_CONSTS_LOG_CLIENT_PUBLISH " time=%d, okbps=%d,%d,%d, ikbps=%d,%d,%d, mr=%d/%d, p1stpt=%d, pnt=%d", - (int)pprint->age(), kbps->get_send_kbps(), kbps->get_send_kbps_30s(), kbps->get_send_kbps_5m(), - kbps->get_recv_kbps(), kbps->get_recv_kbps_30s(), kbps->get_recv_kbps_5m(), mr, srsu2msi(mr_sleep), - srsu2msi(publish_1stpkt_timeout), srsu2msi(publish_normal_timeout)); + srs_trace("<- " SRS_CONSTS_LOG_CLIENT_PUBLISH " time=%d, mr=%d/%d, p1stpt=%d, pnt=%d", (int)pprint->age(), mr, srsu2msi(mr_sleep), srsu2msi(publish_1stpkt_timeout), srsu2msi(publish_normal_timeout)); } } @@ -1313,7 +1303,7 @@ void SrsRtmpConn::http_hooks_on_close() for (int i = 0; i < (int)hooks.size(); i++) { std::string url = hooks.at(i); - SrsHttpHooks::on_close(url, req, kbps->get_send_bytes(), kbps->get_recv_bytes()); + SrsHttpHooks::on_close(url, req, skt->get_send_bytes(), skt->get_recv_bytes()); } } @@ -1468,7 +1458,7 @@ srs_error_t SrsRtmpConn::cycle() // Update statistic when done. SrsStatistic* stat = SrsStatistic::instance(); - stat->kbps_add_delta(get_id().c_str(), this); + stat->kbps_add_delta(get_id().c_str(), delta_); stat->on_disconnect(get_id().c_str()); // Notify manager to remove it. diff --git a/trunk/src/app/srs_app_rtmp_conn.hpp b/trunk/src/app/srs_app_rtmp_conn.hpp index 5e9b7d17c..2b1ff3c4d 100644 --- a/trunk/src/app/srs_app_rtmp_conn.hpp +++ b/trunk/src/app/srs_app_rtmp_conn.hpp @@ -37,6 +37,7 @@ class SrsSecurity; class ISrsWakable; class SrsCommonMessage; class SrsPacket; +class SrsNetworkDelta; // The simple rtmp client for SRS. class SrsSimpleRtmpClient : public SrsBasicRtmpClient @@ -66,7 +67,7 @@ public: }; // The client provides the main logic control for RTMP clients. -class SrsRtmpConn : public ISrsStartableConneciton, public ISrsReloadHandler +class SrsRtmpConn : public ISrsConnection, public ISrsStartable, public ISrsReloadHandler , public ISrsCoroutineHandler, public ISrsExpire { // For the thread to directly access any field of connection. @@ -110,12 +111,8 @@ private: // The ip and port of client. std::string ip; int port; - // The connection total kbps. - // not only the rtmp or http connection, all type of connection are - // need to statistic the kbps of io. - // The SrsStatistic will use it indirectly to statistic the bytes delta of current connection. - SrsKbps* kbps; - SrsWallClock* clk; + // The delta for statistic. + SrsNetworkDelta* delta_; // The create time in milliseconds. // for current connection to log self create time and calculate the living time. int64_t create_time; @@ -134,9 +131,8 @@ public: virtual srs_error_t on_reload_vhost_tcp_nodelay(std::string vhost); virtual srs_error_t on_reload_vhost_realtime(std::string vhost); virtual srs_error_t on_reload_vhost_publish(std::string vhost); -// Interface ISrsKbpsDelta public: - virtual void remark(int64_t* in, int64_t* out); + virtual ISrsKbpsDelta* delta(); private: // When valid and connected to vhost/app, service the client. virtual srs_error_t service_cycle(); diff --git a/trunk/src/app/srs_app_server.cpp b/trunk/src/app/srs_app_server.cpp index adbf10bc4..0b10bd5e0 100644 --- a/trunk/src/app/srs_app_server.cpp +++ b/trunk/src/app/srs_app_server.cpp @@ -1341,11 +1341,21 @@ void SrsServer::resample_kbps() // collect delta from all clients. for (int i = 0; i < (int)conn_manager->size(); i++) { ISrsResource* c = conn_manager->at(i); - ISrsKbpsDelta* conn = dynamic_cast(c); - - // add delta of connection to server kbps., - // for next sample() of server kbps can get the stat. - stat->kbps_add_delta(c->get_id().c_str(), conn); + + SrsRtmpConn* rtmp = dynamic_cast(c); + if (rtmp) { + stat->kbps_add_delta(c->get_id().c_str(), rtmp->delta()); + continue; + } + + SrsHttpxConn* httpx = dynamic_cast(c); + if (httpx) { + stat->kbps_add_delta(c->get_id().c_str(), httpx->delta()); + continue; + } + + // Impossible path, because we only create these connections above. + srs_assert(false); } // Update the global server level statistics. @@ -1356,9 +1366,9 @@ srs_error_t SrsServer::accept_client(SrsListenerType type, srs_netfd_t stfd) { srs_error_t err = srs_success; - ISrsStartableConneciton* conn = NULL; + ISrsResource* resource = NULL; - if ((err = fd_to_resource(type, stfd, &conn)) != srs_success) { + if ((err = fd_to_resource(type, stfd, &resource)) != srs_success) { //close fd on conn error, otherwise will lead to fd leak -gs srs_close_stfd(stfd); if (srs_error_code(err) == ERROR_SOCKET_GET_PEER_IP && _srs_config->empty_ip_ok()) { @@ -1367,11 +1377,12 @@ srs_error_t SrsServer::accept_client(SrsListenerType type, srs_netfd_t stfd) } return srs_error_wrap(err, "fd to resource"); } - srs_assert(conn); + srs_assert(resource); // directly enqueue, the cycle thread will remove the client. - conn_manager->add(conn); + conn_manager->add(resource); + ISrsStartable* conn = dynamic_cast(resource); if ((err = conn->start()) != srs_success) { return srs_error_wrap(err, "start conn coroutine"); } @@ -1384,7 +1395,7 @@ ISrsHttpServeMux* SrsServer::api_server() return http_api_mux; } -srs_error_t SrsServer::fd_to_resource(SrsListenerType type, srs_netfd_t stfd, ISrsStartableConneciton** pr) +srs_error_t SrsServer::fd_to_resource(SrsListenerType type, srs_netfd_t stfd, ISrsResource** pr) { srs_error_t err = srs_success; diff --git a/trunk/src/app/srs_app_server.hpp b/trunk/src/app/srs_app_server.hpp index f7a7775d1..36e6dae69 100644 --- a/trunk/src/app/srs_app_server.hpp +++ b/trunk/src/app/srs_app_server.hpp @@ -311,7 +311,7 @@ public: // TODO: FIXME: Fetch from hybrid server manager. virtual ISrsHttpServeMux* api_server(); private: - virtual srs_error_t fd_to_resource(SrsListenerType type, srs_netfd_t stfd, ISrsStartableConneciton** pr); + virtual srs_error_t fd_to_resource(SrsListenerType type, srs_netfd_t stfd, ISrsResource** pr); // Interface ISrsResourceManager public: // A callback for connection to remove itself. diff --git a/trunk/src/app/srs_app_srt_conn.cpp b/trunk/src/app/srs_app_srt_conn.cpp index f2183a35f..b9c84e8d1 100644 --- a/trunk/src/app/srs_app_srt_conn.cpp +++ b/trunk/src/app/srs_app_srt_conn.cpp @@ -160,9 +160,8 @@ SrsMpegtsSrtConn::SrsMpegtsSrtConn(SrsSrtServer* srt_server, srs_srt_t srt_fd, s srt_fd_ = srt_fd; srt_conn_ = new SrsSrtConnection(srt_fd_); - clock_ = new SrsWallClock(); - kbps_ = new SrsKbps(clock_); - kbps_->set_io(srt_conn_, srt_conn_); + delta_ = new SrsNetworkDelta(); + delta_->set_io(srt_conn_, srt_conn_); ip_ = ip; port_ = port; @@ -177,11 +176,8 @@ SrsMpegtsSrtConn::~SrsMpegtsSrtConn() { srs_freep(trd_); - srs_freep(kbps_); - srs_freep(clock_); - + srs_freep(delta_); srs_freep(srt_conn_); - srs_freep(req_); } @@ -190,9 +186,9 @@ std::string SrsMpegtsSrtConn::desc() return "srt-ts-conn"; } -void SrsMpegtsSrtConn::remark(int64_t* in, int64_t* out) +ISrsKbpsDelta* SrsMpegtsSrtConn::delta() { - kbps_->remark(in, out); + return delta_; } void SrsMpegtsSrtConn::expire() @@ -227,7 +223,7 @@ srs_error_t SrsMpegtsSrtConn::cycle() // Update statistic when done. SrsStatistic* stat = SrsStatistic::instance(); - stat->kbps_add_delta(get_id().c_str(), this); + stat->kbps_add_delta(get_id().c_str(), delta_); stat->on_disconnect(get_id().c_str()); // Notify manager to remove it. @@ -412,11 +408,7 @@ srs_error_t SrsMpegtsSrtConn::do_publishing() s.pktRecv(), s.pktRcvLoss(), s.pktRcvRetrans(), s.pktRcvDrop()); } - kbps_->sample(); - - srs_trace("<- " SRS_CONSTS_LOG_SRT_PUBLISH " time=%d, packets=%d, okbps=%d,%d,%d, ikbps=%d,%d,%d", - (int)pprint->age(), nb_packets, kbps_->get_send_kbps(), kbps_->get_send_kbps_30s(), kbps_->get_send_kbps_5m(), - kbps_->get_recv_kbps(), kbps_->get_recv_kbps_30s(), kbps_->get_recv_kbps_5m()); + srs_trace("<- " SRS_CONSTS_LOG_SRT_PUBLISH " time=%d, packets=%d", (int)pprint->age(), nb_packets); nb_packets = 0; } @@ -493,11 +485,7 @@ srs_error_t SrsMpegtsSrtConn::do_playing() s.pktSent(), s.pktSndLoss(), s.pktRetrans(), s.pktSndDrop()); } - kbps_->sample(); - - srs_trace("-> " SRS_CONSTS_LOG_SRT_PLAY " time=%d, packets=%d, okbps=%d,%d,%d, ikbps=%d,%d,%d", - (int)pprint->age(), nb_packets, kbps_->get_send_kbps(), kbps_->get_send_kbps_30s(), kbps_->get_send_kbps_5m(), - kbps_->get_recv_kbps(), kbps_->get_recv_kbps_30s(), kbps_->get_recv_kbps_5m()); + srs_trace("-> " SRS_CONSTS_LOG_SRT_PLAY " time=%d, packets=%d", (int)pprint->age(), nb_packets); nb_packets = 0; } @@ -600,7 +588,7 @@ void SrsMpegtsSrtConn::http_hooks_on_close() for (int i = 0; i < (int)hooks.size(); i++) { std::string url = hooks.at(i); - SrsHttpHooks::on_close(url, req_, kbps_->get_send_bytes(), kbps_->get_recv_bytes()); + SrsHttpHooks::on_close(url, req_, srt_conn_->get_send_bytes(), srt_conn_->get_recv_bytes()); } } diff --git a/trunk/src/app/srs_app_srt_conn.hpp b/trunk/src/app/srs_app_srt_conn.hpp index fe6acf8f0..709aa98c5 100644 --- a/trunk/src/app/srs_app_srt_conn.hpp +++ b/trunk/src/app/srs_app_srt_conn.hpp @@ -21,6 +21,7 @@ class SrsBuffer; class SrsLiveSource; class SrsSrtSource; class SrsSrtServer; +class SrsNetworkDelta; // The basic connection of SRS, for SRT based protocols, // all srt connections accept from srt listener must extends from this base class, @@ -70,7 +71,7 @@ private: srs_error_t recv_err_; }; -class SrsMpegtsSrtConn : public ISrsStartableConneciton, public ISrsCoroutineHandler, public ISrsExpire +class SrsMpegtsSrtConn : public ISrsConnection, public ISrsStartable, public ISrsCoroutineHandler, public ISrsExpire { public: SrsMpegtsSrtConn(SrsSrtServer* srt_server, srs_srt_t srt_fd, std::string ip, int port); @@ -78,9 +79,8 @@ public: // Interface ISrsResource. public: virtual std::string desc(); -// Interface ISrsKbpsDelta public: - virtual void remark(int64_t* in, int64_t* out); + ISrsKbpsDelta* delta(); // Interface ISrsExpire public: virtual void expire(); @@ -115,8 +115,7 @@ private: SrsSrtServer* srt_server_; srs_srt_t srt_fd_; SrsSrtConnection* srt_conn_; - SrsWallClock* clock_; - SrsKbps* kbps_; + SrsNetworkDelta* delta_; std::string ip_; int port_; SrsCoroutine* trd_; diff --git a/trunk/src/app/srs_app_srt_server.cpp b/trunk/src/app/srs_app_srt_server.cpp index f59ceaa18..2c497d6fa 100644 --- a/trunk/src/app/srs_app_srt_server.cpp +++ b/trunk/src/app/srs_app_srt_server.cpp @@ -212,18 +212,19 @@ srs_error_t SrsSrtServer::accept_srt_client(srs_srt_t srt_fd) { srs_error_t err = srs_success; - ISrsStartableConneciton* conn = NULL; - if ((err = fd_to_resource(srt_fd, &conn)) != srs_success) { + ISrsResource* resource = NULL; + if ((err = fd_to_resource(srt_fd, &resource)) != srs_success) { //close fd on conn error, otherwise will lead to fd leak -gs // TODO: FIXME: Handle error. srs_srt_close(srt_fd); return srs_error_wrap(err, "srt fd to resource"); } - srs_assert(conn); + srs_assert(resource); // directly enqueue, the cycle thread will remove the client. - conn_manager_->add(conn); + conn_manager_->add(resource); + ISrsStartable* conn = dynamic_cast(resource); if ((err = conn->start()) != srs_success) { return srs_error_wrap(err, "start srt conn coroutine"); } @@ -231,7 +232,7 @@ srs_error_t SrsSrtServer::accept_srt_client(srs_srt_t srt_fd) return err; } -srs_error_t SrsSrtServer::fd_to_resource(srs_srt_t srt_fd, ISrsStartableConneciton** pr) +srs_error_t SrsSrtServer::fd_to_resource(srs_srt_t srt_fd, ISrsResource** pr) { srs_error_t err = srs_success; @@ -294,11 +295,13 @@ void SrsSrtServer::resample_kbps() // collect delta from all clients. for (int i = 0; i < (int)conn_manager_->size(); i++) { ISrsResource* c = conn_manager_->at(i); - ISrsKbpsDelta* conn = dynamic_cast(c); + + SrsMpegtsSrtConn* conn = dynamic_cast(c); + srs_assert(conn); // add delta of connection to server kbps., // for next sample() of server kbps can get the stat. - SrsStatistic::instance()->kbps_add_delta(c->get_id().c_str(), conn); + SrsStatistic::instance()->kbps_add_delta(c->get_id().c_str(), conn->delta()); } } diff --git a/trunk/src/app/srs_app_srt_server.hpp b/trunk/src/app/srs_app_srt_server.hpp index 8ae6f7a04..2e0c5d123 100644 --- a/trunk/src/app/srs_app_srt_server.hpp +++ b/trunk/src/app/srs_app_srt_server.hpp @@ -62,7 +62,7 @@ public: // @param srt_fd, the client fd in srt boxed, the underlayer fd. virtual srs_error_t accept_srt_client(srs_srt_t srt_fd); private: - virtual srs_error_t fd_to_resource(srs_srt_t srt_fd, ISrsStartableConneciton** pr); + virtual srs_error_t fd_to_resource(srs_srt_t srt_fd, ISrsResource** pr); // Interface ISrsResourceManager public: // A callback for connection to remove itself. diff --git a/trunk/src/app/srs_app_statistic.cpp b/trunk/src/app/srs_app_statistic.cpp index 94052266b..a8369837f 100644 --- a/trunk/src/app/srs_app_statistic.cpp +++ b/trunk/src/app/srs_app_statistic.cpp @@ -478,11 +478,12 @@ void SrsStatistic::cleanup_stream(SrsStatisticStream* stream) void SrsStatistic::kbps_add_delta(std::string id, ISrsKbpsDelta* delta) { - if (clients.find(id) == clients.end()) { - return; - } - - SrsStatisticClient* client = clients[id]; + if (!delta) return; + + map::iterator it = clients.find(id); + if (it == clients.end()) return; + + SrsStatisticClient* client = it->second; // resample the kbps to collect the delta. int64_t in, out; diff --git a/trunk/src/core/srs_core_version5.hpp b/trunk/src/core/srs_core_version5.hpp index eea783c96..8d0a9e0de 100644 --- a/trunk/src/core/srs_core_version5.hpp +++ b/trunk/src/core/srs_core_version5.hpp @@ -9,6 +9,6 @@ #define VERSION_MAJOR 5 #define VERSION_MINOR 0 -#define VERSION_REVISION 50 +#define VERSION_REVISION 51 #endif diff --git a/trunk/src/protocol/srs_protocol_kbps.cpp b/trunk/src/protocol/srs_protocol_kbps.cpp index f667d41e7..3e2929d8b 100644 --- a/trunk/src/protocol/srs_protocol_kbps.cpp +++ b/trunk/src/protocol/srs_protocol_kbps.cpp @@ -90,6 +90,54 @@ void SrsEphemeralDelta::remark(int64_t* in, int64_t* out) in_ = out_ = 0; } +SrsNetworkDelta::SrsNetworkDelta() +{ + in_ = out_ = NULL; + in_base_ = in_delta_ = 0; + out_base_ = out_delta_ = 0; +} + +SrsNetworkDelta::~SrsNetworkDelta() +{ +} + +void SrsNetworkDelta::set_io(ISrsProtocolStatistic* in, ISrsProtocolStatistic* out) +{ + if (in_) { + in_delta_ += in_->get_recv_bytes() - in_base_; + } + if (in) { + in_base_ = in->get_recv_bytes(); + in_delta_ += in_base_; + } + in_ = in; + + if (out_) { + out_delta_ += out_->get_send_bytes() - out_base_; + } + if (out) { + out_base_ = out->get_send_bytes(); + out_delta_ += out_base_; + } + out_ = out; +} + +void SrsNetworkDelta::remark(int64_t* in, int64_t* out) +{ + if (in_) { + in_delta_ += in_->get_recv_bytes() - in_base_; + in_base_ = in_->get_recv_bytes(); + } + if (out_) { + out_delta_ += out_->get_send_bytes() - out_base_; + out_base_ = out_->get_send_bytes(); + } + + *in = in_delta_; + *out = out_delta_; + in_delta_ = out_delta_ = 0; +} + SrsKbps::SrsKbps(SrsWallClock* c) : is(c), os(c) { clk = c; @@ -249,25 +297,3 @@ int64_t SrsKbps::get_recv_bytes() return bytes; } -void SrsKbps::remark(int64_t* in, int64_t* out) -{ - sample(); - - int64_t inv = is.get_total_bytes() - is.delta_bytes; - is.delta_bytes = is.get_total_bytes(); - if (in) { - *in = inv; - } - - int64_t outv = os.get_total_bytes() - os.delta_bytes; - os.delta_bytes = os.get_total_bytes(); - if (out) { - *out = outv; - } -} - -int SrsKbps::size_memory() -{ - return sizeof(SrsKbps); -} - diff --git a/trunk/src/protocol/srs_protocol_kbps.hpp b/trunk/src/protocol/srs_protocol_kbps.hpp index 336bc54f8..2eb901888 100644 --- a/trunk/src/protocol/srs_protocol_kbps.hpp +++ b/trunk/src/protocol/srs_protocol_kbps.hpp @@ -101,6 +101,26 @@ public: virtual void remark(int64_t* in, int64_t* out); }; +// A network delta data source for SrsKbps. +class SrsNetworkDelta : public ISrsKbpsDelta +{ +private: + ISrsProtocolStatistic* in_; + ISrsProtocolStatistic* out_; + uint64_t in_base_; + uint64_t in_delta_; + uint64_t out_base_; + uint64_t out_delta_; +public: + SrsNetworkDelta(); + virtual ~SrsNetworkDelta(); +public: + virtual void set_io(ISrsProtocolStatistic* in, ISrsProtocolStatistic* out); +// Interface ISrsKbpsDelta. +public: + virtual void remark(int64_t* in, int64_t* out); +}; + /** * to statistic the kbps of io. * itself can be a statistic source, for example, used for SRS bytes stat. @@ -121,14 +141,7 @@ public: * kbps->add_delta(in, out) * kbps->sample() * kbps->get_xxx_kbps(). - * 3. kbps used as ISrsKbpsDelta, to provides delta bytes: - * SrsKbps* kbps = ...; - * kbps->set_io(in, out); - * ISrsKbpsDelta* delta = (ISrsKbpsDelta*)kbps; - * int64_t in, out; - * delta->remark(&in, out); - * printf("delta is %d/%d", in, out); - * 4. kbps used as ISrsProtocolStatistic, to provides raw bytes: + * 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. @@ -137,7 +150,7 @@ public: * user->set_io(kbps, kbps); * the server never know how many bytes already send/recv, for the connection maybe closed. */ -class SrsKbps : public ISrsProtocolStatistic, public ISrsKbpsDelta +class SrsKbps : public ISrsProtocolStatistic { private: SrsKbpsSlice is; @@ -192,12 +205,6 @@ public: public: virtual int64_t get_send_bytes(); virtual int64_t get_recv_bytes(); -// Interface ISrsKbpsDelta -public: - virtual void remark(int64_t* in, int64_t* out); -// Interface ISrsMemorySizer -public: - virtual int size_memory(); }; #endif diff --git a/trunk/src/utest/srs_utest_protocol.cpp b/trunk/src/utest/srs_utest_protocol.cpp index bd2797cd5..dbcd1efc7 100644 --- a/trunk/src/utest/srs_utest_protocol.cpp +++ b/trunk/src/utest/srs_utest_protocol.cpp @@ -6225,37 +6225,18 @@ VOID TEST(ProtocolKbpsTest, Connections) VOID TEST(ProtocolKbpsTest, Delta) { - if (true) { - SrsEphemeralDelta ed; - - ISrsKbpsDelta* delta = (ISrsKbpsDelta*)&ed; - int64_t in, out; - delta->remark(&in, &out); - EXPECT_EQ(0, in); - EXPECT_EQ(0, out); - - ed.add_delta(100 * 1000, 100 * 1000); - delta->remark(&in, &out); - EXPECT_EQ(100 * 1000, in); - EXPECT_EQ(100 * 1000, out); - - delta->remark(&in, &out); - EXPECT_EQ(0, in); - EXPECT_EQ(0, out); - } - if (true) { MockWallClock* clock = new MockWallClock(); SrsAutoFree(MockWallClock, clock); MockStatistic* io = new MockStatistic(); SrsAutoFree(MockStatistic, io); - SrsKbps* conn = new SrsKbps(clock->set_clock(0)); - SrsAutoFree(SrsKbps, conn); + SrsNetworkDelta* conn = new SrsNetworkDelta(); + SrsAutoFree(SrsNetworkDelta, conn); conn->set_io(io, io); // No data. - ISrsKbpsDelta* delta = (ISrsKbpsDelta*)conn; + ISrsKbpsDelta* delta = dynamic_cast(conn); int64_t in, out; delta->remark(&in, &out); EXPECT_EQ(0, in); @@ -6278,13 +6259,13 @@ VOID TEST(ProtocolKbpsTest, Delta) SrsAutoFree(MockWallClock, clock); MockStatistic* io = new MockStatistic(); SrsAutoFree(MockStatistic, io); - - SrsKbps* conn = new SrsKbps(clock->set_clock(0)); - SrsAutoFree(SrsKbps, conn); + + SrsNetworkDelta* conn = new SrsNetworkDelta(); + SrsAutoFree(SrsNetworkDelta, conn); conn->set_io(io, io); // No data. - ISrsKbpsDelta* delta = (ISrsKbpsDelta*)conn; + ISrsKbpsDelta* delta = dynamic_cast(conn); int64_t in, out; delta->remark(&in, &out); EXPECT_EQ(0, in); @@ -6724,3 +6705,51 @@ VOID TEST(ProtocolProtobufTest, FieldKey) } } +VOID TEST(ProtocolKbpsTest, NewDelta) +{ + if (true) { + SrsEphemeralDelta ed; + + ISrsKbpsDelta* delta = (ISrsKbpsDelta*)&ed; + int64_t in, out; + delta->remark(&in, &out); + EXPECT_EQ(0, in); + EXPECT_EQ(0, out); + + ed.add_delta(100 * 1000, 100 * 1000); + delta->remark(&in, &out); + EXPECT_EQ(100 * 1000, in); + EXPECT_EQ(100 * 1000, out); + + delta->remark(&in, &out); + EXPECT_EQ(0, in); + EXPECT_EQ(0, out); + } + + if (true) { + SrsNetworkDelta nd; + + ISrsKbpsDelta* delta = (ISrsKbpsDelta*)&nd; + int64_t in, out; + delta->remark(&in, &out); + EXPECT_EQ(0, in); + EXPECT_EQ(0, out); + + MockStatistic ms; + ms.set_in(100 * 1000)->set_out(100*1000); + nd.set_io(&ms, &ms); + delta->remark(&in, &out); + EXPECT_EQ(100 * 1000, in); + EXPECT_EQ(100 * 1000, out); + + ms.add_in(10 * 1000)->add_out(10 * 1000); + delta->remark(&in, &out); + EXPECT_EQ(10 * 1000, in); + EXPECT_EQ(10 * 1000, out); + + delta->remark(&in, &out); + EXPECT_EQ(0, in); + EXPECT_EQ(0, out); + } +} +