Merge pull request #210 from mikebrady/better-stopping

Better stopping. This improves Shairport Sync's behaviour when it is asked to stop a play session and immediately asked to start another.
This commit is contained in:
Mike Brady
2016-01-24 18:58:48 +00:00
5 changed files with 145 additions and 92 deletions
+48 -22
View File
@@ -81,7 +81,7 @@ static int sampling_rate, frame_size;
static aes_context dctx;
#endif
static pthread_t player_thread;
//static pthread_t player_thread = NULL;
static int please_stop;
static int encrypted; // Normally the audio is encrypted, but it may not be
@@ -308,7 +308,7 @@ void player_put_packet(seq_t seqno, uint32_t timestamp, uint8_t *data, int len)
if ((flush_rtp_timestamp != 0) &&
((timestamp == flush_rtp_timestamp) || seq32_order(timestamp, flush_rtp_timestamp))) {
debug(2, "Dropping flushed packet in player_put_packet, seqno %u, timestamp %u, flushing to "
debug(3, "Dropping flushed packet in player_put_packet, seqno %u, timestamp %u, flushing to "
"timestamp: %u.",
seqno, timestamp, flush_rtp_timestamp);
} else {
@@ -332,7 +332,7 @@ void player_put_packet(seq_t seqno, uint32_t timestamp, uint8_t *data, int len)
// if (ORDINATE(seqno)>(BUFFER_FRAMES*7)/8)
// debug(1,"An interval of %u frames has opened, with ab_read: %u, ab_write: %u and seqno:
// %u.",seq_diff(ab_read,seqno),ab_read,ab_write,seqno);
int32_t gap = seq_diff(ab_write, PREDECESSOR(seqno)) + 1;
int32_t gap = seq_diff(ab_write, seqno);
if (gap <= 0)
debug(1, "Unexpected gap size: %d.", gap);
int i;
@@ -344,8 +344,8 @@ void player_put_packet(seq_t seqno, uint32_t timestamp, uint8_t *data, int len)
}
// debug(1,"N %d s %u.",seq_diff(ab_write,PREDECESSOR(seqno))+1,ab_write);
abuf = audio_buffer + BUFIDX(seqno);
rtp_request_resend(ab_write, gap);
resend_requests++;
// rtp_request_resend(ab_write, gap);
// resend_requests++;
ab_write = SUCCESSOR(seqno);
} else if (seq_order(ab_read, seqno)) { // late but not yet played
late_packets++;
@@ -893,6 +893,13 @@ typedef struct stats { // statistics for running averages
} stats_t;
static void *player_thread_func(void *arg) {
int threads_stop = 0;
// create and start the timing, control and audio receiver threads
pthread_t rtp_audio_thread, rtp_control_thread, rtp_timing_thread;
pthread_create(&rtp_audio_thread, NULL, &rtp_audio_receiver, (void *)&threads_stop);
pthread_create(&rtp_control_thread, NULL, &rtp_control_receiver, (void *)&threads_stop);
pthread_create(&rtp_timing_thread, NULL, &rtp_timing_receiver, (void *)&threads_stop);
session_corrections = 0;
play_segment_reference_frame = 0; // zero signals that we are not in a play segment
// check that there are enough buffers to accommodate the desired latency and the latency offset
@@ -1219,8 +1226,22 @@ static void *player_thread_func(void *arg) {
}
}
}
if (config.output->stop)
config.output->stop();
free(outbuf);
free(silence);
debug(1,"Shut down audio, control and timing threads");
// usleep(1000000);
pthread_kill(rtp_audio_thread, SIGUSR1);
pthread_kill(rtp_control_thread, SIGUSR1);
pthread_kill(rtp_timing_thread, SIGUSR1);
pthread_join(rtp_timing_thread, NULL);
debug(1,"timing thread joined");
pthread_join(rtp_audio_thread, NULL);
debug(1,"audio thread joined");
pthread_join(rtp_control_thread, NULL);
debug(1,"control thread joined");
debug(1,"Player thread exit");
return 0;
}
@@ -1377,7 +1398,7 @@ void player_volume(double airplay_volume) {
}
void player_flush(uint32_t timestamp) {
// debug(1,"Flush requested up to %u. It seems as if 0 is special.",timestamp);
debug(3,"Flush requested up to %u. It seems as if 0 is special.",timestamp);
pthread_mutex_lock(&flush_mutex);
flush_requested = 1;
// if (timestamp!=0)
@@ -1389,7 +1410,9 @@ void player_flush(uint32_t timestamp) {
#endif
}
int player_play(stream_cfg *stream) {
int player_play(stream_cfg *stream, pthread_t *player_thread) {
//if (*player_thread!=NULL)
// die("Trying to create a second player thread for this RTSP session");
packet_count = 0;
encrypted = stream->encrypted;
if (config.buffer_start_fill > BUFFER_FRAMES)
@@ -1434,23 +1457,26 @@ int player_play(stream_cfg *stream) {
rc = pthread_attr_setstacksize(&tattr, size);
if (rc)
debug(1, "Error setting stack size for player_thread: %s", strerror(errno));
pthread_create(&player_thread, &tattr, player_thread_func, NULL);
pthread_create(player_thread, &tattr, player_thread_func, NULL);
pthread_attr_destroy(&tattr);
return 0;
}
void player_stop(void) {
please_stop = 1;
pthread_cond_signal(&flowcontrol); // tell it to give up
pthread_join(player_thread, NULL);
#ifdef CONFIG_METADATA
send_ssnc_metadata('pend', NULL, 0, 1);
#endif
config.output->stop();
command_stop();
free_buffer();
free_decoder();
int rc = pthread_cond_destroy(&flowcontrol);
if (rc)
debug(1, "Error destroying condition variable.");
void player_stop(pthread_t *player_thread) {
//if (*thread==NULL)
// debug(1,"Trying to stop a non-existent player thread");
// else {
please_stop = 1;
pthread_cond_signal(&flowcontrol); // tell it to give up
pthread_join(*player_thread, NULL);
#ifdef CONFIG_METADATA
send_ssnc_metadata('pend', NULL, 0, 1);
#endif
command_stop();
free_buffer();
free_decoder();
int rc = pthread_cond_destroy(&flowcontrol);
if (rc)
debug(1, "Error destroying condition variable.");
// }
}
+2 -2
View File
@@ -14,8 +14,8 @@ typedef uint16_t seq_t;
// wrapped number between two seq_t.
int32_t seq_diff(seq_t a, seq_t b);
int player_play(stream_cfg *cfg);
void player_stop(void);
int player_play(stream_cfg *cfg, pthread_t *thread);
void player_stop(pthread_t *thread);
void player_volume(double f);
void player_flush(uint32_t timestamp);
+43 -41
View File
@@ -61,7 +61,6 @@ typedef struct time_ping_record {
// only one RTP session can be active at a time.
static int running = 0;
static int please_shutdown;
static char client_ip_string[INET6_ADDRSTRLEN]; // the ip string pointing to the client
static short client_ip_family; // AF_INET / AF_INET6
@@ -72,7 +71,7 @@ static SOCKADDR rtp_client_timing_socket; // a socket pointing to the timing po
static int audio_socket; // our local [server] audio socket
static int control_socket; // our local [server] control socket
static int timing_socket; // local timing socket
static pthread_t rtp_audio_thread, rtp_control_thread, rtp_timing_thread;
//static pthread_t rtp_audio_thread, rtp_control_thread, rtp_timing_thread;
static uint32_t reference_timestamp;
static uint64_t reference_timestamp_time;
@@ -96,16 +95,16 @@ static pthread_mutex_t reference_time_mutex = PTHREAD_MUTEX_INITIALIZER;
uint64_t static local_to_remote_time_difference; // used to switch between local and remote clocks
static void *rtp_audio_receiver(void *arg) {
void *rtp_audio_receiver(void *arg) {
// we inherit the signal mask (SIGUSR1)
int *stop = arg; // when set to 1, we should stop
int32_t last_seqno = -1;
uint8_t packet[2048], *pktp;
ssize_t nread;
while (1) {
if (please_shutdown)
break;
while (*stop==0) {
nread = recv(audio_socket, packet, sizeof(packet), 0);
if (nread < 0)
break;
@@ -158,17 +157,18 @@ static void *rtp_audio_receiver(void *arg) {
return NULL;
}
static void *rtp_control_receiver(void *arg) {
void *rtp_control_receiver(void *arg) {
// we inherit the signal mask (SIGUSR1)
int *stop = arg; // when set to 1, we should stop
reference_timestamp = 0; // nothing valid received yet
uint8_t packet[2048], *pktp;
struct timespec tn;
uint64_t remote_time_of_sync, local_time_now, remote_time_now;
uint32_t sync_rtp_timestamp, rtp_timestamp_less_latency;
ssize_t nread;
while (1) {
if (please_shutdown)
break;
while (*stop==0) {
nread = recv(control_socket, packet, sizeof(packet), 0);
local_time_now = get_absolute_time_in_fp();
// clock_gettime(CLOCK_MONOTONIC,&tn);
@@ -259,7 +259,8 @@ static void *rtp_control_receiver(void *arg) {
return NULL;
}
static void *rtp_timing_sender(void *arg) {
void *rtp_timing_sender(void *arg) {
int *stop = arg; // the parameter points to this request to stop thing
struct timing_request {
char leader;
char type;
@@ -280,9 +281,8 @@ static void *rtp_timing_sender(void *arg) {
time_ping_count = 0;
// we inherit the signal mask (SIGUSR1)
while (1) {
if (please_shutdown)
break;
while (*stop==0) {
// debug(1,"Send a timing request");
if (!running)
die("rtp_timing_sender called without active stream!");
@@ -314,12 +314,16 @@ static void *rtp_timing_sender(void *arg) {
return NULL;
}
static void *rtp_timing_receiver(void *arg) {
void *rtp_timing_receiver(void *arg) {
// we inherit the signal mask (SIGUSR1)
int *stop = arg; // when set to 1, we should stop
uint8_t packet[2048], *pktp;
ssize_t nread;
int request_stop = 0;
pthread_t timer_requester;
pthread_create(&timer_requester, NULL, &rtp_timing_sender, NULL);
pthread_create(&timer_requester, NULL, &rtp_timing_sender, (void *)&request_stop);
// struct timespec att;
uint64_t distant_receive_time, distant_transmit_time, arrival_time, return_time, transit_time,
processing_time;
@@ -331,9 +335,7 @@ static void *rtp_timing_receiver(void *arg) {
uint64_t first_local_to_remote_time_difference = 0;
uint64_t first_local_to_remote_time_difference_time;
uint64_t l2rtd = 0;
while (1) {
if (please_shutdown)
break;
while (*stop==0) {
nread = recv(timing_socket, packet, sizeof(packet), 0);
arrival_time = get_absolute_time_in_fp();
// clock_gettime(CLOCK_MONOTONIC,&att);
@@ -497,7 +499,8 @@ static void *rtp_timing_receiver(void *arg) {
}
}
debug(1, "Timing RTP thread interrupted. terminating.");
debug(1, "Timing thread interrupted. terminating.");
request_stop = 1;
void *retval;
pthread_kill(timer_requester, SIGUSR1);
pthread_join(timer_requester, &retval);
@@ -655,11 +658,10 @@ void rtp_setup(SOCKADDR *remote, int cport, int tport, uint32_t active_remote, i
debug(2, "listening for audio, control and timing on ports %d, %d, %d.", *lsport, *lcport,
*ltport);
please_shutdown = 0;
reference_timestamp = 0;
pthread_create(&rtp_audio_thread, NULL, &rtp_audio_receiver, NULL);
pthread_create(&rtp_control_thread, NULL, &rtp_control_receiver, NULL);
pthread_create(&rtp_timing_thread, NULL, &rtp_timing_receiver, NULL);
//pthread_create(&rtp_audio_thread, NULL, &rtp_audio_receiver, NULL);
//pthread_create(&rtp_control_thread, NULL, &rtp_control_receiver, NULL);
//pthread_create(&rtp_timing_thread, NULL, &rtp_timing_receiver, NULL);
running = 1;
request_sent = 0;
@@ -682,27 +684,27 @@ void clear_reference_timestamp(void) {
void rtp_shutdown(void) {
if (!running)
die("rtp_shutdown called without active stream!");
debug(1,"rtp_shutdown called without active stream!");
debug(2, "shutting down RTP thread");
please_shutdown = 1;
void *retval;
reference_timestamp = 0;
pthread_kill(rtp_audio_thread, SIGUSR1);
pthread_join(rtp_audio_thread, &retval);
pthread_kill(rtp_control_thread, SIGUSR1);
pthread_join(rtp_control_thread, &retval);
pthread_kill(rtp_timing_thread, SIGUSR1);
pthread_join(rtp_timing_thread, &retval);
clear_reference_timestamp();
// debug(1,"Shut down audio, control and timing threads");
// usleep(3000000); // hack
// pthread_kill(rtp_audio_thread, SIGUSR1);
// pthread_kill(rtp_control_thread, SIGUSR1);
// pthread_kill(rtp_timing_thread, SIGUSR1);
// pthread_join(rtp_audio_thread, &retval);
// pthread_join(rtp_control_thread, &retval);
// pthread_join(rtp_timing_thread, &retval);
running = 0;
}
void rtp_request_resend(seq_t first, uint32_t count) {
if (running) {
if (!request_sent) {
debug(2, "requesting resend on %d packets starting at %u.", count, first);
request_sent = 1;
}
//if (!request_sent) {
debug(2, "requesting resend of %d packets starting at %u.", count, first);
// request_sent = 1;
//}
char req[8]; // *not* a standard RTCP NACK
req[0] = 0x80;
@@ -721,10 +723,10 @@ void rtp_request_resend(seq_t first, uint32_t count) {
perror("Error sendto-ing to audio socket");
}
} else {
if (!request_sent) {
//if (!request_sent) {
debug(2, "rtp_request_resend called without active stream!");
request_sent = 1;
}
// request_sent = 1;
//}
}
}
+4
View File
@@ -5,6 +5,10 @@
#include "player.h"
void *rtp_audio_receiver(void *arg);
void *rtp_control_receiver(void *arg);
void *rtp_timing_receiver(void *arg);
void rtp_setup(SOCKADDR *remote, int controlport, int timingport, uint32_t active_remote,
int *local_server_port, int *local_control_port, int *local_timing_port);
void rtp_shutdown(void);
+48 -27
View File
@@ -83,15 +83,17 @@ static pthread_mutex_t reference_counter_lock = PTHREAD_MUTEX_INITIALIZER;
// only one thread is allowed to use the player at once.
// it monitors the request variable (at least when interrupted)
static pthread_mutex_t playing_mutex = PTHREAD_MUTEX_INITIALIZER;
static int please_shutdown = 0;
//static int please_shutdown = 0;
static pthread_t playing_thread = 0;
typedef struct {
int fd;
stream_cfg stream;
SOCKADDR remote;
int stop;
int running;
pthread_t thread;
pthread_t player_thread;
} rtsp_conn_info;
#ifdef CONFIG_METADATA
@@ -133,6 +135,8 @@ typedef struct {
rtsp_message *carrier;
} metadata_package;
void ask_other_rtsp_conversation_threads_to_stop(pthread_t except_this_thread);
void pc_queue_init(pc_queue *the_queue, char *items, size_t item_size, uint32_t number_of_items) {
the_queue->item_size = item_size;
the_queue->items = items;
@@ -239,8 +243,8 @@ static inline int rtsp_playing(void) {
}
void rtsp_request_shutdown_stream(void) {
please_shutdown = 1;
pthread_kill(playing_thread, SIGUSR1);
debug(1,"Request to shut down all rtsp conversation threads");
ask_other_rtsp_conversation_threads_to_stop(0); // i.e. ask all playing threads to stop
}
static void rtsp_take_player(void) {
@@ -248,10 +252,8 @@ static void rtsp_take_player(void) {
return;
if (pthread_mutex_trylock(&playing_mutex)) {
debug(1, "shutting down playing thread.");
// XXX minor race condition between please_shutdown and signal delivery
please_shutdown = 1;
pthread_kill(playing_thread, SIGUSR1);
debug(1, "Request to all other playing threads to stop.");
ask_other_rtsp_conversation_threads_to_stop(pthread_self()); // all threads apart from self
pthread_mutex_lock(&playing_mutex);
}
playing_thread = pthread_self(); // make us the currently-playing thread (why?)
@@ -289,6 +291,21 @@ static void cleanup_threads(void) {
}
}
// ask all rtsp_conversation threads to stop -- there should be at most one, but ya never know.
void ask_other_rtsp_conversation_threads_to_stop(pthread_t except_this_thread) {
int i;
debug(2, "asking playing threads to stop");
for (i = 0; i < nconns;i++) {
if (((except_this_thread==0) || (pthread_equal(conns[i]->thread,except_this_thread)==0)) && (conns[i]->running != 0)) {
conns[i]->stop = 1;
pthread_kill(conns[i]->thread,SIGUSR1);
}
}
}
// park a null at the line ending, and return the next line pointer
// accept \r, \n, or \r\n
static char *nextline(char *in, int inbuf) {
@@ -426,7 +443,7 @@ static int msg_handle_line(rtsp_message **pmsg, char *line) {
*p = 0;
p += 2;
msg_add_header(msg, line, p);
debug(2, " %s: %s.", line, p);
debug(3, " %s: %s.", line, p);
return -1;
} else {
char *cl = msg_get_header(msg, "Content-Length");
@@ -442,7 +459,7 @@ fail:
return 0;
}
static enum rtsp_read_request_response rtsp_read_request(int fd, rtsp_message **the_packet) {
static enum rtsp_read_request_response rtsp_read_request(rtsp_conn_info *conn, rtsp_message **the_packet) {
enum rtsp_read_request_response reply = rtsp_read_request_response_ok;
ssize_t buflen = 512;
char *buf = malloc(buflen + 1);
@@ -454,12 +471,12 @@ static enum rtsp_read_request_response rtsp_read_request(int fd, rtsp_message **
int msg_size = -1;
while (msg_size < 0) {
if (please_shutdown) {
if (conn->stop!=0) {
debug(1, "RTSP shutdown requested.");
reply = rtsp_read_request_response_shutdown_requested;
goto shutdown;
}
nread = read(fd, buf + inbuf, buflen - inbuf);
nread = read(conn->fd, buf + inbuf, buflen - inbuf);
if (!nread) {
debug(1, "RTSP connection closed.");
reply = rtsp_read_request_response_shutdown_requested;
@@ -524,7 +541,7 @@ static enum rtsp_read_request_response rtsp_read_request(int fd, rtsp_message **
ssize_t read_chunk = msg_size - inbuf;
if (read_chunk > max_read_chunk)
read_chunk = max_read_chunk;
nread = read(fd, buf + inbuf, read_chunk);
nread = read(conn->fd, buf + inbuf, read_chunk);
if (!nread) {
reply = rtsp_read_request_response_error;
goto shutdown;
@@ -568,7 +585,7 @@ static void msg_write_response(int fd, rtsp_message *resp) {
p += n;
for (i = 0; i < resp->nheaders; i++) {
debug(2, " %s: %s.", resp->name[i], resp->value[i]);
// debug(3, " %s: %s.", resp->name[i], resp->value[i]);
n = snprintf(p, pktfree, "%s: %s\r\n", resp->name[i], resp->value[i]);
pktfree -= n;
p += n;
@@ -623,15 +640,15 @@ static void handle_options(rtsp_conn_info *conn, rtsp_message *req, rtsp_message
static void handle_teardown(rtsp_conn_info *conn, rtsp_message *req, rtsp_message *resp) {
if (!rtsp_playing())
return;
debug(1,"This RTSP conversation thread doesn't think it's playing, but it's sending a response to teardown anyway");
resp->respcode = 200;
msg_add_header(resp, "Connection", "close");
please_shutdown = 1;
conn->stop = 1;
}
static void handle_flush(rtsp_conn_info *conn, rtsp_message *req, rtsp_message *resp) {
if (!rtsp_playing())
return;
debug(1,"This RTSP conversation thread doesn't think it's playing, but it's sending a response to flush anyway");
char *p;
uint32_t rtptime = 0;
char *hdr = msg_get_header(req, "RTP-Info");
@@ -764,7 +781,7 @@ static void handle_setup(rtsp_conn_info *conn, rtsp_message *req, rtsp_message *
strcat(hdr, q); // should unsplice the timing port entry
}
player_play(&conn->stream);
player_play(&conn->stream,&conn->player_thread); // the thread better be 0
char *resphdr = alloca(200);
*resphdr = 0;
@@ -1517,7 +1534,7 @@ static void *rtsp_conversation_thread_func(void *pconn) {
enum rtsp_read_request_response reply;
do {
reply = rtsp_read_request(conn->fd, &req);
reply = rtsp_read_request(conn, &req);
if (reply == rtsp_read_request_response_ok) {
resp = msg_init();
resp->respcode = 400;
@@ -1535,12 +1552,12 @@ static void *rtsp_conversation_thread_func(void *pconn) {
int method_selected = 0;
for (mh = method_handlers; mh->method; mh++) {
if (!strcmp(mh->method, req->method)) {
//debug(1,"RTSP Packet received of type \"%s\":",mh->method),
//msg_print_debug_headers(req);
// debug(1,"RTSP Packet received of type \"%s\":",mh->method),
// msg_print_debug_headers(req);
method_selected = 1;
mh->handler(conn, req, resp);
//debug(1,"RTSP Response:");
//msg_print_debug_headers(resp);
// debug(1,"RTSP Response:");
// msg_print_debug_headers(resp);
break;
}
}
@@ -1557,20 +1574,22 @@ static void *rtsp_conversation_thread_func(void *pconn) {
}
} while (reply != rtsp_read_request_response_shutdown_requested);
debug(1, "closing RTSP connection.");
debug(1, "Now closing RTSP connection.");
if (conn->fd > 0)
close(conn->fd);
if (rtsp_playing()) {
player_stop(&conn->player_thread); // might be less noisy doing this first
rtp_shutdown();
player_stop();
pthread_mutex_unlock(&play_lock);
please_shutdown = 0;
pthread_mutex_unlock(&playing_mutex);
} else {
debug(1,"This RTSP conversation thread doesn't think it's playing for a close RTSP connection.");
}
if (auth_nonce)
free(auth_nonce);
conn->running = 0;
debug(2, "terminating RTSP thread.");
debug(2, "Now terminating RTSP conversation thread.");
// please_shutdown = 0;
return NULL;
}
@@ -1638,6 +1657,7 @@ void rtsp_listen_loop(void) {
// one of the address families will fail on some systems that
// report its availability. do not complain.
if (ret) {
debug(1, "Failed to bind to address %s.", format_address(p->ai_addr));
continue;
@@ -1699,7 +1719,7 @@ void rtsp_listen_loop(void) {
memset(conn, 0, sizeof(rtsp_conn_info));
socklen_t slen = sizeof(conn->remote);
debug(1, "new RTSP connection.");
debug(1, "New RTSP connection on port %d",config.port);
conn->fd = accept(acceptfd, (struct sockaddr *)&conn->remote, &slen);
if (conn->fd < 0) {
perror("failed to accept connection");
@@ -1711,6 +1731,7 @@ void rtsp_listen_loop(void) {
die("Failed to create RTSP receiver thread!");
conn->thread = rtsp_conversation_thread;
conn -> stop = 0;
conn->running = 1;
track_thread(conn);
}