优化getSnap截图接口性能

server/FFmpegSource.cpp

@@ -363,42 +363,88 @@ void FFmpegSource::onGetMediaSource(const MediaSource::Ptr &src) {
 #include "Player/MediaPlayer.h"
 #include "Codec/Transcode.h"
 
-static void makeSnapAsync(const string &play_url, const string &save_path, float timeout_sec, const FFmpegSnap::onSnap &cb) {
+static std::string getStreamName(const std::string &input) {
-    struct Holder {
+    // 先找 '/'
-        MediaPlayer::Ptr player;
+    size_t pos = input.find('/');
-    };
+    if (pos != std::string::npos) {
-    auto holder = std::make_shared<Holder>();
+        return input.substr(0, pos);
-    auto player = std::make_shared<MediaPlayer>();
-    (*player)[mediakit::Client::kTimeoutMS] = timeout_sec * 1000;
-
-    player->setOnPlayResult([holder, save_path, cb, timeout_sec](const SockException &ex) mutable {
-        onceToken token(nullptr, [&]() { holder->player = nullptr; });
-        auto video = ex ? nullptr : dynamic_pointer_cast<VideoTrack>(holder->player->getTrack(TrackVideo, false));
-        if (!video) {
-            cb(false, ex ? ex.what() : "none video track");
-            return;
     }
-        auto decoder = std::make_shared<FFmpegDecoder>(video);
+
-        auto new_holder = std::make_shared<Holder>(*holder);
+    // 如果没有 '/', 再找 '.'
-        auto timer = EventPollerPool::Instance().getPoller()->doDelayTask(1000 * timeout_sec, [cb, new_holder]() {
+    pos = input.find('.');
+    if (pos != std::string::npos) {
+        return input.substr(0, pos);
+    }
+
+    // 都没有，返回原字符串
+    return input;
+}
+
+static bool makeSnapAsync(const string &play_url, const string &save_path, float timeout_sec, const FFmpegSnap::onSnap &cb) {
+    MediaInfo info(play_url);
+    info.stream = getStreamName(info.stream);
+    MediaSource::Ptr src;
+    Track::Ptr track;
+    MultiMediaSourceMuxer::Ptr muxer;
+    if (info.host == "127.0.0.1" || info.host == "localhost" || info.params.find("self=1") != std::string::npos) {
+        // 截图zlm本身
+        src = MediaSource::find(info.vhost, info.app, info.stream, false);
+        track = src ? src->getTrack(TrackVideo) : nullptr;
+        muxer = src ? src->getMuxer() : nullptr;
+    }
+    // 要求开启转码额度
+    if (src && track && muxer) {
+        DebugL << "start snap in fast mode: " << play_url;
+        src->getOwnerPoller()->async([muxer, track, timeout_sec, cb, save_path]() mutable {
+            auto done = std::make_shared<atomic<bool>>(false);
+            auto reader = muxer->getFrameReader();
+            auto timer = EventPollerPool::Instance().getPoller()->doDelayTask(1000 * timeout_sec, [reader, cb, done]() {
+                if (*done) {
+                    return 0;
+                }
+                *done = true;
                 // 防止解码失败导致播放器无法释放
-            new_holder->player = nullptr;
                 cb(false, "decode frame timeout");
                 return 0;
             });
-        auto done = false;
+
-        decoder->setOnDecode([save_path, new_holder, cb, done, timer](const FFmpegFrame::Ptr &frame) mutable {
+            // 单线程解码，没必要太快
-            if (done) {
+            auto decoder = std::make_shared<FFmpegDecoder>(track, 1);
+            decoder->setOnDecode([save_path, cb, done, timer](const FFmpegFrame::Ptr &frame) mutable {
+                if (*done) {
                     return;
                 }
-            onceToken token(nullptr, [&]() { new_holder->player = nullptr; timer->cancel(); done = true; });
+                *done = true;
+                timer->cancel();
+
                 auto ret = FFmpegUtils::saveFrame(frame, save_path.data());
                 cb(std::get<0>(ret), std::get<1>(ret));
             });
-        video->addDelegate([decoder](const Frame::Ptr &frame) { return decoder->inputFrame(frame, false, true); });
+
+            auto decode_poller = WorkThreadPool::Instance().getPoller();
+            reader->setReadCB([decoder, decode_poller, done](const Frame::Ptr &frame) mutable {
+                if (*done) {
+                    // 销毁解码器，防止无谓的解码
+                    decoder = nullptr;
+                    return;
+                }
+                if (frame->getTrackType() == TrackVideo) {
+                    // 异步解码，防止阻塞网络线程
+                    auto copy_frame = Frame::getCacheAbleFrame(frame);
+                    decode_poller->async([decoder, copy_frame, done]() mutable {
+                        if (*done) {
+                            // 销毁解码器，防止无谓的解码
+                            decoder = nullptr;
+                            return;
+                        }
+                        decoder->inputFrame(copy_frame, false, false, true);
                     });
-    player->play(play_url);
+                }
-    holder->player = std::move(player);
+            });
+        });
+        return true;
+    }
+    return false;
 }
 
 #endif
@@ -406,9 +452,10 @@ static void makeSnapAsync(const string &play_url, const string &save_path, float
 void FFmpegSnap::makeSnap(bool async, const string &play_url, const string &save_path, float timeout_sec, const onSnap &cb) {
 #if defined(ENABLE_FFMPEG)
     if (async) {
-        makeSnapAsync(play_url, save_path, timeout_sec, cb);
+        if (makeSnapAsync(play_url, save_path, timeout_sec, cb)) {
             return;
         }
+    }
 #endif
 
     GET_CONFIG(string, ffmpeg_bin, FFmpeg::kBin);

src/Common/MultiMediaSourceMuxer.cpp

@@ -564,7 +564,7 @@ bool MultiMediaSourceMuxer::isRecording(Recorder::type type) {
 
 void MultiMediaSourceMuxer::startSendRtp(const MediaSourceEvent::SendRtpArgs &args, const std::function<void(uint16_t, const toolkit::SockException &)> cb) {
 #if defined(ENABLE_RTPPROXY)
-    createGopCacheIfNeed(1);
+    createGopCacheIfNeed();
 
     auto ring = _ring;
     auto ssrc = args.ssrc;
@@ -636,6 +636,13 @@ bool MultiMediaSourceMuxer::stopSendRtp(const string &ssrc) {
 #endif//ENABLE_RTPPROXY
 }
 
+MultiMediaSourceMuxer::RingType::RingReader::Ptr MultiMediaSourceMuxer::getFrameReader() {
+    auto poller = getOwnerPoller(MediaSource::NullMediaSource());
+    CHECK(poller->isCurrentThread());
+    createGopCacheIfNeed();
+    return _ring->attach(poller);
+}
+
 EventPoller::Ptr MultiMediaSourceMuxer::getOwnerPoller(MediaSource &sender) {
     auto listener = getDelegate();
     if (!listener) {
@@ -754,13 +761,7 @@ void MultiMediaSourceMuxer::onAllTrackReady() {
         listener->onAllTrackReady();
     }
 
-#if defined(ENABLE_RTPPROXY)
+    createGopCacheIfNeed();
-    GET_CONFIG(size_t, gop_cache, RtpProxy::kGopCache);
-    if (gop_cache > 0) {
-        createGopCacheIfNeed(gop_cache);
-    }
-#endif
-
     Stamp *first = nullptr;
     for (auto &pr : _stamps) {
         if (!first) {
@@ -775,10 +776,11 @@ void MultiMediaSourceMuxer::onAllTrackReady() {
     InfoL << "stream: " << shortUrl() << " , codec info: " << getTrackInfoStr(this);
 }
 
-void MultiMediaSourceMuxer::createGopCacheIfNeed(size_t gop_count) {
+void MultiMediaSourceMuxer::createGopCacheIfNeed() {
     if (_ring) {
         return;
     }
+    GET_CONFIG(size_t, gop_cache, RtpProxy::kGopCache);
     weak_ptr<MultiMediaSourceMuxer> weak_self = shared_from_this();
     auto src = std::make_shared<MediaSourceForMuxer>(weak_self.lock());
     _ring = std::make_shared<RingType>(1024, [weak_self, src](int size) {
@@ -789,7 +791,7 @@ void MultiMediaSourceMuxer::createGopCacheIfNeed(size_t gop_count) {
                 strong_self->onReaderChanged(*src, strong_self->totalReaderCount());
             });
         }
-    }, gop_count);
+    }, std::max<size_t>(gop_cache, 1));
 }
 
 void MultiMediaSourceMuxer::resetTracks() {

src/Common/MultiMediaSourceMuxer.h

@@ -194,6 +194,9 @@ public:
      */
     std::shared_ptr<MultiMediaSourceMuxer> getMuxer(MediaSource &sender) const override;
 
+    // 获取frame ring reader
+    RingType::RingReader::Ptr getFrameReader();
+
     const ProtocolOption &getOption() const;
     const MediaTuple &getMediaTuple() const;
     std::string shortUrl() const;
@@ -238,7 +241,7 @@ protected:
     bool onTrackFrame_l(const Frame::Ptr &frame);
 
 private:
-    void createGopCacheIfNeed(size_t gop_count);
+    void createGopCacheIfNeed();
     std::shared_ptr<MediaSinkInterface> makeRecorder(Recorder::type type);
 
 private: