fix #257, support 0.1s+ latency. 2.0.70

README.md

@@ -453,6 +453,7 @@ Supported operating systems and hardware:
 [CN](https://github.com/winlinvip/simple-rtmp-server/wiki/v2_CN_SrsLibrtmp#publish-audio-raw-stream),
 [EN](https://github.com/winlinvip/simple-rtmp-server/wiki/v2_EN_SrsLibrtmp#publish-audio-raw-stream)
 ) by srs-librtmp.
+1. Support 0.1s+ latency, read [#257](https://github.com/winlinvip/simple-rtmp-server/issues/257).
 1. [no-plan] Support <500ms latency, FRSC(Fast RTMP-compatible Stream Channel tech).
 1. [no-plan] Support RTMP 302 redirect [#92](https://github.com/winlinvip/simple-rtmp-server/issues/92).
 1. [no-plan] Support multiple processes, for both origin and edge
@@ -486,6 +487,7 @@ Supported operating systems and hardware:
 * 2013-10-17, Created.<br/>
 
 ## History
+* v2.0, 2014-12-12, fix [#257](https://github.com/winlinvip/simple-rtmp-server/issues/257), support 0.1s+ latency. 2.0.70
 * v2.0, 2014-12-08, update wiki for mr([EN](https://github.com/winlinvip/simple-rtmp-server/wiki/v2_EN_LowLatency#merged-read), [CN](https://github.com/winlinvip/simple-rtmp-server/wiki/v2_CN_LowLatency#merged-read)) and mw([EN](https://github.com/winlinvip/simple-rtmp-server/wiki/v2_EN_LowLatency#merged-write), [CN](https://github.com/winlinvip/simple-rtmp-server/wiki/v2_CN_LowLatency#merged-write)).
 * v2.0, 2014-12-07, fix [#251](https://github.com/winlinvip/simple-rtmp-server/issues/251), 10k+ clients, use queue cond wait and fast vector. 2.0.67
 * v2.0, 2014-12-05, fix [#251](https://github.com/winlinvip/simple-rtmp-server/issues/251), 9k+ clients, use fast cache for msgs queue. 2.0.57

trunk/conf/full.conf

@@ -145,6 +145,13 @@ vhost __defaultVhost__ {
 # the MR(merged-read) setting for publisher.
 # the MW(merged-write) settings for player.
 vhost mrw.srs.com {
+    # whether enable min delay mode for vhost.
+    # for min latence mode:
+    # 1. disable the mr for vhost.
+    # 2. use timeout for cond wait for consumer queue.
+    # @see https://github.com/winlinvip/simple-rtmp-server/issues/257
+    # default: on
+    min_latency     off;
     # about MR, read https://github.com/winlinvip/simple-rtmp-server/issues/241
     mr {
         # whether enable the MR(merged-read)
@@ -440,6 +447,12 @@ vhost debug.srs.com {
 
 # the vhost for min delay, donot cache any stream.
 vhost min.delay.com {
+    # @see vhost mrw.srs.com for detail.
+    min_latency     on;
+    mr {
+        enabled     off;
+    }
+    mw_latency      100;
     # whether cache the last gop.
     # if on, cache the last gop and dispatch to client,
     #   to enabled fast startup for client, client play immediately.

trunk/conf/realtime.conf

@@ -1,12 +1,13 @@
 # the config for srs to delivery realtime RTMP stream
-# @see https://github.com/winlinvip/simple-rtmp-server/wiki/v1_CN_SampleRealtime
+# @see https://github.com/winlinvip/simple-rtmp-server/wiki/v2_CN_SampleRealtime
 # @see full.conf for detail config.
 
 listen              1935;
 max_connections     1000;
 vhost __defaultVhost__ {
     gop_cache       off;
     queue_length    10;
+    min_latency     on;
     mr {
         enabled     off;
     }

trunk/research/players/js/srs.player.js

@@ -22,7 +22,7 @@ function SrsPlayer(container, width, height, private_object) {
     this.height = height;
     this.id = SrsPlayer.__id++;
     this.stream_url = null;
-    this.buffer_time = 0.8; // default to 0.8
+    this.buffer_time = 0.3; // default to 0.3
     this.volume = 1.0; // default to 100%
     this.callbackObj = null;
 

trunk/research/players/srs_player.html

@@ -96,7 +96,7 @@
 
                 srs_player = new SrsPlayer("player_id", srs_get_player_width(), srs_get_player_height());
                 srs_player.on_player_ready = function() {
-                    select_buffer_time("#btn_bt_0_8", 0.8);
+                    select_buffer_time("#btn_bt_0_1", 0.1);
                     this.play(url);
                 };
                 srs_player.on_player_metadata = function(metadata) {
@@ -231,6 +231,15 @@
             }
 
             if (true) {
+                $("#btn_bt_0_1").click(function(){
+                    select_buffer_time("#btn_bt_0_1", 0.1);
+                });
+                $("#btn_bt_0_2").click(function(){
+                    select_buffer_time("#btn_bt_0_2", 0.2);
+                });
+                $("#btn_bt_0_3").click(function(){
+                    select_buffer_time("#btn_bt_0_3", 0.3);
+                });
                 $("#btn_bt_0_5").click(function(){
                     select_buffer_time("#btn_bt_0_5", 0.5);
                 });
@@ -504,6 +513,9 @@ <h3><a href="https://github.com/winlinvip/simple-rtmp-server">SrsPlayer</a></h3>
             <div class="btn-group dropup">
                 <button class="btn dropdown-toggle" data-toggle="dropdown">缓冲区<span class="caret"></span></button>
                 <ul class="dropdown-menu">
+                    <li><a id="btn_bt_0_1" href="#">0.1秒(实时)</a></li>
+                    <li><a id="btn_bt_0_2" href="#">0.2秒(实时)</a></li>
+                    <li><a id="btn_bt_0_3" href="#">0.3秒(实时)</a></li>
                     <li><a id="btn_bt_0_5" href="#">0.5秒(实时)</a></li>
                     <li><a id="btn_bt_0_8" href="#">0.8秒(会议)</a></li>
                     <li><a id="btn_bt_1" href="#">1秒(低延迟)</a></li>

trunk/src/app/srs_app_config.cpp

@@ -852,6 +852,17 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload mw success.", vhost.c_str());
             }
+            // min_latency, only one per vhost
+            if (!srs_directive_equals(new_vhost->get("min_latency"), old_vhost->get("min_latency"))) {
+                for (it = subscribes.begin(); it != subscribes.end(); ++it) {
+                    ISrsReloadHandler* subscribe = *it;
+                    if ((ret = subscribe->on_reload_vhost_realtime(vhost)) != ERROR_SUCCESS) {
+                        srs_error("vhost %s notify subscribes min_latency failed. ret=%d", vhost.c_str(), ret);
+                        return ret;
+                    }
+                }
+                srs_trace("vhost %s reload min_latency success.", vhost.c_str());
+            }
             // http, only one per vhost.
             if (!srs_directive_equals(new_vhost->get("http"), old_vhost->get("http"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -1350,7 +1361,7 @@ int SrsConfig::check_config()
                 && n != "time_jitter" 
                 && n != "atc" && n != "atc_auto"
                 && n != "debug_srs_upnode"
-                && n != "mr" && n != "mw_latency"
+                && n != "mr" && n != "mw_latency" && n != "min_latency"
             ) {
                 ret = ERROR_SYSTEM_CONFIG_INVALID;
                 srs_error("unsupported vhost directive %s, ret=%d", n.c_str(), ret);
@@ -2125,7 +2136,6 @@ int SrsConfig::get_chunk_size(string vhost)
 
 bool SrsConfig::get_mr_enabled(string vhost)
 {
-
     SrsConfDirective* conf = get_vhost(vhost);
 
     if (!conf) {
@@ -2147,7 +2157,6 @@ bool SrsConfig::get_mr_enabled(string vhost)
 
 int SrsConfig::get_mr_sleep_ms(string vhost)
 {
-
     SrsConfDirective* conf = get_vhost(vhost);
 
     if (!conf) {
@@ -2169,7 +2178,6 @@ int SrsConfig::get_mr_sleep_ms(string vhost)
 
 int SrsConfig::get_mw_sleep_ms(string vhost)
 {
-
     SrsConfDirective* conf = get_vhost(vhost);
 
     if (!conf) {
@@ -2184,6 +2192,22 @@ int SrsConfig::get_mw_sleep_ms(string vhost)
     return ::atoi(conf->arg0().c_str());
 }
 
+bool SrsConfig::get_realtime_enabled(string vhost)
+{
+    SrsConfDirective* conf = get_vhost(vhost);
+
+    if (!conf) {
+        return SRS_PERF_MIN_LATENCY_ENABLED;
+    }
+
+    conf = conf->get("min_latency");
+    if (!conf || conf->arg0() != "off") {
+        return SRS_PERF_MIN_LATENCY_ENABLED;
+    }
+
+    return false;
+}
+
 int SrsConfig::get_global_chunk_size()
 {
     SrsConfDirective* conf = root->get("chunk_size");

trunk/src/app/srs_app_config.hpp

@@ -545,6 +545,12 @@ class SrsConfig
     */
     // TODO: FIXME: add utest for mw config.
     virtual int                 get_mw_sleep_ms(std::string vhost);
+    /**
+    * whether min latency mode enabled.
+    * @param vhost, the vhost to get the min_latency.
+    */
+    // TODO: FIXME: add utest for min_latency.
+    virtual bool                get_realtime_enabled(std::string vhost);
 private:
     /**
     * get the global chunk size.

trunk/src/app/srs_app_recv_thread.cpp

@@ -251,6 +251,8 @@ SrsPublishRecvThread::SrsPublishRecvThread(
     mr = _srs_config->get_mr_enabled(req->vhost);
     mr_sleep = _srs_config->get_mr_sleep_ms(req->vhost);
 
+    realtime = _srs_config->get_realtime_enabled(req->vhost);
+
     _srs_config->subscribe(this);
 }
 
@@ -340,6 +342,10 @@ int SrsPublishRecvThread::handle(SrsCommonMessage* msg)
     int ret = ERROR_SUCCESS;
 
     _nb_msgs++;
+
+    // log to show the time of recv thread.
+    srs_verbose("recv thread now=%"PRId64"us, got msg time=%"PRId64"ms, size=%d",
+        srs_update_system_time_ms(), msg->header.timestamp, msg->size);
 
     // the rtmp connection will handle this message
     ret = _conn->handle_publish_message(_source, msg, _is_fmle, _is_edge);
@@ -363,7 +369,7 @@ void SrsPublishRecvThread::on_recv_error(int ret)
 #ifdef SRS_PERF_MERGED_READ
 void SrsPublishRecvThread::on_read(ssize_t nread)
 {
-    if (!mr) {
+    if (!mr || realtime) {
         return;
     }
 
@@ -386,6 +392,10 @@ void SrsPublishRecvThread::on_read(ssize_t nread)
 int SrsPublishRecvThread::on_reload_vhost_mr(string vhost)
 {
     int ret = ERROR_SUCCESS;
+
+    if (req->vhost != vhost) {
+        return ret;
+    }
 
     // the mr settings, 
     // @see https://github.com/winlinvip/simple-rtmp-server/issues/241
@@ -419,6 +429,21 @@ int SrsPublishRecvThread::on_reload_vhost_mr(string vhost)
     return ret;
 }
 
+int SrsPublishRecvThread::on_reload_vhost_realtime(string vhost)
+{
+    int ret = ERROR_SUCCESS;
+
+    if (req->vhost != vhost) {
+        return ret;
+    }
+
+    bool realtime_enabled = _srs_config->get_realtime_enabled(req->vhost);
+    srs_trace("realtime changed %d=>%d", realtime, realtime_enabled);
+    realtime = realtime_enabled;
+
+    return ret;
+}
+
 void SrsPublishRecvThread::set_socket_buffer(int sleep_ms)
 {
     // the bytes:
@@ -446,8 +471,9 @@ void SrsPublishRecvThread::set_socket_buffer(int sleep_ms)
     }
     getsockopt(fd, SOL_SOCKET, SO_RCVBUF, &nb_rbuf, &sock_buf_size);
 
-    srs_trace("mr change sleep %d=>%d, erbuf=%d, rbuf %d=>%d, sbytes=%d",
-        mr_sleep, sleep_ms, socket_buffer_size, onb_rbuf, nb_rbuf, SRS_MR_SMALL_BYTES);
+    srs_trace("mr change sleep %d=>%d, erbuf=%d, rbuf %d=>%d, sbytes=%d, realtime=%d",
+        mr_sleep, sleep_ms, socket_buffer_size, onb_rbuf, nb_rbuf, 
+        SRS_MR_SMALL_BYTES, realtime);
 
     rtmp->set_recv_buffer(nb_rbuf);
 }

trunk/src/app/srs_app_recv_thread.hpp

@@ -153,6 +153,9 @@ class SrsPublishRecvThread : virtual public ISrsMessageHandler
     bool mr;
     int mr_fd;
     int mr_sleep;
+    // for realtime
+    // @see https://github.com/winlinvip/simple-rtmp-server/issues/257
+    bool realtime;
     // the recv thread error code.
     int recv_error_code;
     SrsRtmpConn* _conn;
@@ -193,6 +196,7 @@ class SrsPublishRecvThread : virtual public ISrsMessageHandler
 // interface ISrsReloadHandler
 public:
     virtual int on_reload_vhost_mr(std::string vhost);
+    virtual int on_reload_vhost_realtime(std::string vhost);
 private:
     virtual void set_socket_buffer(int sleep_ms);
 };

trunk/src/app/srs_app_reload.cpp

@@ -150,6 +150,11 @@ int ISrsReloadHandler::on_reload_vhost_mw(string /*vhost*/)
     return ERROR_SUCCESS;
 }
 
+int ISrsReloadHandler::on_reload_vhost_realtime(string /*vhost*/)
+{
+    return ERROR_SUCCESS;
+}
+
 int ISrsReloadHandler::on_reload_vhost_chunk_size(string /*vhost*/)
 {
     return ERROR_SUCCESS;

trunk/src/app/srs_app_reload.hpp

@@ -67,6 +67,7 @@ class ISrsReloadHandler
     virtual int on_reload_vhost_dvr(std::string vhost);
     virtual int on_reload_vhost_mr(std::string vhost);
     virtual int on_reload_vhost_mw(std::string vhost);
+    virtual int on_reload_vhost_realtime(std::string vhost);
     virtual int on_reload_vhost_chunk_size(std::string vhost);
     virtual int on_reload_vhost_transcode(std::string vhost);
     virtual int on_reload_ingest_removed(std::string vhost, std::string ingest_id);