| 800 | SendWithSystemPriority(connection, ms, TTos(), DEFAULT_NETLIBA_COLOR); |
| 801 | } |
| 802 | void DoSends() { |
| 803 | { |
| 804 | TBreakRequest rb; |
| 805 | while (BreakReqList.Dequeue(&rb)) { |
| 806 | InRequests.erase(rb.ReqGuid); |
| 807 | } |
| 808 | } |
| 809 | { |
| 810 | // cancelling requests |
| 811 | TCancelRequest rc; |
| 812 | while (CancelReqList.Dequeue(&rc)) { |
| 813 | TOutRequestHash::iterator i = OutRequests.find(rc.ReqGuid); |
| 814 | if (i == OutRequests.end()) { |
| 815 | AnticipateCancels.insert(rc.ReqGuid); |
| 816 | continue; // cancelling non existing request is ok |
| 817 | } |
| 818 | TOutRequestState& s = i->second; |
| 819 | if (s.State == TOutRequestState::S_SENDING) { |
| 820 | // we are in trouble - have not sent request and we already have to cancel it, |
| 821 | // we send async request to TUdpHost to cancel transfer. |
| 822 | s.State = TOutRequestState::S_CANCEL_AT_SENDING; |
| 823 | Host->Cancel(TTransfer(s.Connection, s.TransferId)); |
| 824 | } else { |
| 825 | DoSendCancel(s.Connection, rc.ReqGuid); |
| 826 | FinishRequest(i, TUdpHttpResponse::CANCELED, nullptr, "request canceled: notify requested side"); |
| 827 | } |
| 828 | } |
| 829 | } |
| 830 | { |
| 831 | // sending replies |
| 832 | for (TSendResponse* rd = nullptr; SendRespList.Dequeue(&rd); delete rd) { |
| 833 | TInRequestHash::iterator i = InRequests.find(rd->ReqGuid); |
| 834 | if (i == InRequests.end()) { |
| 835 | Y_ASSERT(0); |
| 836 | continue; |
| 837 | } |
| 838 | TInRequestState& s = i->second; |
| 839 | if (s.State == TInRequestState::S_CANCELED) { |
| 840 | // need not send response for the canceled request |
| 841 | //printf("Do sends: request %s got canceled, erasing\n", GetGuidAsString(rd->ReqGuid).c_str()); |
| 842 | InRequests.erase(i); |
| 843 | continue; |
| 844 | } |
| 845 | |
| 846 | Y_ASSERT(s.State == TInRequestState::S_WAITING); |
| 847 | s.State = TInRequestState::S_RESPONSE_SENDING; |
| 848 | //printf("InReq %s SendResponse() ... -> S_RESPONSE_SENDING (pkt %s)\n", GetGuidAsString(reqId).c_str(), GetGuidAsString(lowPktGuid).c_str()); |
| 849 | |
| 850 | TAutoPtr<TRopeDataPacket> ms = new TRopeDataPacket; |
| 851 | int dataSize = rd->Data.ysize(); |
| 852 | if (rd->Data.ysize() > MIN_SHARED_MEM_PACKET && Host->IsLocal(s.Connection->GetAddress())) { |
| 853 | TIntrusivePtr<TPosixSharedMemory> shm = new TPosixSharedMemory; |
| 854 | if (shm->Create(dataSize)) { |
| 855 | ms->Write((char)PKT_LOCAL_RESPONSE); |
| 856 | ms->Write(rd->ReqGuid); |
| 857 | memcpy(shm->GetPtr(), &rd->Data[0], dataSize); |
| 858 | TVector<char> empty; |
| 859 | rd->Data.swap(empty); |
nothing calls this directly
no test coverage detected