X-Git-Url: https://git.sesse.net/?a=blobdiff_plain;f=nageru%2Fmixer.cpp;h=22867422858e0913eb6dcd52fcb2ddac5767b09f;hb=896b8bb8e44409738aef56ba7d7b7f300c22d2ce;hp=2911dae1d19cc553846fbe66b8871b7c4dd6673c;hpb=ecaec75dd52d076ba53cafa1fed716ebc0d93da6;p=nageru diff --git a/nageru/mixer.cpp b/nageru/mixer.cpp index 2911dae..2286742 100644 --- a/nageru/mixer.cpp +++ b/nageru/mixer.cpp @@ -41,6 +41,7 @@ #include "shared/context.h" #include "decklink_capture.h" #include "decklink_output.h" +#include "decklink_util.h" #include "defs.h" #include "shared/disk_space_estimator.h" #include "ffmpeg_capture.h" @@ -231,11 +232,34 @@ void JitterHistory::unregister_metrics(const vector> &label global_metrics.remove("input_estimated_max_jitter_seconds", labels); } -void JitterHistory::frame_arrived(steady_clock::time_point now, int64_t frame_duration, size_t dropped_frames) +void JitterHistory::frame_arrived(steady_clock::time_point now, int64_t frame_duration, size_t dropped_frames, bool verbose) { + if (frame_duration != last_duration) { + // If the frame rate changed, the input clock is also going to change, + // so our historical data doesn't make much sense anymore. + // Also, format changes typically introduce blips that are not representative + // of the typical frame stream. (We make the assumption that format changes + // don't happen all the time in regular use; if they did, we should probably + // rather keep the history so that we take jitter they may introduce into account.) + clear(); + last_duration = frame_duration; + if (verbose) { + fprintf(stderr, "JITTER %p: clearing due to format change\n", this); + } + } if (expected_timestamp > steady_clock::time_point::min()) { expected_timestamp += dropped_frames * nanoseconds(frame_duration * 1000000000 / TIMEBASE); double jitter_seconds = fabs(duration(expected_timestamp - now).count()); + if (verbose) { + fprintf(stderr, "JITTER %p: expected_ts=%.6f (after adding %.1f ms for %zu dropped frames, duration=%ld), now=%.6f => %.1f ms late\n", + this, + duration(expected_timestamp.time_since_epoch()).count(), + 1e3 * dropped_frames * duration(nanoseconds(frame_duration * 1000000000 / TIMEBASE)).count(), + dropped_frames, + frame_duration, + duration(now.time_since_epoch()).count(), + 1e3 * duration(now - expected_timestamp).count()); + } history.push_back(orders.insert(jitter_seconds)); if (jitter_seconds > estimate_max_jitter()) { ++metric_input_underestimated_jitter_frames; @@ -248,6 +272,12 @@ void JitterHistory::frame_arrived(steady_clock::time_point now, int64_t frame_du history.pop_front(); } assert(history.size() <= history_length); + } else if (verbose) { + fprintf(stderr, "JITTER %p: now=%.6f, expected=%.6f duration=%ld [initial]\n", + this, + duration(now.time_since_epoch()).count(), + duration((now + nanoseconds(frame_duration * 1000000000 / TIMEBASE)).time_since_epoch()).count(), + frame_duration); } expected_timestamp = now + nanoseconds(frame_duration * 1000000000 / TIMEBASE); } @@ -276,18 +306,18 @@ void QueueLengthPolicy::unregister_metrics(const vector> &l } void QueueLengthPolicy::update_policy(steady_clock::time_point now, - steady_clock::time_point expected_next_frame, + steady_clock::time_point expected_next_input_frame, int64_t input_frame_duration, int64_t master_frame_duration, double max_input_card_jitter_seconds, - double max_master_card_jitter_seconds) + double max_master_card_jitter_seconds, bool verbose) { double input_frame_duration_seconds = input_frame_duration / double(TIMEBASE); double master_frame_duration_seconds = master_frame_duration / double(TIMEBASE); // Figure out when we can expect the next frame for this card, assuming // worst-case jitter (ie., the frame is maximally late). - double seconds_until_next_frame = max(duration(expected_next_frame - now).count() + max_input_card_jitter_seconds, 0.0); + double seconds_until_next_frame = max(duration(expected_next_input_frame - now).count() + max_input_card_jitter_seconds, 0.0); // How many times are the master card expected to tick in that time? // We assume the master clock has worst-case jitter but not any rate @@ -307,6 +337,13 @@ void QueueLengthPolicy::update_policy(steady_clock::time_point now, } else { frames_allowed = frames_needed; } + if (verbose) { + fprintf(stderr, "secs_until_next_frame = %.1f ms, input jitter = %.1f ms, master jitter = %.1f ms, frames_allowed = %.3f\n", + 1e3 * duration(expected_next_input_frame - now).count(), + 1e3 * max_input_card_jitter_seconds, + 1e3 * max_master_card_jitter_seconds, + frames_allowed); + } safe_queue_length = max(floor(frames_allowed), 0); metric_input_queue_safe_length_frames = safe_queue_length; @@ -410,6 +447,12 @@ Mixer::Mixer(const QSurfaceFormat &format) break; } + if (!decklink_card_is_active(decklink, card_index)) { + fprintf(stderr, "DeckLink card %u is inactive in current profile, skipping (try changing it in Desktop Video Setup)\n", card_index); + decklink->Release(); + continue; + } + DeckLinkCapture *capture = new DeckLinkCapture(decklink, card_index); DeckLinkOutput *output = new DeckLinkOutput(resource_pool.get(), decklink_output_surface, global_flags.width, global_flags.height, card_index); if (!output->set_device(decklink)) { @@ -492,10 +535,6 @@ Mixer::Mixer(const QSurfaceFormat &format) } #endif - for (unsigned card_index = 0; card_index < MAX_VIDEO_CARDS; ++card_index) { - cards[card_index].queue_length_policy.reset(card_index); - } - chroma_subsampler.reset(new ChromaSubsampler(resource_pool.get())); if (global_flags.ten_bit_input) { @@ -574,7 +613,9 @@ void Mixer::configure_card(unsigned card_index, CaptureInterface *capture, CardT CaptureCard *card = &cards[card_index]; if (card->capture != nullptr) { + card_mutex.unlock(); // The dequeue thread could be waiting for bm_frame(). card->capture->stop_dequeue_thread(); + card_mutex.lock(); } card->capture.reset(capture); card->is_fake_capture = (card_type == CardType::FAKE_CAPTURE); @@ -629,9 +670,7 @@ void Mixer::configure_card(unsigned card_index, CaptureInterface *capture, CardT assert(card_type == CardType::FFMPEG_INPUT); } - DeviceSpec device; - device = DeviceSpec{InputSourceType::CAPTURE_CARD, card_index}; - audio_mixer->reset_resampler(device); + DeviceSpec device{InputSourceType::CAPTURE_CARD, card_index}; unsigned num_channels = card_type == CardType::LIVE_CARD ? 8 : 2; if (is_active) { audio_mixer->set_device_parameters(device, card->capture->get_description(), card_type, num_channels, /*active=*/true); @@ -641,6 +680,7 @@ void Mixer::configure_card(unsigned card_index, CaptureInterface *capture, CardT snprintf(name, sizeof(name), "Fake card %u", card_index + 1); audio_mixer->set_device_parameters(device, name, card_type, num_channels, /*active=*/false); } + audio_mixer->reset_resampler(device); audio_mixer->trigger_state_changed_callback(); // Unregister old metrics, if any. @@ -882,10 +922,10 @@ void Mixer::set_output_card_internal(int card_index) card->parked_capture = move(card->capture); CaptureInterface *fake_capture = new FakeCapture(global_flags.width, global_flags.height, FAKE_FPS, OUTPUT_FREQUENCY, card_index, global_flags.fake_cards_audio); configure_card(card_index, fake_capture, CardType::FAKE_CAPTURE, card->output.release(), /*is_srt_card=*/false); - card->queue_length_policy.reset(card_index); + card->jitter_history.clear(); card->capture->start_bm_capture(); desired_output_video_mode = output_video_mode = card->output->pick_video_mode(desired_output_video_mode); - card->output->start_output(desired_output_video_mode, pts_int); + card->output->start_output(desired_output_video_mode, pts_int, /*is_master_card=*/slave_to_output); } output_card_index = card_index; output_jitter_history.clear(); @@ -1272,7 +1312,12 @@ void Mixer::thread_func() DeckLinkOutput *output = cards[output_card_index].output.get(); output->end_output(); desired_output_video_mode = output_video_mode = output->pick_video_mode(desired_output_video_mode); - output->start_output(desired_output_video_mode, pts_int); + output->start_output(desired_output_video_mode, pts_int, /*is_master_card=*/slave_to_output); + } + + { + lock_guard lock(card_mutex); + handle_hotplugged_cards(); } CaptureCard::NewFrame new_frames[MAX_VIDEO_CARDS]; @@ -1280,7 +1325,7 @@ void Mixer::thread_func() bool master_card_is_output; unsigned master_card_index; - if (output_card_index != -1) { + if (output_card_index != -1 && slave_to_output) { master_card_is_output = true; master_card_index = output_card_index; } else { @@ -1289,8 +1334,6 @@ void Mixer::thread_func() assert(master_card_index < MAX_VIDEO_CARDS); } - handle_hotplugged_cards(); - vector raw_audio[MAX_VIDEO_CARDS]; // For MJPEG encoding. OutputFrameInfo output_frame_info = get_one_frame_from_each_card(master_card_index, master_card_is_output, new_frames, has_new_frame, raw_audio); schedule_audio_resampling_tasks(output_frame_info.dropped_frames, output_frame_info.num_samples, output_frame_info.frame_duration, output_frame_info.is_preroll, output_frame_info.frame_timestamp); @@ -1311,7 +1354,9 @@ void Mixer::thread_func() // If the first card is reporting a corrupted or otherwise dropped frame, // just increase the pts (skipping over this frame) and don't try to compute anything new. - if (!master_card_is_output && new_frames[master_card_index].frame->len == 0) { + if (!master_card_is_output && + new_frames[master_card_index].frame != nullptr && // Timeout. + new_frames[master_card_index].frame->len == 0) { ++stats_dropped_frames; pts_int += new_frames[master_card_index].length; continue; @@ -1387,7 +1432,7 @@ void Mixer::thread_func() bool Mixer::input_card_is_master_clock(unsigned card_index, unsigned master_card_index) const { - if (output_card_index != -1) { + if (output_card_index != -1 && slave_to_output) { // The output card (ie., cards[output_card_index].output) is the master clock, // so no input card (ie., cards[card_index].capture) is. return false; @@ -1465,26 +1510,45 @@ pair Mixer::get_channel_color_http(unsigned channel_idx) Mixer::OutputFrameInfo Mixer::get_one_frame_from_each_card(unsigned master_card_index, bool master_card_is_output, CaptureCard::NewFrame new_frames[MAX_VIDEO_CARDS], bool has_new_frame[MAX_VIDEO_CARDS], vector raw_audio[MAX_VIDEO_CARDS]) { OutputFrameInfo output_frame_info; + constexpr steady_clock::duration master_card_timeout = milliseconds(200); start: unique_lock lock(card_mutex, defer_lock); + bool timed_out = false; if (master_card_is_output) { // Clocked to the output, so wait for it to be ready for the next frame. cards[master_card_index].output->wait_for_frame(pts_int, &output_frame_info.dropped_frames, &output_frame_info.frame_duration, &output_frame_info.is_preroll, &output_frame_info.frame_timestamp); lock.lock(); } else { // Wait for the master card to have a new frame. - // TODO: Add a timeout. output_frame_info.is_preroll = false; lock.lock(); - cards[master_card_index].new_frames_changed.wait(lock, [this, master_card_index]{ return !cards[master_card_index].new_frames.empty() || cards[master_card_index].capture->get_disconnected(); }); - } - - if (master_card_is_output) { + timed_out = !cards[master_card_index].new_frames_changed.wait_for(lock, + master_card_timeout, + [this, master_card_index] { + return !cards[master_card_index].new_frames.empty() || + cards[master_card_index].capture == nullptr || + cards[master_card_index].capture->get_disconnected(); + }); + if (timed_out) { + fprintf(stderr, "WARNING: Master card (%s) did not deliver a frame for %u ms, creating a fake one.\n", + description_for_card(master_card_index).c_str(), + unsigned(duration_cast(master_card_timeout).count())); + } + } + + if (timed_out) { + // The master card stalled for 200 ms (possible when it's e.g. + // an SRT card). Send a frame no matter what; this also makes sure + // any other cards get to empty their queues, and in general, + // that we make _some_ sort of forward progress. + handle_hotplugged_cards(); + } else if (master_card_is_output) { handle_hotplugged_cards(); } else if (cards[master_card_index].new_frames.empty()) { // We were woken up, but not due to a new frame. Deal with it // and then restart. - assert(cards[master_card_index].capture->get_disconnected()); + assert(cards[master_card_index].capture == nullptr || + cards[master_card_index].capture->get_disconnected()); handle_hotplugged_cards(); lock.unlock(); goto start; @@ -1514,7 +1578,13 @@ start: raw_audio[card_index] = move(card->new_raw_audio); } - if (!master_card_is_output) { + if (timed_out) { + // Pretend the frame happened a while ago and was only processed now, + // so that we get the duration sort-of right. This isn't ideal. + output_frame_info.dropped_frames = 0; // Hard to define, really. + output_frame_info.frame_duration = lrint(TIMEBASE * duration(master_card_timeout).count()); + output_frame_info.frame_timestamp = steady_clock::now() - master_card_timeout; + } else if (!master_card_is_output) { output_frame_info.frame_timestamp = new_frames[master_card_index].received_timestamp; output_frame_info.dropped_frames = new_frames[master_card_index].dropped_frames; output_frame_info.frame_duration = new_frames[master_card_index].length; @@ -1548,6 +1618,14 @@ start: fractional_samples = num_samples_times_timebase % TIMEBASE; assert(output_frame_info.num_samples >= 0); + if (timed_out) { + DeviceSpec device{InputSourceType::CAPTURE_CARD, master_card_index}; + bool success; + do { + success = audio_mixer->add_silence(device, output_frame_info.num_samples, /*dropped_frames=*/0); + } while (!success); + } + return output_frame_info; } @@ -1557,11 +1635,23 @@ void Mixer::handle_hotplugged_cards() for (unsigned card_index = 0; card_index < MAX_VIDEO_CARDS; ++card_index) { CaptureCard *card = &cards[card_index]; if (card->capture != nullptr && card->capture->get_disconnected()) { - fprintf(stderr, "Card %u went away, replacing with a fake card.\n", card_index); - FakeCapture *capture = new FakeCapture(global_flags.width, global_flags.height, FAKE_FPS, OUTPUT_FREQUENCY, card_index, global_flags.fake_cards_audio); - configure_card(card_index, capture, CardType::FAKE_CAPTURE, /*output=*/nullptr, /*is_srt_card=*/false); - card->queue_length_policy.reset(card_index); - card->capture->start_bm_capture(); + bool is_active = card_index < unsigned(global_flags.min_num_cards) || cards[card_index].force_active; + if (is_active) { + fprintf(stderr, "Card %u went away, replacing with a fake card.\n", card_index); + FakeCapture *capture = new FakeCapture(global_flags.width, global_flags.height, FAKE_FPS, OUTPUT_FREQUENCY, card_index, global_flags.fake_cards_audio); + configure_card(card_index, capture, CardType::FAKE_CAPTURE, /*output=*/nullptr, /*is_srt_card=*/false); + card->jitter_history.clear(); + card->capture->start_bm_capture(); + } else { + // NOTE: The theme might end up forcing the card back at some later point + // (ie., force_active is false now, but might immediately be true again on + // e.g. the next frame). That should be rare, though, so we don't bother + // adjusting the message. + fprintf(stderr, "Card %u went away, removing. (To keep a fake card, increase --num-cards.)\n", card_index); + theme->remove_card(card_index); + configure_card(card_index, /*capture=*/nullptr, CardType::FAKE_CAPTURE, /*output=*/nullptr, /*is_srt_card=*/false); + card->jitter_history.clear(); + } } } @@ -1607,7 +1697,7 @@ void Mixer::handle_hotplugged_cards() CaptureCard *card = &cards[free_card_index]; BMUSBCapture *capture = new BMUSBCapture(free_card_index, new_dev); configure_card(free_card_index, capture, CardType::LIVE_CARD, /*output=*/nullptr, /*is_srt_card=*/false); - card->queue_length_policy.reset(free_card_index); + card->jitter_history.clear(); capture->set_card_disconnected_callback(bind(&Mixer::bm_hotplug_remove, this, free_card_index)); capture->start_bm_capture(); } @@ -1615,7 +1705,6 @@ void Mixer::handle_hotplugged_cards() #ifdef HAVE_SRT // Same, for SRT inputs. - // TODO: On disconnect and reconnect, we might want to use the stream ID for (SRTSOCKET sock : hotplugged_srt_cards_copy) { char name[256]; int namelen = sizeof(name); @@ -1664,7 +1753,7 @@ void Mixer::handle_hotplugged_cards() configure_card(free_card_index, capture, CardType::FFMPEG_INPUT, /*output=*/nullptr, /*is_srt_card=*/true); update_srt_stats(sock, card); // Initial zero stats. card->last_srt_stream_id = stream_id; - card->queue_length_policy.reset(free_card_index); + card->jitter_history.clear(); capture->set_card_disconnected_callback(bind(&Mixer::bm_hotplug_remove, this, free_card_index)); capture->start_bm_capture(); } @@ -1677,7 +1766,7 @@ void Mixer::handle_hotplugged_cards() if (card->capture == nullptr && card->force_active) { FakeCapture *capture = new FakeCapture(global_flags.width, global_flags.height, FAKE_FPS, OUTPUT_FREQUENCY, card_index, global_flags.fake_cards_audio); configure_card(card_index, capture, CardType::FAKE_CAPTURE, /*output=*/nullptr, /*is_srt_card=*/false); - card->queue_length_policy.reset(card_index); + card->jitter_history.clear(); card->capture->start_bm_capture(); } } @@ -1923,7 +2012,7 @@ void Mixer::quit() if (global_flags.srt_port >= 0) { // There's seemingly no other reasonable way to wake up the thread // (libsrt's epoll equivalent is busy-waiting). - int sock = srt_socket(AF_INET6, 0, 0); + int sock = srt_create_socket(); if (sock != -1) { sockaddr_in6 addr; memset(&addr, 0, sizeof(addr)); @@ -2132,7 +2221,7 @@ void Mixer::OutputChannel::set_color_updated_callback(Mixer::color_updated_callb #ifdef HAVE_SRT void Mixer::start_srt() { - SRTSOCKET sock = srt_socket(AF_INET6, 0, 0); + SRTSOCKET sock = srt_create_socket(); sockaddr_in6 addr; memset(&addr, 0, sizeof(addr)); addr.sin6_family = AF_INET6; @@ -2160,6 +2249,12 @@ void Mixer::start_srt() } break; } + if (!global_flags.enable_srt) { // Runtime UI toggle. + // Perhaps not as good as never listening in the first place, + // but much simpler to turn on and off. + srt_close(clientsock); + continue; + } lock_guard lock(hotplug_mutex); hotplugged_srt_cards.push_back(clientsock); }