| 169 | } |
| 170 | |
| 171 | void UURLabShmRpcTransport::RunPollLoop() |
| 172 | { |
| 173 | FMjShmHeader* ReqHdr = static_cast<FMjShmHeader*>(ReqRegion.GetData()); |
| 174 | FMjShmHeader* RepHdr = static_cast<FMjShmHeader*>(RepRegion.GetData()); |
| 175 | if (!ReqHdr || !RepHdr) |
| 176 | return; |
| 177 | |
| 178 | uint64 LastSeenReqSeq = ReqHdr->Sequence.load(std::memory_order_acquire); |
| 179 | const uint32 ReqStride = ReqRegion.GetBufferStride(); |
| 180 | const uint32 RepStride = RepRegion.GetBufferStride(); |
| 181 | |
| 182 | while (!bStop.load(std::memory_order_acquire)) |
| 183 | { |
| 184 | #if PLATFORM_WINDOWS |
| 185 | // Block on the bridge's signal that req.shm has fresh content. |
| 186 | // 100 ms timeout so we re-check bStop periodically -- shutdown |
| 187 | // also calls SetEvent on this handle to wake immediately. Falls |
| 188 | // back to polling if event creation failed in TransportInit. |
| 189 | if (ReqReadyEvent) |
| 190 | { |
| 191 | const DWORD WaitResult = ::WaitForSingleObject( |
| 192 | static_cast<HANDLE>(ReqReadyEvent), /*ms=*/100); |
| 193 | if (bStop.load(std::memory_order_acquire)) |
| 194 | break; |
| 195 | // Whether the wait timed out or signalled, fall through to |
| 196 | // the sequence check -- we still need to verify there's a |
| 197 | // real request waiting. |
| 198 | (void)WaitResult; |
| 199 | } |
| 200 | else |
| 201 | { |
| 202 | FPlatformProcess::SleepNoStats(static_cast<float>(PollIntervalUs) * 1e-6f); |
| 203 | } |
| 204 | #else |
| 205 | FPlatformProcess::SleepNoStats(static_cast<float>(PollIntervalUs) * 1e-6f); |
| 206 | #endif |
| 207 | |
| 208 | const uint64 CurSeq = ReqHdr->Sequence.load(std::memory_order_acquire); |
| 209 | if (CurSeq == LastSeenReqSeq) |
| 210 | { |
| 211 | // Spurious wake or just a periodic timeout poll -- keep going. |
| 212 | continue; |
| 213 | } |
| 214 | |
| 215 | // Read the latest request slot. Producer wrote the slot, then |
| 216 | // bumped latest_idx, then bumped sequence -- so at this point |
| 217 | // both writes are visible. Re-check sequence after the copy to |
| 218 | // detect a torn read where the producer raced past us. |
| 219 | const uint32 Idx = ReqHdr->LatestIdx.load(std::memory_order_acquire); |
| 220 | const uint8* Slot = ReqRegion.GetSlot(Idx); |
| 221 | if (!Slot) |
| 222 | { |
| 223 | UE_LOG(LogURLab, Warning, |
| 224 | TEXT("ShmRpcTransport: dropping request seq=%llu with out-of-range latest_idx=%u (nbuffers=%u)"), |
| 225 | static_cast<unsigned long long>(CurSeq), Idx, ReqHdr->NBuffers); |
| 226 | LastSeenReqSeq = CurSeq; |
| 227 | continue; |
| 228 | } |