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

Merge branch '4.0release' into xialixin-dev-28181

Browse source
This commit is contained in:
winlin 2020-03-31 20:09:07 +08:00
commit a342f460e7
29 changed files with 654 additions and 203 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -198,19 +198,32 @@ srs_error_t SrsResponseOnlyHttpConn::pop_message(ISrsHttpMessage** preq)
srs_error_t err = srs_success;
SrsStSocket skt;
if ((err = skt.initialize(stfd)) != srs_success) {
return srs_error_wrap(err, "init socket");
}
if ((err = parser->parse_message(&skt, preq)) != srs_success) {
return srs_error_wrap(err, "parse message");
// Check user interrupt by interval.
skt.set_recv_timeout(3 * SRS_UTIME_SECONDS);
// drop all request body.
char body[4096];
while (true) {
if ((err = trd->pull()) != srs_success) {
return srs_error_wrap(err, "timeout");
}
if ((err = skt.read(body, 4096, NULL)) != srs_success) {
// Because we use timeout to check trd state, so we should ignore any timeout.
if (srs_error_code(err) == ERROR_SOCKET_TIMEOUT) {
srs_freep(err);
continue;
}
return srs_error_wrap(err, "read response");
}
}
// Attach owner connection to message.
SrsHttpMessage* hreq = (SrsHttpMessage*)(*preq);
hreq->set_connection(this);
return err;
}
@ -219,12 +232,12 @@ srs_error_t SrsResponseOnlyHttpConn::on_got_http_message(ISrsHttpMessage* msg)
srs_error_t err = srs_success;
ISrsHttpResponseReader* br = msg->body_reader();
// when not specified the content length, ignore.
if (msg->content_length() == -1) {
return err;
}
// drop all request body.
char body[4096];
while (!br->eof()) {
@ -236,6 +249,11 @@ srs_error_t SrsResponseOnlyHttpConn::on_got_http_message(ISrsHttpMessage* msg)
return err;
}
void SrsResponseOnlyHttpConn::expire()
{
SrsHttpConn::expire();
}
SrsHttpServer::SrsHttpServer(SrsServer* svr)
{
server = svr;

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

@ -101,6 +101,9 @@ public:
virtual srs_error_t pop_message(ISrsHttpMessage** preq);
public:
virtual srs_error_t on_got_http_message(ISrsHttpMessage* msg);
public:
// Set connection to expired.
virtual void expire();
};
// The http server, use http stream or static server to serve requests.

12
trunk/src/app/srs_app_http_stream.cpp
View file

@ -592,10 +592,15 @@ srs_error_t SrsLiveStream::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMess
SrsAutoFree(SrsPithyPrint, pprint);
SrsMessageArray msgs(SRS_PERF_MW_MSGS);
// 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());
// update the statistic when source disconveried.
SrsStatistic* stat = SrsStatistic::instance();
if ((err = stat->on_client(_srs_context->get_id(), req, NULL, SrsRtmpConnPlay)) != srs_success) {
if ((err = stat->on_client(_srs_context->get_id(), req, hc, SrsRtmpConnPlay)) != srs_success) {
return srs_error_wrap(err, "stat on client");
}
@ -613,11 +618,6 @@ srs_error_t SrsLiveStream::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMess
}
SrsFlvStreamEncoder* ffe = dynamic_cast<SrsFlvStreamEncoder*>(enc);
// 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());
// Set the socket options for transport.
bool tcp_nodelay = _srs_config->get_tcp_nodelay(req->vhost);

8
trunk/src/app/srs_app_server.cpp
View file

@ -30,7 +30,9 @@
#include <fcntl.h>
#include <unistd.h>
#include <algorithm>
#ifndef SRS_AUTO_OSX
#include <sys/inotify.h>
#endif
using namespace std;
#include <srs_kernel_log.hpp>
@ -509,6 +511,7 @@ srs_error_t SrsInotifyWorker::start()
{
srs_error_t err = srs_success;
#ifndef SRS_AUTO_OSX
// Whether enable auto reload config.
bool auto_reload = _srs_config->inotify_auto_reload();
if (!auto_reload && _srs_in_docker && _srs_config->auto_reload_for_docker()) {
@ -579,6 +582,7 @@ srs_error_t SrsInotifyWorker::start()
if ((err = trd->start()) != srs_success) {
return srs_error_wrap(err, "inotify");
}
#endif
return err;
}
@ -587,6 +591,7 @@ srs_error_t SrsInotifyWorker::cycle()
{
srs_error_t err = srs_success;
#ifndef SRS_AUTO_OSX
string config_path = _srs_config->config();
string config_file = srs_path_basename(config_path);
string k8s_file = "..data";
@ -626,6 +631,7 @@ srs_error_t SrsInotifyWorker::cycle()
srs_usleep(3000 * SRS_UTIME_MILLISECONDS);
}
#endif
return err;
}
@ -792,7 +798,7 @@ srs_error_t SrsServer::initialize(ISrsServerCycle* ch)
srs_error_t SrsServer::initialize_st()
{
srs_error_t err = srs_success;
// @remark, st alloc segment use mmap, which only support 32757 threads,
// if need to support more, for instance, 100k threads, define the macro MALLOC_STACK.
// TODO: FIXME: maybe can use "sysctl vm.max_map_count" to refine.

82
trunk/src/app/srs_app_utility.cpp
View file

@ -35,6 +35,9 @@
#include <sys/time.h>
#include <math.h>
#include <map>
#ifdef SRS_AUTO_OSX
#include <sys/sysctl.h>
#endif
using namespace std;
#include <srs_kernel_log.hpp>
@ -326,6 +329,7 @@ SrsProcSystemStat* srs_get_system_proc_stat()
bool get_proc_system_stat(SrsProcSystemStat& r)
{
#ifndef SRS_AUTO_OSX
FILE* f = fopen("/proc/stat", "r");
if (f == NULL) {
srs_warn("open system cpu stat failed, ignore");
@ -355,6 +359,7 @@ bool get_proc_system_stat(SrsProcSystemStat& r)
}
fclose(f);
#endif
r.ok = true;
@ -363,6 +368,7 @@ bool get_proc_system_stat(SrsProcSystemStat& r)
bool get_proc_self_stat(SrsProcSelfStat& r)
{
#ifndef SRS_AUTO_OSX
FILE* f = fopen("/proc/self/stat", "r");
if (f == NULL) {
srs_warn("open self cpu stat failed, ignore");
@ -389,6 +395,7 @@ bool get_proc_self_stat(SrsProcSelfStat& r)
&r.guest_time, &r.cguest_time);
fclose(f);
#endif
r.ok = true;
@ -484,6 +491,7 @@ SrsDiskStat* srs_get_disk_stat()
bool srs_get_disk_vmstat_stat(SrsDiskStat& r)
{
#ifndef SRS_AUTO_OSX
FILE* f = fopen("/proc/vmstat", "r");
if (f == NULL) {
srs_warn("open vmstat failed, ignore");
@ -503,6 +511,7 @@ bool srs_get_disk_vmstat_stat(SrsDiskStat& r)
}
fclose(f);
#endif
r.ok = true;
@ -513,13 +522,14 @@ bool srs_get_disk_diskstats_stat(SrsDiskStat& r)
{
r.ok = true;
r.sample_time = srsu2ms(srs_get_system_time());
#ifndef SRS_AUTO_OSX
// if disabled, ignore all devices.
SrsConfDirective* conf = _srs_config->get_stats_disk_device();
if (conf == NULL) {
return true;
}
FILE* f = fopen("/proc/diskstats", "r");
if (f == NULL) {
srs_warn("open vmstat failed, ignore");
@ -584,6 +594,7 @@ bool srs_get_disk_diskstats_stat(SrsDiskStat& r)
}
fclose(f);
#endif
r.ok = true;
@ -675,7 +686,8 @@ SrsMemInfo* srs_get_meminfo()
void srs_update_meminfo()
{
SrsMemInfo& r = _srs_system_meminfo;
#ifndef SRS_AUTO_OSX
FILE* f = fopen("/proc/meminfo", "r");
if (f == NULL) {
srs_warn("open meminfo failed, ignore");
@ -701,6 +713,7 @@ void srs_update_meminfo()
}
fclose(f);
#endif
r.sample_time = srsu2ms(srs_get_system_time());
r.MemActive = r.MemTotal - r.MemFree;
@ -767,7 +780,8 @@ void srs_update_platform_info()
SrsPlatformInfo& r = _srs_system_platform_info;
r.srs_startup_time = srsu2ms(srs_get_system_startup_time());
#ifndef SRS_AUTO_OSX
if (true) {
FILE* f = fopen("/proc/uptime", "r");
if (f == NULL) {
@ -796,7 +810,44 @@ void srs_update_platform_info()
fclose(f);
}
#else
// man 3 sysctl
if (true) {
struct timeval tv;
size_t len = sizeof(timeval);
int mib[2];
mib[0] = CTL_KERN;
mib[1] = KERN_BOOTTIME;
if (sysctl(mib, 2, &tv, &len, NULL, 0) < 0) {
srs_warn("sysctl boottime failed, ignore");
return;
}
time_t bsec = tv.tv_sec;
time_t csec = ::time(NULL);
r.os_uptime = difftime(csec, bsec);
}
// man 3 sysctl
if (true) {
struct loadavg la;
size_t len = sizeof(loadavg);
int mib[2];
mib[0] = CTL_VM;
mib[1] = VM_LOADAVG;
if (sysctl(mib, 2, &la, &len, NULL, 0) < 0) {
srs_warn("sysctl loadavg failed, ignore");
return;
}
r.load_one_minutes = (double)la.ldavg[0] / la.fscale;
r.load_five_minutes = (double)la.ldavg[1] / la.fscale;
r.load_fifteen_minutes = (double)la.ldavg[2] / la.fscale;
}
#endif
r.ok = true;
}
@ -842,6 +893,7 @@ int srs_get_network_devices_count()
void srs_update_network_devices()
{
#ifndef SRS_AUTO_OSX
if (true) {
FILE* f = fopen("/proc/net/dev", "r");
if (f == NULL) {
@ -878,6 +930,7 @@ void srs_update_network_devices()
fclose(f);
}
#endif
}
SrsNetworkRtmpServer::SrsNetworkRtmpServer()
@ -924,7 +977,8 @@ void srs_update_rtmp_server(int nb_conn, SrsKbps* kbps)
int nb_tcp_total = 0;
int nb_tcp_mem = 0;
int nb_udp4 = 0;
#ifndef SRS_AUTO_OSX
if (true) {
FILE* f = fopen("/proc/net/sockstat", "r");
if (f == NULL) {
@ -954,9 +1008,20 @@ void srs_update_rtmp_server(int nb_conn, SrsKbps* kbps)
fclose(f);
}
#else
// TODO: FIXME: impelments it.
nb_socks = 0;
nb_tcp4_hashed = 0;
nb_tcp_orphans = 0;
nb_tcp_tws = 0;
nb_tcp_total = 0;
nb_tcp_mem = 0;
nb_udp4 = 0;
#endif
int nb_tcp_estab = 0;
#ifndef SRS_AUTO_OSX
if (true) {
FILE* f = fopen("/proc/net/snmp", "r");
if (f == NULL) {
@ -986,6 +1051,7 @@ void srs_update_rtmp_server(int nb_conn, SrsKbps* kbps)
fclose(f);
}
#endif
// @see: https://github.com/shemminger/iproute2/blob/master/misc/ss.c
// TODO: FIXME: ignore the slabstat, @see: get_slabstat()

2
trunk/src/core/srs_core_version3.hpp
View file

@ -24,6 +24,6 @@
#ifndef SRS_CORE_VERSION3_HPP
#define SRS_CORE_VERSION3_HPP
#define SRS_VERSION3_REVISION 134
#define SRS_VERSION3_REVISION 139
#endif

21
trunk/src/kernel/srs_kernel_utility.cpp
View file

@ -124,7 +124,9 @@ srs_utime_t srs_get_system_startup_time()
}
// For utest to mock it.
#ifndef SRS_AUTO_OSX
_srs_gettimeofday_t _srs_gettimeofday = ::gettimeofday;
#endif
srs_utime_t srs_update_system_time()
{
@ -652,15 +654,20 @@ bool srs_path_exists(std::string path)
string srs_path_dirname(string path)
{
std::string dirname = path;
// No slash, it must be current dir.
size_t pos = string::npos;
if ((pos = dirname.rfind("/")) != string::npos) {
if (pos == 0) {
return "/";
}
dirname = dirname.substr(0, pos);
if ((pos = dirname.rfind("/")) == string::npos) {
return "./";
}
// Path under root.
if (pos == 0) {
return "/";
}
// Fetch the directory.
dirname = dirname.substr(0, pos);
return dirname;
}

6
trunk/src/kernel/srs_kernel_utility.hpp
View file

@ -166,7 +166,11 @@ extern int srs_chunk_header_c3(int perfer_cid, uint32_t timestamp, char* cache,
// For utest to mock it.
#include <sys/time.h>
typedef int (*_srs_gettimeofday_t)(struct timeval* tv, struct timezone* tz);
#ifdef SRS_AUTO_OSX
#define _srs_gettimeofday gettimeofday
#else
typedef int (*_srs_gettimeofday_t) (struct timeval* tv, struct timezone* tz);
#endif
#endif

8
trunk/src/main/srs_main_server.cpp
View file

@ -127,8 +127,8 @@ srs_error_t do_main(int argc, char** argv)
srs_trace("%s, %s", RTMP_SIG_SRS_SERVER, RTMP_SIG_SRS_LICENSE);
srs_trace("authors: %s", RTMP_SIG_SRS_AUTHORS);
srs_trace("contributors: %s", SRS_AUTO_CONSTRIBUTORS);
srs_trace("cwd=%s, work_dir=%s, build: %s, configure: %s, uname: %s",
_srs_config->cwd().c_str(), cwd.c_str(), SRS_AUTO_BUILD_DATE, SRS_AUTO_USER_CONFIGURE, SRS_AUTO_UNAME);
srs_trace("cwd=%s, work_dir=%s, build: %s, configure: %s, uname: %s, osx: %d",
_srs_config->cwd().c_str(), cwd.c_str(), SRS_AUTO_BUILD_DATE, SRS_AUTO_USER_CONFIGURE, SRS_AUTO_UNAME, SRS_AUTO_OSX_BOOL);
srs_trace("configure detail: " SRS_AUTO_CONFIGURE);
#ifdef SRS_AUTO_EMBEDED_TOOL_CHAIN
srs_trace("crossbuild tool chain: " SRS_AUTO_EMBEDED_TOOL_CHAIN);
@ -335,8 +335,8 @@ void show_macro_features()
#endif
#if VERSION_MAJOR > VERSION_STABLE
#warning "Current branch is unstable."
srs_warn("Develop is unstable, please use branch: git checkout -b %s origin/%s", VERSION_STABLE_BRANCH, VERSION_STABLE_BRANCH);
#warning "Current branch is beta."
srs_warn("%s/%s is beta", RTMP_SIG_SRS_KEY, RTMP_SIG_SRS_VERSION);
#endif
#if defined(SRS_PERF_SO_SNDBUF_SIZE) && !defined(SRS_PERF_MW_SO_SNDBUF)

10
trunk/src/protocol/srs_protocol_utility.cpp
View file

@ -336,21 +336,25 @@ srs_error_t srs_write_large_iovs(ISrsProtocolReadWriter* skt, iovec* iovs, int s
#endif
// send in a time.
if (size < limits) {
if (size <= limits) {
if ((err = skt->writev(iovs, size, pnwrite)) != srs_success) {
return srs_error_wrap(err, "writev");
}
return err;
}
// send in multiple times.
int cur_iov = 0;
ssize_t nwrite = 0;
while (cur_iov < size) {
int cur_count = srs_min(limits, size - cur_iov);
if ((err = skt->writev(iovs + cur_iov, cur_count, pnwrite)) != srs_success) {
if ((err = skt->writev(iovs + cur_iov, cur_count, &nwrite)) != srs_success) {
return srs_error_wrap(err, "writev");
}
cur_iov += cur_count;
if (pnwrite) {
*pnwrite += nwrite;
}
}
return err;

2
trunk/src/service/srs_service_log.cpp
View file

@ -46,7 +46,7 @@ int SrsThreadContext::generate_id()
static int id = 0;
if (id == 0) {
id = (100 + ((int)(int64_t)this)%1000);
id = (100 + ((uint32_t)(int64_t)this)%1000);
}
int gid = id++;

4
trunk/src/utest/srs_utest_kernel.cpp
View file

@ -2565,7 +2565,7 @@ VOID TEST(KernelUtility, BytesUtils)
VOID TEST(KernelUtility, PathUtils)
{
if (true) {
EXPECT_TRUE("" == srs_path_dirname(""));
EXPECT_TRUE("./" == srs_path_dirname(""));
EXPECT_TRUE("/" == srs_path_dirname("/"));
EXPECT_TRUE("/" == srs_path_dirname("//"));
EXPECT_TRUE("/" == srs_path_dirname("/stream"));
@ -4208,6 +4208,7 @@ VOID TEST(KernelUtilityTest, CoverBitsBufferAll)
}
}
#ifndef SRS_AUTO_OSX
extern _srs_gettimeofday_t _srs_gettimeofday;
int mock_gettimeofday(struct timeval* /*tp*/, struct timezone* /*tzp*/) {
return -1;
@ -4238,6 +4239,7 @@ VOID TEST(KernelUtilityTest, CoverTimeSpecial)
EXPECT_TRUE(-1 == srs_update_system_time());
}
}
#endif
extern int64_t _srs_system_time_startup_time;
extern int64_t _srs_system_time_us_cache;

64
trunk/src/utest/srs_utest_protocol.cpp
View file

@ -230,8 +230,6 @@ srs_error_t MockBufferIO::writev(const iovec *iov, int iov_size, ssize_t* nwrite
total += writen;
}
sbytes += total;
if (nwrite) {
*nwrite = total;
}
@ -6412,3 +6410,65 @@ VOID TEST(ProtocolKbpsTest, RAWStatistic)
}
}
VOID TEST(ProtocolKbpsTest, WriteLargeIOVs)
{
srs_error_t err;
if (true) {
iovec iovs[1];
iovs[0].iov_base = (char*)"Hello";
iovs[0].iov_len = 5;
MockBufferIO io;
ssize_t nn = 0;
HELPER_EXPECT_SUCCESS(srs_write_large_iovs(&io, iovs, 1, &nn));
EXPECT_EQ(5, nn);
EXPECT_EQ(5, io.sbytes);
}
if (true) {
iovec iovs[1024];
int nn_iovs = (int)(sizeof(iovs)/sizeof(iovec));
for (int i = 0; i < nn_iovs; i++) {
iovs[i].iov_base = (char*)"Hello";
iovs[i].iov_len = 5;
}
MockBufferIO io;
ssize_t nn = 0;
HELPER_EXPECT_SUCCESS(srs_write_large_iovs(&io, iovs, nn_iovs, &nn));
EXPECT_EQ(5 * nn_iovs, nn);
EXPECT_EQ(5 * nn_iovs, io.sbytes);
}
if (true) {
iovec iovs[1025];
int nn_iovs = (int)(sizeof(iovs)/sizeof(iovec));
for (int i = 0; i < nn_iovs; i++) {
iovs[i].iov_base = (char*)"Hello";
iovs[i].iov_len = 5;
}
MockBufferIO io;
ssize_t nn = 0;
HELPER_EXPECT_SUCCESS(srs_write_large_iovs(&io, iovs, nn_iovs, &nn));
EXPECT_EQ(5 * nn_iovs, nn);
EXPECT_EQ(5 * nn_iovs, io.sbytes);
}
if (true) {
iovec iovs[4096];
int nn_iovs = (int)(sizeof(iovs)/sizeof(iovec));
for (int i = 0; i < nn_iovs; i++) {
iovs[i].iov_base = (char*)"Hello";
iovs[i].iov_len = 5;
}
MockBufferIO io;
ssize_t nn = 0;
HELPER_EXPECT_SUCCESS(srs_write_large_iovs(&io, iovs, nn_iovs, &nn));
EXPECT_EQ(5 * nn_iovs, nn);
EXPECT_EQ(5 * nn_iovs, io.sbytes);
}
}