Merge SRS3

README.md

@@ -168,6 +168,9 @@ For previous versions, please read:
 
 ## V3 changes
 
+* v3.0, 2020-03-12, For [#1630][bug #1630], disable cache for stream changing, and drop dup header. 3.0.128
+* v3.0, 2020-03-12, For [#1594][bug #1594], detect and disable daemon for docker. 3.0.127
+* v3.0, 2020-03-12, For [#1634][bug #1634], always check status in thread loop. 3.0.126
 * v3.0, 2020-03-11, For [#1634][bug #1634], refactor output with datetime for ingest/encoder/exec. 3.0.125
 * v3.0, 2020-03-11, For [#1634][bug #1634], fix quit by accident SIGTERM while killing FFMPEG. 3.0.124
 * <strong>v3.0, 2020-03-05, [3.0 beta2(3.0.123)][r3.0b2] released. 122170 lines.</strong>
@@ -1696,6 +1699,8 @@ Winlin
 [bug #1615]: https://github.com/ossrs/srs/issues/1615
 [bug #1621]: https://github.com/ossrs/srs/issues/1621
 [bug #1634]: https://github.com/ossrs/srs/issues/1634
+[bug #1594]: https://github.com/ossrs/srs/issues/1594
+[bug #1630]: https://github.com/ossrs/srs/issues/1630
 [bug #yyyyyyyyyyyyy]: https://github.com/ossrs/srs/issues/yyyyyyyyyyyyy
 
 [bug #1631]: https://github.com/ossrs/srs/issues/1631

trunk/conf/full.conf

@@ -93,6 +93,10 @@ grace_final_wait 3200;
 # @see https://github.com/ossrs/srs/issues/1579#issuecomment-587475077
 # default: off
 force_grace_quit off;
+# Whether disable daemon for docker.
+# If on, it will set daemon to off in docker, even daemon is on.
+# default: on
+disable_daemon_for_docker on;
 
 #############################################################################################
 # heartbeat/stats sections

trunk/src/app/srs_app_config.cpp

@@ -3489,7 +3489,7 @@ srs_error_t SrsConfig::check_normal_config()
             && n != "http_server" && n != "stream_caster" && n != "srt_server"
             && n != "utc_time" && n != "work_dir" && n != "asprocess"
             && n != "ff_log_level" && n != "grace_final_wait" && n != "force_grace_quit"
-            && n != "grace_start_wait" && n != "empty_ip_ok"
+            && n != "grace_start_wait" && n != "empty_ip_ok" && n != "disable_daemon_for_docker"
             ) {
             return srs_error_new(ERROR_SYSTEM_CONFIG_INVALID, "illegal directive %s", n.c_str());
         }
@@ -4114,6 +4114,18 @@ bool SrsConfig::is_force_grace_quit()
     return SRS_CONF_PERFER_FALSE(conf->arg0());
 }
 
+bool SrsConfig::disable_daemon_for_docker()
+{
+    static bool DEFAULT = true;
+
+    SrsConfDirective* conf = root->get("disable_daemon_for_docker");
+    if (!conf || conf->arg0().empty()) {
+        return DEFAULT;
+    }
+
+    return SRS_CONF_PERFER_TRUE(conf->arg0());
+}
+
 vector<SrsConfDirective*> SrsConfig::get_stream_casters()
 {
     srs_assert(root);

trunk/src/app/srs_app_config.hpp

@@ -476,6 +476,8 @@ public:
     virtual srs_utime_t get_grace_final_wait();
     // Whether force to gracefully quit, never fast quit.
     virtual bool is_force_grace_quit();
+    // Whether disable daemon for docker.
+    virtual bool disable_daemon_for_docker();
 // stream_caster section
 public:
     // Get all stream_caster in config file.

trunk/src/app/srs_app_edge.cpp

@@ -234,15 +234,17 @@ srs_error_t SrsEdgeIngester::cycle()
     srs_error_t err = srs_success;
     
     while (true) {
+        // We always check status first.
+        // @see https://github.com/ossrs/srs/issues/1634#issuecomment-597571561
+        if ((err = trd->pull()) != srs_success) {
+            return srs_error_wrap(err, "edge ingester");
+        }
+
         if ((err = do_cycle()) != srs_success) {
             srs_warn("EdgeIngester: Ignore error, %s", srs_error_desc(err).c_str());
             srs_freep(err);
         }
-        
-        if ((err = trd->pull()) != srs_success) {
-            return srs_error_wrap(err, "edge ingester");
-        }
-        
+
         srs_usleep(SRS_EDGE_INGESTER_CIMS);
     }
     
@@ -314,7 +316,6 @@ srs_error_t SrsEdgeIngester::ingest(string& redirect)
     redirect = "";
     
     while (true) {
-        srs_error_t err = srs_success;
         if ((err = trd->pull()) != srs_success) {
             return srs_error_wrap(err, "thread quit");
         }
@@ -534,14 +535,16 @@ srs_error_t SrsEdgeForwarder::cycle()
     srs_error_t err = srs_success;
     
     while (true) {
-        if ((err = do_cycle()) != srs_success) {
-            return srs_error_wrap(err, "do cycle");
-        }
-        
+        // We always check status first.
+        // @see https://github.com/ossrs/srs/issues/1634#issuecomment-597571561
         if ((err = trd->pull()) != srs_success) {
             return srs_error_wrap(err, "thread pull");
         }
-    
+
+        if ((err = do_cycle()) != srs_success) {
+            return srs_error_wrap(err, "do cycle");
+        }
+
         srs_usleep(SRS_EDGE_FORWARDER_CIMS);
     }
     

trunk/src/app/srs_app_forward.cpp

@@ -181,15 +181,17 @@ srs_error_t SrsForwarder::cycle()
     srs_error_t err = srs_success;
     
     while (true) {
+        // We always check status first.
+        // @see https://github.com/ossrs/srs/issues/1634#issuecomment-597571561
+        if ((err = trd->pull()) != srs_success) {
+            return srs_error_wrap(err, "forwarder");
+        }
+
         if ((err = do_cycle()) != srs_success) {
             srs_warn("Forwarder: Ignore error, %s", srs_error_desc(err).c_str());
             srs_freep(err);
         }
-        
-        if ((err = trd->pull()) != srs_success) {
-            return srs_error_wrap(err, "forwarder");
-        }
-    
+
         srs_usleep(SRS_FORWARDER_CIMS);
     }
     

trunk/src/app/srs_app_http_stream.cpp

@@ -646,13 +646,13 @@ srs_error_t SrsLiveStream::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMess
     // TODO: free and erase the disabled entry after all related connections is closed.
     // TODO: FXIME: Support timeout for player, quit infinite-loop.
     while (entry->enabled) {
-        pprint->elapse();
-        
         // Whether client closed the FD.
         if ((err = trd->pull()) != srs_success) {
             return srs_error_wrap(err, "recv thread");
         }
 
+        pprint->elapse();
+
         // get messages from consumer.
         // each msg in msgs.msgs must be free, for the SrsMessageArray never free them.
         int count = 0;

trunk/src/app/srs_app_ingest.cpp

@@ -201,15 +201,17 @@ srs_error_t SrsIngester::cycle()
     srs_error_t err = srs_success;
     
     while (!disposed) {
+        // We always check status first.
+        // @see https://github.com/ossrs/srs/issues/1634#issuecomment-597571561
+        if ((err = trd->pull()) != srs_success) {
+            return srs_error_wrap(err, "ingester");
+        }
+
         if ((err = do_cycle()) != srs_success) {
             srs_warn("Ingester: Ignore error, %s", srs_error_desc(err).c_str());
             srs_freep(err);
         }
-        
-        if ((err = trd->pull()) != srs_success) {
-            return srs_error_wrap(err, "ingester");
-        }
-    
+
         srs_usleep(SRS_AUTO_INGESTER_CIMS);
     }
     

trunk/src/app/srs_app_ng_exec.cpp

@@ -83,16 +83,18 @@ srs_error_t SrsNgExec::cycle()
     srs_error_t err = srs_success;
     
     while (true) {
-        if ((err = do_cycle()) != srs_success) {
-            srs_warn("EXEC: Ignore error, %s", srs_error_desc(err).c_str());
-            srs_freep(err);
-        }
-        
+        // We always check status first.
+        // @see https://github.com/ossrs/srs/issues/1634#issuecomment-597571561
         if ((err = trd->pull()) != srs_success) {
             err = srs_error_wrap(err, "ng exec cycle");
             break;
         }
-    
+
+        if ((err = do_cycle()) != srs_success) {
+            srs_warn("EXEC: Ignore error, %s", srs_error_desc(err).c_str());
+            srs_freep(err);
+        }
+
         srs_usleep(SRS_RTMP_EXEC_CIMS);
     }
     

trunk/src/app/srs_app_rtmp_conn.cpp

@@ -381,7 +381,6 @@ srs_error_t SrsRtmpConn::service_cycle()
     }
     
     while (true) {
-        srs_error_t err = srs_success;
         if ((err = trd->pull()) != srs_success) {
             return srs_error_wrap(err, "rtmp: thread quit");
         }
@@ -704,14 +703,14 @@ srs_error_t SrsRtmpConn::do_playing(SrsSource* source, SrsConsumer* consumer, Sr
         srsu2msi(send_min_interval), srsu2msi(mw_sleep), mw_enabled, realtime, tcp_nodelay);
     
     while (true) {
-        // collect elapse for pithy print.
-        pprint->elapse();
-        
         // when source is set to expired, disconnect it.
         if ((err = trd->pull()) != srs_success) {
             return srs_error_wrap(err, "rtmp: thread quit");
         }
-        
+
+        // collect elapse for pithy print.
+        pprint->elapse();
+
         // to use isolate thread to recv, can improve about 33% performance.
         // @see: https://github.com/ossrs/srs/issues/196
         // @see: https://github.com/ossrs/srs/issues/217
@@ -872,12 +871,12 @@ srs_error_t SrsRtmpConn::do_publishing(SrsSource* source, SrsPublishRecvThread*
     int64_t nb_msgs = 0;
     uint64_t nb_frames = 0;
     while (true) {
-        pprint->elapse();
-        
         if ((err = trd->pull()) != srs_success) {
             return srs_error_wrap(err, "rtmp: thread quit");
         }
-        
+
+        pprint->elapse();
+
         // cond wait for timeout.
         if (nb_msgs == 0) {
             // when not got msgs, wait for a larger timeout.

trunk/src/app/srs_app_source.cpp

@@ -1506,6 +1506,7 @@ void SrsOriginHub::destroy_forwarders()
 SrsMetaCache::SrsMetaCache()
 {
     meta = video = audio = NULL;
+    previous_video = previous_audio = NULL;
     vformat = new SrsRtmpFormat();
     aformat = new SrsRtmpFormat();
 }
@@ -1516,6 +1517,13 @@ SrsMetaCache::~SrsMetaCache()
 }
 
 void SrsMetaCache::dispose()
+{
+    clear();
+    srs_freep(previous_video);
+    srs_freep(previous_audio);
+}
+
+void SrsMetaCache::clear()
 {
     srs_freep(meta);
     srs_freep(video);
@@ -1570,6 +1578,28 @@ srs_error_t SrsMetaCache::dumps(SrsConsumer* consumer, bool atc, SrsRtmpJitterAl
     return err;
 }
 
+SrsSharedPtrMessage* SrsMetaCache::previous_vsh()
+{
+    return previous_video;
+}
+
+SrsSharedPtrMessage* SrsMetaCache::previous_ash()
+{
+    return previous_audio;
+}
+
+void SrsMetaCache::update_previous_vsh()
+{
+    srs_freep(previous_video);
+    previous_video = video? video->copy() : NULL;
+}
+
+void SrsMetaCache::update_previous_ash()
+{
+    srs_freep(previous_audio);
+    previous_audio = audio? audio->copy() : NULL;
+}
+
 srs_error_t SrsMetaCache::update_data(SrsMessageHeader* header, SrsOnMetaDataPacket* metadata, bool& updated)
 {
     updated = false;
@@ -1636,6 +1666,7 @@ srs_error_t SrsMetaCache::update_ash(SrsSharedPtrMessage* msg)
 {
     srs_freep(audio);
     audio = msg->copy();
+    update_previous_ash();
     return aformat->on_audio(msg);
 }
 
@@ -1643,6 +1674,7 @@ srs_error_t SrsMetaCache::update_vsh(SrsSharedPtrMessage* msg)
 {
     srs_freep(video);
     video = msg->copy();
+    update_previous_vsh();
     return vformat->on_video(msg);
 }
 
@@ -2138,9 +2170,9 @@ srs_error_t SrsSource::on_audio_imp(SrsSharedPtrMessage* msg)
     
     // whether consumer should drop for the duplicated sequence header.
     bool drop_for_reduce = false;
-    if (is_sequence_header && meta->ash() && _srs_config->get_reduce_sequence_header(req->vhost)) {
-        if (meta->ash()->size == msg->size) {
-            drop_for_reduce = srs_bytes_equals(meta->ash()->payload, msg->payload, msg->size);
+    if (is_sequence_header && meta->previous_ash() && _srs_config->get_reduce_sequence_header(req->vhost)) {
+        if (meta->previous_ash()->size == msg->size) {
+            drop_for_reduce = srs_bytes_equals(meta->previous_ash()->payload, msg->payload, msg->size);
             srs_warn("drop for reduce sh audio, size=%d", msg->size);
         }
     }
@@ -2257,9 +2289,9 @@ srs_error_t SrsSource::on_video_imp(SrsSharedPtrMessage* msg)
     
     // whether consumer should drop for the duplicated sequence header.
     bool drop_for_reduce = false;
-    if (is_sequence_header && meta->vsh() && _srs_config->get_reduce_sequence_header(req->vhost)) {
-        if (meta->vsh()->size == msg->size) {
-            drop_for_reduce = srs_bytes_equals(meta->vsh()->payload, msg->payload, msg->size);
+    if (is_sequence_header && meta->previous_vsh() && _srs_config->get_reduce_sequence_header(req->vhost)) {
+        if (meta->previous_vsh()->size == msg->size) {
+            drop_for_reduce = srs_bytes_equals(meta->previous_vsh()->payload, msg->payload, msg->size);
             srs_warn("drop for reduce sh video, size=%d", msg->size);
         }
     }
@@ -2415,6 +2447,10 @@ srs_error_t SrsSource::on_publish()
     
     // reset the mix queue.
     mix_queue->clear();
+
+    // Reset the metadata cache, to make VLC happy when disable/enable stream.
+    // @see https://github.com/ossrs/srs/issues/1630#issuecomment-597979448
+    meta->clear();
     
     // detect the monotonically again.
     is_monotonically_increase = true;
@@ -2450,7 +2486,12 @@ void SrsSource::on_unpublish()
     // donot clear the sequence header, for it maybe not changed,
     // when drop dup sequence header, drop the metadata also.
     gop_cache->clear();
-    
+
+    // Reset the metadata cache, to make VLC happy when disable/enable stream.
+    // @see https://github.com/ossrs/srs/issues/1630#issuecomment-597979448
+    meta->update_previous_vsh();
+    meta->update_previous_ash();
+
     srs_trace("cleanup when unpublish");
     
     _can_publish = true;

trunk/src/app/srs_app_source.hpp

@@ -407,8 +407,10 @@ private:
     SrsSharedPtrMessage* meta;
     // The cached video sequence header, for example, sps/pps for h.264.
     SrsSharedPtrMessage* video;
+    SrsSharedPtrMessage* previous_video;
     // The cached audio sequence header, for example, asc for aac.
     SrsSharedPtrMessage* audio;
+    SrsSharedPtrMessage* previous_audio;
     // The format for sequence header.
     SrsRtmpFormat* vformat;
     SrsRtmpFormat* aformat;
@@ -418,6 +420,8 @@ public:
 public:
     // Dispose the metadata cache.
     virtual void dispose();
+    // For each publishing, clear the metadata cache.
+    virtual void clear();
 public:
     // Get the cached metadata.
     virtual SrsSharedPtrMessage* data();
@@ -431,6 +435,13 @@ public:
     // @param dm Whether dumps the metadata.
     // @param ds Whether dumps the sequence header.
     virtual srs_error_t dumps(SrsConsumer* consumer, bool atc, SrsRtmpJitterAlgorithm ag, bool dm, bool ds);
+public:
+    // Previous exists sequence header.
+    virtual SrsSharedPtrMessage* previous_vsh();
+    virtual SrsSharedPtrMessage* previous_ash();
+    // Update previous sequence header, drop old one, set to new sequence header.
+    virtual void update_previous_vsh();
+    virtual void update_previous_ash();
 public:
     // Update the cached metadata by packet.
     virtual srs_error_t update_data(SrsMessageHeader* header, SrsOnMetaDataPacket* metadata, bool& updated);

trunk/src/core/srs_core_version3.hpp

@@ -24,6 +24,6 @@
 #ifndef SRS_CORE_VERSION3_HPP
 #define SRS_CORE_VERSION3_HPP
 
-#define SRS_VERSION3_REVISION 125
+#define SRS_VERSION3_REVISION 128
 
 #endif

trunk/src/main/srs_main_server.cpp

@@ -48,6 +48,7 @@ using namespace std;
 #include <srs_core_performance.hpp>
 #include <srs_app_utility.hpp>
 #include <srs_core_autofree.hpp>
+#include <srs_kernel_file.hpp>
 #include <srs_app_hybrid.hpp>
 
 #ifdef SRS_AUTO_SRT
@@ -354,12 +355,56 @@ string srs_getenv(const char* name)
     return "";
 }
 
+// Detect docker by https://stackoverflow.com/a/41559867
+srs_error_t srs_detect_docker(bool* is_docker)
+{
+    srs_error_t err = srs_success;
+
+    *is_docker = false;
+
+    SrsFileReader fr;
+    if ((err = fr.open("/proc/1/cgroup")) != srs_success) {
+        return err;
+    }
+
+    ssize_t nn;
+    char buf[1024];
+    if ((err = fr.read(buf, sizeof(buf), &nn)) != srs_success) {
+        return err;
+    }
+
+    if (nn <= 0) {
+        return err;
+    }
+
+    string s(buf, nn);
+    if (srs_string_contains(s, "/docker")) {
+        *is_docker = true;
+    }
+
+    return err;
+}
+
 srs_error_t run_directly_or_daemon()
 {
     srs_error_t err = srs_success;
+
+    // Load daemon from config, disable it for docker.
+    // @see https://github.com/ossrs/srs/issues/1594
+    bool in_daemon = _srs_config->get_daemon();
+    if (in_daemon && _srs_config->disable_daemon_for_docker()) {
+        bool is_docker = false;
+        err = srs_detect_docker(&is_docker);
+        srs_error_reset(err); // Ignore any error while detecting docker.
+
+        if (is_docker) {
+            srs_warn("disable daemon for docker");
+            in_daemon = false;
+        }
+    }
     
     // If not daemon, directly run master.
-    if (!_srs_config->get_daemon()) {
+    if (!in_daemon) {
         if ((err = run_hybrid_server()) != srs_success) {
             return srs_error_wrap(err, "run master");
         }