| 160 | } |
| 161 | |
| 162 | void TContextDistributor::GotResponse(int id, TVector<char>*) { |
| 163 | CHROMIUM_TRACE_FUNCTION(); |
| 164 | TGuard<TMutex> g(Sync); |
| 165 | THashMap<int, TTransferInfo>::iterator i = TransferInfos.find(id); |
| 166 | Y_ASSERT(i != TransferInfos.end()); |
| 167 | const TTransferInfo& xfer = i->second; |
| 168 | |
| 169 | --ComputerSendCount[xfer.SenderComp + 1]; |
| 170 | |
| 171 | TFullCtxInfo& info = EnvId2Info[xfer.EnvId]; |
| 172 | if (xfer.Version == info.MaxVersion) { |
| 173 | // new comp is up to date |
| 174 | Y_ASSERT(info.ReadyMask[xfer.DstComp][xfer.Part] == false); |
| 175 | info.ReadyMask[xfer.DstComp][xfer.Part] = true; |
| 176 | PAR_DEBUG_LOG << Sprintf("Comp %d confirmed env %d version %d part %d\n", xfer.DstComp, xfer.EnvId, xfer.Version, xfer.Part); |
| 177 | } |
| 178 | TransferInfos.erase(i); |
| 179 | |
| 180 | DoSend(); |
| 181 | AtomicAdd(ActiveReqCount, -1); |
| 182 | } |
| 183 | |
| 184 | TContextDistributor::TContextDistributor(TRemoteQueryProcessor* queryProc, TLocalDataBuffer* writeBuffer) |
| 185 | : QueryProc(queryProc) |