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

Send SIGUSR2 to shairport sync and it will send a pause request to the source. Also keep checking buffers.

Browse Source
This commit is contained in:
Mike Brady
2014-09-22 18:31:30 +01:00
parent ed127c7117
commit f2a54dd00c
5 changed files with 155 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
player.c
+64 -9
View File
@@ -135,8 +135,10 @@ static uint64_t missing_packets,late_packets,too_late_packets,resend_requests;
static void ab_resync(void) {
int i;
for (i=0; i<BUFFER_FRAMES; i++)
for (i=0; i<BUFFER_FRAMES; i++) {
audio_buffer[i].ready = 0;
audio_buffer[i].sequence_number = 0;
}
ab_synced = 0;
last_seqno_read = -1;
ab_buffering = 1;
@@ -162,7 +164,6 @@ static inline seq_t PREDECESSOR(seq_t x) {
}
// anything with ORDINATE in it must be proctected by the ab_mutex
static inline int32_t ORDINATE(seq_t x) {
int32_t p = x & 0xffff;
int32_t q = ab_read & 0x0ffff;
@@ -198,6 +199,28 @@ static inline seq_t seq_sum(seq_t a, seq_t b) {
return r;
}
// now for 32-bit wrapping in timestamps
// this returns true if the second arg is strictly after the first
// on the assumption that the gap between them is never greater than (2^31)-1
// Represent a and b in 64 bits
static inline int seq32_order(uint32_t a, uint32_t b) {
if (a==b)
return 0;
int64_t A = a & 0xffffffff;
int64_t B = b & 0xffffffff;
int64_t C = B-A;
// if bit 31 is set, it means either b is before (i.e. less than) a or
// b is (2^31)-1 ahead of b.
// If we assume the gap between b and a should never reach 2 billion, then
// bit 31 == 0 means b is strictly after a
return (C & 0x80000000) == 0;
}
static void alac_decode(short *dest, uint8_t *buf, int len) {
unsigned char packet[MAX_PACKET];
unsigned char packetp[MAX_PACKET];
@@ -285,8 +308,8 @@ void player_put_packet(seq_t seqno,uint32_t timestamp, uint8_t *data, int len) {
abuf = audio_buffer + BUFIDX(seqno);
ab_write = SUCCESSOR(seqno);
} else if (seq_order(ab_write, seqno)) { // newer than expected
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);
//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;
if (gap<=0)
debug(1,"Unexpected gap size: %d.",gap);
@@ -390,9 +413,41 @@ static abuf_t *buffer_get_frame(void) {
flush_requested=0;
}
pthread_mutex_unlock(&flush_mutex);
uint32_t flush_limit = 0;
if (ab_synced) {
do {
curframe = audio_buffer + BUFIDX(ab_read);
if (curframe->ready) {
if (curframe->sequence_number!=ab_read) {
// some kind of sync problem has occurred.
if (BUFIDX(curframe->sequence_number)==BUFIDX(ab_read)) {
// it looks like some kind of aliasing has happened
if (seq_order(ab_read,curframe->sequence_number)) {
ab_read=curframe->sequence_number;
debug(1,"Aliasing of buffer index -- reset.");
}
} else {
debug(1,"Inconsistent sequence numbers detected");
}
}
if ((flush_rtp_timestamp!=0x7fffffff) && ((curframe->timestamp==flush_rtp_timestamp) || seq32_order(curframe->timestamp,flush_rtp_timestamp))) {
debug(1,"Dropping flushed packet seqno %u, timestamp %u",curframe->sequence_number,curframe->timestamp);
curframe->ready=0;
flush_limit++;
ab_read==SUCCESSOR(ab_read);
}
if ((flush_rtp_timestamp!=0x7fffffff) && (!seq32_order(curframe->timestamp,flush_rtp_timestamp))) // if we have gone past the flush boundary time
flush_rtp_timestamp=0x7fffffff;
}
} while ((flush_rtp_timestamp!=0x7fffffff) && (flush_limit<=8820) && (curframe->ready!=0));
if (flush_limit==8820) {
debug(1,"Flush hit the 8820 frame limit!");
flush_limit=0;
}
dac_delay = 0;
curframe = audio_buffer + BUFIDX(ab_read);
if (config.output->delay) {
@@ -717,7 +772,7 @@ static void *player_thread_func(void *arg) {
late_packet_message_sent=0;
missing_packets=late_packets=too_late_packets=resend_requests=0;
flush_rtp_timestamp=0;
flush_rtp_timestamp=0x7fffffff; // it seems this number has a special significance -- it seems to be used as a null operand, so we'll use it like that too
int sync_error_out_of_bounds = 0; // number of times in a row that there's been a serious sync error
while (!please_stop) {
abuf_t *inframe = buffer_get_frame();
@@ -955,8 +1010,8 @@ void player_flush(uint32_t timestamp) {
// debug(1,"Flush requested up to %u. It seems as if 2147483647 is special.",timestamp);
pthread_mutex_lock(&flush_mutex);
flush_requested=1;
if (timestamp!=0x7fffffff)
flush_rtp_timestamp=timestamp; // flush all packets up to (and including?) this
//if (timestamp!=0x7fffffff)
flush_rtp_timestamp=timestamp; // flush all packets up to (and including?) this
pthread_mutex_unlock(&flush_mutex);
}
rtp.c
+67 -6
View File
@@ -56,6 +56,10 @@ typedef struct time_ping_record {
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
static uint32_t client_active_remote; // used when you want to control the client...
static SOCKADDR rtp_client_control_socket; // a socket pointing to the control port of the client
static SOCKADDR rtp_client_timing_socket; // a socket pointing to the timing port of the client
static int audio_socket; // our local [server] audio socket
@@ -421,18 +425,20 @@ static int bind_port(SOCKADDR *remote,int *sock, int desired_port ) {
}
void rtp_setup(SOCKADDR *remote, int cport, int tport, int *lsport, int *lcport, int *ltport) {
void rtp_setup(SOCKADDR *remote, int cport, int tport, uint32_t active_remote, int *lsport, int *lcport, int *ltport) {
if (running)
die("rtp_setup called with active stream!");
debug(2, "rtp_setup: cport=%d tport=%d.", cport, tport);
client_active_remote = active_remote;
// print out what we know about the client
void *addr;
char *ipver;
int port;
char ipstr[INET6_ADDRSTRLEN];
char portstr[20];
client_ip_family = remote->SAFAMILY; // keep information about the kind of ip of the client
#ifdef AF_INET6
if (remote->SAFAMILY == AF_INET6) {
struct sockaddr_in6 *sa6 = (struct sockaddr_in6*)remote;
@@ -447,8 +453,8 @@ void rtp_setup(SOCKADDR *remote, int cport, int tport, int *lsport, int *lcport,
port = ntohs(sa4->sin_port);
ipver = "IPv4";
}
inet_ntop(remote->SAFAMILY,addr,ipstr,sizeof(ipstr));
debug(1,"Connection from %s: %s:%d",ipver,ipstr,port);
inet_ntop(remote->SAFAMILY,addr,client_ip_string,sizeof(client_ip_string)); // keep the client's ip number
debug(1,"Connection from %s: %s:%d",ipver,client_ip_string,port);
@@ -461,7 +467,7 @@ void rtp_setup(SOCKADDR *remote, int cport, int tport, int *lsport, int *lcport,
hints.ai_family = remote->SAFAMILY;
hints.ai_socktype = SOCK_DGRAM;
snprintf(portstr, 20, "%d", cport);
if (getaddrinfo(ipstr,portstr,&hints,&servinfo)!=0)
if (getaddrinfo(client_ip_string,portstr,&hints,&servinfo)!=0)
die("Can't get address of client's control port");
#ifdef AF_INET6
@@ -478,7 +484,7 @@ void rtp_setup(SOCKADDR *remote, int cport, int tport, int *lsport, int *lcport,
hints.ai_family = remote->SAFAMILY;
hints.ai_socktype = SOCK_DGRAM;
snprintf(portstr, 20, "%d", tport);
if (getaddrinfo(ipstr,portstr,&hints,&servinfo)!=0)
if (getaddrinfo(client_ip_string,portstr,&hints,&servinfo)!=0)
die("Can't get address of client's timing port");
#ifdef AF_INET6
if (servinfo->ai_family == AF_INET6)
@@ -566,3 +572,58 @@ void rtp_request_resend(seq_t first, uint32_t count) {
}
}
}
void rtp_request_client_pause() {
if (running) {
debug(1,"Send a request to pause to client at %s on port 3689 with active remote %u.",client_ip_string,client_active_remote);
struct addrinfo hints, *res;
int sockfd;
char message[1000] , server_reply[2000];
// first, load up address structs with getaddrinfo():
memset(&hints, 0, sizeof hints);
hints.ai_family = AF_UNSPEC;
hints.ai_socktype = SOCK_STREAM;
getaddrinfo(client_ip_string, "3689", &hints, &res);
// make a socket:
sockfd = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
if (sockfd == -1) {
die("Could not create socket");
}
debug(1,"Socket created");
// connect!
if (connect(sockfd, res->ai_addr, res->ai_addrlen) < 0) {
die("connect failed. Error");
}
// debug(1,"Connect successful");
sprintf(message,"GET /ctrl-int/1/pause HTTP/1.1\r\nHost: %s\r\nActive-Remote: %u\r\n\r\n",client_ip_string,client_active_remote);
// debug(1,"Sending this message: \"%s\".",message);
//Send some data
if( send(sockfd , message , strlen(message) , 0) < 0) {
debug(1,"Send failed");
}
//Receive a reply from the server
if( recv(sockfd , server_reply , 2000 , 0) < 0) {
debug(1,"recv failed");
}
// debug(1,"Server replied: \"%s\".",server_reply);
close(sockfd);
} else {
debug(1,"Request to pause non-existent play stream -- ignored.");
}
}
rtp.h
+4 -1
View File
@@ -3,9 +3,12 @@
#include <sys/socket.h>
void rtp_setup(SOCKADDR *remote, int controlport, int timingport, int *local_server_port, int *local_control_port, int *local_timing_port);
#include "player.h"
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);
void rtp_request_resend(seq_t first, uint32_t count);
void rtp_request_client_pause(void); // ask the client to pause
void get_reference_timestamp_stuff(uint32_t *timestamp,uint64_t *timestamp_time);
void clear_reference_timestamp(void);
rtsp.c
+11 -1
View File
@@ -436,6 +436,16 @@ static void handle_setup(rtsp_conn_info *conn,
rtsp_message *req, rtsp_message *resp) {
int cport, tport;
int lsport,lcport,ltport;
uint32_t active_remote=0;
char * ar = msg_get_header(req,"Active-Remote");
if (ar) {
debug(1,"Active-Remote string seen: \"%s\".",ar);
// get the active remote
char *p;
active_remote = strtoul(ar,&p,10);
debug(1,"Active Remote is %u.",active_remote);
}
if (config.userSuppliedLatency==0) {
char * ua = msg_get_header(req,"User-Agent");
@@ -473,7 +483,7 @@ static void handle_setup(rtsp_conn_info *conn,
tport = atoi(p);
rtsp_take_player();
rtp_setup(&conn->remote, cport, tport,&lsport,&lcport,&ltport);
rtp_setup(&conn->remote, cport, tport, active_remote, &lsport,&lcport,&ltport);
if (!lsport)
return;
char *q;
shairport.c
+9
View File
@@ -46,6 +46,7 @@
#include "common.h"
#include "rtsp.h"
#include "rtp.h"
#include "mdns.h"
#include <libdaemon/dfork.h>
@@ -91,6 +92,10 @@ static void sig_logrotate(int foo, siginfo_t *bar, void *baz) {
// log_setup();
}
static void sig_pause_client(int foo, siginfo_t *bar, void *baz) {
rtp_request_client_pause();
}
void print_version(void) {
char version_string[200];
strcpy(version_string,PACKAGE_VERSION);
@@ -276,6 +281,7 @@ void signal_setup(void) {
sigdelset(&set, SIGHUP);
sigdelset(&set, SIGSTOP);
sigdelset(&set, SIGCHLD);
sigdelset(&set, SIGUSR2);
pthread_sigmask(SIG_BLOCK, &set, NULL);
// setting this to SIG_IGN would prevent signalling any threads.
@@ -293,6 +299,9 @@ void signal_setup(void) {
sa.sa_sigaction = &sig_logrotate;
sigaction(SIGHUP, &sa, NULL);
sa.sa_sigaction = &sig_pause_client;
sigaction(SIGUSR2, &sa, NULL);
sa.sa_sigaction = &sig_child;
sigaction(SIGCHLD, &sa, NULL);
}