Method MetaThreadFunction

library/cpp/par/par_remote.cpp:473–520

    }

    void TRemoteQueryProcessor::MetaThreadFunction() {
        while (DoRun) {
            TNetworkEvent netEvent;
            while (NetworkEventsQueue.Dequeue(&netEvent)) {
                if (netEvent.EventType == TNetworkEvent::EType::IcomingQueryCancel) {
                    TIntrusivePtr<TQueryResultDst> queryInfoPtr;
                    if (!IncomingRequestsData.ExtractValueIfPresent(netEvent.ReqId, queryInfoPtr)) {
                        continue;
                    }
                    for (auto& callback : queryInfoPtr->CallbackVector) {
                        callback->OnCancel();
                    }
                } else if (netEvent.EventType == TNetworkEvent::EType::IncomingQuery) {
                    if (!CmdProcessors.contains(netEvent.Request->Url)) {
                        if (netEvent.Request->Url == "check_stop") {
                            TVector<char> tmp;
                            tmp.push_back(RequestsData.Empty());
                            Requester->SendResponse(netEvent.Request->ReqId, &tmp);
                        } else {
                            Y_ASSERT(0);
                        }
                    } else {
                        IncomingRequestsData.EmplaceValue(netEvent.Request->ReqId, new TQueryResultDst);
                        CmdProcessors.at(netEvent.Request->Url)->NewRequest(this, netEvent.Request.Get());
                    }
                } else if (netEvent.EventType == TNetworkEvent::EType::ReplyReceived) {
                    TIntrusivePtr<TQueryResultDst> queryInfoPtr;
                    if (!RequestsData.ExtractValueIfPresent(netEvent.Response->ReqId, queryInfoPtr)) {
                        continue;
                    }

                    if (netEvent.Response->Status == TNetworkResponse::EStatus::Canceled) {
                        PAR_DEBUG_LOG << "At " << Requester->GetHostAndPort() << " Query " << GetGuidAsString(netEvent.Response->ReqId) << " cancelled" << Endl;
                    } else if (netEvent.Response->Status == TNetworkResponse::EStatus::Ok) {
                        static TTiming& queryFullTime = TParHostStats::GetTiming(ETimingTag::QueryFullTime);
                        queryFullTime += (TInstant::Now() - queryInfoPtr->QueryCreationTime).SecondsFloat();
                        queryInfoPtr->Proc->GotResponse(queryInfoPtr->Id, &netEvent.Response->Data);
                    } else {
                        Y_ABORT();
                    }
                }
            }
            if (!NetworkEventsQueue.IsEmpty()) {
                continue;
            }
            NetworkEvent.WaitT(TDuration::MilliSeconds(500));
        }
    }
}