HLS: Support reload HLS asynchronously. v5.0.172 v6.0.67 (#3782)

When reloading HLS, it directly operates unpublish and publish. At this
time, if HLS is pushed, an exception may occur.

The reason is that these two coroutines operated on the HLS object at
the same time, causing a null pointer.

Solution: Use asynchronous reload. During reload, only set variables and
let the message processing coroutine implement the reload.

---------

Co-authored-by: Haibo Chen <495810242@qq.com>
Co-authored-by: chundonglinlin <chundonglinlin@163.com>

trunk/doc/CHANGELOG.md

@@ -7,6 +7,7 @@ The changelog for SRS.
 <a name="v6-changes"></a>
 
 ## SRS 6.0 Changelog
+* v6.0, 2023-08-25, Merge [#3782](https://github.com/ossrs/srs/pull/3782): HLS: Support reload HLS asynchronously. v6.0.67 (#3782)
 * v6.0, 2023-08-22, Merge [#3775](https://github.com/ossrs/srs/pull/3775): Bugfix: Log format output type does not match. v6.0.66 (#3699)
 * v6.0, 2023-08-02, Merge [#3750](https://github.com/ossrs/srs/pull/3750): HLS: Ignore empty NALU to avoid error. v6.0.64 (#3750)
 * v6.0, 2023-07-27, Merge [#3611](https://github.com/ossrs/srs/pull/3611): Design and implement helm capabilities to streamline the deployment process of an SRS cluster.. v6.0.63 (#3611)
@@ -78,6 +79,7 @@ The changelog for SRS.
 <a name="v5-changes"></a>
 
 ## SRS 5.0 Changelog
+* v5.0, 2023-08-25, Merge [#3782](https://github.com/ossrs/srs/pull/3782): HLS: Support reload HLS asynchronously.. v5.0.171 (#3782)
 * v5.0, 2023-08-22, Merge [#3775](https://github.com/ossrs/srs/pull/3775): Bugfix: Log format output type does not match. v5.0.171 (#3699)
 * v5.0, 2023-08-02, HLS: Ignore empty NALU to avoid error. v5.0.170
 * v5.0, 2023-07-26, Merge [#3699](https://github.com/ossrs/srs/pull/3699): Bugfix: Eliminate the redundant declaration of the _srs_rtc_manager variable. v5.0.168 (#3699)

trunk/src/app/srs_app_hls.cpp

@@ -1136,6 +1136,7 @@ SrsHls::SrsHls()
     
     enabled = false;
     disposable = false;
+    async_reload_ = reloading_ = false;
     last_update_time = 0;
     hls_dts_directly = false;
     
@@ -1155,6 +1156,53 @@ SrsHls::~SrsHls()
     srs_freep(pprint);
 }
 
+void SrsHls::async_reload()
+{
+    async_reload_ = true;
+}
+
+srs_error_t SrsHls::reload()
+{
+    srs_error_t err = srs_success;
+
+    // Ignore if not active.
+    if (!enabled) return err;
+
+    int reloading = 0, reloaded = 0, refreshed = 0;
+    err = do_reload(&reloading, &reloaded, &refreshed);
+    srs_trace("async reload hls %s, reloading=%d, reloaded=%d, refreshed=%d",
+        req->get_stream_url().c_str(), reloading, reloaded, refreshed);
+
+    return err;
+}
+
+srs_error_t SrsHls::do_reload(int *reloading, int *reloaded, int *refreshed)
+{
+    srs_error_t err = srs_success;
+
+    if (!async_reload_ || reloading_) return err;
+    reloading_ = true;
+    *reloading = 1;
+
+    on_unpublish();
+    if ((err = on_publish()) != srs_success) {
+        return srs_error_wrap(err, "hls publish failed");
+    }
+    *reloaded = 1;
+
+    // Before feed the sequence header, must reset the reloading.
+    reloading_ = false;
+    async_reload_ = false;
+
+    // After reloading, we must request the sequence header again.
+    if ((err = hub->on_hls_request_sh()) != srs_success) {
+        return srs_error_wrap(err, "hls request sh");
+    }
+    *refreshed = 1;
+
+    return err;
+}
+
 void SrsHls::dispose()
 {
     if (enabled) {
@@ -1183,6 +1231,10 @@ srs_error_t SrsHls::cycle()
         return err;
     }
 
+    // When reloading, we must wait for it done.
+    if (async_reload_) return err;
+
+    // If not unpublishing and not reloading, try to dispose HLS stream.
     srs_utime_t hls_dispose = _srs_config->get_hls_dispose(req->vhost);
     if (hls_dispose <= 0) {
         return err;
@@ -1271,9 +1323,9 @@ srs_error_t SrsHls::on_audio(SrsSharedPtrMessage* shared_audio, SrsFormat* forma
 {
     srs_error_t err = srs_success;
 
-    if (!enabled) {
-        return err;
-    }
+    // If not able to transmux to HLS, ignore.
+    if (!enabled) return err;
+    if (async_reload_) return reload();
 
     // Ignore if no format->acodec, it means the codec is not parsed, or unknown codec.
     // @issue https://github.com/ossrs/srs/issues/1506#issuecomment-562079474
@@ -1353,9 +1405,9 @@ srs_error_t SrsHls::on_video(SrsSharedPtrMessage* shared_video, SrsFormat* forma
 {
     srs_error_t err = srs_success;
 
-    if (!enabled) {
-        return err;
-    }
+    // If not able to transmux to HLS, ignore.
+    if (!enabled) return err;
+    if (async_reload_) return reload();
 
     // Ignore if no format->vcodec, it means the codec is not parsed, or unknown codec.
     // @issue https://github.com/ossrs/srs/issues/1506#issuecomment-562079474

trunk/src/app/srs_app_hls.hpp

@@ -275,8 +275,14 @@ private:
     SrsHlsController* controller;
 private:
     SrsRequest* req;
+    // Whether the HLS is enabled.
     bool enabled;
+    // Whether the HLS stream is able to be disposed.
     bool disposable;
+    // Whether requires HLS to do reload asynchronously.
+    bool async_reload_;
+    bool reloading_;
+    // To detect heartbeat and dipose it if configured.
     srs_utime_t last_update_time;
 private:
     // If the diff=dts-previous_audio_dts is about 23,
@@ -293,6 +299,11 @@ private:
 public:
     SrsHls();
     virtual ~SrsHls();
+public:
+    virtual void async_reload();
+private:
+    srs_error_t reload();
+    srs_error_t do_reload(int *reloading, int *reloaded, int *refreshed);
 public:
     virtual void dispose();
     virtual srs_error_t cycle();

trunk/src/app/srs_app_source.cpp

@@ -1228,6 +1228,27 @@ srs_error_t SrsOriginHub::on_dvr_request_sh()
     return err;
 }
 
+srs_error_t SrsOriginHub::on_hls_request_sh()
+{
+    srs_error_t err = srs_success;
+
+    SrsSharedPtrMessage* cache_sh_video = source->meta->vsh();
+    if (cache_sh_video) {
+        if ((err = hls->on_video(cache_sh_video, source->meta->vsh_format())) != srs_success) {
+            return srs_error_wrap(err, "hls video");
+        }
+    }
+
+    SrsSharedPtrMessage* cache_sh_audio = source->meta->ash();
+    if (cache_sh_audio) {
+        if ((err = hls->on_audio(cache_sh_audio, source->meta->ash_format())) != srs_success) {
+            return srs_error_wrap(err, "hls audio");
+        }
+    }
+
+    return err;
+}
+
 srs_error_t SrsOriginHub::on_reload_vhost_forward(string vhost)
 {
     srs_error_t err = srs_success;
@@ -1236,7 +1257,7 @@ srs_error_t SrsOriginHub::on_reload_vhost_forward(string vhost)
         return err;
     }
     
-    // TODO: FIXME: maybe should ignore when publish already stopped?
+    // TODO: FIXME: Must do async reload, see SrsHls::async_reload.
     
     // forwarders
     destroy_forwarders();
@@ -1263,6 +1284,8 @@ srs_error_t SrsOriginHub::on_reload_vhost_dash(string vhost)
         return err;
     }
 
+    // TODO: FIXME: Must do async reload, see SrsHls::async_reload.
+
     dash->on_unpublish();
     
     // Don't start DASH when source is not active.
@@ -1307,46 +1330,7 @@ srs_error_t SrsOriginHub::on_reload_vhost_hls(string vhost)
         return err;
     }
 
-    // TODO: FIXME: maybe should ignore when publish already stopped?
-    
-    hls->on_unpublish();
-    
-    // Don't start HLS when source is not active.
-    if (!is_active) {
-        return err;
-    }
-    
-    if ((err = hls->on_publish()) != srs_success) {
-        return srs_error_wrap(err, "hls publish failed");
-    }
-    srs_trace("vhost %s hls reload success", vhost.c_str());
-
-    SrsRtmpFormat* format = source->format_;
-    
-    // when publish, don't need to fetch sequence header, which is old and maybe corrupt.
-    // when reload, we must fetch the sequence header from source cache.
-    // notice the source to get the cached sequence header.
-    // when reload to start hls, hls will never get the sequence header in stream,
-    // use the SrsLiveSource.on_hls_start to push the sequence header to HLS.
-    SrsSharedPtrMessage* cache_sh_video = source->meta->vsh();
-    if (cache_sh_video) {
-        if ((err = format->on_video(cache_sh_video)) != srs_success) {
-            return srs_error_wrap(err, "format on_video");
-        }
-        if ((err = hls->on_video(cache_sh_video, format)) != srs_success) {
-            return srs_error_wrap(err, "hls on_video");
-        }
-    }
-    
-    SrsSharedPtrMessage* cache_sh_audio = source->meta->ash();
-    if (cache_sh_audio) {
-        if ((err = format->on_audio(cache_sh_audio)) != srs_success) {
-            return srs_error_wrap(err, "format on_audio");
-        }
-        if ((err = hls->on_audio(cache_sh_audio, format)) != srs_success) {
-            return srs_error_wrap(err, "hls on_audio");
-        }
-    }
+    hls->async_reload();
     
     return err;
 }
@@ -1359,7 +1343,7 @@ srs_error_t SrsOriginHub::on_reload_vhost_hds(string vhost)
         return err;
     }
 
-    // TODO: FIXME: maybe should ignore when publish already stopped?
+    // TODO: FIXME: Must do async reload, see SrsHls::async_reload.
     
 #ifdef SRS_HDS
     hds->on_unpublish();
@@ -1386,7 +1370,7 @@ srs_error_t SrsOriginHub::on_reload_vhost_dvr(string vhost)
         return err;
     }
 
-    // TODO: FIXME: maybe should ignore when publish already stopped?
+    // TODO: FIXME: Must do async reload, see SrsHls::async_reload.
     
     // cleanup dvr
     dvr->on_unpublish();
@@ -1423,7 +1407,7 @@ srs_error_t SrsOriginHub::on_reload_vhost_transcode(string vhost)
         return err;
     }
 
-    // TODO: FIXME: maybe should ignore when publish already stopped?
+    // TODO: FIXME: Must do async reload, see SrsHls::async_reload.
     
     encoder->on_unpublish();
     
@@ -1448,7 +1432,7 @@ srs_error_t SrsOriginHub::on_reload_vhost_exec(string vhost)
         return err;
     }
 
-    // TODO: FIXME: maybe should ignore when publish already stopped?
+    // TODO: FIXME: Must do async reload, see SrsHls::async_reload.
     
     ng_exec->on_unpublish();
     

trunk/src/app/srs_app_source.hpp

@@ -368,6 +368,8 @@ public:
     virtual srs_error_t on_forwarder_start(SrsForwarder* forwarder);
     // For the SrsDvr to callback to request the sequence headers.
     virtual srs_error_t on_dvr_request_sh();
+    // For the SrsHls to callback to request the sequence headers.
+    virtual srs_error_t on_hls_request_sh();
 // Interface ISrsReloadHandler
 public:
     virtual srs_error_t on_reload_vhost_forward(std::string vhost);

trunk/src/core/srs_core_version5.hpp

@@ -9,6 +9,6 @@
 
 #define VERSION_MAJOR       5
 #define VERSION_MINOR       0
-#define VERSION_REVISION    171
+#define VERSION_REVISION    172
 
 #endif

trunk/src/core/srs_core_version6.hpp

@@ -9,6 +9,6 @@
 
 #define VERSION_MAJOR       6
 #define VERSION_MINOR       0
-#define VERSION_REVISION    66
+#define VERSION_REVISION    67
 
 #endif