@@ -29,6 +29,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
2929#include < srs_core_autofree.hpp>
3030#include < srs_kernel_utility.hpp>
3131
32+ #include < stdlib.h>
3233using namespace std ;
3334
3435// when got a messae header, there must be some data,
@@ -404,7 +405,15 @@ SrsProtocol::SrsProtocol(ISrsProtocolReaderWriter* io)
404405 in_buffer = new SrsBuffer ();
405406 skt = io;
406407
407- in_chunk_size = out_chunk_size = SRS_CONSTS_RTMP_PROTOCOL_CHUNK_SIZE;
408+ in_chunk_size = SRS_CONSTS_RTMP_PROTOCOL_CHUNK_SIZE;
409+ out_chunk_size = SRS_CONSTS_RTMP_PROTOCOL_CHUNK_SIZE;
410+
411+ nb_out_iovs = SRS_CONSTS_IOVS_MAX;
412+ out_iovs = (iovec*)malloc (sizeof (iovec) * nb_out_iovs);
413+ // each chunk consumers atleast 2 iovs
414+ srs_assert (nb_out_iovs >= 2 );
415+
416+ warned_c0c3_caches = false ;
408417}
409418
410419SrsProtocol::~SrsProtocol ()
@@ -421,6 +430,12 @@ SrsProtocol::~SrsProtocol()
421430 }
422431
423432 srs_freep (in_buffer);
433+
434+ // alloc by malloc, use free directly.
435+ if (out_iovs) {
436+ free (out_iovs);
437+ out_iovs = NULL ;
438+ }
424439}
425440
426441void SrsProtocol::set_recv_timeout (int64_t timeout_us)
@@ -560,7 +575,7 @@ int SrsProtocol::do_send_message(SrsMessage* msg)
560575 // always has header
561576 int nbh = 0 ;
562577 char * header = NULL ;
563- generate_chunk_header (&msg->header , p == msg->payload , &nbh, &header);
578+ generate_chunk_header (out_c0c3_cache, &msg->header , p == msg->payload , &nbh, &header);
564579 srs_assert (nbh > 0 );
565580
566581 // header iov
@@ -590,10 +605,130 @@ int SrsProtocol::do_send_message(SrsMessage* msg)
590605 return ret;
591606}
592607
593- void SrsProtocol::generate_chunk_header (SrsMessageHeader* mh, bool c0, int * pnbh, char ** ph)
608+ int SrsProtocol::do_send_messages (SrsSharedPtrMessage** msgs, int nb_msgs)
594609{
595- char * cache = out_c0c3_cache;
610+ int ret = ERROR_SUCCESS;
611+
612+ // TODO: FIXME: use cache system instead.
613+ int iov_index = 0 ;
614+ iovec* iov = out_iovs + iov_index;
615+
616+ int c0c3_cache_index = 0 ;
617+ char * c0c3_cache = out_c0c3_caches + c0c3_cache_index;
618+
619+ // try to send use the c0c3 header cache,
620+ // if cache is consumed, try another loop.
621+ for (int i = 0 ; i < nb_msgs; i++) {
622+ SrsMessage* msg = msgs[i];
623+
624+ // ignore empty message.
625+ if (!msg->payload || msg->size <= 0 ) {
626+ srs_info (" ignore empty message."
627+ continue ;
628+ }
629+
630+ // we donot use the complex basic header,
631+ // ensure the basic header is 1bytes.
632+ if (msg->header .perfer_cid < 2 ) {
633+ srs_warn (" change the chunk_id=%d to default=%d"
634+ msg->header .perfer_cid , RTMP_CID_ProtocolControl);
635+ msg->header .perfer_cid = RTMP_CID_ProtocolControl;
636+ }
637+
638+ // p set to current write position,
639+ // it's ok when payload is NULL and size is 0.
640+ char * p = msg->payload ;
641+ char * pend = msg->payload + msg->size ;
642+
643+ // always write the header event payload is empty.
644+ while (p < pend) {
645+ // always has header
646+ int nbh = 0 ;
647+ char * header = NULL ;
648+ generate_chunk_header (c0c3_cache, &msg->header , p == msg->payload , &nbh, &header);
649+ srs_assert (nbh > 0 );
650+
651+ // header iov
652+ iov[0 ].iov_base = header;
653+ iov[0 ].iov_len = nbh;
654+
655+ // payload iov
656+ int payload_size = pend - p;
657+ if (payload_size > out_chunk_size) {
658+ payload_size = out_chunk_size;
659+ }
660+ iov[1 ].iov_base = p;
661+ iov[1 ].iov_len = payload_size;
662+
663+ // consume sendout bytes.
664+ p += payload_size;
665+
666+ // realloc the iovs if exceed,
667+ // for we donot know how many messges maybe to send entirely,
668+ // we just alloc the iovs, it's ok.
669+ if (iov_index >= nb_out_iovs - 2 ) {
670+ nb_out_iovs += SRS_CONSTS_IOVS_MAX;
671+ int realloc_size = sizeof (iovec) * nb_out_iovs;
672+ out_iovs = (iovec*)realloc (out_iovs, realloc_size);
673+ }
674+
675+ // to next pair of iovs
676+ iov_index += 2 ;
677+ iov = out_iovs + iov_index;
678+
679+ // to next c0c3 header cache
680+ c0c3_cache_index += nbh;
681+ c0c3_cache = out_c0c3_caches + c0c3_cache_index;
682+
683+ // the cache header should never be realloc again,
684+ // for the ptr is set to iovs, so we just warn user to set larger
685+ // and use another loop to send again.
686+ int c0c3_left = SRS_CONSTS_C0C3_HEADERS_MAX - c0c3_cache_index;
687+ if (c0c3_left < SRS_CONSTS_RTMP_MAX_FMT0_HEADER_SIZE) {
688+ // only warn once for a connection.
689+ if (!warned_c0c3_caches) {
690+ srs_warn (" c0c3 cache header too small, recoment to %d"
691+ SRS_CONSTS_C0C3_HEADERS_MAX + SRS_CONSTS_RTMP_MAX_FMT0_HEADER_SIZE);
692+ warned_c0c3_caches = true ;
693+ }
694+
695+ // when c0c3 cache dry,
696+ // sendout all messages and reset the cache, then send again.
697+ if ((ret = skt->writev (out_iovs, iov_index, NULL )) != ERROR_SUCCESS) {
698+ srs_error (" send with writev failed. ret=%d"
699+ return ret;
700+ }
596701
702+ // reset caches, while these cache ensure
703+ // atleast we can sendout a chunk.
704+ iov_index = 0 ;
705+ iov = out_iovs + iov_index;
706+
707+ c0c3_cache_index = 0 ;
708+ c0c3_cache = out_c0c3_caches + c0c3_cache_index;
709+ }
710+ }
711+ }
712+
713+ // maybe the iovs already sendout when c0c3 cache dry,
714+ // so just ignore when no iovs to send.
715+ if (iov_index <= 0 ) {
716+ return ret;
717+ }
718+
719+ // send by writev
720+ // sendout header and payload by writev.
721+ // decrease the sys invoke count to get higher performance.
722+ if ((ret = skt->writev (out_iovs, iov_index, NULL )) != ERROR_SUCCESS) {
723+ srs_error (" send with writev failed. ret=%d"
724+ return ret;
725+ }
726+
727+ return ret;
728+ }
729+
730+ void SrsProtocol::generate_chunk_header (char * cache, SrsMessageHeader* mh, bool c0, int * pnbh, char ** ph)
731+ {
597732 // to directly set the field.
598733 char * pp = NULL ;
599734
@@ -856,6 +991,34 @@ int SrsProtocol::send_and_free_message(SrsMessage* msg, int stream_id)
856991 return ret;
857992}
858993
994+ int SrsProtocol::send_and_free_messages (SrsSharedPtrMessage** msgs, int nb_msgs, int stream_id)
995+ {
996+ // always not NULL msg.
997+ srs_assert (msgs);
998+ srs_assert (nb_msgs > 0 );
999+
1000+ // update the stream id in header.
1001+ for (int i = 0 ; i < nb_msgs; i++) {
1002+ SrsMessage* msg = msgs[i];
1003+ // we assume that the stream_id in a group must be the same.
1004+ if (msg->header .stream_id == stream_id) {
1005+ break ;
1006+ }
1007+ msg->header .stream_id = stream_id;
1008+ }
1009+
1010+ // donot use the auto free to free the msg,
1011+ // for performance issue.
1012+ int ret = do_send_messages (msgs, nb_msgs);
1013+
1014+ for (int i = 0 ; i < nb_msgs; i++) {
1015+ SrsMessage* msg = msgs[i];
1016+ srs_freep (msg);
1017+ }
1018+
1019+ return ret;
1020+ }
1021+
8591022int SrsProtocol::send_and_free_packet (SrsPacket* packet, int stream_id)
8601023{
8611024 int ret = ERROR_SUCCESS;
0 commit comments