From 75a8aa318ab13be27697e2119102508bbbd6d079 Mon Sep 17 00:00:00 2001 From: Raphael Coeffic Date: Sat, 29 Apr 2006 16:51:16 +0000 Subject: [PATCH] - added adaptive playout buffer with order statistic based delay estimation and adaptive WSOLA playout delay modification. - fixed serious bug in the RTP receiver. git-svn-id: http://svn.berlios.de/svnroot/repos/sems/trunk@41 8eb893ce-cfd4-0310-b710-fb5ebe64c474 --- core/AmPlayoutBuffer.cpp | 341 +++++++++++++++++++++++++++++++++++++++ core/AmPlayoutBuffer.h | 58 +++++++ core/AmRtpAudio.cpp | 14 +- core/AmRtpAudio.h | 6 +- core/AmRtpReceiver.cpp | 7 +- core/AmRtpStream.cpp | 15 +- core/AmStats.h | 92 +++++++++++ 7 files changed, 514 insertions(+), 19 deletions(-) create mode 100644 core/AmPlayoutBuffer.cpp create mode 100644 core/AmPlayoutBuffer.h create mode 100644 core/AmStats.h diff --git a/core/AmPlayoutBuffer.cpp b/core/AmPlayoutBuffer.cpp new file mode 100644 index 00000000..d4e3835c --- /dev/null +++ b/core/AmPlayoutBuffer.cpp @@ -0,0 +1,341 @@ +#include "AmPlayoutBuffer.h" + +#define PACKET_SAMPLES 160 +#define PACKET_SIZE (PACKET_SAMPLES<<1) +#define BUFFER_SIZE (5*PACKET_SIZE) + +#define SEARCH_OFFSET 140 + +#define SEARCH_REGION 110 +#define TEMPLATE_SEG 80 +#define DELTA 5 + +#define TSM_MAX_SCALE 2.0 +#define TSM_MIN_SCALE 0.5 + +#define PI 3.14 + + +AmPlayoutBuffer::AmPlayoutBuffer() + : idx(0), + loss_rate(ORDER_STAT_LOSS_RATE), + wsola_off(WSOLA_START_OFF), + shr_threshold(SHR_THRESHOLD), + plc_cnt(0), + short_scaled(WSOLA_SCALED_WIN) +{ + memset(n_stat,0,sizeof(int32_t)*ORDER_STAT_WIN_SIZE); +} + +u_int32_t AmPlayoutBuffer::next_delay(u_int32_t ref_ts, u_int32_t ts) +{ + int32_t n = (int32_t)(ref_ts - ts); + + multiset::iterator it = o_stat.find(n_stat[idx]); + if(it != o_stat.end()) + o_stat.erase(it); + + n_stat[idx] = n; + o_stat.insert(n); + + + int32_t D_r=0,D_r1=0; + int r = int((double(o_stat.size()) + 1.0)*(1.0 - loss_rate)); + + if((r == 0) || (r >= (int)o_stat.size())){ + + StddevValue n_std; + for(int i=0; i= ORDER_STAT_WIN_SIZE) + idx = 0; + + return D; +} + +void AmPlayoutBuffer::write(u_int32_t ref_ts, u_int32_t ts, + int16_t* buf, u_int32_t len) +{ + u_int32_t p_delay = next_delay(ref_ts,ts); + + u_int32_t old_off = wsola_off; + ts += old_off; + + if(short_scaled.mean() > 2.0){ + if(shr_threshold < 3000) + shr_threshold += 10; + } + else if(short_scaled.mean() < 1.0){ + if(shr_threshold > 100) + shr_threshold -= 2; + } + + if( ts_less()(wsola_off+EXP_THRESHOLD,p_delay) || // expand packet + ts_less()(p_delay+shr_threshold,wsola_off) ) { // shrink packet + + wsola_off = p_delay; + } + else { + if(ts_less()(r_ts,ts+len)){ + plc_cnt = 0; + buffer_put(ts,buf,len); + } + else { + // lost + } + + // statistics + short_scaled.push(0.0); + + return; + } + + int32_t n_len = len + wsola_off - old_off; + if(n_len < 0) + n_len = 1; + + float f = float(n_len) / float(len); + if(f > TSM_MAX_SCALE) + f = TSM_MAX_SCALE; + + n_len = (int32_t)(float(len) * f); + if(ts_less()(ts+n_len,r_ts)){ + + // statistics + short_scaled.push(0.0); + return; + } + + u_int32_t old_wts = w_ts; + buffer_put(ts,buf,len); + + n_len = time_scale(ts,f); + wsola_off = old_off + n_len - len; + + //ts += n_len - len; + + if(w_ts != old_wts) + plc_cnt = 0; + + // statistics + short_scaled.push(100.0); +} + +u_int32_t AmPlayoutBuffer::read(u_int32_t ts, int16_t* buf, u_int32_t len) +{ + bool do_plc=false; + + if(ts_less()(w_ts,ts+len) && (plc_cnt < 6)){ + + if(!plc_cnt){ + time_scale(w_ts-len,2.0); + } + else { + do_plc = true; + } + plc_cnt++; + } + + if(do_plc){ + + short plc_buf[FRAMESZ]; + + for(unsigned int i=0; i<2; i++){ + + fec.dofe(plc_buf); + buffer_put(w_ts,plc_buf,FRAMESZ); + } + + buffer_get(ts,buf,len); + } + else { + + buffer_get(ts,buf,len); + + for(unsigned int i=0; i<2; i++) + fec.addtohistory(buf + i*FRAMESZ); + } + + return len; +} + +void AmPlayoutBuffer::direct_write(unsigned int ts, ShortSample* buf, unsigned int len) +{ + buffer_put(ts+wsola_off,buf,len); +} + +void AmPlayoutBuffer::buffer_put(unsigned int ts, ShortSample* buf, unsigned int len) +{ + buffer.put(ts,buf,len); + + if(ts_less()(w_ts,ts+len)) + w_ts = ts + len; +} + +void AmPlayoutBuffer::buffer_get(unsigned int ts, ShortSample* buf, unsigned int len) +{ + buffer.get(ts,buf,len); + + if(ts_less()(r_ts,ts+len)) + r_ts = ts + len; +} + + +short* find_best_corr(short *ts, short *sr_beg,short* sr_end) +{ + // find best correlation + float corr=0.f,best_corr=0.f; + short *best_sr=0; + short *sr; + + for(sr = sr_beg; sr != sr_end; sr++){ + + corr=0.f; + for(int i=0; i best_corr)){ + best_corr = corr; + best_sr = sr; + } + } + + return best_sr; +} + +unsigned int AmPlayoutBuffer::time_scale(unsigned int ts, float factor) +{ + short p_buf[PACKET_SAMPLES*4]; + short merge_buf[TEMPLATE_SEG]; + + short *tmpl = p_buf + PACKET_SAMPLES; + short *p_buf_beg = p_buf; + short *p_buf_end; + + unsigned int s = PACKET_SAMPLES; + unsigned int s_all = s + PACKET_SAMPLES; + + unsigned int cur_ts = ts; + unsigned int begin_ts = cur_ts-PACKET_SAMPLES; + + if(factor == 1.0) + return s; + + if(factor > TSM_MAX_SCALE) + factor = TSM_MAX_SCALE; + else if(factor < TSM_MIN_SCALE) + factor = TSM_MIN_SCALE; + + short *srch_beg, *srch_end, *srch; + + while(true){ + + buffer_get(begin_ts,p_buf_beg,s_all); + p_buf_end = p_buf_beg + s_all; + + if (factor > 1.0){ + + // expansion + srch_beg = tmpl - (int)((float)TEMPLATE_SEG * (factor - 1.0)) - SEARCH_REGION/2; + srch_end = srch_beg + SEARCH_REGION; + + if(srch_beg < p_buf_beg) + srch_beg = p_buf_beg; + + if(srch_end + DELTA >= tmpl) + srch_end = tmpl - DELTA; + + } + else { + + // compression + srch_end = tmpl + (int)((float)TEMPLATE_SEG * (1.0 - factor)) + SEARCH_REGION/2; + srch_beg = srch_end - SEARCH_REGION; + + if(srch_end + TEMPLATE_SEG > p_buf_end) + srch_end = p_buf_end - TEMPLATE_SEG; + + if(srch_beg - DELTA < tmpl) + srch_beg = tmpl + DELTA; + } + + if (srch_beg >= srch_end) + break; + + srch = find_best_corr(tmpl,srch_beg,srch_end); + memcpy(merge_buf,tmpl,TEMPLATE_SEG<<1); + + float f,v; + for(int k=0; k 32767.) + v = 32767.; + else if(v < -32768.) + v = -32768.; + + merge_buf[k] = (short)v; + } + + buffer_put( cur_ts, merge_buf, TEMPLATE_SEG); + buffer_put( cur_ts + TEMPLATE_SEG, srch + TEMPLATE_SEG, + p_buf_end - srch - TEMPLATE_SEG ); + + s += tmpl - srch; + s_all += tmpl - srch; + + cur_ts += TEMPLATE_SEG/2; + tmpl += TEMPLATE_SEG/2; + + if(p_buf_end - tmpl < TEMPLATE_SEG + DELTA) + break; + + float act_fact = s / (float)PACKET_SAMPLES; + + //DBG("new size = %u, ratio = %f, factor = %f\n",s, act_fact, factor); + if((factor > 1.0) && (act_fact >= factor)) + break; + + else if((factor < 1.0) && (act_fact <= factor)) + break; + + else if(act_fact >= TSM_MAX_SCALE || f <= TSM_MIN_SCALE) + break; + + } + + return s; +} diff --git a/core/AmPlayoutBuffer.h b/core/AmPlayoutBuffer.h new file mode 100644 index 00000000..f38b0139 --- /dev/null +++ b/core/AmPlayoutBuffer.h @@ -0,0 +1,58 @@ +#ifndef _AmPlayoutBuffer_h_ +#define _AmPlayoutBuffer_h_ + +#include "SampleArray.h" +#include "AmStats.h" +#include "LowcFE.h" +#include +using std::multiset; + + +#define ORDER_STAT_WIN_SIZE 35 +#define ORDER_STAT_LOSS_RATE 0.1 + +#define EXP_THRESHOLD 20 +#define SHR_THRESHOLD 180 + +#define WSOLA_START_OFF 80 +#define WSOLA_SCALED_WIN 50 + +class AmPlayoutBuffer +{ + // Order statistics delay estimation + multiset o_stat; + int32_t n_stat[ORDER_STAT_WIN_SIZE]; + int idx; + double loss_rate; + + // adaptive WSOLA + u_int32_t wsola_off; + int shr_threshold; + MeanArray short_scaled; + + // second stage PLC + int plc_cnt; + LowcFE fec; + + // Playout buffer + SampleArrayShort buffer; + u_int32_t r_ts,w_ts; + + void buffer_put(unsigned int ts, ShortSample* buf, unsigned int len); + void buffer_get(unsigned int ts, ShortSample* buf, unsigned int len); + + u_int32_t time_scale(u_int32_t ts, float factor); + u_int32_t next_delay(u_int32_t ref_ts, u_int32_t ts); + +public: + AmPlayoutBuffer(); + + void direct_write(unsigned int ts, ShortSample* buf, unsigned int len); + + void write(u_int32_t ref_ts, u_int32_t ts, int16_t* buf, u_int32_t len); + + u_int32_t read(u_int32_t ts, int16_t* buf, u_int32_t len); +}; + + +#endif diff --git a/core/AmRtpAudio.cpp b/core/AmRtpAudio.cpp index 60ada9fe..2a296754 100644 --- a/core/AmRtpAudio.cpp +++ b/core/AmRtpAudio.cpp @@ -67,9 +67,9 @@ int AmRtpAudio::receive(unsigned int audio_buffer_ts) int l_size = conceal_loss(ts - last_ts); if(l_size>0){ - timed_buffer.put(last_ts, - (ShortSample*)samples.back_buffer(), - PCM16_B2S(l_size)); + playout_buffer.direct_write(last_ts, + (ShortSample*)samples.back_buffer(), + PCM16_B2S(l_size)); } } @@ -82,9 +82,9 @@ int AmRtpAudio::receive(unsigned int audio_buffer_ts) if(use_default_plc) add_to_history(size); - timed_buffer.put(ts, - (ShortSample*)((unsigned char*)samples), - PCM16_B2S(size)); + playout_buffer.write(audio_buffer_ts, ts, + (ShortSample*)((unsigned char*)samples), + PCM16_B2S(size)); last_ts = ts + PCM16_B2S(size); } @@ -101,7 +101,7 @@ int AmRtpAudio::get(unsigned int user_ts, unsigned char* buffer, unsigned int nb int AmRtpAudio::read(unsigned int user_ts, unsigned int size) { - timed_buffer.get(user_ts,(ShortSample*)((unsigned char*)samples),PCM16_B2S(size)); + playout_buffer.read(user_ts,(ShortSample*)((unsigned char*)samples),PCM16_B2S(size)); return size; } diff --git a/core/AmRtpAudio.h b/core/AmRtpAudio.h index 1a50b4f3..1903ce7b 100644 --- a/core/AmRtpAudio.h +++ b/core/AmRtpAudio.h @@ -30,7 +30,8 @@ #include "AmAudio.h" #include "AmRtpStream.h" -#include "SampleArray.h" +#include "AmPlayoutBuffer.h" +//#include "SampleArray.h" #include "LowcFE.h" // Maximum value: AUDIO_BUFFER_SIZE / 2 @@ -39,7 +40,8 @@ class AmRtpAudio: public AmRtpStream, public AmAudio { - SampleArrayShort timed_buffer; + //SampleArrayShort timed_buffer; + AmPlayoutBuffer playout_buffer; LowcFE fec; bool use_default_plc; diff --git a/core/AmRtpReceiver.cpp b/core/AmRtpReceiver.cpp index 8441b4bd..943f597e 100644 --- a/core/AmRtpReceiver.cpp +++ b/core/AmRtpReceiver.cpp @@ -95,7 +95,7 @@ void AmRtpReceiver::run() DBG("error while parsing RTP packet.\n"); continue; } - + gettimeofday(&p.recv_time,NULL); streams_mut.lock(); @@ -138,8 +138,9 @@ void AmRtpReceiver::removeStream(int sd) if(fds[i].fd == sd){ - if(--nfds && (i < nfds)) - fds[i] = fds[nfds-1]; + if(--nfds && (i < nfds)){ + fds[i] = fds[nfds]; + } break; } diff --git a/core/AmRtpStream.cpp b/core/AmRtpStream.cpp index b6b1514a..c096e76f 100644 --- a/core/AmRtpStream.cpp +++ b/core/AmRtpStream.cpp @@ -53,6 +53,7 @@ #include using std::set; +#define MAX_DELAY 8000 /* 1 second */ /* * This function must be called before setLocalPort, because @@ -254,19 +255,19 @@ int AmRtpStream::receive( unsigned char* buffer, unsigned int size, recv_offset_i = true; DBG("initialized recv_offset with %i (%i - %i)\n", recv_offset,audio_buffer_ts,rp.timestamp); - ts = audio_buffer_ts + jitter_delay; + ts = audio_buffer_ts;// + jitter_delay; } else { - ts = rp.timestamp - recv_offset + jitter_delay; + ts = rp.timestamp - recv_offset;// + jitter_delay; // resync - if( ts_less()(ts, audio_buffer_ts) || - !ts_less()(ts, audio_buffer_ts + max_delay) ){ + if( ts_less()(ts, audio_buffer_ts - MAX_DELAY/2) || + !ts_less()(ts, audio_buffer_ts + MAX_DELAY) ){ DBG("resync needed: reference ts = %u; write ts = %u\n", audio_buffer_ts,ts); recv_offset = rp.timestamp - audio_buffer_ts; - ts = audio_buffer_ts + jitter_delay; + ts = audio_buffer_ts;// + jitter_delay; } } @@ -309,8 +310,8 @@ AmRtpStream::AmRtpStream(AmSession* _s) l_saddr.sin_addr.s_addr = INADDR_ANY; #endif - jitter_delay = 80/*ms*/ * 8; - max_delay = 120/*ms*/ * 8; + //jitter_delay = 80/*ms*/ * 8; + //max_delay = 120/*ms*/ * 8; } AmRtpStream::~AmRtpStream() diff --git a/core/AmStats.h b/core/AmStats.h new file mode 100644 index 00000000..a5bd6457 --- /dev/null +++ b/core/AmStats.h @@ -0,0 +1,92 @@ +#ifndef _AmStats_h_ +#define _AmStats_h_ + +#include +#include + +class MeanValue +{ +protected: + double cum_val; + size_t n_val; + +public: + MeanValue() + : cum_val(0.0), + n_val(0) + {} + + void push(double val){ + cum_val += val; + n_val++; + } + + double mean(){ + if(!n_val) return 0.0; + return cum_val / float(n_val); + } +}; + +class StddevValue +{ + protected: + double cum_val; + double sq_cum_val; + size_t n_val; + + public: + StddevValue() + : cum_val(0.0), + sq_cum_val(0.0), + n_val(0) + {} + + void push(double val){ + + cum_val += val; + sq_cum_val += val*val; + n_val++; + } + + double stddev(){ + if(!n_val) return 0.0; + return sqrt((n_val*sq_cum_val - cum_val*cum_val)/(n_val*(n_val-1))); + } +}; + +class MeanArray: public MeanValue +{ + double *buffer; + size_t buf_size; + + double cum_val; + size_t n_val; + + public: + MeanArray(size_t size) + : buf_size(size), + MeanValue() + { + buffer = new double[size]; + } + + ~MeanArray(){ + delete [] buffer; + } + + void push(double val){ + + cum_val -= buffer[n_val % buf_size]; + buffer[n_val % buf_size] = val; + + cum_val += val; + n_val++; + } + + double mean(){ + if(!n_val) return 0.0; + return cum_val / double(n_val > buf_size ? buf_size : n_val); + } +}; + +#endif