| 711 | SendWithHighPriority(addr, ms); |
| 712 | } |
| 713 | void DoSends() { |
| 714 | { |
| 715 | TBreakRequest rb; |
| 716 | while (BreakReqList.Dequeue(&rb)) { |
| 717 | InRequests.erase(rb.ReqGuid); |
| 718 | } |
| 719 | } |
| 720 | { |
| 721 | // cancelling requests |
| 722 | TCancelRequest rc; |
| 723 | while (CancelReqList.Dequeue(&rc)) { |
| 724 | TOutRequestHash::iterator i = OutRequests.find(rc.ReqGuid); |
| 725 | if (i == OutRequests.end()) { |
| 726 | AnticipateCancels.insert(rc.ReqGuid); |
| 727 | continue; // cancelling non existing request is ok |
| 728 | } |
| 729 | TOutRequestState& s = i->second; |
| 730 | if (s.State == TOutRequestState::S_SENDING) { |
| 731 | // we are in trouble - have not sent request and we already have to cancel it, wait send |
| 732 | s.State = TOutRequestState::S_CANCEL_AFTER_SENDING; |
| 733 | } else { |
| 734 | DoSendCancel(s.Address, rc.ReqGuid); |
| 735 | FinishRequest(i, TUdpHttpResponse::CANCELED, nullptr, "request canceled: notify requested side"); |
| 736 | } |
| 737 | } |
| 738 | } |
| 739 | { |
| 740 | // sending replies |
| 741 | for (TSendResponse* rd = nullptr; SendRespList.Dequeue(&rd); delete rd) { |
| 742 | TInRequestHash::iterator i = InRequests.find(rd->ReqGuid); |
| 743 | if (i == InRequests.end()) { |
| 744 | Y_ASSERT(0); |
| 745 | continue; |
| 746 | } |
| 747 | TInRequestState& s = i->second; |
| 748 | if (s.State == TInRequestState::S_CANCELED) { |
| 749 | // need not send response for the canceled request |
| 750 | InRequests.erase(i); |
| 751 | continue; |
| 752 | } |
| 753 | |
| 754 | Y_ASSERT(s.State == TInRequestState::S_WAITING); |
| 755 | s.State = TInRequestState::S_RESPONSE_SENDING; |
| 756 | //printf("InReq %s SendResponse() ... -> S_RESPONSE_SENDING (pkt %s)\n", GetGuidAsString(reqId).c_str(), GetGuidAsString(lowPktGuid).c_str()); |
| 757 | |
| 758 | TAutoPtr<TRopeDataPacket> ms = new TRopeDataPacket; |
| 759 | ui32 crc32 = 0; |
| 760 | int dataSize = rd->Data.ysize(); |
| 761 | if (rd->Data.ysize() > MIN_SHARED_MEM_PACKET && IsLocalFast(s.Address)) { |
| 762 | TIntrusivePtr<TSharedMemory> shm = new TSharedMemory; |
| 763 | if (shm->Create(dataSize)) { |
| 764 | ms->Write((char)PKT_LOCAL_RESPONSE); |
| 765 | ms->Write(rd->ReqGuid); |
| 766 | memcpy(shm->GetPtr(), &rd->Data[0], dataSize); |
| 767 | TVector<char> empty; |
| 768 | rd->Data.swap(empty); |
| 769 | ms->AttachSharedData(shm); |
| 770 | crc32 = CalcChecksum(ms->GetChain()); |
nothing calls this directly
no test coverage detected