From c070f6ece204afcfbedf4ee7539534a812715326 Mon Sep 17 00:00:00 2001
From: John Sully <john@csquare.ca>
Date: Mon, 12 Apr 2021 03:23:49 +0000
Subject: [PATCH] Don't prefetch when lock contention is low, it increases
 latency

Former-commit-id: 9b2629f6a20368cec8e55f0d006f3a67c8b770b7
---
 src/ae.cpp         | 16 ++++++++++++++++
 src/ae.h           |  2 ++
 src/networking.cpp |  3 ++-
 src/server.cpp     | 21 +++++++++++++++++++--
 src/server.h       |  5 +++++
 5 files changed, 44 insertions(+), 3 deletions(-)

diff --git a/src/ae.cpp b/src/ae.cpp
index d88328ded..f96ef4f6c 100644
--- a/src/ae.cpp
+++ b/src/ae.cpp
@@ -870,3 +870,19 @@ int aeThreadOwnsLock()
 {
     return g_lock.fOwnLock();
 }
+
+int aeLockContested(int threshold)
+{
+    return g_lock.m_ticket.m_active < static_cast<uint16_t>(g_lock.m_ticket.m_avail - threshold);
+}
+
+int aeLockContention()
+{
+    ticket ticketT;
+    __atomic_load(&g_lock.m_ticket.u, &ticketT.u, __ATOMIC_RELAXED);
+    int32_t avail = ticketT.m_avail;
+    int32_t active = ticketT.m_active;
+    if (avail < active)
+        avail += 0x10000;
+    return avail - active;
+}
\ No newline at end of file
diff --git a/src/ae.h b/src/ae.h
index 3a240877d..aec1df154 100644
--- a/src/ae.h
+++ b/src/ae.h
@@ -169,6 +169,8 @@ void aeAcquireLock();
 int aeTryAcquireLock(int fWeak);
 void aeReleaseLock();
 int aeThreadOwnsLock();
+int aeLockContested(int threshold);
+int aeLockContention(); // returns the number of instantaneous threads waiting on the lock
 
 #ifdef __cplusplus
 }
diff --git a/src/networking.cpp b/src/networking.cpp
index b8d3ef0b2..58ff54bac 100644
--- a/src/networking.cpp
+++ b/src/networking.cpp
@@ -2371,7 +2371,8 @@ void parseClientCommandBuffer(client *c) {
         }
 
         /* Prefetch outside the lock for better perf */
-        if (g_pserver->prefetch_enabled && cqueriesStart < c->vecqueuedcmd.size() && !GlobalLocksAcquired()) {
+        if (g_pserver->prefetch_enabled && cqueriesStart < c->vecqueuedcmd.size() &&
+            (g_pserver->m_pstorageFactory || aeLockContested(cserver.cthreads/2)) && !GlobalLocksAcquired()) {
             auto &query = c->vecqueuedcmd.back();
             if (query.argc > 0 && query.argc == query.argcMax) {
                 if (c->db->prefetchKeysAsync(c, query, c->vecqueuedcmd.size() == 1)) {
diff --git a/src/server.cpp b/src/server.cpp
index a323053f9..fbcd0cc43 100644
--- a/src/server.cpp
+++ b/src/server.cpp
@@ -2452,6 +2452,14 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
         }
     }
 
+    // Measure lock contention from a different thread to be more accurate
+    g_pserver->asyncworkqueue->AddWorkFunction([]{
+        g_pserver->rglockSamples[g_pserver->ilockRingHead] = (uint16_t)aeLockContention();
+        ++g_pserver->ilockRingHead;
+        if (g_pserver->ilockRingHead >= redisServer::s_lockContentionSamples)
+            g_pserver->ilockRingHead = 0;
+    });
+
     g_pserver->cronloops++;
     return 1000/g_pserver->hz;
 }
@@ -5138,6 +5146,11 @@ sds genRedisInfoString(const char *section) {
 
     /* Stats */
     if (allsections || defsections || !strcasecmp(section,"stats")) {
+        double avgLockContention = 0;
+        for (unsigned i = 0; i < redisServer::s_lockContentionSamples; ++i)
+            avgLockContention += g_pserver->rglockSamples[i];
+        avgLockContention /= redisServer::s_lockContentionSamples;
+
         if (sections++) info = sdscat(info,"\r\n");
         info = sdscatprintf(info,
             "# Stats\r\n"
@@ -5173,7 +5186,9 @@ sds genRedisInfoString(const char *section) {
             "tracking_total_prefixes:%lld\r\n"
             "unexpected_error_replies:%lld\r\n"
             "total_reads_processed:%lld\r\n"
-            "total_writes_processed:%lld\r\n",
+            "total_writes_processed:%lld\r\n"
+            "instantaneous_lock_contention:%d\r\n"
+            "avg_lock_contention:%f\r\n",
             g_pserver->stat_numconnections,
             g_pserver->stat_numcommands,
             getInstantaneousMetric(STATS_METRIC_COMMAND),
@@ -5206,7 +5221,9 @@ sds genRedisInfoString(const char *section) {
             (unsigned long long) trackingGetTotalPrefixes(),
             g_pserver->stat_unexpected_error_replies,
             g_pserver->stat_total_reads_processed.load(std::memory_order_relaxed),
-            g_pserver->stat_total_writes_processed.load(std::memory_order_relaxed));
+            g_pserver->stat_total_writes_processed.load(std::memory_order_relaxed),
+            aeLockContention(),
+            avgLockContention);
     }
 
     /* Replication */
diff --git a/src/server.h b/src/server.h
index 8619ee28e..bb895a814 100644
--- a/src/server.h
+++ b/src/server.h
@@ -2404,6 +2404,11 @@ struct redisServer {
     long long repl_batch_offStart = -1;
     long long repl_batch_idxStart = -1;
 
+    /* Lock Contention Ring Buffer */
+    static const size_t s_lockContentionSamples = 64;
+    uint16_t rglockSamples[s_lockContentionSamples];
+    unsigned ilockRingHead = 0;
+
     bool FRdbSaveInProgress() const { return rdbThreadVars.fRdbThreadActive; }
 };