Edge: Improve stability for state and fd closing. v5.0.214 (#4126)

1. Should always stop coroutine before close fd, see #511, #1784
2. When edge forwarder coroutine quit, always set the error code.
3. Do not unpublish if invalid state.

---------

Co-authored-by: Jacob Su <[email protected]>

Author: winlinvip

trunk/doc/CHANGELOG.md

@@ -7,6 +7,7 @@ The changelog for SRS.
 <a name="v5-changes"></a>
 
 ## SRS 5.0 Changelog
+* v5.0, 2024-07-24, Merge [#4126](https://github.com/ossrs/srs/pull/4126): Edge: Improve stability for state and fd closing. v5.0.214 (#4126)
 * v5.0, 2024-06-03, Merge [#4057](https://github.com/ossrs/srs/pull/4057): RTC: Support dropping h.264 SEI from NALUs. v5.0.213 (#4057)
 * v5.0, 2024-04-23, Merge [#4038](https://github.com/ossrs/srs/pull/4038): RTMP: Do not response publish start message if hooks fail. v5.0.212 (#4038)
 * v5.0, 2024-04-22, Merge [#4033](https://github.com/ossrs/srs/pull/4033): issue #3967: support x509 certification chiain in single pem file. v5.0.211 (#4033)

trunk/src/app/srs_app_edge.cpp

@@ -780,6 +780,10 @@ srs_error_t SrsEdgeForwarder::start()
 
         url = srs_generate_rtmp_url(server, port, req->host, vhost, req->app, req->stream, req->param);
     }
+
+    // We must stop the coroutine before disposing the sdk.
+    srs_freep(trd);
+    trd = new SrsSTCoroutine("edge-fwr", this, _srs_context->get_id());
 
     // open socket.
     srs_freep(sdk);
@@ -804,10 +808,8 @@ srs_error_t SrsEdgeForwarder::start()
     if ((err = sdk->publish(_srs_config->get_chunk_size(req->vhost), false, &stream)) != srs_success) {
         return srs_error_wrap(err, "sdk publish");
     }
-    
-    srs_freep(trd);
-    trd = new SrsSTCoroutine("edge-fwr", this, _srs_context->get_id());
-    
+
+    // Start the forwarding coroutine.
     if ((err = trd->start()) != srs_success) {
         return srs_error_wrap(err, "coroutine");
     }
@@ -819,9 +821,12 @@ srs_error_t SrsEdgeForwarder::start()
 
 void SrsEdgeForwarder::stop()
 {
+    // Make sure the coroutine is stopped before disposing the sdk,
+    // for sdk is used by coroutine.
     trd->stop();
-    queue->clear();
     srs_freep(sdk);
+
+    queue->clear();
 }
 
 // when error, edge ingester sleep for a while and retry.
@@ -839,6 +844,11 @@ srs_error_t SrsEdgeForwarder::cycle()
         }
 
         if ((err = do_cycle()) != srs_success) {
+            // If cycle stopping, we should always set the quit error code.
+            if (send_error_code == 0) {
+                send_error_code = srs_error_code(err);
+            }
+
             return srs_error_wrap(err, "do cycle");
         }
 

trunk/src/app/srs_app_rtmp_conn.cpp

@@ -946,25 +946,22 @@ srs_error_t SrsRtmpConn::publishing(SrsLiveSource* source)
     }
 
     // TODO: FIXME: Should refine the state of publishing.
-    if ((err = acquire_publish(source)) == srs_success) {
+    srs_error_t acquire_err = acquire_publish(source);
+    if ((err = acquire_err) == srs_success) {
         // use isolate thread to recv,
         // @see: https://github.com/ossrs/srs/issues/237
         SrsPublishRecvThread rtrd(rtmp, req, srs_netfd_fileno(stfd), 0, this, source, _srs_context->get_id());
         err = do_publishing(source, &rtrd);
         rtrd.stop();
     }
 
-    // whatever the acquire publish, always release publish.
-    // when the acquire error in the midlle-way, the publish state changed,
-    // but failed, so we must cleanup it.
-    // @see https://github.com/ossrs/srs/issues/474
-    // @remark when stream is busy, should never release it.
-    if (srs_error_code(err) != ERROR_SYSTEM_STREAM_BUSY) {
+    // Release and callback when acquire publishing success, if not, we should ignore, because the source
+    // is not published by this session.
+    if (acquire_err == srs_success) {
         release_publish(source);
+        http_hooks_on_unpublish();
     }
 
-    http_hooks_on_unpublish();
-    
     return err;
 }
 

trunk/src/core/srs_core_version5.hpp

@@ -9,6 +9,6 @@
 
 #define VERSION_MAJOR       5
 #define VERSION_MINOR       0
-#define VERSION_REVISION    213
+#define VERSION_REVISION    214
 
 #endif