diff --git a/trunk/src/app/srs_app_rtmp_conn.cpp b/trunk/src/app/srs_app_rtmp_conn.cpp index f2986c52e..ae83a3e63 100644 --- a/trunk/src/app/srs_app_rtmp_conn.cpp +++ b/trunk/src/app/srs_app_rtmp_conn.cpp @@ -493,10 +493,117 @@ int SrsRtmpConn::check_vhost() return ret; } +class IsolateRecvThread : public ISrsThreadHandler +{ +private: + SrsThread* trd; + SrsRtmpServer* rtmp; + std::vector queue; +public: + IsolateRecvThread(SrsRtmpServer* rtmp_sdk) + { + rtmp = rtmp_sdk; + trd = new SrsThread(this, 0, true); + } + virtual ~IsolateRecvThread() + { + // stop recv thread. + stop(); + + // destroy the thread. + srs_freep(trd); + + // clear all messages. + std::vector::iterator it; + for (it = queue.begin(); it != queue.end(); ++it) { + SrsMessage* msg = *it; + srs_freep(msg); + } + queue.clear(); + } +public: + virtual bool empty() + { + return queue.empty(); + } + virtual SrsMessage* pump() + { + SrsMessage* msg = *queue.begin(); + queue.erase(queue.begin()); + return msg; + } +public: + virtual int start() + { + return trd->start(); + } + virtual void stop() + { + trd->stop(); + } + virtual int cycle() + { + int ret = ERROR_SUCCESS; + + SrsMessage* msg = NULL; + + if ((ret = rtmp->recv_message(&msg)) != ERROR_SUCCESS) { + if (!srs_is_client_gracefully_close(ret)) { + srs_error("recv client control message failed. ret=%d", ret); + } + + // we use no timeout to recv, should never got any error. + trd->stop_loop(); + + return ret; + } + srs_verbose("play loop recv message. ret=%d", ret); + + return ret; + } +}; + int SrsRtmpConn::playing(SrsSource* source) { int ret = ERROR_SUCCESS; + // the multiple messages writev improve performance large, + // but the timeout recv will cause 33% sys call performance, + // to use isolate thread to recv, can improve about 33% performance. + // @see https://github.com/winlinvip/simple-rtmp-server/issues/194 + // @see: https://github.com/winlinvip/simple-rtmp-server/issues/217 + //rtmp->set_recv_timeout(SRS_CONSTS_RTMP_PULSE_TIMEOUT_US); + rtmp->set_recv_timeout(ST_UTIME_NO_TIMEOUT); + + // disable the protocol auto response, + // for the isolate recv thread should never send any messages. + rtmp->set_auto_response(false); + + // use isolate thread to recv, + // start isolate recv thread. + IsolateRecvThread trd(rtmp); + if ((ret = trd.start()) != ERROR_SUCCESS) { + srs_error("start isolate recv thread failed. ret=%d", ret); + return ret; + } + + // delivery messages for clients playing stream. + ret = do_playing(source, &trd); + + // stop isolate recv thread + trd.stop(); + + // enable the protocol auto response, + // for the isolate recv thread terminated. + rtmp->set_auto_response(true); + + return ret; +} + +int SrsRtmpConn::do_playing(SrsSource* source, IsolateRecvThread* trd) +{ + int ret = ERROR_SUCCESS; + if ((ret = refer->check(req->pageUrl, _srs_config->get_refer_play(req->vhost))) != ERROR_SUCCESS) { srs_error("check play_refer failed. ret=%d", ret); return ret; @@ -519,38 +626,19 @@ int SrsRtmpConn::playing(SrsSource* source) bool user_specified_duration_to_stop = (req->duration > 0); int64_t starttime = -1; - // TODO: use isolate thread to recv, - // @see: https://github.com/winlinvip/simple-rtmp-server/issues/196 - // the performance bottleneck not in the timeout recv, but - // in the multiple messages send, so it's ok for timeout recv, - // @see https://github.com/winlinvip/simple-rtmp-server/issues/194 - rtmp->set_recv_timeout(SRS_CONSTS_RTMP_PULSE_TIMEOUT_US); - while (true) { - // TODO: to use isolate thread to recv, can improve about 5% performance. + // to use isolate thread to recv, can improve about 33% performance. // @see: https://github.com/winlinvip/simple-rtmp-server/issues/196 - // read from client. - if (true) { - SrsMessage* msg = NULL; - ret = rtmp->recv_message(&msg); - srs_verbose("play loop recv message. ret=%d", ret); + // @see: https://github.com/winlinvip/simple-rtmp-server/issues/217 + while (!trd->empty()) { + SrsMessage* msg = trd->pump(); + srs_warn("pump client message to process."); - if (ret == ERROR_SOCKET_TIMEOUT) { - // it's ok, do nothing. - ret = ERROR_SUCCESS; - srs_verbose("recv timeout, ignore. ret=%d", ret); - } else if (ret != ERROR_SUCCESS) { - if (!srs_is_client_gracefully_close(ret)) { - srs_error("recv client control message failed. ret=%d", 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); } return 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; - } } } @@ -564,6 +652,12 @@ int SrsRtmpConn::playing(SrsSource* source) srs_error("get messages from consumer failed. ret=%d", ret); return ret; } + + // no message to send, sleep a while. + if (count <= 0) { + srs_verbose("sleep for no messages to send"); + st_usleep(SRS_CONSTS_RTMP_PULSE_TIMEOUT_US); + } // reportable if (pithy_print.can_print()) { @@ -596,7 +690,9 @@ int SrsRtmpConn::playing(SrsSource* source) if (count > 0) { // no need to assert msg, for the rtmp will assert it. if ((ret = rtmp->send_and_free_messages(msgs.msgs, count, res->stream_id)) != ERROR_SUCCESS) { - srs_error("send messages to client failed. ret=%d", ret); + if (!srs_is_client_gracefully_close(ret)) { + srs_error("send messages to client failed. ret=%d", ret); + } return ret; } } diff --git a/trunk/src/app/srs_app_rtmp_conn.hpp b/trunk/src/app/srs_app_rtmp_conn.hpp index 716d32588..4d96be709 100644 --- a/trunk/src/app/srs_app_rtmp_conn.hpp +++ b/trunk/src/app/srs_app_rtmp_conn.hpp @@ -49,6 +49,7 @@ class SrsBandwidth; class SrsKbps; class SrsRtmpClient; class SrsSharedPtrMessage; +class IsolateRecvThread; /** * the client provides the main logic control for RTMP clients. @@ -88,6 +89,7 @@ private: virtual int stream_service_cycle(); virtual int check_vhost(); virtual int playing(SrsSource* source); + virtual int do_playing(SrsSource* source, IsolateRecvThread* trd); virtual int fmle_publishing(SrsSource* source); virtual int do_fmle_publishing(SrsSource* source); virtual int flash_publishing(SrsSource* source); diff --git a/trunk/src/app/srs_app_thread.cpp b/trunk/src/app/srs_app_thread.cpp index bc18333f0..856ff5fa0 100644 --- a/trunk/src/app/srs_app_thread.cpp +++ b/trunk/src/app/srs_app_thread.cpp @@ -161,7 +161,9 @@ void SrsThread::thread_cycle() srs_info("thread on before cycle success"); if ((ret = handler->cycle()) != ERROR_SUCCESS) { - srs_warn("thread cycle failed, ignored and retry, ret=%d", ret); + if (!srs_is_client_gracefully_close(ret)) { + srs_warn("thread cycle failed, ignored and retry, ret=%d", ret); + } goto failed; } srs_info("thread cycle success"); diff --git a/trunk/src/rtmp/srs_protocol_rtmp.cpp b/trunk/src/rtmp/srs_protocol_rtmp.cpp index 1049de187..a4ffd3077 100644 --- a/trunk/src/rtmp/srs_protocol_rtmp.cpp +++ b/trunk/src/rtmp/srs_protocol_rtmp.cpp @@ -735,6 +735,11 @@ SrsRtmpServer::~SrsRtmpServer() srs_freep(hs_bytes); } +void SrsRtmpServer::set_auto_response(bool v) +{ + protocol->set_auto_response(v); +} + void SrsRtmpServer::set_recv_timeout(int64_t timeout_us) { protocol->set_recv_timeout(timeout_us); diff --git a/trunk/src/rtmp/srs_protocol_rtmp.hpp b/trunk/src/rtmp/srs_protocol_rtmp.hpp index 9c8b85d3b..77df8c831 100644 --- a/trunk/src/rtmp/srs_protocol_rtmp.hpp +++ b/trunk/src/rtmp/srs_protocol_rtmp.hpp @@ -334,6 +334,12 @@ public: virtual ~SrsRtmpServer(); // protocol methods proxy public: + /** + * set the auto response message when recv for protocol stack. + * @param v, whether auto response message when recv message. + * @see: https://github.com/winlinvip/simple-rtmp-server/issues/217 + */ + virtual void set_auto_response(bool v); /** * set/get the recv timeout in us. * if timeout, recv/send message return ERROR_SOCKET_TIMEOUT. diff --git a/trunk/src/rtmp/srs_protocol_stack.cpp b/trunk/src/rtmp/srs_protocol_stack.cpp index ef8487983..6f79e202b 100644 --- a/trunk/src/rtmp/srs_protocol_stack.cpp +++ b/trunk/src/rtmp/srs_protocol_stack.cpp @@ -415,6 +415,7 @@ SrsProtocol::SrsProtocol(ISrsProtocolReaderWriter* io) srs_assert(nb_out_iovs >= 2); warned_c0c3_cache_dry = false; + auto_response_when_recv = true; } SrsProtocol::~SrsProtocol() @@ -430,6 +431,15 @@ SrsProtocol::~SrsProtocol() chunk_streams.clear(); } + if (true) { + std::vector::iterator it; + for (it = manual_response_queue.begin(); it != manual_response_queue.end(); ++it) { + SrsPacket* pkt = *it; + srs_freep(pkt); + } + manual_response_queue.clear(); + } + srs_freep(in_buffer); // alloc by malloc, use free directly. @@ -439,6 +449,35 @@ SrsProtocol::~SrsProtocol() } } +void SrsProtocol::set_auto_response(bool v) +{ + auto_response_when_recv = v; +} + +int SrsProtocol::manual_response_flush() +{ + int ret = ERROR_SUCCESS; + + if (manual_response_queue.empty()) { + return ret; + } + + std::vector::iterator it; + for (it = manual_response_queue.begin(); it != manual_response_queue.end();) { + SrsPacket* pkt = *it; + + // erase this packet, the send api always free it. + it = manual_response_queue.erase(it); + + // use underlayer api to send, donot flush again. + if ((ret = do_send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { + return ret; + } + } + + return ret; +} + void SrsProtocol::set_recv_timeout(int64_t timeout_us) { return skt->set_recv_timeout(timeout_us); @@ -638,7 +677,9 @@ int SrsProtocol::do_send_messages(SrsMessage** msgs, int nb_msgs) // when c0c3 cache dry, // sendout all messages and reset the cache, then send again. if ((ret = skt->writev(out_iovs, iov_index, NULL)) != ERROR_SUCCESS) { - srs_error("send with writev failed. ret=%d", ret); + if (!srs_is_client_gracefully_close(ret)) { + srs_error("send with writev failed. ret=%d", ret); + } return ret; } @@ -663,13 +704,57 @@ int SrsProtocol::do_send_messages(SrsMessage** msgs, int nb_msgs) // sendout header and payload by writev. // decrease the sys invoke count to get higher performance. if ((ret = skt->writev(out_iovs, iov_index, NULL)) != ERROR_SUCCESS) { - srs_error("send with writev failed. ret=%d", ret); + if (!srs_is_client_gracefully_close(ret)) { + srs_error("send with writev failed. ret=%d", ret); + } return ret; } return ret; } +int SrsProtocol::do_send_and_free_packet(SrsPacket* packet, int stream_id) +{ + int ret = ERROR_SUCCESS; + + srs_assert(packet); + SrsAutoFree(SrsPacket, packet); + + int size = 0; + char* payload = NULL; + if ((ret = packet->encode(size, payload)) != ERROR_SUCCESS) { + srs_error("encode RTMP packet to bytes oriented RTMP message failed. ret=%d", ret); + return ret; + } + + // encode packet to payload and size. + if (size <= 0 || payload == NULL) { + srs_warn("packet is empty, ignore empty message."); + return ret; + } + + // to message + SrsMessage* msg = new SrsCommonMessage(); + + msg->payload = payload; + msg->size = size; + + msg->header.payload_length = size; + msg->header.message_type = packet->get_message_type(); + msg->header.stream_id = stream_id; + msg->header.perfer_cid = packet->get_prefer_cid(); + + // donot use the auto free to free the msg, + // for performance issue. + ret = do_send_messages(&msg, 1); + if (ret == ERROR_SUCCESS) { + ret = on_send_packet(msg, packet); + } + srs_freep(msg); + + return ret; +} + void SrsProtocol::generate_chunk_header(char* cache, SrsMessageHeader* mh, bool c0, int* pnbh, char** ph) { // to directly set the field. @@ -948,6 +1033,16 @@ int SrsProtocol::send_and_free_messages(SrsMessage** msgs, int nb_msgs, int stre srs_freep(msg); } + // donot flush when send failed + if (ret != ERROR_SUCCESS) { + return ret; + } + + // flush messages in manual queue + if ((ret = manual_response_flush()) != ERROR_SUCCESS) { + return ret; + } + return ret; } @@ -955,41 +1050,15 @@ int SrsProtocol::send_and_free_packet(SrsPacket* packet, int stream_id) { int ret = ERROR_SUCCESS; - srs_assert(packet); - SrsAutoFree(SrsPacket, packet); - - int size = 0; - char* payload = NULL; - if ((ret = packet->encode(size, payload)) != ERROR_SUCCESS) { - srs_error("encode RTMP packet to bytes oriented RTMP message failed. ret=%d", ret); + if ((ret = do_send_and_free_packet(packet, stream_id)) != ERROR_SUCCESS) { return ret; } - // encode packet to payload and size. - if (size <= 0 || payload == NULL) { - srs_warn("packet is empty, ignore empty message."); + // flush messages in manual queue + if ((ret = manual_response_flush()) != ERROR_SUCCESS) { return ret; } - // to message - SrsMessage* msg = new SrsCommonMessage(); - - msg->payload = payload; - msg->size = size; - - msg->header.payload_length = size; - msg->header.message_type = packet->get_message_type(); - msg->header.stream_id = stream_id; - msg->header.perfer_cid = packet->get_prefer_cid(); - - // donot use the auto free to free the msg, - // for performance issue. - ret = do_send_messages(&msg, 1); - if (ret == ERROR_SUCCESS) { - ret = on_send_packet(msg, packet); - } - srs_freep(msg); - return ret; } @@ -1698,7 +1767,15 @@ int SrsProtocol::response_acknowledgement_message() SrsAcknowledgementPacket* pkt = new SrsAcknowledgementPacket(); in_ack_size.acked_size = skt->get_recv_bytes(); pkt->sequence_number = (int32_t)in_ack_size.acked_size; - if ((ret = send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { + + // cache the message and use flush to send. + if (!auto_response_when_recv) { + manual_response_queue.push_back(pkt); + return ret; + } + + // use underlayer api to send, donot flush again. + if ((ret = do_send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { srs_error("send acknowledgement failed. ret=%d", ret); return ret; } @@ -1718,7 +1795,14 @@ int SrsProtocol::response_ping_message(int32_t timestamp) pkt->event_type = SrcPCUCPingResponse; pkt->event_data = timestamp; - if ((ret = send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { + // cache the message and use flush to send. + if (!auto_response_when_recv) { + manual_response_queue.push_back(pkt); + return ret; + } + + // use underlayer api to send, donot flush again. + if ((ret = do_send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { srs_error("send ping response failed. ret=%d", ret); return ret; } diff --git a/trunk/src/rtmp/srs_protocol_stack.hpp b/trunk/src/rtmp/srs_protocol_stack.hpp index 925391fdf..d9ba0ee84 100644 --- a/trunk/src/rtmp/srs_protocol_stack.hpp +++ b/trunk/src/rtmp/srs_protocol_stack.hpp @@ -31,6 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. #include #include +#include #include // for srs-librtmp, @see https://github.com/winlinvip/simple-rtmp-server/issues/213 @@ -213,6 +214,16 @@ private: * input ack size, when to send the acked packet. */ AckWindowSize in_ack_size; + /** + * whether auto response when recv messages. + * default to true for it's very easy to use the protocol stack. + * @see: https://github.com/winlinvip/simple-rtmp-server/issues/217 + */ + bool auto_response_when_recv; + /** + * when not auto response message, manual flush the messages in queue. + */ + std::vector manual_response_queue; // peer out private: /** @@ -244,6 +255,20 @@ public: */ SrsProtocol(ISrsProtocolReaderWriter* io); virtual ~SrsProtocol(); +public: + /** + * set the auto response message when recv for protocol stack. + * @param v, whether auto response message when recv message. + * @see: https://github.com/winlinvip/simple-rtmp-server/issues/217 + */ + virtual void set_auto_response(bool v); + /** + * flush for manual response when the auto response is disabled + * by set_auto_response(false), we default use auto response, so donot + * need to call this api(the protocol sdk will auto send message). + * @see the auto_response_when_recv and manual_response_queue. + */ + virtual int manual_response_flush(); public: /** * set/get the recv timeout in us. @@ -371,6 +396,10 @@ private: */ virtual int do_send_messages(SrsMessage** msgs, int nb_msgs); /** + * underlayer api for send and free packet. + */ + virtual int do_send_and_free_packet(SrsPacket* packet, int stream_id); + /** * generate the chunk header for msg. * @param mh, the header of msg to send. * @param c0, whether the first chunk, the c0 chunk.