胡斌 / srs

转到一个项目
Toggle navigation Toggle navigation pinning
winlin
c3d98fd7 2 parents 0e2c8da2 b4c1d553

Merge branch 'srs.master'

内嵌 并排对比
正在显示 9 个修改的文件 包含 44 行增加 和 328 行删除
... ... @@ -388,8 +388,7 @@ if [ $SRS_EXPORT_LIBRTMP_PROJECT = NO ]; then
"srs_app_thread" "srs_app_bandwidth" "srs_app_st" "srs_app_log" "srs_app_config"
"srs_app_pithy_print" "srs_app_reload" "srs_app_http_api" "srs_app_http_conn" "srs_app_http_hooks"
"srs_app_json" "srs_app_ingest" "srs_app_ffmpeg" "srs_app_utility" "srs_app_dvr" "srs_app_edge"
"srs_app_kbps" "srs_app_heartbeat" "srs_app_empty" "srs_app_http_client" "srs_app_avc_aac"
"srs_app_pipe")
"srs_app_kbps" "srs_app_heartbeat" "srs_app_empty" "srs_app_http_client" "srs_app_avc_aac")
APP_INCS="src/app"; MODULE_DIR=${APP_INCS} . auto/modules.sh
APP_OBJS="${MODULE_OBJS[@]}"
fi
... ...
此 diff 太大无法显示。
/*
The MIT License (MIT)
Copyright (c) 2013-2014 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include <srs_app_pipe.hpp>
#include <unistd.h>
#include <srs_kernel_error.hpp>
#include <srs_kernel_log.hpp>
SrsPipe::SrsPipe()
{
fds[0] = fds[1] = 0;
read_stfd = write_stfd = NULL;
_already_written = false;
}
SrsPipe::~SrsPipe()
{
srs_close_stfd(read_stfd);
srs_close_stfd(write_stfd);
}
int SrsPipe::initialize()
{
int ret = ERROR_SUCCESS;
if (pipe(fds) < 0) {
ret = ERROR_SYSTEM_CREATE_PIPE;
srs_error("create pipe failed. ret=%d", ret);
return ret;
}
if ((read_stfd = st_netfd_open(fds[0])) == NULL) {
ret = ERROR_SYSTEM_CREATE_PIPE;
srs_error("open read pipe failed. ret=%d", ret);
return ret;
}
if ((write_stfd = st_netfd_open(fds[1])) == NULL) {
ret = ERROR_SYSTEM_CREATE_PIPE;
srs_error("open write pipe failed. ret=%d", ret);
return ret;
}
return ret;
}
st_netfd_t SrsPipe::rfd()
{
return read_stfd;
}
bool SrsPipe::already_written()
{
return _already_written;
}
int SrsPipe::active()
{
int ret = ERROR_SUCCESS;
int v = 0;
if (st_write(write_stfd, &v, sizeof(int), ST_UTIME_NO_TIMEOUT) != sizeof(int)) {
ret = ERROR_SYSTEM_WRITE_PIPE;
srs_error("write pipe failed. ret=%d", ret);
return ret;
}
_already_written = true;
return ret;
}
int SrsPipe::reset()
{
int ret = ERROR_SUCCESS;
int v;
if (st_read(read_stfd, &v, sizeof(int), ST_UTIME_NO_TIMEOUT) != sizeof(int)) {
ret = ERROR_SYSTEM_READ_PIPE;
srs_error("read pipe failed. ret=%d", ret);
return ret;
}
_already_written = false;
return ret;
}
/*
The MIT License (MIT)
Copyright (c) 2013-2014 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#ifndef SRS_APP_PIPE_HPP
#define SRS_APP_PIPE_HPP
/*
#include <srs_app_pipe.hpp>
*/
#include <srs_core.hpp>
#include <srs_app_st.hpp>
/**
* convert something to io,
* for example, signal or SrsConsumer event.
* for performance issue, @see: https://github.com/winlinvip/simple-rtmp-server/issues/194
*/
class SrsPipe
{
private:
int fds[2];
st_netfd_t read_stfd;
st_netfd_t write_stfd;
/**
* for the event based service,
* for example, the consumer only care whether there is data writen in pipe,
* and the source will not write to pipe when pipe is already writen.
*/
bool _already_written;
public:
SrsPipe();
virtual ~SrsPipe();
public:
/**
* initialize pipes, open fds.
*/
virtual int initialize();
/**
* get the read fd to poll.
*/
virtual st_netfd_t rfd();
public:
/**
* for event based service, whether already writen data.
*/
virtual bool already_written();
/**
* for event based service,
* write an int to pipe and set the pipe to active.
*/
virtual int active();
/**
* for event based service,
* read an int from pipe and reset the pipe to deactive.
*/
virtual int reset();
};
#endif
... ... @@ -516,12 +516,7 @@ int SrsRtmpConn::playing(SrsSource* source)
SrsAutoFree(SrsConsumer, consumer);
srs_verbose("consumer created success.");
// TODO: FIXME: remove it.
rtmp->set_recv_timeout(SRS_CONSTS_RTMP_PULSE_TIMEOUT_US);
// disable the timeout.
// TODO: FIXME: maybe can use larger timeout?
rtmp->set_recv_timeout(ST_UTIME_NO_TIMEOUT);
rtmp->set_send_timeout(ST_UTIME_NO_TIMEOUT);
SrsPithyPrint pithy_print(SRS_CONSTS_STAGE_PLAY_USER);
... ... @@ -530,30 +525,12 @@ int SrsRtmpConn::playing(SrsSource* source)
bool user_specified_duration_to_stop = (req->duration > 0);
int64_t starttime = -1;
pollfd pds[2];
// poll the client incoming fd.
pds[0].fd = st_netfd_fileno(stfd);
pds[0].events = POLLIN;
// poll the consumer queue pipe.
pds[1].fd = st_netfd_fileno(consumer->pipe_fd());
pds[1].events = POLLIN;
while (true) {
// collect elapse for pithy print.
pithy_print.elapse();
pds[0].revents = 0;
pds[1].revents = 0;
// wait for packet incoming or data outgoing.
if (st_poll(pds, 2, ST_UTIME_NO_TIMEOUT) <= 0) {
srs_error("st_poll failed.");
break;
}
// packet incoming, read from RTMP.
// read from client.
if (pds[0].revents & POLLIN) {
if (true) {
SrsMessage* msg = NULL;
ret = rtmp->recv_message(&msg);
srs_verbose("play loop recv message. ret=%d", ret);
... ... @@ -576,52 +553,49 @@ int SrsRtmpConn::playing(SrsSource* source)
}
}
// data outgoing, sendout packets.
if (pds[1].revents & POLLIN) {
// get messages from consumer.
int count = 0;
if ((ret = consumer->dump_packets(msgs.size, msgs.msgs, count)) != ERROR_SUCCESS) {
srs_error("get messages from consumer failed. ret=%d", ret);
return ret;
}
// reportable
if (pithy_print.can_print()) {
kbps->sample();
srs_trace("-> "SRS_CONSTS_LOG_PLAY
" time=%"PRId64", msgs=%d, okbps=%d,%d,%d, ikbps=%d,%d,%d",
pithy_print.age(), count,
kbps->get_send_kbps(), kbps->get_send_kbps_30s(), kbps->get_send_kbps_5m(),
kbps->get_recv_kbps(), kbps->get_recv_kbps_30s(), kbps->get_recv_kbps_5m());
}
// get messages from consumer.
int count = 0;
if ((ret = consumer->dump_packets(msgs.size, msgs.msgs, count)) != ERROR_SUCCESS) {
srs_error("get messages from consumer failed. ret=%d", ret);
return ret;
}
// reportable
if (pithy_print.can_print()) {
kbps->sample();
srs_trace("-> "SRS_CONSTS_LOG_PLAY
" time=%"PRId64", msgs=%d, okbps=%d,%d,%d, ikbps=%d,%d,%d",
pithy_print.age(), count,
kbps->get_send_kbps(), kbps->get_send_kbps_30s(), kbps->get_send_kbps_5m(),
kbps->get_recv_kbps(), kbps->get_recv_kbps_30s(), kbps->get_recv_kbps_5m());
}
// 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.msgs[i];
// 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.msgs[i];
// the send_message will free the msg,
// so set the msgs[i] to NULL.
msgs.msgs[i] = NULL;
// 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;
// the send_message will free the msg,
// so set the msgs[i] to NULL.
msgs.msgs[i] = NULL;
// 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;
}
// 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;
}
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;
}
}
... ...
... ... @@ -41,7 +41,6 @@ using namespace std;
#include <srs_app_edge.hpp>
#include <srs_kernel_utility.hpp>
#include <srs_app_avc_aac.hpp>
#include <srs_app_pipe.hpp>
#define CONST_MAX_JITTER_MS 500
#define DEFAULT_FRAME_TIME_MS 40
... ... @@ -172,11 +171,6 @@ void SrsMessageQueue::set_queue_size(double queue_size)
queue_size_ms = (int)(queue_size * 1000);
}
bool SrsMessageQueue::empty()
{
return msgs.size() == 0;
}
int SrsMessageQueue::enqueue(SrsSharedPtrMessage* msg)
{
int ret = ERROR_SUCCESS;
... ... @@ -296,7 +290,6 @@ SrsConsumer::SrsConsumer(SrsSource* _source)
jitter = new SrsRtmpJitter();
queue = new SrsMessageQueue();
should_update_source_id = false;
pipe = new SrsPipe();
}
SrsConsumer::~SrsConsumer()
... ... @@ -306,23 +299,6 @@ SrsConsumer::~SrsConsumer()
srs_freep(queue);
}
int SrsConsumer::initialize()
{
int ret = ERROR_SUCCESS;
if ((ret = pipe->initialize()) != ERROR_SUCCESS) {
srs_error("initialize the pipe for consumer failed. ret=%d", ret);
return ret;
}
return ret;
}
st_netfd_t SrsConsumer::pipe_fd()
{
return pipe->rfd();
}
void SrsConsumer::set_queue_size(double queue_size)
{
queue->set_queue_size(queue_size);
... ... @@ -353,18 +329,11 @@ int SrsConsumer::enqueue(SrsSharedPtrMessage* msg, bool atc, int tba, int tbv, S
return ret;
}
// notify the rtmp connection to resume to send packet.
if (!pipe->already_written()) {
pipe->active();
}
return ret;
}
int SrsConsumer::dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& count)
{
int ret = ERROR_SUCCESS;
srs_assert(max_count > 0);
if (should_update_source_id) {
... ... @@ -377,15 +346,7 @@ int SrsConsumer::dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& c
return ERROR_SUCCESS;
}
if ((ret = queue->dump_packets(max_count, pmsgs, count)) != ERROR_SUCCESS) {
return ret;
}
if (queue->empty()) {
return pipe->reset();
}
return ret;
return queue->dump_packets(max_count, pmsgs, count);
}
int SrsConsumer::on_play_client_pause(bool is_pause)
... ... @@ -1493,13 +1454,7 @@ void SrsSource::on_unpublish()
{
int ret = ERROR_SUCCESS;
SrsConsumer* c = new SrsConsumer(this);
if ((ret = c->initialize()) != ERROR_SUCCESS) {
srs_freep(c);
return ret;
}
consumer = c;
consumer = new SrsConsumer(this);
consumers.push_back(consumer);
double queue_size = _srs_config->get_queue_length(_req->vhost);
... ...
... ... @@ -58,7 +58,6 @@ class SrsDvr;
class SrsEncoder;
#endif
class SrsStream;
class SrsPipe;
/**
* the time jitter algorithm:
... ... @@ -123,10 +122,6 @@ public:
virtual void set_queue_size(double queue_size);
public:
/**
* whether queue is empty.
*/
virtual bool empty();
/**
* enqueue the message, the timestamp always monotonically.
* @param msg, the msg to enqueue, user never free it whatever the return code.
*/
... ... @@ -153,7 +148,6 @@ private:
class SrsConsumer
{
private:
SrsPipe* pipe;
SrsRtmpJitter* jitter;
SrsSource* source;
SrsMessageQueue* queue;
... ... @@ -165,16 +159,6 @@ public:
virtual ~SrsConsumer();
public:
/**
* initialize the consumer.
*/
virtual int initialize();
/**
* source can use this fd to poll with the read event,
* for performance issue, @see: https://github.com/winlinvip/simple-rtmp-server/issues/194
*/
virtual st_netfd_t pipe_fd();
public:
/**
* set the size of queue.
*/
virtual void set_queue_size(double queue_size);
... ...
... ... @@ -88,8 +88,6 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#define ERROR_SYSTEM_FILE_SEEK 1049
#define ERROR_SYSTEM_IO_INVALID 1050
#define ERROR_ST_EXCEED_THREADS 1051
#define ERROR_SYSTEM_READ_PIPE 1052
#define ERROR_SYSTEM_WRITE_PIPE 1053
///////////////////////////////////////////////////////
// RTMP protocol error.
... ...
... ... @@ -92,8 +92,6 @@ file
..\app\srs_app_kbps.cpp,
..\app\srs_app_log.hpp,
..\app\srs_app_log.cpp,
..\app\srs_app_pipe.hpp,
..\app\srs_app_pipe.cpp,
..\app\srs_app_refer.hpp,
..\app\srs_app_refer.cpp,
..\app\srs_app_reload.hpp,
... ...