Method LaunchOps

library/cpp/par/par_exec.h:229–318

        TGUID MasterGuid;

        void LaunchOps(int localCompId) {
            CHROMIUM_TRACE_FUNCTION();

            int mapJobCount = JobRequest->Descr.ExecList.ysize();
            LocalMapBuf.ResultData.resize(mapJobCount);
            LocalMapBuf.ResultHasData.resize(mapJobCount, false);
            MapJob2PartId.resize(mapJobCount, -1);

            bool localOnly = JobRequest->ExecPlan.empty();
            if (JobRequest->ExecPlan.ysize() == 1 && JobRequest->ExecPlan[0] == localCompId)
                localOnly = true;

            if (localOnly) {
                // local host only, launch local ops
                LocalUserContext = UserContext;

                LocalMapReqCount = 1;
                for (int i = 0; i < mapJobCount; ++i) {
                    Y_ASSERT(JobRequest->Descr.ExecList[i].CompId == localCompId);
                    int cmdId = JobRequest->Descr.ExecList[i].CmdId;
                    Cmds.Check(cmdId);
                }
                AtomicAdd(LocalMapReqCount, mapJobCount);
                int execRangeFlags = JobRequest->IsLowPriority ? TLocalExecutor::MED_PRIORITY : 0;
                LocalExecutor().ExecRange(this, 0, mapJobCount, execRangeFlags);
                DoneLocalMapTask();
            } else {
                LocalUserContext = UserContext->CreateLocalOnlyContext();

                RemoteMapReqCount = 1;

                // get subtasks
                TVector<TVector<ui16>> subTasks;
                GenerateSubtasks(JobRequest->ExecPlan, &subTasks);
                int partCount = subTasks.ysize();
                Y_ASSERT(partCount > 0); // one remote part is possible

                RemoteMapBuf.ResultData.resize(mapJobCount);
                RemoteMapBuf.ResultHasData.resize(mapJobCount, false);
                PartCompleted.resize(partCount, false);

                // split job on subtasks
                MapParts.resize(partCount);
                for (int i = 0; i < partCount; ++i) {
                    TJobRequest* jr = new TJobRequest;
                    TRemoteMapInfo* part = &MapParts[i];
                    ProjectJob(&jr->Descr, i, &part->ResultMap,
                               &RemoteMapBuf.ResultHasData, &MapJob2PartId,
                               JobRequest->Descr,
                               subTasks[i]);
                    jr->ExecPlan = subTasks[i];
                    jr->HostId2Computer = JobRequest->HostId2Computer;
                    jr->EnvId2Version = JobRequest->EnvId2Version;
                    jr->IsLowPriority = JobRequest->IsLowPriority;

                    part->JobRequest = jr;
                    part->DstHost = SelectRandomHost(jr->ExecPlan);
                }