rtp.c

/*
 * Apple RTP protocol handler. This file is part of Shairport.
 * Copyright (c) James Laird 2013
 * All rights reserved.
 *
 * Permission is hereby granted, free of charge, to any person
 * obtaining a copy of this software and associated documentation
 * files (the "Software"), to deal in the Software without
 * restriction, including without limitation the rights to use,
 * copy, modify, merge, publish, distribute, sublicense, and/or
 * sell copies of the Software, and to permit persons to whom the
 * Software is furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be
 * included in all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
 * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
 * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
 * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
 * OTHER DEALINGS IN THE SOFTWARE.
 */

#include <time.h>
#include <pthread.h>
#include <signal.h>
#include <unistd.h>
#include <memory.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <stdio.h>
#include <errno.h>

#include "common.h"
#include "player.h"
#include "rtp.h"


typedef struct {
	uint32_t seconds;	
	uint32_t fraction;
} ntp_timestamp;

typedef struct time_ping_record {
  uint64_t local_to_remote_difference;
  uint64_t dispersion;
} 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 
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
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 uint32_t reference_timestamp;
static uint64_t reference_timestamp_time;

// debug variables
static int request_sent;

#define time_ping_history  8
#define time_ping_fudge_factor  100000

static uint8_t time_ping_count;
struct time_ping_record time_pings[time_ping_history];

//static struct timespec dtt; // dangerous -- this assumes that there will never be two timing request in flight at the same time
static uint64_t departure_time;  // dangerous -- this assumes that there will never be two timing request in flight at the same time

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) {
    // we inherit the signal mask (SIGUSR1)
    
    int32_t last_seqno = -1;
    uint8_t packet[2048], *pktp;

    ssize_t nread;
    while (1) {
        if (please_shutdown)
            break;
        nread = recv(audio_socket, packet, sizeof(packet), 0);
        if (nread < 0)
            break;

        ssize_t plen = nread;
        uint8_t type = packet[1] & ~0x80;
        if (type == 0x60 || type == 0x56) {   // audio data / resend
            pktp = packet;
            if (type==0x56) {
                pktp += 4;
                plen -= 4;
            }
            seq_t seqno = ntohs(*(unsigned short *)(pktp+2));
            // increment last_seqno and see if it's the same as the incoming seqno

						if (last_seqno==-1)
							last_seqno=seqno;
						else {
							last_seqno = (last_seqno+1)&0xffff;           
							if (seqno!=last_seqno)
								debug(2,"RTP: Packets out of sequence: expected: %d, got %d.",last_seqno,seqno);
								last_seqno=seqno; // reset warning...
            }
            uint32_t timestamp = ntohl(*(unsigned long *)(pktp+4));
            
            //if (packet[1]&0x10)
            //	debug(1,"Audio packet Extension bit set.");

            pktp += 12;
            plen -= 12;

            // check if packet contains enough content to be reasonable
            if (plen >= 16) {
                player_put_packet(seqno,timestamp, pktp, plen);
                continue;
            }
            if (type == 0x56 && seqno == 0) {
                debug(2, "resend-related request packet received, ignoring.");
                continue;
            }
            debug(1, "Audio receiver -- Unknown RTP packet of type 0x%02X length %d seqno %d", type, nread, seqno);
        }
        warn("Audio receiver -- Unknown RTP packet of type 0x%02X length %d.", type, nread);
    }

    debug(1, "Audio receiver -- Server RTP thread interrupted. terminating.");
    close(audio_socket);

    return NULL;
}

static void *rtp_control_receiver(void *arg) {
    // we inherit the signal mask (SIGUSR1)
    reference_timestamp=0; // nothing valid received yet
    uint8_t packet[2048];
    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;
        nread = recv(control_socket, packet, sizeof(packet), 0);
        local_time_now=get_absolute_time_in_fp();
//        clock_gettime(CLOCK_MONOTONIC,&tn);
//        local_time_now=((uint64_t)tn.tv_sec<<32)+((uint64_t)tn.tv_nsec<<32)/1000000000;
        
        if (nread < 0)
            break;

        ssize_t plen = nread;
        if (packet[1] == 0xd4) {   // sync data
        /*
        char obf[4096];
        char *obfp = obf;
        int obfc;
        for (obfc=0;obfc<plen;obfc++) {
          sprintf(obfp,"%02X",packet[obfc]);
          obfp+=2;
        };
        *obfp=0;
        debug(1,"Sync Packet Received: \"%s\"",obf);
        */
          if (local_to_remote_time_difference) { // need a time packet to be interchanged first...
          	
          	remote_time_of_sync = (uint64_t)ntohl(*((uint32_t*)&packet[8]))<<32;
            remote_time_of_sync += ntohl(*((uint32_t*)&packet[12]));
            
            // debug(1,"Remote Sync Time: %0llx.",remote_time_of_sync);
          
          	rtp_timestamp_less_latency = ntohl(*((uint32_t*)&packet[4]));
           	sync_rtp_timestamp = ntohl(*((uint32_t*)&packet[16]));
           	
           	if (packet[0]&0x10) {
           	  // if it's a packet right after a flush or resume
           	  sync_rtp_timestamp += 352; // add frame_size -- can't see a reference to this anywhere, but it seems to get everything into sync.
           	  // it's as if the first sync after a flush or resume is the timing of the next packet after the one whose RTP is given. Weird.
           	}
          	pthread_mutex_lock(&reference_time_mutex);
          	reference_timestamp_time = remote_time_of_sync-local_to_remote_time_difference;
          	reference_timestamp = sync_rtp_timestamp;
          	pthread_mutex_unlock(&reference_time_mutex);
          	// debug(1,"New Reference timestamp and timestamp time...");         	
          	// get estimated remote time now
          	remote_time_now = local_time_now+local_to_remote_time_difference;          	
          	
             //debug(1,"Sync Time is %lld us late (remote times).",((remote_time_now-remote_time_of_sync)*1000000)>>32);      				
             //debug(1,"Sync Time is %lld us late (local times).",((local_time_now-reference_timestamp_time)*1000000)>>32);      				
          } else {
            debug(1,"Sync packet received before we got a timing packet back.");
          }
        } else 
          debug(1,"Control Port -- Unknown RTP packet of type 0x%02X length %d.", packet[1], nread);
    }

    debug(1, "Control RTP thread interrupted. terminating.");
    close(control_socket);

    return NULL;
}

static void *rtp_timing_sender(void *arg) {
    struct timing_request {
    char leader;
    char type;
    uint16_t seqno;
    uint32_t filler;
    uint64_t origin,receive,transmit;
  };
  
  uint64_t request_number=0;
  
  struct timing_request req;  // *not* a standard RTCP NACK
 
  req.leader = 0x80;
  req.type = 0xd2;  // Timing request
  req.filler = 0;
  req.seqno=htons(7);
  
  time_ping_count = 0;

  // we inherit the signal mask (SIGUSR1)
  while (1) {
    if (please_shutdown)
      break;

    if (!running)
      die("rtp_timing_sender called without active stream!");

    //debug(1, "Requesting ntp timestamp exchange.");

    req.filler = 0;
    req.origin = req.receive = req.transmit=0;

//    clock_gettime(CLOCK_MONOTONIC,&dtt);
    departure_time=get_absolute_time_in_fp();
      socklen_t msgsize = sizeof(struct sockaddr_in);
#ifdef AF_INET6
      if (rtp_client_timing_socket.SAFAMILY==AF_INET6) {
          msgsize = sizeof(struct sockaddr_in6);
      }
#endif
      if (sendto(timing_socket, &req, sizeof(req), 0, (struct sockaddr*)&rtp_client_timing_socket, msgsize)==-1) {
        perror("Error sendto-ing to timing socket");
      }
    request_number++;
    if (request_number<=4)
      usleep(500000);
    else
      sleep(3);
  }
  debug(1, "rtp_timing_sender thread interrupted. terminating.");    
  return NULL;
}

static void *rtp_timing_receiver(void *arg) {
    // we inherit the signal mask (SIGUSR1)
    uint8_t packet[2048], *pktp;
    ssize_t nread;
    pthread_t timer_requester;
    pthread_create(&timer_requester, NULL, &rtp_timing_sender, NULL);
//    struct timespec att; 
    uint64_t distant_receive_time,distant_transmit_time,arrival_time,return_time,transit_time,processing_time;
    local_to_remote_time_jitters = 0;
    local_to_remote_time_jitters_count = 0;
    uint64_t l2rtd=0;
    while (1) {
      if (please_shutdown)
          break;
      nread = recv(timing_socket, packet, sizeof(packet), 0);
      arrival_time=get_absolute_time_in_fp();
//      clock_gettime(CLOCK_MONOTONIC,&att);
      
      if (nread < 0)
          break;

      ssize_t plen = nread;
      //debug(1,"Packet Received on Timing Port.");
      if (packet[1] == 0xd3) {   // timing reply
        /*
        char obf[4096];
        char *obfp = obf;
        int obfc;
        for (obfc=0;obfc<plen;obfc++) {
          sprintf(obfp,"%02X",packet[obfc]);
          obfp+=2;
        };
        *obfp=0;
        //debug(1,"Timing Packet Received: \"%s\"",obf);
        */  
        
        //arrival_time = ((uint64_t)att.tv_sec<<32)+((uint64_t)att.tv_nsec<<32)/1000000000;
        //departure_time = ((uint64_t)dtt.tv_sec<<32)+((uint64_t)dtt.tv_nsec<<32)/1000000000;
        
        return_time = arrival_time-departure_time;

        // uint64_t rtus = (return_time*1000000)>>32; debug(1,"Time ping turnaround time: %lld us.",rtus); 
        
        //distant_receive_time = ((uint64_t)ntohl(*((uint32_t*)&packet[16])))<<32+ntohl(*((uint32_t*)&packet[20]));
        
        distant_receive_time = (uint64_t)ntohl(*((uint32_t*)&packet[16]))<<32;
        distant_receive_time += ntohl(*((uint32_t*)&packet[20]));
        
        //distant_transmit_time = ((uint64_t)ntohl(*((uint32_t*)&packet[24])))<<32+ntohl(*((uint32_t*)&packet[28]));
        
        distant_transmit_time = (uint64_t)ntohl(*((uint32_t*)&packet[24]))<<32;
        distant_transmit_time += ntohl(*((uint32_t*)&packet[28]));
        
        processing_time = distant_transmit_time-distant_receive_time;
                
        // debug(1,"Return trip time: %lluuS, remote processing time: %lluuS.",(return_time*1000000)>>32,(processing_time*1000000)>>32); 

        uint64_t local_time_by_remote_clock = distant_transmit_time+return_time/2;
        
        unsigned int cc;       
        for (cc=time_ping_history-1;cc>0;cc--) {
          time_pings[cc]=time_pings[cc-1];
          time_pings[cc].dispersion = (time_pings[cc].dispersion*133)/100; // make the dispersions 'age' by this rational factor
        }
        time_pings[0].local_to_remote_difference = local_time_by_remote_clock-arrival_time;
        time_pings[0].dispersion = return_time;
        if (time_ping_count<time_ping_history)
          time_ping_count++;
        
        
        // now pick the timestamp with the lowest dispersion
        uint64_t l2rtd = time_pings[0].local_to_remote_difference;
        uint64_t tld = time_pings[0].dispersion;
        for (cc=1;cc<time_ping_count;cc++)
          if (time_pings[cc].dispersion<tld) {
            l2rtd=time_pings[cc].local_to_remote_difference;
            tld=time_pings[cc].dispersion;
          }
        int64_t ji;

        if (time_ping_count>1) {
          if (l2rtd>local_to_remote_time_difference) {
            local_to_remote_time_jitters=local_to_remote_time_jitters+l2rtd-local_to_remote_time_difference;
            ji = l2rtd-local_to_remote_time_difference;
          } else {
            local_to_remote_time_jitters=local_to_remote_time_jitters+local_to_remote_time_difference-l2rtd;
            ji = - (local_to_remote_time_difference-l2rtd);
          }
          local_to_remote_time_jitters_count+=1;
        }
        // uncomment below to print jitter between client's clock and oour clock
        // int64_t rtus = (tld*1000000)>>32; ji = (ji*1000000)>>32; debug(1,"Choosing time difference with dispersion of %lld us with delta of %lld us",rtus,ji);

        local_to_remote_time_difference=l2rtd;
        } else {
      	debug(1, "Timing port -- Unknown RTP packet of type 0x%02X length %d.", packet[1], nread);
      }
    }

    debug(1, "Timing RTP thread interrupted. terminating.");
    void *retval;
    pthread_kill(timer_requester, SIGUSR1);
    pthread_join(timer_requester, &retval);
    debug(1,"Closed and terminated timer requester thread.");
    debug(1, "Timing RTP thread terminated.");
    close(timing_socket);

    return NULL;
}

static int bind_port(SOCKADDR *remote,int *sock, int desired_port ) {
    struct 
    addrinfo hints, *info;

    memset(&hints, 0, sizeof(hints));
    hints.ai_family = remote->SAFAMILY;
    hints.ai_socktype = SOCK_DGRAM;
    hints.ai_flags = AI_PASSIVE;
    
    char buffer[10];
    snprintf(buffer, 10, "%d", desired_port);
 
    int ret = getaddrinfo(NULL,buffer, &hints, &info);

    if (ret < 0)
        die("failed to get usable addrinfo?! %s.", gai_strerror(ret));

    *sock = socket(remote->SAFAMILY, SOCK_DGRAM, IPPROTO_UDP);
    ret = bind(*sock, info->ai_addr, info->ai_addrlen);

    freeaddrinfo(info);

    if (ret < 0)
        die("could not bind a UDP port!");

    int sport;
    SOCKADDR local;
    socklen_t local_len = sizeof(local);
    getsockname(*sock, (struct sockaddr*)&local, &local_len);
#ifdef AF_INET6
    if (local.SAFAMILY == AF_INET6) {
        struct sockaddr_in6 *sa6 = (struct sockaddr_in6*)&local;
        sport = ntohs(sa6->sin6_port);
    } else
#endif
    {
        struct sockaddr_in *sa = (struct sockaddr_in*)&local;
        sport = ntohs(sa->sin_port);
    }

    return sport;
}


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 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;
        addr = &(sa6->sin6_addr);
        port = ntohs(sa6->sin6_port);
        ipver = "IPv6";
    }
#endif
    if (remote->SAFAMILY == AF_INET) {
        struct sockaddr_in *sa4 = (struct sockaddr_in*)remote;
        addr = &(sa4->sin_addr);
        port = ntohs(sa4->sin_port);
        ipver = "IPv4";
    }
    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);
    
    

    // set up a the record of the remote's control socket
    struct addrinfo hints;
    struct addrinfo *servinfo;

    memset(&rtp_client_control_socket,0,sizeof(rtp_client_control_socket));
    memset(&hints,0, sizeof hints);
    hints.ai_family = remote->SAFAMILY;
    hints.ai_socktype = SOCK_DGRAM;
    snprintf(portstr, 20, "%d", cport);
    if (getaddrinfo(client_ip_string,portstr,&hints,&servinfo)!=0)
        die("Can't get address of client's control port");

#ifdef AF_INET6
    if (servinfo->ai_family == AF_INET6)
        memcpy(&rtp_client_control_socket,servinfo->ai_addr,sizeof(struct sockaddr_in6));
    else
#endif
        memcpy(&rtp_client_control_socket,servinfo->ai_addr,sizeof(struct sockaddr_in));
    freeaddrinfo(servinfo);
    
    // set up a the record of the remote's timing socket
    memset(&rtp_client_timing_socket,0,sizeof(rtp_client_timing_socket));
    memset(&hints,0, sizeof hints);
    hints.ai_family = remote->SAFAMILY;
    hints.ai_socktype = SOCK_DGRAM;
    snprintf(portstr, 20, "%d", tport);
    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)
        memcpy(&rtp_client_timing_socket,servinfo->ai_addr,sizeof(struct sockaddr_in6));
    else
#endif
        memcpy(&rtp_client_timing_socket,servinfo->ai_addr,sizeof(struct sockaddr_in));
    freeaddrinfo(servinfo);
    
    // now, we open three sockets -- one for the audio stream, one for the timing and one for the control
    
     *lsport = bind_port(remote,&audio_socket,0);
     *lcport = bind_port(remote,&control_socket,0);
     *ltport = bind_port(remote,&timing_socket,0);

    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);

    running = 1;
    request_sent=0;
}

void get_reference_timestamp_stuff(uint32_t *timestamp,uint64_t *timestamp_time) {
  pthread_mutex_lock(&reference_time_mutex);
    *timestamp=reference_timestamp;
    *timestamp_time = reference_timestamp_time;
  pthread_mutex_unlock(&reference_time_mutex);
}

void clear_reference_timestamp(void) {
  pthread_mutex_lock(&reference_time_mutex);
  reference_timestamp=0;
  reference_timestamp_time=0;
  pthread_mutex_unlock(&reference_time_mutex);
}

void rtp_shutdown(void) {
    if (!running)
        die("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);
    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;
      }

      char req[8];    // *not* a standard RTCP NACK
      req[0] = 0x80;
      req[1] = 0x55|0x80;  // Apple 'resend'
      *(unsigned short *)(req+2) = htons(1);  // our seqnum
      *(unsigned short *)(req+4) = htons(first);  // missed seqnum
      *(unsigned short *)(req+6) = htons(count);  // count
        socklen_t msgsize = sizeof(struct sockaddr_in);
#ifdef AF_INET6
        if (rtp_client_timing_socket.SAFAMILY==AF_INET6) {
            msgsize = sizeof(struct sockaddr_in6);
        }
#endif
        if (sendto(audio_socket, req, sizeof(req), 0, (struct sockaddr*)&rtp_client_control_socket,msgsize)==-1) {
          perror("Error sendto-ing to audio socket");
        }
    } else {
      if (!request_sent) {
        debug(2,"rtp_request_resend called without active stream!");
        request_sent=1;
      }
    }
}

void rtp_request_client_pause() {
  if (running) {
  	if (client_active_remote==0) {
  		debug(1,"Can't request a client pause: no valid active remote.");
  	} else {
			// debug(1,"Send a client pause request to %s: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:3689\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);
		
			if (strstr(server_reply,"HTTP/1.1 204 No Content")!=server_reply)
				debug(1,"Client pause request failed.");
				// debug(1,"Client pause request failed: \"%s\".",server_reply);
			close(sockfd);
    }
  } else {
    debug(1,"Request to pause non-existent play stream -- ignored.");
  }
}