database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2026-10-23 21:49:05 +03:00
Code Activity

fix(BRM): MCOL-6434 - BRM access race fix

Browse Source 
In heavy cpimport/query processing loads the access races were observed.
This particular patch attempts to prevent that by using much more tested
upgradable mutext implementation from boost::interprocess.
This commit is contained in:
Serguey Zefirov
2026-07-13 12:43:20 +01:00
committed by Sergey Zefirov
parent 3b0b78547b
commit 8fe717abe7
2 changed files with 92 additions and 447 deletions
Download Patch File Download Diff File Expand all files Collapse all files

519
utils/rwlock/rwlock.cpp
View File

@@ -161,9 +161,7 @@ RWLockShmImpl::RWLockShmImpl(int key, bool excl)
else
fState->writing = 0;
new (&fState->sems[RWLock::MUTEX]) bi::interprocess_semaphore(1);
new (&fState->sems[RWLock::READERS]) bi::interprocess_semaphore(0);
new (&fState->sems[RWLock::WRITERS]) bi::interprocess_semaphore(0);
new (&fState->rwMutex) bi::interprocess_upgradable_mutex();
}
catch (bi::interprocess_exception& e)
{
@@ -208,486 +206,144 @@ RWLock::~RWLock()
{
}
void RWLock::down(int num, bool block)
class not_implemented : public std::exception
{
again:
try
public:
const char* what() const noexcept override
{
if (block)
fPImpl->fState->sems[num].wait();
else
{
bool gotIt = fPImpl->fState->sems[num].try_wait();
if (!gotIt)
throw wouldblock();
}
}
catch (bi::interprocess_exception& bipe)
{
ostringstream os;
os << "RWLock::down(): caught boost ipe: " << bipe.what() << " key = " << fPImpl->keyString()
<< " error code = " << bipe.get_error_code();
if (bipe.get_error_code() == 1) // it passes through EINTR apparently
goto again;
runtime_error rex(os.str());
cerr << bipe.what() << endl;
throw rex;
}
catch (std::exception& ex)
{
cerr <<
__PRETTY_FUNCTION__ <<
":" << __LINE__ << ": caught an exception: " << ex.what() << endl;
throw;
}
catch (...)
{
cerr <<
__PRETTY_FUNCTION__ <<
":" << __LINE__ << ": caught an exception" << endl;
throw runtime_error("RWLock::down(): caught an exception");
return "not implemented";
}
};
void RWLock::down(int /*num*/, bool /*block*/)
{
throw not_implemented();
}
bool RWLock::timed_down(int num, const ptime& delay)
bool RWLock::timed_down(int /*num*/, const ptime& /*delay*/)
{
bool gotTheLock = false;
// cout << "timed_down: current time = " << to_simple_string(microsec_clock::local_time()) <<
// " wake time = " << to_simple_string(delay) << endl;
again:
try
{
// I don't think I've seen timed_wait() ever wait, need to do this 'manually'
// gotTheLock = fPImpl->fState->sems[num].timed_wait(delay);
do
{
gotTheLock = fPImpl->fState->sems[num].try_wait();
if (!gotTheLock)
usleep(100000);
} while (!gotTheLock && microsec_clock::local_time() < delay);
}
catch (boost::thread_interrupted&)
{
// no need to do anything here
}
catch (bi::interprocess_exception& bipe)
{
ostringstream os;
os << "RWLock::timed_down(): caught boost ipe: " << bipe.what() << " key = " << fPImpl->keyString()
<< " error code = " << bipe.get_error_code();
if (bipe.get_error_code() == 1) // it passes through EINTR apparently
goto again;
runtime_error rex(os.str());
cerr << bipe.what() << endl;
throw rex;
}
catch (std::exception& ex)
{
cerr <<
__PRETTY_FUNCTION__ <<
":" << __LINE__ << ": caught an exception: " << ex.what() << endl;
throw;
}
catch (...)
{
cerr <<
__PRETTY_FUNCTION__ <<
":" << __LINE__ << ": caught an exception" << endl;
throw runtime_error("RWLock::timed_down(): caught an exception");
}
return gotTheLock;
throw not_implemented();
return false;
}
void RWLock::up(int num)
void RWLock::up(int /*num*/)
{
try
{
fPImpl->fState->sems[num].post();
}
catch (bi::interprocess_exception& bipe)
{
ostringstream os;
os << "RWLock::up(): caught boost ipe: " << bipe.what() << " key = " << fPImpl->keyString();
runtime_error rex(os.str());
cerr << bipe.what() << endl;
throw rex;
}
catch (...)
{
throw runtime_error("RWLock::up(): caught an exception");
}
throw not_implemented();
}
void RWLock::read_lock(bool block)
{
down(MUTEX, true);
CHECKSAFETY();
CHECKLIVENESS();
if (fPImpl->fState->writerswaiting > 0 || fPImpl->fState->writing > 0)
if (!block)
{
if (!block)
if (!fPImpl->fState->rwMutex.try_lock_sharable())
{
up(MUTEX);
throw wouldblock();
}
fPImpl->fState->readerswaiting++;
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
down(READERS); // unblocked by write_unlock();
#ifdef RWLOCK_DEBUG
down(MUTEX, true);
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
#endif
}
else
{
fPImpl->fState->reading++;
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
fPImpl->fState->reading += 1;
return ;
}
fPImpl->fState->readerswaiting += 1;
fPImpl->fState->rwMutex.lock_sharable();
fPImpl->fState->readerswaiting -= 1;
fPImpl->fState->reading += 1;
return ;
}
void RWLock::read_lock_priority(bool block)
{
down(MUTEX, true);
CHECKSAFETY();
CHECKLIVENESS();
if (fPImpl->fState->writing > 0)
{
if (!block)
{
up(MUTEX);
throw wouldblock();
}
fPImpl->fState->readerswaiting++;
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
down(READERS); // unblocked by write_unlock();
#ifdef RWLOCK_DEBUG
down(MUTEX, true);
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
#endif
}
else
{
fPImpl->fState->reading++;
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
}
// will redirect to read_lock, as the functionality is quite close.
read_lock(block);
}
void RWLock::read_unlock()
{
down(MUTEX, true);
CHECKSAFETY();
CHECKLIVENESS();
/* Made this a bit more tolerant of errors b/c the lock recovery code can technically
* be wrong. In the practically zero chance that happens, better to take the chance
* on a read race than a sure assertion failure
*/
if (fPImpl->fState->reading > 0)
{
--fPImpl->fState->reading;
if (fPImpl->fState->writerswaiting > 0 && fPImpl->fState->reading == 0)
{
--fPImpl->fState->writerswaiting;
fPImpl->fState->writing++;
up(WRITERS);
}
}
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
fPImpl->fState->rwMutex.unlock_sharable();
fPImpl->fState->reading -= 1;
}
void RWLock::write_lock(bool block)
{
down(MUTEX, true);
CHECKSAFETY();
CHECKLIVENESS();
if (fPImpl->fState->writing > 0 || fPImpl->fState->reading > 0)
if (!block)
{
if (!block)
if (!fPImpl->fState->rwMutex.try_lock())
{
up(MUTEX);
throw wouldblock();
}
fPImpl->fState->writerswaiting++;
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
down(WRITERS); // unblocked by write_unlock() or read_unlock()
#ifdef RWLOCK_DEBUG
down(MUTEX, true);
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
#endif
}
else
{
fPImpl->fState->writing++;
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
fPImpl->fState->writing += 1;
return ;
}
// XXX Time to check, time to use??? (TOCTOU, google it)
fPImpl->fState->writerswaiting += 1;
fPImpl->fState->rwMutex.lock();
fPImpl->fState->writerswaiting -= 1;
fPImpl->fState->writing += 1;
return ;
}
// this exists only for the sake of code cleanup
#define RETURN_STATE(mutex_state, state) \
if (state) \
{ \
state->mutexLocked = mutex_state; \
state->readerswaiting = fPImpl->fState->readerswaiting; \
state->reading = fPImpl->fState->reading; \
state->writerswaiting = fPImpl->fState->writerswaiting; \
state->writing = fPImpl->fState->writing; \
#define RETURN_STATE(mutex_state, state) \
if (state) \
{ \
state->mutexLocked = mutex_state; \
state->readerswaiting = fPImpl->fState->readerswaiting.load(); \
state->reading = fPImpl->fState->reading.load(); \
state->writerswaiting = fPImpl->fState->writerswaiting.load(); \
state->writing = fPImpl->fState->writing.load(); \
}
bool RWLock::timed_write_lock(const struct timespec& ts, struct LockState* state)
{
bool gotIt, gotIt2;
ptime delay;
fPImpl->fState->writerswaiting += 1;
delay = microsec_clock::local_time() + seconds(ts.tv_sec) + microsec(ts.tv_nsec / 1000);
gotIt = timed_down(MUTEX, delay);
if (!gotIt)
if (fPImpl->fState->rwMutex.timed_lock(delay))
{
RETURN_STATE(true, state)
return false;
}
CHECKSAFETY();
CHECKLIVENESS();
if (fPImpl->fState->writing > 0 || fPImpl->fState->reading > 0)
{
fPImpl->fState->writerswaiting++;
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
gotIt = timed_down(WRITERS, delay);
if (!gotIt)
{
// need to grab the mutex again to revoke the lock request
// need to adjust the timeout value to make sure the attempt is made.
delay = microsec_clock::local_time() + seconds(10);
gotIt2 = timed_down(MUTEX, delay);
if (!gotIt2)
{
RETURN_STATE(true, state);
return false;
}
/* This thread could have been granted the write lock during that second
lock grab attempt. Observation: the thread that granted it up'd the writers sem,
but didn't wake this thread. If there were other writers waiting, one of those
woke. At this point, if writerswaiting > 0, this thread should consider itself
one that's still waiting and should back out of the request.
If writerswaiting == 0, this thread was the one granted the write lock, and
needs to take possession.
*/
if (fPImpl->fState->writerswaiting == 0)
{
try
{
down(WRITERS, false);
}
catch (const wouldblock&)
{
// Somehow another writer was able to jump in front. This is "impossible".
RETURN_STATE(false, state);
up(MUTEX);
return false;
}
up(MUTEX);
return true;
}
fPImpl->fState->writerswaiting--;
// need to unblock whatever was blocked by this lock attempt
if (fPImpl->fState->writing == 0 && fPImpl->fState->writerswaiting == 0)
{
fPImpl->fState->reading += fPImpl->fState->readerswaiting;
while (fPImpl->fState->readerswaiting > 0)
{
--fPImpl->fState->readerswaiting;
up(READERS);
}
}
// else if there's an active writer, do nothing
CHECKSAFETY();
CHECKLIVENESS();
RETURN_STATE(false, state);
up(MUTEX);
return false;
}
else
return true;
fPImpl->fState->writerswaiting -= 1;
fPImpl->fState->writing += 1;
RETURN_STATE(false, state);
return true;
}
else
{
fPImpl->fState->writing++;
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
return true;
fPImpl->fState->writerswaiting -= 1;
RETURN_STATE(false, state);
return false;
}
}
void RWLock::write_unlock()
{
down(MUTEX, true);
CHECKSAFETY();
CHECKLIVENESS();
assert(fPImpl->fState->writing > 0);
--fPImpl->fState->writing;
if (fPImpl->fState->writerswaiting > 0)
{
--fPImpl->fState->writerswaiting;
fPImpl->fState->writing++;
up(WRITERS);
}
else if (fPImpl->fState->readerswaiting > 0)
{
// let up to state->readerswaiting readers go
fPImpl->fState->reading = fPImpl->fState->readerswaiting;
while (fPImpl->fState->readerswaiting > 0)
{
--fPImpl->fState->readerswaiting;
up(READERS);
}
}
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
fPImpl->fState->rwMutex.unlock();
fPImpl->fState->writing -= 1;
return ;
}
void RWLock::upgrade_to_write()
{
down(MUTEX, true);
CHECKSAFETY();
CHECKLIVENESS();
/* Made this a bit more tolerant of errors b/c the lock recovery code can technically
* be wrong. In the practically zero chance that happens, better to take the chance
* on a read race than a sure assertion failure
*/
if (fPImpl->fState->reading > 0)
fPImpl->fState->writerswaiting += 1;
if (fPImpl->fState->rwMutex.try_unlock_sharable_and_lock())
{
--fPImpl->fState->reading;
// try to cut in line
// On entry we hold a read lock, so reading > 0, and at this point, reading >= 0
if (fPImpl->fState->reading == 0)
{
fPImpl->fState->writing++;
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
return;
}
}
// cut & paste from write_lock()
// The invariants hold that, on entry, writing == 0 and, by now, reading > 0, so this branch is always taken
if (fPImpl->fState->writing > 0 || fPImpl->fState->reading > 0)
{
fPImpl->fState->writerswaiting++;
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
down(WRITERS, true); // unblocked by write_unlock() or read_unlock()
#ifdef RWLOCK_DEBUG
down(MUTEX, true);
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
#endif
}
else // don't think we can get here: reading > 0 && writing == 0 && reading == 0
{
fPImpl->fState->writing++;
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
fPImpl->fState->writerswaiting -= 1;
fPImpl->fState->writing += 1;
fPImpl->fState->reading -= 1;
return ;
}
fPImpl->fState->rwMutex.unlock_sharable();
fPImpl->fState->reading -= 1;
fPImpl->fState->rwMutex.lock();
fPImpl->fState->writing += 1;
fPImpl->fState->writerswaiting -= 1;
}
/* It's safe (and necessary) to simply convert this writer to a reader without
blocking */
void RWLock::downgrade_to_read()
{
down(MUTEX, true);
CHECKSAFETY();
CHECKLIVENESS();
assert(fPImpl->fState->writing > 0);
--fPImpl->fState->writing;
if (fPImpl->fState->readerswaiting > 0)
{
// let up to state->readerswaiting readers go
fPImpl->fState->reading = fPImpl->fState->readerswaiting;
while (fPImpl->fState->readerswaiting > 0)
{
--fPImpl->fState->readerswaiting;
up(READERS);
}
}
fPImpl->fState->reading++;
CHECKSAFETY();
CHECKLIVENESS();
up(MUTEX);
fPImpl->fState->readerswaiting += 1;
fPImpl->fState->rwMutex.unlock_and_lock_sharable();
fPImpl->fState->readerswaiting -= 1;
fPImpl->fState->writing -= 1;
fPImpl->fState->reading += 1;
return ;
}
void RWLock::reset()
@@ -696,32 +352,19 @@ void RWLock::reset()
fPImpl->fState->readerswaiting = 0;
fPImpl->fState->reading = 0;
fPImpl->fState->writing = 0;
for (int i = 0; i < 3; i++)
{
while (fPImpl->fState->sems[i].try_wait())
{
}
}
fPImpl->fState->sems[MUTEX].post();
// XXX Reset mutex?
}
LockState RWLock::getLockState()
{
bool gotIt;
LockState ret;
gotIt = fPImpl->fState->sems[MUTEX].try_wait();
ret.reading = fPImpl->fState->reading;
ret.writing = fPImpl->fState->writing;
ret.readerswaiting = fPImpl->fState->readerswaiting;
ret.writerswaiting = fPImpl->fState->writerswaiting;
ret.mutexLocked = !gotIt;
if (gotIt)
fPImpl->fState->sems[MUTEX].post();
// XXX Possible TOCTOU.
ret.reading = fPImpl->fState->reading.load();
ret.writing = fPImpl->fState->writing.load();
ret.readerswaiting = fPImpl->fState->readerswaiting.load();
ret.writerswaiting = fPImpl->fState->writerswaiting.load();
ret.mutexLocked = false;
return ret;
}

20
utils/rwlock/rwlock.h
View File

@@ -28,6 +28,7 @@
#pragma once
#include <array>
#include <atomic>
#include <unistd.h>
#include <stdexcept>
@@ -36,6 +37,7 @@
#include <boost/interprocess/mapped_region.hpp>
#include <boost/interprocess/sync/interprocess_semaphore.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/interprocess/sync/interprocess_upgradable_mutex.hpp>
#define EXPORT
@@ -48,11 +50,11 @@ const std::array<const std::string, 7> RWLockNames = {{
/// the layout of the shmseg
struct State
{
int writerswaiting;
int writing;
int readerswaiting;
int reading;
boost::interprocess::interprocess_semaphore sems[3];
std::atomic_int writerswaiting;
std::atomic_int writing;
std::atomic_int readerswaiting;
std::atomic_int reading;
boost::interprocess::interprocess_upgradable_mutex rwMutex; // sharable is reading, exclusive is writing.
};
/* the lock state without the semaphores, passed out by timed_write_lock() for
@@ -228,19 +230,19 @@ class RWLock
}
inline int getWriting() const
{
return fPImpl->fState->writing;
return fPImpl->fState->writing.load();
}
inline int getReading() const
{
return fPImpl->fState->reading;
return fPImpl->fState->reading.load();
}
inline int getWritersWaiting() const
{
return fPImpl->fState->writerswaiting;
return fPImpl->fState->writerswaiting.load();
}
inline int getReadersWaiting() const
{
return fPImpl->fState->readerswaiting;
return fPImpl->fState->readerswaiting.load();
}
LockState getLockState();