0

Основная задача заключается в вычитывании строк из файла 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;
}
5
  • я не понял до конца код, но в глаза бросается неатомарный счетчик counter или пробуйте его сделать атомарным или все места уменьшения/увеличения счетчика закрыть мьютексом. Вполне вероятно, что это состояние гонки. link: ru.cppreference.com/w/cpp/atomic/atomic Commented 17 сент. 2018 в 10:41
  • А еще этот counter довольно спорное решение, вот скажем работают у вас во всю 4 потока. counter=4, главный цикл ждет уменьшения. Первый поток закончил работу, сделал counter--; теперь он равен 3, главный цикл читает строку в буфер третьего потока, который еще загружен работой. А первый поток так и остается без новой порции данных, до тех пор пока все остальные потоки не закончат обработку и couner не станет равен 0
    – Mike
    Commented 17 сент. 2018 в 10:48
  • Да, согласен с тем, что не очень рационально, что работа потока разбора данных ориентирована на работу остальных потоков, но пока хочу разобраться с поставленным вопросом и потом доработать сам алгоритм. Commented 17 сент. 2018 в 11:04
  • Я вижу, что в блоке есть данные которые можно преобразовать сразу в массив строк, и те - которые нельзя преобразовать (в начале блока, и в конце блока). Возможно последние нужно как-то складировать - что б позже обработать каждую 123-ю строку. А вот если в блоке не 123 строки - алгоритм усложнится.
    – nick_n_a
    Commented 17 сент. 2018 в 12:06
  • 2
    Потоки нужны для того, чтобы делать другие задачи, когда некоторые задачи простаивают в ожидании входных данных. Если у Вас файл уже существует, то надо его читать в одном потоке. Разбиение чтения на потоки не ускорит чтение существующего файла.
    – pepsicoca1
    Commented 17 сент. 2018 в 14:23

0

Ваш ответ

Нажимая «Отправить ответ», вы соглашаетесь с условиями пользования и подтверждаете, что прочитали политику конфиденциальности.

Посмотрите другие вопросы с метками или задайте свой вопрос.