提高线程安全性

src/Common/MediaSource.cpp

@@ -70,7 +70,6 @@ MediaSource::MediaSource(const string &schema, const string &vhost, const string
     _app = app;
     _stream_id = stream_id;
     _create_stamp = time(NULL);
-    _default_poller = EventPollerPool::Instance().getPoller();
 }
 
 MediaSource::~MediaSource() {
@@ -233,22 +232,23 @@ toolkit::EventPoller::Ptr MediaSource::getOwnerPoller() {
     toolkit::EventPoller::Ptr ret;
     auto listener = _listener.lock();
     if (listener) {
-        ret = listener->getOwnerPoller(*this);
+        return listener->getOwnerPoller(*this);
     }
-    return ret ? ret : _default_poller;
+    throw std::runtime_error(toolkit::demangle(typeid(*this).name()) + "::getOwnerPoller failed:" + getUrl());
 }
 
 void MediaSource::onReaderChanged(int size) {
     weak_ptr<MediaSource> weak_self = shared_from_this();
-    getOwnerPoller()->async([weak_self, size]() {
+    auto listener = _listener.lock();
+    if (!listener) {
+        return;
+    }
+    getOwnerPoller()->async([weak_self, size, listener]() {
         auto strong_self = weak_self.lock();
         if (!strong_self) {
             return;
         }
-        auto listener = strong_self->_listener.lock();
-        if (listener) {
         listener->onReaderChanged(*strong_self, size);
-        }
     });
 }
 
@@ -729,7 +729,7 @@ toolkit::EventPoller::Ptr MediaSourceEventInterceptor::getOwnerPoller(MediaSourc
     if (listener) {
         return listener->getOwnerPoller(sender);
     }
-    return EventPollerPool::Instance().getPoller();
+    throw std::runtime_error(toolkit::demangle(typeid(*this).name()) + "::getOwnerPoller failed");
 }
 
 bool MediaSourceEventInterceptor::setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path, size_t max_second) {

src/Common/MediaSource.h

@@ -137,8 +137,8 @@ private:
 //该对象用于拦截感兴趣的MediaSourceEvent事件
 class MediaSourceEventInterceptor : public MediaSourceEvent {
 public:
-    MediaSourceEventInterceptor(){}
+    MediaSourceEventInterceptor() = default;
-    ~MediaSourceEventInterceptor() override {}
+    ~MediaSourceEventInterceptor() override = default;
 
     void setDelegate(const std::weak_ptr<MediaSourceEvent> &listener);
     std::shared_ptr<MediaSourceEvent> getDelegate() const;
@@ -171,21 +171,18 @@ private:
  */
 class MediaInfo {
 public:
-    ~MediaInfo() {}
+    ~MediaInfo() = default;
-    MediaInfo() {}
+    MediaInfo() = default;
     MediaInfo(const std::string &url) { parse(url); }
     void parse(const std::string &url);
-    std::string shortUrl() const {
+    std::string shortUrl() const { return _vhost + "/" + _app + "/" + _streamid; }
-        return _vhost + "/" + _app + "/" + _streamid;
+    std::string getUrl() const { return _schema + "://" + shortUrl(); }
-    }
+
-    std::string getUrl() const {
-        return _schema + "://" + shortUrl();
-    }
 public:
+    uint16_t _port = 0;
     std::string _full_url;
     std::string _schema;
     std::string _host;
-    uint16_t _port = 0;
     std::string _vhost;
     std::string _app;
     std::string _streamid;
@@ -214,12 +211,9 @@ public:
     // 流id
     const std::string& getId() const;
 
-    std::string shortUrl() const {
+    std::string shortUrl() const { return _vhost + "/" + _app + "/" + _stream_id; }
-        return  _vhost + "/" + _app + "/" + _stream_id;
+
-    }
+    std::string getUrl() const { return _schema + "://" + shortUrl(); }
-    std::string getUrl() const {
-        return _schema + "://" + shortUrl();
-    }
 
     //获取对象所有权
     std::shared_ptr<void> getOwnership();
@@ -327,7 +321,6 @@ private:
     std::string _app;
     std::string _stream_id;
     std::weak_ptr<MediaSourceEvent> _listener;
-    toolkit::EventPoller::Ptr _default_poller;
     // 对象个数统计
     toolkit::ObjectStatistic<MediaSource> _statistic;
 };
@@ -352,9 +345,7 @@ private:
 template<typename packet, typename policy = FlushPolicy, typename packet_list = toolkit::List<std::shared_ptr<packet> > >
 class PacketCache {
 public:
-    PacketCache(){
+    PacketCache() { _cache = std::make_shared<packet_list>(); }
-        _cache = std::make_shared<packet_list>();
-    }
 
     virtual ~PacketCache() = default;
 

src/Common/MultiMediaSourceMuxer.cpp

@@ -91,6 +91,7 @@ const std::string &MultiMediaSourceMuxer::getStreamId() const {
 
 MultiMediaSourceMuxer::MultiMediaSourceMuxer(const string &vhost, const string &app, const string &stream, float dur_sec, const ProtocolOption &option) {
     _poller = EventPollerPool::Instance().getPoller();
+    _create_in_poller = _poller->isCurrentThread();
     _vhost = vhost;
     _app = app;
     _stream_id = stream;
@@ -310,7 +311,12 @@ EventPoller::Ptr MultiMediaSourceMuxer::getOwnerPoller(MediaSource &sender) {
         return _poller;
     }
     try {
-        return listener->getOwnerPoller(sender);
+        auto ret = listener->getOwnerPoller(sender);
+        if (ret != _poller) {
+            WarnL << "OwnerPoller changed:" << _get_origin_url();
+            _poller = ret;
+        }
+        return ret;
     } catch (MediaSourceEvent::NotImplemented &) {
         // listener未重载getOwnerPoller
         return _poller;
@@ -348,6 +354,7 @@ bool MultiMediaSourceMuxer::onTrackReady(const Track::Ptr &track) {
 }
 
 void MultiMediaSourceMuxer::onAllTrackReady() {
+    CHECK(!_create_in_poller || getOwnerPoller(MediaSource::NullMediaSource())->isCurrentThread());
     setMediaListener(getDelegate());
 
     if (_rtmp) {

src/Common/MultiMediaSourceMuxer.h

@@ -217,6 +217,7 @@ protected:
 
 private:
     bool _is_enable = false;
+    bool _create_in_poller = false;
     std::string _vhost;
     std::string _app;
     std::string _stream_id;

src/Rtmp/RtmpSession.cpp

@@ -592,6 +592,10 @@ std::shared_ptr<SockInfo> RtmpSession::getOriginSock(MediaSource &sender) const
     return const_cast<RtmpSession *>(this)->shared_from_this();
 }
 
+toolkit::EventPoller::Ptr RtmpSession::getOwnerPoller(MediaSource &sender) {
+    return getPoller();
+}
+
 void RtmpSession::setSocketFlags(){
     GET_CONFIG(int, merge_write_ms, General::kMergeWriteMS);
     if (merge_write_ms > 0) {

src/Rtmp/RtmpSession.h

@@ -80,6 +80,8 @@ private:
     std::string getOriginUrl(MediaSource &sender) const override;
     // 获取媒体源客户端相关信息
     std::shared_ptr<SockInfo> getOriginSock(MediaSource &sender) const override;
+    // 由于支持断连续推，存在OwnerPoller变更的可能
+    toolkit::EventPoller::Ptr getOwnerPoller(MediaSource &sender) override;
 
     void setSocketFlags();
     std::string getStreamId(const std::string &str);

src/Rtp/RtpProcess.cpp

@@ -66,12 +66,15 @@ RtpProcess::~RtpProcess() {
 }
 
 bool RtpProcess::inputRtp(bool is_udp, const Socket::Ptr &sock, const char *data, size_t len, const struct sockaddr *addr, uint64_t *dts_out) {
-    if (!_sock) {
+    if (_sock != sock) {
         // 第一次运行本函数
+        bool first = !_sock;
         _sock = sock;
         _addr.reset(new sockaddr_storage(*((sockaddr_storage *)addr)));
+        if (first) {
             emitOnPublish();
         }
+    }
 
     _total_bytes += len;
     if (_save_file_rtp) {
@@ -228,7 +231,7 @@ void RtpProcess::emitOnPublish() {
         if (!strong_self) {
             return;
         }
-        auto poller = strong_self->_sock ? strong_self->_sock->getPoller() : EventPollerPool::Instance().getPoller();
+        auto poller = strong_self->getOwnerPoller(MediaSource::NullMediaSource());
         poller->async([weak_self, err, option]() {
             auto strong_self = weak_self.lock();
             if (!strong_self) {
@@ -269,7 +272,10 @@ std::shared_ptr<SockInfo> RtpProcess::getOriginSock(MediaSource &sender) const {
 }
 
 toolkit::EventPoller::Ptr RtpProcess::getOwnerPoller(MediaSource &sender) {
-    return _sock ? _sock->getPoller() : EventPollerPool::Instance().getPoller();
+    if (_sock) {
+        return _sock->getPoller();
+    }
+    throw std::runtime_error("RtpProcess::getOwnerPoller failed:" + _media_info._streamid);
 }
 
 float RtpProcess::getLossRate(MediaSource &sender, TrackType type) {

src/Rtsp/RtspSession.cpp

@@ -1159,6 +1159,10 @@ std::shared_ptr<SockInfo> RtspSession::getOriginSock(MediaSource &sender) const
     return const_cast<RtspSession *>(this)->shared_from_this();
 }
 
+toolkit::EventPoller::Ptr RtspSession::getOwnerPoller(MediaSource &sender) {
+    return getPoller();
+}
+
 void RtspSession::onBeforeRtpSorted(const RtpPacket::Ptr &rtp, int track_index){
     updateRtcpContext(rtp);
 }

src/Rtsp/RtspSession.h

@@ -91,6 +91,8 @@ protected:
     std::string getOriginUrl(MediaSource &sender) const override;
     // 获取媒体源客户端相关信息
     std::shared_ptr<SockInfo> getOriginSock(MediaSource &sender) const override;
+    // 由于支持断连续推，存在OwnerPoller变更的可能
+    toolkit::EventPoller::Ptr getOwnerPoller(MediaSource &sender) override;
 
     /////TcpSession override////
     ssize_t send(toolkit::Buffer::Ptr pkt) override;

webrtc/WebRtcPusher.cpp

@@ -74,6 +74,10 @@ std::shared_ptr<SockInfo> WebRtcPusher::getOriginSock(MediaSource &sender) const
     return static_pointer_cast<SockInfo>(getSession());
 }
 
+toolkit::EventPoller::Ptr WebRtcPusher::getOwnerPoller(MediaSource &sender) {
+    return getPoller();
+}
+
 void WebRtcPusher::onRecvRtp(MediaTrack &track, const string &rid, RtpPacket::Ptr rtp) {
     if (!_simulcast) {
         assert(_push_src);

webrtc/WebRtcPusher.h

@@ -44,6 +44,8 @@ protected:
     std::string getOriginUrl(MediaSource &sender) const override;
     // 获取媒体源客户端相关信息
     std::shared_ptr<SockInfo> getOriginSock(MediaSource &sender) const override;
+    // 由于支持断连续推，存在OwnerPoller变更的可能
+    toolkit::EventPoller::Ptr getOwnerPoller(MediaSource &sender) override;
     // 获取丢包率
     float getLossRate(MediaSource &sender,TrackType type) override;