C#多线程编程(6)--线程安全2 互锁构造Interlocked

时间:2024-01-23 11:44:19

在线程安全1中,我介绍了线程同步的意义和一种实现线程同步的方法:volatile。volatile关键字属于原子操作的一种,若对一个关键字使用volatile,很多时候会显得很“浪费”,因为只有在并发访问的情况下才需要“易变”读写,单线程访问时并不需要。在命名空间System.Threading命名空间中提供了InterLock类,该类中提供了一些原子方法。本文来介绍如何使用这些方法。

  •   互锁

  在抢占式系统中,一个线程在执行到任何阶段都有可能被其他线程“打断”,原子操作能够保证该操作不被其他线程打断,其他线程的“打断”只可能发生在该操作的之前或之后。InterLock类中的方法就是”原子“式的,最常用的方法有:

public static class InterLock{
    // return (++location)
    public static int Increment(ref int location);
    // return(--location)
    public static int Decrement(ref int location);
    // return(location += value)
    //注意,也可能是负数,从而实现减法运算。
    public static int Add(ref int location, int value)
    // int old = location; location = value; return old;
    public static int Exchange(ref int location, int value);
    //old = location1;
    //if(location1 = comparand) location1 = value;
    //return old;
    public static int CompareExchange(ref int location1,
           int value, int comparand);
    ...
}

  《CLR via C#》的作者在书中说他喜欢Interlocked中的方法,因为它不但很快,而且不会阻塞线程。为了验证InterLock真的很快,我们对变量进行一百万次的写,与Volatile.Write()来进行对比,看看是否真的快。

 static void Main(string[] args){
     TestInterLock();
     Console.ReadLine();
 }
 private static volatile int v = 0;
 private static int n = 0;
 static void TestInterLock(){
     Stopwatch sw = Stopwatch.StartNew();
     for (int i = 0; i < 1000000; i++)
         v++;
     Console.WriteLine("volatile write 1000000 times takes:{0}", sw.ElapsedMilliseconds);
     sw = Stopwatch.StartNew();
     for (int i = 0; i < 1000000; i++)
         Interlocked.Increment(ref n);
     Console.WriteLine("InterLock write 1000000 times takes:{0}", sw.ElapsedMilliseconds);
     n = 0;
     sw = Stopwatch.StartNew();
     for (int i = 0; i < 1000000; i++)
         n++;
     Console.WriteLine("n++ write 1000000 times takes:{0}", sw.ElapsedMilliseconds);
 }

 

  运行结果如下:

volatile write 1000000 times takes:2

InterLock write 1000000 times takes:8

n++ write 1000000 times takes:2

  我运行了好几次,结果会有些出入,但是大部分的结果都是volatile的写入速度和原生的n++的速度是一样的。InterLock也确实如Jeffrey Richter所说,很快,只是没有volatile关键字修饰的变量的读写快。这是在非并发情况下,下面来看一下并发情况下是否还是很快。

static void TestInterLock1(){
    Stopwatch sw = Stopwatch.StartNew();
    Task[] t2 = new[] { new Task(() =>    {
        for (int i = 0; i < 1000000; i++)
            Interlocked.Increment(ref n);
    }), new Task(() => { 
        for (int i = 0; i < 1000000; i++)
            Interlocked.Increment(ref n);}) };
    Task[] t1 = new[] { new Task(() =>    {
        for (int i = 0; i < 1000000; i++)
            v++;
    }), new Task(() => { 
        for (int i = 0; i < 1000000; i++)
            v++;}) };
    Task.WhenAll(t1)
        .ContinueWith(t => Console.WriteLine("volatile write 2000000 times takes:{0}", sw.ElapsedMilliseconds));
    Task.WhenAll(t2)
        .ContinueWith(t => Console.WriteLine("InterLock write 2000000 times takes:{0}", sw.ElapsedMilliseconds));
    t2[0].Start();
    t1[0].Start();
    t1[1].Start();
    t2[1].Start();
}

  先看运行结果,

volatile write 2000000 times takes:94
InterLock write 2000000 times takes:101

  多次运行,运行时间会有不同,但是在并发情况下,volatile的写入和InterLock的写入速度几乎相同。上述代码写的如此丑陋,而不是直接写Task.Run(),是为了保证初始化部分都运行完成后,再Start(),且两个任务的先后顺序进行了打乱,最大限度减少误差。可以看到并发情况下,volatile和InterLock几乎一样,且在InterLock中的方法要比Volatile的功能要全,但是在串行时,Volatile的性能要比InterLock要好。结论是,若只对变量读写,没有替换或者其他复杂操作时,可以使用volatile关键字,但是一些复杂操作,需要原子操作时,就得使用InterLock中的方法了,如果使用volatile关键字修饰的变量来进行交换的话,很难保证原子性,只有引入锁才能保证线程同步。且InterLock中提供了几个重载方法,能够接受object类型,还有泛型版本。

可以利用InterLock来实现一个简单的自旋锁,代码如下:

public struct SimpleSpinLock{
    private int m_Lock;

    public void Enter(){
        while (true){
            if (Interlocked.Exchange(ref m_Lock, 1) == 0) return;
            //此处可以添加黑科技
        }
    }

    public void Leave(){
        Volatile.Write(ref m_Lock, 0);
    }
}

//下面是如何使用SimpleSpinLock的例子
public class Simple{
    private SimpleSpinLock m_lock = new SimpleSpinLock();

    public void AccessResource(){
        m_lock.Enter();
        //执行某些程序,只有一个线程可以进入这里
        m_lock.Leave();
    }
}

  这个简单的自旋锁在一个线程调用Enter()时,其他线程在调用m_lock.Enter()方法时,if (Interlocked.Exchange(ref m_lock, 1) == 0)会失败,因为该方法会将m_lock和1交换,并返回旧值,在已有线程调用m_lock.Enter()时,m_lock的旧值是1,因此该方法会在whle(true)处自旋,不断尝试获得锁。该锁的问题是,该线程没有被阻塞(挂起),而是一直在占用CPU资源,其他需要CPU资源的线程无法运行(可以在while内,我加注释的地方,加入”黑科技“来尝试解决此问题。其思路是在线程自旋的过程中,立刻交出CPU资源,可通过调用Thread.Sleep(0)或者Thread.Yield()来实现。尝试获得锁的线程交出时间片,这样当前获得锁的线程能够有更多的资源来运行程序,从而运行结束并交出锁,具体细节在这里不展开)。因此,自旋锁只适合那些运行非常快的方法。

  • Interlocked Anything 模式

  Interlocked中全部是原子性操作,那是否提供了一个方法,该方法可以接受一个委托,保证该委托在运行时是原子的。答案是没有,但是可以利用Interlocked.CompareExchange来自己实现一个。我们先来看一下利用CompareExchange来实现原子性的Maximum方法。

public static int Maximum(ref int target, int value){
    int currentValue = target, startValue, desireValue;
    do{
        startValue = currentValue;
        //可以在此处添加任何想要保证“原子性”的操作,此处是求最大值。
        desireValue = Math.Max(startValue, value);
        //注意,此处有可能被其他线程抢占,也有可能target的值被修改,因此if()语句会出问题,要使用Interlocked的方法。
        //if (startValue = target) target = desireValue;
        currentValue = Interlocked.CompareExchange(ref target, desireValue, startValue);
    } 
    //若在此时target已被修改,则重新计算最大值
    while (startValue != currentValue);
    return currentValue;
}

  此方法的思路是:从CPU将target的值读到寄存器中,到计算最大值结束,期间的任何时间target都有可能被其他线程修改。因此保证原子性就被转换成保证计算最大值时,target的值没有变过,如果变过,就重新计算。因此,在最开始的时候,startValue=currentValue,currentValue是开始计算时target的值。然后求得最大值,并保存到desireValue中。注意,此时target有可能被修改,因此调用CompareExchange方法,该方法会将target与startValue比较,如果此时两值相等,那相当于我之前说的,target从开始到最后没有改变,那么这个最大值是准确的,并将target的旧值付给currentValue,最后如果startValue==currentValue,则计算完成,否则继续循环。

  《CLR via C#》的作者Jeff很喜欢上面的方法,他在实际开发中,都是使用上面的方法,并对其进行了包装,使之能够支持Interlocked Anything。其原理就是在desireValue=Math.Max()处替换成其他方法,只要在返回结果时保证旧值和最开始读取的值一致就可以。我们来看一下他的封装:

delegate int Morpher<TResult, in TArgument>(int startValue, TArgument argument, out TResult result);

static TResult Morph<TResult, TArgumen>(ref int target, TArgumen argumen, Morpher<TResult, TArgumen> morpher){
    TResult mophorResult;
    int currentValue = target, desireValue, startValue;
    do{
        startValue = currentValue;
        desireValue = morpher(startValue, argumen, out mophorResult);
        currentValue = Interlocked.CompareExchange(ref target, desireValue, startValue);
    } 
    while (currentValue != startValue);
    return mophorResult;
}

说实话,我并不能非常好了理解这个封装,并不是不能理解做法,而是不能确定此方法到底能不能实现效果,那我们来测试一下。测试的基本思路是对一个变量执行1000次的result+=10。分别是不带线程同步的和利用Morph方法对result+=10的方法进行互锁,保证其原子性。我省去了Morpher和Morph的声明部分。之所以要在DelayAdd和DelayAdd1方法中调用ThreadSleep(20),是为了模拟当在运行较长的方法时,Morph方法是否还能够保证该方法运行的原子性。

static void Main(string[] args){
    Test(new TestAction(Add), "Add");
    Test(new TestAction(MorphAdd), "MorphAdd");
    Console.ReadLine();
}
//DelayAdd1的变种,使之能够符合Morpher的签名
static int DelayAdd(int startValue, int argument, out int result){
    Thread.Sleep(20);
    result = startValue + argument;
    return result;
}

static void DelayAdd1(int argument, ref int result){
    Thread.Sleep(20);
    result += argument;
}
//测试的不具有线程同步的方法。
static void Add(ref int result){
    DelayAdd1(10, ref result);
}
//具有线程同步的方法。
static void MorphAdd(ref int result){
    Morph(ref result, 10, new Morpher<int, int>(DelayAdd));
}
//要测试的委托签名
delegate void TestAction(ref int result);
//公共测试方法
static void Test(TestAction action, string actionName){
    int result = 0;
    var tList = new Task[1000];
    for (int i = 0; i < 1000; i++)
        tList[i] = Task.Run(() =>
        {
            action(ref result);
        });
    Task.WhenAll(tList).GetAwaiter().OnCompleted(
        () => Console.WriteLine("{0}, Result is {1}", actionName, result));
}

运行,得到的结果是:

Add, Result is 8440
MorphAdd, Result is 10000

运行1000次result += 10,普通的Add不能够得到正确结果,但是MorphAdd可以。这是因为在1000次的Add中,某几个Add是同时调用的,result+=10在同一时间调用了多次,因此有156次的Add因为并行而被吞噬了。值得注意的是,MorphAdd方法因为需要线程同步,因此执行时间要慢很多。但是这些付出是值得的,因为这保证了结果的正确。

  上述例子证明了Morph方法能够保证委托的原子性,且该方法既不会阻塞线程也不会长时间的自旋,推荐大家在实际中使用该方法。

  本文中,我先介绍了Interlocked类中较常用的方法,以及Interlocked.Increment()方法与volatile关键字的对比,结论是虽然将变量设置为所有读写都是“易变的”看起来很浪费,但是该关键字能够保证在单线程时几乎没有性能损失,大部分情况下和原生的读写是一样的速度,且volatile比Interlocked类中提供的写要快2-4倍,但是在并发状态下其性能和volatile关键字是没有差别的。之后我介绍了用Interlocked类中的方法来实现简单的自旋锁,该锁的优点是在非竟态情况下非常快,但是在竟态情况下,未获得锁的线程会一直处于自旋状态,白白浪费CPU。最后介绍了《CLR via C#》书中提到的Interlocked Anything的方法(文中其他的知识点大多也是提取自《CLR via C#》),并测试了该方法确实可以保证委托的原子性,且不会阻塞线程,没有锁,不会造成死锁。至此线程同步中互锁构造就讲完了,后面我会给大家介绍内核构造的信号量和其他锁。