I was looking for long time around to have a portable robust solution for multiprocessing synchronization. Who touche this things know that good solution are boost::iterprocess named sync objects.
When your process have a named_mutex locked and your process die (there are many normal situations when process die, not just bug or others.) In that case a named_mutex will remain in locked state. There were attempt to make a robust_mutex in boost code done by Ion Gaztanaga: Robust Emulation
He had a nice idea on how to resolve abandoning state check. Each process, in game, has its own lock file and while is alive it hold that file locked. Then Ion's robust_mutex check, in case of failed lock attempt, current owner process lock file, and can determine if current mutex owner is alive or not. In case it is death mutex can be taken. The trick with file lock is nice idea cause file locks are unlocked by OS in case process die, and this seems to be well portable. This solution wraps base spin_mutex and hold current owner process id in the internal field. I made intensive testing and found 2 big problems.
File lock handling and way how is implemented slows down mutex in manner that it is faster just use file lock.
Decoupling effective lock gate variable and owner process id cause situations where mutex can be stolen by different processes.
I'm proposing solution for both problems, and I wood like to have some pro opinion about it.
Do not use for each existing process separate lock file but to use one file for all eventual process id (there should be enough 4MB) and for each process lock just one byte. Position of that byte is determined by the process id itself. (this is not my idea but I found it in code of Howard Chu and his excellent LMDB)
Do not wrap
spin_mutexas is, but rewrite it's code so it use as lock gate current owner process id instead just 0/1, so lock and unlock can happen in one atomic CAS operation.
I did a try to implement it and tested on windows. I use original boost code and call boost where necessary. I post here content of principal file robust_mutex.hpp
//////////////////////////////////////////////////////////////////////////////
//
// (C) Copyright Ion Gaztanaga 2010-2012. Distributed under the Boost
// Software License, Version 1.0. (See accompanying file
// LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//
// See http://www.boost.org/libs/interprocess for documentation.
//
//////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////
//
// (C) Copyright Ladislav Sopko, 3D Informatica s.r.l 2014
//
//////////////////////////////////////////////////////////////////////////////
#if defined(_MSC_VER)&&(_MSC_VER>=1200)
#pragma once
#endif
#ifndef XW_IPC_ROBUST_SPIN_MUTEX_HPP
#define XW_IPC_ROBUST_SPIN_MUTEX_HPP
#include <boost/interprocess/detail/config_begin.hpp>
#include <boost/interprocess/detail/workaround.hpp>
#include <xw-incl/ipc/exception.hpp>
#include <xw-incl/ipc/os_file_functions_extended.hpp>
#include <boost/interprocess/detail/os_thread_functions.hpp>
#include <boost/interprocess/detail/posix_time_types_wrk.hpp>
#include <boost/interprocess/detail/atomic.hpp>
#include <boost/interprocess/sync/spin/wait.hpp>
#include <boost/cstdint.hpp>
#ifdef strRev
# undef strRev
#endif
#define strRev "$Revision: 1.7 $"
//2^22 of possible processes
#define MAX_PIDS_NUMBER 4194304
//#define _ROBUST_MUTEX_TRACE
#if defined(_DEBUG) && defined(_ROBUST_MUTEX_TRACE)
#define IPC_DBG_TRACE(msg) { std::stringstream __str; __str << bipc::ipcdetail::get_current_system_highres_count() << " " << msg; xw::ipc::dbg_tracer::trace(__str); }
#else
#define IPC_DBG_TRACE(msg)
#endif
// there are 3 mutex states, correct, fixing and broken, if mutex is broken it can be
// automatically put into fixing state if XW_IPC_STRICT_ROBUST_MUTEX_BROKEN_POLICY is not defined
#if !defined(XW_IPC_STRICT_ROBUST_MUTEX_BROKEN_POLICY)
#define XW_IPC_HANDLE_BROKEN_STATE() \
boost::uint32_t oldState = bipc::ipcdetail::atomic_cas32(&this->_state, robust_mutex_state::fixing_state, robust_mutex_state::broken_state);\
if(oldState == robust_mutex_state::broken_state){\
IPC_DBG_TRACE("Recovered from broken ... (" << _name << ")");\
}
#else
#define XW_IPC_HANDLE_BROKEN_STATE() \
if(bipc::ipcdetail::atomic_read32(&this->_state) == robust_mutex_state::broken_state){\
XW_THROW_EXCEPTION(ex_ipc_mutex_broken, "Robust mutex is broken!");\
}
#endif
namespace bipc = boost::interprocess;
namespace xw{ namespace ipc{
#ifdef _DEBUG
struct dbg_tracer{
static FILE* f ;
static void trace(const std::stringstream& str){
char tmp[1024];
sprintf(tmp, "%s/xw_ipc_robust_mutex_log_trace.log", bipc::ipcdetail::get_temporary_path());
if ( NULL == f ) {
f = fopen(tmp, "a+b");
}
if(f){
fprintf(f, "%6d %s\r\n", _getpid(), str.str().c_str());
fflush(f) ;
// fclose(f);
}
}
};
#ifdef _ROBUST_MUTEX_TRACE_FILE
FILE* dbg_tracer::f = (FILE *)NULL ;
#endif
#endif
namespace helper {
inline void get_pids_locking_file_base_name(std::string &s)
{
char tmp[1024];
sprintf(tmp, "%s/xw_ipc_robust_mutex_pids", bipc::ipcdetail::get_temporary_path());
s.assign(tmp);
}
// This class will be a intermodule_singleton.
// The constructor will check existence of robust_mutexes central pids
// lock file and it will lock current pid byte in.
// destructor will release that locked byte
// if we have some pid byte locked we know that process is alive
class process_life_guard
{
public:
static process_life_guard* instance(){
static process_life_guard _instance;
return &_instance;
}
private:
struct scoped_file_lock{
bipc::file_handle_t fd;
bool _locked;
scoped_file_lock(const bipc::file_handle_t& fd) : fd(fd), _locked(false){
bipc::ipcdetail::try_acquire_file_lock(fd, _locked);
if(!_locked){
XW_THROW_EXCEPTION(ex_ipc_generic, "Robust mutex can't lock file!");
}
}
~scoped_file_lock(){
if(_locked){
bipc::ipcdetail::release_file_lock(fd);
}
}
};
struct _deleter{
char* _ptr;
_deleter(char* p) : _ptr(p){}
~_deleter(){
if(_ptr) delete[] _ptr;
}
};
bipc::file_handle_t pids_file_lock_fd; // while we are creating "main pids file" we have to sync!!
bipc::file_handle_t pids_lock_fd; // 2^22 big file with reserved 1byte for each eventually alive process
int self_pid;
bool im_a_locker;
public:
bool is_procces_dead(bipc::ipcdetail::OS_process_id_t pid){
bool _locked = false;
if ( pid == bipc::ipcdetail::OS_process_id_t(0) ) {
return false ; // If process is unknown, than the process is not dead.
}
bipc::ipcdetail::try_acquire_file_lock_ex(pids_lock_fd, _locked, pid, 1);
if(_locked){
IPC_DBG_TRACE("check process " << pid << " was dead");
bipc::ipcdetail::release_file_lock_ex(pids_lock_fd, pid, 1);
return true;
}
return false;
}
private:
process_life_guard(const process_life_guard &);
process_life_guard &operator=(const process_life_guard &);
process_life_guard()
: pids_file_lock_fd(bipc::ipcdetail::invalid_file()), pids_lock_fd(bipc::ipcdetail::invalid_file()), self_pid(bipc::ipcdetail::get_current_process_id()), im_a_locker(false)
{
bipc::permissions p;
p.set_unrestricted();
std::string lockNameBase;
get_pids_locking_file_base_name(lockNameBase);
std::string mainFileName = (lockNameBase + "/main.lck").c_str();
std::string pidsFileName = (lockNameBase + "/pids.lck").c_str();
//ensure dir exists
bipc::ipcdetail::create_directory(lockNameBase.c_str());
//main global lock fd
pids_file_lock_fd = bipc::ipcdetail::create_or_open_file(mainFileName.c_str(), bipc::read_write, p);
if(pids_file_lock_fd == bipc::ipcdetail::invalid_file()){
XW_THROW_EXCEPTION(ex_ipc_generic, "Robust mutex can't open main lock file!");
}
//lock it in scope
{
//IPC_DBG_TRACE("start locking " << mainFileName);
scoped_file_lock scoped_lock(pids_file_lock_fd);
pids_lock_fd = bipc::ipcdetail::create_new_file(pidsFileName.c_str(), bipc::read_write, p);
if(pids_lock_fd != bipc::ipcdetail::invalid_file()){
//IPC_DBG_TRACE("create new " << pidsFileName);
_deleter d(new char[MAX_PIDS_NUMBER]);
memset(d._ptr, 0, MAX_PIDS_NUMBER);
bipc::ipcdetail::write_file(pids_lock_fd, d._ptr, MAX_PIDS_NUMBER);
}else{
//IPC_DBG_TRACE("use existing " << pidsFileName);
pids_lock_fd = bipc::ipcdetail::open_existing_file(pidsFileName.c_str(), bipc::read_write);
}
}
if(pids_lock_fd == bipc::ipcdetail::invalid_file()){
XW_THROW_EXCEPTION(ex_ipc_generic, "Robust mutex can't open pids lock file!");
}
//lock my slot, if we did lock im_a_locker will become true, if not, there is other instance of singleton in the same process
bipc::ipcdetail::try_acquire_file_lock_ex(pids_lock_fd, im_a_locker, self_pid, 1);
IPC_DBG_TRACE("locking " << self_pid << " in " << pidsFileName << " -> " << (im_a_locker ? "locker" : "non locker"));
}
~process_life_guard()
{
if(im_a_locker){
bipc::ipcdetail::release_file_lock_ex(pids_lock_fd, self_pid, 1);
IPC_DBG_TRACE("released proccess lock " << self_pid);
}
bipc::ipcdetail::close_file(pids_lock_fd);
bipc::ipcdetail::close_file(pids_file_lock_fd);
}
};
}//helper
struct robust_mutex_state{
static const boost::uint32_t correct_state = 0;
static const boost::uint32_t fixing_state = 1;
static const boost::uint32_t broken_state = 2;
};
template<typename mutexT>
class set_name_trait{
public:
void name(mutexT & m, const char* n){
m.name(n);
}
};
template<typename mutexT>
class fix_it_trait{
public:
void fix_it(mutexT & m){
m.fix_it();
}
};
class robust_recursive_mutex;
class robust_mutex
{
typedef robust_mutex self_type;
friend class fix_it_trait<self_type>;
friend class set_name_trait<self_type>;
//robust mutex will be wrapped in recursive one
friend class robust_recursive_mutex;
robust_mutex(const robust_mutex &);
robust_mutex &operator=(const robust_mutex &);
public:
robust_mutex();
~robust_mutex();
void lock();
bool try_lock();
bool timed_lock(const boost::posix_time::ptime &abs_time);
void unlock();
void take_ownership(){};
boost::uint32_t get_state();
const char* name() const;
private:
//eventual name for mutex
char _name[MAX_PATH + 1];
//The pid of the owner, this field will be also lock gate
volatile boost::uint32_t _owner;
//The state of the mutex (correct, fixing, broken)
volatile boost::uint32_t _state;
//just some time check process alive
static const unsigned int _spin_threshold = 255u;
//invalid handle
static const boost::uint32_t _invalidProcesshandle = (boost::uint32_t)bipc::ipcdetail::OS_process_id_t(0);
bool robust_check();
bool check_if_owner_dead_and_take_ownership_atomically();
bool init_life_guard();
void fix_it(); //who need call must have trait for it!!!
void name(const char*);
};
/************************************************************************/
/* IPC types */
/************************************************************************/
typedef robust_mutex robust_ipc_mutex;
/************************************************************************/
/* IMPL */
/************************************************************************/
inline const char* robust_mutex::name() const { return _name; }
inline void robust_mutex::name(const char* n) {
strncpy(_name, n, MAX_PATH);
size_t len = strlen(n);
_name[len < MAX_PATH ? len : MAX_PATH] = 0;
}
inline robust_mutex::robust_mutex()
: _owner(robust_mutex::_invalidProcesshandle), _state(robust_mutex_state::correct_state)
{
_name[0] = 0;
}
inline robust_mutex::~robust_mutex(){}
inline void robust_mutex::lock(void)
{
XW_IPC_HANDLE_BROKEN_STATE();
//This function provokes intermodule_singleton instantiation
if(!this->init_life_guard()){
XW_THROW_EXCEPTION(ex_ipc_generic, "Can't make life guard");
}
bipc::spin_wait swait;
boost::uint32_t currentPid = bipc::ipcdetail::get_current_process_id();
do{
// 0 -> current
boost::uint32_t prev_s = bipc::ipcdetail::atomic_cas32(const_cast<boost::uint32_t*>(&_owner), currentPid, 0);
if (_owner == currentPid && prev_s == 0){
//locked
break;
}else{
// relinquish current time slice
swait.yield();
//Do the dead owner checking each spin_threshold lock tries
if(0 == (swait.count() & _spin_threshold)){
// Check if owner dead and take ownership if possible
// it return true if ownership was done internal mutex is still locked!!!!
// if ownership is taken then locks count will be set to 1
if(this->robust_check()){
break;
}
}
}
}while (true);
}
inline bool robust_mutex::try_lock(void)
{
XW_IPC_HANDLE_BROKEN_STATE();
//This function provokes intermodule_singleton instantiation
if(!this->init_life_guard()){
XW_THROW_EXCEPTION(ex_ipc_generic, "Can't make life guard");
}
boost::uint32_t currentPid = bipc::ipcdetail::get_current_process_id();
// 0 -> current
boost::uint32_t prev_s = bipc::ipcdetail::atomic_cas32(const_cast<boost::uint32_t*>(&_owner), currentPid, 0);
//if not locked do one time also robust check
return _owner == currentPid && prev_s == 0 ? true : this->robust_check();
}
inline bool robust_mutex::timed_lock(const boost::posix_time::ptime &abs_time)
{
if(abs_time == boost::posix_time::pos_infin){
this->lock();
return true;
}
//Obtain current count and target time
boost::posix_time::ptime now = bipc::microsec_clock::universal_time();
bipc::spin_wait swait;
do{
if(this->try_lock()){
break;
}
now = bipc::microsec_clock::universal_time();
if(now >= abs_time){
return false;
}
// relinquish current time slice
swait.yield();
}while (true);
return true;
}
inline void robust_mutex::unlock(void)
{
//If in "fixing" state, unlock and mark the mutex as unrecoverable
//so next locks will fail and all threads will be notified that the
//data protected by the mutex was not recoverable.
boost::uint32_t oldState = bipc::ipcdetail::atomic_cas32(&this->_state, robust_mutex_state::broken_state, robust_mutex_state::fixing_state);
if(oldState == robust_mutex_state::fixing_state){
IPC_DBG_TRACE("Marked BROKEN ! (" << _name << ")");
XW_THROW_EXCEPTION(ex_ipc_mutex_broken, "Robust mutex is broken ! [status:" << this->_state << ", owner:" << this->_owner << ", process:" << (boost::uint32_t)bipc::ipcdetail::get_current_process_id() << "]" );
}
boost::uint32_t currentPid = bipc::ipcdetail::get_current_process_id();
if( bipc::ipcdetail::atomic_read32(&this->_owner) != currentPid){
XW_THROW_EXCEPTION(ex_ipc_just_owner_can_unlock, "Not owner atemp to unlock ! [status:" << this->_state << ", owner:" << this->_owner << ", process:" << (boost::uint32_t)bipc::ipcdetail::get_current_process_id() << "]" );
}
// current -> 0
bipc::ipcdetail::atomic_cas32(const_cast<boost::uint32_t*>(&_owner), 0, currentPid);
}
inline void robust_mutex::fix_it()
{
// This function supposes the previous state was "fixing"
// and the current process holds the mutex
// so this should be called on locked mutex in fixing state!!!!
if(bipc::ipcdetail::atomic_read32(&this->_state) != robust_mutex_state::fixing_state &&
bipc::ipcdetail::atomic_read32(&this->_owner) != (boost::uint32_t)bipc::ipcdetail::get_current_process_id()){
XW_THROW_EXCEPTION(ex_ipc_mutex_broken, "Robust mutex is broken ! [status:" << this->_state << ", owner:" << this->_owner << ", process:" << (boost::uint32_t)bipc::ipcdetail::get_current_process_id() << "]" );
}
//If that's the case, just update mutex state
bipc::ipcdetail::atomic_write32(&this->_state, robust_mutex_state::correct_state);
IPC_DBG_TRACE("putting mutex in correct state ... (" << _name << ")");
}
inline boost::uint32_t robust_mutex::get_state()
{
//Notifies if a owner recovery has been performed in the last lock()
return bipc::ipcdetail::atomic_read32(&this->_state);
};
inline bool robust_mutex::robust_check()
{
//If the old owner was dead, and we've acquired ownership, mark
//the mutex as 'fixing'. This means that a "consistent()" is needed
//to avoid marking the mutex as "broken" when the mutex is unlocked.
if(!this->check_if_owner_dead_and_take_ownership_atomically()){
return false;
}
IPC_DBG_TRACE("putting mutex in fixing state ... (" << _name << ")");
bipc::ipcdetail::atomic_write32(&this->_state, robust_mutex_state::fixing_state);
return true;
}
inline bool robust_mutex::check_if_owner_dead_and_take_ownership_atomically()
{
boost::uint32_t cur_owner = bipc::ipcdetail::get_current_process_id();
boost::uint32_t old_owner = bipc::ipcdetail::atomic_read32(&this->_owner), old_owner2;
// The cas loop guarantees that only one thread from this or another process
// will succeed taking ownership
// we will accept only mutex with non invalid owner handle
do{
//Check if owner process is dead
if(! helper::process_life_guard::instance()->is_procces_dead(old_owner)){
// current owner is running, and its not and invalid handle so
// we let it be and go away
return false;
}
//If it's dead, try to mark this process as the owner in the owner field
old_owner2 = old_owner;
old_owner = bipc::ipcdetail::atomic_cas32(&this->_owner, cur_owner, old_owner);
}while(old_owner2 != old_owner);
IPC_DBG_TRACE("Mutex (" << _name << ")" << " taken from : " << old_owner << " by: " << cur_owner);
return true;
}
inline bool robust_mutex::init_life_guard()
{
helper::process_life_guard* guard = helper::process_life_guard::instance();
return guard != 0;
}
}} //xw::ipc
#include <boost/interprocess/detail/config_end.hpp>
#endif
Here is rest of code. It's taken from our projects tree, so if you want to test it you have to adapt some include maybe. It's proposal, so please do not blame me for code style or something else. If idea and mode is good, I'll continue to make it more perfect. If not I'll just use something else, but I can't find anything else.
There are also versions for recursive_mutex, and named_mutex too. Then there is sort of
fixing proposal, cause if one process take in owner ship precedentaly abandoned mutex, there is high probability that there must be done some sort of integrity check.
I'd like to discuss eventual improvements
