database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-08-07 03:22:57 +03:00
Code Activity

MCOL-5009 fix deadlock

Browse Source 
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.
This commit is contained in:
David Hall
2022-03-07 14:24:54 -06:00
parent b1d5a46f3f
commit 7ebd0d3b3c
3 changed files with 5 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
primitives/primproc/bppsendthread.cpp
View File

@@ -75,11 +75,9 @@ void BPPSendThread::sendResult(const Msg_t& msg, bool newConnection)
std::unique_lock<std::mutex> sl1(respondLock); std::unique_lock<std::mutex> sl1(respondLock);
while (currentByteSize >= maxByteSize && msgQueue.size() > 3 && !die) while (currentByteSize >= maxByteSize && msgQueue.size() > 3 && !die)
{ {
respondWait = true;
fProcessorPool->incBlockedThreads(); fProcessorPool->incBlockedThreads();
okToRespond.wait(sl1); okToRespond.wait(sl1);
fProcessorPool->decBlockedThreads(); fProcessorPool->decBlockedThreads();
respondWait = false;
} }
} }
if (die) if (die)
@@ -122,11 +120,9 @@ void BPPSendThread::sendResults(const vector<Msg_t>& msgs, bool newConnection)
std::unique_lock<std::mutex> sl1(respondLock); std::unique_lock<std::mutex> sl1(respondLock);
while (currentByteSize >= maxByteSize && msgQueue.size() > 3 && !die) while (currentByteSize >= maxByteSize && msgQueue.size() > 3 && !die)
{ {
respondWait = true;
fProcessorPool->incBlockedThreads(); fProcessorPool->incBlockedThreads();
okToRespond.wait(sl1); okToRespond.wait(sl1);
fProcessorPool->decBlockedThreads(); fProcessorPool->decBlockedThreads();
respondWait = false;
} }
} }
if (die) if (die)
@@ -278,7 +274,7 @@ void BPPSendThread::mainLoop()
msg[msgsSent].msg.reset(); msg[msgsSent].msg.reset();
} }
if (respondWait && currentByteSize < maxByteSize) if (fProcessorPool->blockedThreadCount() > 0 && currentByteSize < maxByteSize)
{ {
okToRespond.notify_one(); okToRespond.notify_one();
} }

1
primitives/primproc/bppsendthread.h
View File

@@ -120,7 +120,6 @@ class BPPSendThread
std::mutex ackLock; std::mutex ackLock;
std::condition_variable okToSend; std::condition_variable okToSend;
// Condition to prevent run away queue // Condition to prevent run away queue
bool respondWait;
std::mutex respondLock; std::mutex respondLock;
std::condition_variable okToRespond; std::condition_variable okToRespond;

4
utils/threadpool/prioritythreadpool.h
View File

@@ -113,6 +113,10 @@ class PriorityThreadPool
{ {
blockedThreads--; blockedThreads--;
} }
uint32_t blockedThreadCount()
{
return blockedThreads;
}
protected: protected:
private: private: