winlin

fix the log cid display error, merge the publish recv thread log to publish connection.

... ... @@ -57,6 +57,20 @@ int SrsThreadContext::get_id()
return cache[st_thread_self()];
}
int SrsThreadContext::set_id(int v)
{
st_thread_t self = st_thread_self();
int ov = 0;
if (cache.find(self) != cache.end()) {
ov = cache[self];
}
cache[self] = v;
return ov;
}
// the max size of a line of log.
#define LOG_MAX_SIZE 4096
... ...
... ... @@ -53,6 +53,7 @@ public:
public:
virtual int generate_id();
virtual int get_id();
virtual int set_id(int v);
};
/**
... ...
... ... @@ -62,6 +62,11 @@ SrsRecvThread::~SrsRecvThread()
srs_freep(trd);
}
int SrsRecvThread::cid()
{
return trd->cid();
}
int SrsRecvThread::start()
{
return trd->start();
... ... @@ -253,6 +258,7 @@ SrsPublishRecvThread::SrsPublishRecvThread(
recv_error_code = ERROR_SUCCESS;
_nb_msgs = 0;
error = st_cond_new();
ncid = cid = 0;
req = _req;
mr_fd = mr_sock_fd;
... ... @@ -297,9 +303,21 @@ int SrsPublishRecvThread::error_code()
return recv_error_code;
}
void SrsPublishRecvThread::set_cid(int v)
{
ncid = v;
}
int SrsPublishRecvThread::get_cid()
{
return ncid;
}
int SrsPublishRecvThread::start()
{
return trd.start();
int ret = trd.start();
ncid = cid = trd.cid();
return ret;
}
void SrsPublishRecvThread::stop()
... ... @@ -352,6 +370,12 @@ int SrsPublishRecvThread::handle(SrsCommonMessage* msg)
{
int ret = ERROR_SUCCESS;
// when cid changed, change it.
if (ncid != cid) {
_srs_context->set_id(ncid);
cid = ncid;
}
_nb_msgs++;
// log to show the time of recv thread.
... ...
... ... @@ -91,6 +91,8 @@ public:
SrsRecvThread(ISrsMessageHandler* msg_handler, SrsRtmpServer* rtmp_sdk, int timeout_ms);
virtual ~SrsRecvThread();
public:
virtual int cid();
public:
virtual int start();
virtual void stop();
virtual void stop_loop();
... ... @@ -170,6 +172,9 @@ private:
// the error timeout cond
// @see https://github.com/simple-rtmp-server/srs/issues/244
st_cond_t error;
// merged context id.
int cid;
int ncid;
public:
SrsPublishRecvThread(SrsRtmpServer* rtmp_sdk,
SrsRequest* _req, int mr_sock_fd, int timeout_ms,
... ... @@ -182,6 +187,8 @@ public:
virtual int wait(int timeout_ms);
virtual int64_t nb_msgs();
virtual int error_code();
virtual void set_cid(int v);
virtual int get_cid();
public:
virtual int start();
virtual void stop();
... ...
... ... @@ -851,6 +851,11 @@ int SrsRtmpConn::do_publishing(SrsSource* source, SrsPublishRecvThread* trd)
return ret;
}
// change the isolate recv thread context id,
// merge its log to current thread.
int receive_thread_cid = trd->get_cid();
trd->set_cid(_srs_context->get_id());
// initialize the publish timeout.
publish_1stpkt_timeout = _srs_config->get_publish_1stpkt_timeout(req->vhost);
publish_normal_timeout = _srs_config->get_publish_1stpkt_timeout(req->vhost);
... ... @@ -861,8 +866,8 @@ int SrsRtmpConn::do_publishing(SrsSource* source, SrsPublishRecvThread* trd)
if (true) {
bool mr = _srs_config->get_mr_enabled(req->vhost);
int mr_sleep = _srs_config->get_mr_sleep_ms(req->vhost);
srs_trace("start publish mr=%d/%d, p1stpt=%d, pnt=%d, tcp_nodelay=%d",
mr, mr_sleep, publish_1stpkt_timeout, publish_normal_timeout, tcp_nodelay);
srs_trace("start publish mr=%d/%d, p1stpt=%d, pnt=%d, tcp_nodelay=%d, rtcid=%d",
mr, mr_sleep, publish_1stpkt_timeout, publish_normal_timeout, tcp_nodelay, receive_thread_cid);
}
int64_t nb_msgs = 0;
... ... @@ -1109,7 +1114,9 @@ int SrsRtmpConn::process_play_control_msg(SrsConsumer* consumer, SrsCommonMessag
res->command_object = SrsAmf0Any::null();
res->response = SrsAmf0Any::null();
if ((ret = rtmp->send_and_free_packet(res, 0)) != ERROR_SUCCESS) {
if (!srs_is_system_control_error(ret) && !srs_is_client_gracefully_close(ret)) {
srs_warn("response call failed. ret=%d", ret);
}
return ret;
}
}
... ...
... ... @@ -1415,6 +1415,7 @@ int SrsSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata
if ((prop = metadata->metadata->ensure_property_number("audiocodecid")) != NULL) {
ss << ", acodec=" << (int)prop->to_number();
}
srs_trace("got metadata%s", ss.str().c_str());
// add server info to metadata
metadata->metadata->set("server", SrsAmf0Any::str(RTMP_SIG_SRS_SERVER));
... ... @@ -1479,7 +1480,6 @@ int SrsSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata
return ret;
}
}
srs_trace("got metadata%s", ss.str().c_str());
}
// copy to all forwarders
... ...
... ... @@ -76,4 +76,9 @@ int ISrsThreadContext::get_id()
return 0;
}
int ISrsThreadContext::set_id(int /*v*/)
{
return 0;
}
... ...
... ... @@ -95,6 +95,13 @@ public:
virtual void error(const char* tag, int context_id, const char* fmt, ...);
};
/**
* the context id manager to identify context, for instance, the green-thread.
* usage:
* _srs_context->generate_id(); // when thread start.
* _srs_context->get_id(); // get current generated id.
* int old_id = _srs_context->set_id(1000); // set context id if need to merge thread context.
*/
// the context for multiple clients.
class ISrsThreadContext
{
... ... @@ -102,8 +109,19 @@ public:
ISrsThreadContext();
virtual ~ISrsThreadContext();
public:
/**
* generate the id for current context.
*/
virtual int generate_id();
/**
* get the generated id of current context.
*/
virtual int get_id();
/**
* set the id of current context.
* @return the previous id value; 0 if no context.
*/
virtual int set_id(int v);
};
// user must provides a log object
... ...
... ... @@ -2633,7 +2633,9 @@ int SrsRtmpServer::identify_client(int stream_id, SrsRtmpConnType& type, string&
res->command_object = SrsAmf0Any::null();
res->response = SrsAmf0Any::null();
if ((ret = protocol->send_and_free_packet(res, 0)) != ERROR_SUCCESS) {
if (!srs_is_system_control_error(ret) && !srs_is_client_gracefully_close(ret)) {
srs_warn("response call failed. ret=%d", ret);
}
return ret;
}
continue;
... ...