database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-07-27 21:01:50 +03:00
Code Activity

MCOL-4841 Fix for MCOL-5009

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 15:25:00 -06:00
parent 27dea733c5
commit a98834e31c
3 changed files with 6 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
primitives/primproc/bppsendthread.cpp
View File

@ -76,11 +76,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)
@ -123,11 +121,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)
@ -279,7 +275,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

@ -118,7 +118,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

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