Simplification

dbms/programs/server/HTTPHandler.cpp

@@ -141,15 +141,15 @@ static Poco::Net::HTTPResponse::HTTPStatus exceptionCodeToHTTPStatus(int excepti
 }
 
 
-static std::chrono::steady_clock::duration parseSessionTimeout(
+static uint32_t parseSessionTimeout(
     const Poco::Util::AbstractConfiguration & config,
     const HTMLForm & params)
 {
-    unsigned session_timeout = config.getInt("default_session_timeout", 60);
+    uint32_t session_timeout = config.getInt("default_session_timeout", 60);
 
     if (params.has("session_timeout"))
     {
-        unsigned max_session_timeout = config.getUInt("max_session_timeout", 3600);
+        uint32_t max_session_timeout = config.getUInt("max_session_timeout", 3600);
         std::string session_timeout_str = params.get("session_timeout");
 
         ReadBufferFromString buf(session_timeout_str);
@@ -162,7 +162,7 @@ static std::chrono::steady_clock::duration parseSessionTimeout(
                 ErrorCodes::INVALID_SESSION_TIMEOUT);
     }
 
-    return std::chrono::seconds(session_timeout);
+    return session_timeout;
 }
 
 
@@ -275,7 +275,7 @@ void HTTPHandler::processQuery(
 
     std::shared_ptr<NamedSession> session;
     String session_id;
-    std::chrono::steady_clock::duration session_timeout;
+    uint32_t session_timeout;
     bool session_is_set = params.has("session_id");
     const auto & config = server.config();
 

dbms/src/Interpreters/Context.cpp

@@ -124,7 +124,7 @@ public:
     std::shared_ptr<NamedSession> acquireSession(
         const String & session_id,
         Context & context,
-        std::chrono::steady_clock::duration timeout,
+        uint32_t timeout,
         bool throw_if_not_found)
     {
         std::unique_lock lock(mutex);
@@ -162,7 +162,7 @@ public:
     void releaseSession(NamedSession & session)
     {
         std::unique_lock lock(mutex);
-        scheduleCloseSession(session, lock);
+        close_times.emplace(time(nullptr) + session.timeout, session.key);
     }
 
 private:
@@ -178,31 +178,11 @@ private:
         }
     };
 
-    /// TODO it's very complicated. Make simple std::map with time_t or boost::multi_index.
     using Container = std::unordered_map<Key, std::shared_ptr<NamedSession>, SessionKeyHash>;
-    using CloseTimes = std::deque<std::vector<Key>>;
+    using CloseTimes = std::multimap<time_t, Key>;
+
     Container sessions;
     CloseTimes close_times;
-    std::chrono::steady_clock::duration close_interval = std::chrono::seconds(1);
-    std::chrono::steady_clock::time_point close_cycle_time = std::chrono::steady_clock::now();
-    UInt64 close_cycle = 0;
-
-    void scheduleCloseSession(NamedSession & session, std::unique_lock<std::mutex> &)
-    {
-        /// Push it on a queue of sessions to close, on a position corresponding to the timeout.
-        /// (timeout is measured from current moment of time)
-
-        const UInt64 close_index = session.timeout / close_interval + 1;
-        const auto new_close_cycle = close_cycle + close_index;
-
-        if (session.close_cycle != new_close_cycle)
-        {
-            session.close_cycle = new_close_cycle;
-            if (close_times.size() < close_index + 1)
-                close_times.resize(close_index + 1);
-            close_times[close_index].emplace_back(session.key);
-        }
-    }
 
     void cleanThread()
     {
@@ -212,51 +192,32 @@ private:
 
         while (true)
         {
-            auto interval = closeSessions(lock);
-
-            if (cond.wait_for(lock, interval, [this]() -> bool { return quit; }))
+            closeSessions(lock);
+            if (cond.wait_for(lock, std::chrono::seconds(1), [this]() -> bool { return quit; }))
                 break;
         }
     }
 
     /// Close sessions, that has been expired. Returns how long to wait for next session to be expired, if no new sessions will be added.
-    std::chrono::steady_clock::duration closeSessions(std::unique_lock<std::mutex> & lock)
+    void closeSessions(std::unique_lock<std::mutex> &)
     {
-        const auto now = std::chrono::steady_clock::now();
-
-        /// The time to close the next session did not come
-        if (now < close_cycle_time)
-            return close_cycle_time - now;  /// Will sleep until it comes.
-
-        const auto current_cycle = close_cycle;
-
-        ++close_cycle;
-        close_cycle_time = now + close_interval;
+        time_t now = time(nullptr);
+        for (auto it = close_times.begin(); it != close_times.end();)
+        {
+            if (it->first >= now)
+                break;
 
-        if (close_times.empty())
-            return close_interval;
+            const auto session_it = sessions.find(it->second);
+            it = close_times.erase(it);
 
-        auto & sessions_to_close = close_times.front();
+            if (session_it == sessions.end())
+                continue;
 
-        for (const auto & key : sessions_to_close)
-        {
-            const auto session = sessions.find(key);
-
-            if (session != sessions.end() && session->second->close_cycle <= current_cycle)
-            {
-                if (!session->second.unique())
-                {
-                    /// Skip but move it to close on the next cycle.
-                    session->second->timeout = std::chrono::steady_clock::duration{0};
-                    scheduleCloseSession(*session->second, lock);
-                }
-                else
-                    sessions.erase(session);
-            }
+            if (session_it->second.unique())
+                sessions.erase(session_it);
+            else
+                close_times.emplace(now + session_it->second->timeout, session_it->second->key);    /// Does not invalidate iterators.
         }
-
-        close_times.pop_front();
-        return close_interval;
     }
 
     std::mutex mutex;
@@ -520,7 +481,7 @@ void Context::enableNamedSessions()
     shared->named_sessions.emplace();
 }
 
-std::shared_ptr<NamedSession> Context::acquireNamedSession(const String & session_id, std::chrono::steady_clock::duration timeout, bool session_check)
+std::shared_ptr<NamedSession> Context::acquireNamedSession(const String & session_id, uint32_t timeout, bool session_check)
 {
     if (!shared->named_sessions)
         throw Exception("Support for named sessions is not enabled", ErrorCodes::NOT_IMPLEMENTED);

dbms/src/Interpreters/Context.h

@@ -426,7 +426,7 @@ public:
     /// The method must be called at the server startup.
     void enableNamedSessions();
 
-    std::shared_ptr<NamedSession> acquireNamedSession(const String & session_id, std::chrono::steady_clock::duration timeout, bool session_check);
+    std::shared_ptr<NamedSession> acquireNamedSession(const String & session_id, uint32_t timeout, bool session_check);
 
     /// For methods below you may need to acquire a lock by yourself.
     std::unique_lock<std::recursive_mutex> getLock() const;
@@ -672,12 +672,11 @@ using NamedSessionKey = std::pair<String, String>;
 struct NamedSession
 {
     NamedSessionKey key;
-    UInt64 close_cycle = 0;
     Context context;
-    std::chrono::steady_clock::duration timeout;
+    uint32_t timeout;
     NamedSessions & parent;
 
-    NamedSession(NamedSessionKey key_, Context & context_, std::chrono::steady_clock::duration timeout_, NamedSessions & parent_)
+    NamedSession(NamedSessionKey key_, Context & context_, uint32_t timeout_, NamedSessions & parent_)
         : key(key_), context(context_), timeout(timeout_), parent(parent_)
     {
     }