Основная задача заключается в вычитывании строк из файла txt с 8 млн. строк. Данную задачу хотелось реализовать в многопоточном режиме. Поскольку использование одного fstream в нескольких потоках, по высказыванию некоторых людей на форумах, неразумно, было принято решение читать блоки данных (в файле 123 строки - это единичный блок данных). Схема программы следующая:
Создаётся вектор в котором в свою очередь элементами также являются вектора, в которых хранятся строки, то есть vector<vector<string>>
. Число вложенных векторов равно числу физических ядер процессора.
После того, как вложенные вектора заполнены строками (каждый вложенный вектор имеет именно 123 строки), основной main поток чтения файла засыпает в бесконечном цикле, из которого он может выйти только, если глобальная переменная counter уменьшится во всех 4-х (в моём случае) потоках до 0. В дочерних потоках происходит считывание данных из стринговых векторов и запись их в файл, плюс происходит очистка вектора строк.Таким образом каждый дочерний поток читает только свою ячейку данных внутри вектора.
Наконец-то подошёл к самому вопросу: в дочернем потоке, первоначально для контроля за работой потоков производился вывод name_id потока в консоль через cout. Далее за ним следовала процедура записи данных в файл. Так вот, после того как программа стала работать правильно и потоки стали разбирать данные корректно, cout был удалён. Однако его удаление привело к тому, программа стала работать в бесконечном цикле дочернего потока. То есть, к примеру первая операция чтения данных первого потока отрабатывает, второго наполовину и программа начинает крутиться бесконечно в цикле дочернего потока. Я попробовал внести операцию усыпления потока вместо cout на 1/100000 миллисекунды и это помогло.
В итоге вопрос: как можно поправить программу и удалить данный костыль в виде cout или усыпления потока?
#include "pch.h"
#include "libraries.h"
using namespace std;
mutex mtx; //mutex for blocking resourses
//float timer = 0.00000f;
unsigned short int counter = 0;
const int amountOfRowsInBlock = 123; // amount of rows in the one item in groo file
bool flagOut = false; // flag for outing from child thread
void funcRead(vector<string> &str) { //function for analyze data from the large file in some threads
std::ostringstream ss;
ss << std::this_thread::get_id();
std::string idstr = ss.str(); // three rows for getting name_id of thred
fstream fileWrite("result_" + idstr + ".txt",ios::out | ios::app | ios::ate);
for (;;) {
if (flagOut) {
fileWrite.close();
return;
}
if (str.size() == amountOfRowsInBlock) {
for (int i = 0; i < str.size(); i++) {
//cout << /*"Thread name " << this_thread::get_id() << "---" << str[i]*/"*" << endl;
//this_thread::sleep_for(std::chrono::milliseconds(1/100000));
fileWrite << "Thread name " << this_thread::get_id() << "---" << str[i] << "\n";
}
str.clear();
counter--;
}
}
}
int main() {
const int amountOfThreads = 4; //for the faster work i think better use amount of threads equal amount of physical cores of CPU
vector<thread> threads;
vector<vector<string>> stringsOfGRO(amountOfThreads, vector<string>());
fstream fileRead("C:\\Users\\Denis\\Desktop\\newFolder\\Files\\fullC8_70W_30ACN\\short.txt", ios::in);
if (!fileRead) throw exception("The operation of openning has have problem!!!");
for (int i = 0; i < amountOfThreads; i++) {
thread thr(funcRead, ref(stringsOfGRO[i])); //creation of threads
threads.emplace_back(move(thr));
}
string bufferString = "";
char bufferChar[100];
while (1) {
if (fileRead.eof()) {
for (int i = 1; i < amountOfThreads; i++) {
flagOut = true;
counter = 0;
}
break;
}
if (counter == amountOfThreads) {
while (counter != 0) {
this_thread::sleep_for(std::chrono::milliseconds(1));
}
}
for (short int j = 0; j < amountOfRowsInBlock; j++) {
fileRead.getline(bufferChar, sizeof(bufferChar));
bufferString = bufferChar;
{
lock_guard<std::mutex> lock(mtx);
stringsOfGRO[counter].emplace_back(bufferString);
}
}
counter++;
}
fileRead.close();
// can't use const auto& here since .join() is not marked const
for (auto& thr : threads) {
thr.join();
}
cout << clock()/1000 << endl;
system("pause");
return 0;
}