HTTP-FLV: Crash when multiple viewers. v6.0.148 (#4144)

I did some preliminary code inspection. The two playback endpoints share
the same `SrsLiveStream` instance. After the first one disconnects,
`alive_` is set to false.
```
  alive_ = true;
  err = do_serve_http(w, r);
  alive_ = false;
```

In the `SrsHttpStreamServer::http_unmount(SrsRequest* r)` function,
`stream->alive()` is already false, so `mux.unhandle` will free the
`SrsLiveStream`. This causes the other connection coroutine to return to
its execution environment after the `SrsLiveStream` instance has already
been freed.
```
    // Wait for cache and stream to stop.
    int i = 0;
    for (; i < 1024; i++) {
        if (!cache->alive() && !stream->alive()) {
            break;
        }
        srs_usleep(100 * SRS_UTIME_MILLISECONDS);
    }

    // Unmount the HTTP handler, which will free the entry. Note that we must free it after cache and
    // stream stopped for it uses it.
    mux.unhandle(entry->mount, stream.get());
```

`alive_` was changed from a `bool` to an `int` to ensure that
`mux.unhandle` is only executed after each connection's `serve_http` has
exited.

---------

Co-authored-by: liumengte <liumengte@visionular.com>
Co-authored-by: winlin <winlinvip@gmail.com>

trunk/doc/CHANGELOG.md

@@ -7,6 +7,7 @@ The changelog for SRS.
 <a name="v6-changes"></a>
 
 ## SRS 6.0 Changelog
+* v6.0, 2024-08-15, Merge [#4144](https://github.com/ossrs/srs/pull/4144): HTTP-FLV: Crash when multiple viewers. v6.0.148 (#4144)
 * v6.0, 2024-08-15, Merge [#4142](https://github.com/ossrs/srs/pull/4142): Config: Add more utest for env config. v6.0.147 (#4142)
 * v6.0, 2024-08-14, Merge [#4141](https://github.com/ossrs/srs/pull/4141): Live: Crash for invalid live stream state when unmount HTTP. v6.0.146 (#4141)
 * v6.0, 2024-08-13, Merge [#4092](https://github.com/ossrs/srs/pull/4092): Config: Improve env config to support multi values. v6.0.146 (#4092)

trunk/src/app/srs_app_http_stream.cpp

@@ -583,7 +583,7 @@ SrsLiveStream::SrsLiveStream(SrsRequest* r, SrsBufferCache* c)
     cache = c;
     req = r->copy()->as_http();
     security_ = new SrsSecurity();
-    alive_ = false;
+    alive_viewers_ = 0;
 }
 
 SrsLiveStream::~SrsLiveStream()
@@ -634,9 +634,9 @@ srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage
         return srs_error_wrap(err, "http hook");
     }
 
-    alive_ = true;
+    alive_viewers_++;
     err = do_serve_http(w, r);
-    alive_ = false;
+    alive_viewers_--;
     
     http_hooks_on_stop(r);
     
@@ -645,7 +645,7 @@ srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage
 
 bool SrsLiveStream::alive()
 {
-    return alive_;
+    return alive_viewers_ > 0;
 }
 
 srs_error_t SrsLiveStream::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)

trunk/src/app/srs_app_http_stream.hpp

@@ -182,7 +182,10 @@ private:
     SrsRequest* req;
     SrsBufferCache* cache;
     SrsSecurity* security_;
-    bool alive_;
+    // For multiple viewers, which means there will more than one alive viewers for a live stream, so we must
+    // use an int value to represent if there is any viewer is alive. We should never do cleanup unless all
+    // viewers closed the connection.
+    int alive_viewers_;
 public:
     SrsLiveStream(SrsRequest* r, SrsBufferCache* c);
     virtual ~SrsLiveStream();

trunk/src/app/srs_app_source.cpp

@@ -1886,7 +1886,7 @@ SrsLiveSource::SrsLiveSource()
     mix_correct = false;
     mix_queue = new SrsMixQueue();
     
-    _can_publish = true;
+    can_publish_ = true;
     stream_die_at_ = 0;
     publisher_idle_at_ = 0;
 
@@ -1952,7 +1952,7 @@ srs_error_t SrsLiveSource::cycle()
 bool SrsLiveSource::stream_is_dead()
 {
     // still publishing?
-    if (!_can_publish || !publish_edge->can_publish()) {
+    if (!can_publish_ || !publish_edge->can_publish()) {
         return false;
     }
     
@@ -2151,7 +2151,7 @@ SrsContextId SrsLiveSource::pre_source_id()
 
 bool SrsLiveSource::inactive()
 {
-    return _can_publish;
+    return can_publish_;
 }
 
 void SrsLiveSource::update_auth(SrsRequest* r)
@@ -2167,7 +2167,7 @@ bool SrsLiveSource::can_publish(bool is_edge)
         return publish_edge->can_publish();
     }
     
-    return _can_publish;
+    return can_publish_;
 }
 
 srs_error_t SrsLiveSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata)
@@ -2566,7 +2566,7 @@ srs_error_t SrsLiveSource::on_publish()
     // update the request object.
     srs_assert(req);
     
-    _can_publish = false;
+    can_publish_ = false;
     
     // whatever, the publish thread is the source or edge source,
     // save its id to srouce id.
@@ -2614,7 +2614,7 @@ srs_error_t SrsLiveSource::on_publish()
 void SrsLiveSource::on_unpublish()
 {
     // ignore when already unpublished.
-    if (_can_publish) {
+    if (can_publish_) {
         return;
     }
     
@@ -2655,7 +2655,10 @@ void SrsLiveSource::on_unpublish()
         stream_die_at_ = srs_get_system_time();
     }
 
-    _can_publish = true;
+    // Note that we should never set to unpublish before any other handler is done, especially the handler
+    // which is actually an http stream that unmounts the HTTP path for streaming, because there maybe some
+    // coroutine switch in these handlers.
+    can_publish_ = true;
 }
 
 srs_error_t SrsLiveSource::create_consumer(SrsLiveConsumer*& consumer)
@@ -2735,7 +2738,7 @@ void SrsLiveSource::on_consumer_destroy(SrsLiveConsumer* consumer)
         play_edge->on_all_client_stop();
 
         // If no publishers, the stream is die.
-        if (_can_publish) {
+        if (can_publish_) {
             stream_die_at_ = srs_get_system_time();
         }
 

trunk/src/app/srs_app_source.hpp

@@ -529,7 +529,7 @@ private:
     SrsRtmpFormat* format_;
 private:
     // Whether source is avaiable for publishing.
-    bool _can_publish;
+    bool can_publish_;
     // The last die time, while die means neither publishers nor players.
     srs_utime_t stream_die_at_;
     // The last idle time, while idle means no players.

trunk/src/core/srs_core_version6.hpp

@@ -9,6 +9,6 @@
 
 #define VERSION_MAJOR       6
 #define VERSION_MINOR       0
-#define VERSION_REVISION    147
+#define VERSION_REVISION    148
 
 #endif