<commit_before>#include "ffmpegdecoder.h"
#include "makeguard.h"

void FFmpegDecoder::parseRunnable()
{
    CHANNEL_LOG(ffmpeg_threads) << "Parse thread started";
    AVPacket packet;
    bool eof = false;

    // detect real framesize
    fixDuration();

    startAudioThread();
    startVideoThread();

    for (;;)
    {
        if (boost::this_thread::interruption_requested())
        {
            return;
        }

        int64_t seekDuration = m_seekDuration.exchange(AV_NOPTS_VALUE);
        if (seekDuration != AV_NOPTS_VALUE)
        {
            if (!resetDecoding(seekDuration, false))
                return;
        }
        seekDuration = m_videoResetDuration.exchange(AV_NOPTS_VALUE);
        if (seekDuration != AV_NOPTS_VALUE)
        {
            if (!resetDecoding(seekDuration, true))
                return;
        }

        const int readStatus = av_read_frame(m_formatContext, &packet);
        if (readStatus >= 0)
        {
            dispatchPacket(packet);
            eof = false;
        }
        else
        {
            if (eof)
            {
                using namespace boost;
                if (m_decoderListener
                    && m_videoPacketsQueue.empty()
                    && m_audioPacketsQueue.empty()
                    && (lock_guard<mutex>(m_videoFramesMutex), !m_videoFramesQueue.canPop()))
                {
                    m_decoderListener->onEndOfStream();
                }

                this_thread::sleep_for(chrono::milliseconds(10));
            }
            eof = readStatus == AVERROR_EOF;
        }

        // Continue packet reading
    }

    CHANNEL_LOG(ffmpeg_threads) << "Decoding ended";
}

void FFmpegDecoder::dispatchPacket(AVPacket& packet)
{
    auto guard = MakeGuard(&packet, av_packet_unref);

    auto seekLambda = [this]
    {
        return m_seekDuration != AV_NOPTS_VALUE || m_videoResetDuration != AV_NOPTS_VALUE;
    };

    if (seekLambda())
    {
        return; // guard frees packet
    }

    if (packet.stream_index == m_videoStreamNumber)
    { 
        if (!m_videoPacketsQueue.push(packet, seekLambda))
        {
            return; // guard frees packet
        }
    }
    else if (packet.stream_index == m_audioStreamNumber)
    { 
        if (!m_audioPacketsQueue.push(packet, seekLambda))
        {
            return; // guard frees packet
        }
    }
    else
    {
        //auto codec = m_formatContext->streams[packet.stream_index]->codec;
        //if (codec->codec_type == AVMEDIA_TYPE_SUBTITLE)
        //{
        //    AVSubtitle subtitle {};
        //    int has_subtitle = 0;
        //    avcodec_decode_subtitle2(codec, &subtitle, &has_subtitle, &packet);
        //    if (has_subtitle)
        //    {
        //        auto format = subtitle.format;
        //    }
        //}

        return; // guard frees packet
    }

    guard.release();
}

void FFmpegDecoder::startAudioThread()
{
    if (m_audioStreamNumber >= 0)
    {
        m_mainAudioThread.reset(new boost::thread(&FFmpegDecoder::audioParseRunnable, this));
    }
}

void FFmpegDecoder::startVideoThread()
{
    if (m_videoStreamNumber >= 0)
    {
        m_mainVideoThread.reset(new boost::thread(&FFmpegDecoder::videoParseRunnable, this));
    }
}

bool FFmpegDecoder::resetDecoding(int64_t seekDuration, bool resetVideo)
{
    CHANNEL_LOG(ffmpeg_seek) << __FUNCTION__ << " resetVideo=" << resetVideo;

    const bool hasVideo = m_mainVideoThread != nullptr;

    if (avformat_seek_file(m_formatContext, 
                           hasVideo ? m_videoStreamNumber : m_audioStreamNumber,
                           0, seekDuration, seekDuration, AVSEEK_FLAG_FRAME) < 0)
    {
        CHANNEL_LOG(ffmpeg_seek) << "Seek failed";
        return false;
    }

    const bool hasAudio = m_mainAudioThread != nullptr;

    if (hasVideo)
    {
        m_mainVideoThread->interrupt();
    }
    if (hasAudio)
    {
        m_mainAudioThread->interrupt();
    }

    if (hasVideo)
    {
        m_mainVideoThread->join();
    }
    if (hasAudio)
    {
        m_mainAudioThread->join();
    }

    // Reset stuff<commit_msg>a fix for .weba files<commit_after>#include "ffmpegdecoder.h"
#include "makeguard.h"

void FFmpegDecoder::parseRunnable()
{
    CHANNEL_LOG(ffmpeg_threads) << "Parse thread started";
    AVPacket packet;
    bool eof = false;

    // detect real framesize
    fixDuration();

    startAudioThread();
    startVideoThread();

    for (;;)
    {
        if (boost::this_thread::interruption_requested())
        {
            return;
        }

        int64_t seekDuration = m_seekDuration.exchange(AV_NOPTS_VALUE);
        if (seekDuration != AV_NOPTS_VALUE)
        {
            if (!resetDecoding(seekDuration, false))
                return;
        }
        seekDuration = m_videoResetDuration.exchange(AV_NOPTS_VALUE);
        if (seekDuration != AV_NOPTS_VALUE)
        {
            if (!resetDecoding(seekDuration, true))
                return;
        }

        const int readStatus = av_read_frame(m_formatContext, &packet);
        if (readStatus >= 0)
        {
            dispatchPacket(packet);
            eof = false;
        }
        else
        {
            if (eof)
            {
                using namespace boost;
                if (m_decoderListener
                    && m_videoPacketsQueue.empty()
                    && m_audioPacketsQueue.empty()
                    && (lock_guard<mutex>(m_videoFramesMutex), !m_videoFramesQueue.canPop()))
                {
                    m_decoderListener->onEndOfStream();
                }

                this_thread::sleep_for(chrono::milliseconds(10));
            }
            eof = readStatus == AVERROR_EOF;
        }

        // Continue packet reading
    }

    CHANNEL_LOG(ffmpeg_threads) << "Decoding ended";
}

void FFmpegDecoder::dispatchPacket(AVPacket& packet)
{
    auto guard = MakeGuard(&packet, av_packet_unref);

    auto seekLambda = [this]
    {
        return m_seekDuration != AV_NOPTS_VALUE || m_videoResetDuration != AV_NOPTS_VALUE;
    };

    if (seekLambda())
    {
        return; // guard frees packet
    }

    if (packet.stream_index == m_videoStreamNumber)
    { 
        if (!m_videoPacketsQueue.push(packet, seekLambda))
        {
            return; // guard frees packet
        }
    }
    else if (packet.stream_index == m_audioStreamNumber)
    { 
        if (!m_audioPacketsQueue.push(packet, seekLambda))
        {
            return; // guard frees packet
        }
    }
    else
    {
        //auto codec = m_formatContext->streams[packet.stream_index]->codec;
        //if (codec->codec_type == AVMEDIA_TYPE_SUBTITLE)
        //{
        //    AVSubtitle subtitle {};
        //    int has_subtitle = 0;
        //    avcodec_decode_subtitle2(codec, &subtitle, &has_subtitle, &packet);
        //    if (has_subtitle)
        //    {
        //        auto format = subtitle.format;
        //    }
        //}

        return; // guard frees packet
    }

    guard.release();
}

void FFmpegDecoder::startAudioThread()
{
    if (m_audioStreamNumber >= 0)
    {
        m_mainAudioThread.reset(new boost::thread(&FFmpegDecoder::audioParseRunnable, this));
    }
}

void FFmpegDecoder::startVideoThread()
{
    if (m_videoStreamNumber >= 0)
    {
        m_mainVideoThread.reset(new boost::thread(&FFmpegDecoder::videoParseRunnable, this));
    }
}

bool FFmpegDecoder::resetDecoding(int64_t seekDuration, bool resetVideo)
{
    CHANNEL_LOG(ffmpeg_seek) << __FUNCTION__ << " resetVideo=" << resetVideo;

    const bool hasVideo = m_mainVideoThread != nullptr;

    if (avformat_seek_file(m_formatContext, 
                           hasVideo ? m_videoStreamNumber : m_audioStreamNumber,
                           0, seekDuration, seekDuration, AVSEEK_FLAG_FRAME) < 0
        && (seekDuration >= 0 || avformat_seek_file(m_formatContext, 
                           hasVideo ? m_videoStreamNumber : m_audioStreamNumber,
                           0, 0, 0, AVSEEK_FLAG_FRAME) < 0))
    {
        CHANNEL_LOG(ffmpeg_seek) << "Seek failed";
        return false;
    }

    const bool hasAudio = m_mainAudioThread != nullptr;

    if (hasVideo)
    {
        m_mainVideoThread->interrupt();
    }
    if (hasAudio)
    {
        m_mainAudioThread->interrupt();
    }

    if (hasVideo)
    {
        m_mainVideoThread->join();
    }
    if (hasAudio)
    {
        m_mainAudioThread->join();
    }

    // Reset stuff
{'fraction_non_alphanumeric': 0.1073072836737103, 'fraction_numerical': 0.003659544327706292, 'mean_word_length': 1.9355509355509355, 'pattern_counts': {'":': 0, '<': 32, '<?xml version=': 0, '>': 28, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '321683', 'n_tokens_mistral': 2640, 'n_tokens_neox': 2516, 'n_words': 434}