| 1009 | } |
| 1010 | |
| 1011 | void TDualStackSocket::RecvLoop() { |
| 1012 | for (;;) { |
| 1013 | TUdpRecvPacket* p = nullptr; |
| 1014 | sockaddr_in6 srcAddr; |
| 1015 | sockaddr_in6 dstAddr; |
| 1016 | while (AtomicAdd(ShouldDie, 0) == 0 && (p = TBase::Recv(&srcAddr, &dstAddr, NETLIBA_ANY_VERSION))) { |
| 1017 | Y_ASSERT(p->DataStart == 0); |
| 1018 | if (p->DataSize < 12) { |
| 1019 | continue; |
| 1020 | } |
| 1021 | |
| 1022 | TFilteredPacketQueue& q = GetRecvQueue(p->Data.get()[8]); |
| 1023 | const ui8 res = q.Push(p, {srcAddr, dstAddr}); |
| 1024 | if (res == TFilteredPacketQueue::PR_OK) { |
| 1025 | GetQueueEvent(q).Signal(); |
| 1026 | } else { |
| 1027 | // simulate OS behavior on buffer overflow - drop packets. |
| 1028 | const NHPTimer::STime time = AtomicGet(RecvLag); |
| 1029 | const float sec = NHPTimer::GetSeconds(time); |
| 1030 | fprintf(stderr, "TDualStackSocket::RecvLoop netliba v%d queue overflow, recv lag: %f sec, dropping packet, res: %u\n", |
| 1031 | &q == &RecvQueue12 ? 12 : 6, sec, res); |
| 1032 | delete p; |
| 1033 | } |
| 1034 | } |
| 1035 | |
| 1036 | if (AtomicAdd(ShouldDie, 0)) { |
| 1037 | DieEvent.Signal(); |
| 1038 | return; |
| 1039 | } |
| 1040 | |
| 1041 | TBase::Wait(0.1f, NETLIBA_ANY_VERSION); |
| 1042 | } |
| 1043 | } |
| 1044 | |
| 1045 | void TDualStackSocket::Wait(float timeoutSec, int netlibaVersion) const { |
| 1046 | TFilteredPacketQueue& q = GetRecvQueue(netlibaVersion); |