/** rtp.c * * Copyright (C) 2013-2015 Tox project All Rights Reserved. * * This file is part of Tox. * * Tox is free software: you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation, either version 3 of the License, or * (at your option) any later version. * * Tox is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with Tox. If not, see . * */ #ifdef HAVE_CONFIG_H #include "config.h" #endif /* HAVE_CONFIG_H */ #include "../toxcore/logger.h" #include "../toxcore/util.h" #include "rtp.h" #include #define size_32 4 #define RTCP_REPORT_INTERVAL_MS 500 #define ADD_FLAG_VERSION(_h, _v) do { ( _h->flags ) &= 0x3F; ( _h->flags ) |= ( ( ( _v ) << 6 ) & 0xC0 ); } while(0) #define ADD_FLAG_PADDING(_h, _v) do { if ( _v > 0 ) _v = 1; ( _h->flags ) &= 0xDF; ( _h->flags ) |= ( ( ( _v ) << 5 ) & 0x20 ); } while(0) #define ADD_FLAG_EXTENSION(_h, _v) do { if ( _v > 0 ) _v = 1; ( _h->flags ) &= 0xEF;( _h->flags ) |= ( ( ( _v ) << 4 ) & 0x10 ); } while(0) #define ADD_FLAG_CSRCC(_h, _v) do { ( _h->flags ) &= 0xF0; ( _h->flags ) |= ( ( _v ) & 0x0F ); } while(0) #define ADD_SETTING_MARKER(_h, _v) do { if ( _v > 1 ) _v = 1; ( _h->marker_payloadt ) &= 0x7F; ( _h->marker_payloadt ) |= ( ( ( _v ) << 7 ) /*& 0x80 */ ); } while(0) #define ADD_SETTING_PAYLOAD(_h, _v) do { if ( _v > 127 ) _v = 127; ( _h->marker_payloadt ) &= 0x80; ( _h->marker_payloadt ) |= ( ( _v ) /* & 0x7F */ ); } while(0) #define GET_FLAG_VERSION(_h) (( _h->flags & 0xd0 ) >> 6) #define GET_FLAG_PADDING(_h) (( _h->flags & 0x20 ) >> 5) #define GET_FLAG_EXTENSION(_h) (( _h->flags & 0x10 ) >> 4) #define GET_FLAG_CSRCC(_h) ( _h->flags & 0x0f ) #define GET_SETTING_MARKER(_h) (( _h->marker_payloadt ) >> 7) #define GET_SETTING_PAYLOAD(_h) ((_h->marker_payloadt) & 0x7f) typedef struct { uint64_t timestamp; /* in ms */ uint32_t received_packets; uint32_t expected_packets; /* ... other stuff in the future */ } RTCPReport; typedef struct RTCPSession_s { RTPSession *rtp_session; uint8_t prefix; uint64_t last_sent_report_ts; uint32_t last_received_packets; uint32_t last_expected_packets; RingBuffer* pl_stats; /* Packet loss stats over time */ } RTCPSession; /* queue_message() is defined in codec.c */ void queue_message(RTPSession *session, RTPMessage *msg); RTPHeader *parse_header_in ( const uint8_t *payload, int length ); RTPExtHeader *parse_ext_header_in ( const uint8_t *payload, uint16_t length ); RTPMessage *msg_parse ( const uint8_t *data, int length ); uint8_t *parse_header_out ( const RTPHeader* header, uint8_t* payload ); uint8_t *parse_ext_header_out ( const RTPExtHeader* header, uint8_t* payload ); void build_header ( RTPSession* session, RTPHeader* header ); void send_rtcp_report ( RTCPSession* session, Messenger* m, int32_t friendnumber ); int handle_rtp_packet ( Messenger *m, int32_t friendnumber, const uint8_t *data, uint32_t length, void *object ); int handle_rtcp_packet ( Messenger *m, int32_t friendnumber, const uint8_t *data, uint32_t length, void *object ); RTPSession *rtp_new ( int payload_type, Messenger *messenger, int friend_num ) { RTPSession *retu = calloc(1, sizeof(RTPSession)); if ( !retu ) { LOGGER_WARNING("Alloc failed! Program might misbehave!"); return NULL; } retu->version = RTP_VERSION; /* It's always 2 */ retu->ssrc = random_int(); retu->payload_type = payload_type % 128; retu->tstate = rtp_StateNormal; retu->m = messenger; retu->dest = friend_num; if ( !(retu->csrc = calloc(1, sizeof(uint32_t))) ) { LOGGER_WARNING("Alloc failed! Program might misbehave!"); free(retu); return NULL; } retu->csrc[0] = retu->ssrc; /* Set my ssrc to the list receive */ /* Also set payload type as prefix */ retu->prefix = payload_type; /* Initialize rtcp session */ if (!(retu->rtcp_session = calloc(1, sizeof(RTCPSession)))) { LOGGER_WARNING("Alloc failed! Program might misbehave!"); free(retu->csrc); free(retu); return NULL; } retu->rtcp_session->prefix = 222 + payload_type % 192; retu->rtcp_session->pl_stats = rb_new(4); retu->rtcp_session->rtp_session = retu; return retu; } void rtp_kill ( RTPSession *session ) { if ( !session ) return; rtp_stop_receiving (session); free ( session->ext_header ); free ( session->csrc ); void* t; while (!rb_empty(session->rtcp_session->pl_stats)) { rb_read(session->rtcp_session->pl_stats, (void**) &t); free(t); } rb_free(session->rtcp_session->pl_stats); LOGGER_DEBUG("Terminated RTP session: %p", session); /* And finally free session */ free ( session ); } void rtp_do(RTPSession *session) { if (!session || !session->rtcp_session) return; if (current_time_monotonic() - session->rtcp_session->last_sent_report_ts >= RTCP_REPORT_INTERVAL_MS) { send_rtcp_report(session->rtcp_session, session->m, session->dest); } if (rb_full(session->rtcp_session->pl_stats)) { RTCPReport* reports[4]; int i = 0; for (; rb_read(session->rtcp_session->pl_stats, (void**) reports + i); i++); /* Check for timed out reports (> 6 sec) */ uint64_t now = current_time_monotonic(); for (i = 0; i < 4 && now - reports[i]->timestamp < 6000; i ++); for (; i < 4; i ++) { rb_write(session->rtcp_session->pl_stats, reports[i]); reports[i] = NULL; } if (!rb_empty(session->rtcp_session->pl_stats)) { for (i = 0; reports[i] != NULL; i ++) free(reports[i]); return; /* As some reports are timed out, we need more... */ } /* We have 4 on-time reports so we can proceed */ uint32_t quality = 100; for (i = 0; i < 4; i++) { uint32_t idx = reports[i]->received_packets * 100 / reports[i]->expected_packets; quality = MIN(quality, idx); free(reports[i]); } if (quality <= 70) { session->tstate = rtp_StateBad; LOGGER_WARNING("Stream quality: BAD"); } else if (quality >= 99) { session->tstate = rtp_StateGood; LOGGER_DEBUG("Stream quality: GOOD"); } else { session->tstate = rtp_StateNormal; LOGGER_DEBUG("Stream quality: NORMAL"); } } } int rtp_start_receiving(RTPSession* session) { if (session == NULL) return -1; if (custom_lossy_packet_registerhandler(session->m, session->dest, session->prefix, handle_rtp_packet, session) == -1) { LOGGER_WARNING("Failed to register rtp receive handler"); return -1; } if (custom_lossy_packet_registerhandler(session->m, session->dest, session->rtcp_session->prefix, handle_rtcp_packet, session->rtcp_session) == -1) { LOGGER_WARNING("Failed to register rtcp receive handler"); custom_lossy_packet_registerhandler(session->m, session->dest, session->prefix, NULL, NULL); return -1; } return 0; } int rtp_stop_receiving(RTPSession* session) { if (session == NULL) return -1; custom_lossy_packet_registerhandler(session->m, session->dest, session->prefix, NULL, NULL); custom_lossy_packet_registerhandler(session->m, session->dest, session->rtcp_session->prefix, NULL, NULL); /* RTCP */ return 0; } int rtp_send_msg ( RTPSession *session, const uint8_t *data, uint16_t length ) { if ( !session ) { LOGGER_WARNING("No session!"); return -1; } uint8_t parsed[MAX_RTP_SIZE]; uint8_t *it; RTPHeader header[1]; build_header(session, header); uint32_t parsed_len = length + header->length + 1; parsed[0] = session->prefix; it = parse_header_out ( header, parsed + 1 ); if ( session->ext_header ) { parsed_len += ( 4 /* Minimum ext header len */ + session->ext_header->length * size_32 ); it = parse_ext_header_out ( session->ext_header, it ); } memcpy ( it, data, length ); if ( -1 == send_custom_lossy_packet(session->m, session->dest, parsed, parsed_len) ) { LOGGER_WARNING("Failed to send full packet (len: %d)! std error: %s", length, strerror(errno)); return -1; } /* Set sequ number */ session->sequnum = session->sequnum >= MAX_SEQU_NUM ? 0 : session->sequnum + 1; return 0; } void rtp_free_msg ( RTPSession *session, RTPMessage *msg ) { if ( !session ) { if ( msg->ext_header ) { free ( msg->ext_header->table ); free ( msg->ext_header ); } } else { if ( msg->ext_header && session->ext_header != msg->ext_header ) { free ( msg->ext_header->table ); free ( msg->ext_header ); } } free ( msg->header ); free ( msg ); } RTPHeader *parse_header_in ( const uint8_t *payload, int length ) { if ( !payload || !length ) { LOGGER_WARNING("No payload to extract!"); return NULL; } RTPHeader *retu = calloc(1, sizeof (RTPHeader)); if ( !retu ) { LOGGER_WARNING("Alloc failed! Program might misbehave!"); return NULL; } memcpy(&retu->sequnum, payload, sizeof(retu->sequnum)); retu->sequnum = ntohs(retu->sequnum); const uint8_t *it = payload + 2; retu->flags = *it; ++it; /* This indicates if the first 2 bits are valid. * Now it may happen that this is out of order but * it cuts down chances of parsing some invalid value */ if ( GET_FLAG_VERSION(retu) != RTP_VERSION ) { /* Deallocate */ LOGGER_WARNING("Invalid version!"); free(retu); return NULL; } /* * Added a check for the size of the header little sooner so * I don't need to parse the other stuff if it's bad */ uint8_t cc = GET_FLAG_CSRCC ( retu ); int total = 12 /* Minimum header len */ + ( cc * 4 ); if ( length < total ) { /* Deallocate */ LOGGER_WARNING("Length invalid!"); free(retu); return NULL; } retu->marker_payloadt = *it; ++it; retu->length = total; memcpy(&retu->timestamp, it, sizeof(retu->timestamp)); retu->timestamp = ntohl(retu->timestamp); it += 4; memcpy(&retu->ssrc, it, sizeof(retu->ssrc)); retu->ssrc = ntohl(retu->ssrc); uint8_t x; for ( x = 0; x < cc; x++ ) { it += 4; memcpy(&retu->csrc[x], it, sizeof(retu->csrc[x])); retu->csrc[x] = ntohl(retu->csrc[x]); } return retu; } RTPExtHeader *parse_ext_header_in ( const uint8_t *payload, uint16_t length ) { const uint8_t *it = payload; RTPExtHeader *retu = calloc(1, sizeof (RTPExtHeader)); if ( !retu ) { LOGGER_WARNING("Alloc failed! Program might misbehave!"); return NULL; } uint16_t ext_length; memcpy(&ext_length, it, sizeof(ext_length)); ext_length = ntohs(ext_length); it += 2; if ( length < ( ext_length * sizeof(uint32_t) ) ) { LOGGER_WARNING("Length invalid!"); free(retu); return NULL; } retu->length = ext_length; memcpy(&retu->type, it, sizeof(retu->type)); retu->type = ntohs(retu->type); it += 2; if ( !(retu->table = calloc(ext_length, sizeof (uint32_t))) ) { LOGGER_WARNING("Alloc failed! Program might misbehave!"); free(retu); return NULL; } uint16_t x; for ( x = 0; x < ext_length; x++ ) { it += 4; memcpy(&(retu->table[x]), it, sizeof(retu->table[x])); retu->table[x] = ntohl(retu->table[x]); } return retu; } RTPMessage *msg_parse ( const uint8_t *data, int length ) { RTPMessage *retu = calloc(1, sizeof (RTPMessage)); retu->header = parse_header_in ( data, length ); /* It allocates memory and all */ if ( !retu->header ) { LOGGER_WARNING("Header failed to extract!"); free(retu); return NULL; } uint16_t from_pos = retu->header->length; retu->length = length - from_pos; if ( GET_FLAG_EXTENSION ( retu->header ) ) { retu->ext_header = parse_ext_header_in ( data + from_pos, length ); if ( retu->ext_header ) { retu->length -= ( 4 /* Minimum ext header len */ + retu->ext_header->length * size_32 ); from_pos += ( 4 /* Minimum ext header len */ + retu->ext_header->length * size_32 ); } else { /* Error */ LOGGER_WARNING("Ext Header failed to extract!"); rtp_free_msg(NULL, retu); return NULL; } } else { retu->ext_header = NULL; } if ( length - from_pos <= MAX_RTP_SIZE ) memcpy ( retu->data, data + from_pos, length - from_pos ); else { LOGGER_WARNING("Invalid length!"); rtp_free_msg(NULL, retu); return NULL; } return retu; } uint8_t *parse_header_out ( const RTPHeader *header, uint8_t *payload ) { uint8_t cc = GET_FLAG_CSRCC ( header ); uint8_t *it = payload; uint16_t sequnum; uint32_t timestamp; uint32_t ssrc; uint32_t csrc; /* Add sequence number first */ sequnum = htons(header->sequnum); memcpy(it, &sequnum, sizeof(sequnum)); it += 2; *it = header->flags; ++it; *it = header->marker_payloadt; ++it; timestamp = htonl(header->timestamp); memcpy(it, ×tamp, sizeof(timestamp)); it += 4; ssrc = htonl(header->ssrc); memcpy(it, &ssrc, sizeof(ssrc)); uint8_t x; for ( x = 0; x < cc; x++ ) { it += 4; csrc = htonl(header->csrc[x]); memcpy(it, &csrc, sizeof(csrc)); } return it + 4; } uint8_t *parse_ext_header_out ( const RTPExtHeader *header, uint8_t *payload ) { uint8_t *it = payload; uint16_t length; uint16_t type; uint32_t entry; length = htons(header->length); memcpy(it, &length, sizeof(length)); it += 2; type = htons(header->type); memcpy(it, &type, sizeof(type)); it -= 2; /* Return to 0 position */ if ( header->table ) { uint16_t x; for ( x = 0; x < header->length; x++ ) { it += 4; entry = htonl(header->table[x]); memcpy(it, &entry, sizeof(entry)); } } return it + 4; } void build_header ( RTPSession *session, RTPHeader *header ) { ADD_FLAG_VERSION ( header, session->version ); ADD_FLAG_PADDING ( header, session->padding ); ADD_FLAG_EXTENSION ( header, session->extension ); ADD_FLAG_CSRCC ( header, session->cc ); ADD_SETTING_MARKER ( header, session->marker ); ADD_SETTING_PAYLOAD ( header, session->payload_type ); header->sequnum = session->sequnum; header->timestamp = current_time_monotonic(); /* milliseconds */ header->ssrc = session->ssrc; int i; for ( i = 0; i < session->cc; i++ ) header->csrc[i] = session->csrc[i]; header->length = 12 /* Minimum header len */ + ( session->cc * size_32 ); } void send_rtcp_report(RTCPSession* session, Messenger* m, int32_t friendnumber) { if (session->last_expected_packets == 0) return; uint8_t parsed[9]; parsed[0] = session->prefix; uint32_t received_packets = htonl(session->last_received_packets); uint32_t expected_packets = htonl(session->last_expected_packets); memcpy(parsed + 1, &received_packets, 4); memcpy(parsed + 5, &expected_packets, 4); if (-1 == send_custom_lossy_packet(m, friendnumber, parsed, sizeof(parsed))) LOGGER_WARNING("Failed to send full packet (len: %d)! std error: %s", sizeof(parsed), strerror(errno)); else { LOGGER_DEBUG("Sent rtcp report: ex: %d rc: %d", session->last_expected_packets, session->last_received_packets); session->last_received_packets = 0; session->last_expected_packets = 0; session->last_sent_report_ts = current_time_monotonic(); } } int handle_rtp_packet ( Messenger *m, int32_t friendnumber, const uint8_t *data, uint32_t length, void *object ) { RTPSession *session = object; RTPMessage *msg; if ( !session || length < 13 ) { /* 12 is the minimum length for rtp + desc. byte */ LOGGER_WARNING("No session or invalid length of received buffer!"); return -1; } msg = msg_parse ( data + 1, length - 1 ); if ( !msg ) { LOGGER_WARNING("Could not parse message!"); return -1; } /* Check if message came in late */ if ( msg->header->sequnum > session->rsequnum || msg->header->timestamp > session->rtimestamp ) { /* Not late */ if (msg->header->sequnum > session->rsequnum) session->rtcp_session->last_expected_packets += msg->header->sequnum - session->rsequnum; else if (msg->header->sequnum < session->rsequnum) session->rtcp_session->last_expected_packets += (msg->header->sequnum + 65535) - session->rsequnum; else /* Usual case when transmission starts */ session->rtcp_session->last_expected_packets ++; session->rsequnum = msg->header->sequnum; session->rtimestamp = msg->header->timestamp; } session->rtcp_session->last_received_packets ++; queue_message(session, msg); return 0; } int handle_rtcp_packet ( Messenger *m, int32_t friendnumber, const uint8_t *data, uint32_t length, void *object ) { if (length < 9) return -1; RTCPSession* session = object; RTCPReport* report = malloc(sizeof(RTCPReport)); memcpy(&report->received_packets, data + 1, 4); memcpy(&report->expected_packets, data + 5, 4); report->received_packets = ntohl(report->received_packets); report->expected_packets = ntohl(report->expected_packets); /* Invalid values */ if (report->expected_packets == 0 || report->received_packets > report->expected_packets) { LOGGER_WARNING("Malformed rtcp report! %d %d", report->expected_packets, report->received_packets); free(report); return 0; } report->timestamp = current_time_monotonic(); free(rb_write(session->pl_stats, report)); LOGGER_DEBUG("Got rtcp report: ex: %d rc: %d", report->expected_packets, report->received_packets); return 0; }