for #730, support config in/out ack size. 3.0.13

@@ -184,6 +184,7 @@ Please select your language:
 
 
 ### V3 changes
 ### V3 changes
 
 
+* v3.0, 2017-01-06, for [#730][bug #730] support config in/out ack size. 3.0.13
 * v3.0, 2017-01-06, for [#711][bug #711] support perfile for transcode. 3.0.12
 * v3.0, 2017-01-06, for [#711][bug #711] support perfile for transcode. 3.0.12
 * v3.0, 2017-01-05, patch ST for valgrind and ARM. 3.0.11
 * v3.0, 2017-01-05, patch ST for valgrind and ARM. 3.0.11
 * v3.0, 2017-01-05, for [#324][bug #324], always enable hstrs. 3.0.10
 * v3.0, 2017-01-05, for [#324][bug #324], always enable hstrs. 3.0.10
@@ -1355,6 +1356,7 @@ Winlin
 [bug #717]: https://github.com/ossrs/srs/issues/717
 [bug #717]: https://github.com/ossrs/srs/issues/717
 [bug #691]: https://github.com/ossrs/srs/issues/691
 [bug #691]: https://github.com/ossrs/srs/issues/691
 [bug #711]: https://github.com/ossrs/srs/issues/711
 [bug #711]: https://github.com/ossrs/srs/issues/711
+[bug #730]: https://github.com/ossrs/srs/issues/730
 [bug #xxxxxxxxxx]: https://github.com/ossrs/srs/issues/xxxxxxxxxx
 [bug #xxxxxxxxxx]: https://github.com/ossrs/srs/issues/xxxxxxxxxx
 
 
 [exo #828]: https://github.com/google/ExoPlayer/pull/828
 [exo #828]: https://github.com/google/ExoPlayer/pull/828

@@ -300,6 +300,19 @@ vhost scope.vhost.srs.com {
     # vhost chunk size will override the global value.
     # vhost chunk size will override the global value.
     # default: global chunk size.
     # default: global chunk size.
     chunk_size      128;
     chunk_size      128;
+    
+    # The input ack size, 0 to not set.
+    # Generally, it's set by the message from peer,
+    # but for some peer(encoder), it never send message but use a different ack size.
+    # We can chnage the default ack size in server-side, to send acknowledge message,
+    # or the encoder maybe blocked after publishing for some time.
+    # Default: 0
+    in_ack_size     0;
+    
+    # The output ack size, 0 to not set.
+    # This is used to notify the peer(player) to send acknowledge to server.
+    # Default: 2500000
+    out_ack_size    2500000;
 }
 }
 
 
 # set the chunk size of vhost.
 # set the chunk size of vhost.

@@ -3826,6 +3826,7 @@ int SrsConfig::check_config()
                 && n != "play" && n != "publish" && n != "cluster"
                 && n != "play" && n != "publish" && n != "cluster"
                 && n != "security" && n != "http_remux"
                 && n != "security" && n != "http_remux"
                 && n != "http_static" && n != "hds" && n != "exec"
                 && n != "http_static" && n != "hds" && n != "exec"
+                && n != "in_ack_size" && n != "out_ack_size"
             ) {
             ) {
                 ret = ERROR_SYSTEM_CONFIG_INVALID;
                 ret = ERROR_SYSTEM_CONFIG_INVALID;
                 srs_error("unsupported vhost directive %s, ret=%d", n.c_str(), ret);
                 srs_error("unsupported vhost directive %s, ret=%d", n.c_str(), ret);
@@ -4694,6 +4695,40 @@ SrsConfDirective* SrsConfig::get_refer_publish(string vhost)
     return conf->get("publish");
     return conf->get("publish");
 }
 }
 
 
+int SrsConfig::get_in_ack_size(string vhost)
+{
+    static int DEFAULT = 0;
+    
+    SrsConfDirective* conf = get_vhost(vhost);
+    if (!conf) {
+        return DEFAULT;
+    }
+    
+    conf = conf->get("in_ack_size");
+    if (!conf || conf->arg0().empty()) {
+        return DEFAULT;
+    }
+    
+    return ::atoi(conf->arg0().c_str());
+}
+
+int SrsConfig::get_out_ack_size(string vhost)
+{
+    static int DEFAULT = 2500000;
+    
+    SrsConfDirective* conf = get_vhost(vhost);
+    if (!conf) {
+        return DEFAULT;
+    }
+    
+    conf = conf->get("out_ack_size");
+    if (!conf || conf->arg0().empty()) {
+        return DEFAULT;
+    }
+    
+    return ::atoi(conf->arg0().c_str());
+}
+
 int SrsConfig::get_chunk_size(string vhost)
 int SrsConfig::get_chunk_size(string vhost)
 {
 {
     if (vhost.empty()) {
     if (vhost.empty()) {

@@ -754,6 +754,10 @@ public:
      * @return the refer, NULL for not configed.
      * @return the refer, NULL for not configed.
      */
      */
     virtual SrsConfDirective*   get_refer_publish(std::string vhost);
     virtual SrsConfDirective*   get_refer_publish(std::string vhost);
+    // Get the input default ack size, which is generally set by message from peer.
+    virtual int                 get_in_ack_size(std::string vhost);
+    // Get the output default ack size, to notify the peer to send acknowledge to server.
+    virtual int                 get_out_ack_size(std::string vhost);
     /**
     /**
     * get the chunk size of vhost.
     * get the chunk size of vhost.
     * @param vhost, the vhost to get the chunk size. use global if not specified.
     * @param vhost, the vhost to get the chunk size. use global if not specified.

@@ -578,11 +578,17 @@ int SrsRtmpConn::service_cycle()
 {    
 {    
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
-    if ((ret = rtmp->set_window_ack_size((int)(2.5 * 1000 * 1000))) != ERROR_SUCCESS) {
-        srs_error("set window acknowledgement size failed. ret=%d", ret);
+    int out_ack_size = _srs_config->get_out_ack_size(req->vhost);
+    if (out_ack_size && (ret = rtmp->set_window_ack_size(out_ack_size)) != ERROR_SUCCESS) {
+        srs_error("set output window acknowledgement size failed. ret=%d", ret);
+        return ret;
+    }
+    
+    int in_ack_size = _srs_config->get_in_ack_size(req->vhost);
+    if (in_ack_size && (ret = rtmp->set_in_window_ack_size(in_ack_size)) != ERROR_SUCCESS) {
+        srs_error("set input window acknowledgement size failed. ret=%d", ret);
         return ret;
         return ret;
     }
     }
-    srs_verbose("set window acknowledgement size success");
         
         
     if ((ret = rtmp->set_peer_bandwidth((int)(2.5 * 1000 * 1000), 2)) != ERROR_SUCCESS) {
     if ((ret = rtmp->set_peer_bandwidth((int)(2.5 * 1000 * 1000), 2)) != ERROR_SUCCESS) {
         srs_error("set peer bandwidth failed. ret=%d", ret);
         srs_error("set peer bandwidth failed. ret=%d", ret);

@@ -355,6 +355,12 @@ int64_t SrsProtocol::get_send_bytes()
     return skt->get_send_bytes();
     return skt->get_send_bytes();
 }
 }
 
 
+int SrsProtocol::set_in_window_ack_size(int ack_size)
+{
+    in_ack_size.window = ack_size;
+    return ERROR_SUCCESS;
+}
+
 int SrsProtocol::recv_message(SrsCommonMessage** pmsg)
 int SrsProtocol::recv_message(SrsCommonMessage** pmsg)
 {
 {
     *pmsg = NULL;
     *pmsg = NULL;
@@ -1620,7 +1626,7 @@ int SrsProtocol::response_acknowledgement_message()
     }
     }
     
     
     // ignore when delta bytes not exceed half of window(ack size).
     // ignore when delta bytes not exceed half of window(ack size).
-    uint32_t delta = (uint32_t)(skt->get_recv_bytes() - in_ack_size.nb_recv_bytes);
+    uint32_t delta = (uint32_t)(skt->get_recv_bytes() - in_ack_size.nb_recv_bytes)*100;
     if (delta < in_ack_size.window / 2) {
     if (delta < in_ack_size.window / 2) {
         return ret;
         return ret;
     }
     }
@@ -1635,7 +1641,6 @@ int SrsProtocol::response_acknowledgement_message()
     
     
     SrsAcknowledgementPacket* pkt = new SrsAcknowledgementPacket();
     SrsAcknowledgementPacket* pkt = new SrsAcknowledgementPacket();
     pkt->sequence_number = sequence_number;
     pkt->sequence_number = sequence_number;
-    srs_warn("ack sequence=%#x", sequence_number);
     
     
     // cache the message and use flush to send.
     // cache the message and use flush to send.
     if (!auto_response_when_recv) {
     if (!auto_response_when_recv) {
@@ -2548,6 +2553,11 @@ int SrsRtmpServer::set_window_ack_size(int ack_size)
     return ret;
     return ret;
 }
 }
 
 
+int SrsRtmpServer::set_in_window_ack_size(int ack_size)
+{
+    return protocol->set_in_window_ack_size(ack_size);
+}
+
 int SrsRtmpServer::set_peer_bandwidth(int bandwidth, int type)
 int SrsRtmpServer::set_peer_bandwidth(int bandwidth, int type)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;

@@ -330,6 +330,14 @@ public:
     virtual int64_t get_recv_bytes();
     virtual int64_t get_recv_bytes();
     virtual int64_t get_send_bytes();
     virtual int64_t get_send_bytes();
 public:
 public:
+    // Set the input default ack size. This is generally set by the message from peer,
+    // but for some encoder, it never send the ack message while it default to a none zone size.
+    // This will cause the encoder to block after publishing some messages to server,
+    // because it wait for server to send acknowledge, but server default to 0 which means no need
+    // to ack encoder. We can change the default input ack size. We will always response the
+    // ack size whatever the encoder set or not.
+    virtual int set_in_window_ack_size(int ack_size);
+public:
     /**
     /**
     * recv a RTMP message, which is bytes oriented.
     * recv a RTMP message, which is bytes oriented.
     * user can use decode_message to get the decoded RTMP packet.
     * user can use decode_message to get the decoded RTMP packet.
@@ -928,9 +936,11 @@ public:
      */
      */
     virtual int connect_app(SrsRequest* req);
     virtual int connect_app(SrsRequest* req);
     /**
     /**
-     * set ack size to client, client will send ack-size for each ack window
+     * set output ack size to client, client will send ack-size for each ack window
      */
      */
     virtual int set_window_ack_size(int ack_size);
     virtual int set_window_ack_size(int ack_size);
+    // Set the default input ack size value.
+    virtual int set_in_window_ack_size(int ack_size);
     /**
     /**
      * @type: The sender can mark this message hard (0), soft (1), or dynamic (2)
      * @type: The sender can mark this message hard (0), soft (1), or dynamic (2)
      * using the Limit type field.
      * using the Limit type field.