From 7ebd0d3b3c3b6c4a1d57f6053753d4891af8277e Mon Sep 17 00:00:00 2001
From: David Hall <david.hall@mariadb.com>
Date: Mon, 7 Mar 2022 14:24:54 -0600
Subject: [PATCH] MCOL-5009 fix deadlock respondWait could be set to false
 while other threads were waiting. With respondWait false, okToRrespond
 wouldn't ever get notify_one(). Get rid of respondWait and use
 fProcessorPool->blockedThreadCount to determine if any threads may be
 waiting.

---
 primitives/primproc/bppsendthread.cpp | 6 +-----
 primitives/primproc/bppsendthread.h   | 1 -
 utils/threadpool/prioritythreadpool.h | 4 ++++
 3 files changed, 5 insertions(+), 6 deletions(-)

diff --git a/primitives/primproc/bppsendthread.cpp b/primitives/primproc/bppsendthread.cpp
index 580598dfc..11b00ec6e 100644
--- a/primitives/primproc/bppsendthread.cpp
+++ b/primitives/primproc/bppsendthread.cpp
@@ -75,11 +75,9 @@ void BPPSendThread::sendResult(const Msg_t& msg, bool newConnection)
     std::unique_lock<std::mutex> sl1(respondLock);
     while (currentByteSize >= maxByteSize && msgQueue.size() > 3 && !die)
     {
-      respondWait = true;
       fProcessorPool->incBlockedThreads();
       okToRespond.wait(sl1);
       fProcessorPool->decBlockedThreads();
-      respondWait = false;
     }
   }
   if (die)
@@ -122,11 +120,9 @@ void BPPSendThread::sendResults(const vector<Msg_t>& msgs, bool newConnection)
     std::unique_lock<std::mutex> sl1(respondLock);
     while (currentByteSize >= maxByteSize && msgQueue.size() > 3 && !die)
     {
-      respondWait = true;
       fProcessorPool->incBlockedThreads();
       okToRespond.wait(sl1);
       fProcessorPool->decBlockedThreads();
-      respondWait = false;
     }
   }
   if (die)
@@ -278,7 +274,7 @@ void BPPSendThread::mainLoop()
         msg[msgsSent].msg.reset();
       }
 
-      if (respondWait && currentByteSize < maxByteSize)
+      if (fProcessorPool->blockedThreadCount() > 0 && currentByteSize < maxByteSize)
       {
         okToRespond.notify_one();
       }
diff --git a/primitives/primproc/bppsendthread.h b/primitives/primproc/bppsendthread.h
index deb6ff3d6..17994617a 100644
--- a/primitives/primproc/bppsendthread.h
+++ b/primitives/primproc/bppsendthread.h
@@ -120,7 +120,6 @@ class BPPSendThread
   std::mutex ackLock;
   std::condition_variable okToSend;
   // Condition to prevent run away queue
-  bool respondWait;
   std::mutex respondLock;
   std::condition_variable okToRespond;
 
diff --git a/utils/threadpool/prioritythreadpool.h b/utils/threadpool/prioritythreadpool.h
index ac5b74821..c37990416 100644
--- a/utils/threadpool/prioritythreadpool.h
+++ b/utils/threadpool/prioritythreadpool.h
@@ -113,6 +113,10 @@ class PriorityThreadPool
   {
     blockedThreads--;
   }
+  uint32_t blockedThreadCount()
+  {
+    return blockedThreads;
+  }
 
  protected:
  private: