2. 程式人生 > >Doubango RTP包傳輸使用UDT可靠傳輸協議，解決RTP丟包問題

Doubango RTP包傳輸使用UDT可靠傳輸協議，解決RTP丟包問題

阿新 • • 發佈：2018-11-04
使用SIP做過VOIP通話的同學，肯定被RTP丟包弄的焦頭爛額，必定嘗試過不少的辦法，比方：
1、丟包重傳(NACK)
2、前向糾錯(FEC)
3、丟幀處理
但效果往往不盡如人意，那有沒有一勞永逸的方法？確實，網路發展到現在，家庭頻寬隨隨便便都是百兆的今天，為什麼還有讓人困擾不已的丟包問題？為何不換成TCP傳輸，假如延時容許的情況下，UDT就是為了解決這個問題的。

UDT建立於UDP之上，並引入新的擁塞控制和資料可靠性控制機制。UDT是面向連線的雙向的應用層協議。它同時支援可靠的資料流傳輸和部分可靠的資料報傳輸，UDT的特點是不用進行開發，直接利用庫的傳送函式就可以實現可靠的資料傳輸。

Doubango rtp接收和傳送函式修改：

/* use transport udt*/
#include "udt_api.h"

void *udt_transport_mainthread(void *param)
{
    trtp_manager_t *self = param;


    if(!self){
        TSK_DEBUG_ERROR("udt_transport_mainthread Invalid parameter");
        return 0;
    }
    TSK_DEBUG_INFO("ii udt_transport_mainthread enter, self->is_originated:%d", self->is_originated);

    int rcv_buf = (int)tmedia_defaults_get_rtpbuff_size();
    int snd_buf = (int)tmedia_defaults_get_rtpbuff_size();

    int clientFD = -1;               
    int socketfDUDT = -1;         
    int clientSocket = -1; 
    if (self->is_originated){
        //first listen
        int ret = init_udt(&socketfDUDT, (self->rtp.public_port), 0);
        self->serverSocket = socketfDUDT;
        TSK_DEBUG_INFO("socketfDUDT:%d listent", socketfDUDT);

        set_connect_buff_udt(&socketfDUDT, rcv_buf, snd_buf);//set buff size
        
        listen_udt(&socketfDUDT);        
        clientFD = accept_udt(&socketfDUDT);
        if (clientFD == -1){
            TSK_DEBUG_ERROR("clientFd is failed");
            tsk_safeobj_lock(self);
            self->mainThreadId[0] = NULL;
            tsk_safeobj_unlock(self);//
 
[email protected] add for crash.

            goto FAILED;
        } 
        TSK_DEBUG_INFO("clientFD:%d enter", clientFD);

        if (!tmedia_defaults_get_single_connect()){
            //after connect 2 user
            ret = init_udt(&clientSocket, (self->rtp.public_port + 3), 0);
            set_connect_buff_udt(&clientSocket, rcv_buf, snd_buf);//set buff size
        
            if (0 != ret || clientSocket == -1){
                TSK_DEBUG_ERROR("clientSocket connect init is failed");
                tsk_safeobj_lock(self);
                self->mainThreadId[0] = NULL;
                tsk_safeobj_unlock(self);//
 
[email protected] add for crash.

                goto FAILED;
            }
            ret = connect_udt(&clientSocket, self->rtp.remote_ip, self->rtp.remote_port);
            if (-1 == ret){
                TSK_DEBUG_ERROR("clientSocket connect is failed");
                tsk_safeobj_lock(self);
                self->mainThreadId[0] = NULL;
                tsk_safeobj_unlock(self);//
 
[email protected] add for crash.

                goto FAILED;
            }        
            TSK_DEBUG_INFO("clientSocket:%d enter", clientSocket);

            self->clientSocket = clientSocket;
        }
    }else{
        //first connect 2 user
        int ret = init_udt(&clientSocket, (self->rtp.public_port + 3), 0);
        if (0 != ret || clientSocket == -1){
            TSK_DEBUG_ERROR("ii clientSocket connect init is failed");
            tsk_safeobj_lock(self);
            self->mainThreadId[0] = NULL;
            tsk_safeobj_unlock(self);//[email protected] add for crash.

            goto FAILED;
        }
        set_connect_buff_udt(&clientSocket, rcv_buf, snd_buf);//set buff size
        ret = connect_udt(&clientSocket, self->rtp.remote_ip, self->rtp.remote_port);
        if (-1 == ret){
            TSK_DEBUG_ERROR("clientSocket connect is failed");
            tsk_safeobj_lock(self);
            self->mainThreadId[0] = NULL;
            tsk_safeobj_unlock(self);//[email protected] add for crash.

            goto FAILED;
        }
        self->clientSocket = clientSocket;
        TSK_DEBUG_INFO("ii clientSocket:%d enter", clientSocket);
        
        if (!tmedia_defaults_get_single_connect()){
            //second listen
            ret = init_udt(&socketfDUDT, (self->rtp.public_port), 0);
            self->serverSocket = socketfDUDT;
            TSK_DEBUG_INFO("ii socketfDUDT:%d listent", socketfDUDT);
            
            set_connect_buff_udt(&socketfDUDT, rcv_buf, snd_buf);//set buff size
            listen_udt(&socketfDUDT);
            clientFD = accept_udt(&socketfDUDT);
            if (clientFD == -1){
                TSK_DEBUG_ERROR("clientFd is failed");
                tsk_safeobj_lock(self);
                self->mainThreadId[0] = NULL;
                tsk_safeobj_unlock(self);//[email protected] add for crash.

                goto FAILED;
            }         
            TSK_DEBUG_INFO("ii clientFD:%d enter", clientFD);
        }else{
                tsk_safeobj_lock(self);
                self->mainThreadId[0] = NULL;
                tsk_safeobj_unlock(self);//[email protected] add for crash.
        } 
    }

    if (tmedia_defaults_get_single_connect()){
        if (self->is_originated){
            char buffer[4096];
            while(self->is_started){
                int received = recvdata_udt(&clientFD, buffer, sizeof(buffer), 0);

                if (received <= -1) {
                    continue;
                }

                _trtp_manager_recv_data(self, buffer, received, -1, NULL);
            }
        }

    }else{
            char buffer[4096];
            while(self->is_started){
                int received = recvdata_udt(&clientFD, buffer, sizeof(buffer), 0);

                if (received <= -1) {
                    continue;
                }

                _trtp_manager_recv_data(self, buffer, received, -1, NULL);
            }
    }

FAILED:        
    if (-1 !=clientFD){
        close_udt(clientFD);
    }
    if (-1 !=socketfDUDT){
        close_udt(socketfDUDT);
    }
    
    TSK_DEBUG_INFO("udt_transport_mainthread end.");
    return 0;
}

int udt_transport_send(trtp_manager_t* self, char* encodedData, int encodedDataLen){
    TSK_DEBUG_INFO("senddata_udt START:%d", encodedDataLen);

    int ret = senddata_udt(&self->clientSocket, encodedData, encodedDataLen, 0);
    if (-1 == ret){
	TSK_DEBUG_ERROR("senddata_udt failed");
        return -1;
    }
    TSK_DEBUG_INFO("senddata_udt ret:%d", ret);
    
    return ret;
}
/* add end */

啟動：在 int trtp_manager_start(trtp_manager_t* self)中： 

	if(!self->transport || !self->transport->master){
              //use other transport --udt:
              if (tmedia_defaults_get_use_udt_socket() != 0){
                    //use other transport.
                    self->clientSocket = -1;
                    self->serverSocket = -1;
                    self->is_started = tsk_true;

                    if ((ret = tsk_thread_create(self->mainThreadId, udt_transport_mainthread, self))){ /* More important than "tsk_runnable_start" ==> start it first. */
                        TSK_DEBUG_FATAL("Failed to create udt main thread [%d]", ret);
                    }else{           
		         TSK_DEBUG_INFO("udt thread create success.");      	
                 
#if !TNET_UNDER_APPLE
                        ret = tsk_thread_set_priority(self->mainThreadId[0], TSK_THREAD_PRIORITY_TIME_CRITICAL);
#endif
                        tsk_safeobj_unlock(self);
   
                        return 0;
                    }
              }
              //add end.
              
		TSK_DEBUG_ERROR("RTP/RTCP manager not prepared");
		ret = -2;
		goto bail;
	}



UDT協議封裝：

int init_udt(int *usock, int port, bool rendezvous);
int connect_udt(void* usocket, char *peerip, int port);
int set_connect_buff_udt(void* usocket, int send_buf_size, int recv_buf_size);
int recvdata_udt(void* usocket, char* buf, int size, int flags);
int senddata_udt(void* usocket, char* buf, int size, int flags);
int listen_udt(void* usocket);
int accept_udt(void* usocket);
int close_udt(int usock);

實現：  

#include <winsock2.h>
#include <ws2tcpip.h>
#include <wspiapi.h>
#include <iostream>
#include <udt.h>
#include <arpa/inet.h> 
#include <android/log.h>
#define TAG "UDT_API"

extern "C" int init_udt(int *usock, int port, bool rendezvous);
extern "C" int connect_udt(void* usocket, char *peerip, int port);
extern "C" int set_connect_buff_udt(void* usocket, int send_buf_size, int recv_buf_size);
extern "C" int recvdata_udt(void* usocket, char* buf, int size, int flags);
extern "C" int senddata_udt(void* usocket, char* buf, int size, int flags);
extern "C" int listen_udt(void* usocket);
extern "C" int accept_udt(void* usocket);
extern "C" int close_udt(int usock);

using namespace std;

const int g_IP_Version = AF_INET;
const int g_Socket_Type = SOCK_DGRAM;
const char g_Localhost[] = "127.0.0.1";
  
int init_udt(int *usock, int port, bool rendezvous){
	addrinfo hints;
	addrinfo* res;
	memset(&hints, 0, sizeof(struct addrinfo));
	hints.ai_flags = AI_PASSIVE;
	hints.ai_family = AF_INET;
	hints.ai_socktype = g_Socket_Type;

	if (usock == NULL){
		return -1;
	}

	char service[16];
	sprintf(service, "%d", port);

	if (0 != getaddrinfo(NULL, service, &hints, &res))
	{
		cout << "illegal port number or port is busy.\n" << endl;        	
             __android_log_print(ANDROID_LOG_WARN, TAG, "illegal port number or port is busy:%s", UDT::getlasterror().getErrorMessage() );

		return -1;
	}

	UDTSOCKET retFD = UDT::socket(res->ai_family, res->ai_socktype, res->ai_protocol);

	// since we will start a lot of connections, we set the buffer size to smaller value.
	int snd_buf = 16000;
	int rcv_buf = 16000;
	UDT::setsockopt(retFD, 0, UDT_SNDBUF, &snd_buf, sizeof(int));
	UDT::setsockopt(retFD, 0, UDT_RCVBUF, &rcv_buf, sizeof(int)); 
	//int fc = 16;
	//UDT::setsockopt(retFD, 0, UDT_FC, &fc, sizeof(int));
	bool reuse = true;
	UDT::setsockopt(retFD, 0, UDT_REUSEADDR, &reuse, sizeof(bool));
	UDT::setsockopt(retFD, 0, UDT_RENDEZVOUS, &rendezvous, sizeof(bool));

	if (UDT::ERROR == UDT::bind(retFD, res->ai_addr, res->ai_addrlen))
	{
        	__android_log_print(ANDROID_LOG_WARN, TAG, "err, init failed:%s", UDT::getlasterror().getErrorMessage() );
		cout << "bind: " << UDT::getlasterror().getErrorMessage() << endl;
		return -1;
	} 

	*usock = retFD;//set output socket fd.

	freeaddrinfo(res);
	return 0;
}
 
int connect_udt(void* usocket, char *peerip, int port){
	if (usocket == NULL || peerip == NULL){
		return -1;
	}
	UDTSOCKET usock = *(UDTSOCKET*)usocket;
	
	addrinfo hints, *peer;
	memset(&hints, 0, sizeof(struct addrinfo));
	hints.ai_flags = AI_PASSIVE;
	hints.ai_family =  g_IP_Version;
	hints.ai_socktype = g_Socket_Type;

	char buffer[16];
	sprintf(buffer, "%d", port);
 
	if (0 != getaddrinfo(peerip, buffer, &hints, &peer))
	{
		return -1;
	}

       int newsock = UDT::connect(usock, peer->ai_addr, peer->ai_addrlen);

	freeaddrinfo(peer);
	return newsock;
}

int recvdata_udt(void* usocket, char* buf, int size, int flags){	
	if (usocket == NULL){
		return -1;
	}
	UDTSOCKET recver = *(UDTSOCKET*)usocket;  
	int rs;  
	//if (UDT::ERROR == (rs = UDT::recv(recver, buf, size, 0)))
	if (UDT::ERROR == (rs = UDT::recvmsg(recver, buf, size)))
	{
		cout << "recv:" << UDT::getlasterror().getErrorMessage() << endl;
              if (UDT::getlasterror().getErrorCode() == 2001 || UDT::getlasterror().getErrorCode() == 5004){
                  return  -1;
              }
        	__android_log_print(ANDROID_LOG_WARN, TAG, "err, recvdata_udt failed:%s", UDT::getlasterror().getErrorMessage() );
		return -1;
	} 
    return rs;
}

int set_connect_buff_udt(void* usocket, int send_buf_size, int recv_buf_size){
	if (usocket == NULL){
		return -1;
	}
	UDTSOCKET sock = *(UDTSOCKET*)usocket;   
	UDT::setsockopt(sock, 0, UDT_SNDBUF, &send_buf_size, sizeof(int));
	UDT::setsockopt(sock, 0, UDT_RCVBUF, &recv_buf_size, sizeof(int)); 
	
	return 0;
}

int senddata_udt(void* usocket, char* buf, int size, int flags){
	   
    if (usocket == NULL){
	   return -1;
    }
	UDTSOCKET client = *(UDTSOCKET*)usocket;
	
	int sent = 0; 
    while (sent < size){
        //int ret = UDT::send(client, (((const char*)buf) + sent), (int)(size - sent), flags);
        int ret = UDT::sendmsg(client, (((const char*)buf) + sent), (int)(size - sent), -1, true);
        if (sent < 0)
        {        	
            __android_log_print(ANDROID_LOG_WARN, TAG, "err, senddata_udt failed:%s", UDT::getlasterror().getErrorMessage() );

            cout << "send: " << UDT::getlasterror().getErrorMessage() << endl;
            return -1;
        } 
        else{
            __android_log_print(ANDROID_LOG_WARN, TAG, " senddata_udt ret:%d, size:%d", ret, size);

            sent += ret;
        }
    }
	
    return sent;
}
  
int listen_udt(void* usocket){
    if (usocket == NULL){
	   return -1;
    }
	UDTSOCKET serv = *(UDTSOCKET*)usocket; 
	
	return UDT::listen(serv, 1024);
}

int accept_udt(void* usocket){
    if (usocket == NULL){
	   return -1;
    }
	UDTSOCKET serv = *(UDTSOCKET*)usocket; 
	sockaddr_in  clientaddr;
	int addrlen = sizeof(clientaddr);
	UDTSOCKET new_sock = UDT::accept(serv, (sockaddr*)&clientaddr, &addrlen); 

	if (new_sock == UDT::INVALID_SOCK)
	{
		return -1;
	}
	char ip[16];
	cout << "new connection: " << inet_ntoa(clientaddr.sin_addr) << ":" << ntohs(clientaddr.sin_port) << endl;

	
	return new_sock;
}

int close_udt(int usock){
	UDT::close(usock);
	return 0;
}



相關推薦

Doubango RTP包傳輸使用UDT可靠傳輸協議，解決RTP丟包問題

使用SIP做過VOIP通話的同學，肯定被RTP丟包弄的焦頭爛額，必定嘗試過不少的辦法，比方： 1、丟包重傳(NACK) 2、前向糾錯(FEC) 3、丟幀處理 但效果往往不盡如人意，那有沒有一勞永逸的方法？確實，網路發展到現在，家庭頻寬隨隨便便都是百兆的今天，為什麼還有讓人困擾不已

Wireshark抓包分析Yersinia 根據DHCP協議傳送的偽造資料包攻擊（sending RAW packet）

抓包結果如下： 抓包與分析 幀號時間源地址目的地址高層協議包內資訊概況No.  Time   Source        Destination      Protocol    Info 5  1

TCP協議的定義和丟包時的重傳機制

TCP是一個巨複雜的協議，因為他要解決很多問題，而這些問題又帶出了很多子問題和陰暗面。所以學習TCP本身是個比較痛苦的過程，但對於

使用Windows 2008R2中的NFS替代Samba協議，解決Windows 與Linux共享文件的問題

acc 無法 xen 右鍵 nfs服務 sina 存儲 選擇 顯示 一、在Windows服務器上進行安裝NFS服務 首先，打開服務管理器，選擇添加角色； 選中文件服務，下一步； 出現一個提示，不管它，繼續下一步； 在接下來

Dart Socket 程式設計，通過使用JSON方式，解決業務粘包的問題的最佳實踐

一、背景 Socket程式設計程式設計主用於資料交換，而粘包的問題，其實本身不是問題，TCP已經對於傳輸的封包進行了很好的處理，業務粘包，只是業務處理上的問題，網路上很多處理方法，最常見的有以下幾種： 定義業務傳輸頭，在頭裡面描述了開始識別符號，再加資料長度，如0xAA

程式包org.apache.http不存在，解決方式

類裡提示爆紅 import org.apache.http.HttpEntity; import org.apache.http.HttpResponse; import org.apache.http.StatusLine; import org.apache.http.

用好這幾個技巧，解決Maven Jar包衝突易如反掌

## 前言 大家在專案中肯定有碰到過`Maven`的Jar包衝突問題，經常出現的場景為： 本地執行報`NoSuchMethodError`，`ClassNotFoundException`。明明在依賴裡有這個Jar包啊。怎麼執行不了！？ 專案中明明定義著某個jar包版本為`2.0.2`，怎麼打包之後變成

UDT:基於UDP的可靠傳輸協議

 基於UDP 上的UDT ，比TCP傳輸效率高 UDT:基於UDP的資料傳輸協議(初譯) (譯者:Jack) Status of this Memo Internet-Drafts are working documents of th

基於位圖索引的無線傳感器網絡可靠傳輸協議_愛學術

線圖 nbsp shu www mage search 可靠 com 分享圖片 【摘要】根據無線傳感器網絡中大量上行數據流實時傳輸的需求,提出一種基於位圖索引的可靠傳輸協議(BRDT)。該協議使用叠代方式完成一組大量數據的傳輸。每次叠代的上行數據流采用無重傳傳輸以減少延遲,

用wireshark抓包分析TCP三次握手、四次揮手以及TCP實現可靠傳輸的機制（轉）

關於TCP三次握手和四次揮手大家都在《計算機網路》課程裡學過，還記得當時高超老師耐心地講解。大學裡我遇到的最好的老師大概就是這位了，雖然他只給我講過《java程式設計》和《計算機網路》，但每次課幾乎都動手敲程式碼或者當場做實驗。好了不扯了，下面進入正題。      關

我是如何講清楚TCP協議是如何保證可靠傳輸的

1、UDP： （1）UDP，user datagram protocol，使用者資料報協議，不提供複雜的控制機制，利用IP提供面向無連線的通訊服務，並且它是將應用程式傳送過來的資料包在收到的那一刻，立即按照原樣傳送到上的一種機制。 （2）即使在網路擁堵的情況下，UDP也無法進行流量控制等避免網路

TCP-可靠傳輸的實現-滑動視窗協議

 TCP協議作為一個可靠的面向流的傳輸協議，其可靠性和流量控制由滑動視窗協議保證，而擁塞控制則由控制視窗結合一系列的控制演算法實現。一、滑動視窗協議     關於這部分自己不曉得怎麼敘述才好，因為理解的部分更多，下面就用自己的理解來介紹下TCP的精髓：滑動視窗協議。   

計算機網路課程實驗——可靠傳輸協議：（1）問題描述

Programming Assignment #2: Implementing a Reliable Transport Protocol Overview In this assignmentoratory programming assignment, you wi

TCP協議如何實現可靠傳輸

1. 可靠傳輸的工作原理————停止等待協議 1. 使用確認和重傳機制，我們就可以在不可靠的傳輸網路上實現可靠的通訊 2. 這種可靠傳輸協議常稱為自動重傳請求ARQ(Automatic Repeat reQuest)。

計算機網路課程實驗——可靠傳輸協議：（3）GBN（unidirectional）

/* Zhang Jingtun(Ordinary Crazy) */ #include <stdio.h> #include <stdlib.h> #include <string.h> //#include <time.h&g

計算機網路課程實驗——可靠傳輸協議：（2）ABP（unidirectional）

/* Zhang Jingtun(Ordinary Crazy) */ #include <stdio.h> #include <stdlib.h> #include <string.h> /* *******************

TCP協議如何保證可靠傳輸

1、應用資料被分割成TCP認為最適合傳送的資料塊。 2、超時重傳：當TCP發出一個段後，它啟動一個定時器，等待目的端確認收到這個報文段。如果不能及時收到一個確認，將重發這個報文段。 3、TCP給傳送

《TCP/IP協議族》：TCP可靠傳輸的實現

可靠傳輸 為方便描述可靠傳輸原理，假定資料傳輸只在一個方向上進行，即A傳送資料，B給出確認 1. 以位元組為單位的滑動視窗 TCP的滑動視窗是以位元組為單位的。為了便於說明，位元組編號取得很小。先假定A收到B發來的確認報文欄位，其中視窗是20位元組，而

26-tcp可靠傳輸——停止等待協議

1. tcp可靠傳輸   通過前面的學習可知，網路層傳輸資料時是盡最大努力傳輸到目的地，並不保障資料的可靠傳輸，對於網路擁塞，延遲，資料丟失等問題沒有采取有效的措施。因此我們需要一種資料可靠傳輸的通訊方式，即tcp來實現傳送端和接收端之間的可靠通訊。   那

UDT--UDP可靠傳輸

官網 http://udt.sourceforge.net/ https://www.cnblogs.com/lidabo/p/6846678.html http://jimmee.iteye.com/blog/2037451