winlin

refine the tcp nodelay, effect connected connections.

@@ -768,6 +768,17 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root) @@ -768,6 +768,17 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
768 } 768 }
769 srs_trace("vhost %s reload smi success.", vhost.c_str()); 769 srs_trace("vhost %s reload smi success.", vhost.c_str());
770 } 770 }
  771 + // tcp_nodelay, only one per vhost
  772 + if (!srs_directive_equals(new_vhost->get("tcp_nodelay"), old_vhost->get("tcp_nodelay"))) {
  773 + for (it = subscribes.begin(); it != subscribes.end(); ++it) {
  774 + ISrsReloadHandler* subscribe = *it;
  775 + if ((ret = subscribe->on_reload_vhost_tcp_nodelay(vhost)) != ERROR_SUCCESS) {
  776 + srs_error("vhost %s notify subscribes tcp_nodelay failed. ret=%d", vhost.c_str(), ret);
  777 + return ret;
  778 + }
  779 + }
  780 + srs_trace("vhost %s reload tcp_nodelay success.", vhost.c_str());
  781 + }
771 // publish_1stpkt_timeout, only one per vhost 782 // publish_1stpkt_timeout, only one per vhost
772 if (!srs_directive_equals(new_vhost->get("publish_1stpkt_timeout"), old_vhost->get("publish_1stpkt_timeout"))) { 783 if (!srs_directive_equals(new_vhost->get("publish_1stpkt_timeout"), old_vhost->get("publish_1stpkt_timeout"))) {
773 for (it = subscribes.begin(); it != subscribes.end(); ++it) { 784 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -175,6 +175,11 @@ int ISrsReloadHandler::on_reload_vhost_smi(string /*vhost*/) @@ -175,6 +175,11 @@ int ISrsReloadHandler::on_reload_vhost_smi(string /*vhost*/)
175 return ERROR_SUCCESS; 175 return ERROR_SUCCESS;
176 } 176 }
177 177
  178 +int ISrsReloadHandler::on_reload_vhost_tcp_nodelay(string /*vhost*/)
  179 +{
  180 + return ERROR_SUCCESS;
  181 +}
  182 +
178 int ISrsReloadHandler::on_reload_vhost_realtime(string /*vhost*/) 183 int ISrsReloadHandler::on_reload_vhost_realtime(string /*vhost*/)
179 { 184 {
180 return ERROR_SUCCESS; 185 return ERROR_SUCCESS;
@@ -74,6 +74,7 @@ public: @@ -74,6 +74,7 @@ public:
74 virtual int on_reload_vhost_mr(std::string vhost); 74 virtual int on_reload_vhost_mr(std::string vhost);
75 virtual int on_reload_vhost_mw(std::string vhost); 75 virtual int on_reload_vhost_mw(std::string vhost);
76 virtual int on_reload_vhost_smi(std::string vhost); 76 virtual int on_reload_vhost_smi(std::string vhost);
  77 + virtual int on_reload_vhost_tcp_nodelay(std::string vhost);
77 virtual int on_reload_vhost_realtime(std::string vhost); 78 virtual int on_reload_vhost_realtime(std::string vhost);
78 virtual int on_reload_vhost_p1stpt(std::string vhost); 79 virtual int on_reload_vhost_p1stpt(std::string vhost);
79 virtual int on_reload_vhost_pnt(std::string vhost); 80 virtual int on_reload_vhost_pnt(std::string vhost);
@@ -94,6 +94,7 @@ SrsRtmpConn::SrsRtmpConn(SrsServer* svr, st_netfd_t c) @@ -94,6 +94,7 @@ SrsRtmpConn::SrsRtmpConn(SrsServer* svr, st_netfd_t c)
94 mw_enabled = false; 94 mw_enabled = false;
95 realtime = SRS_PERF_MIN_LATENCY_ENABLED; 95 realtime = SRS_PERF_MIN_LATENCY_ENABLED;
96 send_min_interval = 0; 96 send_min_interval = 0;
  97 + tcp_nodelay = false;
97 98
98 _srs_config->subscribe(this); 99 _srs_config->subscribe(this);
99 } 100 }
@@ -265,6 +266,19 @@ int SrsRtmpConn::on_reload_vhost_smi(string vhost) @@ -265,6 +266,19 @@ int SrsRtmpConn::on_reload_vhost_smi(string vhost)
265 return ret; 266 return ret;
266 } 267 }
267 268
  269 +int SrsRtmpConn::on_reload_vhost_tcp_nodelay(string vhost)
  270 +{
  271 + int ret = ERROR_SUCCESS;
  272 +
  273 + if (req->vhost != vhost) {
  274 + return ret;
  275 + }
  276 +
  277 + set_sock_options();
  278 +
  279 + return ret;
  280 +}
  281 +
268 int SrsRtmpConn::on_reload_vhost_realtime(string vhost) 282 int SrsRtmpConn::on_reload_vhost_realtime(string vhost)
269 { 283 {
270 int ret = ERROR_SUCCESS; 284 int ret = ERROR_SUCCESS;
@@ -651,8 +665,8 @@ int SrsRtmpConn::do_playing(SrsSource* source, SrsConsumer* consumer, SrsQueueRe @@ -651,8 +665,8 @@ int SrsRtmpConn::do_playing(SrsSource* source, SrsConsumer* consumer, SrsQueueRe
651 // set the sock options. 665 // set the sock options.
652 set_sock_options(); 666 set_sock_options();
653 667
654 - srs_trace("start play smi=%.2f, mw_sleep=%d, mw_enabled=%d, realtime=%d",  
655 - send_min_interval, mw_sleep, mw_enabled, realtime); 668 + srs_trace("start play smi=%.2f, mw_sleep=%d, mw_enabled=%d, realtime=%d, tcp_nodelay=%d",
  669 + send_min_interval, mw_sleep, mw_enabled, realtime, tcp_nodelay);
656 670
657 while (!disposed) { 671 while (!disposed) {
658 // collect elapse for pithy print. 672 // collect elapse for pithy print.
@@ -837,15 +851,18 @@ int SrsRtmpConn::do_publishing(SrsSource* source, SrsPublishRecvThread* trd) @@ -837,15 +851,18 @@ int SrsRtmpConn::do_publishing(SrsSource* source, SrsPublishRecvThread* trd)
837 return ret; 851 return ret;
838 } 852 }
839 853
  854 + // initialize the publish timeout.
  855 + publish_1stpkt_timeout = _srs_config->get_publish_1stpkt_timeout(req->vhost);
  856 + publish_normal_timeout = _srs_config->get_publish_1stpkt_timeout(req->vhost);
  857 +
840 // set the sock options. 858 // set the sock options.
841 set_sock_options(); 859 set_sock_options();
842 860
843 if (true) { 861 if (true) {
844 bool mr = _srs_config->get_mr_enabled(req->vhost); 862 bool mr = _srs_config->get_mr_enabled(req->vhost);
845 int mr_sleep = _srs_config->get_mr_sleep_ms(req->vhost); 863 int mr_sleep = _srs_config->get_mr_sleep_ms(req->vhost);
846 - publish_1stpkt_timeout = _srs_config->get_publish_1stpkt_timeout(req->vhost);  
847 - publish_normal_timeout = _srs_config->get_publish_1stpkt_timeout(req->vhost);  
848 - srs_trace("start publish mr=%d/%d, p1stpt=%d, pnt=%d", mr, mr_sleep, publish_1stpkt_timeout, publish_normal_timeout); 864 + srs_trace("start publish mr=%d/%d, p1stpt=%d, pnt=%d, tcp_nodelay=%d",
  865 + mr, mr_sleep, publish_1stpkt_timeout, publish_normal_timeout, tcp_nodelay);
849 } 866 }
850 867
851 int64_t nb_msgs = 0; 868 int64_t nb_msgs = 0;
@@ -1173,22 +1190,24 @@ void SrsRtmpConn::change_mw_sleep(int sleep_ms) @@ -1173,22 +1190,24 @@ void SrsRtmpConn::change_mw_sleep(int sleep_ms)
1173 1190
1174 void SrsRtmpConn::set_sock_options() 1191 void SrsRtmpConn::set_sock_options()
1175 { 1192 {
1176 - if (_srs_config->get_tcp_nodelay(req->vhost)) { 1193 + bool nvalue = _srs_config->get_tcp_nodelay(req->vhost);
  1194 + if (nvalue != tcp_nodelay) {
  1195 + tcp_nodelay = nvalue;
1177 #ifdef SRS_PERF_TCP_NODELAY 1196 #ifdef SRS_PERF_TCP_NODELAY
1178 int fd = st_netfd_fileno(stfd); 1197 int fd = st_netfd_fileno(stfd);
1179 - 1198 +
1180 socklen_t nb_v = sizeof(int); 1199 socklen_t nb_v = sizeof(int);
1181 - 1200 +
1182 int ov = 0; 1201 int ov = 0;
1183 getsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &ov, &nb_v); 1202 getsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &ov, &nb_v);
1184 -  
1185 - int v = 1; 1203 +
  1204 + int v = tcp_nodelay;
1186 // set the socket send buffer when required larger buffer 1205 // set the socket send buffer when required larger buffer
1187 if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &v, nb_v) < 0) { 1206 if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &v, nb_v) < 0) {
1188 srs_warn("set sock TCP_NODELAY=%d failed.", v); 1207 srs_warn("set sock TCP_NODELAY=%d failed.", v);
1189 } 1208 }
1190 getsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &v, &nb_v); 1209 getsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &v, &nb_v);
1191 - 1210 +
1192 srs_trace("set TCP_NODELAY %d=>%d", ov, v); 1211 srs_trace("set TCP_NODELAY %d=>%d", ov, v);
1193 #else 1212 #else
1194 srs_warn("SRS_PERF_TCP_NODELAY is disabled but tcp_nodelay configed."); 1213 srs_warn("SRS_PERF_TCP_NODELAY is disabled but tcp_nodelay configed.");
@@ -91,6 +91,8 @@ private: @@ -91,6 +91,8 @@ private:
91 int publish_1stpkt_timeout; 91 int publish_1stpkt_timeout;
92 // publish normal packet timeout in ms 92 // publish normal packet timeout in ms
93 int publish_normal_timeout; 93 int publish_normal_timeout;
  94 + // whether enable the tcp_nodelay.
  95 + bool tcp_nodelay;
94 public: 96 public:
95 SrsRtmpConn(SrsServer* svr, st_netfd_t c); 97 SrsRtmpConn(SrsServer* svr, st_netfd_t c);
96 virtual ~SrsRtmpConn(); 98 virtual ~SrsRtmpConn();
@@ -103,6 +105,7 @@ public: @@ -103,6 +105,7 @@ public:
103 virtual int on_reload_vhost_removed(std::string vhost); 105 virtual int on_reload_vhost_removed(std::string vhost);
104 virtual int on_reload_vhost_mw(std::string vhost); 106 virtual int on_reload_vhost_mw(std::string vhost);
105 virtual int on_reload_vhost_smi(std::string vhost); 107 virtual int on_reload_vhost_smi(std::string vhost);
  108 + virtual int on_reload_vhost_tcp_nodelay(std::string vhost);
106 virtual int on_reload_vhost_realtime(std::string vhost); 109 virtual int on_reload_vhost_realtime(std::string vhost);
107 virtual int on_reload_vhost_p1stpt(std::string vhost); 110 virtual int on_reload_vhost_p1stpt(std::string vhost);
108 virtual int on_reload_vhost_pnt(std::string vhost); 111 virtual int on_reload_vhost_pnt(std::string vhost);