external/srs
1
0
Fork 0
mirror of https://github.com/ossrs/srs.git synced 2025-03-09 15:49:59 +00:00
Code Issues Releases Wiki Activity

For #1657: Refine TCP connections arch

Browse source
This commit is contained in:
winlin 2020-11-05 12:25:54 +08:00
parent 4ba66b388b
commit 2a14dc0ebd
10 changed files with 29 additions and 286 deletions
Download patch file Download diff file Expand all files Collapse all files

209
trunk/src/app/srs_app_conn.cpp
View file

@ -330,45 +330,19 @@ 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;
@ -376,10 +350,6 @@ srs_error_t SrsTcpConnection::start()
return srs_error_wrap(err, "init socket");
}
if ((err = trd->start()) != srs_success) {
return srs_error_wrap(err, "coroutine");
}
return err;
}
@ -466,207 +436,52 @@ srs_error_t SrsTcpConnection::set_socket_buffer(srs_utime_t 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;
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)
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);
}

76
trunk/src/app/srs_app_conn.hpp
View file

@ -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:

2
trunk/src/app/srs_app_http_api.cpp
View file

@ -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;

2
trunk/src/app/srs_app_http_api.hpp
View file

@ -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;

2
trunk/src/app/srs_app_http_conn.cpp
View file

@ -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;

2
trunk/src/app/srs_app_http_conn.hpp
View file

@ -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;

1
trunk/src/app/srs_app_rtc_source.hpp
View file

@ -38,7 +38,6 @@
#include <srs_app_source.hpp>
class SrsRequest;
class SrsTcpConnection;
class SrsMetaCache;
class SrsSharedPtrMessage;
class SrsCommonMessage;

2
trunk/src/app/srs_app_rtmp_conn.cpp
View file

@ -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;

2
trunk/src/app/srs_app_rtmp_conn.hpp
View file

@ -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;

1
trunk/src/app/srs_app_server.hpp
View file

@ -40,7 +40,6 @@
#include <srs_app_gb28181_sip.hpp>
class SrsServer;
class SrsTcpConnection;
class SrsHttpServeMux;
class SrsHttpServer;
class SrsIngester;