From 7187b63eb6f8f58875b2d8cfeb814539a43ce135 Mon Sep 17 00:00:00 2001 From: Mike Brady Date: Wed, 30 Apr 2014 20:54:52 +0100 Subject: [PATCH] too numerous to mention: watch timing of everything, add pthread_cond_wait signalling, monitor latency of DAC, try to guess initial delay for first packet, trim or stuff packets to keep latency within bounds, much much more... --- player.c | 470 +++++++++++++++++++++++++++++++++++++++++++------------ 1 file changed, 368 insertions(+), 102 deletions(-) diff --git a/player.c b/player.c index 639354ce..8f221d79 100644 --- a/player.c +++ b/player.c @@ -37,6 +37,7 @@ #include #include #include +#include #include "common.h" #include "player.h" @@ -68,6 +69,9 @@ static SRC_STATE *src; #endif +// debug variables +static int late_packet_message_sent; + // interthread variables static double volume = 1.0; static int fix_volume = 0x10000; @@ -80,6 +84,7 @@ static pthread_mutex_t vol_mutex = PTHREAD_MUTEX_INITIALIZER; typedef struct audio_buffer_entry { // decoded audio packets int ready; + uint32_t timestamp; signed short *data; } abuf_t; static abuf_t audio_buffer[BUFFER_FRAMES]; @@ -88,7 +93,14 @@ static abuf_t audio_buffer[BUFFER_FRAMES]; // mutex-protected variables static seq_t ab_read, ab_write; static int ab_buffering = 1, ab_synced = 0; +static uint32_t first_packet_timestamp = 0; + static pthread_mutex_t ab_mutex = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t flowcontrol; + +static int64_t first_packet_time_to_play; // nanoseconds + + static void bf_est_reset(short fill); @@ -98,6 +110,10 @@ static void ab_resync(void) { audio_buffer[i].ready = 0; ab_synced = 0; ab_buffering = 1; + first_packet_timestamp = 0; + first_packet_time_to_play = 0; + clear_reference_timestamp(); + } // the sequence numbers will wrap pretty often. @@ -188,7 +204,7 @@ static void free_buffer(void) { free(audio_buffer[i].data); } -void player_put_packet(seq_t seqno, uint8_t *data, int len) { +void player_put_packet(seq_t seqno,uint32_t timestamp, uint8_t *data, int len) { abuf_t *abuf = 0; int16_t buf_fill; @@ -208,8 +224,11 @@ void player_put_packet(seq_t seqno, uint8_t *data, int len) { ab_write = seqno; } else if (seq_order(ab_read, seqno)) { // late but not yet played abuf = audio_buffer + BUFIDX(seqno); - } else { // too late. - warn("late packet %04X (%04X:%04X)", seqno, ab_read, ab_write); + } else { // too late. + if (!late_packet_message_sent) { + debug(1, "late packet %04X (%04X:%04X).\n", seqno, ab_read, ab_write); + late_packet_message_sent=1; + } } buf_fill = seq_diff(ab_read, ab_write); pthread_mutex_unlock(&ab_mutex); @@ -217,14 +236,16 @@ void player_put_packet(seq_t seqno, uint8_t *data, int len) { if (abuf) { alac_decode(abuf->data, data, len); abuf->ready = 1; + abuf->timestamp = timestamp; } + pthread_mutex_lock(&ab_mutex); - if (ab_buffering && buf_fill >= config.buffer_start_fill) { - debug(1, "buffering over. starting play\n"); - ab_buffering = 0; - bf_est_reset(buf_fill); - } + + int rc = pthread_cond_signal(&flowcontrol); + if (rc) + debug(1,"Error signalling flowcontrol.\n"); + pthread_mutex_unlock(&ab_mutex); } @@ -236,17 +257,17 @@ static short lcg_rand(void) { } static inline short dithered_vol(short sample) { - static short rand_a, rand_b; - long out; + static short rand_a, rand_b; + long out; - out = (long)sample * fix_volume; - if (fix_volume < 0x10000) { - rand_b = rand_a; - rand_a = lcg_rand(); - out += rand_a; - out -= rand_b; - } - return out>>16; + out = (long)sample * fix_volume; + if (fix_volume < 0x10000) { + rand_b = rand_a; + rand_a = lcg_rand(); + out += rand_a; + out -= rand_b; + } + return out>>16; } typedef struct { @@ -339,69 +360,146 @@ static void bf_est_update(short fill) { } // get the next frame, when available. return 0 if underrun/stream reset. -static short *buffer_get_frame(void) { - int16_t buf_fill; - seq_t read, next; - abuf_t *abuf = 0; - int i; +static abuf_t *buffer_get_frame(void) { + int16_t buf_fill; + uint64_t local_time_now; + seq_t read, next; + abuf_t *abuf = 0; + int i; + abuf_t *curframe; + pthread_mutex_lock(&ab_mutex); + int wait; + uint32_t dac_delay = 0; + do { + curframe = audio_buffer + BUFIDX(ab_read); + if (config.output->delay) + dac_delay = config.output->delay(); + + if (curframe->ready) { + if (ab_buffering) { // if we are getting packets but not yet forwarding them to the player + if (first_packet_timestamp==0) { // if this is the very first packet + uint32_t reference_timestamp; + uint64_t reference_timestamp_time; + get_reference_timestamp_stuff(&reference_timestamp,&reference_timestamp_time); + if (reference_timestamp) { // if we have a reference time - if (ab_buffering) - return 0; + first_packet_timestamp=curframe->timestamp; // we will keep buffering until we are supposed to start playing this + + // here, see if we should start playing. We need to know when to allow the packets to be sent to the player + // we will get a fix every second or so, which will be stored as a pair consisting of + // the time when the packet with a particular timestamp should be played. + // it might not be the timestamp of our first packet, however, so we might have to do some calculations. + + struct timespec tn; + clock_gettime(CLOCK_MONOTONIC,&tn); + local_time_now=((uint64_t)tn.tv_sec<<32)+((uint64_t)tn.tv_nsec<<32)/1000000000; + + int64_t delta = ((int64_t)first_packet_timestamp-(int64_t)reference_timestamp); - pthread_mutex_lock(&ab_mutex); + first_packet_time_to_play = reference_timestamp_time+((delta+(int64_t)config.latency)<<32)/44100; // using the latency requested... - buf_fill = seq_diff(ab_read, ab_write); - if (buf_fill < 1 || !ab_synced) { - if (buf_fill < 1) - warn("underrun."); - ab_buffering = 1; - pthread_mutex_unlock(&ab_mutex); - return 0; - } - if (buf_fill >= BUFFER_FRAMES) { // overrunning! uh-oh. restart at a sane distance - warn("overrun."); - ab_read = ab_write - config.buffer_start_fill; - } - read = ab_read; - ab_read++; - buf_fill = seq_diff(ab_read, ab_write); - bf_est_update(buf_fill); + } + } - // check if t+16, t+32, t+64, t+128, ... (buffer_start_fill / 2) - // packets have arrived... last-chance resend - if (!ab_buffering) { - for (i = 16; i < (config.buffer_start_fill / 2); i = (i * 2)) { - next = ab_read + i; - abuf = audio_buffer + BUFIDX(next); - if (!abuf->ready) { - rtp_request_resend(next, next); - } + if (first_packet_time_to_play!=0) { + uint32_t filler_size = frame_size; + + if (config.output->delay) { + dac_delay = config.output->delay(); + if (dac_delay==0) + filler_size = 2048*15; + } + struct timespec time_now; + clock_gettime(CLOCK_MONOTONIC,&time_now); + uint64_t tn = ((uint64_t)time_now.tv_sec<<32)+((uint64_t)time_now.tv_nsec<<32)/1000000000; + uint64_t exact_frame_gap = (((first_packet_time_to_play-tn)*44100)>>32)-dac_delay; + if (dac_delay==0) + debug(2,"exact_frame_gap is %llu.\n",exact_frame_gap); + uint32_t fs=filler_size; + if (exact_frame_gap<=filler_size) { + fs=exact_frame_gap; + ab_buffering = 0; + } + signed short *silence; + silence = malloc(FRAME_BYTES(fs)); + memset(silence, 0, FRAME_BYTES(fs)); + config.output->play(silence, fs); + free(silence); } + } } + wait = (ab_buffering || (dac_delay>=4410) || (!ab_synced)) && (!please_stop); +// wait = (ab_buffering || (seq_diff(ab_read, ab_write) < (config.latency-22000)/(352)) || (!ab_synced)) && (!please_stop); + if (wait) { + struct timespec to; + clock_gettime(CLOCK_MONOTONIC, &to); + // Watch out: should be 1^9/44100, but this causes integer overflow + uint64_t calc = (uint64_t)(4*252*1000000)/(44*3); //four thirds of a packet time + uint32_t calcs = calc&0xFFFFFFFF; + to.tv_nsec+=calcs; + if (to.tv_nsec>1000000000) { + to.tv_nsec-=1000000000; + to.tv_sec+=1; + } + pthread_cond_timedwait(&flowcontrol,&ab_mutex,&to); + } + } while (wait); - abuf_t *curframe = audio_buffer + BUFIDX(read); - if (!curframe->ready) { - warn("missing frame %04X.", read); - memset(curframe->data, 0, FRAME_BYTES(frame_size)); + if (!please_stop) { +// buf_fill = seq_diff(ab_read, ab_write); +// if (buf_fill < 1 || !ab_synced) { +// if (buf_fill < 1) + if (dac_delay==0) {// we just got underrun + debug(1,"Underrun!\n"); + ab_resync(); // starting over } - curframe->ready = 0; + } + + if (please_stop) { pthread_mutex_unlock(&ab_mutex); + debug(1,"Exiting from buffer_get_frame.\n"); + return 0; + } - return curframe->data; +/* + if (buf_fill >= BUFFER_FRAMES) { // overrunning! uh-oh. restart at a sane distance + warn("overrun."); + ab_read = ab_write - config.buffer_start_fill; + } +*/ + read = ab_read; + ab_read++; + + + // check if t+16, t+32, t+64, t+128, ... (buffer_start_fill / 2) + // packets have arrived... last-chance resend + + + if (!ab_buffering) { + for (i = 16; i < (seq_diff(ab_read,ab_write) / 2); i = (i * 2)) { + next = ab_read + i; + abuf = audio_buffer + BUFIDX(next); + if (!abuf->ready) { + rtp_request_resend(next, next); + } + } + } + + + if (!curframe->ready) { + // debug(1, "missing frame %04X. Supplying a silent frame.\n", read); + memset(curframe->data, 0, FRAME_BYTES(frame_size)); + } + curframe->ready = 0; + pthread_mutex_unlock(&ab_mutex); + return curframe; } -static int stuff_buffer(double playback_rate, short *inptr, short *outptr) { +static int stuff_buffer(short *inptr, short *outptr, int32_t stuff) { int i; int stuffsamp = frame_size; - int stuff = 0; - double p_stuff; - - p_stuff = 1.0 - pow(1.0 - fabs(playback_rate-1.0), frame_size); - - if (rand() < p_stuff * RAND_MAX) { - stuff = playback_rate > 1.0 ? -1 : 1; - stuffsamp = rand() % (frame_size - 1); - } + if (stuff) + stuffsamp = rand() % (frame_size - 1); pthread_mutex_lock(&vol_mutex); for (i=0; i> 1); *outptr++ = dithered_vol(((long)inptr[-1] + (long)inptr[1]) >> 1); } else if (stuff==-1) { - debug(2, "---------\n"); + debug(3, "---------\n"); inptr++; inptr++; } @@ -430,8 +528,35 @@ static int stuff_buffer(double playback_rate, short *inptr, short *outptr) { } static void *player_thread_func(void *arg) { +#define averaging_interval 71 +#define trend_interval 44100 + int32_t corrections[trend_interval]; + int32_t sum_of_corrections; + double moving_average_correction; + int oldest_correction, newest_correction, number_of_corrections; + + int64_t delays[averaging_interval]; + int64_t sum_of_delays,moving_average_delay; + int oldest_delay,newest_delay,number_of_delays; + int64_t additions,deletions = 0; + additions = deletions = 0; + int64_t frames = 0; int play_samples; + int64_t initial_latency = 0; + int64_t previous_latency,current_latency,change_in_latency; + int64_t current_delay; + int play_number = 0; + uint64_t accumulated_buffers_in_use = 0; + int64_t accumulated_delay = 0; + int64_t accumulated_da_delay = 0; + const int print_interval = 71; + // I think it's useful to keep this prime to prevent it from falling into a pattern with some other process. + + struct drand48_data randBuffer; + srand48_r(time(NULL), &randBuffer); + + signed short *inbuf, *outbuf, *silence; outbuf = malloc(OUTFRAME_BYTES(frame_size)); silence = malloc(OUTFRAME_BYTES(frame_size)); @@ -452,56 +577,184 @@ static void *player_thread_func(void *arg) { srcdat.end_of_input = 0; } #endif - - while (!please_stop) { - inbuf = buffer_get_frame(); - if (!inbuf) - inbuf = silence; - -#ifdef FANCY_RESAMPLING + change_in_latency = 0; + late_packet_message_sent=0; + oldest_delay=newest_delay=number_of_delays=0; + sum_of_delays=0; + oldest_correction=newest_correction=number_of_corrections=0; + sum_of_corrections=0; + while (!please_stop) { + abuf_t *inframe = buffer_get_frame(); + if (inframe) { + inbuf = inframe->data; + if (inbuf) { + #ifdef FANCY_RESAMPLING if (fancy_resampling) { - int i; - pthread_mutex_lock(&vol_mutex); - for (i=0; i<2*FRAME_BYTES(frame_size); i++) { - frame[i] = (float)inbuf[i] / 32768.0; - frame[i] *= volume; - } - pthread_mutex_unlock(&vol_mutex); - srcdat.src_ratio = bf_playback_rate; - src_process(src, &srcdat); - assert(srcdat.input_frames_used == FRAME_BYTES(frame_size)); - src_float_to_short_array(outframe, outbuf, FRAME_BYTES(frame_size)*2); - play_samples = srcdat.output_frames_gen; - } else -#endif - play_samples = stuff_buffer(bf_playback_rate, inbuf, outbuf); + int i; + pthread_mutex_lock(&vol_mutex); + for (i=0; i<2*FRAME_BYTES(frame_size); i++) { + frame[i] = (float)inbuf[i] / 32768.0; + frame[i] *= volume; + } + pthread_mutex_unlock(&vol_mutex); + srcdat.src_ratio = bf_playback_rate; + src_process(src, &srcdat); + assert(srcdat.input_frames_used == FRAME_BYTES(frame_size)); + src_float_to_short_array(outframe, outbuf, FRAME_BYTES(frame_size)*2); + play_samples = srcdat.output_frames_gen; + } else + #endif + + // we need to see whether the player is a bit slow or a bit fast, so we track the overall latency + struct timespec tn; + clock_gettime(CLOCK_MONOTONIC,&tn); + uint64_t local_time_now=((uint64_t)tn.tv_sec<<32)+((uint64_t)tn.tv_nsec<<32)/1000000000; + double x; + drand48_r(&randBuffer, &x); + play_number++; + int32_t abs_change_in_latency = change_in_latency; + + if (abs_change_in_latency<1) + abs_change_in_latency=-abs_change_in_latency; + + int32_t amount_to_stuff=0; + + if (x<=(double)1.0*abs_change_in_latency/print_interval) { + amount_to_stuff= -abs_change_in_latency/change_in_latency; + } + + if (number_of_corrections==trend_interval) { // the array of corrections is full + sum_of_corrections-=corrections[oldest_correction]; + oldest_correction=(oldest_correction+1)%trend_interval; + number_of_corrections--; + } + + corrections[newest_correction]=change_in_latency; + sum_of_corrections+=amount_to_stuff; + newest_correction=(newest_correction+1)%trend_interval; + number_of_corrections++; + + moving_average_correction = (1.0*sum_of_corrections)/number_of_corrections; + + + play_samples = stuff_buffer(inbuf, outbuf,amount_to_stuff); + if (amount_to_stuff > 0) + additions += amount_to_stuff; + else + deletions -= amount_to_stuff; + frames += play_samples; + + if (config.output->delay) + current_delay = config.output->delay(); config.output->play(outbuf, play_samples); - } + // now see if we can calculate how early or late the first frame would be + uint32_t reference_timestamp; + uint64_t reference_timestamp_time; - return 0; + get_reference_timestamp_stuff(&reference_timestamp,&reference_timestamp_time); + + int64_t rt,nt; + rt = reference_timestamp; + nt = inframe->timestamp; + + int64_t td_in_frames; + int64_t td = local_time_now-reference_timestamp_time; + //debug(1,"td is %lld.\n",td); + if (td>=0) + td_in_frames = (td*44100)>>32; + else { + td_in_frames = -((-td*44100)>>32); + } + // this is the actual delay, which will fluctuate a good bit about a potentially rising or falling trend. + int64_t delay = td_in_frames+rt-(nt-current_delay); + + if (number_of_delays==averaging_interval) { // the array of delays is full + sum_of_delays-=delays[oldest_delay]; + oldest_delay=(oldest_delay+1)%averaging_interval; + number_of_delays--; + } + + delays[newest_delay]=delay; + sum_of_delays+=delay; + newest_delay=(newest_delay+1)%averaging_interval; + number_of_delays++; + + moving_average_delay = sum_of_delays/number_of_delays; // this doesn't seem to be right + + change_in_latency=moving_average_delay-config.latency; + + accumulated_da_delay+=current_delay; + accumulated_delay+=delay; + + accumulated_buffers_in_use += seq_diff(ab_read, ab_write); + if (play_number%print_interval==0) { + current_latency=accumulated_delay/print_interval; + if ((play_number/print_interval)%50==0) { // only print every fiftieth one, in verbose mode + debug(1,"Valid frames: %lld; overall frames added/subtracted %lld; frames added + frames deleted %lld; average D/A delay, average latency (frames): %llu, %llu; average buffers in use: %llu, moving average delay (number of delays): %llu (%lu).\n", + frames-(additions-deletions), additions-deletions, additions+deletions, accumulated_da_delay/print_interval,current_latency,accumulated_buffers_in_use/print_interval,moving_average_delay,number_of_delays); + // this is misleading! debug(1,"Moving average correction (ppm) (number of frames): %.4f (%d).\n", moving_average_correction*1000000/252,number_of_corrections); + } + if (initial_latency==0) { + initial_latency=moving_average_delay; + debug(1,"Initial latency error: %d frames.\n",config.latency-initial_latency); + } + if (previous_latency==0) + previous_latency=current_latency; + else { + previous_latency=current_latency; + } + accumulated_delay=0; + accumulated_da_delay=0; + accumulated_buffers_in_use=0; + } + } + } + } + return 0; } + // takes the volume as specified by the airplay protocol void player_volume(double f) { - double linear_volume = pow(10.0, 0.05*f); +// the volume ranges -144.0 (mute) or -30 -- 0. See http://git.zx2c4.com/Airtunes2/about/#setting-volume +// see http://sound.westhost.com/project01.htm +// see also http://tangentsoft.net/audio/atten.html for data on good attenuators. Seems as if 30 dB power (60 dB amplitude) is about right +// by examination, the -30 -- 0 range is linear on the slider; i.e. the slider is calibrated in 30 equal increments +// so, we will pass this on without any weighting if we have a hardware mixer, as we expect the mixer to be calibrated in dB. + +// I'm guessing the dB are often expressing amplitude ratios, as -57 dB on the iMic mixer sounds quite like a ratio of about 0.001 on the soft volume control + + double linear_volume = pow(10.0, 0.10*f); // gives an exponentially weighted volume control from about 0.001 to 1 + // 0.001 is just audible on an iMic with headphones -- to 1 + // the level sorta corresponds to the same setting via the iMic's hardware mixer + + if(f == -144.0) + linear_volume = 0.0; + + debug(2,"iTunes volume: %f, linear volume: %f.\n",f,linear_volume); + if (config.output->volume) { - config.output->volume(linear_volume); - } else { - pthread_mutex_lock(&vol_mutex); - volume = linear_volume; - fix_volume = 65536.0 * volume; - pthread_mutex_unlock(&vol_mutex); - } + config.output->volume(f); + linear_volume=1.0; // no attenuation needed + } + pthread_mutex_lock(&vol_mutex); + volume = linear_volume; + fix_volume = 65536.0 * volume; + pthread_mutex_unlock(&vol_mutex); } + void player_flush(void) { + if (config.output->flush) + config.output->flush(); pthread_mutex_lock(&ab_mutex); ab_resync(); pthread_mutex_unlock(&ab_mutex); } int player_play(stream_cfg *stream) { + debug(1,"player_play called...\n"); if (config.buffer_start_fill > BUFFER_FRAMES) die("specified buffer starting fill %d > buffer size %d", config.buffer_start_fill, BUFFER_FRAMES); @@ -514,13 +767,20 @@ int player_play(stream_cfg *stream) { #ifdef FANCY_RESAMPLING init_src(); #endif - please_stop = 0; if (config.cmd_start && !fork()) { if (system(config.cmd_start)) warn("exec of external start command failed"); exit(0); } + + // set the flowcontrol condition variable to wait on a monotonic clock + pthread_condattr_t attr; + pthread_condattr_init(&attr); + pthread_condattr_setclock( &attr, CLOCK_MONOTONIC); + int rc = pthread_cond_init(&flowcontrol,&attr); + if (rc) + debug(1,"Error initialising condition variable.\n"); config.output->start(sampling_rate); pthread_create(&player_thread, NULL, player_thread_func, NULL); @@ -528,7 +788,9 @@ int player_play(stream_cfg *stream) { } void player_stop(void) { + debug(1,"player_stop called...\n"); please_stop = 1; + pthread_cond_signal(&flowcontrol); // tell it to give up pthread_join(player_thread, NULL); config.output->stop(); if (config.cmd_stop && !fork()) { @@ -541,4 +803,8 @@ void player_stop(void) { #ifdef FANCY_RESAMPLING free_src(); #endif + int rc = pthread_cond_destroy(&flowcontrol); + if (rc) + debug(1,"Error destroying condition variable.\n"); + debug(1,"player_stop finished...\n"); }