boost::unordered_map<int,struct>和shared_mutex的執行緒安全-有解無憂

我正在嘗試決議來自具有 4 個執行緒的套接字的 ts 流資料。我決定使用 boost 共享互斥鎖來管理連接和資料接收。但我完全是 C 的新手，我不確定我是否會在胎面安全方面做得對。我正在使用 boost unordered_map<int, TsStreams>，當一個新用戶連接時，我用一個唯一的鎖鎖定互斥鎖并將一個用戶添加到地圖，當這個用戶斷開連接時，我鎖定互斥鎖一個獨特的鎖并將他從地圖中洗掉。TsStreams 結構在用戶發送資料時包含向量和一些附加變數，我使用共享鎖從地圖中獲取用戶的 TsStreams 參考，將新資料添加到向量并修改其他變數。以這種方式修改 TsStreams 是否是執行緒安全的？

class Demuxer {

public:
    Demuxer();
    typedef signal<void (int, TsStream)> PacketSignal;
    void onUserConnected(User);
    void onUserDisconnected(int);
    void onUserData(Data);
    void addPacketSignal(const PacketSignal::slot_type& slot);
private:
    mutable PacketSignal packetSignal;
    void onPacketReady(int, TsStream);
    TsDemuxer tsDemuxer;
    boost::unordered_map<int, TsStreams> usersData;
    boost::shared_mutex mtx_;
};

#include "Demuxer.h"

Demuxer::Demuxer() {
    tsDemuxer.addPacketSignal(boost::bind(&Demuxer::onPacketReady, this, _1, _2));
}

void Demuxer::onUserConnected(User user){
    boost::unique_lock<boost::shared_mutex> lock(mtx_);
    if(usersData.count(user.socket)){
        usersData.erase(user.socket);
    }
    TsStreams streams;
    streams.video.isVideo = true;
    usersData.insert(std::make_pair(user.socket, streams));
}
void Demuxer::onUserDisconnected(int socket){
    boost::unique_lock<boost::shared_mutex> lock(mtx_);
    if(usersData.count(socket)){
        usersData.erase(socket);
    }
}

void Demuxer::onUserData(Data data) {
    boost::shared_lock<boost::shared_mutex> lock(mtx_);
    if(!usersData.count(data.socket)){
        return;
    }
    tsDemuxer.parsePacket(data.socket, std::ref(usersData.at(data.socket)), (uint8_t *) data.buffer, data.length);
    }

void Demuxer::onPacketReady(int socket, TsStream data) {
    packetSignal(socket, data);
}

void Demuxer::addPacketSignal(const PacketSignal::slot_type& slot){
    packetSignal.connect(slot);
}

struct TsStreams{
    TsStreams() = default;
    TsStreams(const TsStreams &p1) {}
    TsStream video;
    TsStream audio;
};

struct TsStream
{
    TsStream() = default;
    TsStream(const TsStream &p1) {}
    boost::recursive_mutex mtx_; // to make sure to have the queue, it may not be necessary
    uint64_t PTS = 0;
    uint64_t DTS = 0;
    std::vector<char> buffer;
    uint32_t bytesDataLength = 0;
    bool isVideo = false;
};

class TsDemuxer {
public:
    typedef signal<void (int, TsStream)> PacketSignal;
    void parsePacket(int socket, TsStreams &streams, uint8_t *data, int size);
    connection addPacketSignal(const PacketSignal::slot_type& slot);
private:
    PacketSignal packetSignal;
    void parseTSPacket(int socket, TsStream &stream, uint8_t *data, int size);
    void parseAdaptationField(BitReader &bitReader);
    void parseStream(int socket, TsStream &stream, BitReader &bitReader, uint32_t payload_unit_start_indicator);
    void parsePES(TsStream &stream, BitReader &bitReader);
    int64_t parseTSTimestamp(BitReader &bitReader);
};

void TsDemuxer::parsePacket(int socket, TsStreams &streams, uint8_t *data, int size) {
//some parsing
    if(video){
            streams.video.mtx_.lock();
            parseTSPacket(socket, streams.video, (uint8_t *)buf, 188);
        }else{
            streams.audio.mtx_.lock();
            parseTSPacket(socket, streams.audio, (uint8_t *)buf, 188);
        }
}

void TsDemuxer::parseTSPacket(int socket, TsStream &stream, uint8_t *data, int size)
{
 //some more parsing
parseStream(socket, stream, bitReader, payload_unit_start_indicator);
}

void TsDemuxer::parseStream(int socket, TsStream &stream, BitReader &bitReader, uint32_t payload_unit_start_indicator) {

    if(payload_unit_start_indicator)
    {
        if(!stream.buffer.empty()){
            packetSignal(socket, stream);
            stream.buffer = vector<char>();
            stream.bytesDataLength = 0;
        }
        parsePES(stream, bitReader);
    }

    size_t payloadSizeBytes = bitReader.numBitsLeft() / 8;

    copy(bitReader.getBitReaderData(), bitReader.getBitReaderData() payloadSizeBytes,back_inserter(stream.buffer));
    stream.mtx_.unlock();
}

uj5u.com熱心網友回復：

解復用器對我來說看起來是正確的。不過也有一些低效的地方：

你不需要count在你之前erase。抹去就好。如果一個元素不存在，這將什么都不做。這為您節省了一次查找。同樣，不要使用count后跟at. 使用find（見下文使用）。
您可能希望將盡可能多的作業移出關鍵部分。您的敵人示例onUserConnected可以在獲取鎖之前創建 TsStreams 物件。
請注意，更改無序映射永遠不會使映射中元素的指標或參考無效，除非它們被擦除。這意味著onUserData您在決議資料包時不必持有地圖上的鎖。

也就是說，假設您沒有從兩個不同的執行緒為同一個用戶呼叫 onUserData。您可以通過對 TsStream 物件引入第二個鎖來防止這種情況發生。同樣，您應該防止在另一個執行緒仍可能決議最后一個資料包時擦除元素。我會為此使用 a shared_ptr。像這樣的東西：

class Demuxer {
...
    boost::unordered_map<int, boost::shared_ptr<TsStreams> > usersData;
    boost::shared_mutex mtx_;
};
void Demuxer::onUserData(Data data) {
    boost::shared_lock<boost::shared_mutex> maplock(mtx_);
    auto found = usersData.find(data.socket);
    if(found == usersData.end())
        return;
    boost::shared_ptr<TsStreams> stream = found->second;
    boost::unique_lock<boost::recursive_mutex> datalock(stream->mtx_);
    maplock.unlock();
    tsDemuxer.parsePacket(data.socket, *stream, (uint8_t *) data.buffer, data.length);
}

如果您使用這種方法減少使用 Demuxer 鎖的時間，您可能應該用普通的互斥鎖替換該共享互斥鎖。共享互斥體的開銷要高得多，對于如此短的關鍵部分不值得。

TsDemuxer 看起來有點不穩定：

在TsDemuxer::parsePacket你永遠不會解鎖互斥鎖。那不應該是一個unique_lock嗎？同樣，在parseStream解鎖似乎未配對。一般來說，unique_lock與手動鎖定和解鎖相比，使用物件始終是最佳選擇。如果有的話，鎖定和解鎖unique_lock，而不是互斥鎖。

與多執行緒無關的備注

stream.buffer.clear()比stream.buffer = vector<char>()因為這將重用緩沖記憶體而不是完全釋放它更有效。
正如其他人所指出的，boost 的這些部分現在是標準庫的一部分。替換boost::為std::并啟用最新的 C 標準，如 C 14 或 17，就可以了。在最壞的情況下，您必須替換shared_mutex為shared_timed_mutex.
在 Demuxer 中，您按值傳遞 User 和 Data 物件。你確定那些不應該是 const 參考嗎？

轉載請註明出處，本文鏈接：https://www.uj5u.com/net/481901.html

標籤：C 多线程线程安全无序映射

上一篇：將Python嵌入到C 分段錯誤

下一篇：佇列如何解決執行緒競爭問題？