| 81 | } |
| 82 | |
| 83 | void UURLabShmPublishTransport::Publish(const FString& /*Topic*/, |
| 84 | const TArray<uint8>& Bytes) |
| 85 | { |
| 86 | if (!StateRegion.IsOpen()) |
| 87 | return; |
| 88 | |
| 89 | FMjShmHeader* Hdr = static_cast<FMjShmHeader*>(StateRegion.GetData()); |
| 90 | if (!Hdr) |
| 91 | return; |
| 92 | |
| 93 | const uint32 Stride = StateRegion.GetBufferStride(); |
| 94 | if (static_cast<uint32>(Bytes.Num()) + sizeof(uint32) > Stride) |
| 95 | { |
| 96 | UE_LOG(LogURLabNet, Warning, |
| 97 | TEXT("UURLabShmPublishTransport: payload %d bytes exceeds slot stride %u; dropping"), |
| 98 | Bytes.Num(), Stride); |
| 99 | return; |
| 100 | } |
| 101 | |
| 102 | // Write into the slot the consumer is *not* reading. The header's |
| 103 | // latest_idx points at the last fully-published slot; we target the |
| 104 | // next one in ring order, populate it, then atomically flip |
| 105 | // latest_idx and bump the sequence counter. |
| 106 | const uint32 CurrentLatest = Hdr->LatestIdx.load(std::memory_order_acquire); |
| 107 | const uint32 NBuffers = Hdr->NBuffers > 0 ? Hdr->NBuffers : 1; |
| 108 | const uint32 Target = (CurrentLatest + 1) % NBuffers; |
| 109 | uint8* Slot = StateRegion.GetSlot(Target); |
| 110 | if (!Slot) |
| 111 | return; |
| 112 | |
| 113 | const uint32 PayloadSize = static_cast<uint32>(Bytes.Num()); |
| 114 | FMemory::Memcpy(Slot, &PayloadSize, sizeof(uint32)); |
| 115 | if (PayloadSize > 0) |
| 116 | { |
| 117 | FMemory::Memcpy(Slot + sizeof(uint32), Bytes.GetData(), PayloadSize); |
| 118 | } |
| 119 | |
| 120 | Hdr->LatestIdx.store(Target, std::memory_order_release); |
| 121 | Hdr->Sequence.fetch_add(1, std::memory_order_release); |
| 122 | } |