| 184 | } |
| 185 | |
| 186 | void QueryExecMgr::ReleaseQueryState(QueryState* qs) { |
| 187 | DCHECK(qs != nullptr); |
| 188 | TUniqueId query_id = qs->query_id(); |
| 189 | int32_t cnt = qs->refcnt_.Add(-1); |
| 190 | // don't reference anything from 'qs' beyond this point, 'qs' might get |
| 191 | // gc'd out from under us |
| 192 | qs = nullptr; |
| 193 | VLOG(2) << "ReleaseQueryState(): query_id=" << PrintId(query_id) |
| 194 | << " refcnt=" << cnt + 1; |
| 195 | DCHECK_GE(cnt, 0); |
| 196 | if (cnt > 0) return; |
| 197 | |
| 198 | QueryState* qs_from_map = nullptr; |
| 199 | { |
| 200 | ScopedShardedMapRef<QueryState*> map_ref(query_id, |
| 201 | &ExecEnv::GetInstance()->query_exec_mgr()->qs_map_); |
| 202 | DCHECK(map_ref.get() != nullptr); |
| 203 | |
| 204 | auto it = map_ref->find(query_id); |
| 205 | // someone else might have gc'd the entry |
| 206 | if (it == map_ref->end()) return; |
| 207 | qs_from_map = it->second; |
| 208 | DCHECK(qs_from_map->query_ctx().query_id == query_id); |
| 209 | int32_t cnt = qs_from_map->refcnt_.Load(); |
| 210 | DCHECK_GE(cnt, 0); |
| 211 | // someone else might have increased the refcnt in the meantime |
| 212 | if (cnt > 0) return; |
| 213 | map_ref->erase(it); |
| 214 | } |
| 215 | delete qs_from_map; |
| 216 | VLOG(1) << "ReleaseQueryState(): deleted query_id=" << PrintId(query_id); |
| 217 | // BACKEND_NUM_QUERIES_EXECUTING is used to detect the backend being quiesced, so we |
| 218 | // decrement it after we're completely done with the query. |
| 219 | ImpaladMetrics::BACKEND_NUM_QUERIES_EXECUTING->Increment(-1); |
| 220 | } |
| 221 | |
| 222 | void QueryExecMgr::AcquireQueryStateLocked(QueryState* qs) { |
| 223 | if (qs == nullptr) return; |