For #1657: Refine TCP connections arch

trunk/src/app/srs_app_caster_flv.cpp

@@ -97,7 +97,7 @@ srs_error_t SrsAppCasterFlv::on_tcp_client(srs_netfd_t stfd)
 
 void SrsAppCasterFlv::remove(ISrsResource* c)
 {
-    SrsTcpConnection* conn = dynamic_cast<SrsTcpConnection*>(c);
+    SrsHttpConn* conn = dynamic_cast<SrsHttpConn*>(c);
 
     std::vector<SrsHttpConn*>::iterator it;
     if ((it = std::find(conns.begin(), conns.end(), conn)) != conns.end()) {

trunk/src/app/srs_app_conn.cpp

@@ -314,6 +314,14 @@ void SrsResourceManager::dispose(ISrsResource* c)
     srs_freep(c);
 }
 
+ISrsExpire::ISrsExpire()
+{
+}
+
+ISrsExpire::~ISrsExpire()
+{
+}
+
 ISrsStartableConneciton::ISrsStartableConneciton()
 {
 }
@@ -329,7 +337,7 @@ SrsTcpConnection::SrsTcpConnection(ISrsResourceManager* cm, srs_netfd_t c, strin
     ip = cip;
     port = cport;
     create_time = srsu2ms(srs_get_system_time());
-    
+
     skt = new SrsStSocket();
     clk = new SrsWallClock();
     kbps = new SrsKbps(clk);
@@ -507,4 +515,159 @@ void SrsTcpConnection::expire()
     trd->interrupt();
 }
 
+SrsTcpConnection2::SrsTcpConnection2(srs_netfd_t c)
+{
+    stfd = c;
+    skt = new SrsStSocket();
+}
+
+SrsTcpConnection2::~SrsTcpConnection2()
+{
+    srs_freep(skt);
+    srs_close_stfd(stfd);
+}
+
+srs_error_t SrsTcpConnection2::initialize()
+{
+    srs_error_t err = srs_success;
+
+    if ((err = skt->initialize(stfd)) != srs_success) {
+        return srs_error_wrap(err, "init socket");
+    }
+
+    return err;
+}
+
+srs_error_t SrsTcpConnection2::set_tcp_nodelay(bool v)
+{
+    srs_error_t err = srs_success;
+
+    int r0 = 0;
+    socklen_t nb_v = sizeof(int);
+    int fd = srs_netfd_fileno(stfd);
+
+    int ov = 0;
+    if ((r0 = getsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &ov, &nb_v)) != 0) {
+        return srs_error_new(ERROR_SOCKET_NO_NODELAY, "getsockopt fd=%d, r0=%d", fd, r0);
+    }
+
+#ifndef SRS_PERF_TCP_NODELAY
+    srs_warn("ignore TCP_NODELAY, fd=%d, ov=%d", fd, ov);
+    return err;
+#endif
+
+    int iv = (v? 1:0);
+    if ((r0 = setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &iv, nb_v)) != 0) {
+        return srs_error_new(ERROR_SOCKET_NO_NODELAY, "setsockopt fd=%d, r0=%d", fd, r0);
+    }
+    if ((r0 = getsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &iv, &nb_v)) != 0) {
+        return srs_error_new(ERROR_SOCKET_NO_NODELAY, "getsockopt fd=%d, r0=%d", fd, r0);
+    }
+
+    srs_trace("set fd=%d TCP_NODELAY %d=>%d", fd, ov, iv);
+
+    return err;
+}
+
+srs_error_t SrsTcpConnection2::set_socket_buffer(srs_utime_t buffer_v)
+{
+    srs_error_t err = srs_success;
+
+    int r0 = 0;
+    int fd = srs_netfd_fileno(stfd);
+    socklen_t nb_v = sizeof(int);
+
+    int ov = 0;
+    if ((r0 = getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &ov, &nb_v)) != 0) {
+        return srs_error_new(ERROR_SOCKET_SNDBUF, "getsockopt fd=%d, r0=%d", fd, r0);
+    }
+
+#ifndef SRS_PERF_MW_SO_SNDBUF
+    srs_warn("ignore SO_SNDBUF, fd=%d, ov=%d", fd, ov);
+    return err;
+#endif
+
+    // the bytes:
+    //      4KB=4096, 8KB=8192, 16KB=16384, 32KB=32768, 64KB=65536,
+    //      128KB=131072, 256KB=262144, 512KB=524288
+    // the buffer should set to sleep*kbps/8,
+    // for example, your system delivery stream in 1000kbps,
+    // sleep 800ms for small bytes, the buffer should set to:
+    //      800*1000/8=100000B(about 128KB).
+    // other examples:
+    //      2000*3000/8=750000B(about 732KB).
+    //      2000*5000/8=1250000B(about 1220KB).
+    int kbps = 4000;
+    int iv = srsu2ms(buffer_v) * kbps / 8;
+
+    // socket send buffer, system will double it.
+    iv = iv / 2;
+
+    // override the send buffer by macro.
+#ifdef SRS_PERF_SO_SNDBUF_SIZE
+    iv = SRS_PERF_SO_SNDBUF_SIZE / 2;
+#endif
+
+    // set the socket send buffer when required larger buffer
+    if (setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &iv, nb_v) < 0) {
+        return srs_error_new(ERROR_SOCKET_SNDBUF, "setsockopt fd=%d, r0=%d", fd, r0);
+    }
+    if ((r0 = getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &iv, &nb_v)) != 0) {
+        return srs_error_new(ERROR_SOCKET_SNDBUF, "getsockopt fd=%d, r0=%d", fd, r0);
+    }
+
+    srs_trace("set fd=%d, SO_SNDBUF=%d=>%d, buffer=%dms", fd, ov, iv, srsu2ms(buffer_v));
+
+    return err;
+}
+
+void SrsTcpConnection2::set_recv_timeout(srs_utime_t tm)
+{
+    skt->set_recv_timeout(tm);
+}
+
+srs_utime_t SrsTcpConnection2::get_recv_timeout()
+{
+    return skt->get_recv_timeout();
+}
+
+srs_error_t SrsTcpConnection2::read_fully(void* buf, size_t size, ssize_t* nread)
+{
+    return skt->read_fully(buf, size, nread);
+}
+
+int64_t SrsTcpConnection2::get_recv_bytes()
+{
+    return skt->get_recv_bytes();
+}
+
+int64_t SrsTcpConnection2::get_send_bytes()
+{
+    return skt->get_send_bytes();
+}
+
+srs_error_t SrsTcpConnection2::read(void* buf, size_t size, ssize_t* nread)
+{
+    return skt->read(buf, size, nread);
+}
+
+void SrsTcpConnection2::set_send_timeout(srs_utime_t tm)
+{
+    skt->set_send_timeout(tm);
+}
+
+srs_utime_t SrsTcpConnection2::get_send_timeout()
+{
+    return skt->get_send_timeout();
+}
+
+srs_error_t SrsTcpConnection2::write(void* buf, size_t size, ssize_t* nwrite)
+{
+    return skt->write(buf, size, nwrite);
+}
+
+srs_error_t SrsTcpConnection2::writev(const iovec *iov, int iov_size, ssize_t* nwrite)
+{
+    return skt->writev(iov, iov_size, nwrite);
+}
 

trunk/src/app/srs_app_conn.hpp

@@ -106,6 +106,18 @@ class SrsResourceManager : virtual public ISrsCoroutineHandler, virtual public I
     void dispose(ISrsResource* c);
 };
 
+// If a connection is able to be expired,
+// user can use HTTP-API to kick-off it.
+class ISrsExpire
+{
+public:
+    ISrsExpire();
+    virtual ~ISrsExpire();
+public:
+    // Set connection to expired to kick-off it.
+    virtual void expire() = 0;
+};
+
 // Interface for connection that is startable.
 class ISrsStartableConneciton : virtual public ISrsConnection
     , virtual public ISrsStartable, virtual public ISrsKbpsDelta
@@ -119,7 +131,7 @@ class ISrsStartableConneciton : virtual public ISrsConnection
 // all connections accept from listener must extends from this base class,
 // server will add the connection to manager, and delete it when remove.
 class SrsTcpConnection : virtual public ISrsStartableConneciton
-    , virtual public ISrsReloadHandler, virtual public ISrsCoroutineHandler
+    , virtual public ISrsReloadHandler, virtual public ISrsCoroutineHandler, virtual public ISrsExpire
 {
 protected:
     // Each connection start a green thread,
@@ -185,4 +197,38 @@ class SrsTcpConnection : virtual public ISrsStartableConneciton
     virtual srs_error_t do_cycle() = 0;
 };
 
+// The basic connection of SRS, for TCP based protocols,
+// all connections accept from listener must extends from this base class,
+// server will add the connection to manager, and delete it when remove.
+class SrsTcpConnection2 : virtual public ISrsProtocolReadWriter
+{
+private:
+    // The underlayer st fd handler.
+    srs_netfd_t stfd;
+    // The underlayer socket.
+    SrsStSocket* skt;
+public:
+    SrsTcpConnection2(srs_netfd_t c);
+    virtual ~SrsTcpConnection2();
+public:
+    virtual srs_error_t initialize();
+public:
+    // Set socket option TCP_NODELAY.
+    virtual srs_error_t set_tcp_nodelay(bool v);
+    // Set socket option SO_SNDBUF in srs_utime_t.
+    virtual srs_error_t set_socket_buffer(srs_utime_t buffer_v);
+// Interface ISrsProtocolReadWriter
+public:
+    virtual void set_recv_timeout(srs_utime_t tm);
+    virtual srs_utime_t get_recv_timeout();
+    virtual srs_error_t read_fully(void* buf, size_t size, ssize_t* nread);
+    virtual int64_t get_recv_bytes();
+    virtual int64_t get_send_bytes();
+    virtual srs_error_t read(void* buf, size_t size, ssize_t* nread);
+    virtual void set_send_timeout(srs_utime_t tm);
+    virtual srs_utime_t get_send_timeout();
+    virtual srs_error_t write(void* buf, size_t size, ssize_t* nwrite);
+    virtual srs_error_t writev(const iovec *iov, int iov_size, ssize_t* nwrite);
+};
+
 #endif

trunk/src/app/srs_app_http_api.cpp

@@ -852,7 +852,7 @@ srs_error_t SrsGoApiClients::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessa
         if (!client) {
             return srs_api_response_code(w, r, ERROR_RTMP_CLIENT_NOT_FOUND);
         }
-        
+
         client->conn->expire();
         srs_warn("kickoff client id=%s ok", client_id.c_str());
     } else {
@@ -1674,22 +1674,38 @@ srs_error_t SrsGoApiTcmalloc::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMess
 }
 #endif
 
-SrsHttpApi::SrsHttpApi(ISrsResourceManager* cm, srs_netfd_t fd, SrsHttpServeMux* m, string cip, int port)
-    : SrsTcpConnection(cm, fd, cip, port)
+SrsHttpApi::SrsHttpApi(ISrsResourceManager* cm, srs_netfd_t fd, SrsHttpServeMux* m, string cip, int cport)
 {
     mux = m;
     cors = new SrsHttpCorsMux();
     parser = new SrsHttpParser();
+
+    skt = new SrsTcpConnection2(fd);
+    manager = cm;
+    ip = cip;
+    port = cport;
+    create_time = srsu2ms(srs_get_system_time());
+    clk = new SrsWallClock();
+    kbps = new SrsKbps(clk);
+    kbps->set_io(skt, skt);
+    trd = new SrsSTCoroutine("api", this);
 
     _srs_config->subscribe(this);
 }
 
 SrsHttpApi::~SrsHttpApi()
 {
+    _srs_config->unsubscribe(this);
+
+    trd->interrupt();
+    srs_freep(trd);
+
     srs_freep(parser);
     srs_freep(cors);
-
-    _srs_config->unsubscribe(this);
+
+    srs_freep(kbps);
+    srs_freep(clk);
+    srs_freep(skt);
 }
 
 std::string SrsHttpApi::desc()
@@ -1699,7 +1715,7 @@ std::string SrsHttpApi::desc()
 
 void SrsHttpApi::remark(int64_t* in, int64_t* out)
 {
-    // TODO: FIXME: implements it
+    kbps->remark(in, out);
 }
 
 srs_error_t SrsHttpApi::do_cycle()
@@ -1804,3 +1820,62 @@ srs_error_t SrsHttpApi::on_reload_http_api_crossdomain()
     return err;
 }
 
+srs_error_t SrsHttpApi::start()
+{
+    srs_error_t err = srs_success;
+
+    if ((err = skt->initialize()) != srs_success) {
+        return srs_error_wrap(err, "init socket");
+    }
+
+    if ((err = trd->start()) != srs_success) {
+        return srs_error_wrap(err, "coroutine");
+    }
+
+    return err;
+}
+
+srs_error_t SrsHttpApi::cycle()
+{
+    srs_error_t err = do_cycle();
+
+    // Notify manager to remove it.
+    manager->remove(this);
+
+    // success.
+    if (err == srs_success) {
+        srs_trace("client finished.");
+        return err;
+    }
+
+    // It maybe success with message.
+    if (srs_error_code(err) == ERROR_SUCCESS) {
+        srs_trace("client finished%s.", srs_error_summary(err).c_str());
+        srs_freep(err);
+        return err;
+    }
+
+    // client close peer.
+    // TODO: FIXME: Only reset the error when client closed it.
+    if (srs_is_client_gracefully_close(err)) {
+        srs_warn("client disconnect peer. ret=%d", srs_error_code(err));
+    } else if (srs_is_server_gracefully_close(err)) {
+        srs_warn("server disconnect. ret=%d", srs_error_code(err));
+    } else {
+        srs_error("serve error %s", srs_error_desc(err).c_str());
+    }
+
+    srs_freep(err);
+    return srs_success;
+}
+
+string SrsHttpApi::remote_ip()
+{
+    return ip;
+}
+
+const SrsContextId& SrsHttpApi::get_id()
+{
+    return trd->cid();
+}
+