#include <windows.h>
#include <time.h>
#include <process.h>
#include <stdio.h>

#define BUF_COUNT 128
#define BUF_SIZE 1024 * 8

// Создание файла
void create_file(const char* filename, int size) {
	FILE* file = fopen(filename, "rb");   
	if (file != NULL) { 
		// Файл уже есть
		printf("Use file %s\n", filename);
		fclose(file);
		return;
	}
	printf("Create file %s size %d ...", filename, size);
	file = fopen(filename, "wb");   
	if (file != NULL) { 
		for(int i = 0; i < size; i++) {
			fprintf(file, "%d\n", rand());
		}
		fclose(file);
	}
}

// Замер скорости чтения
void read_speed(const char* filename) {
	int count = 0;
	clock_t start = clock();
	FILE* file = fopen(filename, "rb");   
	if (file != NULL) { 
		int n = 0;
		bool store = false;
		char buffer[BUF_SIZE];
		size_t bytesRead = 0;
		while ((bytesRead = fread(buffer, 1, sizeof(buffer), file)) > 0) {
			count++;
		}
		fclose(file);
	}
	printf("ONLY READ: Count %d blocks Time %f sec.\n", count, (double)(clock() - start) / CLOCKS_PER_SEC);
}

// Однопоточный вариант
void read_ints(const char* filename) {
	clock_t start = clock();
	int count = 0;
	int check = 0;
	FILE* file = fopen(filename, "rb");   
	if (file != NULL) { 
		int n = 0;
		bool minus = false;
		bool store = false;
		bool overflow = false;
		char buffer[1024 * 8];
		size_t bytesRead = 0;
		while ((bytesRead = fread(buffer, 1, sizeof(buffer), file)) > 0) {
			char* end = buffer + bytesRead;
			for(char* cur = buffer; cur < end; cur++) {
				// считаем что все числа целые и разделены любыми разделителями
				if(*cur >= '0' && *cur <= '9') {
					int k = n * 10 + *cur - '0';
					if(!overflow) overflow = k < n;
					n = k;
					store = true;
				} else {
					if(store) {
						// тут сохраняем результат
						if(overflow) {
							printf("overflow\n");// переполнение
							overflow = false;
						} else {
							if(minus) {
								n = -n;
								minus = false;
							}
							check ^= n;
							//printf("%d\n", n);
							count++;
						}
						store = false;
						n = 0;
					}
					minus = (*cur == '-');
				}
			}
		}
		fclose(file);
	}
	printf("Read %d numbers CheckSum %X Time %f sec.\n", count, check, (double)(clock() - start) / CLOCKS_PER_SEC);
}

//Синхронный printf() из разных потоков
void mt_printf(const char* text, int value = -1) {
	static CRITICAL_SECTION cs;
	volatile static bool init = false;
	if(!init) {
		init = true;
		InitializeCriticalSectionAndSpinCount(&cs, 4000);
	}
	EnterCriticalSection(&cs);
	if(value == -1) {
		printf("%05d: %s\n", clock(), text);
	} else {
		printf("%05d: %s %d\n", clock(), text, value);
	}
	LeaveCriticalSection(&cs);
}

// Буфер
typedef struct {
	volatile long size;
	char data[BUF_SIZE];
} buf_t;

// Параметры потока чтения
typedef struct {
	const char* filename;
	HANDLE wait; // Event для ожидания
	volatile long wait_count; // Количество ожидающих
	CRITICAL_SECTION cs;
	buf_t buf[BUF_COUNT];
	volatile long buf_count; // Количество заполненных буферов
} param_t;


// Ожидание в пробуждением другого потока
void wait_one(param_t* param, HANDLE thread = NULL) {
	// Возобновление другого потока
	EnterCriticalSection(&param->cs);
	if(param->wait_count != 0) {
		SetEvent(param->wait); 
		Sleep(0);
		while(param->wait_count != 0) {
			//mt_printf("Warning: Sleep(1)");
			Sleep(1);
		}
	}
	// Ожидание
	ResetEvent(param->wait); 
	if(param->wait_count != 0) {
		mt_printf("Error: wake up");
		LeaveCriticalSection(&param->cs);
	} else {
		InterlockedExchangeAdd(&param->wait_count, 1);
		LeaveCriticalSection(&param->cs);
		if(thread) { // Дополнительно ожидать завершение потока
			HANDLE wait[2];
			wait[0] = param->wait;
			wait[1] = thread;
			if(WaitForMultipleObjects(2, wait, FALSE, 1000) == WAIT_TIMEOUT) {
				mt_printf("Error: wait timeout");
			}
		} else if(WaitForSingleObject(param->wait, 1000) == WAIT_TIMEOUT) {
			mt_printf("Error: wait timeout");
		}
		InterlockedExchangeAdd(&param->wait_count, -1);
	}
}

// Пробуждение спящих
void wake_up(param_t* param) {
	if(param->wait_count != 0 && param->buf_count > BUF_COUNT / 2) {
		EnterCriticalSection(&param->cs);
		if(param->wait_count != 0) {
			SetEvent(param->wait); 
			Sleep(0);
			while(param->wait_count != 0) {
				//mt_printf("Warning: Sleep(1)");
				Sleep(1);
			}
		}
		LeaveCriticalSection(&param->cs);
	}
}
// Поток чтения
unsigned __stdcall read_thread(void* par) {
	int count = 0; // счетчик запусков ожидания разбора
	param_t* param = (param_t*) par;
	FILE* file = fopen(param->filename, "rb");   
	if (file != NULL) { 
		long cur_buf = 0;
		size_t read = 0;
		// Чтение в buf[cur_buf]
		while ((read = fread(param->buf[cur_buf].data, 1, BUF_SIZE, file)) > 0) {
			// Установка размера прочитанного
			if(InterlockedExchange(&param->buf[cur_buf].size, read) != 0) {
				mt_printf("Error: Bufer not empty (1)");
				break;
			}
			#ifdef _DEBUG
			mt_printf("read end", cur_buf);
			#endif
			InterlockedExchangeAdd(&param->buf_count, 1);
			wake_up(param); // пробуждение разбора
			// следующий буфер
			cur_buf++;
			if(cur_buf == BUF_COUNT) cur_buf = 0;

			// Проверка что буфер пустой
			if(param->buf[cur_buf].size != 0) {
				// Ожидание пустого буфера
				#ifdef _DEBUG
				mt_printf("read wait", cur_buf);
				#endif
				wait_one(param);
				count++;
			}
			#ifdef _DEBUG
			mt_printf("read start", cur_buf);
			#endif
		}
		if(count > 0) mt_printf("read wait times", count);
	}
	//if(param->wait_count != 0) SetEvent(param->wait);
	_endthreadex(0);
	return 0;
}

void read_ints_mt(const char* filename) {
	clock_t start = clock();
	// Подготовка и запуск потока чтения
	param_t* param = (param_t*) calloc(1, sizeof(param_t));
	if(!param) {
		mt_printf("Error: no memory");
		return;
	}
	param->filename = filename;
	param->wait = CreateEvent(NULL, FALSE, FALSE, NULL);
	InitializeCriticalSection(&param->cs);
    unsigned th_id;
    HANDLE th = (HANDLE)_beginthreadex(NULL, 0, &read_thread, param, 0, &th_id );
	int wait_count = 0;
	// Разбор
	int n = 0;
	bool minus = false;
	bool store = false;
	bool overflow = false;
	int check = 0;
	int numbers_count = 0;
	size_t cur_buf = 0;
	buf_t* buf = NULL;
	while(1) {
		if(param->buf[cur_buf].size == 0) { // Нечего разбирать
			if(WaitForSingleObject(th, 0) != WAIT_TIMEOUT) break; // Поток приема завершен
			// Ожидание чтения
			#ifdef _DEBUG
			mt_printf("parse wait", cur_buf);
			#endif
			wait_count++;
			wait_one(param, th);
		}
		#ifdef _DEBUG
		mt_printf("parse start", cur_buf);
		#endif
		// Разбор буфера
		char* end = param->buf[cur_buf].data + param->buf[cur_buf].size;
		for(char* cur = param->buf[cur_buf].data; cur < end; cur++) {
			// считаем что все числа целые и разделены любыми разделителями
			if(*cur >= '0' && *cur <= '9') {
				int k = n * 10 + *cur - '0';
				if(!overflow) overflow = k < n;
				n = k;
				store = true;
			} else {
				if(store) {
					// тут сохраняем результат
					if(overflow) {
						mt_printf("overflow");// переполнение
						overflow = false;
					} else {
						if(minus) {
							n = -n;
							minus = false;
						}
						check ^= n;
						//printf("%d\n", n);
						numbers_count++;
					}
					store = false;
					n = 0;
				}
				minus = (*cur == '-');
			}
		}
		// Пометка буфера разобранным
		if(InterlockedExchange(&(param->buf[cur_buf].size), 0) == 0) {
			if(WaitForSingleObject(th, 0) != WAIT_TIMEOUT) break; // Поток приема завершен
			mt_printf("Error: size == 0 (2)");
			break;
		}
		InterlockedExchangeAdd(&param->buf_count, -1);
		#ifdef _DEBUG
		mt_printf("parse end", cur_buf);
		#endif
		// Переход на следующий буфер
		cur_buf++;
		if(cur_buf == BUF_COUNT) cur_buf = 0;
	}
	WaitForSingleObject(th, INFINITE);
	printf("MT Read %d numbers CheckSum %X Wait %d times Time %f sec.\n", numbers_count, check, wait_count, (double)(clock() - start) / CLOCKS_PER_SEC);
	CloseHandle(th);
	CloseHandle(param->wait);
	DeleteCriticalSection(&param->cs);
	free(param);

}


int main(int argc, char* argv[])
{
	char file[] = "big.txt";
	create_file(file, 1000000000);
	//char file[] = "mini.txt";
	//create_file(file, 100000000);
	read_speed(file);
	read_ints(file);
	read_ints_mt(file);

	return 0;
}

Use file big.txt
ONLY READ: Count 86380 blocks Time 57.236000 sec.
Read 1000000000 numbers CheckSum 4231 Time 57.205000 sec.
MT Read 1000000000 numbers CheckSum 4231 Wait 680 times Time 57.034000 sec.

Use file big.txt
ONLY READ: Count 86380 blocks Time 58.854000 sec.
Read 1000000000 numbers CheckSum 4231 Time 57.595000 sec.
MT Read 1000000000 numbers CheckSum 4231 Wait 701 times Time 58.828000 sec.

Use file big.txt
ONLY READ: Count 86380 blocks Time 64.600000 sec.
Read 1000000000 numbers CheckSum 4231 Time 60.949000 sec.
MT Read 1000000000 numbers CheckSum 4231 Wait 736 times Time 54.694000 sec.

#include <windows.h>
#include <time.h>
#include <process.h>
#include <stdio.h>

#define BUF_COUNT 128
#define BUF_SIZE 1024 * 64

// Создание файла
void create_file(const char* filename, int size) {
	FILE* file = fopen(filename, "rb");   
	if (file != NULL) { 
		// Файл уже есть
		printf("Use file %s\n", filename);
		fclose(file);
		return;
	}
	printf("Create file %s size %d ...\n", filename, size);
	file = fopen(filename, "wb");   
	if (file != NULL) { 
		for(int i = 0; i < size; i++) {
			fprintf(file, "%d\n", rand());
		}
		fclose(file);
	}
}

// Замер скорости чтения
void read_speed(const char* filename) {
	int count = 0;
	clock_t start = clock();
	FILE* file = fopen(filename, "rb");   
	if (file != NULL) { 
		int n = 0;
		bool store = false;
		char buffer[BUF_SIZE];
		size_t bytesRead = 0;
		while ((bytesRead = fread(buffer, 1, sizeof(buffer), file)) > 0) {
			count++;
		}
		fclose(file);
	}
	printf("ONLY READ: Count %d blocks Time %f sec.\n", count, (double)(clock() - start) / CLOCKS_PER_SEC);
}

// Блочный парсер
class parser {
	int n;
	bool minus;
	bool store;
	bool overflow;
public:
	int count;
	int check;

	parser() {
		n = 0;
		count = 0;
		check = 0;
		minus = false;
		store = false;
		overflow = false;
	}

	void parse_block(const char* data, int size) {
		const char* end = data + size;
		for(const char* cur = data; cur < end; cur++) {
			// считаем что все числа целые и разделены любыми разделителями
			if(*cur >= '0' && *cur <= '9') {
				int k = n * 10 + *cur - '0';
				if(!overflow) overflow = k < n;
				n = k;
				store = true;
			} else {
				if(store) {
					// тут сохраняем результат
					if(overflow) {
						printf("overflow\n");// переполнение
						overflow = false;
					} else {
						if(minus) {
							n = -n;
							minus = false;
						}
						check ^= n;
						//printf("%d\n", n);
						count++;
					}
					store = false;
					n = 0;
				}
				minus = (*cur == '-');
			}
		}
	}
};

// Однопоточный вариант
void read_ints(const char* filename) {
	clock_t start = clock();
	FILE* file = fopen(filename, "rb");   
	parser p;
	if (file != NULL) { 
		char buffer[BUF_SIZE];
		size_t bytesRead = 0;
		while ((bytesRead = fread(buffer, 1, sizeof(buffer), file)) > 0) p.parse_block(buffer, bytesRead);
		fclose(file);
	}
	printf("Read %d numbers CheckSum %X Time %f sec.\n", p.count, p.check, (double)(clock() - start) / CLOCKS_PER_SEC);
}

// printf()
void mt_printf(const char* text, int value = -1) {
	if(value == -1) {
		printf("%05d: %s\n", clock(), text);
	} else {
		printf("%05d: %s %d\n", clock(), text, value);
	}
}

// Буфер
typedef struct {
	volatile long size;
	char data[BUF_SIZE];
} buf_t;

// Параметры потока чтения
typedef struct {
	const char* filename;
	HANDLE read_wait; // Event для ожидания чтения
	HANDLE parse_wait; // Event для ожидания разбора
	buf_t buf[BUF_COUNT];
	volatile long buf_count; // Количество заполненных буферов
	volatile long read_wait_count; // Счетчик засыпаний чтения
	volatile long read_result; // Результат завершения чтения
} param_t;

// Поток чтения
unsigned __stdcall read_thread(void* par) {
	param_t* param = (param_t*) par;
	FILE* file = fopen(param->filename, "rb");   
	if (file != NULL) { 
		long cur_buf = 0;
		size_t read = 0;
		while (1) {
			// Проверка что буфер пустой
			if(param->buf[cur_buf].size != 0) {
				// Ожидание пустого буфера
				if(WaitForSingleObject(param->read_wait, 1000) == WAIT_TIMEOUT) {
					mt_printf("Error: read wait timeout");
					break;
				}
				if(param->buf[cur_buf].size != 0) continue;
				param->read_wait_count++;
			}
			// Чтение в buf[cur_buf]
			size_t read = fread(param->buf[cur_buf].data, 1, BUF_SIZE, file);
			if(read == 0) { // Файл прочитан
				param->read_result = 1;
				break;
			}
			// Установка размера прочитанного
			if(InterlockedExchange(&param->buf[cur_buf].size, read) != 0) {
				mt_printf("Error: Bufer not empty (1)");
				break;
			}
			// пробуждение разбора если занято более половины буферов
			if(InterlockedExchangeAdd(&param->buf_count, 1) >= BUF_COUNT/2) {
				SetEvent(param->parse_wait);
			}
			// следующий буфер
			cur_buf++;
			if(cur_buf == BUF_COUNT) cur_buf = 0;
		}
	}
	_endthreadex(0);
	return 0;
}

void read_ints_mt(const char* filename) {
	clock_t start = clock();
	parser p;
	// Подготовка и запуск потока чтения
	param_t* param = (param_t*) calloc(1, sizeof(param_t));
	if(!param) {
		mt_printf("Error: no memory");
		return;
	}
	param->filename = filename;
	param->read_wait = CreateEvent(NULL, FALSE, FALSE, NULL);
	param->parse_wait = CreateEvent(NULL, FALSE, FALSE, NULL);
    unsigned th_id;
	HANDLE waits[2];
	waits[0] = param->parse_wait;
    waits[1] = (HANDLE)_beginthreadex(NULL, 0, &read_thread, param, 0, &th_id );
	int wait_count = 0;
	// Разбор
	size_t cur_buf = 0;
	buf_t* buf = NULL;
	while(1) {
		if(param->buf[cur_buf].size == 0) { 
			// Нечего разбирать
			if(WaitForSingleObject(waits[1], 0) != WAIT_TIMEOUT) break; // Поток приема завершен
			// Ожидание чтения
			if(WaitForMultipleObjects(2, waits, FALSE, 1000) == WAIT_TIMEOUT) {
				mt_printf("Error: parse wait timeout");
				break;
			}
			if(param->buf[cur_buf].size == 0) continue;
			wait_count++;
		}
		// Разбор буфера
		p.parse_block(param->buf[cur_buf].data, param->buf[cur_buf].size);
		// Пометка буфера разобранным
		if(InterlockedExchange(&(param->buf[cur_buf].size), 0) == 0) {
			mt_printf("Error: size == 0 (2)");
			break;
		}
		// пробуждение чтения если занято менее половины буферов
		if(InterlockedExchangeAdd(&param->buf_count, -1) <= BUF_COUNT/2) {
			SetEvent(param->read_wait);
		}
		// Переход на следующий буфер
		cur_buf++;
		if(cur_buf == BUF_COUNT) cur_buf = 0;
	}
	WaitForSingleObject(waits[1], INFINITE); // Ожидание завершения потока чтения
	if(param->read_result != 1) {
		mt_printf("Error: read not all");
	} else if(param->buf_count != 0) {
		mt_printf("Error: parse not all");
	}
	printf("MT Read %d numbers CheckSum %X Wait (R/P) %d/%d Time %f sec.\n", p.count, p.check, param->read_wait_count, wait_count, (double)(clock() - start) / CLOCKS_PER_SEC);
	CloseHandle(waits[1]);
	CloseHandle(param->parse_wait);
	CloseHandle(param->read_wait);
	free(param);
}


int main(int argc, char* argv[])
{
	printf("\nTest: size %d Kb count %d  Compile %s %s\n", BUF_SIZE / 1024, BUF_COUNT, __DATE__, __TIME__);
	char file[] = "big.txt";
	create_file(file, 1000000000);
	read_speed(file);
	read_ints(file);
	read_ints_mt(file);

	return 0;
}