nils/shairport-sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Try to add proper cancellation code to all threads -- not complete, buggy.

Browse Source
This commit is contained in:
Mike Brady
2018-07-20 13:50:38 +01:00
parent c8c70b6016
commit f1d4503431
19 changed files with 330 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
audio_alsa.c
+6 -4
View File
@@ -407,8 +407,9 @@ static int init(int argc, char **argv) {
(1.0 * alsa_mix_maxdb) / 100.0);
} else {
// use the linear scale and do the db conversion ourselves
debug(1, "note: the hardware mixer specified -- \"%s\" -- does not have "
"a dB volume scale.",
debug(1,
"note: the hardware mixer specified -- \"%s\" -- does not have "
"a dB volume scale.",
alsa_mix_ctrl);
if (snd_ctl_open(&ctl, alsa_mix_dev, 0) < 0) {
@@ -661,8 +662,9 @@ int open_alsa_device(void) {
buffer_size);
}
*/
debug(1, "The alsa buffer is smaller (%lu bytes) than the desired backend buffer "
"length (%ld) you have chosen.",
debug(1,
"The alsa buffer is smaller (%lu bytes) than the desired backend buffer "
"length (%ld) you have chosen.",
actual_buffer_length, config.audio_backend_buffer_desired_length);
}
audio_soundio.c
+3 -2
View File
@@ -27,8 +27,9 @@ static void write_callback(struct SoundIoOutStream *outstream, int frame_count_m
int fill_bytes = soundio_ring_buffer_fill_count(ring_buffer);
int fill_count = fill_bytes / outstream->bytes_per_frame;
debug(3, "[--->>] frame_count_min: %d , frame_count_max: %d , fill_bytes: %d , fill_count: %d , "
"outstream->bytes_per_frame: %d",
debug(3,
"[--->>] frame_count_min: %d , frame_count_max: %d , fill_bytes: %d , fill_count: %d , "
"outstream->bytes_per_frame: %d",
frame_count_min, frame_count_max, fill_bytes, fill_count, outstream->bytes_per_frame);
if (frame_count_min > fill_count) {
common.c
+11 -18
View File
@@ -120,7 +120,6 @@ void die(const char *format, ...) {
daemon_log(LOG_EMERG, "% 20.9f|*fatal error: %s", tss, s);
else
daemon_log(LOG_EMERG, "fatal error: %s", s);
shairport_shutdown();
exit(1);
}
@@ -748,21 +747,17 @@ double flat_vol2attn(double vol, long max_db, long min_db) {
double vol2attn(double vol, long max_db, long min_db) {
// We use a little coordinate geometry to build a transfer function from the volume passed in to the
// device's dynamic range.
// (See the diagram in the documents folder.)
// The x axis is the "volume in" which will be from -30 to 0. The y axis will be the "volume out"
// which will be from the bottom of the range to the top.
// We build the transfer function from one or more lines. We characterise each line with two
// numbers:
// the first is where on x the line starts when y=0 (x can be from 0 to -30); the second is where on
// y the line stops when when x is -30.
// thus, if the line was characterised as {0,-30}, it would be an identity transfer.
// Assuming, for example, a dynamic range of lv=-60 to hv=0
// Typically we'll use three lines -- a three order transfer function
// First: {0,30} giving a gentle slope -- the 30 comes from half the dynamic range
// Second: {-5,-30-(lv+30)/2} giving a faster slope from y=0 at x=-12 to y=-42.5 at x=-30
// Third: {-17,lv} giving a fast slope from y=0 at x=-19 to y=-60 at x=-30
// We use a little coordinate geometry to build a transfer function from the volume passed in to
// the device's dynamic range. (See the diagram in the documents folder.) The x axis is the
// "volume in" which will be from -30 to 0. The y axis will be the "volume out" which will be from
// the bottom of the range to the top. We build the transfer function from one or more lines. We
// characterise each line with two numbers: the first is where on x the line starts when y=0 (x
// can be from 0 to -30); the second is where on y the line stops when when x is -30. thus, if the
// line was characterised as {0,-30}, it would be an identity transfer. Assuming, for example, a
// dynamic range of lv=-60 to hv=0 Typically we'll use three lines -- a three order transfer
// function First: {0,30} giving a gentle slope -- the 30 comes from half the dynamic range
// Second: {-5,-30-(lv+30)/2} giving a faster slope from y=0 at x=-12 to y=-42.5 at x=-30
// Third: {-17,lv} giving a fast slope from y=0 at x=-19 to y=-60 at x=-30
#define order 3
@@ -966,8 +961,6 @@ int64_t r64i() { return (ranval(&rx) >> 1); }
/* generate an array of 64-bit random numbers */
const int ranarraylength = 1009; // these will be 8-byte numbers.
uint64_t *ranarray;
int ranarraynext;
void ranarrayinit() {
common.h
+1
View File
@@ -229,6 +229,7 @@ void r64init(uint64_t seed);
uint64_t r64u();
int64_t r64i();
uint64_t *ranarray;
void r64arrayinit();
uint64_t ranarray64u();
int64_t ranarray64i();
dacp.c
+31 -8
View File
@@ -110,7 +110,10 @@ static void response_code(void *opaque, int code) {
}
static const struct http_funcs responseFuncs = {
response_realloc, response_body, response_header, response_code,
response_realloc,
response_body,
response_header,
response_code,
};
// static pthread_mutex_t dacp_conversation_lock = PTHREAD_MUTEX_INITIALIZER;
@@ -261,8 +264,9 @@ int dacp_send_command(const char *command, char **body, ssize_t *bodysize) {
// debug(1,"Sent command\"%s\" with a response body of size %d.",command,response.size);
// debug(1,"dacp_conversation_lock released.");
} else {
debug(3, "Could not acquire a lock on the dacp transmit/receive section when attempting to "
"send the command \"%s\". Possible timeout?",
debug(3,
"Could not acquire a lock on the dacp transmit/receive section when attempting to "
"send the command \"%s\". Possible timeout?",
command);
response.code = 494; // This client is already busy
}
@@ -384,6 +388,12 @@ void dacp_monitor_port_update_callback(char *dacp_id, uint16_t port) {
pthread_cond_signal(&dacp_server_information_cv);
pthread_mutex_unlock(&dacp_server_information_lock);
}
void dacp_monitor_thread_code_cleanup(__attribute__((unused)) void *arg) {
debug(1, "dacp_monitor_thread_code_cleanup called.");
pthread_mutex_unlock(&dacp_server_information_lock);
}
void *dacp_monitor_thread_code(__attribute__((unused)) void *na) {
int scan_index = 0;
// char server_reply[10000];
@@ -397,14 +407,18 @@ void *dacp_monitor_thread_code(__attribute__((unused)) void *na) {
sps_pthread_mutex_timedlock(
&dacp_server_information_lock, 500000,
"dacp_monitor_thread_code couldn't get DACP server information lock in 0.5 second!.", 2);
int32_t the_volume;
pthread_cleanup_push(dacp_monitor_thread_code_cleanup, NULL);
if (dacp_server.scan_enable == 0) {
metadata_hub_modify_prolog();
int ch = (metadata_store.dacp_server_active != 0) ||
(metadata_store.advanced_dacp_server_active != 0);
metadata_store.dacp_server_active = 0;
metadata_store.advanced_dacp_server_active = 0;
debug(2, "setting dacp_server_active and advanced_dacp_server_active to 0 with an update "
"flag value of %d",
debug(2,
"setting dacp_server_active and advanced_dacp_server_active to 0 with an update "
"flag value of %d",
ch);
metadata_hub_modify_epilog(ch);
while (dacp_server.scan_enable == 0) {
@@ -416,7 +430,6 @@ void *dacp_monitor_thread_code(__attribute__((unused)) void *na) {
idle_scan_count = 0;
}
scan_index++;
int32_t the_volume;
result = dacp_get_volume(&the_volume); // just want the http code
if ((result == 496) || (result == 403) || (result == 501)) {
@@ -438,7 +451,9 @@ void *dacp_monitor_thread_code(__attribute__((unused)) void *na) {
debug(1, "DACP server status scanning stopped.");
dacp_server.scan_enable = 0;
}
pthread_mutex_unlock(&dacp_server_information_lock);
pthread_cleanup_pop(1);
// pthread_mutex_unlock(&dacp_server_information_lock);
// debug(1, "DACP Server ID \"%u\" at \"%s:%u\", scan %d.", dacp_server.active_remote_id,
// dacp_server.ip_string, dacp_server.port, scan_index);
@@ -764,7 +779,7 @@ void *dacp_monitor_thread_code(__attribute__((unused)) void *na) {
sleep(config.scan_interval_when_inactive);
}
}
debug(1, "DACP monitor thread exiting.");
debug(1, "DACP monitor thread exiting -- should never happen.");
pthread_exit(NULL);
}
@@ -816,6 +831,14 @@ void dacp_monitor_start() {
pthread_create(&dacp_monitor_thread, NULL, dacp_monitor_thread_code, NULL);
}
void dacp_monitor_stop() {
debug(1, "dacp_monitor_stop");
pthread_cancel(dacp_monitor_thread);
pthread_join(dacp_monitor_thread, NULL);
pthread_mutex_destroy(&dacp_server_information_lock);
pthread_mutex_destroy(&dacp_conversation_lock);
}
uint32_t dacp_tlv_crawl(char **p, int32_t *length) {
char typecode[5];
memcpy(typecode, *p, 4);
dacp.h
+1
View File
@@ -14,6 +14,7 @@ typedef struct dacp_speaker_stuff {
} dacp_spkr_stuff;
void dacp_monitor_start();
void dacp_monitor_stop();
uint32_t dacp_tlv_crawl(
char **p,
dbus-service.c
+14 -2
View File
@@ -19,6 +19,8 @@ ShairportSyncDiagnostics *shairportSyncDiagnosticsSkeleton = NULL;
ShairportSyncRemoteControl *shairportSyncRemoteControlSkeleton = NULL;
ShairportSyncAdvancedRemoteControl *shairportSyncAdvancedRemoteControlSkeleton = NULL;
guint ownerID = 0;
void dbus_metadata_watcher(struct metadata_bundle *argc, __attribute__((unused)) void *userdata) {
char response[100];
const char *th;
@@ -744,6 +746,7 @@ static void on_dbus_name_acquired(GDBusConnection *connection, const gchar *name
static void on_dbus_name_lost_again(__attribute__((unused)) GDBusConnection *connection,
__attribute__((unused)) const gchar *name,
__attribute__((unused)) gpointer user_data) {
debug(1, "Name lost again.");
warn("Could not acquire a Shairport Sync native D-Bus interface \"%s\" on the %s bus.", name,
(config.dbus_service_bus_type == DBT_session) ? "session" : "system");
}
@@ -751,6 +754,7 @@ static void on_dbus_name_lost_again(__attribute__((unused)) GDBusConnection *con
static void on_dbus_name_lost(__attribute__((unused)) GDBusConnection *connection,
__attribute__((unused)) const gchar *name,
__attribute__((unused)) gpointer user_data) {
debug(1, "Name lost.");
// debug(1, "Could not acquire a Shairport Sync native D-Bus interface \"%s\" on the %s bus --
// will try adding the process "
// "number to the end of it.",
@@ -774,7 +778,15 @@ int start_dbus_service() {
dbus_bus_type = G_BUS_TYPE_SESSION;
// debug(1, "Looking for a Shairport Sync native D-Bus interface \"org.gnome.ShairportSync\" on
// the %s bus.",(config.dbus_service_bus_type == DBT_session) ? "session" : "system");
g_bus_own_name(dbus_bus_type, "org.gnome.ShairportSync", G_BUS_NAME_OWNER_FLAGS_NONE, NULL,
on_dbus_name_acquired, on_dbus_name_lost, NULL, NULL);
ownerID = g_bus_own_name(dbus_bus_type, "org.gnome.ShairportSync", G_BUS_NAME_OWNER_FLAGS_NONE,
NULL, on_dbus_name_acquired, on_dbus_name_lost, NULL, NULL);
return 0; // this is just to quieten a compiler warning
}
void stop_dbus_service() {
debug(1, "stopping dbus service");
if (ownerID)
g_bus_unown_name(ownerID);
else
debug(1, "Zero OwnerID for \"org.gnome.ShairportSync\".");
}
dbus-service.h
+1
View File
@@ -6,5 +6,6 @@
ShairportSync *shairportSyncSkeleton;
int start_dbus_service();
void stop_dbus_service();
#endif /* #ifndef DBUS_SERVICE_H */
mdns_avahi.c
+9 -3
View File
@@ -385,21 +385,27 @@ static int avahi_register(char *srvname, int srvport) {
static void avahi_unregister(void) {
debug(1, "avahi: avahi_unregister.");
if (tpoll) {
avahi_threaded_poll_stop(tpoll);
// debug(1, "avahi: stop the threaded poll.");
// avahi_threaded_poll_stop(tpoll);
if (client) {
debug(1, "avahi: free the client.");
avahi_client_free(client);
client = NULL;
} else {
debug(1, "avahi attempting to unregister a NULL client");
}
debug(1, "avahi: free the threaded poll.");
avahi_threaded_poll_free(tpoll);
tpoll = NULL;
} else {
debug(1, "No avahi threaded poll.");
}
if (service_name)
if (service_name) {
debug(1, "avahi: free the service name.");
free(service_name);
else
} else
debug(1, "avahi attempt to free NULL service name");
service_name = NULL;
}
mdns_dns_sd.c
+1 -1
View File
@@ -24,8 +24,8 @@
* OTHER DEALINGS IN THE SOFTWARE.
*/
#include "mdns.h"
#include "common.h"
#include "mdns.h"
#include <arpa/inet.h>
#include <dns_sd.h>
#include <stdlib.h>
mdns_external.c
+1 -1
View File
@@ -24,8 +24,8 @@
* OTHER DEALINGS IN THE SOFTWARE.
*/
#include "mdns.h"
#include "common.h"
#include "mdns.h"
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
metadata_hub.c
+35 -15
View File
@@ -90,12 +90,30 @@ void metadata_hub_release_track_metadata(struct track_metadata_bundle *track_met
}
}
void metadata_hub_release_track_artwork(void) {
// debug(1,"release track artwork");
release_char_string(&metadata_store.cover_art_pathname);
}
void metadata_hub_init(void) {
// debug(1, "Metadata bundle initialisation.");
memset(&metadata_store, 0, sizeof(metadata_store));
track_metadata = NULL;
}
void metadata_hub_stop(void) {
debug(1, "metadata_hub_stop.");
metadata_hub_release_track_artwork();
if (metadata_store.track_metadata) {
metadata_hub_release_track_metadata(metadata_store.track_metadata);
metadata_store.track_metadata = NULL;
}
if (track_metadata) {
metadata_hub_release_track_metadata(track_metadata);
track_metadata = NULL;
}
}
void add_metadata_watcher(metadata_watcher fn, void *userdata) {
int i;
for (i = 0; i < number_of_watchers; i++) {
@@ -108,32 +126,34 @@ void add_metadata_watcher(metadata_watcher fn, void *userdata) {
}
}
void metadata_hub_modify_prolog(void) {
// always run this before changing an entry or a sequence of entries in the metadata_hub
// debug(1, "locking metadata hub for writing");
if (pthread_rwlock_trywrlock(&metadata_hub_re_lock) != 0) {
debug(2, "Metadata_hub write lock is already taken -- must wait.");
pthread_rwlock_wrlock(&metadata_hub_re_lock);
debug(2, "Okay -- acquired the metadata_hub write lock.");
}
}
void metadata_hub_release_track_artwork(void) {
// debug(1,"release track artwork");
release_char_string(&metadata_store.cover_art_pathname);
void metadata_hub_unlock_hub_mutex_cleanup(__attribute__((unused)) void *arg) {
debug(1, "metadata_hub_unlock_hub_mutex_cleanup called.");
pthread_rwlock_wrlock(&metadata_hub_re_lock);
}
void run_metadata_watchers(void) {
int i;
// debug(1, "locking metadata hub for reading");
pthread_rwlock_rdlock(&metadata_hub_re_lock);
pthread_cleanup_push(metadata_hub_unlock_hub_mutex_cleanup, NULL);
for (i = 0; i < number_of_watchers; i++) {
if (metadata_store.watchers[i]) {
metadata_store.watchers[i](&metadata_store, metadata_store.watchers_data[i]);
}
}
// debug(1, "unlocking metadata hub for reading");
pthread_rwlock_unlock(&metadata_hub_re_lock);
// pthread_rwlock_unlock(&metadata_hub_re_lock);
pthread_cleanup_pop(1);
}
void metadata_hub_modify_prolog(void) {
// always run this before changing an entry or a sequence of entries in the metadata_hub
// debug(1, "locking metadata hub for writing");
if (pthread_rwlock_trywrlock(&metadata_hub_re_lock) != 0) {
debug(2, "Metadata_hub write lock is already taken -- must wait.");
pthread_rwlock_unlock(&metadata_hub_re_lock);
debug(2, "Okay -- acquired the metadata_hub write lock.");
}
}
void metadata_hub_modify_epilog(int modified) {
@@ -197,7 +217,7 @@ char *metadata_write_image_file(const char *buf, int len) {
char *path = NULL; // this will be what is returned
uint8_t img_md5[16];
// uint8_t ap_md5[16];
// uint8_t ap_md5[16];
#ifdef HAVE_LIBSSL
MD5_CTX ctx;
metadata_hub.h
+1
View File
@@ -92,6 +92,7 @@ struct metadata_bundle metadata_store;
void add_metadata_watcher(metadata_watcher fn, void *userdata);
void metadata_hub_init(void);
void metadata_hub_stop(void);
void metadata_hub_process_metadata(uint32_t type, uint32_t code, char *data, uint32_t length);
void metadata_hub_reset_track_metadata(void);
void metadata_hub_release_track_artwork(void);
player.c
+53 -49
View File
@@ -315,16 +315,18 @@ static int alac_decode(short *dest, int *destlen, uint8_t *buf, int len, rtsp_co
}
if (outsize > toutsize) {
debug(2, "Output from alac_decode larger (%d bytes, not frames) than expected (%d bytes) -- "
"truncated, but buffer overflow possible! Encrypted = %d.",
debug(2,
"Output from alac_decode larger (%d bytes, not frames) than expected (%d bytes) -- "
"truncated, but buffer overflow possible! Encrypted = %d.",
outsize, toutsize, conn->stream.encrypted);
reply = -1; // output packet is the wrong size
}
*destlen = outsize / conn->input_bytes_per_frame;
if ((outsize % conn->input_bytes_per_frame) != 0)
debug(1, "Number of audio frames (%d) does not correspond exactly to the number of bytes (%d) "
"and the audio frame size (%d).",
debug(1,
"Number of audio frames (%d) does not correspond exactly to the number of bytes (%d) "
"and the audio frame size (%d).",
*destlen, outsize, conn->input_bytes_per_frame);
return reply;
}
@@ -614,8 +616,9 @@ void player_put_packet(seq_t seqno, uint32_t actual_timestamp, int64_t timestamp
if (config.disable_resend_requests == 0) {
if (((int)(resend_interval * pow(j + 1, step_exponent)) + k) >=
seq_diff(conn->ab_read, conn->ab_write, conn->ab_read))
debug(3, "Last-ditch (#%d) resend request for packet %u in range %u to %u. "
"Looking back %d packets.",
debug(3,
"Last-ditch (#%d) resend request for packet %u in range %u to %u. "
"Looking back %d packets.",
j, next, conn->ab_read, conn->ab_write, back_step + k);
debug_mutex_unlock(&conn->ab_mutex, 3);
rtp_request_resend(next, 1, conn);
@@ -807,8 +810,9 @@ static abuf_t *buffer_get_frame(rtsp_conn_info *conn) {
// if (conn->packet_count>500) { //for testing -- about 4 seconds of play first
if ((local_time_now > conn->time_of_last_audio_packet) &&
(local_time_now - conn->time_of_last_audio_packet >= ct << 32)) {
debug(1, "As Yeats almost said, \"Too long a silence / can make a stone of the heart\" "
"from RTSP conversation %d.",
debug(1,
"As Yeats almost said, \"Too long a silence / can make a stone of the heart\" "
"from RTSP conversation %d.",
conn->connection_number);
conn->stop = 1;
pthread_kill(conn->thread, SIGUSR1);
@@ -1649,17 +1653,17 @@ void *player_thread_func(void *arg) {
// if ((input_rate!=config.output_rate) || (input_bit_depth!=output_bit_depth)) {
// debug(1,"Define tbuf of length
// %d.",output_bytes_per_frame*(max_frames_per_packet*output_sample_ratio+max_frame_size_change));
conn->tbuf =
malloc(sizeof(int32_t) * 2 * (conn->max_frames_per_packet * conn->output_sample_ratio +
conn->max_frame_size_change));
conn->tbuf = malloc(
sizeof(int32_t) * 2 *
(conn->max_frames_per_packet * conn->output_sample_ratio + conn->max_frame_size_change));
if (conn->tbuf == NULL)
die("Failed to allocate memory for the transition buffer.");
// initialise this, because soxr stuffing might be chosen later
conn->sbuf =
malloc(sizeof(int32_t) * 2 * (conn->max_frames_per_packet * conn->output_sample_ratio +
conn->max_frame_size_change));
conn->sbuf = malloc(
sizeof(int32_t) * 2 *
(conn->max_frames_per_packet * conn->output_sample_ratio + conn->max_frame_size_change));
if (conn->sbuf == NULL)
die("Failed to allocate memory for the sbuf buffer.");
@@ -1770,8 +1774,9 @@ void *player_thread_func(void *arg) {
} else {
// the player may change the contents of the buffer, so it has to be zeroed each time;
// might as well malloc and freee it locally
memset(silence, 0, conn->output_bytes_per_frame * conn->max_frames_per_packet *
conn->output_sample_ratio);
memset(silence, 0,
conn->output_bytes_per_frame * conn->max_frames_per_packet *
conn->output_sample_ratio);
config.output->play(silence, conn->max_frames_per_packet * conn->output_sample_ratio);
free(silence);
}
@@ -1791,8 +1796,9 @@ void *player_thread_func(void *arg) {
} else {
// the player may change the contents of the buffer, so it has to be zeroed each time;
// might as well malloc and freee it locally
memset(silence, 0, conn->output_bytes_per_frame * conn->max_frames_per_packet *
conn->output_sample_ratio);
memset(silence, 0,
conn->output_bytes_per_frame * conn->max_frames_per_packet *
conn->output_sample_ratio);
config.output->play(silence, conn->max_frames_per_packet * conn->output_sample_ratio);
free(silence);
}
@@ -1939,8 +1945,9 @@ void *player_thread_func(void *arg) {
SUCCESSOR(conn->last_seqno_read); // int32_t from seq_t, i.e. uint16_t, so okay.
if (inframe->sequence_number !=
conn->last_seqno_read) { // seq_t, ei.e. uint16_t and int32_t, so okay
debug(2, "Player: packets out of sequence: expected: %u, got: %u, with ab_read: %u "
"and ab_write: %u.",
debug(2,
"Player: packets out of sequence: expected: %u, got: %u, with ab_read: %u "
"and ab_write: %u.",
conn->last_seqno_read, inframe->sequence_number, conn->ab_read, conn->ab_write);
conn->last_seqno_read = inframe->sequence_number; // reset warning...
}
@@ -2091,7 +2098,7 @@ void *player_thread_func(void *arg) {
#ifdef CONFIG_CONVOLUTION
|| config.convolution
#endif
) {
) {
int32_t *tbuf32 = (int32_t *)conn->tbuf;
float fbuf_l[inbuflength];
float fbuf_r[inbuflength];
@@ -2280,26 +2287,25 @@ void *player_thread_func(void *arg) {
if (at_least_one_frame_seen) {
if ((config.output->delay)) {
if (config.no_sync == 0) {
inform("|%*.1f," /* Sync error in milliseconds */
"%*.1f," /* net correction in ppm */
"%*.1f," /* corrections in ppm */
"%*d," /* total packets */
"%*llu," /* missing packets */
"%*llu," /* late packets */
"%*llu," /* too late packets */
"%*llu," /* resend requests */
"%*lli," /* min DAC queue size */
"%*d," /* min buffer occupancy */
"%*d", /* max buffer occupancy */
10,
1000 * moving_average_sync_error / config.output_rate, 10,
moving_average_correction * 1000000 / (352 * conn->output_sample_ratio),
10, moving_average_insertions_plus_deletions * 1000000 /
(352 * conn->output_sample_ratio),
12, play_number, 7, conn->missing_packets, 7, conn->late_packets, 7,
conn->too_late_packets, 7, conn->resend_requests, 7,
minimum_dac_queue_size, 5, minimum_buffer_occupancy, 5,
maximum_buffer_occupancy);
inform(
"|%*.1f," /* Sync error in milliseconds */
"%*.1f," /* net correction in ppm */
"%*.1f," /* corrections in ppm */
"%*d," /* total packets */
"%*llu," /* missing packets */
"%*llu," /* late packets */
"%*llu," /* too late packets */
"%*llu," /* resend requests */
"%*lli," /* min DAC queue size */
"%*d," /* min buffer occupancy */
"%*d", /* max buffer occupancy */
10, 1000 * moving_average_sync_error / config.output_rate, 10,
moving_average_correction * 1000000 / (352 * conn->output_sample_ratio), 10,
moving_average_insertions_plus_deletions * 1000000 /
(352 * conn->output_sample_ratio),
12, play_number, 7, conn->missing_packets, 7, conn->late_packets, 7,
conn->too_late_packets, 7, conn->resend_requests, 7, minimum_dac_queue_size,
5, minimum_buffer_occupancy, 5, maximum_buffer_occupancy);
} else {
inform("|%*.1f," /* Sync error in milliseconds */
"%*d," /* total packets */
@@ -2310,10 +2316,9 @@ void *player_thread_func(void *arg) {
"%*lli," /* min DAC queue size */
"%*d," /* min buffer occupancy */
"%*d", /* max buffer occupancy */
10,
1000 * moving_average_sync_error / config.output_rate, 12, play_number, 7,
conn->missing_packets, 7, conn->late_packets, 7, conn->too_late_packets, 7,
conn->resend_requests, 7, minimum_dac_queue_size, 5,
10, 1000 * moving_average_sync_error / config.output_rate, 12, play_number,
7, conn->missing_packets, 7, conn->late_packets, 7, conn->too_late_packets,
7, conn->resend_requests, 7, minimum_dac_queue_size, 5,
minimum_buffer_occupancy, 5, maximum_buffer_occupancy);
}
} else {
@@ -2325,10 +2330,9 @@ void *player_thread_func(void *arg) {
"%*llu," /* resend requests */
"%*d," /* min buffer occupancy */
"%*d", /* max buffer occupancy */
10,
1000 * moving_average_sync_error / config.output_rate, 12, play_number, 7,
conn->missing_packets, 7, conn->late_packets, 7, conn->too_late_packets, 7,
conn->resend_requests, 5, minimum_buffer_occupancy, 5,
10, 1000 * moving_average_sync_error / config.output_rate, 12, play_number,
7, conn->missing_packets, 7, conn->late_packets, 7, conn->too_late_packets,
7, conn->resend_requests, 5, minimum_buffer_occupancy, 5,
maximum_buffer_occupancy);
}
} else {
rtp.c
+10 -7
View File
@@ -108,8 +108,9 @@ void *rtp_audio_receiver(void *arg) {
stat_mean += stat_delta / stat_n;
stat_M2 += stat_delta * (time_interval_us - stat_mean);
if (stat_n % 2500 == 0) {
debug(2, "Packet reception interval stats: mean, standard deviation and max for the last "
"2,500 packets in microseconds: %10.1f, %10.1f, %10.1f.",
debug(2,
"Packet reception interval stats: mean, standard deviation and max for the last "
"2,500 packets in microseconds: %10.1f, %10.1f, %10.1f.",
stat_mean, sqrtf(stat_M2 / (stat_n - 1)), longest_packet_time_interval_us);
stat_n = 0;
stat_mean = 0.0;
@@ -325,9 +326,10 @@ void *rtp_control_receiver(void *arg) {
if (la != conn->latency) {
conn->latency = la;
debug(3, "New latency detected: %" PRId64 ", sync latency: %" PRId64
", minimum latency: %" PRId64 ", maximum "
"latency: %" PRId64 ", fixed offset: %" PRId64 ".",
debug(3,
"New latency detected: %" PRId64 ", sync latency: %" PRId64
", minimum latency: %" PRId64 ", maximum "
"latency: %" PRId64 ", fixed offset: %" PRId64 ".",
la, sync_rtp_timestamp - rtp_timestamp_less_latency, conn->minimum_latency,
conn->maximum_latency, config.fixedLatencyOffset);
}
@@ -957,8 +959,9 @@ void rtp_request_resend(seq_t first, uint32_t count, rtsp_conn_info *conn) {
(struct sockaddr *)&conn->rtp_client_control_socket, msgsize) == -1) {
char em[1024];
strerror_r(errno, em, sizeof(em));
debug(1, "Error %d using sendto to an audio socket: \"%s\". Backing off for 1/16th of a "
"second.",
debug(1,
"Error %d using sendto to an audio socket: \"%s\". Backing off for 1/16th of a "
"second.",
errno, em);
conn->rtp_time_of_last_resend_request_error_fp = time_of_sending_fp;
} else {
rtsp.c
+91 -33
View File
@@ -101,8 +101,6 @@ static pthread_mutex_t reference_counter_lock = PTHREAD_MUTEX_INITIALIZER;
// static int please_shutdown = 0;
// static pthread_t playing_thread = 0;
static rtsp_conn_info **conns = NULL;
int RTSP_connection_index = 1;
#ifdef CONFIG_METADATA
@@ -156,6 +154,12 @@ void pc_queue_init(pc_queue *the_queue, char *items, size_t item_size, uint32_t
the_queue->eoq = 0;
}
void pc_queue_delete(pc_queue *the_queue) {
pthread_cond_destroy(&the_queue->pc_queue_item_removed_signal);
pthread_cond_destroy(&the_queue->pc_queue_item_added_signal);
pthread_mutex_destroy(&the_queue->pc_queue_lock);
}
int send_metadata(uint32_t type, uint32_t code, char *data, uint32_t length, rtsp_message *carrier,
int block);
@@ -266,6 +270,21 @@ static void track_thread(rtsp_conn_info *conn) {
}
}
void cancel_all_RTSP_threads(void) {
int i;
for (i = 0; i < nconns; i++) {
debug(1, "Connection %d: cancelling.", conns[i]->connection_number);
pthread_cancel(conns[i]->thread);
}
for (i = 0; i < nconns; i++) {
debug(1, "Connection %d: joining.", conns[i]->connection_number);
pthread_join(conns[i]->thread, NULL);
if (conns[i] == playing_conn)
playing_conn = NULL;
free(conns[i]);
}
}
static void cleanup_threads(void) {
void *retval;
int i;
@@ -719,9 +738,10 @@ static void handle_options(rtsp_conn_info *conn, __attribute__((unused)) rtsp_me
rtsp_message *resp) {
debug(3, "Connection %d: OPTIONS", conn->connection_number);
resp->respcode = 200;
msg_add_header(resp, "Public", "ANNOUNCE, SETUP, RECORD, "
"PAUSE, FLUSH, TEARDOWN, "
"OPTIONS, GET_PARAMETER, SET_PARAMETER");
msg_add_header(resp, "Public",
"ANNOUNCE, SETUP, RECORD, "
"PAUSE, FLUSH, TEARDOWN, "
"OPTIONS, GET_PARAMETER, SET_PARAMETER");
}
static void handle_teardown(rtsp_conn_info *conn, __attribute__((unused)) rtsp_message *req,
@@ -853,10 +873,11 @@ static void handle_setup(rtsp_conn_info *conn, rtsp_message *req, rtsp_message *
}
char resphdr[256] = "";
snprintf(resphdr, sizeof(resphdr), "RTP/AVP/"
"UDP;unicast;interleaved=0-1;mode=record;control_port=%d;"
"timing_port=%d;server_"
"port=%d",
snprintf(resphdr, sizeof(resphdr),
"RTP/AVP/"
"UDP;unicast;interleaved=0-1;mode=record;control_port=%d;"
"timing_port=%d;server_"
"port=%d",
conn->local_control_port, conn->local_timing_port, conn->local_audio_port);
msg_add_header(resp, "Transport", resphdr);
@@ -1051,9 +1072,9 @@ static char *metadata_sockmsg;
#define metadata_queue_size 500
metadata_package metadata_queue_items[metadata_queue_size];
static pthread_t metadata_thread;
pthread_t metadata_thread;
void metadata_create(void) {
void metadata_create_multicast_socket(void) {
if (config.metadata_enabled == 0)
return;
@@ -1075,7 +1096,7 @@ void metadata_create(void) {
if (metadata_sockmsg) {
memset(metadata_sockmsg, 0, config.metadata_sockmsglength);
} else {
die("Could not malloc metadata socket buffer");
die("Could not malloc metadata multicast socket buffer");
}
}
}
@@ -1094,6 +1115,15 @@ void metadata_create(void) {
umask(oldumask);
}
void metadata_delete_multicast_socket(void) {
if (config.metadata_enabled == 0)
return;
shutdown(metadata_sock, SHUT_RDWR); // we want to immediately deallocate the buffer
close(metadata_sock);
if (metadata_sockmsg)
free(metadata_sockmsg);
}
void metadata_open(void) {
if (config.metadata_enabled == 0)
return;
@@ -1111,12 +1141,12 @@ void metadata_open(void) {
free(path);
}
/*
static void metadata_close(void) {
if (fd < 0)
return;
close(fd);
fd = -1;
}
*/
void metadata_process(uint32_t type, uint32_t code, char *data, uint32_t length) {
// debug(2, "Process metadata with type %x, code %x and length %u.", type, code, length);
@@ -1238,11 +1268,32 @@ void metadata_process(uint32_t type, uint32_t code, char *data, uint32_t length)
}
}
void metadata_thread_cleanup_function(__attribute__((unused)) void *arg) {
debug(1, "metadata_thread_cleanup_function called");
metadata_delete_multicast_socket();
metadata_close();
pc_queue_delete(&metadata_queue);
}
void metadata_pack_cleanup_function(void *arg) {
debug(1, "metadata_pack_cleanup_function called");
metadata_package *pack = (metadata_package *)arg;
if (pack->carrier)
msg_free(pack->carrier); // release the message
else if (pack->data)
free(pack->data);
}
void *metadata_thread_function(__attribute__((unused)) void *ignore) {
metadata_create();
// create a pc_queue for passing information to a threaded metadata handler
pc_queue_init(&metadata_queue, (char *)&metadata_queue_items, sizeof(metadata_package),
metadata_queue_size);
metadata_create_multicast_socket();
metadata_package pack;
pthread_cleanup_push(metadata_thread_cleanup_function, NULL);
while (1) {
pc_queue_get_item(&metadata_queue, &pack);
pthread_cleanup_push(metadata_pack_cleanup_function, (void *)&pack);
if (config.metadata_enabled) {
metadata_process(pack.type, pack.code, pack.data, pack.length);
#ifdef HAVE_METADATA_HUB
@@ -1254,23 +1305,24 @@ void *metadata_thread_function(__attribute__((unused)) void *ignore) {
}
#endif
}
if (pack.carrier)
msg_free(pack.carrier); // release the message
else if (pack.data)
free(pack.data);
pthread_cleanup_pop(1);
}
pthread_cleanup_pop(1); // will never happen
pthread_exit(NULL);
}
void metadata_init(void) {
// create a pc_queue for passing information to a threaded metadata handler
pc_queue_init(&metadata_queue, (char *)&metadata_queue_items, sizeof(metadata_package),
metadata_queue_size);
int ret = pthread_create(&metadata_thread, NULL, metadata_thread_function, NULL);
if (ret)
debug(1, "Failed to create metadata thread!");
}
void metadata_stop(void) {
debug(1, "metadata_stop called.");
pthread_cancel(metadata_thread);
pthread_join(metadata_thread, NULL);
}
int send_metadata(uint32_t type, uint32_t code, char *data, uint32_t length, rtsp_message *carrier,
int block) {
@@ -1379,7 +1431,7 @@ static void handle_set_parameter(rtsp_conn_info *conn, rtsp_message *req, rtsp_m
char *ct = msg_get_header(req, "Content-Type");
if (ct) {
// debug(2, "SET_PARAMETER Content-Type:\"%s\".", ct);
// debug(2, "SET_PARAMETER Content-Type:\"%s\".", ct);
#ifdef CONFIG_METADATA
// It seems that the rtptime of the message is used as a kind of an ID that
@@ -1889,8 +1941,9 @@ authenticate:
}
void rtsp_conversation_thread_cleanup_function(void *arg) {
debug(1, "rtsp_conversation_thread_func_cleanup_function called.");
rtsp_conn_info *conn = (rtsp_conn_info *)arg;
debug(1, "Connection %d: rtsp_conversation_thread_func_cleanup_function called.",
conn->connection_number);
if (conn->fd > 0)
close(conn->fd);
if (conn->auth_nonce) {
@@ -1974,8 +2027,9 @@ static void *rtsp_conversation_thread_func(void *pconn) {
if (strcmp(req->method, "OPTIONS") !=
0) // the options message is very common, so don't log it until level 3
debug_level = 2;
debug(debug_level, "RTSP thread %d received an RTSP Packet of type \"%s\":",
conn->connection_number, req->method),
debug(debug_level,
"RTSP thread %d received an RTSP Packet of type \"%s\":", conn->connection_number,
req->method),
debug_print_msg_headers(debug_level, req);
apple_challenge(conn->fd, req, resp);
@@ -2077,6 +2131,14 @@ static const char *format_address(struct sockaddr *fsa) {
}
*/
void rtsp_listen_loop_cleanup_handler(__attribute__((unused)) void *arg) {
debug(1, "rtsp_listen_loop_cleanup_handler called.");
int *sockfd = (int *)arg;
mdns_unregister();
if (sockfd)
free(sockfd);
}
void rtsp_listen_loop(void) {
struct addrinfo hints, *info, *p;
char portstr[6];
@@ -2175,6 +2237,7 @@ void rtsp_listen_loop(void) {
int acceptfd;
struct timeval tv;
pthread_cleanup_push(rtsp_listen_loop_cleanup_handler, (void *)sockfd);
do {
pthread_testcancel();
tv.tv_sec = 60;
@@ -2272,11 +2335,6 @@ void rtsp_listen_loop(void) {
}
} while (1);
mdns_unregister();
if (sockfd)
free(sockfd);
// perror("select");
// die("fell out of the RTSP select loop");
pthread_cleanup_pop(0);
debug(1, "Oops -- fell out of the RTSP select loop");
}
rtsp.h
+5 -1
View File
@@ -4,14 +4,18 @@
#include "player.h"
rtsp_conn_info *playing_conn;
rtsp_conn_info **conns;
void rtsp_listen_loop(void);
// void rtsp_shutdown_stream(void);
void rtsp_request_shutdown_stream(void);
// initialise the metadata stuff
void cancel_all_RTSP_threads(void);
// initialise and completely delete the metadata stuff
void metadata_init(void);
void metadata_stop(void);
// sends metadata out to the metadata pipe, if enabled.
// It is sent with the type 'ssnc' the given code, data and length
shairport.c
+54 -27
View File
@@ -36,7 +36,6 @@
#include <stdio.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/wait.h>
@@ -117,22 +116,11 @@ static int shutting_down = 0;
char configuration_file_path[4096 + 1];
char actual_configuration_file_path[4096 + 1];
void shairport_shutdown() {
if (shutting_down)
return;
shutting_down = 1;
mdns_unregister();
rtsp_request_shutdown_stream();
if (config.output)
config.output->deinit();
}
static void sig_ignore(__attribute__((unused)) int foo, __attribute__((unused)) siginfo_t *bar,
__attribute__((unused)) void *baz) {}
static void sig_shutdown(__attribute__((unused)) int foo, __attribute__((unused)) siginfo_t *bar,
__attribute__((unused)) void *baz) {
debug(1, "shutdown requested...");
shairport_shutdown();
// daemon_log(LOG_NOTICE, "exit...");
daemon_retval_send(255);
daemon_pid_file_remove();
@@ -392,6 +380,7 @@ int parse_options(int argc, char **argv) {
debug(2, "Looking for configuration file at full path \"%s\"", config_file_real_path);
/* Read the file. If there is an error, report it and exit. */
if (config_read_file(&config_file_stuff, config_file_real_path)) {
free(config_file_real_path);
// make config.cfg point to it
config.cfg = &config_file_stuff;
/* Get the Service Name. */
@@ -938,7 +927,6 @@ int parse_options(int argc, char **argv) {
config_set_lookup_bool(config.cfg, "mqtt.publish_cover", &config.mqtt_publish_cover);
config_set_lookup_bool(config.cfg, "mqtt.enable_remote", &config.mqtt_enable_remote);
#endif
free(config_file_real_path);
}
// now, do the command line options again, but this time do them fully -- it's a unix convention
@@ -1063,13 +1051,14 @@ int parse_options(int argc, char **argv) {
}
#if defined(HAVE_DBUS) || defined(HAVE_MPRIS)
GMainLoop *loop;
GMainLoop *g_main_loop;
pthread_t dbus_thread;
void *dbus_thread_func(__attribute__((unused)) void *arg) {
loop = g_main_loop_new(NULL, FALSE);
g_main_loop_run(loop);
return NULL;
g_main_loop = g_main_loop_new(NULL, FALSE);
g_main_loop_run(g_main_loop);
debug(1, "g_main_loop thread exit");
pthread_exit(NULL);
}
#endif
@@ -1132,6 +1121,17 @@ const char *pid_file_proc(void) {
void exit_function() {
debug(1, "exit function called...");
cancel_all_RTSP_threads();
if (conns)
free(conns); // make sure the connections have been deleted first
if (config.service_name)
free(config.service_name);
if (config.regtype)
free(config.regtype);
if (config.computed_piddir)
free(config.computed_piddir);
if (ranarray)
free((void *)ranarray);
if (config.cfg)
config_destroy(config.cfg);
if (config.appName)
@@ -1142,7 +1142,38 @@ void exit_function() {
void main_cleanup_handler(__attribute__((unused)) void *arg) {
debug(1, "main cleanup handler called.");
shairport_shutdown();
#ifdef HAVE_MQTT
if (config.mqtt_enabled) {
// terminate_mqtt();
}
#endif
#if defined(HAVE_DBUS) || defined(HAVE_MPRIS)
#ifdef HAVE_MPRIS
// stop_mpris_service();
#endif
#ifdef HAVE_DBUS
stop_dbus_service();
#endif
debug(1, "Stopping DBUS Loop Thread");
g_main_loop_quit(g_main_loop);
pthread_join(dbus_thread, NULL);
#endif
#ifdef HAVE_DACP_CLIENT
debug(1, "Stopping DACP Monitor");
dacp_monitor_stop();
#endif
#ifdef HAVE_METADATA_HUB
debug(1, "Stopping metadata hub");
metadata_hub_stop();
#endif
#ifdef CONFIG_METADATA
metadata_stop(); // close down the metadata pipe
#endif
daemon_log(LOG_NOTICE, "Unexpected exit...");
daemon_retval_send(0);
daemon_pid_file_remove();
@@ -1151,6 +1182,7 @@ void main_cleanup_handler(__attribute__((unused)) void *arg) {
}
int main(int argc, char **argv) {
conns = NULL; // no connections active
memset((void *)&main_thread_id, 0, sizeof(main_thread_id));
fp_time_at_startup = get_absolute_time_in_fp();
fp_time_at_last_debug_message = fp_time_at_startup;
@@ -1408,13 +1440,8 @@ int main(int argc, char **argv) {
/* end libdaemon stuff */
}
// int old_cancel_state = 0;
// pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
// pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &old_cancel_state);
main_thread_id = pthread_self();
if (main_thread_id)
debug(1, "Main thread ID set up.");
else
if (!main_thread_id)
debug(1, "Main thread is set up to be NULL!");
signal_setup();
@@ -1552,8 +1579,9 @@ int main(int argc, char **argv) {
debug(1, "loudness is %d.", config.loudness);
debug(1, "loudness reference level is %f", config.loudness_reference_volume_db);
debug(1, "disable resend requests is %s.", config.disable_resend_requests ? "on" : "off");
debug(1, "diagnostic_drop_packet_fraction is %f. A value of 0.0 means no packets will be dropped "
"deliberately.",
debug(1,
"diagnostic_drop_packet_fraction is %f. A value of 0.0 means no packets will be dropped "
"deliberately.",
config.diagnostic_drop_packet_fraction);
uint8_t ap_md5[16];
@@ -1615,7 +1643,6 @@ int main(int argc, char **argv) {
rtsp_listen_loop();
// should not reach this...
// shairport_shutdown();
// daemon_log(LOG_NOTICE, "Unexpected exit...");
// daemon_retval_send(0);
// daemon_pid_file_remove();
tinysvcmdns.c
+2 -1
View File
@@ -1713,7 +1713,8 @@ void mdnsd_stop(struct mdnsd *s) {
assert(s != NULL);
struct timeval tv = {
.tv_sec = 0, .tv_usec = 500 * 1000,
.tv_sec = 0,
.tv_usec = 500 * 1000,
};
s->stop_flag = 1;