multithreading - 保护多个线程对文件的读/写访问

Question

这是上一个问题here的后续文章-我收到了一些很棒的建议，这些建议可以帮助我继续编写代码。对于下一个难题，我认为它值得一个新职位-我希望可以。
我有一些代码可以在主循环中创建请求，以从文件中读取或写入文件，并在其自己的线程中执行每个请求。在上一篇文章的帮助下，我能够扩展代码以添加具有多个条目的请求队列和读/写功能，这些功能仅休眠一小段时间即可模拟文件访问。
现在，我想实际学习如何在可能存在一个或多个线程尝试同时读取和/或写入同一文件时对文件进行读取/写入。
为了使测试更容易，我将文件限制为单个实例，否则我需要考虑文件不存在的情况等。在实际的应用程序中，将有数百个文件在运行，但是我的有限理解表明如果我可以使锁定机制适用于单个文件，则当有多个文件时，它将起作用。
我仍在尝试提高对线程的理解，首先尝试在std::mutex和read_file()函数中添加带有全局锁定变量的write_file()，但我陷入了一片困惑。
有人能给我指出正确的方法，我应该研究一下以健壮的方式进行这项工作。

#include <fstream>
#include <future>
#include <iostream>
#include <sstream>
#include <iomanip>
#include <string>
#include <random>

std::vector< std::future<std::string> > requests;

int random_int(int start, int end)
{
    std::random_device rd;
    std::mt19937 generator(rd());
    std::uniform_int_distribution<> distrib(start, end);

    return distrib(generator);
}

const std::string generate_filename()
{
    std::ostringstream filename;

    // use a single file for testing
    //filename << "file_" << std::setfill('0') << std::setw(2) << random_int(1, 20) << ".txt";

    filename << "file.txt";

    return filename.str();
}

std::string write_file(const std::string filename)
{
    std::cout << "write_file: filename is " << filename << std::endl;

    // slow things down so i can follow
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    std::ofstream ofs(filename);
    if (!ofs)
    {
        return std::string("ERROR");
    }

    const char chr = 'A' + random_int(0, 25);
    for (int i = 0; i < 64; ++i)
    {
        ofs << chr;
    }
    ofs << std::endl;
    ofs.close();

    std::cout << "write_file: written to " << filename << std::endl;

    return std::string("WRITTEN");
}

std::string read_file(const std::string filename)
{
    std::cout << "read_file: filename is " << filename << std::endl;

    // slow things down so i can follow
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    std::ifstream ifs(filename);
    if (!ifs.is_open())
    {
        return std::string("ERROR OPEINING FILE");
    }

    std::string contents;
    if (std::getline(ifs, contents))
    {
        std::cout << "    read_file: read from " << filename << std::endl;
        return std::string(contents);
    }

    return std::string("ERROR READING CONTENTS");
}

void add_request()
{
    // randomly add a read or a write request
    if (random_int(1, 50) > 25)
        requests.push_back(std::async(std::launch::async, write_file, generate_filename()));
    else
        requests.push_back(std::async(std::launch::async, read_file, generate_filename()));
}

int main(int argc, char* argv[])
{
    int max_requests = 10;

    // avoid falling out of the loop on first pass
    add_request();

    do {
        std::cout << "working: requests in queue = " << requests.size() << std::endl;

        // randomly add a request if we still have not added the max
        if (random_int(1, 5) == 1)
        {
            if (--max_requests > 0)
            {
                add_request();
            }
        }

        // service the future for each item in the request queue
        for (auto iter = requests.begin(); iter != requests.end(); )
        {
            if ((*iter).wait_for(std::chrono::milliseconds(1)) == std::future_status::ready)
            {    
                std::cout << "Request completed, removing it from the queue: result: " << (*iter).get() << std::endl;
                iter = requests.erase(iter);
            }
            else
            {
                ++iter;
            }
        }

    // once the queue is empty we exit - in the real app, we do not 
    // and keep processing requests until the app exits normally
    } while (requests.size() > 0);
}

Answer 1

这是每个线程应遵循的算法:

获取保护文件共享状态的锁。

查看表中是否存在我们尝试访问的文件。

如果不存在，请创建它。

检查我们尝试访问的文件的条目。

如果没有其他线程正在访问该文件，请跳至步骤9。

释放保护文件共享状态的锁。

等待

转到步骤1。

标记正在使用的文件。

释放保护文件共享状态的锁。

适本地读取或写入文件。

获取保护文件共享状态的锁。

标记文件未使用。

释放保护文件共享状态的锁。

请注意，如果使用条件变量使等待更有效，则步骤6、7和8变为等待条件变量，然后跳到步骤2。而且，您将需要广播条件变量(全部通知)在步骤14之前或之后。(最好在之前。)