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

fix(DEC):MCOL-5805,5808 to resolve UM-only node crash inside DEC when there is no local PP to send the local requests to. (#3349)

Browse Source
This commit is contained in:
drrtuy
2024-11-11 18:29:53 +00:00
committed by GitHub
parent 6f4274760e
commit c6b747b7c0
2 changed files with 23 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
dbcon/joblist/distributedenginecomm.cpp
View File

@@ -181,6 +181,17 @@ struct QueueShutdown
namespace joblist namespace joblist
{ {
bool isDefaultLocalConnectionId(const uint32_t connectionId)
{
return defaultLocalConnectionId() == connectionId;
}
bool needToSendToLocalPM(const bool isExeMgr, const uint32_t connectionId)
{
return isExeMgr && !isDefaultLocalConnectionId(connectionId);
}
DistributedEngineComm* DistributedEngineComm::fInstance = 0; DistributedEngineComm* DistributedEngineComm::fInstance = 0;
/*static*/ /*static*/
@@ -725,7 +736,7 @@ void DistributedEngineComm::sendAcks(uint32_t uniqueID, const vector<SBS>& msgs,
nextPMToACK(mqe, l_msgCount, &sockIndex, toAck); nextPMToACK(mqe, l_msgCount, &sockIndex, toAck);
idbassert(*toAck <= l_msgCount); idbassert(*toAck <= l_msgCount);
l_msgCount -= *toAck; l_msgCount -= *toAck;
if (sockIndex == localConnectionId_ && fIsExeMgr) if (sockIndex == localConnectionId_ && needToSendToLocalPM(fIsExeMgr, sockIndex))
{ {
sendToLocal = true; sendToLocal = true;
continue; continue;
@@ -766,7 +777,7 @@ void DistributedEngineComm::sendAcks(uint32_t uniqueID, const vector<SBS>& msgs,
writeToClient(i, ackCommand); writeToClient(i, ackCommand);
} }
} }
if (!pmAcked[localConnectionId_] && fIsExeMgr) if (needToSendToLocalPM(fIsExeMgr, localConnectionId_) && !pmAcked[localConnectionId_])
{ {
writeToClient(localConnectionId_, msg); writeToClient(localConnectionId_, msg);
} }
@@ -857,9 +868,11 @@ void DistributedEngineComm::setFlowControl(bool enabled, uint32_t uniqueID, boos
} }
writeToClient(i, msg); writeToClient(i, msg);
} }
if (fIsExeMgr) if (needToSendToLocalPM(fIsExeMgr, localConnectionId_))
{
writeToClient(localConnectionId_, msg); writeToClient(localConnectionId_, msg);
} }
}
int32_t DistributedEngineComm::write(uint32_t senderID, const SBS& msg) int32_t DistributedEngineComm::write(uint32_t senderID, const SBS& msg)
{ {
@@ -898,7 +911,7 @@ int32_t DistributedEngineComm::write(uint32_t senderID, const SBS& msg)
return rc; return rc;
} }
} }
if (fIsExeMgr) if (needToSendToLocalPM(fIsExeMgr, localConnectionId_))
{ {
return writeToClient(localConnectionId_, msg); return writeToClient(localConnectionId_, msg);
} }

7
dbcon/joblist/distributedenginecomm.h
View File

@@ -69,6 +69,11 @@ class Config;
*/ */
namespace joblist namespace joblist
{ {
constexpr uint32_t defaultLocalConnectionId()
{
return std::numeric_limits<uint32_t>::max();
}
class DECEventListener class DECEventListener
{ {
public: public:
@@ -315,7 +320,7 @@ class DistributedEngineComm
boost::mutex ackLock; boost::mutex ackLock;
// localConnectionId_ is set running Setup() method // localConnectionId_ is set running Setup() method
uint32_t localConnectionId_ = std::numeric_limits<uint32_t>::max(); uint32_t localConnectionId_ = defaultLocalConnectionId();
std::vector<struct in_addr> localNetIfaceSins_; std::vector<struct in_addr> localNetIfaceSins_;
std::mutex inMemoryEM2PPExchMutex_; std::mutex inMemoryEM2PPExchMutex_;
std::condition_variable inMemoryEM2PPExchCV_; std::condition_variable inMemoryEM2PPExchCV_;