锟斤拷锟斤拷位锟矫ｏ拷锟斤拷页 > 锟斤拷锟斤拷锟教筹拷 > 锟教筹拷 > iceoryx源锟斤拷锟侥讹拷锟斤拷锟侥ｏ拷锟斤拷锟斤拷锟斤拷锟斤拷锟节达拷通锟脚ｏ拷锟斤拷锟斤拷

iceoryx源锟斤拷锟侥讹拷锟斤拷锟侥ｏ拷锟斤拷锟斤拷锟斤拷锟斤拷锟节达拷通锟脚ｏ拷锟斤拷锟斤拷

锟斤拷源锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷|锟斤拷时锟戒：2024-05-29 08:55:57 |锟斤拷锟侥讹拷锟斤拷143锟斤拷|锟斤拷锟斤拷签锟斤拷 C 锟侥讹拷通锟斤拷 CEO 锟斤拷 |锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷

目录0 锟斤拷锟斤拷1 锟斤拷锟斤拷锟斤拷锟捷结构2 锟斤拷锟斤拷锟节达拷锟饺�2.1 PublisherImpl::loan2.2 PublisherImpl::loanSample2.3 PublisherPortUser::tryAllocateChunk2.4 ChunkSender::tryAllocate3 锟斤拷息锟斤拷锟斤拷锟竭硷拷3

0 锟斤拷锟斤拷

iceoryx源锟斤拷锟侥讹拷锟斤拷一锟斤拷锟斤拷锟斤拷全锟街革拷锟斤拷
iceoryx源锟斤拷锟侥讹拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟节达拷锟斤拷锟�
iceoryx源锟斤拷锟侥讹拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟节达拷锟斤拷锟斤拷锟揭伙拷锟�
iceoryx源锟斤拷锟侥讹拷锟斤拷锟侥ｏ拷锟斤拷锟斤拷锟斤拷锟斤拷锟节达拷通锟脚ｏ拷锟斤拷锟斤拷
iceoryx源锟斤拷锟侥讹拷锟斤拷锟藉）锟斤拷锟斤拷锟斤拷锟斤拷锟节达拷通锟脚ｏ拷锟斤拷锟斤拷
iceoryx源锟斤拷锟侥讹拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟节存创锟斤拷
iceoryx源锟斤拷锟侥讹拷锟斤拷锟竭ｏ拷锟斤拷锟斤拷锟斤拷锟斤拷锟街伙拷锟斤拷
iceoryx源锟斤拷锟侥讹拷锟斤拷锟剿ｏ拷锟斤拷锟斤拷IPC通锟脚伙拷锟斤拷

锟斤拷锟斤拷锟侥讹拷锟诫共锟斤拷锟节达拷通锟斤拷锟斤拷氐锟斤拷呒锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷然锟饺∫伙拷楣诧拷锟斤拷诖妫拷锟斤拷锟斤拷锟叫达拷锟斤拷锟斤拷荩锟饺伙拷锟斤拷锟斤拷锟较拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟较拷锟斤拷锟斤拷锟斤拷荩锟斤拷锟斤拷锟斤拷叽锟斤拷锟较拷锟斤拷锟斤拷卸锟饺★拷锟较拷锟斤拷锟斤拷锟斤拷荨锟斤拷锟斤拷拇锟斤拷姆锟斤拷锟斤拷锟叫斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷萁峁癸拷锟斤拷锟斤拷锟斤拷诖锟斤拷取锟斤拷锟斤拷息锟斤拷锟斤拷锟竭硷拷锟斤拷锟斤拷息锟斤拷锟斤拷锟竭硷拷锟斤拷

1 锟斤拷锟斤拷锟斤拷锟捷结构

锟斤拷锟斤拷前锟斤拷知锟斤拷锟斤拷锟斤拷锟斤拷元锟斤拷为 ShmSafeUnmanagedChunk 锟斤拷锟斤拷锟叫达拷诺锟斤拷锟� ChunkManagement 锟斤拷锟节癸拷锟斤拷锟节达拷蔚锟絠d锟斤拷锟斤拷愿霉锟斤拷锟斤拷诖锟斤拷椎锟街凤拷锟狡拷疲锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟绞撅拷锟�

iceoryx源码阅读（四）——共享内存通信（二）

锟斤拷息锟斤拷锟斤拷锟斤拷锟斤拷锟铰达拷锟诫定锟藉：

struct ChunkQueueData : public LockingPolicy
{
    // ...
    static constexpr uint64_t MAX_CAPACITY = ChunkQueueDataProperties_t::MAX_QUEUE_CAPACITY;
    cxx::VariantQueue m_queue;
    // ...
};

struct ChunkDistributorData : public LockingPolicy
{
    // ...
    using QueueContainer_t =
    cxx::vector, ChunkDistributorDataProperties_t::MAX_QUEUES>;
    QueueContainer_t m_queues;
    // ...
};

struct ChunkReceiverData : public ChunkQueueDataType
{
    // ...
};

ChunkDistributorData 锟角凤拷锟斤拷锟斤拷锟斤拷锟斤拷锟叫的讹拷锟斤拷锟斤拷锟捷结构锟斤拷锟斤拷锟斤拷一锟斤拷锟斤拷锟斤拷锟竭伙拷址锟斤拷锟斤拷锟斤拷锟斤拷锟侥端ｏ拷锟斤拷锟皆筹拷锟叫讹拷锟斤拷锟斤拷小锟�
ChunkReceiverData 锟角讹拷锟斤拷锟竭碉拷锟斤拷锟斤拷锟斤拷锟斤拷坛锟斤拷锟� ChunkQueueData 锟斤拷锟节诧拷只锟斤拷一锟斤拷锟斤拷锟叫ｏ拷锟斤拷锟斤拷元锟斤拷锟斤拷锟斤拷为 ShmSafeUnmanagedChunk 锟斤拷

锟斤拷锟斤拷锟斤拷锟斤拷锟叫ｏ拷锟斤拷锟斤拷锟斤拷锟捷结构锟斤拷锟斤拷锟斤拷为 cxx::VariantQueue 锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷一锟斤拷锟戒长锟斤拷锟介，锟斤拷实锟斤拷锟斤拷锟斤拷锟斤拷一锟斤拷锟斤拷锟斤拷锟斤拷锟介，锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷萁峁癸拷锟斤拷澹�

enum class VariantQueueTypes : uint64_t
{
    FiFo_SingleProducerSingleConsumer = 0,
    SoFi_SingleProducerSingleConsumer = 1,
    FiFo_MultiProducerSingleConsumer = 2,
    SoFi_MultiProducerSingleConsumer = 3
};

template 
class VariantQueue
{
public:
    using fifo_t = variant,
                           concurrent::SoFi,
                           concurrent::ResizeableLockFreeQueue,
                           concurrent::ResizeableLockFreeQueue>;
    // ...

private:
    VariantQueueTypes m_type;
    fifo_t m_fifo;
};

fifo_t 锟角讹拷锟叫底诧拷峁癸拷锟斤拷停锟斤拷锟斤拷锟斤拷锟� concurrent::FiFo 锟斤拷 concurrent::SoFi 锟斤拷 concurrent::ResizeableLockFreeQueue 之一锟斤拷锟斤拷锟斤拷使锟斤拷锟斤拷一锟街ｏ拷锟斤拷枚锟斤拷值 m_type 确锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟节诧拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟捷结构锟斤拷

template 
struct NonZeroedBuffer
{
    struct alignas(ElementType) element_t
    {
        cxx::byte_t data[sizeof(ElementType)];
    };
    element_t value[Capacity];
};

锟斤拷锟斤拷锟斤拷一锟结构锟斤拷锟绞撅拷锟斤拷一锟斤拷锟斤拷锟介，锟斤拷元锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷为Element锟斤拷

2 锟斤拷锟斤拷锟节达拷锟饺�

锟斤拷锟斤拷锟斤拷锟斤拷前锟斤拷应锟矫筹拷锟斤拷锟斤拷锟斤拷锟斤拷要锟饺伙拷取一锟斤拷锟斤拷蚀锟叫★拷锟紺hunk锟斤拷锟斤拷锟斤拷锟斤拷写锟斤拷锟斤拷锟捷ｏ拷然锟斤拷锟斤拷锟斤拷锟较拷锟斤拷徒涌诮锟斤拷蟹锟斤拷汀锟�

2.1 PublisherImpl::loan

职锟斤拷

锟斤拷取一锟介共锟斤拷锟节存，锟斤拷锟斤拷锟矫癸拷锟届函锟斤拷锟斤拷锟叫筹拷始锟斤拷锟斤拷

锟斤拷危锟�

args锟斤拷模锟斤拷锟轿ｏ拷锟斤拷锟节碉拷锟矫达拷锟斤拷锟斤拷锟酵的癸拷锟届函锟斤拷锟斤拷也锟斤拷锟皆诧拷锟斤拷锟斤拷

锟斤拷锟截ｏ拷

Sample锟斤拷锟斤拷实锟斤拷锟斤拷锟斤拷锟斤拷锟角讹拷锟矫伙拷锟缴诧拷锟斤拷锟侥癸拷锟斤拷锟节达拷蔚姆锟阶帮拷锟�

template 
template 
inline cxx::expected, AllocationError>
PublisherImpl::loan(Args&&... args) noexcept
{
    return std::move(loanSample().and_then([&](auto& sample) { new (sample.get()) T(std::forward(args)...); }));
}

锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷

锟斤拷锟饺碉拷锟斤拷loanSample锟斤拷锟斤拷锟斤拷取锟斤拷锟斤拷锟节存，然锟斤拷锟斤拷霉锟斤拷旌拷锟斤拷锟斤拷谐锟绞硷拷锟斤拷锟斤拷锟斤拷锟绞癸拷锟絇lacement new锟斤法锟斤拷锟斤拷要指锟斤拷锟斤拷锟角ｏ拷loanSample锟斤拷锟截碉拷锟角斤拷锟斤拷锟节达拷锟斤拷没锟斤拷锟斤拷莸锟斤拷椎锟街凤拷锟斤拷锟斤拷锟斤拷锟紺hunk锟斤拷锟阶碉拷址锟斤拷

2.2 PublisherImpl::loanSample

职锟斤拷

锟斤拷锟戒共锟斤拷锟节存，锟斤拷锟斤拷锟斤拷转锟斤拷为Sample锟斤拷锟酵ｏ拷锟斤拷锟斤拷锟截★拷

锟斤拷锟截ｏ拷

Sample锟斤拷锟斤拷实锟斤拷锟斤拷

template 
inline cxx::expected, AllocationError> PublisherImpl::loanSample() noexcept
{
    static constexpr uint32_t USER_HEADER_SIZE{std::is_same::value ? 0U : sizeof(H)};

    auto result = port().tryAllocateChunk(sizeof(T), alignof(T), USER_HEADER_SIZE, alignof(H));
    if (result.has_error())
    {
        return cxx::error(result.get_error());
    }
    else
    {
        return cxx::success>(convertChunkHeaderToSample(result.value()));
    }
}

锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷

锟斤拷锟饺碉拷锟斤拷 tryAllocateChunk 锟斤拷锟揭伙拷楣诧拷锟斤拷诖妫拷锟斤拷锟斤拷锟絊ample实锟斤拷锟斤拷

2.3 PublisherPortUser::tryAllocateChunk

职锟斤拷

锟斤拷锟戒共锟斤拷锟节存，锟斤拷锟斤拷锟斤拷转锟斤拷为Sample锟斤拷锟酵ｏ拷锟斤拷锟斤拷锟截★拷

锟斤拷危锟�

4锟斤拷锟斤拷锟节硷拷锟斤拷锟斤拷锟借共锟斤拷锟节达拷锟叫★拷牟锟斤拷锟斤拷锟斤拷锟斤拷锊徽癸拷锟斤拷锟斤拷锟斤拷恕锟�

锟斤拷锟斤拷值锟斤拷

锟斤拷锟斤拷锟节达拷锟阶碉拷址锟斤拷锟斤拷锟斤拷为 ChunkHeader * 锟斤拷锟斤拷 4.1 Chunk锟斤拷锟斤拷锟结构锟斤拷

cxx::expected
PublisherPortUser::tryAllocateChunk(const uint32_t userPayloadSize,
                                    const uint32_t userPayloadAlignment,
                                    const uint32_t userHeaderSize,
                                    const uint32_t userHeaderAlignment) noexcept
{
    return m_chunkSender.tryAllocate(
        getUniqueID(), userPayloadSize, userPayloadAlignment, userHeaderSize, userHeaderAlignment);
}

锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷

锟斤拷锟斤拷锟斤拷锟斤拷只锟角简单地碉拷锟斤拷 ChunkSender 锟斤拷 tryAllocate 锟斤拷锟斤拷锟斤拷

2.4 ChunkSender::tryAllocate

职锟斤拷

锟斤拷锟斤拷 MemoryManager锟侥筹拷员锟斤拷锟斤拷getChunk 锟矫碉拷锟斤拷锟斤拷锟节达拷锟斤拷锟斤拷锟斤拷锟揭伙拷锟绞癸拷玫墓锟斤拷锟斤拷诖锟介。

锟斤拷危锟�

同锟较ｏ拷锟皆ｏ拷

锟斤拷锟斤拷值锟斤拷

指锟斤拷锟斤拷锟节达拷锟斤拷椎锟街凤拷锟街革拷耄拷锟斤拷锟轿� ChunkHeader 锟斤拷

template 
inline cxx::expected
ChunkSender::tryAllocate(const UniquePortId originId,
                                              const uint32_t userPayloadSize,
                                              const uint32_t userPayloadAlignment,
                                              const uint32_t userHeaderSize,
                                              const uint32_t userHeaderAlignment) noexcept
{
    const auto chunkSettingsResult =
        mepoo::ChunkSettings::create(userPayloadSize, userPayloadAlignment, userHeaderSize, userHeaderAlignment);
    if (chunkSettingsResult.has_error())
    {
        return cxx::error(AllocationError::INVALID_PARAMETER_FOR_USER_PAYLOAD_OR_USER_HEADER);
    }

    const auto& chunkSettings = chunkSettingsResult.value();
    const uint32_t requiredChunkSize = chunkSettings.requiredChunkSize();

    auto& lastChunkUnmanaged = getMembers()->m_lastChunkUnmanaged;
    mepoo::ChunkHeader* lastChunkChunkHeader =
        lastChunkUnmanaged.isNotLogicalNullptrAndHasNoOtherOwners() ? lastChunkUnmanaged.getChunkHeader() : nullptr;

    if (lastChunkChunkHeader && (lastChunkChunkHeader->chunkSize() >= requiredChunkSize))
    {
        /* * * * *  锟斤拷锟斤拷锟斤拷锟�2-4-1锟斤拷锟斤拷锟斤拷锟斤拷锟揭伙拷畏锟斤拷锟侥癸拷锟斤拷锟节达拷  * * * * */
    }
    else
    {
        /* * * * *  锟斤拷锟斤拷锟斤拷锟�2-4-2锟斤拷锟斤拷锟斤拷一锟斤拷锟铰碉拷未使锟矫的癸拷锟斤拷锟节达拷 * * * * */
    }
}

锟斤拷未锟斤拷锟斤拷锟斤拷锟斤拷

LINE 09 锟斤拷 LINE 17锟斤拷 锟斤拷锟斤拷锟斤拷锟借共锟斤拷锟节达拷锟叫★拷锟�
LINE 19 锟斤拷 LINE 30锟斤拷 锟叫讹拷锟斤拷锟揭伙拷畏锟斤拷锟侥癸拷锟斤拷锟节达拷锟斤拷欠锟斤拷锟斤拷卸锟斤拷锟斤拷叨锟斤拷讯锟饺★拷锟斤拷锟斤拷掖锟叫★拷锟斤拷锟斤拷锟斤拷锟斤拷小锟斤拷锟斤拷锟斤拷锟斤拷锟揭伙拷畏锟斤拷锟侥癸拷锟斤拷锟节达拷椋拷锟斤拷锟斤拷路锟斤拷涔诧拷锟斤拷诖锟介。

锟斤拷锟斤拷锟�2-4-1锟斤拷锟斤拷锟斤拷锟斤拷锟揭伙拷畏锟斤拷锟侥癸拷锟斤拷锟节达拷

auto sharedChunk = lastChunkUnmanaged.cloneToSharedChunk();
if (getMembers()->m_chunksInUse.insert(sharedChunk))
{
    auto chunkSize = lastChunkChunkHeader->chunkSize();
    lastChunkChunkHeader->~ChunkHeader();
    new (lastChunkChunkHeader) mepoo::ChunkHeader(chunkSize, chunkSettings);
    lastChunkChunkHeader->setOriginId(originId);
    return cxx::success(lastChunkChunkHeader);
}
else
{
    return cxx::error(AllocationError::TOO_MANY_CHUNKS_ALLOCATED_IN_PARALLEL);
}

锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷

锟斤拷锟斤拷锟斤拷锟绞癸拷玫墓锟斤拷锟斤拷诖锟斤拷未锟斤拷锟斤拷锟斤拷锟斤拷耄拷锟斤拷锟斤拷锟街帮拷锟斤拷锟斤拷荩锟酵憋拷锟斤拷锟斤拷锟节达拷锟较癸拷锟斤拷锟铰碉拷 ChunkHeader 锟斤拷锟斤拷锟津返回达拷锟斤拷

锟斤拷锟斤拷锟�2-4-2锟斤拷锟斤拷锟斤拷一锟斤拷锟铰碉拷未使锟矫的癸拷锟斤拷锟节达拷

auto getChunkResult = getMembers()->m_memoryMgr->getChunk(chunkSettings);

if (!getChunkResult.has_error())
{
    auto& chunk = getChunkResult.value();

    // if the application allocated too much chunks, return no more chunks
    if (getMembers()->m_chunksInUse.insert(chunk))
    {
        // END of critical section
        chunk.getChunkHeader()->setOriginId(originId);
        return cxx::success(chunk.getChunkHeader());
    }
    else
    {
        // release the allocated chunk
        chunk = nullptr;
        return cxx::error(AllocationError::TOO_MANY_CHUNKS_ALLOCATED_IN_PARALLEL);
    }
}
else
{
    /// @todo iox-#1012 use cxx::error::from(E1); once available
    return cxx::error(cxx::into(getChunkResult.get_error()));
}

锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷

锟斤拷锟斤拷MemoryManager锟侥筹拷员锟斤拷锟斤拷getChunk锟斤拷取锟斤拷锟斤拷锟节达拷椋拷锟斤拷锟斤拷取锟缴癸拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷 m_chunksInUse 锟斤拷锟斤拷锟斤拷锟饺∈э拷芑锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷蚍祷鼗锟饺∈э拷埽锟斤拷锟绞憋拷锟斤拷锟絉AII原锟斤拷锟斤拷 SharedChunk 锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟皆讹拷锟斤拷锟斤拷锟斤拷锟节达拷榉碉拷锟斤拷锟� MemPool 锟斤拷

m_chunksInUse 锟节诧拷锟斤拷装锟斤拷锟斤拷锟斤拷元锟截碉拷锟斤拷锟斤拷为锟斤拷锟斤拷锟斤拷锟斤拷一篇锟斤拷锟斤拷锟叫斤拷锟杰碉拷 ShmSafeUnmanagedChunk 锟斤拷锟斤拷锟斤拷锟斤拷筒锟斤拷锟斤拷锟斤拷锟斤拷眉锟斤拷锟斤拷锟轿裁达拷顺锟斤拷锟斤拷锟斤拷虿换岜伙拷锟斤拷锟斤拷锟�

为什么要锟斤拷 m_chunksInUse 锟斤拷锟介？原锟斤拷锟斤拷锟铰ｏ拷锟斤拷锟角匡拷锟斤拷 tryAllocate 锟斤拷锟截碉拷锟斤拷锟斤拷息锟节达拷锟斤拷指锟诫，锟斤拷锟斤拷息锟斤拷锟酵碉拷时锟斤拷锟斤拷要使锟斤拷 SharedChunk 锟斤拷锟斤拷锟斤拷锟睫凤拷锟斤拷前锟斤拷转锟斤拷为锟斤拷锟竭★拷锟斤拷锟皆ｏ拷锟剿达拷锟斤拷锟斤拷锟斤拷锟介，锟斤拷息锟斤拷锟酵猴拷锟斤拷锟斤拷通锟斤拷锟斤拷息锟节达拷锟斤拷指锟斤拷锟斤拷叶锟接︼拷锟斤拷锟皆拷兀锟斤拷指锟斤拷锟� SharedChunk 实锟斤拷锟斤拷锟斤拷锟斤拷锟�3.3 锟斤拷

3 锟斤拷息锟斤拷锟斤拷锟竭硷拷

锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷息锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷息锟斤拷锟斤拷锟结构 ShmSafeUnmanagedChunk 锟斤拷

3.1 PublisherImpl::publish

职锟斤拷

锟较诧拷应锟矫筹拷锟斤拷锟斤拷么朔锟斤拷锟斤拷锟斤拷锟斤拷锟较拷锟�

锟斤拷危锟�

sample 锟斤拷锟矫伙拷锟斤拷锟斤拷锟斤拷锟捷的凤拷装实锟斤拷锟斤拷

template 
inline void PublisherImpl::publish(Sample&& sample) noexcept
{
    auto userPayload = sample.release(); // release the Samples ownership of the chunk before publishing
    auto chunkHeader = mepoo::ChunkHeader::fromUserPayload(userPayload);
    port().sendChunk(chunkHeader);
}

锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷

锟斤拷锟斤拷锟斤拷锟斤拷锟� sample 锟斤拷取锟斤拷锟矫伙拷锟斤拷锟斤拷锟斤拷锟斤拷指锟诫，锟捷此硷拷锟斤拷 Chunk 锟阶碉拷址锟斤拷然锟斤拷锟斤拷锟� sendChunk 锟斤拷锟叫凤拷锟酵★拷

锟斤拷锟斤拷锟矫伙拷锟斤拷锟斤拷锟斤拷锟斤拷指锟斤拷锟斤拷锟� Chunk 锟阶碉拷址锟斤拷实锟斤拷锟角硷拷去一锟斤拷偏锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷惴斤拷锟斤拷锟斤拷拢锟�

ChunkHeader* ChunkHeader::fromUserPayload(void* const userPayload) noexcept
{
    if (userPayload == nullptr)
    {
        return nullptr;
    }
    uint64_t userPayloadAddress = reinterpret_cast(userPayload);
    auto backOffset = reinterpret_cast(userPayloadAddress - sizeof(UserPayloadOffset_t));
    return reinterpret_cast(userPayloadAddress - *backOffset);
}

锟斤拷锟斤拷偏锟狡凤拷锟斤拷payload之前锟斤拷锟斤拷锟斤拷 *backOffset 锟斤拷

3.2 PublisherPortUser::sendChunk

职锟斤拷

锟斤拷锟斤拷锟矫伙拷锟斤拷锟捷★拷

锟斤拷危锟�

chunkHeader 锟斤拷 ChunkHeader 锟斤拷锟酵碉拷指锟诫， Chunk 锟斤拷锟阶碉拷址锟斤拷

void PublisherPortUser::sendChunk(mepoo::ChunkHeader* const chunkHeader) noexcept
{
    const auto offerRequested = getMembers()->m_offeringRequested.load(std::memory_order_relaxed);

    if (offerRequested)
    {
        m_chunkSender.send(chunkHeader);
    }
    else
    {
        m_chunkSender.pushToHistory(chunkHeader);
    }
}

锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷

3.3 ChunkSender::send

职锟斤拷

锟斤拷锟斤拷锟矫伙拷锟斤拷锟捷★拷

锟斤拷危锟�

chunkHeader 锟斤拷 ChunkHeader 指锟诫， Chunk 锟斤拷锟阶碉拷址锟斤拷

template 
inline uint64_t ChunkSender::send(mepoo::ChunkHeader* const chunkHeader) noexcept
{
    uint64_t numberOfReceiverTheChunkWasDelivered{0};
    mepoo::SharedChunk chunk(nullptr);
    // BEGIN of critical section, chunk will be lost if the process terminates in this section
    if (getChunkReadyForSend(chunkHeader, chunk))
    {
        numberOfReceiverTheChunkWasDelivered = this->deliverToAllStoredQueues(chunk);

        getMembers()->m_lastChunkUnmanaged.releaseToSharedChunk();
        getMembers()->m_lastChunkUnmanaged = chunk;
    }
    // END of critical section

    return numberOfReceiverTheChunkWasDelivered;
}

锟斤拷未锟斤拷锟斤拷锟斤拷锟斤拷

LINE 05 锟斤拷 LINE 07锟斤拷 锟斤拷锟斤拷 chunkHeader 指锟斤拷锟� m_chunksInUse 锟斤拷锟介，锟街革拷 SharedChunk 实锟斤拷锟斤拷
LINE 09 锟斤拷 LINE 09锟斤拷 锟斤拷锟矫伙拷锟斤拷某锟皆憋拷锟斤拷锟� deliverToAllStoredQueues 锟斤拷锟斤拷锟斤拷蟹锟斤拷停锟斤拷锟斤拷耄╋拷锟较拷锟�
LINE 11 锟斤拷 LINE 12锟斤拷 锟斤拷锟斤拷 m_lastChunkUnmanaged 实锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟杰★拷

3.4 ChunkDistributor::deliverToAllStoredQueues

template 
inline uint64_t ChunkDistributor::deliverToAllStoredQueues(mepoo::SharedChunk chunk) noexcept
{
    uint64_t numberOfQueuesTheChunkWasDeliveredTo{0U};
    typename ChunkDistributorDataType::QueueContainer_t remainingQueues;

    /* * * * *  锟斤拷锟斤拷锟斤拷锟�3-3-1锟斤拷锟斤拷锟斤拷蟹锟斤拷锟斤拷锟较拷锟绞э拷锟斤拷锟絩emainingQueues  * * * * */

    /* * * * *  锟斤拷锟斤拷锟斤拷锟�3-3-2锟斤拷锟斤拷锟斤拷失锟杰的诧拷锟较筹拷锟斤拷锟斤拷锟铰凤拷锟斤拷  * * * * */

    addToHistoryWithoutDelivery(chunk);

    return numberOfQueuesTheChunkWasDeliveredTo;
}

锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷

锟解部锟斤拷没锟斤拷什么锟斤拷锟捷ｏ拷锟斤拷要实锟斤拷锟节达拷锟斤拷锟�3-3-1锟酵达拷锟斤拷锟�3-3-2锟斤拷

锟斤拷锟斤拷锟�3-3-1锟斤拷

{
    {
    typename MemberType_t::LockGuard_t lock(*getMembers());

    bool willWaitForConsumer = getMembers()->m_consumerTooSlowPolicy == ConsumerTooSlowPolicy::WAIT_FOR_CONSUMER;
    // send to all the queues
    for (auto& queue : getMembers()->m_queues)
    {
        bool isBlockingQueue = (willWaitForConsumer && queue->m_queueFullPolicy == QueueFullPolicy::BLOCK_PRODUCER);

        if (pushToQueue(queue.get(), chunk))
        {
            ++numberOfQueuesTheChunkWasDeliveredTo;
        }
        else
        {
            if (isBlockingQueue)
            {
                remainingQueues.emplace_back(queue);
            }
            else
            {
                ++numberOfQueuesTheChunkWasDeliveredTo;
                ChunkQueuePusher_t(queue.get()).lostAChunk();
            }
        }
    }
}

锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷

锟斤拷未锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷潜锟斤拷锟斤拷锟斤拷卸锟斤拷锟斤拷叨锟斤拷校锟斤拷锟斤拷锟� pushToQueue 锟斤拷锟斤拷息锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷息锟斤拷实锟斤拷锟斤拷息锟斤拷锟酵★拷锟斤拷锟斤拷锟斤拷息锟斤拷锟叫的筹拷锟斤拷锟斤拷锟斤拷锟睫的ｏ拷锟斤拷锟斤拷锟斤拷诙锟斤拷锟斤拷叽锟斤拷锟斤拷俣锟教拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟接︼拷锟斤拷锟矫达拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷茫锟斤拷锟斤拷锟窖★拷锟斤拷锟斤拷锟接︼拷圆锟斤拷裕锟�

锟斤拷锟斤拷锟叫憋拷锟斤拷锟斤拷锟斤拷锟斤拷LINE 17 锟斤拷 LINE 20锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷些锟斤拷锟叫诧拷锟较筹拷锟皆凤拷锟酵ｏ拷直锟斤拷锟斤拷锟叫讹拷锟斤拷锟斤拷锟酵成癸拷锟斤拷锟斤拷锟斤拷锟斤拷锟�3-3-2锟斤拷
锟斤拷锟斤拷锟叫憋拷锟轿� 锟斤拷锟斤拷息锟斤拷失 锟斤拷LINE 22 锟斤拷 LINE 25锟斤拷锟斤拷

template 
inline void ChunkQueuePusher::lostAChunk() noexcept
{
    getMembers()->m_queueHasLostChunks.store(true, std::memory_order_relaxed);
}

锟斤拷锟斤拷锟�3-3-2锟斤拷锟斤拷锟较筹拷锟皆凤拷锟酵ｏ拷直锟斤拷锟斤拷锟斤拷锟斤拷息锟斤拷锟酵成癸拷

cxx::internal::adaptive_wait adaptiveWait;
while (!remainingQueues.empty())
{
    adaptiveWait.wait();
    {
        typename MemberType_t::LockGuard_t lock(*getMembers());

        /* * * * *  锟斤拷锟斤拷锟斤拷锟�3-3-3锟斤拷锟斤拷锟皆撅拷锟斤拷锟斤拷锟�  * * * * */

        for (uint64_t i = remainingQueues.size() - 1U; !remainingQueues.empty(); --i)
        {
            if (pushToQueue(remainingQueues[i].get(), chunk))
            {
                remainingQueues.erase(remainingQueues.begin() + i);
                ++numberOfQueuesTheChunkWasDeliveredTo;
            }

            if (i == 0U)
            {
                break;
            }
        }
    }
}

锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷

锟解部锟街达拷锟斤拷锟斤拷嵌锟绞ｏ拷锟轿达拷锟斤拷统晒锟斤拷亩锟斤拷薪锟斤拷锟斤拷锟斤拷路锟斤拷停锟街憋拷锟斤拷锟斤拷卸锟斤拷蟹锟斤拷统晒锟斤拷锟矫匡拷殖锟斤拷锟斤拷屑锟斤拷使锟斤拷yield锟斤拷sleep锟斤拷锟斤拷锟饺达拷一锟斤拷时锟戒，锟斤拷锟解不锟斤拷要锟斤拷锟斤拷锟斤拷锟剿费★拷同时锟斤拷锟斤拷锟酵癸拷锟斤拷锟叫ｏ拷锟斤拷锟斤拷锟诫当前锟斤拷跃锟斤拷锟斤拷锟襟交ｏ拷锟斤拷锟铰ｏ拷

锟斤拷锟斤拷锟�3-3-3锟斤拷锟斤拷锟皆撅拷锟斤拷锟斤拷锟�

typename ChunkDistributorDataType::QueueContainer_t queueIntersection(remainingQueues.size());
auto greaterThan = [](memory::RelativePointer& a,
                  memory::RelativePointer& b) -> bool {
return reinterpret_cast(a.get()) > reinterpret_cast(b.get());
};
std::sort(getMembers()->m_queues.begin(), getMembers()->m_queues.end(), greaterThan);
std::sort(remainingQueues.begin(), remainingQueues.end(), greaterThan);

auto iter = std::set_intersection(getMembers()->m_queues.begin(),
                              getMembers()->m_queues.end(),
                              remainingQueues.begin(),
                              remainingQueues.end(),
                              queueIntersection.begin(),
                              greaterThan);
queueIntersection.resize(static_cast(iter - queueIntersection.begin()));
remainingQueues = queueIntersection;

锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷

锟斤拷锟斤拷锟斤拷未锟斤拷锟斤拷锟斤拷锟斤拷锟� remainingQueues 锟酵碉拷前锟斤拷跃锟斤拷锟斤拷 m_queues 锟斤拷锟斤拷锟斤拷锟斤拷锟解发锟斤拷锟斤拷锟斤拷循锟斤拷锟斤拷 set_intersection 锟斤拷C++锟斤拷准锟解函锟斤拷锟斤拷锟斤拷锟斤拷锟� https://en.cppreference.com/w/cpp/algorithm/set_intersection

锟斤拷锟剿ｏ拷锟斤拷息锟斤拷锟酵碉拷锟斤拷锟教凤拷锟斤拷锟斤拷稀锟�