Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

FLV: Refine source and http handler. v6.0.155 v7.0.14 #4165

Merged
merged 8 commits into from
Sep 1, 2024
Merged

FLV: Refine source and http handler. v6.0.155 v7.0.14 #4165

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
e094cf8
Live: Unmount flv when disposing.
winlinvip Aug 31, 2024
c677a7e
Live: Fail when flv stream is disposing.
winlinvip Aug 31, 2024
ba1f89a
Refine error code.
winlinvip Aug 31, 2024
76acb86
Refine http stream cache.
winlinvip Aug 31, 2024
a151b53
Use async worker to free the stream.
winlinvip Aug 31, 2024
8c3d701
Refine code.
winlinvip Aug 31, 2024
943d540
Refine code.
winlinvip Aug 31, 2024
50ca250
Update release to v6.0.155 v7.0.14
winlinvip Sep 1, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions trunk/doc/CHANGELOG.md
View file
Open in desktop
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
View file
Open in desktop
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
Loading