Fix #636, FD leak for requesting empty HTTP stream. 2.0.241

README.md

@@ -337,6 +337,7 @@ Remark:
 
 ## History
 
+* v2.0, 2017-04-30, Fix [#636][bug #636], FD leak for requesting empty HTTP stream. 2.0.241
 * v2.0, 2017-04-23, Fix [#851][bug #851], HTTP API support number of video frames for FPS. 2.0.240
 * <strong>v2.0, 2017-04-18, [2.0 release1(2.0.239)][r2.0r1] released. 86515 lines.</strong>
 * v2.0, 2017-04-18, Fix [#848][bug #848], crash at HTTP fast buffer grow. 2.0.239
@@ -1293,6 +1294,7 @@ Winlin
 [bug #844]: https://github.com/ossrs/srs/issues/844
 [bug #848]: https://github.com/ossrs/srs/issues/848
 [bug #851]: https://github.com/ossrs/srs/issues/851
+[bug #636]: https://github.com/ossrs/srs/issues/636
 [bug #xxxxxxxxxx]: https://github.com/ossrs/srs/issues/xxxxxxxxxx
 
 [exo #828]: https://github.com/google/ExoPlayer/pull/828

trunk/src/app/srs_app_http_conn.cpp

@@ -1310,6 +1310,19 @@ SrsResponseOnlyHttpConn::~SrsResponseOnlyHttpConn()
 {
 }
 
+int SrsResponseOnlyHttpConn::pop_message(ISrsHttpMessage** preq)
+{
+    int ret = ERROR_SUCCESS;
+    
+    SrsStSocket skt(stfd);
+    
+    if ((ret = parser->parse_message(&skt, this, preq)) != ERROR_SUCCESS) {
+        return ret;
+    }
+    
+    return ret;
+}
+
 int SrsResponseOnlyHttpConn::on_got_http_message(ISrsHttpMessage* msg)
 {
     int ret = ERROR_SUCCESS;

trunk/src/app/srs_app_http_conn.hpp

@@ -390,7 +390,7 @@ private:
 
 class SrsHttpConn : public SrsConnection
 {
-private:
+protected:
     SrsHttpParser* parser;
     ISrsHttpServeMux* http_mux;
 public:
@@ -421,6 +421,13 @@ class SrsResponseOnlyHttpConn : public SrsHttpConn
 public:
     SrsResponseOnlyHttpConn(IConnectionManager* cm, st_netfd_t fd, ISrsHttpServeMux* m);
     virtual ~SrsResponseOnlyHttpConn();
+public:
+    // Directly read a HTTP request message.
+    // It's exported for HTTP stream, such as HTTP FLV, only need to write to client when
+    // serving it, but we need to start a thread to read message to detect whether FD is closed.
+    // @see https://github.com/ossrs/srs/issues/636#issuecomment-298208427
+    // @remark Should only used in HTTP-FLV streaming connection.
+    virtual int pop_message(ISrsHttpMessage** preq);
 public:
     virtual int on_got_http_message(ISrsHttpMessage* msg);
 };

trunk/src/app/srs_app_http_stream.cpp

@@ -54,6 +54,7 @@ using namespace std;
 #include <srs_app_pithy_print.hpp>
 #include <srs_app_source.hpp>
 #include <srs_app_server.hpp>
+#include <srs_app_recv_thread.hpp>
 
 #endif
 
@@ -536,10 +537,28 @@ int SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
     SrsFastFlvStreamEncoder* ffe = dynamic_cast<SrsFastFlvStreamEncoder*>(enc);
 #endif
     
+    // Use receive thread to accept the close event to avoid FD leak.
+    // @see https://github.com/ossrs/srs/issues/636#issuecomment-298208427
+    SrsHttpMessage* hr = dynamic_cast<SrsHttpMessage*>(r);
+    SrsResponseOnlyHttpConn* hc = dynamic_cast<SrsResponseOnlyHttpConn*>(hr->connection());
+    
+    SrsHttpRecvThread* trd = new SrsHttpRecvThread(hc);
+    SrsAutoFree(SrsHttpRecvThread, trd);
+    
+    if ((ret = trd->start()) != ERROR_SUCCESS) {
+        srs_error("http: start notify thread failed, ret=%d", ret);
+        return ret;
+    }
+
     // TODO: free and erase the disabled entry after all related connections is closed.
     while (entry->enabled) {
         pprint->elapse();
         
+        // Whether client closed the FD.
+        if ((ret = trd->error_code()) != ERROR_SUCCESS) {
+            return ret;
+        }
+
         // 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_recv_thread.cpp

@@ -31,6 +31,8 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_core_performance.hpp>
 #include <srs_app_config.hpp>
 #include <srs_app_source.hpp>
+#include <srs_app_http_conn.hpp>
+#include <srs_core_autofree.hpp>
 
 using namespace std;
 
@@ -523,3 +525,42 @@ void SrsPublishRecvThread::set_socket_buffer(int sleep_ms)
     rtmp->set_recv_buffer(nb_rbuf);
 }
 
+SrsHttpRecvThread::SrsHttpRecvThread(SrsResponseOnlyHttpConn* c)
+{
+    conn = c;
+    error = ERROR_SUCCESS;
+    trd = new SrsOneCycleThread("http-receive", this);
+}
+
+SrsHttpRecvThread::~SrsHttpRecvThread()
+{
+    srs_freep(trd);
+}
+
+int SrsHttpRecvThread::start()
+{
+    return trd->start();
+}
+
+int SrsHttpRecvThread::error_code()
+{
+    return error;
+}
+
+int SrsHttpRecvThread::cycle()
+{
+    int ret = ERROR_SUCCESS;
+    
+    while (true) {
+        ISrsHttpMessage* req = NULL;
+        SrsAutoFree(ISrsHttpMessage, req);
+        
+        if ((ret = conn->pop_message(&req)) != ERROR_SUCCESS) {
+            error = ret;
+            break;
+        }
+    }
+    
+    return ret;
+}
+

trunk/src/app/srs_app_recv_thread.hpp

@@ -43,6 +43,8 @@ class SrsRtmpConn;
 class SrsSource;
 class SrsRequest;
 class SrsConsumer;
+class SrsHttpConn;
+class SrsResponseOnlyHttpConn;
 
 /**
  * for the recv thread to handle the message.
@@ -215,5 +217,29 @@ private:
     virtual void set_socket_buffer(int sleep_ms);
 };
 
+/**
+ * The HTTP receive thread, try to read messages util EOF.
+ * For example, the HTTP FLV serving thread will use the receive thread to break
+ * when client closed the request, to avoid FD leak.
+ * @see https://github.com/ossrs/srs/issues/636#issuecomment-298208427
+ */
+class SrsHttpRecvThread : public ISrsOneCycleThreadHandler
+{
+private:
+    SrsResponseOnlyHttpConn* conn;
+    SrsOneCycleThread* trd;
+    int error;
+public:
+    SrsHttpRecvThread(SrsResponseOnlyHttpConn* c);
+    virtual ~SrsHttpRecvThread();
+public:
+    virtual int start();
+public:
+    virtual int error_code();
+// interface ISrsOneCycleThreadHandler
+public:
+    virtual int cycle();
+};
+
 #endif
 

trunk/src/core/srs_core.hpp

@@ -31,7 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 // current release version
 #define VERSION_MAJOR       2
 #define VERSION_MINOR       0
-#define VERSION_REVISION    240
+#define VERSION_REVISION    241
 
 // generated by configure, only macros.
 #include <srs_auto_headers.hpp>