Function AsyncStartGroupRequest

library/cpp/par/par_mr.cpp:214–248

/////////////////////////////////////////////////////////////////////

    //////////////////////////////////////////////////////////////////////////
    void AsyncStartGroupRequest(TJobDescription* descr,
                                TRemoteQueryProcessor* queryProc, IUserContext* userContext,
                                IMRCommandCompleteNotify* mrNotify) {
        TIntrusivePtr<TJobRequest> jr = new TJobRequest;
        jr->Descr.Cmds.swap(descr->Cmds);
        jr->Descr.ParamsData.swap(descr->ParamsData);
        jr->Descr.ParamsPtr.swap(descr->ParamsPtr);
        jr->EnvId2Version = userContext->GetEnvId2Version();
        jr->IsLowPriority = false;

        int hostIdCount = userContext->GetHostIdCount();

        for (int i = 0; i < descr->ExecList.ysize(); ++i) {
            TJobParams jp = descr->ExecList[i];
            if (jp.HostId == TJobDescription::MAP_HOST_ID) {
                // its a map request, assign to several hosts
                for (int hostId = 0; hostId < hostIdCount; ++hostId) {
                    jp.HostId = hostId;
                    jr->Descr.ExecList.push_back(jp);
                }
            } else {
                // direct request
                Y_ABORT_UNLESS(
                    (jp.HostId >= 0 && jp.HostId < hostIdCount) || jp.HostId == TJobDescription::ANYWHERE_HOST_ID,
                    "jp.HostId=%d, hostIdCount=%d",
                    jp.HostId, hostIdCount);
                jr->Descr.ExecList.push_back(jp);
            }
        }

        if (queryProc)
            TSplitMRExec::Launch(jr.Get(), queryProc, userContext, mrNotify);
        else
            LaunchLocalJobRequest(jr.Get(), -1, userContext, mrNotify);
    }
}