Re: [boost] Copyable mutex

From: Howard Hinnant (howard.hinnant_at_[hidden])
Date: 2007-07-24 19:27:50

On Jul 23, 2007, at 3:53 AM, Anthony Williams wrote:

> "Peter Dimov" <pdimov_at_[hidden]> writes:
>
>> Phillip Hellewell wrote:
>>
>>> However, even the simplest _assignable_ class seems almost  
>>> impossible
>>> to me to make thread-safe.  You can synchronize operator= using both
>>> its own mutex(es) and rhs's mutex(es), which seems to solve the
>>> problem.  But how do you prevent deadlock from something like this?
>>>
>>> Thread 1: x = y
>>> Thread 2: y = x
>>>
>>> Thread 2 is going to lock the mutexes in the opposite order of  
>>> thread
>>> 1. Ouch!
>>
>> The classic solution is to lock the mutexes in ascending order of  
>> their
>> addresses, but I prefer
>>
>> X& X::operator=( X const& rhs )
>> {
>>    X tmp( rhs ); // temporarily locks rhs.mutex
>>
>>    scoped_lock lock( this->mutex );
>>    swap( tmp ); // or move_from( tmp )
>>
>>    return *this;
>> }
>>
>> It's always a good idea to only lock one mutex at a time if you can  
>> afford
>> it.
>
> I like the idea, but surely the publicly-visible swap should lock both
> mutexes? Wouldn't this be better as internal_swap()?

Coming to the party late...

The above is inefficient for types that are not cheaply swappable.   
Here is another, admittedly C++0X suggestion which works both for  
expensive-to-swap and cheap-to-swap types:

X& X::operator=( X const& rhs )
{
    namespace pstd = proposed_std;

    pstd::unique_lock<pstd::mutex> this_lock(this->mutex,  
pstd::defer_lock);
    pstd::unique_lock<pstd::mutex> that_lock(rhs.mutex,  
pstd::defer_lock);
    pstd::lock(this_lock, that_lock);  // avoids deadlock

    // do the assign, both *this and rhs are locked

    return *this;
}

The above might be better implemented if there is a rw_mutex available:

X& X::operator=( X const& rhs )
{
    namespace pstd = proposed_std;

    pstd::unique_lock<pstd::mutex> this_lock(this->mutex,  
pstd::defer_lock);
    pstd::shared_lock<pstd::mutex> that_lock(rhs.mutex,  
pstd::defer_lock);
    pstd::lock(this_lock, that_lock);  // ok with different lock types

    // do the assign, *this write-locked, rhs is only read-locked

    return *this;
}

With variadic templates, one can implement pstd::lock with an  
arbitrary number of mutexes/locks requiring nothing more than lock(),  
try_lock() and unlock().  And it has strong exception safety assuming  
a nothrow unlock().  I.e. when it exits exceptionally (if lock() or  
try_lock() throws), all locks are unlocked.  When it exits normally,  
all locks are locked, no deadlock guaranteed.

And for bonus points, this algorithm blows away the performance of  
locking in order (which is just dead-slow in the measurements I've  
made - and don't forget or ignore the yields).

The below can be implemented today for two locks by just striking out  
the variadic stuff.  Or the greater than two locks overloads could be  
added manually (what a pain!).

-Howard

// returns index of failed lock or -1 on success

template <class L1, class L2>
int
try_lock(L1& l1, L2& l2)
{
     unique_lock<L1> u(l1, try_to_lock);
     if (u.owns())
     {
         if (l2.try_lock())
         {
             u.release();
             return -1;
         }
         return 1;
     }
     return 0;
}

template <class L1, class L2, class ...L3>
int
try_lock(L1& l1, L2& l2, L3& l3...)
{
     unsigned r = 0;
     unique_lock<L1> u(l1, try_to_lock);
     if (u.owns())
     {
         r = try_lock(l2, l3...);
         if (r == -1)
             u.release();
         else
             ++r;
     }
     return r;
}

template <class L1, class L2>
void
lock(L1& l1, L2& l2)
{
     while (true)
     {
         {
         unique_lock<L1> u1(l1);
         if (l2.try_lock())
         {
             u1.release();
             break;
         }
         }
         std::this_thread::yield();
         {
         unique_lock<L2> u2(l2);
         if (l1.try_lock())
         {
             u2.release();
             break;
         }
         }
         std::this_thread::yield();
     }
}

namespace detail
{

// On failed attempts, next try always starts
//   with lock that failed the try_lock on the
//   previous iteration (i.e. the algorithm blocks on it).
template <class L1, class L2, class ...L3>
void
lock_first(int i, L1& l1, L2& l2, L3& l3...)
{
     while (true)
     {
         switch (i)
         {
         case 0:
             {
             unique_lock<L1> u1(l1);
             i = try_lock(l2, l3...);
             if (i == -1)
             {
                 u1.release();
                 return;
             }
             }
             ++i;
             std::this_thread::yield();
             break;
         case 1:
             {
             unique_lock<L2> u2(l2);
             i = try_lock(l3..., l1);
             if (i == -1)
             {
                 u2.release();
                 return;
             }
             }
             if (i == sizeof...(L3))
                 i = 0;
             else
                 i += 2;
             std::this_thread::yield();
             break;
         default:
             lock_first(i - 2, l3..., l1, l2);
             return;
         }
     }
}

}  // detail

template <class L1, class L2, class ...L3>
inline
void
lock(L1& l1, L2& l2, L3& l3...)
{
     detail::lock_first(0, l1, l2, l3...);
}