<commit_before>#include <libavcodec/avcodec.h>
#include <libavformat/avformat.h>
#include <libavutil/imgutils.h>
#include <libswscale/swscale.h>
#include <signal.h>

// Hacky macros to disable alarm signals temporarily.
// FFmpeg doesn't handle syscall interruptions properly,
// and GHC runtime uses alarms a lot.
#define BEGIN_DISABLE_ALARMS() \
	sigset_t disableAlarmsMask, oldMask; \
	sigemptyset(&disableAlarmsMask); \
	sigaddset(&disableAlarmsMask, SIGALRM); \
	sigaddset(&disableAlarmsMask, SIGVTALRM); \
	pthread_sigmask(SIG_BLOCK, &disableAlarmsMask, &oldMask)
#define END_DISABLE_ALARMS() \
	pthread_sigmask(SIG_SETMASK, &oldMask, NULL)

void flaw_ffmpeg_init()
{
	av_register_all();
	avformat_network_init();
}

AVFormatContext* flaw_ffmpeg_openInput(const char* url)
{
	AVFormatContext* ctx = NULL;
	BEGIN_DISABLE_ALARMS();
	int r = avformat_open_input(&ctx, url, NULL, NULL);
	END_DISABLE_ALARMS();
	if(r == 0) return ctx;
	avformat_free_context(ctx);
	return NULL;
}

int flaw_ffmpeg_getStreamsCount(AVFormatContext* ctx)
{
	return ctx->nb_streams;
}

void flaw_ffmpeg_getStreamsTypes(AVFormatContext* ctx, int* outStreamsTypes)
{
	for(int i = 0; i < ctx->nb_streams; ++i)
	{
		outStreamsTypes[i] = ctx->streams[i]->codec->codec_type;
	}
}

// Decode streams using decode callback provided.
// If callback returns non-zero value, decoding is stopped and that value is returned.
int flaw_ffmpeg_decode(AVFormatContext* ctx, int streamsMask, int (*callback)(AVFrame*, void*), void* callbackOpaque)
{
	// open decoders for streams needed
	for(int i = 0; i < ctx->nb_streams; ++i)
	{
		if(streamsMask & (1 << i))
		{
			AVCodecContext* codec = ctx->streams[i]->codec;
			AVCodec* decoder = avcodec_find_decoder(codec->codec_id);
			int ret = decoder ? avcodec_open2(codec, decoder, NULL) : -1;
			if(ret < 0)
			{
				// close already opened decoders
				for(int j = 0; j < i; ++j)
				{
					if(streamsMask & (1 << j))
					{
						avcodec_close(ctx->streams[j]->codec);
					}
				}
				return ret;
			}
		}
	}

	// init frame
	AVFrame* frame = av_frame_alloc();

	// init packet struct
	AVPacket pkt;
	av_init_packet(&pkt);
	pkt.data = NULL;
	pkt.size = 0;

	// loop reading packets
	int err = 0, gotPacket;
	do
	{
		gotPacket = av_read_frame(ctx, &pkt) >= 0;
		if(!gotPacket)
		{
			pkt.data = NULL;
			pkt.size = 0;
		}
		if(streamsMask & (1 << pkt.stream_index))
		{
			AVCodecContext* codec = ctx->streams[pkt.stream_index]->codec;
			AVPacket origPkt = pkt;
			// loop decoding frames
			int gotFrame;
			do
			{
				// decode frame
				gotFrame = 0;
				int decoded = 0;
				switch(codec->codec_type)
				{
				case AVMEDIA_TYPE_VIDEO:
					decoded = avcodec_decode_video2(codec, frame, &gotFrame, &pkt);
					break;
				case AVMEDIA_TYPE_AUDIO:
					decoded = avcodec_decode_audio4(codec, frame, &gotFrame, &pkt);
					break;
				case AVMEDIA_TYPE_DATA:
					break;
				case AVMEDIA_TYPE_SUBTITLE:
					break;
				case AVMEDIA_TYPE_ATTACHMENT:
					break;
				}
				if(decoded < 0)
				{
					err = decoded;
				}
				else
				{
					// output frame
					if(gotFrame)
					{
						int ret = callback(frame, callbackOpaque);
						if(ret != 0)
						{
							err = ret;
						}
					}
					// advance packet
					if(gotPacket)
					{
						pkt.data += decoded;
						pkt.size -= decoded;
					}
				}
			}
			while(err >= 0 && (gotPacket ? pkt.size > 0 : gotFrame));
		}
		// free packet
		av_packet_unref(&pkt);
	}
	while(err >= 0 && gotPacket);

	// close frame
	av_frame_free(&frame);

	// close decoders
	for(int i = 0; i < ctx->nb_streams; ++i)
	{
		if(streamsMask & (1 << i))
		{
			avcodec_close(ctx->streams[i]->codec);
		}
	}

	return err;
}

struct ChainedDecodeCallback
{
	void* callback;
	void* opaque;
};
<commit_msg>refactor ffmpeg again, add separate demux stage<commit_after>#include <libavcodec/avcodec.h>
#include <libavformat/avformat.h>
#include <libavutil/imgutils.h>
#include <libswscale/swscale.h>
#include <signal.h>

// Hacky macros to disable alarm signals temporarily.
// FFmpeg doesn't handle syscall interruptions properly,
// and GHC runtime uses alarms a lot.
#define BEGIN_DISABLE_ALARMS() \
	sigset_t disableAlarmsMask, oldMask; \
	sigemptyset(&disableAlarmsMask); \
	sigaddset(&disableAlarmsMask, SIGALRM); \
	sigaddset(&disableAlarmsMask, SIGVTALRM); \
	pthread_sigmask(SIG_BLOCK, &disableAlarmsMask, &oldMask)
#define END_DISABLE_ALARMS() \
	pthread_sigmask(SIG_SETMASK, &oldMask, NULL)

void flaw_ffmpeg_init()
{
	av_register_all();
	avformat_network_init();
}

AVFormatContext* flaw_ffmpeg_openInput(const char* url)
{
	AVFormatContext* ctx = NULL;
	BEGIN_DISABLE_ALARMS();
	int r = avformat_open_input(&ctx, url, NULL, NULL);
	END_DISABLE_ALARMS();
	if(r == 0) return ctx;
	avformat_free_context(ctx);
	return NULL;
}

int flaw_ffmpeg_getStreamsCount(AVFormatContext* ctx)
{
	return ctx->nb_streams;
}

void flaw_ffmpeg_getStreamsTypes(AVFormatContext* ctx, int* outStreamsTypes)
{
	for(int i = 0; i < ctx->nb_streams; ++i)
	{
		outStreamsTypes[i] = ctx->streams[i]->codec->codec_type;
	}
}

int flaw_ffmpeg_demux(AVFormatContext* ctx, int (*callback)(AVPacket*, void*), void* opaque)
{
	// init packet struct
	AVPacket pkt;
	av_init_packet(&pkt);
	pkt.data = NULL;
	pkt.size = 0;

	// loop reading packets
	int err = 0, gotPacket;
	do
	{
		gotPacket = av_read_frame(ctx, &pkt) >= 0;
		if(!gotPacket)
		{
			pkt.data = NULL;
			pkt.size = 0;
		}

		AVPacket origPkt = pkt;

		int ret = callback(&pkt, opaque);
		if(ret != 0)
		{
			err = ret;
		}

		// free packet
		av_packet_unref(&origPkt);
	}
	while(err == 0 && gotPacket);

	return err;
}

struct DecodeState
{
	AVFormatContext* ctx;
	AVFrame* frame;
	int streamsMask;
	int (*callback)(AVFrame*, void*);
	void* callbackOpaque;
};

static int decodeCallback(AVPacket* pkt, void* opaque)
{
	struct DecodeState* state = (struct DecodeState*)opaque;

	// skip streams we don't care about
	if(!(state->streamsMask & (1 << pkt->stream_index))) return 0;

	AVCodecContext* codec = state->ctx->streams[pkt->stream_index]->codec;
	// loop decoding frames
	int gotFrame, err = 0;
	do
	{
		// decode frame
		gotFrame = 0;
		int decoded = 0;
		switch(codec->codec_type)
		{
		case AVMEDIA_TYPE_VIDEO:
			decoded = avcodec_decode_video2(codec, state->frame, &gotFrame, pkt);
			break;
		case AVMEDIA_TYPE_AUDIO:
			decoded = avcodec_decode_audio4(codec, state->frame, &gotFrame, pkt);
			break;
		case AVMEDIA_TYPE_DATA:
			break;
		case AVMEDIA_TYPE_SUBTITLE:
			break;
		case AVMEDIA_TYPE_ATTACHMENT:
			break;
		}
		if(decoded < 0)
		{
			err = decoded;
		}
		else
		{
			// output frame
			if(gotFrame)
			{
				int ret = state->callback(state->frame, state->callbackOpaque);
				if(ret != 0)
				{
					err = ret;
				}
			}
			// advance packet
			if(pkt->data)
			{
				pkt->data += decoded;
				pkt->size -= decoded;
			}
		}
	}
	while(err == 0 && (pkt->data ? pkt->size > 0 : gotFrame));

	return err;
}

// Decode streams using decode callback provided.
// If callback returns non-zero value, decoding is stopped and that value is returned.
int flaw_ffmpeg_decode(AVFormatContext* ctx, int streamsMask, int (*callback)(AVFrame*, void*), void* callbackOpaque)
{
	// open decoders for streams needed
	for(int i = 0; i < ctx->nb_streams; ++i)
	{
		if(streamsMask & (1 << i))
		{
			AVCodecContext* codec = ctx->streams[i]->codec;
			AVCodec* decoder = avcodec_find_decoder(codec->codec_id);
			int ret = decoder ? avcodec_open2(codec, decoder, NULL) : -1;
			if(ret < 0)
			{
				// close already opened decoders
				for(int j = 0; j < i; ++j)
				{
					if(streamsMask & (1 << j))
					{
						avcodec_close(ctx->streams[j]->codec);
					}
				}
				return ret;
			}
		}
	}

	// init state
	struct DecodeState state;
	state.ctx = ctx;
	state.frame = av_frame_alloc();
	state.streamsMask = streamsMask;
	state.callback = callback;
	state.callbackOpaque = callbackOpaque;

	int err = flaw_ffmpeg_demux(ctx, decodeCallback, &state);

	// close frame
	av_frame_free(&state.frame);

	// close decoders
	for(int i = 0; i < ctx->nb_streams; ++i)
	{
		if(streamsMask & (1 << i))
		{
			avcodec_close(ctx->streams[i]->codec);
		}
	}

	return err;
}

struct ChainedDecodeCallback
{
	void* callback;
	void* opaque;
};

{'fraction_non_alphanumeric': 0.13390139987827146, 'fraction_numerical': 0.0064516129032258064, 'mean_word_length': 3.529217199558986, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 67, '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': '3803331', 'n_tokens_mistral': 3542, 'n_tokens_neox': 3120, 'n_words': 683}