| 2963 | } |
| 2964 | |
| 2965 | [[noreturn]] void ImpalaServer::SessionMaintenance() { |
| 2966 | while (true) { |
| 2967 | { |
| 2968 | unique_lock<mutex> timeout_lock(session_timeout_lock_); |
| 2969 | if (session_timeout_set_.empty()) { |
| 2970 | session_timeout_cv_.Wait(timeout_lock); |
| 2971 | } else { |
| 2972 | // Sleep for a second before doing maintenance. |
| 2973 | session_timeout_cv_.WaitFor(timeout_lock, MICROS_PER_SEC); |
| 2974 | } |
| 2975 | } |
| 2976 | |
| 2977 | int64_t now = UnixMillis(); |
| 2978 | int expired_cnt = 0; |
| 2979 | VLOG(3) << "Session maintenance thread waking up"; |
| 2980 | { |
| 2981 | // TODO: If holding session_state_map_lock_ for the duration of this loop is too |
| 2982 | // expensive, consider a priority queue. |
| 2983 | lock_guard<mutex> map_lock(session_state_map_lock_); |
| 2984 | vector<TUniqueId> sessions_to_remove; |
| 2985 | for (SessionStateMap::value_type& map_entry : session_state_map_) { |
| 2986 | const TUniqueId& session_id = map_entry.first; |
| 2987 | std::shared_ptr<SessionState> session_state = map_entry.second; |
| 2988 | unordered_set<TUniqueId> inflight_queries; |
| 2989 | Status query_cancel_status; |
| 2990 | { |
| 2991 | lock_guard<mutex> state_lock(session_state->lock); |
| 2992 | if (session_state->ref_count > 0) continue; |
| 2993 | // A session closed by other means is in the process of being removed, and it's |
| 2994 | // best not to interfere. |
| 2995 | if (session_state->closed) continue; |
| 2996 | |
| 2997 | if (session_state->connections.size() == 0 |
| 2998 | && (now - session_state->disconnected_ms) |
| 2999 | >= FLAGS_disconnected_session_timeout * 1000L) { |
| 3000 | // This session has no active connections and is past the disconnected session |
| 3001 | // timeout, so close it. |
| 3002 | DCHECK(session_state->session_type == TSessionType::HIVESERVER2 || |
| 3003 | session_state->session_type == TSessionType::EXTERNAL_FRONTEND); |
| 3004 | LOG(INFO) << "Closing session: " << PrintId(session_id) |
| 3005 | << ", user: " << session_state->connected_user |
| 3006 | << ", because it no longer has any open connections. The last " |
| 3007 | << "connection was closed at: " |
| 3008 | << ToStringFromUnixMillis(session_state->disconnected_ms); |
| 3009 | session_state->closed = true; |
| 3010 | sessions_to_remove.push_back(session_id); |
| 3011 | ImpaladMetrics::IMPALA_SERVER_NUM_OPEN_HS2_SESSIONS->Increment(-1L); |
| 3012 | UnregisterSessionTimeout(FLAGS_disconnected_session_timeout); |
| 3013 | query_cancel_status = |
| 3014 | Status::Expected(TErrorCode::DISCONNECTED_SESSION_CLOSED); |
| 3015 | DecrementSessionCount(session_state->connected_user); |
| 3016 | } else { |
| 3017 | // Check if the session should be expired. |
| 3018 | if (session_state->expired || session_state->session_timeout == 0) { |
| 3019 | continue; |
| 3020 | } |
| 3021 | |
| 3022 | int64_t last_accessed_ms = session_state->last_accessed_ms; |
nothing calls this directly
no test coverage detected