must be called only after succes aquiring output
| 745 | |
| 746 | //must be called only after succes aquiring output |
| 747 | void SendMessages(bool asioThread) { |
| 748 | //DBGOUT("SendMessages"); |
| 749 | if (Y_UNLIKELY(AtomicGet(State_) == Closed)) { |
| 750 | if (asioThread) { |
| 751 | OnError(Error_); |
| 752 | } else { |
| 753 | SafeOnError(); |
| 754 | } |
| 755 | return; |
| 756 | } |
| 757 | |
| 758 | do { |
| 759 | if (asioThread) { |
| 760 | NeedCheckCancelsQueue_.store(false); |
| 761 | TRequestId reqId; |
| 762 | |
| 763 | ProcessReqsInFlyQueue(); |
| 764 | while (Cancels_.Dequeue(&reqId)) { |
| 765 | TReqsInFly::iterator it = ReqsInFly_.find(reqId); |
| 766 | if (it == ReqsInFly_.end()) { |
| 767 | continue; |
| 768 | } |
| 769 | |
| 770 | ReqsInFly_.erase(it); |
| 771 | OutputBuffers_.AddCancelRequest(reqId); |
| 772 | if (Y_UNLIKELY(!OutputBuffers_.HasFreeSpace())) { |
| 773 | if (!FlushOutputBuffers(asioThread, 0)) { |
| 774 | return; |
| 775 | } |
| 776 | } |
| 777 | } |
| 778 | } else if (NeedCheckCancelsQueue_.load()) { |
| 779 | AS_.GetIOService().Post(std::bind(&TConnection::SendMessages, TConnectionRef(this), true)); |
| 780 | return; |
| 781 | } |
| 782 | |
| 783 | TRequestId lastReqId = 0; |
| 784 | { |
| 785 | NeedCheckReqsQueue_.store(false); |
| 786 | TRequest* reqPtr; |
| 787 | |
| 788 | while (Reqs_.Dequeue(&reqPtr)) { |
| 789 | TRequestRef reqTmp(reqPtr); |
| 790 | reqPtr->UnRef(); |
| 791 | reqPtr->SetReqId(GenerateReqId()); |
| 792 | if (reqPtr->Canceled()) { |
| 793 | continue; |
| 794 | } |
| 795 | lastReqId = reqPtr->ReqId(); |
| 796 | if (asioThread) { |
| 797 | TRequestRef& req = ReqsInFly_[(TRequestId)reqPtr->ReqId()]; |
| 798 | req.Swap(reqTmp); |
| 799 | OutputBuffers_.AddRequest(req); |
| 800 | } else { //can access to ReqsInFly_ only from asio thread, so enqueue req to update ReqsInFly_ queue |
| 801 | try { |
| 802 | reqTmp->Ref(); |
| 803 | ReqsInFlyQueue_.Enqueue(reqPtr); |
| 804 | } catch (...) { |
nothing calls this directly
no test coverage detected