Revert "SmartPtr: Use shared ptr in RTC TCP connection. v6.0.127 (oss…

…rs#4083)"

This reverts commit 242152b.

trunk/doc/CHANGELOG.md

@@ -7,7 +7,6 @@ The changelog for SRS.
 <a name="v6-changes"></a>
 
 ## SRS 6.0 Changelog
-* v6.0, 2024-06-13, Merge [#4083](https://github.com/ossrs/srs/pull/4083): SmartPtr: Use shared ptr in RTC TCP connection. v6.0.127 (#4083)
 * v6.0, 2024-06-12, Merge [#4080](https://github.com/ossrs/srs/pull/4080): SmartPtr: Use shared ptr to manage GB objects. v6.0.126 (#4080)
 * v6.0, 2024-06-03, Merge [#4057](https://github.com/ossrs/srs/pull/4057): RTC: Support dropping h.264 SEI from NALUs. v6.0.125 (#4057)
 * v6.0, 2024-04-26, Merge [#4044](https://github.com/ossrs/srs/pull/4044): fix: correct SRS_ERRNO_MAP_HTTP duplicate error code. v6.0.124 (#4044)

trunk/src/app/srs_app_rtc_conn.cpp

@@ -1924,7 +1924,6 @@ std::string SrsRtcConnection::desc()
 
 void SrsRtcConnection::expire()
 {
-    // TODO: FIXME: Should set session to expired and remove it by heartbeat checking. Should not remove it directly.
     _srs_rtc_manager->remove(this);
 }
 

trunk/src/app/srs_app_rtc_network.cpp

@@ -436,19 +436,19 @@ srs_error_t SrsRtcUdpNetwork::write(void* buf, size_t size, ssize_t* nwrite)
     return sendonly_skt_->sendto(buf, size, SRS_UTIME_NO_TIMEOUT);
 }
 
-SrsRtcTcpNetwork::SrsRtcTcpNetwork(SrsRtcConnection* conn, SrsEphemeralDelta* delta) : owner_(new SrsRtcTcpConn())
+SrsRtcTcpNetwork::SrsRtcTcpNetwork(SrsRtcConnection* conn, SrsEphemeralDelta* delta)
 {
     conn_ = conn;
     delta_ = delta;
     sendonly_skt_ = NULL;
     transport_ = new SrsSecurityTransport(this);
     peer_port_ = 0;
     state_ = SrsRtcNetworkStateInit;
+    owner_ = NULL;
 }
 
 SrsRtcTcpNetwork::~SrsRtcTcpNetwork()
 {
-    owner_->interrupt();
     srs_freep(transport_);
 }
 
@@ -694,72 +694,54 @@ void SrsRtcTcpNetwork::dispose()
 
 #define SRS_RTC_TCP_PACKET_MAX 1500
 
-SrsRtcTcpConn::SrsRtcTcpConn()
-{
-    wrapper_ = NULL;
-    owner_coroutine_ = NULL;
-    owner_cid_ = NULL;
-    cid_ = _srs_context->get_id();
-
-    pkt_ = NULL;
-    delta_ = NULL;
-    skt_ = NULL;
-}
-
-SrsRtcTcpConn::SrsRtcTcpConn(ISrsProtocolReadWriter* skt, std::string cip, int port) : SrsRtcTcpConn()
+SrsRtcTcpConn::SrsRtcTcpConn(ISrsProtocolReadWriter* skt, std::string cip, int port, ISrsResourceManager* cm)
 {
+    manager_ = cm;
     ip_ = cip;
     port_ = port;
     skt_ = skt;
     delta_ = new SrsNetworkDelta();
     delta_->set_io(skt_, skt_);
+    trd_ = new SrsSTCoroutine("tcp", this, _srs_context->get_id());
     session_ = NULL;
     pkt_ = new char[SRS_RTC_TCP_PACKET_MAX];
+    _srs_rtc_manager->subscribe(this);
 }
 
 SrsRtcTcpConn::~SrsRtcTcpConn()
 {
+    _srs_rtc_manager->unsubscribe(this);
+    trd_->interrupt();
+    srs_freep(trd_);
+
     srs_freepa(pkt_);
     srs_freep(delta_);
     srs_freep(skt_);
 }
 
-void SrsRtcTcpConn::setup_owner(SrsSharedResource<SrsRtcTcpConn>* wrapper, ISrsInterruptable* owner_coroutine, ISrsContextIdSetter* owner_cid)
-{
-    wrapper_ = wrapper;
-    owner_coroutine_ = owner_coroutine;
-    owner_cid_ = owner_cid;
-}
-
 ISrsKbpsDelta* SrsRtcTcpConn::delta()
 {
     return delta_;
 }
 
-void SrsRtcTcpConn::interrupt()
-{
-    session_ = NULL;
-    if (owner_coroutine_) owner_coroutine_->interrupt();
-}
-
 std::string SrsRtcTcpConn::desc()
 {
     return "Tcp";
 }
 
 const SrsContextId& SrsRtcTcpConn::get_id()
 {
-    return cid_;
+    return trd_->cid();
 }
 
 std::string SrsRtcTcpConn::remote_ip()
 {
     return ip_;
 }
 
-void SrsRtcTcpConn::on_executor_done(ISrsInterruptable* executor)
+srs_error_t SrsRtcTcpConn::start()
 {
-    owner_coroutine_ = NULL;
+    return trd_->start();
 }
 
 srs_error_t SrsRtcTcpConn::cycle()
@@ -770,10 +752,15 @@ srs_error_t SrsRtcTcpConn::cycle()
     SrsStatistic::instance()->on_disconnect(get_id().c_str(), err);
     SrsStatistic::instance()->kbps_add_delta(get_id().c_str(), delta_);
 
+    // TODO: FIXME: Should manage RTC TCP connection by _srs_rtc_manager.
+    // Because we use manager to manage this object, not the http connection object, so we must remove it here.
+    manager_->remove(this);
+
+    // TODO: FIXME: When TCP connection(transport) closed, should notify session to dispose, should not free them simultaneously.
     // Only remove session when network is established, because client might use other UDP network.
     if(session_ && session_->tcp()->is_establelished()) {
         session_->tcp()->set_state(SrsRtcNetworkStateClosed);
-        session_->expire();
+        _srs_rtc_manager->remove(session_);
     }
 
     // For HTTP-API timeout, we think it's done successfully,
@@ -814,18 +801,13 @@ srs_error_t SrsRtcTcpConn::do_cycle()
 {
     srs_error_t err = srs_success;
 
-    // Update all context id to cid of session.
-    _srs_context->set_id(cid_);
-    owner_cid_->set_cid(cid_);
-
     if((err = handshake()) != srs_success) {
         return srs_error_wrap(err, "process rtc tcp pkt");
     }
 
     // TODO: FIXME: Handle all bytes of TCP Connection.
     while(true) {
-        if (!owner_coroutine_) return err;
-        if ((err = owner_coroutine_->pull()) != srs_success) {
+        if((err = trd_->pull()) != srs_success) {
             return srs_error_wrap(err, "rtc tcp conn");
         }
 
@@ -877,11 +859,11 @@ srs_error_t SrsRtcTcpConn::handshake()
 
     // Should support only one TCP candidate.
     SrsRtcTcpNetwork* network = dynamic_cast<SrsRtcTcpNetwork*>(session->tcp());
-    if (network->owner().get() != this) {
-        network->set_owner(*wrapper_);
+    if (!network->owner()) {
+        network->set_owner(this);
         session_ = session;
     }
-    if (network->owner().get() != this) {
+    if (network->owner() != this) {
         return srs_error_new(ERROR_RTC_TCP_UNIQUE, "only support one network");
     }
 
@@ -957,3 +939,21 @@ srs_error_t SrsRtcTcpConn::on_tcp_pkt(char* pkt, int nb_pkt)
     return srs_error_new(ERROR_RTC_UDP, "unknown packet");
 }
 
+void SrsRtcTcpConn::on_before_dispose(ISrsResource* c)
+{
+    if (!session_) return;
+
+    SrsRtcConnection* conn = dynamic_cast<SrsRtcConnection*>(c);
+    if(conn == session_) {
+        session_ = NULL;
+        // the related rtc connection will be disposed
+        srs_trace("RTC: tcp conn diposing, because of rtc connection");
+        trd_->interrupt();
+    }
+}
+
+void SrsRtcTcpConn::on_disposing(ISrsResource* c)
+{
+    return;
+}
+

trunk/src/app/srs_app_rtc_network.hpp

@@ -180,7 +180,7 @@ class SrsRtcTcpNetwork: public ISrsRtcNetwork
 private:
      // The DTLS transport over this network.
     ISrsRtcTransport* transport_;
-    SrsSharedResource<SrsRtcTcpConn> owner_;
+    SrsRtcTcpConn* owner_;
 private:
     std::string peer_ip_;
     int peer_port_;
@@ -189,8 +189,8 @@ class SrsRtcTcpNetwork: public ISrsRtcNetwork
     SrsRtcTcpNetwork(SrsRtcConnection* conn, SrsEphemeralDelta* delta);
     virtual ~SrsRtcTcpNetwork();
 public:
-    void set_owner(SrsSharedResource<SrsRtcTcpConn> v) { owner_ = v; }
-    SrsSharedResource<SrsRtcTcpConn> owner() { return owner_; }
+    void set_owner(SrsRtcTcpConn* v) { owner_ = v; }
+    SrsRtcTcpConn* owner() { return owner_; }
     void update_sendonly_socket(ISrsProtocolReadWriter* skt);
 //ISrsRtcNetwork 
 public:
@@ -232,9 +232,13 @@ class SrsRtcTcpNetwork: public ISrsRtcNetwork
 };
 
 // For WebRTC over TCP.
-class SrsRtcTcpConn : public ISrsConnection, public ISrsCoroutineHandler, public ISrsExecutorHandler
+class SrsRtcTcpConn : public ISrsConnection, public ISrsStartable, public ISrsCoroutineHandler, public ISrsDisposingHandler
 {
 private:
+    // The manager object to manage the connection.
+    ISrsResourceManager* manager_;
+    // Use a coroutine to serve the TCP connection.
+    SrsCoroutine* trd_;
     // The ip and port of client.
     std::string ip_;
     int port_;
@@ -245,34 +249,21 @@ class SrsRtcTcpConn : public ISrsConnection, public ISrsCoroutineHandler, public
     ISrsProtocolReadWriter* skt_;
     // Packet cache.
     char* pkt_;
-private:
-    // The shared resource which own this object, we should never free it because it's managed by shared ptr.
-    SrsSharedResource<SrsRtcTcpConn>* wrapper_;
-    // The owner coroutine, allow user to interrupt the loop.
-    ISrsInterruptable* owner_coroutine_;
-    ISrsContextIdSetter* owner_cid_;
-    SrsContextId cid_;
-public:
-    SrsRtcTcpConn();
-    SrsRtcTcpConn(ISrsProtocolReadWriter* skt, std::string cip, int port);
-    virtual ~SrsRtcTcpConn();
 public:
-    // Setup the owner, the wrapper is the shared ptr, the interruptable object is the coroutine, and the cid is the context id.
-    void setup_owner(SrsSharedResource<SrsRtcTcpConn>* wrapper, ISrsInterruptable* owner_coroutine, ISrsContextIdSetter* owner_cid);
+    SrsRtcTcpConn(ISrsProtocolReadWriter* skt, std::string cip, int port, ISrsResourceManager* cm);
+    virtual ~SrsRtcTcpConn();
 public:
     ISrsKbpsDelta* delta();
-    // Interrupt transport by session.
-    void interrupt();
 // Interface ISrsResource.
 public:
     virtual std::string desc();
     virtual const SrsContextId& get_id();
 // Interface ISrsConnection.
 public:
     virtual std::string remote_ip();
-// Interface ISrsExecutorHandler
+// Interface ISrsStartable
 public:
-    virtual void on_executor_done(ISrsInterruptable* executor);
+    virtual srs_error_t start();
 // Interface ISrsCoroutineHandler
 public:
     virtual srs_error_t cycle();
@@ -282,6 +273,10 @@ class SrsRtcTcpConn : public ISrsConnection, public ISrsCoroutineHandler, public
     srs_error_t read_packet(char* pkt, int* nb_pkt);
     srs_error_t on_stun(char* pkt, int nb_pkt);
     srs_error_t on_tcp_pkt(char* pkt, int nb_pkt);
+// Interface of ISrsDisposingHandler
+public:
+    virtual void on_before_dispose(ISrsResource* c);
+    virtual void on_disposing(ISrsResource* c);
 };
 
 #endif

trunk/src/app/srs_app_rtc_source.cpp

@@ -59,7 +59,6 @@ const int kVideoSamplerate  = 90000;
 
 using namespace std;
 
-#ifdef SRS_FFMPEG_FIT
 // The RTP payload max size, reserved some paddings for SRTP as such:
 //      kRtpPacketSize = kRtpMaxPayloadSize + paddings
 // For example, if kRtpPacketSize is 1500, recommend to set kRtpMaxPayloadSize to 1400,
@@ -69,7 +68,6 @@ using namespace std;
 // so we set kRtpMaxPayloadSize = 1200.
 // see @doc https://groups.google.com/g/discuss-webrtc/c/gH5ysR3SoZI
 const int kRtpMaxPayloadSize = kRtpPacketSize - 300;
-#endif
 
 // TODO: Add this function into SrsRtpMux class.
 srs_error_t aac_raw_append_adts_header(SrsSharedPtrMessage* shared_audio, SrsFormat* format, char** pbuf, int* pnn_buf)

trunk/src/app/srs_app_server.cpp

@@ -39,7 +39,6 @@ using namespace std;
 #include <srs_app_conn.hpp>
 #ifdef SRS_RTC
 #include <srs_app_rtc_network.hpp>
-#include <srs_app_rtc_server.hpp>
 #endif
 #ifdef SRS_GB28181
 #include <srs_app_gb28181.hpp>
@@ -1194,7 +1193,8 @@ srs_error_t SrsServer::do_on_tcp_client(ISrsListener* listener, srs_netfd_t& stf
         if (nn == 10 && b[0] == 0 && b[2] == 0 && b[3] == 1 && b[1] - b[5] == 20
             && b[6] == 0x21 && b[7] == 0x12 && b[8] == 0xa4 && b[9] == 0x42
         ) {
-            resource = new SrsRtcTcpConn(io, ip, port);
+            // TODO: FIXME: Should manage this connection by _srs_rtc_manager
+            resource = new SrsRtcTcpConn(io, ip, port, this);
         } else {
             resource = new SrsHttpxConn(listener == http_listener_, this, io, http_server, ip, port);
         }
@@ -1213,7 +1213,8 @@ srs_error_t SrsServer::do_on_tcp_client(ISrsListener* listener, srs_netfd_t& stf
             resource = new SrsHttpxConn(is_https, this, new SrsTcpConnection(stfd2), http_server, ip, port);
 #ifdef SRS_RTC
         } else if (listener == webrtc_listener_) {
-            resource = new SrsRtcTcpConn(new SrsTcpConnection(stfd2), ip, port);
+            // TODO: FIXME: Should manage this connection by _srs_rtc_manager
+            resource = new SrsRtcTcpConn(new SrsTcpConnection(stfd2), ip, port, this);
 #endif
         } else if (listener == exporter_listener_) {
             // TODO: FIXME: Maybe should support https metrics.
@@ -1226,28 +1227,11 @@ srs_error_t SrsServer::do_on_tcp_client(ISrsListener* listener, srs_netfd_t& stf
         }
     }
 
-#ifdef SRS_RTC
-    // For RTC TCP connection, use resource executor to manage the resource.
-    SrsRtcTcpConn* raw_conn = dynamic_cast<SrsRtcTcpConn*>(resource);
-    if (raw_conn) {
-        SrsSharedResource<SrsRtcTcpConn>* conn = new SrsSharedResource<SrsRtcTcpConn>(raw_conn);
-        SrsExecutorCoroutine* executor = new SrsExecutorCoroutine(_srs_rtc_manager, conn, raw_conn, raw_conn);
-        raw_conn->setup_owner(conn, executor, executor);
-        if ((err = executor->start()) != srs_success) {
-            srs_freep(executor);
-            return srs_error_wrap(err, "start executor");
-        }
-        return err;
-    }
-#endif
-
     // Use connection manager to manage all the resources.
-    srs_assert(resource);
     conn_manager->add(resource);
 
     // If connection is a resource to start, start a coroutine to handle it.
     ISrsStartable* conn = dynamic_cast<ISrsStartable*>(resource);
-    srs_assert(conn);
     if ((err = conn->start()) != srs_success) {
         return srs_error_wrap(err, "start conn coroutine");
     }