Mirror/squeezelite-esp32
1
0
Fork 0
mirror of https://github.com/sle118/squeezelite-esp32.git synced 2025-12-07 12:07:09 +03:00
Code Issues Packages Projects Releases Wiki Activity

AirPlay progress with synchro

Browse Source
This commit is contained in:
philippe44
2019-08-18 17:38:48 -07:00
parent 0f6e240cc6
commit 876720f9e2
8 changed files with 108 additions and 216 deletions
Download Patch File Download Diff File Expand all files Collapse all files

236
components/raop/rtp.c
View File

@@ -41,6 +41,7 @@
#include "platform.h"
#include "rtp.h"
#include "raop_sink.h"
#include "log_util.h"
#include "util.h"
@@ -63,16 +64,13 @@
#define MS2TS(ms, rate) ((((u64_t) (ms)) * (rate)) / 1000)
#define TS2MS(ts, rate) NTP2MS(TS2NTP(ts,rate))
#define GAP_THRES 8
#define GAP_COUNT 20
extern log_level raop_loglevel;
static log_level *loglevel = &raop_loglevel;
//#define __RTP_STORE
// default buffer size
#define BUFFER_FRAMES ( (120 * 88200) / (352 * 100) )
#define MAX_PACKET 1408
@@ -103,7 +101,8 @@ typedef struct rtp_s {
#ifdef WIN32
AES_KEY aes;
#else
mbedtls_aes_context aes;
mbedtls_aes_context aes;
#endif
bool decrypt;
u8_t *decrypt_buf;
int frame_size, frame_duration;
@@ -113,15 +112,11 @@ typedef struct rtp_s {
struct {
unsigned short rport, lport;
int sock;
} rtp_sockets[3]; // data, control, timing
} rtp_sockets[3]; // data, control, timing
bool drift;
u64_t local, remote;
struct timing_s {
u64_t local, remote;
} timing;
struct {
u32_t rtp, time;
u32_t rtp, time;
u8_t status;
} synchro;
@@ -129,9 +124,10 @@ typedef struct rtp_s {
u32_t time;
seq_t seqno;
u32_t rtptime;
} record;
} record;
int latency; // rtp hold depth in samples
u32_t resent_req, resent_rec; // total resent + recovered frames
u32_t silent_frames; // total silence frames
u32_t filled_frames; // silence frames in current silence episode
u32_t discarded;
int skip; // number of frames to skip to keep sync alignement
@@ -144,7 +140,9 @@ typedef struct rtp_s {
TaskHandle_t rtp_thread, joiner;
#endif
alac_file *alac_codec;
int flush_seqno;
int flush_seqno;
bool playing;
raop_data_cb_t data_cb;
raop_cmd_cb_t cmd_cb;
u16_t syncS, syncN;
} rtp_t;
@@ -192,10 +190,9 @@ static alac_file* alac_init(int fmtp[32]) {
return alac;
}
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
int latency, char *aeskey, char *aesiv, char *fmtpstr,
rtp_resp_t rtp_init(struct in_addr host, int latency, char *aeskey, char *aesiv, char *fmtpstr,
short unsigned pCtrlPort, short unsigned pTimingPort,
raop_cmd_cb_t cmd_cb, raop_data_cb_t data_cb)
{
@@ -208,15 +205,13 @@ rtp_resp_t rtp_init(struct in_addr host, bool sync, bool drift, bool range,
if (!ctx) return resp;
ctx->host = host;
ctx->host = host;
ctx->decrypt = false;
ctx->cmd_cb = cmd_cb;
ctx->data_cb = data_cb;
ctx->rtp_host.sin_family = AF_INET;
ctx->rtp_host.sin_addr.s_addr = INADDR_ANY;
pthread_mutex_init(&ctx->ab_mutex, 0);
ctx->flush_seqno = -1;
ctx->latency = latency;
ctx->synchro.required = sync;
ctx->flush_seqno = -1;
ctx->latency = latency;
@@ -238,6 +233,7 @@ rtp_resp_t rtp_init(struct in_addr host, bool sync, bool drift, bool range,
#else
memset(&ctx->aes, 0, sizeof(mbedtls_aes_context));
mbedtls_aes_setkey_dec(&ctx->aes, (unsigned char*) aeskey, 128);
#endif
ctx->decrypt = true;
ctx->decrypt_buf = malloc(MAX_PACKET);
}
@@ -301,6 +297,7 @@ void rtp_end(rtp_t *ctx)
for (i = 0; i < 3; i++) shutdown_socket(ctx->rtp_sockets[i].sock);
delete_alac(ctx->alac_codec);
if (ctx->decrypt_buf) free(ctx->decrypt_buf);
buffer_release(ctx->audio_buffer);
@@ -324,7 +321,6 @@ bool rtp_flush(rtp_t *ctx, unsigned short seqno, unsigned int rtptime)
} else {
pthread_mutex_lock(&ctx->ab_mutex);
buffer_reset(ctx->audio_buffer);
ctx->playing = false;
ctx->playing = false;
ctx->flush_seqno = seqno;
pthread_mutex_unlock(&ctx->ab_mutex);
@@ -376,7 +372,6 @@ static int seq_order(seq_t a, seq_t b) {
return d > 0;
}
/*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*/
static void alac_decode(rtp_t *ctx, s16_t *dest, char *buf, int len, int *outsize) {
unsigned char iv[16];
@@ -385,12 +380,12 @@ static void alac_decode(rtp_t *ctx, s16_t *dest, char *buf, int len, int *outsiz
if (ctx->decrypt) {
aeslen = len & ~0xf;
memcpy(iv, ctx->aesiv, sizeof(iv));
memcpy(iv, ctx->aesiv, sizeof(iv));
#ifdef WIN32
AES_cbc_encrypt((unsigned char*)buf, packet, aeslen, &ctx->aes, iv, AES_DECRYPT);
AES_cbc_encrypt((unsigned char*)buf, ctx->decrypt_buf, aeslen, &ctx->aes, iv, AES_DECRYPT);
#else
mbedtls_aes_crypt_cbc(&ctx->aes, MBEDTLS_AES_DECRYPT, aeslen, iv, (unsigned char*) buf, packet);
#endif
mbedtls_aes_crypt_cbc(&ctx->aes, MBEDTLS_AES_DECRYPT, aeslen, iv, (unsigned char*) buf, ctx->decrypt_buf);
#endif
memcpy(ctx->decrypt_buf+aeslen, buf+aeslen, len-aeslen);
decode_frame(ctx->alac_codec, ctx->decrypt_buf, dest, outsize);
} else decode_frame(ctx->alac_codec, (unsigned char*) buf, dest, outsize);
@@ -398,19 +393,21 @@ static void alac_decode(rtp_t *ctx, s16_t *dest, char *buf, int len, int *outsiz
/*---------------------------------------------------------------------------*/
static void buffer_put_packet(rtp_t *ctx, seq_t seqno, unsigned rtptime, bool first, char *data, int len) {
abuf_t *abuf = NULL;
u32_t playtime;
pthread_mutex_lock(&ctx->ab_mutex);
if (!ctx->playing) {
if (!ctx->playing) {
if ((ctx->flush_seqno == -1 || seq_order(ctx->flush_seqno, seqno)) &&
(ctx->synchro.status & RTP_SYNC && ctx->synchro.status & NTP_SYNC)) {
ctx->ab_write = seqno-1;
ctx->ab_read = seqno;
ctx->skip = 0;
ctx->flush_seqno = -1;
ctx->playing = true;
ctx->flush_seqno = -1;
ctx->playing = true;
ctx->resent_req = ctx->resent_rec = ctx->silent_frames = ctx->discarded = 0;
playtime = ctx->synchro.time + (((s32_t)(rtptime - ctx->synchro.rtp)) * 10) / 441;
ctx->cmd_cb(RAOP_PLAY, &playtime);
} else {
@@ -422,6 +419,7 @@ static void buffer_put_packet(rtp_t *ctx, seq_t seqno, unsigned rtptime, bool fi
if (seqno == ctx->ab_write+1) {
// expected packet
abuf = ctx->audio_buffer + BUFIDX(seqno);
ctx->ab_write = seqno;
LOG_SDEBUG("packet expected seqno:%hu rtptime:%u (W:%hu R:%hu)", seqno, rtptime, ctx->ab_write, ctx->ab_read);
} else if (seq_order(ctx->ab_write, seqno)) {
@@ -448,6 +446,7 @@ static void buffer_put_packet(rtp_t *ctx, seq_t seqno, unsigned rtptime, bool fi
abuf = ctx->audio_buffer + BUFIDX(seqno);
ctx->ab_write = seqno;
} else if (seq_order(ctx->ab_read, seqno + 1)) {
// recovered packet, not yet sent
abuf = ctx->audio_buffer + BUFIDX(seqno);
ctx->resent_rec++;
LOG_DEBUG("[%p]: packet recovered seqno:%hu rtptime:%u (W:%hu R:%hu)", ctx, seqno, rtptime, ctx->ab_write, ctx->ab_read);
@@ -455,7 +454,7 @@ static void buffer_put_packet(rtp_t *ctx, seq_t seqno, unsigned rtptime, bool fi
// too late
LOG_DEBUG("[%p]: packet too late seqno:%hu rtptime:%u (W:%hu R:%hu)", ctx, seqno, rtptime, ctx->ab_write, ctx->ab_read);
}
if (ctx->in_frames++ > 1000) {
LOG_INFO("[%p]: fill [level:%hd rec:%u] [W:%hu R:%hu]", ctx, (seq_t) (ctx->ab_write - ctx->ab_read + 1), ctx->resent_rec, ctx->ab_write, ctx->ab_read);
ctx->in_frames = 0;
@@ -479,7 +478,7 @@ static void buffer_put_packet(rtp_t *ctx, seq_t seqno, unsigned rtptime, bool fi
/*---------------------------------------------------------------------------*/
// push as many frames as possible through callback
static void buffer_push_packet(rtp_t *ctx) {
static void buffer_push_packet(rtp_t *ctx) {
abuf_t *curframe = NULL;
u32_t now, playtime, hold = max((ctx->latency * 1000) / (8 * 44100), 100);
int i;
@@ -492,27 +491,17 @@ static void buffer_push_packet(rtp_t *ctx) {
// there is always at least one frame in the buffer
do {
curframe = ctx->audio_buffer + BUFIDX(ctx->ab_read);
playtime = ctx->synchro.time + (((s32_t)(curframe->rtptime - ctx->synchro.rtp)) * 1000) / 44100;
/*
playtime = ctx->synchro.time + (((s32_t)(curframe->rtptime - ctx->synchro.rtp)) * 10) / 441;
if (now > playtime) {
LOG_DEBUG("[%p]: discarded frame now:%u missed by %d (W:%hu R:%hu)", ctx, now, now - playtime, ctx->ab_write, ctx->ab_read);
//LOG_INFO("[%p]: discarded frame (W:%hu R:%hu)", ctx, ctx->ab_write, ctx->ab_read);
ctx->discarded++;
} else if (curframe->ready) {
ctx->callback((const u8_t*) curframe->data, curframe->len);
ctx->data_cb((const u8_t*) curframe->data, curframe->len);
curframe->ready = 0;
} else if (now >= playtime) {
LOG_DEBUG("[%p]: created zero frame (W:%hu R:%hu)", ctx, ctx->ab_write, ctx->ab_read);
ctx->callback(silence_frame, ctx->frame_size * 4);
ctx->silent_frames++;
} else break;
*/
if (curframe->ready) {
ctx->callback((const u8_t*) curframe->data, curframe->len);
curframe->ready = 0;
} else if (now >= playtime) {
} else if (playtime - now <= hold) {
LOG_DEBUG("[%p]: created zero frame (W:%hu R:%hu)", ctx, ctx->ab_write, ctx->ab_read);
ctx->data_cb(silence_frame, ctx->frame_size * 4);
ctx->silent_frames++;
@@ -522,16 +511,16 @@ static void buffer_push_packet(rtp_t *ctx) {
ctx->out_frames++;
} while (ctx->ab_write - ctx->ab_read + 1 > 0);
if (ctx->out_frames > 1000) {
LOG_INFO("[%p]: drain [level:%hd gap:%d] [W:%hu R:%hu] [R:%u S:%u F:%u]",
LOG_INFO("[%p]: drain [level:%hd gap:%d] [W:%hu R:%hu] [R:%u S:%u F:%u D:%u] (head in %u ms) ",
ctx, ctx->ab_write - ctx->ab_read, playtime - now, ctx->ab_write, ctx->ab_read,
ctx->resent_req, ctx->silent_frames, ctx->filled_frames, ctx->discarded, playtime - now);
ctx->out_frames = 0;
}
LOG_SDEBUG("playtime %u %d [W:%hu R:%hu] %d", playtime, playtime - now, ctx->ab_write, ctx->ab_read, curframe->ready);
// each missing packet will be requested up to (latency_frames / 16) times
for (i = 1; seq_order(ctx->ab_read + i, ctx->ab_write); i += 16) {
abuf_t *frame = ctx->audio_buffer + BUFIDX(ctx->ab_read + i);
@@ -626,8 +615,8 @@ static void *rtp_thread_func(void *arg) {
case 0x54: {
u32_t rtp_now_latency = ntohl(*(u32_t*)(pktp+4));
u64_t remote = (((u64_t) ntohl(*(u32_t*)(pktp+8))) << 32) + ntohl(*(u32_t*)(pktp+12));
u32_t rtp_now = ntohl(*(u32_t*)(pktp+16));
u32_t rtp_now = ntohl(*(u32_t*)(pktp+16));
pthread_mutex_lock(&ctx->ab_mutex);
// re-align timestamp and expected local playback time (and magic 11025 latency)
@@ -636,7 +625,6 @@ static void *rtp_thread_func(void *arg) {
ctx->synchro.time = ctx->timing.local + (u32_t) NTP2MS(remote - ctx->timing.remote);
// now we are synced on RTP frames
ctx->synchro.status |= RTP_SYNC;
ctx->synchro.status |= RTP_SYNC;
// 1st sync packet received (signals a restart of playback)
@@ -656,7 +644,6 @@ static void *rtp_thread_func(void *arg) {
break;
}
// NTP timing packet
case 0x53: {
@@ -676,50 +663,17 @@ static void *rtp_thread_func(void *arg) {
elapsed local time between the two request, corrected by the
drifting
*/
expected = ctx->timing.remote + MS2NTP(reference - ctx->timing.local);
expected = ctx->timing.remote + MS2NTP(reference - ctx->timing.local);
ctx->timing.remote = remote;
ctx->timing.local = reference;
ctx->timing.count++;
if (!ctx->timing.drift && (ctx->synchro.status & NTP_SYNC)) {
delta = NTP2MS((s64_t) expected - (s64_t) ctx->timing.remote);
ctx->timing.gap_sum += delta;
pthread_mutex_lock(&ctx->ab_mutex);
/*
if expected time is more than remote, then our time is
running faster and we are transmitting frames too quickly,
so we'll run out of frames, need to add one
*/
if (ctx->timing.gap_sum > GAP_THRES && ctx->timing.gap_count++ > GAP_COUNT) {
LOG_INFO("[%p]: Sending packets too fast %Ld [W:%hu R:%hu]", ctx, ctx->timing.gap_sum, ctx->ab_write, ctx->ab_read);
ctx->ab_read--;
ctx->audio_buffer[BUFIDX(ctx->ab_read)].ready = 1;
ctx->timing.gap_sum -= GAP_THRES;
ctx->timing.gap_adjust -= GAP_THRES;
/*
if expected time is less than remote, then our time is
running slower and we are transmitting frames too slowly,
so we'll overflow frames buffer, need to remove one
*/
} else if (ctx->timing.gap_sum < -GAP_THRES && ctx->timing.gap_count++ > GAP_COUNT) {
if (seq_order(ctx->ab_read, ctx->ab_write + 1)) {
ctx->audio_buffer[BUFIDX(ctx->ab_read)].ready = 0;
ctx->ab_read++;
} else ctx->skip++;
ctx->timing.gap_sum += GAP_THRES;
ctx->timing.gap_adjust += GAP_THRES;
LOG_INFO("[%p]: Sending packets too slow %Ld (skip: %d) [W:%hu R:%hu]", ctx, ctx->timing.gap_sum, ctx->skip, ctx->ab_write, ctx->ab_read);
}
ctx->timing.remote = remote;
ctx->timing.local = reference;
if (ctx->synchro.status & NTP_SYNC) {
s32_t delta = NTP2MS((s64_t) expected - (s64_t) ctx->timing.remote);
ctx->cmd_cb(RAOP_TIMING, &delta);
}
// now we are synced on NTP (mutex not needed)
// now we are synced on NTP (mutex not needed)
ctx->synchro.status |= NTP_SYNC;
LOG_DEBUG("[%p]: Timing references local:%Lu, remote:%Lx (delta:%Ld, sum:%Ld, adjust:%Ld, gaps:%d)",
@@ -778,7 +732,7 @@ static bool rtp_request_resend(rtp_t *ctx, seq_t first, seq_t last) {
/*---------------------------------------------------------------------------*/
static bool rtp_request_resend(rtp_t *ctx, seq_t first, seq_t last) {
unsigned char req[8]; // *not* a standard RTCP NACK
// do not request silly ranges (happens in case of network large blackouts)
if (seq_order(last, first) || last - first > BUFFER_FRAMES / 2) return false;
@@ -797,99 +751,3 @@ static bool rtp_request_resend(rtp_t *ctx, seq_t first, seq_t last) {
if (sizeof(req) != sendto(ctx->rtp_sockets[CONTROL].sock, req, sizeof(req), 0, (struct sockaddr*) &ctx->rtp_host, sizeof(ctx->rtp_host))) {
LOG_WARN("[%p]: SENDTO failed (%s)", ctx, strerror(errno));
}
return true;
}
#if 0
/*---------------------------------------------------------------------------*/
// get the next frame, when available. return 0 if underrun/stream reset.
static short *_buffer_get_frame(rtp_t *ctx, int *len) {
short buf_fill;
abuf_t *curframe = 0;
int i;
u32_t now, playtime;
if (!ctx->playing) return NULL;
// skip frames if we are running late and skip could not be done in SYNC
while (ctx->skip && seq_order(ctx->ab_read, ctx->ab_write + 1)) {
ctx->audio_buffer[BUFIDX(ctx->ab_read)].ready = 0;
ctx->ab_read++;
ctx->skip--;
LOG_INFO("[%p]: Sending packets too slow (skip: %d) [W:%hu R:%hu]", ctx, ctx->skip, ctx->ab_write, ctx->ab_read);
}
buf_fill = ctx->ab_write - ctx->ab_read + 1;
if (buf_fill >= BUFFER_FRAMES) {
LOG_ERROR("[%p]: Buffer overrun %hu", ctx, buf_fill);
ctx->ab_read = ctx->ab_write - (BUFFER_FRAMES - 64);
buf_fill = ctx->ab_write - ctx->ab_read + 1;
}
now = gettime_ms();
curframe = ctx->audio_buffer + BUFIDX(ctx->ab_read);
// use next frame when buffer is empty or silence continues to be sent
if (!buf_fill) curframe->rtptime = ctx->audio_buffer[BUFIDX(ctx->ab_read - 1)].rtptime + ctx->frame_size;
playtime = ctx->synchro.time + (((s32_t)(curframe->rtptime - ctx->synchro.rtp))*1000)/44100;
LOG_SDEBUG("playtime %u %d [W:%hu R:%hu] %d", playtime, playtime - now, ctx->ab_write, ctx->ab_read, curframe->ready);
// wait if not ready but have time, otherwise send silence
if (!buf_fill || ctx->synchro.status != (RTP_SYNC | NTP_SYNC) || (now < playtime && !curframe->ready)) {
LOG_SDEBUG("[%p]: waiting (fill:%hd, W:%hu R:%hu) now:%u, playtime:%u, wait:%d", ctx, buf_fill, ctx->ab_write, ctx->ab_read, now, playtime, playtime - now);
// look for "blocking" frames at the top of the queue and try to catch-up
for (i = 0; i < min(16, buf_fill); i++) {
abuf_t *frame = ctx->audio_buffer + BUFIDX(ctx->ab_read + i);
if (!frame->ready && now - frame->last_resend > RESEND_TO) {
rtp_request_resend(ctx, ctx->ab_read + i, ctx->ab_read + i);
frame->last_resend = now;
}
}
return NULL;
}
// when silence is inserted at the top, need to move write pointer
if (!buf_fill) {
if (!ctx->filled_frames) {
LOG_WARN("[%p]: start silence (late %d ms) [W:%hu R:%hu]", ctx, now - playtime, ctx->ab_write, ctx->ab_read);
}
ctx->ab_write++;
ctx->filled_frames++;
} else ctx->filled_frames = 0;
if (!(ctx->out_frames++ & 0x1ff)) {
LOG_INFO("[%p]: drain [level:%hd gap:%d] [W:%hu R:%hu] [R:%u S:%u F:%u]",
ctx, buf_fill-1, playtime - now, ctx->ab_write, ctx->ab_read,
ctx->resent_frames, ctx->silent_frames, ctx->filled_frames);
}
// each missing packet will be requested up to (latency_frames / 16) times
for (i = 16; seq_order(ctx->ab_read + i, ctx->ab_write); i += 16) {
abuf_t *frame = ctx->audio_buffer + BUFIDX(ctx->ab_read + i);
if (!frame->ready && now - frame->last_resend > RESEND_TO) {
rtp_request_resend(ctx, ctx->ab_read + i, ctx->ab_read + i);
frame->last_resend = now;
}
}
if (!curframe->ready) {
LOG_DEBUG("[%p]: created zero frame (W:%hu R:%hu)", ctx, ctx->ab_write, ctx->ab_read);
memset(curframe->data, 0, ctx->frame_size*4);
curframe->len = ctx->frame_size * 4;
ctx->silent_frames++;
} else {
LOG_SDEBUG("[%p]: prepared frame (fill:%hd, W:%hu R:%hu)", ctx, buf_fill-1, ctx->ab_write, ctx->ab_read);
}
*len = curframe->len;
curframe->ready = 0;
ctx->ab_read++;
return curframe->data;
}