add connection idle time support. if the connection has been idle for…

… a long time, reconnect it.
youkingkeep · Jul 29, 2021 · e2e1bbf · e2e1bbf
1 parent e8e12e2
commit e2e1bbf
Show file tree

Hide file tree

Showing 8 changed files with 56 additions and 11 deletions.
diff --git a/src/sw/redis++/async_connection.cpp b/src/sw/redis++/async_connection.cpp
@@ -67,7 +67,8 @@ AsyncConnection::AsyncConnection(const ConnectionOptions &opts,
         AsyncConnectionMode mode) :
     _opts(opts),
     _loop(loop),
-    _create_time(std::chrono::steady_clock::now()) {
+    _create_time(std::chrono::steady_clock::now()),
+    _last_active(std::chrono::steady_clock::now()) {
     assert(_loop != nullptr);
 
     switch (mode) {

diff --git a/src/sw/redis++/async_connection.h b/src/sw/redis++/async_connection.h
@@ -87,6 +87,11 @@ class AsyncConnection : public std::enable_shared_from_this<AsyncConnection> {
         return _create_time;
     }
 
+    auto last_active() const
+        -> std::chrono::time_point<std::chrono::steady_clock> {
+        return _last_active;
+    }
+
     void disconnect(std::exception_ptr err);
 
     template <typename Result, typename ResultParser>
@@ -123,6 +128,8 @@ class AsyncConnection : public std::enable_shared_from_this<AsyncConnection> {
     redisAsyncContext& _context() {
         assert(_ctx != nullptr);
 
+        _last_active = std::chrono::steady_clock::now();
+
         return *_ctx;
     }
 
@@ -179,6 +186,10 @@ class AsyncConnection : public std::enable_shared_from_this<AsyncConnection> {
     // The time that the connection is created.
     std::chrono::time_point<std::chrono::steady_clock> _create_time{};
 
+    // The time that the connection is created or the time that
+    // the connection is recently used, i.e. `_context()` is called.
+    std::atomic<std::chrono::time_point<std::chrono::steady_clock>> _last_active{};
+
     std::vector<std::unique_ptr<AsyncEvent>> _events;
 
     std::atomic<State> _state{State::NOT_CONNECTED};

diff --git a/src/sw/redis++/async_connection_pool.cpp b/src/sw/redis++/async_connection_pool.cpp
@@ -141,6 +141,7 @@ AsyncConnectionSPtr AsyncConnectionPool::fetch() {
     auto connection = _fetch();
 
     auto connection_lifetime = _pool_opts.connection_lifetime;
+    auto connection_idle_time = _pool_opts.connection_idle_time;
 
     if (_sentinel) {
         auto opts = _opts;
@@ -149,7 +150,7 @@ AsyncConnectionSPtr AsyncConnectionPool::fetch() {
 
         lock.unlock();
 
-        if (role_changed || _need_reconnect(*connection, connection_lifetime)) {
+        if (role_changed || _need_reconnect(*connection, connection_lifetime, connection_idle_time)) {
             try {
                 auto tmp_connection = sentinel.create(opts, shared_from_this(), _loop.get());
 
@@ -172,7 +173,7 @@ AsyncConnectionSPtr AsyncConnectionPool::fetch() {
 
     assert(connection);
 
-    if (_need_reconnect(*connection, connection_lifetime)) {
+    if (_need_reconnect(*connection, connection_lifetime, connection_idle_time)) {
         try {
             auto tmp_connection = _create();
 
@@ -310,18 +311,25 @@ void AsyncConnectionPool::_wait_for_connection(std::unique_lock<std::mutex> &loc
 }
 
 bool AsyncConnectionPool::_need_reconnect(const AsyncConnection &connection,
-                                    const std::chrono::milliseconds &connection_lifetime) const {
+                                    const std::chrono::milliseconds &connection_lifetime,
+                                    const std::chrono::milliseconds &connection_idle_time) const {
     if (connection.broken()) {
         return true;
     }
 
+    auto now = std::chrono::steady_clock::now();
     if (connection_lifetime > std::chrono::milliseconds(0)) {
-        auto now = std::chrono::steady_clock::now();
         if (now - connection.create_time() > connection_lifetime) {
             return true;
         }
     }
 
+    if (connection_idle_time > std::chrono::milliseconds(0)) {
+        if (now - connection.last_active() > connection_idle_time) {
+            return true;
+        }
+    }
+
     return false;
 }
 

diff --git a/src/sw/redis++/async_connection_pool.h b/src/sw/redis++/async_connection_pool.h
@@ -118,7 +118,8 @@ class AsyncConnectionPool : public std::enable_shared_from_this<AsyncConnectionP
     void _wait_for_connection(std::unique_lock<std::mutex> &lock);
 
     bool _need_reconnect(const AsyncConnection &connection,
-                            const std::chrono::milliseconds &connection_lifetime) const;
+                            const std::chrono::milliseconds &connection_lifetime,
+                            const std::chrono::milliseconds &connection_idle_time) const;
 
     void _update_connection_opts(const std::string &host, int port) {
         _opts.host = host;

diff --git a/src/sw/redis++/connection.cpp b/src/sw/redis++/connection.cpp
@@ -364,6 +364,7 @@ void swap(Connection &lhs, Connection &rhs) noexcept {
 Connection::Connection(const ConnectionOptions &opts) :
             _ctx(Connector(opts).connect()),
             _create_time(std::chrono::steady_clock::now()),
+            _last_active(std::chrono::steady_clock::now()),
             _opts(opts) {
     assert(_ctx && !broken());
 

diff --git a/src/sw/redis++/connection.h b/src/sw/redis++/connection.h
@@ -142,6 +142,11 @@ class Connection {
         return _create_time;
     }
 
+    auto last_active() const
+        -> std::chrono::time_point<std::chrono::steady_clock> {
+        return _last_active;
+    }
+
     template <typename ...Args>
     void send(const char *format, Args &&...args);
 
@@ -185,6 +190,10 @@ class Connection {
     // The time that the connection is created.
     std::chrono::time_point<std::chrono::steady_clock> _create_time{};
 
+    // The time that the connection is created or the time that
+    // the connection is recently used, i.e. `_context()` is called.
+    std::chrono::time_point<std::chrono::steady_clock> _last_active{};
+
     ConnectionOptions _opts;
 
     // TODO: define _tls_ctx before _ctx
@@ -216,6 +225,8 @@ inline void Connection::send(const char *format, Args &&...args) {
 }
 
 inline redisContext* Connection::_context() {
+    _last_active = std::chrono::steady_clock::now();
+
     return _ctx.get();
 }
 

diff --git a/src/sw/redis++/connection_pool.cpp b/src/sw/redis++/connection_pool.cpp
@@ -93,6 +93,7 @@ Connection ConnectionPool::fetch() {
     auto connection = _fetch();
 
     auto connection_lifetime = _pool_opts.connection_lifetime;
+    auto connection_idle_time = _pool_opts.connection_idle_time;
 
     if (_sentinel) {
         auto opts = _opts;
@@ -101,7 +102,7 @@ Connection ConnectionPool::fetch() {
 
         lock.unlock();
 
-        if (role_changed || _need_reconnect(connection, connection_lifetime)) {
+        if (role_changed || _need_reconnect(connection, connection_lifetime, connection_idle_time)) {
             try {
                 connection = _create(sentinel, opts, false);
             } catch (const Error &e) {
@@ -116,7 +117,7 @@ Connection ConnectionPool::fetch() {
 
     lock.unlock();
 
-    if (_need_reconnect(connection, connection_lifetime)) {
+    if (_need_reconnect(connection, connection_lifetime, connection_idle_time)) {
         try {
             connection.reconnect();
         } catch (const Error &e) {
@@ -248,18 +249,25 @@ void ConnectionPool::_wait_for_connection(std::unique_lock<std::mutex> &lock) {
 }
 
 bool ConnectionPool::_need_reconnect(const Connection &connection,
-                                    const std::chrono::milliseconds &connection_lifetime) const {
+                                    const std::chrono::milliseconds &connection_lifetime,
+                                    const std::chrono::milliseconds &connection_idle_time) const {
     if (connection.broken()) {
         return true;
     }
 
+    auto now = std::chrono::steady_clock::now();
     if (connection_lifetime > std::chrono::milliseconds(0)) {
-        auto now = std::chrono::steady_clock::now();
         if (now - connection.create_time() > connection_lifetime) {
             return true;
         }
     }
 
+    if (connection_idle_time > std::chrono::milliseconds(0)) {
+        if (now - connection.last_active() > connection_idle_time) {
+            return true;
+        }
+    }
+
     return false;
 }
 

diff --git a/src/sw/redis++/connection_pool.h b/src/sw/redis++/connection_pool.h
@@ -39,6 +39,9 @@ struct ConnectionPoolOptions {
 
     // Max lifetime of a connection. 0ms means we never expire the connection.
     std::chrono::milliseconds connection_lifetime{0};
+
+    // Max idle time of a connection. 0ms means we never expire the connection.
+    std::chrono::milliseconds connection_idle_time{0};
 };
 
 class ConnectionPool {
@@ -85,7 +88,8 @@ class ConnectionPool {
     void _wait_for_connection(std::unique_lock<std::mutex> &lock);
 
     bool _need_reconnect(const Connection &connection,
-                            const std::chrono::milliseconds &connection_lifetime) const;
+                            const std::chrono::milliseconds &connection_lifetime,
+                            const std::chrono::milliseconds &connection_idle_time) const;
 
     void _update_connection_opts(const std::string &host, int port) {
         _opts.host = host;