Mutable/Promotable Recursive Mutex

#include "mutable_recursive_mutex.hpp"


mutable_recursive_mutex::mutable_recursive_mutex()
{
#ifdef FAIR_LOCKING
	m_is_main_thread_id_assigned = false;
	m_thread = new boost::thread(&mutable_recursive_mutex::unlock_thread, this);
#endif
}

mutable_recursive_mutex::~mutable_recursive_mutex()
{
#ifdef FAIR_LOCKING
	m_thread->interrupt();
	m_thread->join();
	delete m_thread;
	m_thread = NULL;
#endif
}


void mutable_recursive_mutex::set_main_thread_id()
{
#ifdef FAIR_LOCKING
	m_main_thread_id = boost::this_thread::get_id();
	m_is_main_thread_id_assigned = true;
#endif
}

void mutable_recursive_mutex::reset_main_thread_id()
{
#ifdef FAIR_LOCKING
	m_is_main_thread_id_assigned = false;
#endif
}

mutable_recursive_mutex::thread_recursion_stack_t * mutable_recursive_mutex::get_stack()
{
	thread_recursion_stack_t * stack = m_thread_states.get();
	if (stack == NULL)
	{
		stack = new thread_recursion_stack_t();
		m_thread_states.reset(stack);
		stack->push_front(lst_none);
	}
	return stack;
}

#ifdef FAIR_LOCKING

void mutable_recursive_mutex::unlock_thread()
{
	boost::chrono::milliseconds ms(100);
	try
	{
		while (true)
		{
			boost::this_thread::sleep_for(ms);
			boost::unique_lock<boost::mutex> promote_lock(m_promote_mutex);
			if (!m_waiting_promote_threads.empty())
				m_promote_condition.notify_all();
		}
	} catch (boost::thread_interrupted)
	{
	}
}

void mutable_recursive_mutex::promote(thread_recursion_stack_t * stack)
{
	boost::thread::id thread_id = boost::this_thread::get_id();
	boost::mutex helper_mutex;
	boost::unique_lock<boost::mutex> helper_lock(helper_mutex);
	bool bnotify = (stack->front() == lst_shared);

	boost::unique_lock<boost::mutex> promote_lock(m_promote_mutex);
	if (m_is_main_thread_id_assigned && (thread_id == m_main_thread_id))
		m_waiting_promote_threads.push_front(thread_id);
	else
		m_waiting_promote_threads.push_back(thread_id);
	if (bnotify) 
		m_mutex.unlock_shared();
	while (true)
	{
		if (m_waiting_promote_threads.front() == thread_id)
		{
			bnotify = false;
			if (m_mutex.try_lock())
			{
				m_waiting_promote_threads.pop_front();
				break;
			}
		} 
		if (bnotify)
			m_promote_condition.notify_all();
		bnotify = false;
		promote_lock.unlock();
		m_promote_condition.wait(helper_lock);
		promote_lock.lock();
	}
}

#else

void mutable_recursive_mutex::promote(thread_recursion_stack_t * stack)
{
	if (stack->front() == lst_shared)
		m_mutex.unlock_shared();
	m_mutex.lock();
}

#endif

void mutable_recursive_mutex::do_unlock_from_shared(thread_recursion_stack_t * stack, lock_state_t new_state)
{
	switch(new_state)
	{
	case lst_none:
		{
#ifdef FAIR_LOCKING
			boost::unique_lock<boost::mutex> promote_lock(m_promote_mutex);
#endif
			m_mutex.unlock_shared();
#ifdef FAIR_LOCKING
			m_promote_condition.notify_all();
#endif
		}
		break;

	case lst_unique:
		promote(stack);
		break;
	}
}

void mutable_recursive_mutex::do_unlock_from_unique(lock_state_t new_state)
{
	switch(new_state)
	{
	case lst_none:
		{
#ifdef FAIR_LOCKING
			boost::unique_lock<boost::mutex> promote_lock(m_promote_mutex);
#endif
			m_mutex.unlock();
#ifdef FAIR_LOCKING
			m_promote_condition.notify_all();
#endif
		}
		break;

	case lst_shared:
		m_mutex.unlock_and_lock_shared();
		break;
	}
}

void mutable_recursive_mutex::do_unlock()
{
	assert(m_thread_states.get() != NULL);
	thread_recursion_stack_t * stack = get_stack();
	assert(stack->size() > 1);
	lock_state_t current_state = stack->front();
	lock_state_t new_state = stack->at(1);

	switch(current_state)
	{
	case lst_shared:
		do_unlock_from_shared(stack, new_state);
		break;

	case lst_unique:
		do_unlock_from_unique(new_state);
		break;

	case lst_suspended:
		return;
	}

	stack->pop_front();
	if (stack->size() == 1)
	{
		assert(stack->front() == lst_none);
		m_thread_states.reset(NULL);
	}
}

void mutable_recursive_mutex::lock_shared()
{
	thread_recursion_stack_t * stack = get_stack();
	switch(stack->front())
	{
	case lst_none:
		m_mutex.lock_shared();
		break;

	case lst_unique:
		m_mutex.unlock_and_lock_shared();
		break;

	case lst_suspended:
		return;
	}
	stack->push_front(lst_shared);
}

void mutable_recursive_mutex::lock()
{
	thread_recursion_stack_t * stack = get_stack();
	switch(stack->front())
	{
	case lst_none:
	case lst_shared:
		promote(stack);
		break;

	case lst_suspended:
		return;
	}
	stack->push_front(lst_unique);
}

void mutable_recursive_mutex::unlock()
{
	do_unlock();
}

void mutable_recursive_mutex::unlock_shared()
{
	do_unlock();
}

void mutable_recursive_mutex::suspend()
{
	thread_recursion_stack_t * stack = get_stack();
	switch(stack->front())
	{
	case lst_shared:
		{
#ifdef FAIR_LOCKING
			boost::unique_lock<boost::mutex> promote_lock(m_promote_mutex);
#endif
			m_mutex.unlock_shared();
#ifdef FAIR_LOCKING
			m_promote_condition.notify_all();
#endif
		}
		break;

	case lst_unique:
		{
#ifdef FAIR_LOCKING
			boost::unique_lock<boost::mutex> promote_lock(m_promote_mutex);
#endif
			m_mutex.unlock();
#ifdef FAIR_LOCKING
			m_promote_condition.notify_all();
#endif
		}
		break;
	}
	stack->push_front(lst_suspended);
}

void mutable_recursive_mutex::resume()
{
	assert(m_thread_states.get() != NULL);
	thread_recursion_stack_t * stack = get_stack();
	assert(stack->size() > 1);

	assert(stack->front() == lst_suspended);

	switch(stack->at(1))
	{
	case lst_shared:
		m_mutex.lock_shared();
		break;

	case lst_unique:
		promote(stack);
		break;
	}

	stack->pop_front();
	if (stack->size() == 1)
	{
		assert(stack->front() == lst_none);
		m_thread_states.reset(NULL);
	}
}

mutable_unique_lock::mutable_unique_lock()
{
	m_mutex = NULL;
	m_is_locked = false;
}

mutable_unique_lock::mutable_unique_lock(mutable_recursive_mutex& mutex)
{
	m_mutex = &mutex;
	m_mutex->lock();
	m_is_locked = true;
}

mutable_unique_lock::~mutable_unique_lock()
{
	if (m_is_locked)
		m_mutex->unlock();
}

bool mutable_unique_lock::owns_lock() const
{
	return m_is_locked;
}

void mutable_unique_lock::lock()
{
    if(m_mutex == NULL)
    {
        boost::throw_exception(boost::lock_error(boost::system::errc::operation_not_permitted, "boost mutable_unique_lock has no mutex"));
    }
    if(owns_lock())
    {
        boost::throw_exception(boost::lock_error(boost::system::errc::resource_deadlock_would_occur, "boost mutable_unique_lock owns already the mutex"));
    }
	m_mutex->lock();
	m_is_locked = true;
}

void mutable_unique_lock::unlock()
{
    if(m_mutex == NULL)
    {
        boost::throw_exception(boost::lock_error(boost::system::errc::operation_not_permitted, "boost mutable_unique_lock has no mutex"));
    }
    if(!owns_lock())
    {
        boost::throw_exception(boost::lock_error(boost::system::errc::operation_not_permitted, "boost mutable_unique_lock doesn't own the mutex"));
    }
	m_mutex->unlock();
	m_is_locked = false;
}

mutable_shared_lock::mutable_shared_lock()
{
	m_mutex = NULL;
	m_is_locked = false;
}

mutable_shared_lock::mutable_shared_lock(mutable_recursive_mutex& mutex)
{
	m_mutex = &mutex;
	m_mutex->lock_shared();
	m_is_locked = true;
}

mutable_shared_lock::~mutable_shared_lock()
{
	if (m_is_locked)
		m_mutex->unlock_shared();
}

bool mutable_shared_lock::owns_lock() const
{
	return m_is_locked;
}

void mutable_shared_lock::lock()
{
    if(m_mutex == NULL)
    {
        boost::throw_exception(boost::lock_error(boost::system::errc::operation_not_permitted, "boost mutable_shared_lock has no mutex"));
    }
    if(owns_lock())
    {
        boost::throw_exception(boost::lock_error(boost::system::errc::resource_deadlock_would_occur, "boost mutable_shared_lock owns already the mutex"));
    }
	m_mutex->lock_shared();
	m_is_locked = true;
}

void mutable_shared_lock::unlock()
{
    if(m_mutex == NULL)
    {
        boost::throw_exception(boost::lock_error(boost::system::errc::operation_not_permitted, "boost mutable_shared_lock has no mutex"));
    }
    if(!owns_lock())
    {
        boost::throw_exception(boost::lock_error(boost::system::errc::operation_not_permitted, "boost mutable_shared_lock doesn't own the mutex"));
    }
	m_mutex->unlock_shared();
	m_is_locked = false;
}

mutable_suspend_lock::mutable_suspend_lock()
{
	m_mutex = NULL;
	m_is_locked = false;
}

mutable_suspend_lock::mutable_suspend_lock(mutable_recursive_mutex& mutex)
{
	m_mutex = &mutex;
	m_mutex->suspend();
	m_is_locked = true;
}

mutable_suspend_lock::mutable_suspend_lock(mutable_recursive_mutex& mutex, boost::function<void()> f)
{
	m_mutex = &mutex;
	m_mutex->suspend();
	m_is_locked = true;

	f();
}

mutable_suspend_lock::~mutable_suspend_lock()
{
	if (m_is_locked)
		m_mutex->resume();
}

void suspend_lock(mutable_recursive_mutex& mutex, boost::function<void()> f)
{
	mutable_suspend_lock suspend_lock(mutex);
	f();
}