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

fix(pp-threadpool): MCOL-5565 the workaround for a stuck tests001 in CI (#2947)

Browse Source 
CI ocassionaly stuck running test001 b/c PP threadpool endlessly reschedules
    meta jobs, e.g. BATCH_PRIMITIVE_CREATE, which ByteStreams were somehow damaged or read out.

Co-authored-by: Leonid Fedorov <leonid.fedorov@mariadb.com>
This commit is contained in:
drrtuy
2023-09-01 17:09:44 +02:00
committed by GitHub
parent a0e7bf4fd6
commit 2802e2810f
3 changed files with 19 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
primitives/primproc/primitiveserver.cpp
View File

@ -31,7 +31,7 @@
#include <mutex>
#include <stdexcept>
//#define NDEBUG
// #define NDEBUG
#include <cassert>
#include <boost/thread.hpp>
#include <boost/thread/condition.hpp>
@ -350,7 +350,6 @@ uint32_t loadBlocks(LBID_t* lbids, QueryContext qc, VER_t txn, int compType, uin
*blocksWereVersioned = false;
if (LBIDTrace)
{
for (i = 0; i < blockCount; i++)
@ -412,7 +411,6 @@ uint32_t loadBlocks(LBID_t* lbids, QueryContext qc, VER_t txn, int compType, uin
{
prefetchBlocks(lbids[0], compType, &blksRead);
if (fPMProfOn)
pmstats.markEvent(lbids[0], (pthread_t)-1, sessionID, 'M');
@ -496,11 +494,9 @@ void loadBlock(uint64_t lbid, QueryContext v, uint32_t t, int compType, void* bu
uint32_t blksRead = 0;
VSSCache::iterator it;
if (LBIDTrace)
stats.touchedLBID(lbid, pthread_self(), sessionID);
if (vssCache)
{
it = vssCache->find(lbid);
@ -556,14 +552,12 @@ void loadBlock(uint64_t lbid, QueryContext v, uint32_t t, int compType, void* bu
SUMMARY_INFO2("open failed: ", fileNamePtr);
char errbuf[80];
string errMsg;
//#if STRERROR_R_CHAR_P
// #if STRERROR_R_CHAR_P
const char* p;
if ((p = strerror_r(errCode, errbuf, 80)) != 0)
errMsg = p;
if (errCode == EINVAL)
{
throw logging::IDBExcept(logging::IDBErrorInfo::instance()->errorMsg(logging::ERR_O_DIRECT),
@ -811,7 +805,6 @@ void loadBlock(uint64_t lbid, QueryContext v, uint32_t t, int compType, void* bu
{
prefetchBlocks(lbid, compType, &blksRead);
if (fPMProfOn)
pmstats.markEvent(lbid, (pthread_t)-1, sessionID, 'M');
@ -983,7 +976,7 @@ void loadBlockAsync(uint64_t lbid, const QueryContext& c, uint32_t txn, int comp
} // namespace primitiveprocessor
//#define DCT_DEBUG 1
// #define DCT_DEBUG 1
#define SETUP_GUARD \
{ \
unsigned char* o = outputp.get(); \
@ -1565,6 +1558,14 @@ struct BPPHandler
int destroyBPP(ByteStream& bs, const posix_time::ptime& dieTime)
{
// This is a corner case that damages bs so its length becomes less than a header length.
// The damaged bs doesn't pass the if that checks bs at least has header + 3x int32_t.
// The if block below works around the issue.
if (posix_time::second_clock::universal_time() > dieTime)
{
return 0;
}
uint32_t uniqueID, sessionID, stepID;
BPPMap::iterator it;
@ -1944,7 +1945,7 @@ struct ReadThread
const uint32_t txnId = *((uint32_t*)&buf[pos + 2]);
const uint32_t stepID = *((uint32_t*)&buf[pos + 6]);
const uint32_t uniqueID = *((uint32_t*)&buf[pos + 10]);
const uint32_t weight = 1;
const uint32_t weight = threadpool::MetaJobsInitialWeight;
const uint32_t priority = 0;
uint32_t id = 0;
boost::shared_ptr<FairThreadPool::Functor> functor;