去除rtp_sender_mtx (#1722)

* 去除rtp_sender_mtx
* 无人观看定时器指定为MediaSource对象所属poller以确保读取观看数是线程安全的

Co-authored-by: xiongziliang <771730766@qq.com>

src/Common/MediaSource.cpp

@@ -233,10 +233,17 @@ toolkit::EventPoller::Ptr MediaSource::getOwnerPoller() {
 }
 
 void MediaSource::onReaderChanged(int size) {
-    auto listener = _listener.lock();
+    weak_ptr<MediaSource> weak_self = shared_from_this();
-    if (listener) {
+    getOwnerPoller()->async([weak_self, size]() {
-        listener->onReaderChanged(*this, size);
+        auto strong_self = weak_self.lock();
+        if (!strong_self) {
+            return;
         }
+        auto listener = strong_self->_listener.lock();
+        if (listener) {
+            listener->onReaderChanged(*strong_self, size);
+        }
+    });
 }
 
 bool MediaSource::setupRecord(Recorder::type type, bool start, const string &custom_path, size_t max_second){

src/Common/MultiMediaSourceMuxer.cpp

@@ -230,7 +230,6 @@ void MultiMediaSourceMuxer::startSendRtp(MediaSource &, const MediaSourceEvent::
             rtp_sender->addTrack(track);
         }
         rtp_sender->addTrackCompleted();
-        lock_guard<mutex> lck(strong_self->_rtp_sender_mtx);
         strong_self->_rtp_sender[args.ssrc] = rtp_sender;
     });
 #else
@@ -249,13 +248,11 @@ bool MultiMediaSourceMuxer::stopSendRtp(MediaSource &sender, const string &ssrc)
     }
     if (ssrc.empty()) {
         //关闭全部
-        lock_guard<mutex> lck(_rtp_sender_mtx);
         auto size = _rtp_sender.size();
         _rtp_sender.clear();
         return size;
     }
     //关闭特定的
-    lock_guard<mutex> lck(_rtp_sender_mtx);
     return _rtp_sender.erase(ssrc);
 #else
     return false;
@@ -340,7 +337,6 @@ void MultiMediaSourceMuxer::resetTracks() {
 #endif
 
 #if defined(ENABLE_RTPPROXY)
-    lock_guard<mutex> lck(_rtp_sender_mtx);
     for (auto &pr : _rtp_sender) {
         pr.second->resetTracks();
     }
@@ -447,7 +443,6 @@ bool MultiMediaSourceMuxer::onTrackFrame(const Frame::Ptr &frame_in) {
 #endif
 
 #if defined(ENABLE_RTPPROXY)
-    lock_guard<mutex> lck(_rtp_sender_mtx);
     for (auto &pr : _rtp_sender) {
         ret = pr.second->inputFrame(frame) ? true : ret;
     }

src/Common/MultiMediaSourceMuxer.h

@@ -180,7 +180,6 @@ private:
     std::weak_ptr<Listener> _track_listener;
     std::function<std::string()> _get_origin_url;
 #if defined(ENABLE_RTPPROXY)
-    std::mutex _rtp_sender_mtx;
     std::unordered_map<std::string, RtpSender::Ptr> _rtp_sender;
 #endif //ENABLE_RTPPROXY
 

src/Rtp/RtpProcess.cpp

@@ -277,5 +277,9 @@ std::shared_ptr<SockInfo> RtpProcess::getOriginSock(MediaSource &sender) const {
     return const_cast<RtpProcess *>(this)->shared_from_this();
 }
 
+toolkit::EventPoller::Ptr RtpProcess::getOwnerPoller(MediaSource &sender) {
+    return _sock ? _sock->getPoller() : nullptr;
+}
+
 }//namespace mediakit
 #endif//defined(ENABLE_RTPPROXY)

src/Rtp/RtpProcess.h

@@ -76,6 +76,7 @@ protected:
     MediaOriginType getOriginType(MediaSource &sender) const override;
     std::string getOriginUrl(MediaSource &sender) const override;
     std::shared_ptr<SockInfo> getOriginSock(MediaSource &sender) const override;
+    toolkit::EventPoller::Ptr getOwnerPoller(MediaSource &sender) override;
 
 private:
     void emitOnPublish();