胡斌 / srs

转到一个项目
Toggle navigation Toggle navigation pinning
winlin
7150a99f 1 parent 149d5863

for bug #251, remove the mic(message iovs cache), no use.

内嵌 并排对比
正在显示 7 个修改的文件 包含 2 行增加 和 452 行删除
... ... @@ -822,10 +822,6 @@ SrsSource::SrsSource(SrsRequest* req)
_srs_config->subscribe(this);
atc = _srs_config->get_atc(_req->vhost);
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
chunk_size = 0;
#endif
}
SrsSource::~SrsSource()
... ... @@ -1052,26 +1048,6 @@ int SrsSource::on_reload_vhost_dvr(string vhost)
return ret;
}
int SrsSource::on_reload_vhost_chunk_size(string vhost)
{
int ret = ERROR_SUCCESS;
if (_req->vhost != vhost) {
return ret;
}
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
int size = _srs_config->get_chunk_size(_req->vhost);
if (chunk_size != size) {
srs_warn("connected clients will error for mic chunk_size changed %d=>%d",
chunk_size, size);
}
chunk_size = size;
#endif
return ret;
}
int SrsSource::on_reload_vhost_transcode(string vhost)
{
int ret = ERROR_SUCCESS;
... ... @@ -1301,14 +1277,6 @@ int SrsSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata
}
srs_verbose("initialize shared ptr metadata success.");
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
if ((ret = cache_metadata->mic_evaluate(chunk_size)) != ERROR_SUCCESS) {
srs_error("mic metadata iovs failed, chunk_size=%d. ret=%d", chunk_size, ret);
return ret;
}
srs_info("mic metadata iovs ok, chunk_size=%d", chunk_size);
#endif
// copy to all consumer
if (true) {
std::vector<SrsConsumer*>::iterator it;
... ... @@ -1350,14 +1318,6 @@ int SrsSource::on_audio(SrsCommonMessage* __audio)
}
srs_verbose("initialize shared ptr audio success.");
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
if ((ret = msg.mic_evaluate(chunk_size)) != ERROR_SUCCESS) {
srs_error("mic audio iovs failed, chunk_size=%d. ret=%d", chunk_size, ret);
return ret;
}
srs_info("mic audio iovs ok, chunk_size=%d", chunk_size);
#endif
#ifdef SRS_AUTO_HLS
if ((ret = hls->on_audio(&msg)) != ERROR_SUCCESS) {
srs_warn("hls process audio message failed, ignore and disable hls. ret=%d", ret);
... ... @@ -1468,14 +1428,6 @@ int SrsSource::on_video(SrsCommonMessage* __video)
}
srs_verbose("initialize shared ptr video success.");
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
if ((ret = msg.mic_evaluate(chunk_size)) != ERROR_SUCCESS) {
srs_error("mic video iovs failed, chunk_size=%d. ret=%d", chunk_size, ret);
return ret;
}
srs_info("mic video iovs ok, chunk_size=%d", chunk_size);
#endif
#ifdef SRS_AUTO_HLS
if ((ret = hls->on_video(&msg)) != ERROR_SUCCESS) {
srs_warn("hls process video message failed, ignore and disable hls. ret=%d", ret);
... ... @@ -1727,11 +1679,6 @@ int SrsSource::on_publish()
return ret;
}
#endif
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
chunk_size = _srs_config->get_chunk_size(_req->vhost);
srs_trace("mic use chunk_size=%d to send msgs", chunk_size);
#endif
return ret;
}
... ...
... ... @@ -403,11 +403,6 @@ private:
std::vector<SrsForwarder*> forwarders;
// for aggregate message
SrsStream* aggregate_stream;
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
// the chunk size for mic,
// update when publish stream.
int chunk_size;
#endif
private:
/**
* the sample rate of audio in metadata.
... ... @@ -452,7 +447,6 @@ public:
virtual int on_reload_vhost_forward(std::string vhost);
virtual int on_reload_vhost_hls(std::string vhost);
virtual int on_reload_vhost_dvr(std::string vhost);
virtual int on_reload_vhost_chunk_size(std::string vhost);
virtual int on_reload_vhost_transcode(std::string vhost);
// for the tools callback
public:
... ...
... ... @@ -31,7 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
// current release version
#define VERSION_MAJOR 2
#define VERSION_MINOR 0
#define VERSION_REVISION 67
#define VERSION_REVISION 68
// server info.
#define RTMP_SIG_SRS_KEY "SRS"
#define RTMP_SIG_SRS_ROLE "origin/edge server"
... ...
... ... @@ -97,25 +97,12 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
// the default config of mw.
#define SRS_PERF_MW_SLEEP 350
/**
* use iovs cache in each msg,
* for the shared ptr message, we calc once and used for every copy.
* @see https://github.com/winlinvip/simple-rtmp-server/issues/251
* @remark if enable this, donot use protocol iovs cache.
* @remark when reload change the chunk size, previous clients error.
*/
#undef SRS_PERF_MW_MSG_IOVS_CACHE
/**
* how many msgs can be send entirely.
* for play clients to get msgs then totally send out.
* for the mw sleep set to 1800, the msgs is about 133.
* @remark, recomment to 128.
* @remark, when mic enabled, use larger iovs cache, to 512.
*/
#ifndef SRS_PERF_MW_MSG_IOVS_CACHE
#define SRS_PERF_MW_MSGS 128
#else
#define SRS_PERF_MW_MSGS 512
#endif
#define SRS_PERF_MW_MSGS 128
/**
* whether set the socket send buffer size.
... ...
... ... @@ -149,11 +149,6 @@ void show_macro_features()
#endif
srs_trace("system default latency in ms: mw(0-%d) + mr(0-%d) + play-queue(0-%d)",
SRS_PERF_MW_SLEEP, possible_mr_latency, SRS_PERF_PLAY_QUEUE*1000);
#ifndef SRS_PERF_MW_MSG_IOVS_CACHE
srs_warn("MIC(message iovs cache) enabled, the connected clients will be"
"disconneted when reload changed the chunk_size.");
#endif
}
void check_macro_features()
... ...
... ... @@ -387,109 +387,16 @@ SrsSharedPtrMessage::__SrsSharedPtr::__SrsSharedPtr()
payload = NULL;
size = 0;
shared_count = 0;
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
nb_iovs = 0;
iovs = NULL;
chunk_size = 0;
#endif
}
SrsSharedPtrMessage::__SrsSharedPtr::~__SrsSharedPtr()
{
srs_freep(payload);
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
srs_freep(iovs);
#endif
}
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
int SrsSharedPtrMessage::__SrsSharedPtr::mic_evaluate(int chunk_size)
{
int ret = ERROR_SUCCESS;
// use the chunk size, shuold not be changed.
this->chunk_size = chunk_size;
// c0 header
int nbh = srs_chunk_header_c0(
header.perfer_cid, 0, header.payload_length,
header.message_type, 0,
mic_c0, sizeof(mic_c0));
srs_assert(nbh > 0);;
// c3 header
mic_c3 = 0xC0 | (header.perfer_cid & 0x3F);
// calc number of iovs
nb_chunks = header.payload_length / chunk_size;
if (header.payload_length % chunk_size) {
nb_chunks++;
}
nb_iovs = 1/*cid*/ + 1/*size*//*type*/+ 1/*chunk*/;
// left chunks, always cid+chunk.
if (nb_chunks > 0) {
nb_iovs += (nb_chunks - 1) * 2;
}
// create iovs
srs_freep(iovs);
iovs = new iovec[nb_iovs];
// for payload chunks.
char* p = payload;
char* end = p + size;
iovec* iov = iovs + 0;
while (p < end) {
// size of payload.
int payload_size = srs_min(chunk_size, end - p);
// header, c0 or c3
if (p == payload) {
// c0, cid+size+type
// cid, 1B
iov[0].iov_base = mic_c0;
iov[0].iov_len = 1;
// size(payload length), 3B
// type(message type), 1B
iov[1].iov_base = mic_c0 + 4;
iov[1].iov_len = 4;
// chunk
iov[2].iov_base = p;
iov[2].iov_len = payload_size;
// move to next iovs.
iov += 3;
} else {
// c3
iov[0].iov_base = &mic_c3;
iov[0].iov_len = 1;
// chunk
iov[1].iov_base = p;
iov[1].iov_len = payload_size;
// move to next iovs.
iov += 2;
}
// to next chunk
p += payload_size;
}
return ret;
}
#endif
SrsSharedPtrMessage::SrsSharedPtrMessage()
{
ptr = NULL;
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
mic_etime_present = false;
#endif
}
SrsSharedPtrMessage::~SrsSharedPtrMessage()
... ... @@ -592,7 +499,6 @@ bool SrsSharedPtrMessage::is_video()
return ptr->header.message_type == RTMP_MSG_VideoMessage;
}
#ifndef SRS_PERF_MW_MSG_IOVS_CACHE
int SrsSharedPtrMessage::chunk_header(char* cache, int nb_cache, bool c0)
{
if (c0) {
... ... @@ -606,7 +512,6 @@ int SrsSharedPtrMessage::chunk_header(char* cache, int nb_cache, bool c0)
cache, nb_cache);
}
}
#endif
SrsSharedPtrMessage* SrsSharedPtrMessage::copy()
{
... ... @@ -625,155 +530,6 @@ SrsSharedPtrMessage* SrsSharedPtrMessage::copy()
return copy;
}
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
int SrsSharedPtrMessage::mic_evaluate(int chunk_size)
{
int ret = ERROR_SUCCESS;
// when chunk size changed, error to disconnect the client..
if (ptr->chunk_size > 0 && chunk_size != ptr->chunk_size) {
ret = ERROR_RTMP_MIC_CHUNKSIZE_CHANGED;
srs_warn("mic chunk size changed %d=>%d, ret=%d",
ptr->chunk_size, chunk_size, ret);
return ret;
}
// calc the shared ptr iovs at the first time.
if (ptr->chunk_size <= 0) {
if ((ret = ptr->mic_evaluate(chunk_size)) != ERROR_SUCCESS) {
srs_warn("mic evaluate source iovs failed. ret=%d", ret);
return ret;
}
}
return ret;
}
int SrsSharedPtrMessage::mic_iovs_dump(iovec* iovs, int max_nb_iovs)
{
// calc the private iovs
char* pp = NULL;
// calc number of iovs.
int nb_iovs = 1/*time*/ + 1/*sid*/;
// insert etime before all chunks.
if (mic_etime_present) {
nb_iovs += ptr->nb_chunks;
}
// not enough, return nagetive to try another loop.
if (max_nb_iovs < nb_iovs + ptr->nb_iovs) {
return -1;
}
// timestamp for c0/c3
u_int32_t timestamp = (u_int32_t)this->timestamp;
mic_etime_present = timestamp >= RTMP_EXTENDED_TIMESTAMP;
// chunk message header, 11 bytes
// timestamp, 3bytes, big-endian
char* p = mic_c0_time;
if (!mic_etime_present) {
pp = (char*)&timestamp;
*p++ = pp[2];
*p++ = pp[1];
*p++ = pp[0];
} else {
*p++ = 0xFF;
*p++ = 0xFF;
*p++ = 0xFF;
}
// stream_id, 4bytes, little-endian
p = mic_c0_sid;
pp = (char*)&stream_id;
*p++ = pp[0];
*p++ = pp[1];
*p++ = pp[2];
*p++ = pp[3];
// for c0
// chunk extended timestamp header, 0 or 4 bytes, big-endian
//
// for c3:
// chunk extended timestamp header, 0 or 4 bytes, big-endian
// 6.1.3. Extended Timestamp
// This field is transmitted only when the normal time stamp in the
// chunk message header is set to 0x00ffffff. If normal time stamp is
// set to any value less than 0x00ffffff, this field MUST NOT be
// present. This field MUST NOT be present if the timestamp field is not
// present. Type 3 chunks MUST NOT have this field.
// adobe changed for Type3 chunk:
// FMLE always sendout the extended-timestamp,
// must send the extended-timestamp to FMS,
// must send the extended-timestamp to flash-player.
// @see: ngx_rtmp_prepare_message
// @see: http://blog.csdn.net/win_lin/article/details/13363699
// TODO: FIXME: extract to outer.
p = mic_etime;
if (mic_etime_present) {
pp = (char*)&timestamp;
*p++ = pp[3];
*p++ = pp[2];
*p++ = pp[1];
*p++ = pp[0];
}
// dumps all ovs
iovec* shared = ptr->iovs;
iovec* iov = iovs;
// dump the c0 chunk
// cid
iov->iov_len = shared->iov_len;
iov->iov_base = shared->iov_base;
iov++; shared++;
// time, 3B
iov->iov_base = mic_c0_time;
iov->iov_len = 3;
iov++;
// size, type
iov->iov_len = shared->iov_len;
iov->iov_base = shared->iov_base;
iov++; shared++;
// sid, 4B
iov->iov_base = mic_c0_sid;
iov->iov_len = 4;
iov++;
// etime, 4B
if (mic_etime_present) {
// etime
iov->iov_base = mic_etime;
iov->iov_len = 4;
iov++;
}
// chunk
iov->iov_len = shared->iov_len;
iov->iov_base = shared->iov_base;
iov++; shared++;
// dump left c3 chunks
for (int i = 1; i < ptr->nb_chunks; i++) {
// cid
iov->iov_len = shared->iov_len;
iov->iov_base = shared->iov_base;
iov++; shared++;
if (mic_etime_present) {
// etime
iov->iov_base = mic_etime;
iov->iov_len = 4;
iov++;
}
// chunk
iov->iov_len = shared->iov_len;
iov->iov_base = shared->iov_base;
iov++; shared++;
}
return nb_iovs + ptr->nb_iovs;
}
#endif
SrsProtocol::AckWindowSize::AckWindowSize()
{
ack_window_size = 0;
... ... @@ -793,10 +549,8 @@ SrsProtocol::SrsProtocol(ISrsProtocolReaderWriter* io)
// each chunk consumers atleast 2 iovs
srs_assert(nb_out_iovs >= 2);
#ifndef SRS_PERF_MW_MSG_IOVS_CACHE
warned_c0c3_cache_dry = false;
auto_response_when_recv = true;
#endif
cs_cache = NULL;
if (SRS_PERF_CHUNK_STREAM_CACHE > 0) {
... ... @@ -1004,7 +758,6 @@ int SrsProtocol::do_send_messages(SrsSharedPtrMessage** msgs, int nb_msgs)
{
int ret = ERROR_SUCCESS;
#ifndef SRS_PERF_MW_MSG_IOVS_CACHE
int iov_index = 0;
iovec* iov = out_iovs + iov_index;
... ... @@ -1105,51 +858,6 @@ int SrsProtocol::do_send_messages(SrsSharedPtrMessage** msgs, int nb_msgs)
nb_msgs, iov_index, SRS_PERF_MW_MSGS, nb_out_iovs);
return do_iovs_send(out_iovs, iov_index);
#else
// send all iovs for all msgs.
int msg_sent = 0;
while (msg_sent < nb_msgs) {
int iov_index = 0;
for (int i = msg_sent; i < nb_msgs; i++) {
SrsSharedPtrMessage* msg = msgs[i];
// evaluate the first
if (i == 0 && (ret = msg->mic_evaluate(out_chunk_size)) != ERROR_SUCCESS) {
return ret;
}
// dump msg to iovec.
int ok_iovs = msg->mic_iovs_dump(
out_iovs + iov_index, nb_out_iovs - iov_index
);
// protocol iovs cache exceed.
if (ok_iovs <= 0) {
break;
}
// ok, dump next.
msg_sent++;
iov_index += ok_iovs;
}
srs_info("mic nb_iovs=%d, msgs=%d, msg_sent=%d, iovs_sent=%d",
nb_out_iovs, nb_msgs, msg_sent, iov_index);
// cache not enough.
if (iov_index <= 0) {
ret = ERROR_RTMP_MIC_CACHE_OVERFLOW;
srs_warn("mic iovs overflow, nb_iovs=%d, msgs=%d, msg_sent=%d, iovs_sent=%d, ret=%d",
nb_out_iovs, nb_msgs, msg_sent, iov_index, ret);
return ret;
}
// send out these iovs.
if ((ret = do_iovs_send(out_iovs, iov_index)) != ERROR_SUCCESS) {
return ret;
}
}
return ret;
#endif
}
int SrsProtocol::do_iovs_send(iovec* iovs, int size)
... ...
... ... @@ -265,72 +265,10 @@ private:
// the reference count
int shared_count;
public:
// the iovs cache in shared ptr message.
// @see https://github.com/winlinvip/simple-rtmp-server/issues/251
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
/**
* the mic(msg iovs cache).
* why share the cache in msg?
* all msgs of a source are same for:
* 1. cid, all use the same cid, copy from src msg.
* 2. size, all msg size never changed.
* 3. type, type never changed.
* 4. chunk size, all connections in a vhost use the same chunk size.
* the different:
* 1. time and etime, almost different.
* 2. stream id, maybe different, but almost the same.
* @remark, when reload change the chunk size, clients will be disconnected.
*/
// the c0 shared section for all msgs
// 1. cid, 1B, same.
// 2. [*]time, 3B, not same.
// 3. size, 3B, same.
// 4. type, 1B, same.
// 5. [*]stream id, 4B, not same, little-endian.
// 6. [*]etime, 4B, not same.
// the stared field must be calced in each msg.
char mic_c0[16];
// the c3 headers.
char mic_c3;
// the calced iovs for all msg,
// we assumpt that the chunk size is not changed for a vhost,
// if do changed, the client will got an error msg and disconnect.
iovec* iovs;
int nb_iovs;
// the msgs source chunk size,
// which is evaluated the iovs first time,
// this cannot be changed.
int chunk_size;
// the number of chunks.
int nb_chunks;
#endif
public:
__SrsSharedPtr();
virtual ~__SrsSharedPtr();
public:
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
/**
* for iovs msg cache, calc the iovs.
* @param chunk_size use the specified chunk size to evaluate the iovs.
*/
virtual int mic_evaluate(int chunk_size);
#endif
};
__SrsSharedPtr* ptr;
private:
// msgs level cache.
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
// the c0 private section for this
// 1. time, 3B, not same, not used.
// 2. stream id, 4B, almost the same, little-endian.
// 3. etime, 4B, optional, always same for all chunk when present.
// the stared field must be calced in each msg.
char mic_c0_time[3];
char mic_c0_sid[4];
char mic_etime[4];
// whether etime present.
bool mic_etime_present;
#endif
public:
SrsSharedPtrMessage();
virtual ~SrsSharedPtrMessage();
... ... @@ -366,32 +304,17 @@ public:
virtual bool is_audio();
virtual bool is_video();
public:
#ifndef SRS_PERF_MW_MSG_IOVS_CACHE
/**
* generate the chunk header to cache.
* @return the size of header.
*/
virtual int chunk_header(char* cache, int nb_cache, bool c0);
#endif
public:
/**
* copy current shared ptr message, use ref-count.
* @remark, assert object is created.
*/
virtual SrsSharedPtrMessage* copy();
public:
#ifdef SRS_PERF_MW_MSG_IOVS_CACHE
/**
* for iovs msg cache, calc the iovs.
* @param chunk_size use the specified chunk size to evaluate the iovs.
*/
virtual int mic_evaluate(int chunk_size);
/**
* dump all iovs, the _nb_iovs must equals to mic_iovs_count().
* @return the dumped count. -1 if not enough iovs.
*/
virtual int mic_iovs_dump(iovec* iovs, int max_nb_iovs);
#endif
};
/**
... ... @@ -465,9 +388,6 @@ private:
*/
iovec* out_iovs;
int nb_out_iovs;
// if use iovs cache in each msg,
// donot use protocol level c0c3 cache.
#ifndef SRS_PERF_MW_MSG_IOVS_CACHE
/**
* output header cache.
* used for type0, 11bytes(or 15bytes with extended timestamp) header.
... ... @@ -479,7 +399,6 @@ private:
char out_c0c3_caches[SRS_CONSTS_C0C3_HEADERS_MAX];
// whether warned user to increase the c0c3 header cache.
bool warned_c0c3_cache_dry;
#endif
/**
* output chunk size, default to 128, set by config.
*/
... ...