Из-за чего может быть ошибка?

Question

[Blunker]

Задача: На вход поступает несколько текстовых файлов и нужно в нескольких потоках записать слова в вектор.

#include <iostream>
#include <string>
#include <fstream>
#include <sstream>
#include <vector>
#include <deque>
#include <map>
#include <thread>
#include <mutex>

std::mutex lock;

void pipeline(std::ifstream&, std::vector<std::string>&);


std::string files[4] =
        {
                "file1.txt",
                "file2.txt",
                "file3.txt",
                "file4.txt"
        };

int main()
{
    std::string string;
    std::vector<std::string> str;
    std::map<std::string, std::size_t> freq;
    std::deque<std::thread> pool;
    std::ifstream in;
    std::ofstream out("out.txt");

    for (std::size_t i = 0; i < 4; i++)
    {
        in.open(files[i].c_str());

        pool.push_back(std::thread(pipeline, std::ref(in), std::ref(str)));

    }

    in.close();

    while (pool.size())
    {
        pool.front().join();
        pool.pop_front();
    }

    std::cout << str.size();

    return 0;
}

void pipeline(std::ifstream &in, std::vector<std::string> &str)
{
    lock.lock();

    while (!in.eof())
    {
        std::string temp;

        std::getline(in, temp);

        str.push_back(temp);
    }

    lock.unlock();
}

Программа компилируется без ошибок, но при выполнение появляется следующая ошибка:

terminate called after throwing an instance of 'std::bad_alloc'
  what():  std::bad_alloc

Как это можно исправить?

Answer 1

[monah_tuk]

Как минимум, вы один и тот же стрим ввода передаёте всем тредам. Тут имеет место быть не только гонка, но и вообще каша с доступом. Затем тред на продолжении ВСЕЙ своей работы блокирует мутекс. Ваш вариант с тем же успехом мог бы работать в один поток, т.к. пока поток не окончит чтение, другие будут ждать на мутексе. Блокируйте только на доступе к вектору.

Как вариант такой код:

#include <iostream>
#include <string>
#include <fstream>
#include <sstream>
#include <vector>
#include <deque>
#include <map>
#include <thread>
#include <mutex>

std::mutex lock;

void pipeline(const std::string&, std::vector<std::string>&);


std::string files[4] =
        {
                "file1.txt",
                "file2.txt",
                "file3.txt",
                "file4.txt"
        };

int main()
{
    std::string string;
    std::vector<std::string> str;
    std::map<std::string, std::size_t> freq;
    std::deque<std::thread> pool;
    std::ofstream out("out.txt");

    for (std::size_t i = 0; i < 3; i++)
    {
        pool.push_back(std::thread(pipeline, std::ref(files[i]), std::ref(str)));
    }

    while (pool.size())
    {
        pool.front().join();
        pool.pop_front();
    }

    std::cout << str.size();

    return 0;
}

void pipeline(const std::string &infile, std::vector<std::string> &str)
{
    std::ifstream in(infile.c_str());

    while (!in.eof())
    {
        std::string temp;

        std::getline(in, temp);

        {
            std::unique_lock<std::mutex> guard(lock);
            str.push_back(temp);
        }
    }
}

Изменения:
- стрим создаётся в потоке, потоку передаётся имя файла
- блокировка доступа только на модификации вектора
- использование охранного класса для блокирования и освобождения верктора.
- close на стриме можно в данном случае не делать - он закроется по выходу из области видимости

Comments

[Blunker]

Спасибо большое.

Answer 2

[mamkaololosha]

Вот исправленный вариант.
Вы передавали один in на всех и попадали в гонку/deadlock.

#include <iostream>
#include <string>
#include <fstream>
#include <sstream>
#include <vector>
#include <deque>
#include <map>
#include <thread>
#include <mutex>

std::mutex mutex;

std::string files[4] =
        {
                "file1.txt",
                "file2.txt",
                "file3.txt",
                "file4.txt"
        };

void pipeline(std::string &in, std::vector<std::string> &str)
{
	std::lock_guard<std::mutex> lock(mutex);

	std::ifstream file;
	file.open(in);

    while (!file.eof())
    {
        std::string temp;
        std::getline(file, temp);
        str.push_back(temp);
    }

	std::cout << in << " finished" << std::endl;
	file.close();
}

int main()
{
    std::string string;
    std::vector<std::string> str;
    std::map<std::string, std::size_t> freq;
    std::deque<std::thread> pool;
    std::ofstream out("out.txt");

    for (std::size_t i = 0; i < 4; i++)
    {
        pool.push_back(std::thread(pipeline, std::ref(files[i]), std::ref(str)));
    }

    while (pool.size())
    {
        pool.front().join();
        pool.pop_front();
    }

	out << str.size();
	std::cout << str.size();

	std::cin.get();

    return 0;
}