diff --git a/player.c b/player.c index bd1f2ce0..754d30d1 100644 --- a/player.c +++ b/player.c @@ -437,13 +437,11 @@ void player_put_packet(int original_format, seq_t seqno, uint32_t actual_timesta if (conn->connection_state_to_output) { // if we are supposed to be processing these packets abuf_t *abuf = 0; if (!conn->ab_synced) { - // if this is the first packet... - debug(3, "syncing to seqno %u.", seqno); conn->ab_write = seqno; conn->ab_read = seqno; conn->ab_synced = 1; conn->first_packet_timestamp = 0; - debug(1, "synced by first packet"); + debug(2, "Connection %d: synced by first packet, seqno %u.", conn->connection_number, seqno); } else if (original_format == 0) { // if the packet is coming in original format, the sequence number is important // otherwise, ignore is by setting it equal to the expected sequence number in ab_write @@ -904,8 +902,8 @@ static abuf_t *buffer_get_frame(rtsp_conn_info *conn) { conn->flush_rtp_timestamp = 0; debug_mutex_unlock(&conn->flush_mutex, 0); } - debug_mutex_lock(&conn->flush_mutex, 1000, 0); + pthread_cleanup_push(mutex_unlock, &conn->flush_mutex); if (conn->flush_requested == 1) { if (conn->flush_output_flushed == 0) if (config.output->flush) { @@ -999,7 +997,7 @@ static abuf_t *buffer_get_frame(rtsp_conn_info *conn) { conn->flush_rtp_timestamp = 0; conn->flush_output_flushed = 0; } - debug_mutex_unlock(&conn->flush_mutex, 0); + pthread_cleanup_pop(1); // unlock the conn->flush_mutex if (conn->ab_synced) { curframe = conn->audio_buffer + BUFIDX(conn->ab_read); @@ -1543,10 +1541,16 @@ void player_thread_cleanup_handler(void *arg) { pthread_cleanup_push(mutex_unlock, &playing_conn_lock); if (playing_conn == conn) { - if (config.output->stop) + if (config.output->stop) { + debug(3, "Connection %d: Stop the output backend.", conn->connection_number); config.output->stop(); + } + } else { + debug(1, "This is not the playing conn."); } pthread_cleanup_pop(1); // unlock the mutex + } else { + debug(1, "Can not acquire play lock."); } int oldState; @@ -1586,65 +1590,33 @@ void player_thread_cleanup_handler(void *arg) { #ifdef CONFIG_AIRPLAY_2 if (conn->airplay_type == ap_2) { - debug(1, "In AP2 mode"); + debug(2, "Cancelling AP2 timing, control and audio threads..."); if (conn->airplay_stream_type == realtime_stream) { - debug(2, "Connection %d: TEARDOWN Delete Realtime Audio Stream Thread", + debug(2, "Connection %d: Delete Realtime Audio Stream thread", conn->connection_number); pthread_cancel(conn->rtp_realtime_audio_thread); pthread_join(conn->rtp_realtime_audio_thread, NULL); - if (conn->realtime_audio_socket) - close(conn->realtime_audio_socket); + } else if (conn->airplay_stream_type == buffered_stream) { - debug(2, "Connection %d: TEARDOWN Delete Buffered Audio Stream Thread", - conn->connection_number); + + debug(2, "Connection %d: Delete Buffered Audio Stream thread", conn->connection_number); pthread_cancel(conn->rtp_buffered_audio_thread); pthread_join(conn->rtp_buffered_audio_thread, NULL); - if (conn->buffered_audio_socket) - close(conn->buffered_audio_socket); + } else { die("Unrecognised Stream Type"); } - if (conn->rtp_ap2_control_thread) { - debug(1, "Cancelling rtp_ap2_control_thread"); - pthread_cancel(conn->rtp_ap2_control_thread); - pthread_join(conn->rtp_ap2_control_thread, NULL); - } - if (conn->ap2_control_socket) { - close(conn->ap2_control_socket); - conn->ap2_control_socket = 0; - conn->ap2_remote_control_socket_addr_length = - 0; // indicates to the control receiver thread that the socket address need to be - // recreated. - } + debug(2, "Connection %d: Delete AirPlay 2 Control thread"); + pthread_cancel(conn->rtp_ap2_control_thread); + pthread_join(conn->rtp_ap2_control_thread, NULL); - if (conn->airplay_stream_type == realtime_stream) { - // realtime stuff - if (conn->rtp_realtime_audio_thread) { - debug(1, "Cancelling rtp_realtime_audio_thread"); - pthread_cancel(conn->rtp_realtime_audio_thread); - pthread_join(conn->rtp_realtime_audio_thread, NULL); - } - if (conn->realtime_audio_socket) { - close(conn->realtime_audio_socket); - conn->realtime_audio_socket = 0; - } - } else if (conn->airplay_stream_type == buffered_stream) { - // buffered stuff - if (conn->rtp_buffered_audio_thread) { - debug(3, "Cancelling rtp_buffered_audio_thread"); - pthread_cancel(conn->rtp_buffered_audio_thread); - pthread_join(conn->rtp_buffered_audio_thread, NULL); - } - if (conn->buffered_audio_socket) { - close(conn->buffered_audio_socket); - conn->buffered_audio_socket = 0; - } - } } else { + debug(1, "Cancelling AP1-compatible timing, control and audio threads..."); +#else + debug(2, "Cancelling AP1 timing, control and audio threads..."); #endif - debug(1, "Cancelling AP1 timing, control and audio threads..."); debug(3, "Cancel timing thread."); pthread_cancel(conn->rtp_timing_thread); debug(3, "Join timing thread."); @@ -1689,6 +1661,7 @@ void player_thread_cleanup_handler(void *arg) { release_play_lock(conn); conn->rtp_running = 0; pthread_setcancelstate(oldState, NULL); + debug(2, "Connection %d: player terminated.", conn->connection_number); } void *player_thread_func(void *arg) { @@ -3029,7 +3002,7 @@ void player_volume(double airplay_volume, rtsp_conn_info *conn) { void do_flush(uint32_t timestamp, rtsp_conn_info *conn) { - debug(2, "do_flush: flush to %u.", timestamp); + debug(3, "do_flush: flush to %u.", timestamp); debug_mutex_lock(&conn->flush_mutex, 1000, 1); conn->flush_requested = 1; conn->flush_rtp_timestamp = timestamp; // flush all packets up to, but not including, this one. diff --git a/rtp.c b/rtp.c index 84f0d800..8877285d 100644 --- a/rtp.c +++ b/rtp.c @@ -290,8 +290,8 @@ void *rtp_control_receiver(void *arg) { obfp += 2; }; *obfp = 0; - - + + // get raw timestamp information // I think that a good way to understand these timestamps is that // (1) the rtlt below is the timestamp of the frame that should be playing at the @@ -302,19 +302,19 @@ void *rtp_control_receiver(void *arg) { // Thus, (3) the latency can be calculated by subtracting the second from the // first. // There must be more to it -- there something missing. - + // In addition, it seems that if the value of the short represented by the second // pair of bytes in the packet is 7 // then an extra time lag is expected to be added, presumably by // the AirPort Express. - + // Best guess is that this delay is 11,025 frames. - + uint32_t rtlt = nctohl(&packet[4]); // raw timestamp less latency uint32_t rt = nctohl(&packet[16]); // raw timestamp - + uint32_t fl = nctohs(&packet[2]); // - + debug(1,"Sync Packet of %d bytes received: \"%s\", flags: %d, timestamps %u and %u, giving a latency of %d frames.",plen,obf,fl,rt,rtlt,rt-rtlt); //debug(1,"Monotonic timestamps are: %" PRId64 " and %" PRId64 " @@ -1330,12 +1330,12 @@ int local_ptp_time_to_frame(uint64_t time, uint32_t *frame, rtsp_conn_info *conn void rtp_event_receiver_cleanup_handler(void *arg) { rtsp_conn_info *conn = (rtsp_conn_info *)arg; - debug(2, "Connection %d: Event Receiver Cleanup.", conn->connection_number); + debug(2, "Connection %d: AP2 Event Receiver Cleanup.", conn->connection_number); } void *rtp_event_receiver(void *arg) { - debug(2, "Event Receiver started"); rtsp_conn_info *conn = (rtsp_conn_info *)arg; + debug(2, "Connection %d: AP2 Event Receiver started", conn->connection_number); pthread_cleanup_push(rtp_event_receiver_cleanup_handler, arg); listen(conn->event_socket, 5); @@ -1355,7 +1355,7 @@ void *rtp_event_receiver(void *arg) { if (nread < 0) { char errorstring[1024]; strerror_r(errno, (char *)errorstring, sizeof(errorstring)); - debug(1, "error in rtp_event_receiver %d: \"%s\". Could not recv a packet.", errno, + debug(1, "Connection %d: error in ap2 rtp_event_receiver %d: \"%s\". Could not recv a packet.", conn->connection_number, errno, errorstring); // if ((config.diagnostic_drop_packet_fraction == 0.0) || // (drand48() > config.diagnostic_drop_packet_fraction)) { @@ -1364,10 +1364,10 @@ void *rtp_event_receiver(void *arg) { // ssize_t plen = nread; debug(1, "Packet Received on Event Port."); if (packet[1] == 0xD7) { - debug(1, "Event Receiver -- Time Announce RTP packet of type 0x%02X length %d received.", + debug(1, "Connection %d: AP2 Event Receiver -- Time Announce RTP packet of type 0x%02X length %d received.", conn->connection_number, packet[1], nread); } else { - debug(1, "Event Receiver -- Unknown RTP packet of type 0x%02X length %d received.", + debug(1, "Connection %d: AP2 Event Receiver -- Unknown RTP packet of type 0x%02X length %d received.", conn->connection_number, packet[1], nread); } // } else { @@ -1377,17 +1377,22 @@ void *rtp_event_receiver(void *arg) { finished = 1; } } while (finished == 0); - debug(1, "Event Receiver RTP thread \"normal\" exit."); + debug(1, "Connection %d: AP2 Event Receiver RTP thread \"normal\" exit.", conn->connection_number); pthread_cleanup_pop(1); // close the socket pthread_cleanup_pop(1); // do the cleanup - debug(1, "Connection %d: Event Receiver RTP thread exit.", conn->connection_number); + debug(1, "Connection %d: AP2 Event Receiver RTP thread exit.", conn->connection_number); pthread_exit(NULL); } void rtp_ap2_control_handler_cleanup_handler(void *arg) { rtsp_conn_info *conn = (rtsp_conn_info *)arg; debug(2, "Connection %d: AP2 Control Receiver Cleanup.", conn->connection_number); + close(conn->ap2_control_socket); + conn->ap2_control_socket = 0; + conn->ap2_remote_control_socket_addr_length = + 0; // indicates to the control receiver thread that the socket address need to be + // recreated (needed for resend requests in the realtime mode) } int32_t decipher_player_put_packet(uint8_t *ciphered_audio_alt, ssize_t nread, @@ -1463,7 +1468,8 @@ void *rtp_ap2_control_receiver(void *arg) { rtsp_conn_info *conn = (rtsp_conn_info *)arg; uint8_t packet[4096]; ssize_t nread; - while (1) { + int keep_going = 1; + while (keep_going) { SOCKADDR from_sock_addr; socklen_t from_sock_addr_length = sizeof(SOCKADDR); memset(&from_sock_addr, 0, sizeof(SOCKADDR)); @@ -1471,7 +1477,7 @@ void *rtp_ap2_control_receiver(void *arg) { nread = recvfrom(conn->ap2_control_socket, packet, sizeof(packet), 0, (struct sockaddr *)&from_sock_addr, &from_sock_addr_length); - if (nread >= 0) { + if (nread > 0) { // debug(1,"rtp_ap2_control_receiver coded: %u, %u", packet[0], packet[1]); if ((config.diagnostic_drop_packet_fraction == 0.0) || @@ -1552,22 +1558,25 @@ void *rtp_ap2_control_receiver(void *arg) { } else { debug(1, "AP2 Control Receiver -- dropping a packet."); } + } else if (nread == 0) { + debug(1, "AP2 Control Receiver -- connection closed."); + keep_going = 0; } else { debug(1, "AP2 Control Receiver -- error %d receiving a packet.", errno); } } debug(1, "AP2 Control RTP thread \"normal\" exit -- this can't happen. Hah!"); - pthread_cleanup_pop(0); // don't execute anything here. + pthread_cleanup_pop(1); debug(1, "AP2 Control RTP thread exit."); pthread_exit(NULL); } void rtp_realtime_audio_cleanup_handler(__attribute__((unused)) void *arg) { - debug(1, "Realtime Audio Receiver Cleanup Start."); + debug(2, "Realtime Audio Receiver Cleanup Start."); rtsp_conn_info *conn = (rtsp_conn_info *)arg; close(conn->realtime_audio_socket); conn->realtime_audio_socket = 0; - debug(1, "Realtime Audio Receiver Cleanup Done."); + debug(2, "Realtime Audio Receiver Cleanup Done."); } void *rtp_realtime_audio_receiver(void *arg) { @@ -1627,14 +1636,6 @@ void *rtp_realtime_audio_receiver(void *arg) { pthread_exit(NULL); } -void rtp_buffered_audio_receiver_cleanup_handler(__attribute__((unused)) void *arg) { - debug(1, "Buffered Audio Receiver Cleanup Start."); - rtsp_conn_info *conn = (rtsp_conn_info *)arg; - close(conn->buffered_audio_socket); - conn->buffered_audio_socket = 0; - debug(1, "Buffered Audio Receiver Cleanup Done."); -} - ssize_t buffered_read(buffered_tcp_desc *descriptor, void *buf, size_t count) { ssize_t response; // usleep(1500000); @@ -1676,10 +1677,10 @@ ssize_t buffered_read(buffered_tcp_desc *descriptor, void *buf, size_t count) { return response; } -#define STANDARD_PACKET_SIZE 65536 +#define STANDARD_PACKET_SIZE 4096 void buffered_tcp_reader_cleanup_handler(__attribute__((unused)) void *arg) { - debug(2, "Buffered TCP Reader Thread Exit via Cleanup."); + debug(1, "Buffered TCP Reader Thread Exit via Cleanup."); } void *buffered_tcp_reader(void *arg) { @@ -1849,11 +1850,11 @@ void addADTStoPacket(uint8_t *packet, int packetLen) { } void rtp_buffered_audio_cleanup_handler(__attribute__((unused)) void *arg) { - debug(3, "Buffered Audio Receiver Cleanup Start."); + debug(1, "Buffered Audio Receiver Cleanup Start."); rtsp_conn_info *conn = (rtsp_conn_info *)arg; close(conn->buffered_audio_socket); conn->buffered_audio_socket = 0; - debug(2, "Buffered Audio Receiver Cleanup Done."); + debug(1, "Buffered Audio Receiver Cleanup Done."); } void *rtp_buffered_audio_processor(void *arg) { @@ -2093,7 +2094,7 @@ void *rtp_buffered_audio_processor(void *arg) { if ((lead_time >= (int64_t)(reqested_lead_time * 1000000000)) || (streaming_has_started == 1)) { if (streaming_has_started == 0) - debug(1, "rtp lead time is %f ms.", 0.000001 * lead_time); + debug(2, "Connection %d: buffered audio lead time is %f seconds.", 0.000000001 * lead_time); streaming_has_started = 1; player_put_packet(0, 0, pcm_buffer_read_point_rtptime, pcm_buffer + pcm_buffer_read_point, 352, conn); @@ -2371,6 +2372,7 @@ void *rtp_buffered_audio_processor(void *arg) { pthread_cleanup_pop(1); // do the cleanup. pthread_exit(NULL); } + int frame_to_local_time(uint32_t timestamp, uint64_t *time, rtsp_conn_info *conn) { if (conn->timing_type == ts_ptp) return frame_to_ptp_local_time(timestamp, time, conn); diff --git a/rtsp.c b/rtsp.c index dde6cff7..c82c6bd7 100644 --- a/rtsp.c +++ b/rtsp.c @@ -1339,6 +1339,7 @@ void handle_setrateanchori(rtsp_conn_info *conn, rtsp_message *req, rtsp_message plist_t messagePlist = plist_from_rtsp_content(req); if (messagePlist != NULL) { + pthread_cleanup_push(plist_cleanup, (void *)messagePlist); plist_t item = plist_dict_get_item(messagePlist, "networkTimeSecs"); if (item != NULL) { plist_t item_2 = plist_dict_get_item(messagePlist, "networkTimeTimelineID"); @@ -1388,16 +1389,9 @@ void handle_setrateanchori(rtsp_conn_info *conn, rtsp_message *req, rtsp_message plist_get_uint_val(item, &rate); debug(2, "anchor rate 0x%016" PRIx64 ".", rate); debug_mutex_lock(&conn->flush_mutex, 1000, 1); + pthread_cleanup_push(mutex_unlock, &conn->flush_mutex); conn->ap2_rate = rate; if ((rate & 1) != 0) { - - get_play_lock(conn); - - if (conn->ap2_timing_peer_list_message) - ptp_send_control_message_string(conn->ap2_timing_peer_list_message); - else - debug(1, "Connection %d: No timing peer list!", conn->connection_number); - debug(2, "Connection %d: Start playing.", conn->connection_number); conn->ap2_play_enabled = 1; } else { @@ -1405,13 +1399,21 @@ void handle_setrateanchori(rtsp_conn_info *conn, rtsp_message *req, rtsp_message conn->ap2_play_enabled = 0; // not sure this is needed yet reset_anchor_info(conn); // needed if the player resumes - release_play_lock(conn); } - debug_mutex_unlock(&conn->flush_mutex, 3); - if ((rate & 1) == 0) + pthread_cleanup_pop(1); // unlock the conn->flush_mutex + + if ((rate & 1) != 0) { + // keep this outside the flush_mutex lock + if (conn->ap2_timing_peer_list_message) { + ptp_send_control_message_string(conn->ap2_timing_peer_list_message); + } else { + debug(1, "Connection %d: No timing peer list!", conn->connection_number); + } + } else { player_full_flush(conn); + } } - plist_free(messagePlist); + pthread_cleanup_pop(1); // plist_free the messagePlist; } else { debug(1, "missing plist!"); } @@ -1945,7 +1947,7 @@ void handle_setup_2(rtsp_conn_info *conn, rtsp_message *req, rtsp_message *resp) case 103: { debug(1, "Connection %d. Buffered Audio Stream Detected.", conn->connection_number); debug_log_rtsp_message(2, "Buffered Audio Stream SETUP incoming message", req); - // for buffered audio, delay getting the play lock until we actually start playing... + get_play_lock(conn); conn->airplay_stream_type = buffered_stream; // get needed stuff @@ -3730,10 +3732,10 @@ void rtsp_conversation_thread_cleanup_function(void *arg) { debug(3, "Delete watchdog mutex."); pthread_mutex_destroy(&conn->watchdog_mutex); - debug(3, "Connection %d: Checking play lock.", conn->connection_number); - release_play_lock(conn); + // debug(3, "Connection %d: Checking play lock.", conn->connection_number); + // release_play_lock(conn); - debug(1, "Connection %d: terminated.", conn->connection_number); + debug(2, "Connection %d: Closed.", conn->connection_number); conn->running = 0; pthread_setcancelstate(oldState, NULL); }