原子操作(1) - 用匯編實現原子操作

“最輕量級的鎖”,通常也叫”原子操作”,之所以加引號是因為他們在匯編級別并不是原子操作,是用多條指令完成的,這些操作大多都是利用CPU支持的匯編指令.在某些構架過時的CPU體系結構上,它們應該是用比較重量級的線程內鎖實現的吧(我的猜測).

最常見的原子操作有Compare and Exchange,Self Increase/Decrease等等

80486 CPU相關指令：

LOCK:這是一個指令前綴，在所對應的指令操作期間使此指令的目標操作數指定的存儲區域鎖定，以得到保護。

XADD:先交換兩個操作數的值，再進行算術加法操作。多處理器安全，在80486及以上CPU中支持。

CMPXCHG:比較交換指令，第一操作數先和AL/AX/EAX比較，如果相等ZF置1，第二操作數賦給第一操作數，否則ZF清0，第一操作數賦給AL/AX/EAX。多處理器安全，在80486及以上CPU中支持。

XCHG:交換兩個操作數,其中至少有一個是寄存器尋址.其他寄存器和標志位不受影響.

80486以上都支持這四個操作,因此當今幾乎100%CPU都支持這兩個指令,也能由此用標準C和C++寫出一系列幾乎可以跨平臺的”原子操作”函數和Lock-Free數據結構和算法.

64位平臺也有一系列的相關指令,當然他們也有以上提到的指令,其下的64位原子操作函數應該和32位的分開(要問為什么?我現在告訴你恐怕你印象不深,接著看這一系列吧),而道理完全一樣.因此,不存在跨CPU體系結構的問題)

“原子操作”函數:

由以上提供的幾個匯編指令,我們可以做出以下實現,這些都是最常用的原語.

比較后交換

long __stdcall CompareExchange(long volatile*Destination,long Exchange,long Comperand)

{

   __asm

   {

      mov     ecx, Destination;

      mov     edx, Exchange;

      mov     eax, Comperand;

      lock cmpxchg [ecx], edx;

   }

}

交換

long __stdcall Exchange(long volatile* Target,long Value)

{

   __asm

   {

      mov      ecx, Target;

      mov      edx, Value;

label:

      lock cmpxchg [ecx], edx;//加

      jnz      short label;

   }

}

自減

long __stdcall Decrement(long volatile* Addend)

{

   __asm

   {

      mov     ecx, Addend;

      mov     eax, 0FFFFFFFFh;//-1

      lock xadd [ecx], eax; //加-1

      dec     eax;

   }

}

自增

long __stdcall Increment(long volatile* Addend)

{

   __asm

   {

      mov      ecx, Addend;

      mov      eax, 1;

      lock xadd [ecx], eax; //加

      inc      eax;

   }

}

相加后交換

long __stdcall ExchangeAdd(long volatile* Addend,long Value)

{

   __asm

   {

      mov      ecx, Addend;

      mov      eax, Value;

      lock xadd [ecx], eax;

   }

}

32位的數據類型有4種,但是上面的只支持long,怎么辦?手工hack?太土了,當然是C++的大規模殺傷性武器template了.

同時把幾個不跨平臺的地方抽出來用macro表示.

目前模板還沒有得到concept的支持,所以只能用boost.type_traits低級的手動判斷,要求只有32位的整數類型才能實例化這些函數.

#include

#include

#define CALL_METHOD __stdcall

#define VOLATILE volatile

template<typename T>

T CALL_METHOD compare_exchange32(T VOLATILE*Destination,T exchange32,T Comperand)

{

   BOOST_STATIC_ASSERT(sizeof(T) == 4 && boost::is_integral<T>::value);

   __asm

   {

      mov     ecx, Destination;

      mov     edx, exchange32;

      mov     eax, Comperand;

      lock cmpxchg [ecx], edx;

   }

}

template<typename T>

T CALL_METHOD exchange32(T VOLATILE* Target,T Value)

{

   BOOST_STATIC_ASSERT(sizeof(T) == 4 && boost::is_integral<T>::value);

   __asm

   {

      //     mov    ecx, Target;

      //     mov    edx, Value;      

      //label:

      //     lock   cmpxchg [ecx], edx;//加

      //     jnz    short label;

      mov      ecx, Target;

      mov      eax, Value;

      xchg [ecx],eax;

   }

}

template<typename T>

T CALL_METHOD decrement32(T VOLATILE* Addend)

{

   BOOST_STATIC_ASSERT(sizeof(T) == 4 && boost::is_integral<T>::value);

   __asm

   {

      mov     ecx, Addend;

      mov    eax, 0FFFFFFFFh;//-1

      lock xadd [ecx], eax; //加-1

      dec     eax;

   }

}

template<typename T>

T CALL_METHOD increment32(T VOLATILE* Addend)

{

   BOOST_STATIC_ASSERT(sizeof(T) == 4 && boost::is_integral<T>::value);

   __asm

   {

      mov      ecx, Addend;

      mov      eax, 1;

      lock xadd [ecx], eax; //加

      inc      eax;

   }

}

template<typename T>

T CALL_METHOD exchange_add32(T VOLATILE* Addend,T Value)

{

   BOOST_STATIC_ASSERT(sizeof(T) == 4 && boost::is_integral<T>::value);

   __asm

   {

      mov      ecx, Addend;

      mov      eax, Value;

      lock xadd [ecx], eax;

   }

}

根據上面的5個函數就能做出一個原子操作的整數數字類,這將是下一節中,我的最輕量級鎖的基礎和原型,他不依賴于操作系統,當然,所以你也可以不叫他是鎖,只是一種類似鎖的機制.

一切細節看源碼中穿插的注釋.

#ifndef __ATOM_VALUE_H__

#define __ATOM_VALUE_H__

#include "atom.hpp"

#include <boost/static_assert.hpp>

#include <boost/type_traits.hpp>

template<typename T>

class atomic_value32

{

   //恩,用boost.type_traits來保證是位的整數類型

   BOOST_STATIC_ASSERT(sizeof(T) == 4 && boost::is_integral<T>::value);

private:

   volatile T value_;

public:

   atomic_value32(T v = 0)

      :value_(v){}

   atomic_value32(atomic_value32& v){//??? 這里留給大家,我不給出源碼了

   }

   //需要這么一個轉型符號,因為大部分時候我們將atomic_value32<T>當作一個T使用

   operator T(){return exchange_add32(&value_,0);}

   //賦值

   atomic_value32& operator=(T v){exchange32(&value_, v);return *this;}

   atomic_value32& operator=(atomic_value32& v){exchange32(&value_, v);return *this;}

   //比較并交換,好像沒有什么operator與之對應,就直接拿出來了

   T compare_exchange(T to_exchange, T to_compare)

   {return compare_exchange32<T>(&value_, to_exchange, to_compare);}

   //只提供前置,后置似乎沒什么必要,我也懶得去實現了:)

   T operator++(){return increment32(&value_);}

   T operator--(){return decrement32(&value_);}

   //千萬不能返回引用,因為線程安全考慮,

   T operator+=(T add){return exchange_add32(&value_,add);}

   T operator+=(atomic_value32& add){return exchange_add32(&value_,add);}

   T operator-=(T add){return exchange_add32(&value_,-add);}

   T operator-=(atomic_value32& add){return exchange_add32(&value_,-add);}

   //6個比較符號

   bool operator==(T rhs){return operator T()==rhs;}

   bool operator==(atomic_value32& rhs){return operator T()==rhs.operator T();}

   bool operator<(T rhs){return operator T()<rhs;}

   bool operator<(atomic_value32& rhs){return operator T()<rhs.operator T();}

   bool operator!=(T rhs){return !this->operator ==(rhs);}

   bool operator!=(atomic_value32& rhs){return !this->operator ==(rhs);}

   bool operator>=(T rhs){return !this->operator <(rhs);}

   bool operator>=(atomic_value32& rhs){return !this->operator <(rhs);}

   bool operator>(T rhs){return ((*this)!=(rhs)) && !((*this)<(rhs));}

   bool operator>(atomic_value32& rhs){return ((*this)!=(rhs)) && !((*this)<(rhs));}

   bool operator<=(T rhs){return !((*this)>(rhs));}

   bool operator<=(atomic_value32& rhs){return !((*this)>(rhs));}

};

#endif//__ATOM_VALUE_H__