From 8fe717abe7880a761fb2a208d16da8d477e6049d Mon Sep 17 00:00:00 2001 From: Serguey Zefirov Date: Mon, 13 Jul 2026 12:43:20 +0100 Subject: [PATCH] fix(BRM): MCOL-6434 - BRM access race fix 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. --- utils/rwlock/rwlock.cpp | 519 +++++++--------------------------------- utils/rwlock/rwlock.h | 20 +- 2 files changed, 92 insertions(+), 447 deletions(-) diff --git a/utils/rwlock/rwlock.cpp b/utils/rwlock/rwlock.cpp index fde95640c..e128746e0 100644 --- a/utils/rwlock/rwlock.cpp +++ b/utils/rwlock/rwlock.cpp @@ -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; } diff --git a/utils/rwlock/rwlock.h b/utils/rwlock/rwlock.h index 575c8854f..34923aaa2 100644 --- a/utils/rwlock/rwlock.h +++ b/utils/rwlock/rwlock.h @@ -28,6 +28,7 @@ #pragma once #include +#include #include #include @@ -36,6 +37,7 @@ #include #include #include +#include #define EXPORT @@ -48,11 +50,11 @@ const std::array 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();