| 135 | } |
| 136 | } |
| 137 | void StartReduce() { |
| 138 | ReduceReqCount = 1; |
| 139 | int totalJobCount = JobRequest->Descr.ExecList.ysize(); |
| 140 | RemapTable.resize(totalJobCount); |
| 141 | int reduceResultCount = 0; |
| 142 | for (int i = 0; i < totalJobCount;) { |
| 143 | int firstId = i; |
| 144 | int hasDataCount = CountHasData(&i); |
| 145 | |
| 146 | if (hasDataCount > 1) { |
| 147 | PAR_DEBUG_LOG << "Launch reduce task " << firstId << Endl; |
| 148 | int cmdId = JobRequest->Descr.ExecList[firstId].CmdId; |
| 149 | Cmds.Check(cmdId); |
| 150 | AtomicAdd(ReduceReqCount, 1); |
| 151 | if (JobRequest->IsLowPriority) |
| 152 | LocalExecutor().Exec(this, firstId, TLocalExecutor::MED_PRIORITY); |
| 153 | else |
| 154 | LocalExecutor().Exec(this, firstId, 0); |
| 155 | RemapTable[reduceResultCount++] = firstId; |
| 156 | } else |
| 157 | RemapTable[reduceResultCount++] = firstId; |
| 158 | } |
| 159 | if (reduceResultCount == totalJobCount) |
| 160 | RemapTable.resize(0); |
| 161 | else |
| 162 | RemapTable.resize(reduceResultCount); |
| 163 | DoneReduceTask(); |
| 164 | } |
| 165 | TReduceExec( |
| 166 | TJobRequest* jobRequest, |
| 167 | IMRCommandCompleteNotify* completeNotify, |