Function RemoteMapReduceImpl

library/cpp/par/par_util.cpp:115–156

    const int N_MAX_PART_COUNT = 100;
    static void RemoteMapReduceImpl(TJobDescription* job, IDistrCmd* finalMap, ERROp op) {
        CHROMIUM_TRACE_FUNCTION();

        TObj<IDistrCmd> hold(finalMap);
        if (job->ExecList.empty())
            return;

        int jobCount = job->ExecList.ysize();
        int partCount = Min(jobCount, N_MAX_PART_COUNT);
        int jobPerPart = (jobCount + partCount - 1) / partCount;

        TVector<bool> hasData;
        hasData.resize(jobCount);

        TJobDescription newJob;
        {
            newJob.Cmds.resize(1);
            TObj<TRemoteReduce> rr = new TRemoteReduce(finalMap, op);
            SerializeToMem(&newJob.Cmds[0], rr);
        }
        newJob.ExecList.resize(partCount);

        for (int part = 0; part < partCount; ++part) {
            int startIdx = part * jobPerPart;
            int finishIdx = Min(startIdx + jobPerPart, jobCount);
            if (finishIdx <= startIdx) {
                newJob.ExecList.resize(part);
                break;
            }
            TJobDescription descr;
            TVector<int> resultMap;
            ProjectJob(&descr, startIdx, finishIdx - startIdx, &resultMap, &hasData, *job);
            int paramId = newJob.AddParam(&descr);
            TJobParams& jp = newJob.ExecList[part];
            jp = TJobParams(0, paramId, part, -1, TJobDescription::ANYWHERE_HOST_ID);
        }
        job->Swap(&newJob);
#ifdef _DEBUG
        for (int i = 0; i < hasData.ysize(); ++i)
            Y_ASSERT(hasData[i]);
#endif
    }

    void RemoteMap(TJobDescription* job, IDistrCmd* finalMap) {
        CHROMIUM_TRACE_FUNCTION();