Skip to content

Commit

Permalink
FLV: Refine source and http handler. v6.0.155 v7.0.14 (#4165)
Browse files Browse the repository at this point in the history
1. Do not create a source when mounting FLV because it may not unmount
FLV when freeing the source. If you access the FLV stream without any
publisher, then wait for source cleanup and review the FLV stream again,
there is an annoying warning message.

```bash
# View HTTP FLV stream by curl, wait for stream to be ready.
# curl http://localhost:8080/live/livestream.flv -v >/dev/null
HTTP #0 127.0.0.1:58026 GET http://localhost:8080/live/livestream.flv, content-length=-1
new live source, stream_url=/live/livestream
http: mount flv stream for sid=/live/livestream, mount=/live/livestream.flv

# Cancel the curl and trigger source cleanup without http unmount.
client disconnect peer. ret=1007
Live: cleanup die source, id=[], total=1

# View the stream again, it fails.
# curl http://localhost:8080/live/livestream.flv -v >/dev/null
HTTP #0 127.0.0.1:58040 GET http://localhost:8080/live/livestream.flv, content-length=-1
serve error code=1097(NoSource)(No source found) : process request=0 : cors serve : serve http : no source for /live/livestream
serve_http() [srs_app_http_stream.cpp:641]
```

> Note: There is an inconsistency. The first time, you can access the
FLV stream and wait for the publisher, but the next time, you cannot.

2. Create a source when starting to serve the FLV client. We do not need
to create the source when creating the HTTP handler. Instead, we should
try to create the source in the cache or stream. Because the source
cleanup does not unmount the HTTP handler, the handler remains after the
source is destroyed. The next time you access the FLV stream, the source
is not found.

```cpp
srs_error_t SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph) {
    SrsSharedPtr<SrsLiveSource> live_source;
    if ((err = _srs_sources->fetch_or_create(r.get(), server, live_source)) != srs_success) { }
    if ((err = http_mount(r.get())) != srs_success) { }

srs_error_t SrsBufferCache::cycle() {
    SrsSharedPtr<SrsLiveSource> live_source = _srs_sources->fetch(req);
    if (!live_source.get()) {
        return srs_error_new(ERROR_NO_SOURCE, "no source for %s", req->get_stream_url().c_str());
    }

srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) {
    SrsSharedPtr<SrsLiveSource> live_source = _srs_sources->fetch(req);
    if (!live_source.get()) {
        return srs_error_new(ERROR_NO_SOURCE, "no source for %s", req->get_stream_url().c_str());
    }
```

> Note: We should not create the source in hijack, instead, we create it
in cache or stream:

```cpp
srs_error_t SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph) {
    if ((err = http_mount(r.get())) != srs_success) { }

srs_error_t SrsBufferCache::cycle() {
    SrsSharedPtr<SrsLiveSource> live_source;
    if ((err = _srs_sources->fetch_or_create(req, server_, live_source)) != srs_success) { }

srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) {
    SrsSharedPtr<SrsLiveSource> live_source;
    if ((err = _srs_sources->fetch_or_create(req, server_, live_source)) != srs_success) { }
```

> Note: This fixes the failure and annoying warning message, and
maintains consistency by always waiting for the stream to be ready if
there is no publisher.

3. Fail the http request if the HTTP handler is disposing, and also keep
the handler entry when disposing the stream, because we should dispose
the handler entry and stream at the same time.

```cpp
srs_error_t SrsHttpStreamServer::http_mount(SrsRequest* r) {
        entry = streamHandlers[sid];
        if (entry->disposing) {
            return srs_error_new(ERROR_STREAM_DISPOSING, "stream is disposing");
        }

void SrsHttpStreamServer::http_unmount(SrsRequest* r) {
    std::map<std::string, SrsLiveEntry*>::iterator it = streamHandlers.find(sid);
    SrsUniquePtr<SrsLiveEntry> entry(it->second);
    entry->disposing = true;
```

> Note: If the disposal process takes a long time, this will prevent
unexpected behavior or access to the resource that is being disposed of.

4. In edge mode, the edge ingester will unpublish the source when the
last consumer quits, which is actually triggered by the HTTP stream.
While it also waits for the stream to quit when the HTTP unmounts, there
is a self-destruction risk: the HTTP live stream object destroys itself.

```cpp
srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) {
    SrsUniquePtr<SrsLiveConsumer> consumer(consumer_raw); // Trigger destroy.

void SrsHttpStreamServer::http_unmount(SrsRequest* r) {
    for (;;) { if (!cache->alive() && !stream->alive()) { break; } // A circle reference.
    mux.unhandle(entry->mount, stream.get()); // Free the SrsLiveStream itself.
```

> Note: It also introduces a circular reference in the object
relationships, the stream reference to itself when unmount:

```text
SrsLiveStream::serve_http 
    -> SrsLiveConsumer::~SrsLiveConsumer -> SrsEdgeIngester::stop 
    -> SrsLiveSource::on_unpublish -> SrsHttpStreamServer::http_unmount 
        -> SrsLiveStream::alive
```

> Note: We should use an asynchronous worker to perform the cleanup to
avoid the stream destroying itself and to prevent self-referencing.

```cpp
void SrsHttpStreamServer::http_unmount(SrsRequest* r) {
    entry->disposing = true;
    if ((err = async_->execute(new SrsHttpStreamDestroy(&mux, &streamHandlers, sid))) != srs_success) { }
```

> Note: This also ensures there are no circular references and no
self-destruction.

---------

Co-authored-by: Jacob Su <[email protected]>
  • Loading branch information
winlinvip and suzp1984 authored Sep 1, 2024
1 parent 740f0d3 commit 15fbe45
Show file tree
Hide file tree
Showing 6 changed files with 154 additions and 64 deletions.
2 changes: 2 additions & 0 deletions trunk/doc/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ The changelog for SRS.
<a name="v7-changes"></a>

## SRS 7.0 Changelog
* v7.0, 2024-09-01, Merge [#4165](https://github.com/ossrs/srs/pull/4165): FLV: Refine source and http handler. v7.0.14 (#4165)
* v7.0, 2024-09-01, Merge [#4166](https://github.com/ossrs/srs/pull/4166): Edge: Fix flv edge crash when http unmount. v7.0.13 (#4166)
* v7.0, 2024-08-31, Merge [#4162](https://github.com/ossrs/srs/pull/4162): Fix #3767: RTMP: Do not response empty data packet. v7.0.12 (#4162)
* v7.0, 2024-08-31, Merge [#4164](https://github.com/ossrs/srs/pull/4164): HTTP-FLV: Notify connection to expire when unpublishing. v7.0.11 (#4164)
Expand All @@ -25,6 +26,7 @@ The changelog for SRS.
<a name="v6-changes"></a>

## SRS 6.0 Changelog
* v6.0, 2024-09-01, Merge [#4165](https://github.com/ossrs/srs/pull/4165): FLV: Refine source and http handler. v6.0.155 (#4165)
* v6.0, 2024-09-01, Merge [#4166](https://github.com/ossrs/srs/pull/4166): Edge: Fix flv edge crash when http unmount. v6.0.154 (#4166)
* v6.0, 2024-08-31, Merge [#4162](https://github.com/ossrs/srs/pull/4162): Fix #3767: RTMP: Do not response empty data packet. v6.0.153 (#4162)
* v6.0, 2024-08-31, Merge [#4164](https://github.com/ossrs/srs/pull/4164): HTTP-FLV: Notify connection to expire when unpublishing. v6.0.152 (#4164)
Expand Down
183 changes: 124 additions & 59 deletions trunk/src/app/srs_app_http_stream.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -39,15 +39,17 @@ using namespace std;
#include <srs_app_statistic.hpp>
#include <srs_app_recv_thread.hpp>
#include <srs_app_http_hooks.hpp>
#include <srs_app_async_call.hpp>

SrsBufferCache::SrsBufferCache(SrsRequest* r)
SrsBufferCache::SrsBufferCache(SrsServer* s, SrsRequest* r)
{
req = r->copy()->as_http();
queue = new SrsMessageQueue(true);
trd = new SrsSTCoroutine("http-stream", this);

// TODO: FIXME: support reload.
fast_cache = _srs_config->get_vhost_http_remux_fast_cache(req->vhost);
server_ = s;
}

SrsBufferCache::~SrsBufferCache()
Expand All @@ -69,6 +71,11 @@ srs_error_t SrsBufferCache::update_auth(SrsRequest* r)
srs_error_t SrsBufferCache::start()
{
srs_error_t err = srs_success;

// Not enabled.
if (fast_cache <= 0) {
return err;
}

if ((err = trd->start()) != srs_success) {
return srs_error_wrap(err, "corotine");
Expand All @@ -79,11 +86,21 @@ srs_error_t SrsBufferCache::start()

void SrsBufferCache::stop()
{
// Not enabled.
if (fast_cache <= 0) {
return;
}

trd->stop();
}

bool SrsBufferCache::alive()
{
// Not enabled.
if (fast_cache <= 0) {
return false;
}

srs_error_t err = trd->pull();
if (err == srs_success) {
return true;
Expand Down Expand Up @@ -115,17 +132,12 @@ srs_error_t SrsBufferCache::dump_cache(SrsLiveConsumer* consumer, SrsRtmpJitterA
srs_error_t SrsBufferCache::cycle()
{
srs_error_t err = srs_success;

// TODO: FIXME: support reload.
if (fast_cache <= 0) {
srs_usleep(SRS_STREAM_CACHE_CYCLE);
return err;
}

SrsSharedPtr<SrsLiveSource> live_source = _srs_sources->fetch(req);
if (!live_source.get()) {
return srs_error_new(ERROR_NO_SOURCE, "no source for %s", req->get_stream_url().c_str());
SrsSharedPtr<SrsLiveSource> live_source;
if ((err = _srs_sources->fetch_or_create(req, server_, live_source)) != srs_success) {
return srs_error_wrap(err, "source create");
}
srs_assert(live_source.get() != NULL);

// the stream cache will create consumer to cache stream,
// which will trigger to fetch stream from origin for edge.
Expand Down Expand Up @@ -578,11 +590,12 @@ srs_error_t SrsBufferWriter::writev(const iovec* iov, int iovcnt, ssize_t* pnwri
return writer->writev(iov, iovcnt, pnwrite);
}

SrsLiveStream::SrsLiveStream(SrsRequest* r, SrsBufferCache* c)
SrsLiveStream::SrsLiveStream(SrsServer* s, SrsRequest* r, SrsBufferCache* c)
{
cache = c;
req = r->copy()->as_http();
security_ = new SrsSecurity();
server_ = s;
}

SrsLiveStream::~SrsLiveStream()
Expand Down Expand Up @@ -636,10 +649,17 @@ srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage
return srs_error_wrap(err, "http hook");
}

SrsSharedPtr<SrsLiveSource> live_source = _srs_sources->fetch(req);
if (!live_source.get()) {
return srs_error_new(ERROR_NO_SOURCE, "no source for %s", req->get_stream_url().c_str());
// Always try to create the source, because http handler won't create it.
SrsSharedPtr<SrsLiveSource> live_source;
if ((err = _srs_sources->fetch_or_create(req, server_, live_source)) != srs_success) {
return srs_error_wrap(err, "source create");
}
srs_assert(live_source.get() != NULL);

bool enabled_cache = _srs_config->get_gop_cache(req->vhost);
int gcmf = _srs_config->get_gop_cache_max_frames(req->vhost);
live_source->set_cache(enabled_cache);
live_source->set_gop_cache_max_frames(gcmf);

// Create consumer of source, ignore gop cache, use the audio gop cache.
SrsLiveConsumer* consumer_raw = NULL;
Expand Down Expand Up @@ -926,6 +946,7 @@ srs_error_t SrsLiveStream::streaming_send_messages(ISrsBufferEncoder* enc, SrsSh
SrsLiveEntry::SrsLiveEntry(std::string m)
{
mount = m;
disposing = false;

stream = NULL;
cache = NULL;
Expand Down Expand Up @@ -967,6 +988,7 @@ bool SrsLiveEntry::is_mp3()
SrsHttpStreamServer::SrsHttpStreamServer(SrsServer* svr)
{
server = svr;
async_ = new SrsAsyncCallWorker();

mux.hijack(this);
_srs_config->subscribe(this);
Expand All @@ -976,6 +998,9 @@ SrsHttpStreamServer::~SrsHttpStreamServer()
{
mux.unhijack(this);
_srs_config->unsubscribe(this);

async_->stop();
srs_freep(async_);

if (true) {
std::map<std::string, SrsLiveEntry*>::iterator it;
Expand Down Expand Up @@ -1003,6 +1028,10 @@ srs_error_t SrsHttpStreamServer::initialize()
if ((err = initialize_flv_streaming()) != srs_success) {
return srs_error_wrap(err, "http flv stream");
}

if ((err = async_->start()) != srs_success) {
return srs_error_wrap(err, "async start");
}

return err;
}
Expand Down Expand Up @@ -1037,8 +1066,8 @@ srs_error_t SrsHttpStreamServer::http_mount(SrsRequest* r)
entry = new SrsLiveEntry(mount);

entry->req = r->copy()->as_http();
entry->cache = new SrsBufferCache(r);
entry->stream = new SrsLiveStream(r, entry->cache);
entry->cache = new SrsBufferCache(server, r);
entry->stream = new SrsLiveStream(server, r, entry->cache);

// TODO: FIXME: maybe refine the logic of http remux service.
// if user push streams followed:
Expand Down Expand Up @@ -1067,6 +1096,12 @@ srs_error_t SrsHttpStreamServer::http_mount(SrsRequest* r)
} else {
// The entry exists, we reuse it and update the request of stream and cache.
entry = streamHandlers[sid];

// Fail if system is disposing the entry.
if (entry->disposing) {
return srs_error_new(ERROR_STREAM_DISPOSING, "stream is disposing");
}

entry->stream->update_auth(r);
entry->cache->update_auth(r);
}
Expand All @@ -1088,36 +1123,19 @@ void SrsHttpStreamServer::http_unmount(SrsRequest* r)
return;
}

// Free all HTTP resources.
SrsUniquePtr<SrsLiveEntry> entry(it->second);
streamHandlers.erase(it);

SrsUniquePtr<SrsLiveStream> stream(entry->stream);
SrsUniquePtr<SrsBufferCache> cache(entry->cache);

// Notify cache and stream to stop.
if (stream->entry) stream->entry->enabled = false;
stream->expire();
cache->stop();

// 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);
// Set the entry to disposing, which will prevent the stream to be reused.
SrsLiveEntry* entry = it->second;
if (entry->disposing) {
return;
}
entry->disposing = true;

if (cache->alive() || stream->alive()) {
srs_warn("http: try to free a alive stream, cache=%d, stream=%d", cache->alive(), stream->alive());
// Use async worker to execute the task, which will destroy the stream.
srs_error_t err = srs_success;
if ((err = async_->execute(new SrsHttpStreamDestroy(&mux, &streamHandlers, sid))) != srs_success) {
srs_warn("http: ignore unmount stream failed, sid=%s, err=%s", sid.c_str(), srs_error_desc(err).c_str());
srs_freep(err);
}

// 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());

srs_trace("http: unmount flv stream for sid=%s, i=%d", sid.c_str(), i);
}

srs_error_t SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph)
Expand Down Expand Up @@ -1214,17 +1232,6 @@ srs_error_t SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandle
}
}

SrsSharedPtr<SrsLiveSource> live_source;
if ((err = _srs_sources->fetch_or_create(r.get(), server, live_source)) != srs_success) {
return srs_error_wrap(err, "source create");
}
srs_assert(live_source.get() != NULL);

bool enabled_cache = _srs_config->get_gop_cache(r->vhost);
int gcmf = _srs_config->get_gop_cache_max_frames(r->vhost);
live_source->set_cache(enabled_cache);
live_source->set_gop_cache_max_frames(gcmf);

// create http streaming handler.
if ((err = http_mount(r.get())) != srs_success) {
return srs_error_wrap(err, "http mount");
Expand All @@ -1235,11 +1242,8 @@ srs_error_t SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandle
entry = streamHandlers[sid];
*ph = entry->stream;
}

// trigger edge to fetch from origin.
bool vhost_is_edge = _srs_config->get_vhost_is_edge(r->vhost);
srs_trace("flv: source url=%s, is_edge=%d, source_id=%s/%s",
r->get_stream_url().c_str(), vhost_is_edge, live_source->source_id().c_str(), live_source->pre_source_id().c_str());

srs_trace("flv: hijack %s ok", upath.c_str());

return err;
}
Expand Down Expand Up @@ -1281,3 +1285,64 @@ srs_error_t SrsHttpStreamServer::initialize_flv_entry(std::string vhost)
return err;
}

SrsHttpStreamDestroy::SrsHttpStreamDestroy(SrsHttpServeMux* mux, map<std::string, SrsLiveEntry*>* handlers, string sid)
{
mux_ = mux;
sid_ = sid;
streamHandlers_ = handlers;
}

SrsHttpStreamDestroy::~SrsHttpStreamDestroy()
{
}

srs_error_t SrsHttpStreamDestroy::call()
{
srs_error_t err = srs_success;

std::map<std::string, SrsLiveEntry*>::iterator it = streamHandlers_->find(sid_);
if (it == streamHandlers_->end()) {
return err;
}

// Free all HTTP resources.
SrsUniquePtr<SrsLiveEntry> entry(it->second);
srs_assert(entry->disposing);

SrsUniquePtr<SrsLiveStream> stream(entry->stream);
SrsUniquePtr<SrsBufferCache> cache(entry->cache);

// Notify cache and stream to stop.
if (stream->entry) stream->entry->enabled = false;
stream->expire();
cache->stop();

// 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);
}

if (cache->alive() || stream->alive()) {
srs_warn("http: try to free a alive stream, cache=%d, stream=%d", cache->alive(), stream->alive());
}

// Remove the entry from handlers.
streamHandlers_->erase(it);

// 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());

srs_trace("http: unmount flv stream for sid=%s, i=%d", sid_.c_str(), i);
return err;
}

string SrsHttpStreamDestroy::to_string()
{
return "destroy";
}

Loading

0 comments on commit 15fbe45

Please sign in to comment.