Частичная запись POSIX, безопасность потоков и блокировка

Question

Даже если _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;
}

Answer 1

Простым решением будет использование одного мьютекса для каждого дескриптора файла.

Вам потребуется только глобальный мьютекс, чтобы создать один уникальный мьютекс для данного дескриптора и сохранить его в карте, пока так как вы можете компилировать по крайней мере с C ++ 11 (например, local stati c thread thread )

Но вам нужно сохранить результат создания карты / поиска карты во что-то ( поскольку контейнеры STL сами по себе не являются потокобезопасными). Я использовал общие указатели здесь, чтобы служить этому. они автоматически c удаляют.

Если вы хотите использовать исключение, std :: lock_guard RAII поможет вам освободить мьютекс, если произойдет что-то плохое (см. QA вроде Разблокировать мьютекс при исключении )

Здесь (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;
}

В системе coliru

Размер трубы stdout: 65536

С my_write (...) вы получите такой результат

bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb ...

aaaaaaaaaaaaaaaaaaaaaaaaaaaa30aaaa

И при обычной записи (...) вы иногда можете получить

bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb ... bbbaaaaaaaaaaa ... aaaaaaaabababa .... ааа

пока!