[streaming] Change how metadata is delivered to http streaming

This gets rid of player locks + the special header file outputs/streaming.h
This commit is contained in:
ejurgensen 2023-05-08 20:46:16 +02:00
parent 6364515fb7
commit f998b1f3dd
7 changed files with 217 additions and 177 deletions

View File

@ -118,7 +118,7 @@ owntone_SOURCES = main.c \
outputs/rtp_common.h outputs/rtp_common.c \
outputs/raop.c outputs/airplay.c $(PAIR_AP_SRC) \
outputs/airplay_events.c outputs/airplay_events.h \
outputs/streaming.c outputs/streaming.h \
outputs/streaming.c \
outputs/dummy.c outputs/fifo.c outputs/rcp.c \
$(ALSA_SRC) $(PULSEAUDIO_SRC) $(CHROMECAST_SRC) \
evrtsp/rtsp.c evrtsp/evrtsp.h evrtsp/rtsp-internal.h evrtsp/log.h \

View File

@ -32,20 +32,24 @@
#include "httpd_internal.h"
#include "player.h"
#include "outputs/streaming.h"
#include "logger.h"
#include "conffile.h"
#define STREAMING_ICY_METALEN_MAX 4080 // 255*16 incl header/footer (16bytes)
#define STREAMING_ICY_METATITLELEN_MAX 4064 // STREAMING_ICY_METALEN_MAX -16 (not incl header/footer)
struct streaming_session {
struct httpd_request *hreq;
int fd;
struct event *readev;
struct evbuffer *readbuf;
int id;
struct event *audioev;
struct event *metadataev;
struct evbuffer *audiobuf;
size_t bytes_sent;
bool icy_is_requested;
size_t icy_remaining;
char icy_title[STREAMING_ICY_METATITLELEN_MAX];
};
static struct media_quality streaming_default_quality = {
@ -55,25 +59,20 @@ static struct media_quality streaming_default_quality = {
.bit_rate = 128000,
};
static void
session_free(struct streaming_session *session);
/* ------------------------------ ICY metadata -------------------------------*/
// To test mp3 and ICY tagm it is good to use:
// mpv --display-tags=* http://localhost:3689/stream.mp3
#define STREAMING_ICY_METALEN_MAX 4080 // 255*16 incl header/footer (16bytes)
#define STREAMING_ICY_METATITLELEN_MAX 4064 // STREAMING_ICY_METALEN_MAX -16 (not incl header/footer)
// As streaming quality goes up, we send more data to the remote client. With a
// smaller ICY_METAINT value we have to splice metadata more frequently - on
// some devices with small input buffers, a higher quality stream and low
// ICY_METAINT can lead to stuttering as observed on a Roku Soundbridge
static unsigned short streaming_icy_metaint = 16384;
static pthread_mutex_t streaming_metadata_lck;
static char streaming_icy_title[STREAMING_ICY_METATITLELEN_MAX];
// We know that the icymeta is limited to 1+255*16 (ie 4081) bytes so caller must
// provide a buf of this size to avoid needless mallocs
//
@ -125,7 +124,7 @@ icy_meta_create(uint8_t buf[STREAMING_ICY_METALEN_MAX+1], unsigned *buflen, cons
}
static void
icy_meta_splice(struct evbuffer *out, struct evbuffer *in, size_t *icy_remaining)
icy_meta_splice(struct evbuffer *out, struct evbuffer *in, size_t *icy_remaining, char *title)
{
uint8_t meta[STREAMING_ICY_METALEN_MAX + 1];
unsigned metalen;
@ -139,9 +138,7 @@ icy_meta_splice(struct evbuffer *out, struct evbuffer *in, size_t *icy_remaining
*icy_remaining -= consume;
if (*icy_remaining == 0)
{
pthread_mutex_lock(&streaming_metadata_lck);
icy_meta_create(meta, &metalen, streaming_icy_title);
pthread_mutex_unlock(&streaming_metadata_lck);
icy_meta_create(meta, &metalen, title);
evbuffer_add(out, meta, metalen);
*icy_remaining = streaming_icy_metaint;
@ -149,50 +146,6 @@ icy_meta_splice(struct evbuffer *out, struct evbuffer *in, size_t *icy_remaining
}
}
// Thread: player. TODO Would be nice to avoid the lock. Consider moving all the
// ICY tag stuff to streaming.c and make a STREAMING_FORMAT_MP3_ICY?
static void
icy_metadata_cb(char *metadata)
{
pthread_mutex_lock(&streaming_metadata_lck);
snprintf(streaming_icy_title, sizeof(streaming_icy_title), "%s", metadata);
pthread_mutex_unlock(&streaming_metadata_lck);
}
/* ----------------------------- Session helpers ---------------------------- */
static void
session_free(struct streaming_session *session)
{
if (!session)
return;
if (session->readev)
{
player_streaming_deregister(session->fd);
event_free(session->readev);
}
evbuffer_free(session->readbuf);
free(session);
}
static struct streaming_session *
session_new(struct httpd_request *hreq, bool icy_is_requested)
{
struct streaming_session *session;
CHECK_NULL(L_STREAMING, session = calloc(1, sizeof(struct streaming_session)));
CHECK_NULL(L_STREAMING, session->readbuf = evbuffer_new());
session->hreq = hreq;
session->icy_is_requested = icy_is_requested;
session->icy_remaining = streaming_icy_metaint;
return session;
}
/* ----------------------------- Event callbacks ---------------------------- */
@ -205,7 +158,7 @@ conn_close_cb(void *arg)
}
static void
read_cb(evutil_socket_t fd, short event, void *arg)
audio_cb(evutil_socket_t fd, short event, void *arg)
{
struct streaming_session *session = arg;
struct httpd_request *hreq;
@ -213,7 +166,7 @@ read_cb(evutil_socket_t fd, short event, void *arg)
CHECK_NULL(L_STREAMING, hreq = session->hreq);
len = evbuffer_read(session->readbuf, fd, -1);
len = evbuffer_read(session->audiobuf, fd, -1);
if (len < 0 && errno != EAGAIN)
{
DPRINTF(E_INFO, L_STREAMING, "Stopping mp3 streaming to %s:%d\n", session->hreq->peer_address, (int)session->hreq->peer_port);
@ -224,15 +177,87 @@ read_cb(evutil_socket_t fd, short event, void *arg)
}
if (session->icy_is_requested)
icy_meta_splice(hreq->out_body, session->readbuf, &session->icy_remaining);
icy_meta_splice(hreq->out_body, session->audiobuf, &session->icy_remaining, session->icy_title);
else
evbuffer_add_buffer(hreq->out_body, session->readbuf);
evbuffer_add_buffer(hreq->out_body, session->audiobuf);
httpd_send_reply_chunk(hreq, NULL, NULL);
session->bytes_sent += len;
}
static void
metadata_cb(evutil_socket_t fd, short event, void *arg)
{
struct streaming_session *session = arg;
struct evbuffer *evbuf;
int len;
CHECK_NULL(L_STREAMING, evbuf = evbuffer_new());
len = evbuffer_read(evbuf, fd, -1);
if (len < 0)
goto out;
len = sizeof(session->icy_title);
evbuffer_remove(evbuf, session->icy_title, len);
session->icy_title[len - 1] = '\0';
out:
evbuffer_free(evbuf);
}
/* ----------------------------- Session helpers ---------------------------- */
static void
session_free(struct streaming_session *session)
{
if (!session)
return;
player_streaming_deregister(session->id);
if (session->audioev)
event_free(session->audioev);
if (session->metadataev)
event_free(session->metadataev);
evbuffer_free(session->audiobuf);
free(session);
}
static struct streaming_session *
session_new(struct httpd_request *hreq, bool icy_is_requested, enum player_format format, struct media_quality quality)
{
struct streaming_session *session;
int audio_fd;
int metadata_fd;
CHECK_NULL(L_STREAMING, session = calloc(1, sizeof(struct streaming_session)));
CHECK_NULL(L_STREAMING, session->audiobuf = evbuffer_new());
session->hreq = hreq;
session->icy_is_requested = icy_is_requested;
session->icy_remaining = streaming_icy_metaint;
// Ask streaming output module for a fd to read mp3 from
session->id = player_streaming_register(&audio_fd, &metadata_fd, format, quality);
if (session->id < 0)
goto error;
CHECK_NULL(L_STREAMING, session->audioev = event_new(hreq->evbase, audio_fd, EV_READ | EV_PERSIST, audio_cb, session));
event_add(session->audioev, NULL);
CHECK_NULL(L_STREAMING, session->metadataev = event_new(hreq->evbase, metadata_fd, EV_READ | EV_PERSIST, metadata_cb, session));
event_add(session->metadataev, NULL);
return session;
error:
session_free(session);
return NULL;
}
/* -------------------------- Module implementation ------------------------- */
@ -254,21 +279,9 @@ streaming_mp3_handler(struct httpd_request *hreq)
httpd_header_add(hreq->out_headers, "icy-metaint", buf);
}
session = session_new(hreq, icy_is_requested);
session = session_new(hreq, icy_is_requested, PLAYER_FORMAT_MP3, streaming_default_quality);
if (!session)
{
goto error;
}
// Ask streaming output module for a fd to read mp3 from
session->fd = player_streaming_register(STREAMING_FORMAT_MP3, streaming_default_quality);
if (session->fd < 0)
{
goto error;
}
CHECK_NULL(L_STREAMING, session->readev = event_new(hreq->evbase, session->fd, EV_READ | EV_PERSIST, read_cb, session));
event_add(session->readev, NULL);
return -1; // Error sent by caller
httpd_request_close_cb_set(hreq, conn_close_cb, session);
@ -281,11 +294,6 @@ streaming_mp3_handler(struct httpd_request *hreq)
httpd_send_reply_start(hreq, HTTP_OK, "OK");
return 0;
error:
session_free(session);
// Error message is sent by streaming_request()
return -1;
}
static struct httpd_uri_map streaming_handlers[] =
@ -357,9 +365,6 @@ streaming_init(void)
else
DPRINTF(E_INFO, L_STREAMING, "Unsupported icy_metaint=%d, supported range: 4096..131072, defaulting to %d\n", val, streaming_icy_metaint);
CHECK_ERR(L_STREAMING, mutex_init(&streaming_metadata_lck));
streaming_metadatacb_register(icy_metadata_cb);
return 0;
}

View File

@ -142,6 +142,10 @@ struct output_device
short v4_port;
short v6_port;
// Only used for streaming
int audio_fd;
int metadata_fd;
struct event *stop_timer;
// Opaque pointers to device and session data

View File

@ -28,7 +28,6 @@
#include <uninorm.h>
#include <fcntl.h>
#include "streaming.h"
#include "outputs.h"
#include "misc.h"
#include "worker.h"
@ -63,9 +62,10 @@ struct pipepair
struct streaming_wanted
{
int num_sessions; // for refcounting
struct pipepair pipes[WANTED_PIPES_MAX];
struct pipepair audio[WANTED_PIPES_MAX];
struct pipepair metadata[WANTED_PIPES_MAX];
enum streaming_format format;
enum player_format format;
struct media_quality quality;
struct evbuffer *audio_in;
@ -86,12 +86,11 @@ struct streaming_ctx
struct timeval silencetv;
struct media_quality last_quality;
char title[4064]; // See STREAMING_ICY_METALEN_MAX in http_streaming.c
// seqnum may wrap around so must be unsigned
unsigned int seqnum;
unsigned int seqnum_encode_next;
// callback with new metadata, e.g. for ICY tags
void (*metadatacb)(char *metadata);
};
struct encode_cmdarg
@ -114,7 +113,7 @@ extern struct event_base *evbase_player;
/* ------------------------------- Helpers ---------------------------------- */
static struct encode_ctx *
encoder_setup(enum streaming_format format, struct media_quality *quality)
encoder_setup(enum player_format format, struct media_quality *quality)
{
struct decode_ctx *decode_ctx = NULL;
struct encode_ctx *encode_ctx = NULL;
@ -133,7 +132,7 @@ encoder_setup(enum streaming_format format, struct media_quality *quality)
goto out;
}
if (format == STREAMING_FORMAT_MP3)
if (format == PLAYER_FORMAT_MP3)
encode_ctx = transcode_encode_setup(XCODE_MP3, quality, decode_ctx, NULL, 0, 0);
if (!encode_ctx)
@ -194,7 +193,9 @@ wanted_free(struct streaming_wanted *w)
return;
for (int i = 0; i < WANTED_PIPES_MAX; i++)
pipe_close(&w->pipes[i]);
pipe_close(&w->audio[i]);
for (int i = 0; i < WANTED_PIPES_MAX; i++)
pipe_close(&w->metadata[i]);
transcode_encode_cleanup(&w->xcode_ctx);
evbuffer_free(w->audio_in);
@ -203,8 +204,20 @@ wanted_free(struct streaming_wanted *w)
free(w);
}
static int
pipe_index_find_byreadfd(struct pipepair *p, int readfd)
{
for (int i = 0; i < WANTED_PIPES_MAX; i++, p++)
{
if (p->readfd == readfd)
return i;
}
return -1;
}
static struct streaming_wanted *
wanted_new(enum streaming_format format, struct media_quality quality)
wanted_new(enum player_format format, struct media_quality quality)
{
struct streaming_wanted *w;
@ -225,8 +238,10 @@ wanted_new(enum streaming_format format, struct media_quality quality)
for (int i = 0; i < WANTED_PIPES_MAX; i++)
{
w->pipes[i].writefd = -1;
w->pipes[i].readfd = -1;
w->audio[i].writefd = -1;
w->audio[i].readfd = -1;
w->metadata[i].writefd = -1;
w->metadata[i].readfd = -1;
}
return w;
@ -262,7 +277,7 @@ wanted_remove(struct streaming_wanted **wanted, struct streaming_wanted *remove)
}
static struct streaming_wanted *
wanted_add(struct streaming_wanted **wanted, enum streaming_format format, struct media_quality quality)
wanted_add(struct streaming_wanted **wanted, enum player_format format, struct media_quality quality)
{
struct streaming_wanted *w;
@ -274,7 +289,7 @@ wanted_add(struct streaming_wanted **wanted, enum streaming_format format, struc
}
static struct streaming_wanted *
wanted_find_byformat(struct streaming_wanted *wanted, enum streaming_format format, struct media_quality quality)
wanted_find_byformat(struct streaming_wanted *wanted, enum player_format format, struct media_quality quality)
{
struct streaming_wanted *w;
@ -294,31 +309,36 @@ wanted_find_byreadfd(struct streaming_wanted *wanted, int readfd)
int i;
for (w = wanted; w; w = w->next)
for (i = 0; i < WANTED_PIPES_MAX; i++)
{
if (w->pipes[i].readfd == readfd)
return w;
}
{
i = pipe_index_find_byreadfd(w->audio, readfd);
if (i != -1)
return w;
}
return NULL;
}
static int
wanted_session_add(struct pipepair *p, struct streaming_wanted *w)
wanted_session_add(int *audiofd, int *metadatafd, struct streaming_wanted *w)
{
int ret;
int i;
for (i = 0; i < WANTED_PIPES_MAX; i++)
{
if (w->pipes[i].writefd != -1) // In use
if (w->audio[i].writefd != -1) // In use
continue;
ret = pipe_open(&w->pipes[i]);
ret = pipe_open(&w->audio[i]);
if (ret < 0)
return -1;
memcpy(p, &w->pipes[i], sizeof(struct pipepair));
ret = pipe_open(&w->metadata[i]);
if (ret < 0)
return -1;
*audiofd = w->audio[i].readfd;
*metadatafd = w->metadata[i].readfd;
break;
}
@ -329,31 +349,25 @@ wanted_session_add(struct pipepair *p, struct streaming_wanted *w)
}
w->num_sessions++;
DPRINTF(E_DBG, L_STREAMING, "Session register readfd %d, wanted->num_sessions=%d\n", p->readfd, w->num_sessions);
DPRINTF(E_DBG, L_STREAMING, "Session register audiofd %d, metadatafd %d, wanted->num_sessions=%d\n", *audiofd, *metadatafd, w->num_sessions);
return 0;
}
static void
wanted_session_remove(struct streaming_wanted *w, int readfd)
{
int i;
for (i = 0; i < WANTED_PIPES_MAX; i++)
{
if (w->pipes[i].readfd != readfd)
continue;
pipe_close(&w->pipes[i]);
break;
}
if (i == WANTED_PIPES_MAX)
i = pipe_index_find_byreadfd(w->audio, readfd);
if (i < 0)
{
DPRINTF(E_LOG, L_STREAMING, "Cannot remove streaming session, readfd %d not found\n", readfd);
return;
}
pipe_close(&w->audio[i]);
pipe_close(&w->metadata[i]);
w->num_sessions--;
DPRINTF(E_DBG, L_STREAMING, "Session deregister readfd %d, wanted->num_sessions=%d\n", readfd, w->num_sessions);
}
@ -421,7 +435,6 @@ encode_buffer(struct streaming_wanted *w, uint8_t *buf, size_t bufsize)
static void
encode_and_write(int *failed_pipe_readfd, struct streaming_wanted *w, struct output_buffer *obuf)
{
struct pipepair *p;
uint8_t *buf;
size_t bufsize;
size_t len;
@ -444,10 +457,8 @@ encode_and_write(int *failed_pipe_readfd, struct streaming_wanted *w, struct out
{
for (i = 0; i < WANTED_PIPES_MAX; i++)
{
p = &w->pipes[i];
if (p->writefd < 0)
continue;
*failed_pipe_readfd = p->readfd;
if (w->audio[i].writefd != -1)
*failed_pipe_readfd = w->audio[i].readfd;
}
return;
@ -462,15 +473,14 @@ encode_and_write(int *failed_pipe_readfd, struct streaming_wanted *w, struct out
buf = evbuffer_pullup(w->audio_out, -1);
for (i = 0; i < WANTED_PIPES_MAX; i++)
{
p = &w->pipes[i];
if (p->writefd < 0)
if (w->audio[i].writefd == -1)
continue;
ret = write(p->writefd, buf, len);
ret = write(w->audio[i].writefd, buf, len);
if (ret < 0)
{
DPRINTF(E_LOG, L_STREAMING, "Error writing to stream pipe %d (format %d): %s\n", p->writefd, w->format, strerror(errno));
*failed_pipe_readfd = p->readfd;
DPRINTF(E_LOG, L_STREAMING, "Error writing to stream pipe %d (format %d): %s\n", w->audio[i].writefd, w->format, strerror(errno));
*failed_pipe_readfd = w->audio[i].readfd;
}
}
@ -506,11 +516,44 @@ encode_data_cb(void *arg)
player_streaming_deregister(failed_pipe_readfd);
}
static void
metadata_write(struct streaming_wanted *w, int readfd, const char *metadata)
{
size_t metadata_size;
int i;
int ret;
for (i = 0; i < WANTED_PIPES_MAX; i++)
{
if (w->metadata[i].writefd == -1)
continue;
if (readfd >= 0 && w->metadata[i].readfd != readfd)
continue;
metadata_size = strlen(metadata) + 1;
ret = write(w->metadata[i].writefd, metadata, metadata_size);
if (ret < 0)
DPRINTF(E_WARN, L_STREAMING, "Error writing metadata '%s' to fd %d\n", metadata, w->metadata[i].writefd);
}
}
static void
metadata_startup_cb(void *arg)
{
int *metadata_fd = arg;
struct streaming_wanted *w;
pthread_mutex_lock(&streaming_wanted_lck);
for (w = streaming.wanted; w; w = w->next)
metadata_write(w, *metadata_fd, streaming.title);
pthread_mutex_unlock(&streaming_wanted_lck);
}
static void *
streaming_metadata_prepare(struct output_metadata *metadata)
{
struct db_queue_item *queue_item;
char *title;
struct streaming_wanted *w;
queue_item = db_queue_fetch_byitemid(metadata->item_id);
if (!queue_item)
@ -519,22 +562,19 @@ streaming_metadata_prepare(struct output_metadata *metadata)
return NULL;
}
title = safe_asprintf("%s - %s", queue_item->title, queue_item->artist);
pthread_mutex_lock(&streaming_wanted_lck);
// Save it here, we might need it later if a new session starts up
snprintf(streaming.title, sizeof(streaming.title), "%s - %s", queue_item->title, queue_item->artist);
for (w = streaming.wanted; w; w = w->next)
metadata_write(w, -1, streaming.title);
pthread_mutex_unlock(&streaming_wanted_lck);
free_queue_item(queue_item, 0);
return title;
return NULL;
}
/* ----------------------------- Thread: httpd ------------------------------ */
// Not thread safe, but only called once during httpd init
void
streaming_metadatacb_register(streaming_metadatacb cb)
{
streaming.metadatacb = cb;
}
/* ----------------------------- Thread: Player ----------------------------- */
static void
@ -570,14 +610,7 @@ silenceev_cb(evutil_socket_t fd, short event, void *arg)
static void
streaming_metadata_send(struct output_metadata *metadata)
{
char *title = metadata->priv;
// Calls back to httpd_streaming to update the title
if (streaming.metadatacb)
streaming.metadatacb(title);
free(title);
outputs_metadata_free(metadata);
// Nothing to do, metadata_prepare() did all we needed in a worker thread
}
// Since this is streaming and there is no actual device, we will be called with
@ -587,20 +620,22 @@ static int
streaming_start(struct output_device *device, int callback_id)
{
struct streaming_wanted *w;
struct pipepair pipe;
int ret;
pthread_mutex_lock(&streaming_wanted_lck);
w = wanted_find_byformat(streaming.wanted, device->format, device->quality);
if (!w)
w = wanted_add(&streaming.wanted, device->format, device->quality);
ret = wanted_session_add(&pipe, w);
ret = wanted_session_add(&device->audio_fd, &device->metadata_fd, w);
if (ret < 0)
goto error;
pthread_mutex_unlock(&streaming_wanted_lck);
worker_execute(metadata_startup_cb, &(device->metadata_fd), sizeof(device->metadata_fd), 0);
outputs_quality_subscribe(&device->quality);
device->id = pipe.readfd; // Not super clean
device->id = device->audio_fd;
return 0;
error:

View File

@ -1,15 +0,0 @@
#ifndef __STREAMING_H__
#define __STREAMING_H__
typedef void (*streaming_metadatacb)(char *metadata);
enum streaming_format
{
STREAMING_FORMAT_MP3,
};
void
streaming_metadatacb_register(streaming_metadatacb cb);
#endif /* !__STREAMING_H__ */

View File

@ -152,7 +152,10 @@ struct speaker_attr_param
bool busy;
struct media_quality quality;
int format;
enum player_format format;
int audio_fd;
int metadata_fd;
const char *pin;
};
@ -2904,10 +2907,10 @@ streaming_register(void *arg, int *retval)
};
*retval = outputs_device_start(&device, NULL, false);
if (*retval < 0)
return COMMAND_END;
*retval = device.id; // Actually the fd that the called needs
param->spk_id = device.id;
param->audio_fd = device.audio_fd;
param->metadata_fd = device.metadata_fd;
return COMMAND_END;
}
@ -3470,7 +3473,7 @@ player_speaker_authorize(uint64_t id, const char *pin)
}
int
player_streaming_register(int format, struct media_quality quality)
player_streaming_register(int *audio_fd, int *metadata_fd, enum player_format format, struct media_quality quality)
{
struct speaker_attr_param param;
int ret;
@ -3479,8 +3482,12 @@ player_streaming_register(int format, struct media_quality quality)
param.quality = quality;
ret = commands_exec_sync(cmdbase, streaming_register, NULL, &param);
if (ret < 0)
return ret;
return ret;
*audio_fd = param.audio_fd;
*metadata_fd = param.metadata_fd;
return param.spk_id;
}
int

View File

@ -28,6 +28,10 @@ enum player_seek_mode {
PLAYER_SEEK_RELATIVE = 2,
};
enum player_format {
PLAYER_FORMAT_MP3,
};
struct player_speaker_info {
uint64_t id;
uint32_t active_remote;
@ -119,7 +123,7 @@ int
player_speaker_authorize(uint64_t id, const char *pin);
int
player_streaming_register(int format, struct media_quality quality);
player_streaming_register(int *audio_fd, int *metadata_fd, enum player_format format, struct media_quality quality);
int
player_streaming_deregister(int id);