| 148 | } |
| 149 | |
| 150 | void SetDataFromQueueLocked() { |
| 151 | for (TSetDataOp* op; SetQueue.Dequeue(&op); delete op) { |
| 152 | AtomicAdd(QueuedDataSize, -TotalSize(op->Data)); |
| 153 | TTableInfo& tbl = Tables[op->TblId]; |
| 154 | |
| 155 | if (tbl.Version > op->VersionId) { |
| 156 | continue; // ignore stale updates |
| 157 | } |
| 158 | if (tbl.Version < op->VersionId) { |
| 159 | tbl.Version = op->VersionId; |
| 160 | RemoveAllTableBlocks(op->TblId); |
| 161 | } |
| 162 | tbl.Blocks.push_back(op->Id); |
| 163 | if (op->Object) { |
| 164 | Data[op->Id].SetObject(op->Object); |
| 165 | } else { |
| 166 | Data[op->Id].SetData(op->Data); |
| 167 | } |
| 168 | } |
| 169 | } |
| 170 | void SetDataFromQueue() { |
| 171 | TGuard<TMutex> gg(Lock); |
| 172 | SetDataFromQueueLocked(); |