POSIX 部分写入、线程安全和锁定

Posted 2023-02-22

【中文标题】POSIX 部分写入、线程安全和锁定

【英文标题】：POSIX partial writes, thread safety and locking

【发布时间】：2020-03-06 18:13:27

【问题描述】：

即使_write 是线程安全的，它也不能保证完全写入，可能会发生部分写入。

如果两个线程写入同一个文件描述符，有没有办法只阻塞文件描述符，而不是使用全局互斥锁来阻塞整个函数？

因此，如果两个线程尝试写入fd 1，则一个必须等​​待另一个完成；如果一个人试图写信给fd 1，而另一个人试图写信给fd 2，那么它们将同时被执行。

我正在寻找 C++ 解决方案。

#include <io.h>

struct IOError ;

void
write(int const fd, char const * buffer, int unsigned size) 
    int result;
    while (size != 0) 
        result = ::_write(fd, buffer, size);
        if (result < 0) 
            throw IOError();
        
        buffer += result;
        size -= result;
    


int
main() 
    write(1, "Hello, world!\n", 14);
    return 0;

【参考方案1】：

一个简单的解决方案是为每个文件描述符使用一个互斥锁。

您只需要一个全局互斥锁即可为给定描述符创建一个唯一互斥锁并将其存储到映射中，只要您至少可以使用 C++11 进行编译（例如 local static thread safety）

但是您需要将地图创建/地图搜索的结果存储到某些东西中（因为 STL 容器本身不是线程安全的）。 我在这里使用共享指针来提供服务。他们会自动删除。

如果您想使用异常，std::lock_guard RAII 可以帮助您在发生不良情况时释放互斥锁（参见 QA，如 Unlock mutex on exception）

您可以在此处复制/粘贴（面向 Linux 的）代码。只需将 NB_ELEM 调整为高于系统管道尺寸即可。

#include <unistd.h>
#include <mutex>
#include <map>
#include <memory>

#include <future> // For async and testing
#include <vector> // For testing
#include <iostream> // here for testing std::cout 
#include <fcntl.h> // For testing fcntl to display pipe (std out) size

void my_write(int const fd, char const * buffer, ssize_t size) 


    static std::map<int,std::shared_ptr<std::mutex>> MM;
    static std::mutex global_mutex;
    
    ssize_t result;
    std::shared_ptr<std::mutex> msptr;
    
    
        std::lock_guard<std::mutex> lock(global_mutex);
        if ( MM.cend() == MM.find(fd) ) 
            msptr = std::make_shared<std::mutex>();
            MM[fd] = msptr;
           
        else 
            msptr = MM[fd];
        
    
    
    std::lock_guard<std::mutex> lock(*msptr);

    while (size != 0) 
        result = write(fd, buffer, size);
        if (result < 0) 
            //throw if you want
        
        buffer += result;
        size -= result;
    
    


const size_t NB_ELEM = 100000u;

std::vector<char> va(NB_ELEM,'a');

std::vector<char> vb(NB_ELEM,'b');


int
main() 

    va.push_back('\n');
    vb.push_back('\n');
    
    std::cout << "stdout pipe size is : " << fcntl( 1, F_GETPIPE_SZ ) << "\n" << std::flush;
    
    
    #if 1
        auto fut2 = std::async([]()my_write(1, vb.data(), vb.size()););
        auto fut1 = std::async([]()my_write(1, va.data(), va.size()););
    #else
        auto fut2 = std::async([]()write(1, vb.data(), vb.size()););
        auto fut1 = std::async([]()write(1, va.data(), va.size()););
    #endif
    
    
    std::cout << "Bye ! \n" << std::flush;
    
    return 0;

关于大肠杆菌系统

标准输出管道尺寸为：65536

使用 my_write(...) 你会得到这样的结果

bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb...

呸呸呸呸呸呸呸呸呸呸呸……

再见！

而使用普通的 write(...) 你有时可能会得到

bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb...bbbaaaaaaaaaaa...aaaaaaaaaaabbb...

bbbbbbbbbbbbbbbbbbbbbbbaaaaaaaa....aaa

再见！