diff --git a/trunk/src/app/srs_app_conn.cpp b/trunk/src/app/srs_app_conn.cpp index b32151f80..b62419832 100644 --- a/trunk/src/app/srs_app_conn.cpp +++ b/trunk/src/app/srs_app_conn.cpp @@ -330,77 +330,47 @@ ISrsStartableConneciton::~ISrsStartableConneciton() { } -SrsTcpConnection::SrsTcpConnection(ISrsResourceManager* cm, srs_netfd_t c, string cip, int cport) +SrsTcpConnection::SrsTcpConnection(srs_netfd_t c) { - manager = cm; stfd = c; - ip = cip; - port = cport; - create_time = srsu2ms(srs_get_system_time()); - skt = new SrsStSocket(); - clk = new SrsWallClock(); - kbps = new SrsKbps(clk); - kbps->set_io(skt, skt); - - trd = new SrsSTCoroutine("conn", this); } SrsTcpConnection::~SrsTcpConnection() { - dispose(); - - srs_freep(kbps); - srs_freep(clk); srs_freep(skt); - srs_freep(trd); - srs_close_stfd(stfd); } -void SrsTcpConnection::remark(int64_t* in, int64_t* out) -{ - kbps->remark(in, out); -} - -void SrsTcpConnection::dispose() -{ - trd->interrupt(); -} - -srs_error_t SrsTcpConnection::start() +srs_error_t SrsTcpConnection::initialize() { srs_error_t err = srs_success; - + if ((err = skt->initialize(stfd)) != 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 SrsTcpConnection::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); @@ -408,168 +378,13 @@ srs_error_t SrsTcpConnection::set_tcp_nodelay(bool v) 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 SrsTcpConnection::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; -} - -srs_error_t SrsTcpConnection::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 SrsTcpConnection::remote_ip() -{ - return ip; -} - -const SrsContextId& SrsTcpConnection::get_id() -{ - return trd->cid(); -} - -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; @@ -621,52 +436,52 @@ srs_error_t SrsTcpConnection2::set_socket_buffer(srs_utime_t buffer_v) return err; } -void SrsTcpConnection2::set_recv_timeout(srs_utime_t tm) +void SrsTcpConnection::set_recv_timeout(srs_utime_t tm) { skt->set_recv_timeout(tm); } -srs_utime_t SrsTcpConnection2::get_recv_timeout() +srs_utime_t SrsTcpConnection::get_recv_timeout() { return skt->get_recv_timeout(); } -srs_error_t SrsTcpConnection2::read_fully(void* buf, size_t size, ssize_t* nread) +srs_error_t SrsTcpConnection::read_fully(void* buf, size_t size, ssize_t* nread) { return skt->read_fully(buf, size, nread); } -int64_t SrsTcpConnection2::get_recv_bytes() +int64_t SrsTcpConnection::get_recv_bytes() { return skt->get_recv_bytes(); } -int64_t SrsTcpConnection2::get_send_bytes() +int64_t SrsTcpConnection::get_send_bytes() { return skt->get_send_bytes(); } -srs_error_t SrsTcpConnection2::read(void* buf, size_t size, ssize_t* nread) +srs_error_t SrsTcpConnection::read(void* buf, size_t size, ssize_t* nread) { return skt->read(buf, size, nread); } -void SrsTcpConnection2::set_send_timeout(srs_utime_t tm) +void SrsTcpConnection::set_send_timeout(srs_utime_t tm) { skt->set_send_timeout(tm); } -srs_utime_t SrsTcpConnection2::get_send_timeout() +srs_utime_t SrsTcpConnection::get_send_timeout() { return skt->get_send_timeout(); } -srs_error_t SrsTcpConnection2::write(void* buf, size_t size, ssize_t* nwrite) +srs_error_t SrsTcpConnection::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) +srs_error_t SrsTcpConnection::writev(const iovec *iov, int iov_size, ssize_t* nwrite) { return skt->writev(iov, iov_size, nwrite); } diff --git a/trunk/src/app/srs_app_conn.hpp b/trunk/src/app/srs_app_conn.hpp index ac0619fad..a338046ed 100644 --- a/trunk/src/app/srs_app_conn.hpp +++ b/trunk/src/app/srs_app_conn.hpp @@ -130,77 +130,7 @@ public: // 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 SrsTcpConnection : virtual public ISrsStartableConneciton - , virtual public ISrsReloadHandler, virtual public ISrsCoroutineHandler, virtual public ISrsExpire -{ -protected: - // Each connection start a green thread, - // when thread stop, the connection will be delete by server. - SrsCoroutine* trd; - // The manager object to manage the connection. - ISrsResourceManager* manager; - // The underlayer st fd handler. - srs_netfd_t stfd; - // The ip and port of client. - std::string ip; - int port; - // The underlayer socket. - SrsStSocket* skt; - // The connection total kbps. - // not only the rtmp or http connection, all type of connection are - // need to statistic the kbps of io. - // The SrsStatistic will use it indirectly to statistic the bytes delta of current connection. - SrsKbps* kbps; - SrsWallClock* clk; - // The create time in milliseconds. - // for current connection to log self create time and calculate the living time. - int64_t create_time; -public: - SrsTcpConnection(ISrsResourceManager* cm, srs_netfd_t c, std::string cip, int cport); - virtual ~SrsTcpConnection(); -// Interface ISrsKbpsDelta -public: - virtual void remark(int64_t* in, int64_t* out); -public: - // To dipose the connection. - virtual void dispose(); -// Interface ISrsStartable -public: - // Start the client green thread. - // when server get a client from listener, - // 1. server will create an concrete connection(for instance, RTMP connection), - // 2. then add connection to its connection manager, - // 3. start the client thread by invoke this start() - // when client cycle thread stop, invoke the on_thread_stop(), which will use server - // To remove the client by server->remove(this). - virtual srs_error_t start(); -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 ISrsOneCycleThreadHandler -public: - // The thread cycle function, - // when serve connection completed, terminate the loop which will terminate the thread, - // thread will invoke the on_thread_stop() when it terminated. - virtual srs_error_t cycle(); -// Interface ISrsConnection. -public: - virtual std::string remote_ip(); - virtual const SrsContextId& get_id(); -public: - // Set connection to expired. - virtual void expire(); -protected: - // For concrete connection to do the cycle. - 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 +class SrsTcpConnection : virtual public ISrsProtocolReadWriter { private: // The underlayer st fd handler. @@ -208,8 +138,8 @@ private: // The underlayer socket. SrsStSocket* skt; public: - SrsTcpConnection2(srs_netfd_t c); - virtual ~SrsTcpConnection2(); + SrsTcpConnection(srs_netfd_t c); + virtual ~SrsTcpConnection(); public: virtual srs_error_t initialize(); public: diff --git a/trunk/src/app/srs_app_http_api.cpp b/trunk/src/app/srs_app_http_api.cpp index 7bea05f90..c7f9ce21e 100644 --- a/trunk/src/app/srs_app_http_api.cpp +++ b/trunk/src/app/srs_app_http_api.cpp @@ -1680,7 +1680,7 @@ SrsHttpApi::SrsHttpApi(ISrsResourceManager* cm, srs_netfd_t fd, SrsHttpServeMux* cors = new SrsHttpCorsMux(); parser = new SrsHttpParser(); - skt = new SrsTcpConnection2(fd); + skt = new SrsTcpConnection(fd); manager = cm; ip = cip; port = cport; diff --git a/trunk/src/app/srs_app_http_api.hpp b/trunk/src/app/srs_app_http_api.hpp index ae4cb63d2..f4cf06670 100644 --- a/trunk/src/app/srs_app_http_api.hpp +++ b/trunk/src/app/srs_app_http_api.hpp @@ -263,7 +263,7 @@ private: SrsHttpCorsMux* cors; SrsHttpServeMux* mux; private: - SrsTcpConnection2* skt; + SrsTcpConnection* skt; // Each connection start a green thread, // when thread stop, the connection will be delete by server. SrsCoroutine* trd; diff --git a/trunk/src/app/srs_app_http_conn.cpp b/trunk/src/app/srs_app_http_conn.cpp index d2161ed8c..978e42973 100644 --- a/trunk/src/app/srs_app_http_conn.cpp +++ b/trunk/src/app/srs_app_http_conn.cpp @@ -65,7 +65,7 @@ SrsHttpConn::SrsHttpConn(ISrsResourceManager* cm, srs_netfd_t fd, ISrsHttpServeM cors = new SrsHttpCorsMux(); http_mux = m; - skt = new SrsTcpConnection2(fd); + skt = new SrsTcpConnection(fd); manager = cm; ip = cip; port = cport; diff --git a/trunk/src/app/srs_app_http_conn.hpp b/trunk/src/app/srs_app_http_conn.hpp index 234db8027..d86593d33 100644 --- a/trunk/src/app/srs_app_http_conn.hpp +++ b/trunk/src/app/srs_app_http_conn.hpp @@ -63,7 +63,7 @@ protected: ISrsHttpServeMux* http_mux; SrsHttpCorsMux* cors; protected: - SrsTcpConnection2* skt; + SrsTcpConnection* skt; // Each connection start a green thread, // when thread stop, the connection will be delete by server. SrsCoroutine* trd; diff --git a/trunk/src/app/srs_app_rtc_source.hpp b/trunk/src/app/srs_app_rtc_source.hpp index 7a2f126c5..6eaf89c2b 100644 --- a/trunk/src/app/srs_app_rtc_source.hpp +++ b/trunk/src/app/srs_app_rtc_source.hpp @@ -38,7 +38,6 @@ #include class SrsRequest; -class SrsTcpConnection; class SrsMetaCache; class SrsSharedPtrMessage; class SrsCommonMessage; diff --git a/trunk/src/app/srs_app_rtmp_conn.cpp b/trunk/src/app/srs_app_rtmp_conn.cpp index b487b2d59..8a8a00c31 100644 --- a/trunk/src/app/srs_app_rtmp_conn.cpp +++ b/trunk/src/app/srs_app_rtmp_conn.cpp @@ -109,7 +109,7 @@ SrsRtmpConn::SrsRtmpConn(SrsServer* svr, srs_netfd_t c, string cip, int cport) server = svr; stfd = c; - skt = new SrsTcpConnection2(c); + skt = new SrsTcpConnection(c); manager = svr; ip = cip; port = cport; diff --git a/trunk/src/app/srs_app_rtmp_conn.hpp b/trunk/src/app/srs_app_rtmp_conn.hpp index 138dd1ab4..1151378b7 100644 --- a/trunk/src/app/srs_app_rtmp_conn.hpp +++ b/trunk/src/app/srs_app_rtmp_conn.hpp @@ -119,7 +119,7 @@ private: SrsClientInfo* info; private: srs_netfd_t stfd; - SrsTcpConnection2* skt; + SrsTcpConnection* skt; // Each connection start a green thread, // when thread stop, the connection will be delete by server. SrsCoroutine* trd; diff --git a/trunk/src/app/srs_app_server.hpp b/trunk/src/app/srs_app_server.hpp index 11315bf35..c60205a54 100644 --- a/trunk/src/app/srs_app_server.hpp +++ b/trunk/src/app/srs_app_server.hpp @@ -40,7 +40,6 @@ #include class SrsServer; -class SrsTcpConnection; class SrsHttpServeMux; class SrsHttpServer; class SrsIngester;