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

performance refine, support 3k+ connections(270kbps). 0.9.130

Browse source
This commit is contained in:
winlin 2014-06-22 20:01:25 +08:00
parent e9c96af91a
commit 1ae3e6c64c
18 changed files with 230 additions and 162 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1403,7 +1403,7 @@ int SrsConfig::get_max_connections()
SrsConfDirective* conf = root->get("max_connections");
if (!conf || conf->arg0().empty()) {
return 2000;
return SRS_CONF_DEFAULT_MAX_CONNECTIONS;
}
return ::atoi(conf->arg0().c_str());

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

@ -38,6 +38,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#define SRS_CONF_DEFAULT_PID_FILE "./objs/srs.pid"
#define SRS_DEFAULT_CONF "conf/srs.conf"
#define SRS_CONF_DEFAULT_MAX_CONNECTIONS 12345
#define SRS_CONF_DEFAULT_HLS_PATH "./objs/nginx/html"
#define SRS_CONF_DEFAULT_HLS_FRAGMENT 10
#define SRS_CONF_DEFAULT_HLS_WINDOW 60

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

@ -44,6 +44,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <srs_app_socket.hpp>
#include <srs_app_kbps.hpp>
#include <srs_kernel_utility.hpp>
#include <srs_protocol_msg_array.hpp>
// when error, edge ingester sleep for a while and retry.
#define SRS_EDGE_INGESTER_SLEEP_US (int64_t)(1*1000*1000LL)
@ -431,6 +432,7 @@ void SrsEdgeForwarder::stop()
kbps->set_io(NULL, NULL);
}
#define SYS_MAX_EDGE_SEND_MSGS 128
int SrsEdgeForwarder::cycle()
{
int ret = ERROR_SUCCESS;
@ -438,6 +440,8 @@ int SrsEdgeForwarder::cycle()
client->set_recv_timeout(SRS_PULSE_TIMEOUT_US);
SrsPithyPrint pithy_print(SRS_STAGE_EDGE);
SrsSharedPtrMessageArray msgs(SYS_MAX_EDGE_SEND_MSGS);
while (pthread->can_loop()) {
// switch to other st-threads.
@ -465,8 +469,7 @@ int SrsEdgeForwarder::cycle()
// forward all messages.
int count = 0;
SrsSharedPtrMessage** msgs = NULL;
if ((ret = queue->get_packets(0, msgs, count)) != ERROR_SUCCESS) {
if ((ret = queue->dump_packets(msgs.size, msgs.msgs, count)) != ERROR_SUCCESS) {
srs_error("get message to forward to origin failed. ret=%d", ret);
return ret;
}
@ -488,16 +491,15 @@ int SrsEdgeForwarder::cycle()
srs_verbose("no packets to forward.");
continue;
}
SrsAutoFreeArray(SrsSharedPtrMessage, msgs, count);
// all msgs to forward to origin.
// @remark, becareful, all msgs must be free explicitly,
// free by send_and_free_message or srs_freep.
for (int i = 0; i < count; i++) {
SrsSharedPtrMessage* msg = msgs[i];
SrsSharedPtrMessage* msg = msgs.msgs[i];
srs_assert(msg);
msgs[i] = NULL;
msgs.msgs[i] = NULL;
if ((ret = client->send_and_free_message(msg, stream_id)) != ERROR_SUCCESS) {
srs_error("edge publish forwarder send message to server failed. ret=%d", ret);

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

@ -41,6 +41,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <srs_protocol_rtmp.hpp>
#include <srs_app_kbps.hpp>
#include <srs_kernel_utility.hpp>
#include <srs_protocol_msg_array.hpp>
// when error, forwarder sleep for a while and retry.
#define SRS_FORWARDER_SLEEP_US (int64_t)(3*1000*1000LL)
@ -309,6 +310,7 @@ int SrsForwarder::connect_server()
return ret;
}
#define SYS_MAX_FORWARD_SEND_MSGS 128
int SrsForwarder::forward()
{
int ret = ERROR_SUCCESS;
@ -317,6 +319,8 @@ int SrsForwarder::forward()
SrsPithyPrint pithy_print(SRS_STAGE_FORWARDER);
SrsSharedPtrMessageArray msgs(SYS_MAX_FORWARD_SEND_MSGS);
while (pthread->can_loop()) {
// switch to other st-threads.
st_usleep(0);
@ -339,8 +343,7 @@ int SrsForwarder::forward()
// forward all messages.
int count = 0;
SrsSharedPtrMessage** msgs = NULL;
if ((ret = queue->get_packets(0, msgs, count)) != ERROR_SUCCESS) {
if ((ret = queue->dump_packets(msgs.size, msgs.msgs, count)) != ERROR_SUCCESS) {
srs_error("get message to forward failed. ret=%d", ret);
return ret;
}
@ -360,16 +363,15 @@ int SrsForwarder::forward()
srs_verbose("no packets to forward.");
continue;
}
SrsAutoFreeArray(SrsSharedPtrMessage, msgs, count);
// all msgs to forward.
// @remark, becareful, all msgs must be free explicitly,
// free by send_and_free_message or srs_freep.
for (int i = 0; i < count; i++) {
SrsSharedPtrMessage* msg = msgs[i];
SrsSharedPtrMessage* msg = msgs.msgs[i];
srs_assert(msg);
msgs[i] = NULL;
msgs.msgs[i] = NULL;
if ((ret = client->send_and_free_message(msg, stream_id)) != ERROR_SUCCESS) {
srs_error("forwarder send message to server failed. ret=%d", ret);

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

@ -50,6 +50,7 @@ using namespace std;
#include <srs_app_utility.hpp>
#include <srs_protocol_utility.hpp>
#include <srs_kernel_utility.hpp>
#include <srs_protocol_msg_array.hpp>
// when stream is busy, for example, streaming is already
// publishing, when a new client to request to publish,
@ -382,7 +383,7 @@ int SrsRtmpConn::stream_service_cycle()
}
srs_info("start to publish stream %s success", req->stream.c_str());
ret = fmle_publish(source);
ret = fmle_publishing(source);
// when edge, notice edge to change state.
// when origin, notice all service to unpublish.
@ -416,7 +417,7 @@ int SrsRtmpConn::stream_service_cycle()
}
srs_info("flash start to publish stream %s success", req->stream.c_str());
ret = flash_publish(source);
ret = flash_publishing(source);
// when edge, notice edge to change state.
// when origin, notice all service to unpublish.
@ -476,6 +477,8 @@ int SrsRtmpConn::check_vhost()
return ret;
}
#define SYS_MAX_PLAY_SEND_MSGS 128
int SrsRtmpConn::playing(SrsSource* source)
{
int ret = ERROR_SUCCESS;
@ -499,38 +502,43 @@ int SrsRtmpConn::playing(SrsSource* source)
rtmp->set_recv_timeout(SRS_PULSE_TIMEOUT_US);
SrsPithyPrint pithy_print(SRS_STAGE_PLAY_USER);
SrsSharedPtrMessageArray msgs(SYS_MAX_PLAY_SEND_MSGS);
bool user_specified_duration_to_stop = (req->duration > 0);
int64_t starttime = -1;
while (true) {
// switch to other st-threads.
st_usleep(0);
// collect elapse for pithy print.
pithy_print.elapse();
// read from client.
if (true) {
SrsMessage* msg = NULL;
ret = rtmp->recv_message(&msg);
srs_verbose("play loop recv message. ret=%d", ret);
if (ret != ERROR_SUCCESS && ret != ERROR_SOCKET_TIMEOUT) {
if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) {
if (ret == ERROR_SOCKET_TIMEOUT) {
// it's ok, do nothing.
ret = ERROR_SUCCESS;
} else if (ret != ERROR_SUCCESS) {
if (!srs_is_client_gracefully_close(ret)) {
srs_error("recv client control message failed. ret=%d", ret);
}
return ret;
}
if ((ret = process_play_control_msg(consumer, msg)) != ERROR_SUCCESS) {
if (!srs_is_system_control_error(ret)) {
srs_error("process play control message failed. ret=%d", ret);
} else {
if ((ret = process_play_control_msg(consumer, msg)) != ERROR_SUCCESS) {
if (!srs_is_system_control_error(ret)) {
srs_error("process play control message failed. ret=%d", ret);
}
return ret;
}
return ret;
}
}
// get messages from consumer.
SrsSharedPtrMessage** msgs = NULL;
int count = 0;
if ((ret = consumer->get_packets(0, msgs, count)) != ERROR_SUCCESS) {
if ((ret = consumer->dump_packets(msgs.size, msgs.msgs, count)) != ERROR_SUCCESS) {
srs_error("get messages from consumer failed. ret=%d", ret);
return ret;
}
@ -545,32 +553,29 @@ int SrsRtmpConn::playing(SrsSource* source)
kbps->get_recv_kbps(), kbps->get_recv_kbps_30s(), kbps->get_recv_kbps_5m());
}
if (count <= 0) {
srs_verbose("no packets in queue.");
continue;
}
SrsAutoFreeArray(SrsSharedPtrMessage, msgs, count);
// sendout messages
// @remark, becareful, all msgs must be free explicitly,
// free by send_and_free_message or srs_freep.
for (int i = 0; i < count; i++) {
SrsSharedPtrMessage* msg = msgs[i];
SrsSharedPtrMessage* msg = msgs.msgs[i];
// the send_message will free the msg,
// so set the msgs[i] to NULL.
msgs[i] = NULL;
msgs.msgs[i] = NULL;
srs_assert(msg);
// foreach msg, collect the duration.
// @remark: never use msg when sent it, for the protocol sdk will free it.
if (starttime < 0 || starttime > msg->header.timestamp) {
// only when user specifies the duration,
// we start to collect the durations for each message.
if (user_specified_duration_to_stop) {
// foreach msg, collect the duration.
// @remark: never use msg when sent it, for the protocol sdk will free it.
if (starttime < 0 || starttime > msg->header.timestamp) {
starttime = msg->header.timestamp;
}
duration += msg->header.timestamp - starttime;
starttime = msg->header.timestamp;
}
duration += msg->header.timestamp - starttime;
starttime = msg->header.timestamp;
// no need to assert msg, for the rtmp will assert it.
if ((ret = rtmp->send_and_free_message(msg, res->stream_id)) != ERROR_SUCCESS) {
srs_error("send message to client failed. ret=%d", ret);
return ret;
@ -579,17 +584,22 @@ int SrsRtmpConn::playing(SrsSource* source)
// if duration specified, and exceed it, stop play live.
// @see: https://github.com/winlinvip/simple-rtmp-server/issues/45
if (req->duration > 0 && duration >= (int64_t)req->duration) {
ret = ERROR_RTMP_DURATION_EXCEED;
srs_trace("stop live for duration exceed. ret=%d", ret);
return ret;
if (user_specified_duration_to_stop) {
if (duration >= (int64_t)req->duration) {
ret = ERROR_RTMP_DURATION_EXCEED;
srs_trace("stop live for duration exceed. ret=%d", ret);
return ret;
}
}
// switch to other threads, to anti dead loop.
st_usleep(0);
}
return ret;
}
int SrsRtmpConn::fmle_publish(SrsSource* source)
int SrsRtmpConn::fmle_publishing(SrsSource* source)
{
int ret = ERROR_SUCCESS;
@ -668,7 +678,7 @@ int SrsRtmpConn::fmle_publish(SrsSource* source)
return ret;
}
int SrsRtmpConn::flash_publish(SrsSource* source)
int SrsRtmpConn::flash_publishing(SrsSource* source)
{
int ret = ERROR_SUCCESS;

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

@ -48,6 +48,7 @@ class SrsHttpHooks;
class SrsBandwidth;
class SrsKbps;
class SrsRtmpClient;
class SrsSharedPtrMessage;
/**
* the client provides the main logic control for RTMP clients.
@ -87,8 +88,8 @@ private:
virtual int stream_service_cycle();
virtual int check_vhost();
virtual int playing(SrsSource* source);
virtual int fmle_publish(SrsSource* source);
virtual int flash_publish(SrsSource* source);
virtual int fmle_publishing(SrsSource* source);
virtual int flash_publishing(SrsSource* source);
virtual int process_publish_message(SrsSource* source, SrsMessage* msg, bool vhost_is_edge);
virtual int process_play_control_msg(SrsConsumer* consumer, SrsMessage* msg);
private:

66
trunk/src/app/srs_app_source.cpp
View file

@ -136,7 +136,7 @@ int SrsMessageQueue::enqueue(SrsSharedPtrMessage* msg)
{
int ret = ERROR_SUCCESS;
if (msg->header.is_video() || msg->header.is_audio()) {
if (msg->header.is_audio() || msg->header.is_video()) {
if (av_start_time == -1) {
av_start_time = msg->header.timestamp;
}
@ -153,7 +153,7 @@ int SrsMessageQueue::enqueue(SrsSharedPtrMessage* msg)
return ret;
}
int SrsMessageQueue::get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& count)
int SrsMessageQueue::dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& count)
{
int ret = ERROR_SUCCESS;
@ -161,17 +161,8 @@ int SrsMessageQueue::get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, in
return ret;
}
if (max_count == 0) {
count = (int)msgs.size();
} else {
count = srs_min(max_count, (int)msgs.size());
}
if (count <= 0) {
return ret;
}
pmsgs = new SrsSharedPtrMessage*[count];
srs_assert(max_count > 0);
count = srs_min(max_count, (int)msgs.size());
for (int i = 0; i < count; i++) {
pmsgs[i] = msgs[i];
@ -275,11 +266,11 @@ int SrsConsumer::get_time()
return jitter->get_time();
}
int SrsConsumer::enqueue(SrsSharedPtrMessage* msg, int tba, int tbv)
int SrsConsumer::enqueue(SrsSharedPtrMessage* msg, bool atc, int tba, int tbv)
{
int ret = ERROR_SUCCESS;
if (!source->is_atc()) {
if (!atc) {
if ((ret = jitter->correct(msg, tba, tbv)) != ERROR_SUCCESS) {
srs_freep(msg);
return ret;
@ -293,8 +284,10 @@ int SrsConsumer::enqueue(SrsSharedPtrMessage* msg, int tba, int tbv)
return ret;
}
int SrsConsumer::get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& count)
int SrsConsumer::dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& count)
{
srs_assert(max_count > 0);
if (should_update_source_id) {
srs_trace("update source_id=%d", source->source_id());
should_update_source_id = false;
@ -305,7 +298,7 @@ int SrsConsumer::get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& c
return ERROR_SUCCESS;
}
return queue->get_packets(max_count, pmsgs, count);
return queue->dump_packets(max_count, pmsgs, count);
}
int SrsConsumer::on_play_client_pause(bool is_pause)
@ -391,14 +384,15 @@ void SrsGopCache::clear()
cached_video_count = 0;
}
int SrsGopCache::dump(SrsConsumer* consumer, int tba, int tbv)
int SrsGopCache::dump(SrsConsumer* consumer, bool atc, int tba, int tbv)
{
int ret = ERROR_SUCCESS;
std::vector<SrsSharedPtrMessage*>::iterator it;
for (it = gop_cache.begin(); it != gop_cache.end(); ++it) {
SrsSharedPtrMessage* msg = *it;
if ((ret = consumer->enqueue(msg->copy(), tba, tbv)) != ERROR_SUCCESS) {
SrsSharedPtrMessage* copy = msg->copy();
if ((ret = consumer->enqueue(copy, atc, tba, tbv)) != ERROR_SUCCESS) {
srs_error("dispatch cached gop failed. ret=%d", ret);
return ret;
}
@ -926,7 +920,8 @@ int SrsSource::on_meta_data(SrsMessage* msg, SrsOnMetaDataPacket* metadata)
std::vector<SrsConsumer*>::iterator it;
for (it = consumers.begin(); it != consumers.end(); ++it) {
SrsConsumer* consumer = *it;
if ((ret = consumer->enqueue(cache_metadata->copy(), sample_rate, frame_rate)) != ERROR_SUCCESS) {
SrsSharedPtrMessage* copy = cache_metadata->copy();
if ((ret = consumer->enqueue(copy, atc, sample_rate, frame_rate)) != ERROR_SUCCESS) {
srs_error("dispatch the metadata failed. ret=%d", ret);
return ret;
}
@ -987,17 +982,17 @@ int SrsSource::on_audio(SrsMessage* audio)
// copy to all consumer
if (true) {
std::vector<SrsConsumer*>::iterator it;
for (it = consumers.begin(); it != consumers.end(); ++it) {
SrsConsumer* consumer = *it;
if ((ret = consumer->enqueue(msg->copy(), sample_rate, frame_rate)) != ERROR_SUCCESS) {
for (int i = 0; i < (int)consumers.size(); i++) {
SrsConsumer* consumer = consumers.at(i);
SrsSharedPtrMessage* copy = msg->copy();
if ((ret = consumer->enqueue(copy, atc, sample_rate, frame_rate)) != ERROR_SUCCESS) {
srs_error("dispatch the audio failed. ret=%d", ret);
return ret;
}
}
srs_info("dispatch audio success.");
}
// copy to all forwarders.
if (true) {
std::vector<SrsForwarder*>::iterator it;
@ -1077,10 +1072,10 @@ int SrsSource::on_video(SrsMessage* video)
// copy to all consumer
if (true) {
std::vector<SrsConsumer*>::iterator it;
for (it = consumers.begin(); it != consumers.end(); ++it) {
SrsConsumer* consumer = *it;
if ((ret = consumer->enqueue(msg->copy(), sample_rate, frame_rate)) != ERROR_SUCCESS) {
for (int i = 0; i < (int)consumers.size(); i++) {
SrsConsumer* consumer = consumers.at(i);
SrsSharedPtrMessage* copy = msg->copy();
if ((ret = consumer->enqueue(copy, atc, sample_rate, frame_rate)) != ERROR_SUCCESS) {
srs_error("dispatch the video failed. ret=%d", ret);
return ret;
}
@ -1327,27 +1322,27 @@ void SrsSource::on_unpublish()
}
// copy metadata.
if (cache_metadata && (ret = consumer->enqueue(cache_metadata->copy(), sample_rate, frame_rate)) != ERROR_SUCCESS) {
if (cache_metadata && (ret = consumer->enqueue(cache_metadata->copy(), atc, sample_rate, frame_rate)) != ERROR_SUCCESS) {
srs_error("dispatch metadata failed. ret=%d", ret);
return ret;
}
srs_info("dispatch metadata success");
// copy sequence header
if (cache_sh_video && (ret = consumer->enqueue(cache_sh_video->copy(), sample_rate, frame_rate)) != ERROR_SUCCESS) {
if (cache_sh_video && (ret = consumer->enqueue(cache_sh_video->copy(), atc, sample_rate, frame_rate)) != ERROR_SUCCESS) {
srs_error("dispatch video sequence header failed. ret=%d", ret);
return ret;
}
srs_info("dispatch video sequence header success");
if (cache_sh_audio && (ret = consumer->enqueue(cache_sh_audio->copy(), sample_rate, frame_rate)) != ERROR_SUCCESS) {
if (cache_sh_audio && (ret = consumer->enqueue(cache_sh_audio->copy(), atc, sample_rate, frame_rate)) != ERROR_SUCCESS) {
srs_error("dispatch audio sequence header failed. ret=%d", ret);
return ret;
}
srs_info("dispatch audio sequence header success");
// copy gop cache to client.
if ((ret = gop_cache->dump(consumer, sample_rate, frame_rate)) != ERROR_SUCCESS) {
if ((ret = gop_cache->dump(consumer, atc, sample_rate, frame_rate)) != ERROR_SUCCESS) {
return ret;
}
@ -1375,11 +1370,6 @@ void SrsSource::set_cache(bool enabled)
gop_cache->set(enabled);
}
bool SrsSource::is_atc()
{
return atc;
}
int SrsSource::on_edge_start_play()
{
return play_edge->on_client_play();

23
trunk/src/app/srs_app_source.hpp
View file

@ -110,11 +110,11 @@ public:
virtual int enqueue(SrsSharedPtrMessage* msg);
/**
* get packets in consumer queue.
* @pmsgs SrsMessages*[], output the prt array.
* @count the count in array.
* @max_count the max count to dequeue, 0 to dequeue all.
* @pmsgs SrsMessages*[], used to store the msgs, user must alloc it.
* @count the count in array, output param.
* @max_count the max count to dequeue, must be positive.
*/
virtual int get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& count);
virtual int dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& count);
private:
/**
* remove a gop from the front.
@ -155,19 +155,20 @@ public:
virtual int get_time();
/**
* enqueue an shared ptr message.
* @param whether atc, donot use jitter correct if true.
* @param tba timebase of audio.
* used to calc the audio time delta if time-jitter detected.
* @param tbv timebase of video.
* used to calc the video time delta if time-jitter detected.
*/
virtual int enqueue(SrsSharedPtrMessage* msg, int tba, int tbv);
virtual int enqueue(SrsSharedPtrMessage* msg, bool atc, int tba, int tbv);
/**
* get packets in consumer queue.
* @pmsgs SrsMessages*[], output the prt array.
* @count the count in array.
* @max_count the max count to dequeue, 0 to dequeue all.
* @pmsgs SrsMessages*[], used to store the msgs, user must alloc it.
* @count the count in array, output param.
* @max_count the max count to dequeue, must be positive.
*/
virtual int get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& count);
virtual int dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& count);
/**
* when client send the pause message.
*/
@ -208,7 +209,7 @@ public:
*/
virtual int cache(SrsSharedPtrMessage* msg);
virtual void clear();
virtual int dump(SrsConsumer* consumer, int tba, int tbv);
virtual int dump(SrsConsumer* consumer, bool atc, int tba, int tbv);
/**
* used for atc to get the time of gop cache,
* the atc will adjust the sequence header timestamp to gop cache.
@ -346,8 +347,6 @@ public:
virtual void set_cache(bool enabled);
// internal
public:
// for consumer, atc feature.
virtual bool is_atc();
// for edge, when play edge stream, check the state
virtual int on_edge_start_play();
// for edge, when publish edge stream, check the state