并发追加到文件:写入丢失

Posted

技术标签:

【中文标题】并发追加到文件:写入丢失【英文标题】:Concurrent appends to a file : writes getting lost 【发布时间】:2021-01-25 01:24:34 【问题描述】:

并发追加到文件支持功能吗?

我使用并发线程 + 每个线程的 fstream 对此进行了测试。我看到数据没有损坏,但是一些写入丢失了。 写入完成后文件大小小于预期。写入不会重叠。

如果我在每个 fstream 中使用自定义搜索来编写每个线程将写入的偏移量,则不会丢失任何写入。

这里是示例代码:

#include <fstream>
#include <vector>
#include <thread>
#include "gtest/gtest.h"

void append_concurrently(string filename, const int data_in_gb, const int num_threads, const char start_char,
    bool stream_cache = true) 
    const int offset = 1024;
    const long long num_records_each_thread = (data_in_gb * 1024 * ((1024 * 1024) / (num_threads * offset)));

    
        auto write_file_fn = [&](int index) 
            // each thread has its own handle
            fstream file_handle(filename, fstream::app | fstream::binary);
            if (!stream_cache) 
                file_handle.rdbuf()->pubsetbuf(nullptr, 0); // no bufferring in fstream
            

            vector<char> data(offset, (char)(index + start_char));

            for (long long i = 0; i < num_records_each_thread; ++i) 
                file_handle.write(data.data(), offset);
                if (!file_handle) 
                    std::cout << "File write failed: "
                        << file_handle.fail() << " " << file_handle.bad() << " " << file_handle.eof() << std::endl;
                    break;
                
            

            // file_handle.flush();
        ;

        auto start_time = chrono::high_resolution_clock::now();
        vector<thread> writer_threads;
        for (int i = 0; i < num_threads; ++i) 
            writer_threads.push_back(std::thread(write_file_fn, i));
        

        for (int i = 0; i < num_threads; ++i) 
            writer_threads[i].join();
        

        auto end_time = chrono::high_resolution_clock::now();

        std::cout << filename << " Data written : " << data_in_gb << " GB, " << num_threads << " threads "
            << ", cache " << (stream_cache ? "true " : "false ") << ", size " << offset << " bytes ";
        std::cout << "Time taken: " << (end_time - start_time).count() / 1000 << " micro-secs" << std::endl;
    

    
        ifstream file(filename, fstream::in | fstream::binary);
        file.seekg(0, ios_base::end);
        
        // This EXPECT_EQ FAILS as file size is smaller than EXPECTED
        EXPECT_EQ(num_records_each_thread * num_threads * offset, file.tellg());
        file.seekg(0, ios_base::beg);
        EXPECT_TRUE(file);

        char data[offset] 0 ;
        for (long long i = 0; i < (num_records_each_thread * num_threads); ++i) 
            file.read(data, offset);
            EXPECT_TRUE(file || file.eof()); // should be able to read until eof
            char expected_char = data[0]; // should not have any interleaving of data.

            bool same = true;
            for (auto & c : data) 
                same = same && (c == expected_char) && (c != 0);
            

            EXPECT_TRUE(same); // THIS PASSES
            if (!same) 
                std::cout << "corruption detected !!!" << std::endl;
                break;
            

            if (file.eof())  // THIS FAILS as file size is smaller
                EXPECT_EQ(num_records_each_thread * num_threads, i + 1);
                break;
            
        
    


TEST(fstream, file_concurrent_appends) 
    string filename = "file6.log";
    const int data_in_gb = 1;
    
    
        // trunc file before write threads start.
        
            fstream file(filename, fstream::in | fstream::out | fstream::trunc | fstream::binary);
        
        append_concurrently(filename, data_in_gb, 4, 'B', false);
    
    std::remove(filename.c_str());

编辑:

我将 fstream 移动到由所有线程共享。现在,对于 512 字节的缓冲区大小,我看到 8 次写入,总共丢失了 4 KB。

    const int offset = 512;
    const long long num_records_each_thread = (data_in_gb * 1024 * ((1024 * 1024) / (num_threads * offset)));

    fstream file_handle(filename, fstream::app | fstream::binary);
    if (!stream_cache) 
        file_handle.rdbuf()->pubsetbuf(nullptr, 0); // no bufferring in fstream
    

4KB 缓冲区大小不会重现问题。

Running main() from gtest_main.cc
Note: Google Test filter = *file_conc*_*append*
[==========] Running 1 test from 1 test case.
[----------] Global test environment set-up.
[----------] 1 test from fstream
[ RUN      ] fstream.file_concurrent_appends
file6.log Data written : 1 GB, 1 threads , cache true , size 512 bytes Time taken: 38069289 micro-secs
d:\projs\logpoc\tests\test.cpp(279): error:       Expected: num_records_each_thread * num_threads * offset
      Which is: 1073741824
To be equal to: file.tellg()
      Which is: 1073737728
d:\projs\logpoc\tests\test.cpp(301): error:       Expected: num_records_each_thread * num_threads
      Which is: 2097152
To be equal to: i + 1
      Which is: 2097145

编辑 2: 在加入所有线程以从内部缓冲区刷新数据后关闭file_handle。这解决了上述问题。

【问题讨论】:

@Andreas Wenzel 不支持使用相同 fstream 的多线程,因为它内部有一个偏移位置。我以附加模式打开文件。所以,我希望操作系统会关心偏移量 【参考方案1】:

根据§29.4.2 ¶7 of the official ISO C++20 standard,std::fstream提供的函数一般都是线程安全的。

但是,如果每个线程都有自己的std::fstream 对象,那么就 C++ 标准库而言,这些是不同的流,不会发生同步。只有操作系统的kernel 会知道所有文件句柄都指向同一个文件。因此,任何同步都必须由内核完成。但是内核可能甚至不知道写入应该到达文件的末尾。根据您的平台,内核可能只接收特定文件位置的写入请求。如果文件的末尾同时被另一个线程的追加移动了,那么线程之前的写请求的位置可能不再是文件的末尾。

根据std::fstream::open 上的文档,以附加模式打开文件将导致流在每次写入之前查找文件末尾。这种行为似乎正是您想要的。但是,由于上述原因,这可能仅在所有线程共享相同的std::fstream 对象时才有效。在这种情况下,std::fstream 对象应该能够同步所有写入。特别是,它应该能够以原子方式执行对文件末尾的查找和后续写入。

【讨论】:

谢谢.. 我尝试分享 fstream。缓冲区大小为 512,我看到 8 次写入,总共 4 KB 写入一直丢失。 @AshishNegi:我怀疑您的函数调用 file_handle.rdbuf()-&gt;pubsetbuf(nullptr, 0); 可能会失败。根据this answer to another question,您必须在打开文件之前调用该函数。由于您似乎没有刷新输出缓冲区,我怀疑丢失的 4096 字节仍在缓冲区中。 @AshishNegi:经过进一步研究,我发现根据this page,gcc编译器要求在打开文件之前调用该函数,但其​​他编译器允许您调用该函数打开文件后也是如此。不过,我不确定该页面是否仍然是最新的。 情况好像没变。使用 libstdc++(由 gcc 使用),在打开任何文件之前仍必须调用 setbufpubsetbuf。否则,函数调用无效。请参阅this page from the official documentation 了解更多信息。 你是正确的关于不刷新/关闭用于写入的早期文件句柄。谢谢..这解决了这个谜..

以上是关于并发追加到文件:写入丢失的主要内容,如果未能解决你的问题,请参考以下文章

file_put_contents记录的日志内容丢失

windows环境,多线程情况下,C语言向文件写入数据。

数据库事务:“写入偏差”和“丢失更新”之间的区别

Kafka如何保证百万级写入速度 并 保证不丢失不重复消费

日志文件丢失

在 Cassandra 中使用轻量级事务 (CAS) 时,我们如何避免丢失写入?